wspr: Make Lookup requests made during server shutdown return an error.
PROBLEM:
When an RPC is made to a JS server after the tab has closed,
the RPC just hangs forever, since the Go Server's Listeners only close
after the server has completed Unmounting, and requests made after Stop
is called aren't handle by WSPR currently.
SOLUTION:
WSPR returns retryable errors when the the tab is closed, I am working on
an additional CL to support this in the Go Client retry code.
I also cleaned some minor things up while I was here.
Change-Id: I0d63d09b9ffd512d7462eb4a8d4d6b9154969665
diff --git a/services/wspr/browsprd/main_nacl.go b/services/wspr/browsprd/main_nacl.go
index a56e332..d421119 100644
--- a/services/wspr/browsprd/main_nacl.go
+++ b/services/wspr/browsprd/main_nacl.go
@@ -299,13 +299,11 @@
func (inst *browsprInstance) HandleBrowsprMessage(instanceId int32, origin string, message ppapi.Var) error {
str, err := message.AsString()
if err != nil {
- // TODO(bprosnitz) Remove. We shouldn't panic on user input.
return fmt.Errorf("Error while converting message to string: %v", err)
}
vlog.VI(1).Infof("Calling browspr's HandleMessage: instanceId %d origin %s message %s", instanceId, origin, str)
if err := inst.browspr.HandleMessage(instanceId, origin, str); err != nil {
- // TODO(bprosnitz) Remove. We shouldn't panic on user input.
return fmt.Errorf("Error while handling message in browspr: %v", err)
}
return nil
diff --git a/services/wspr/internal/browspr/browspr.go b/services/wspr/internal/browspr/browspr.go
index a8f1f37..97bc317 100644
--- a/services/wspr/internal/browspr/browspr.go
+++ b/services/wspr/internal/browspr/browspr.go
@@ -30,9 +30,8 @@
postMessage func(instanceId int32, ty, msg string)
principalManager *principal.PrincipalManager
- // Locks activeInstances
mu sync.Mutex
- activeInstances map[int32]*pipe
+ activeInstances map[int32]*pipe // GUARDED_BY mu
}
// Create a new Browspr instance.
diff --git a/services/wspr/internal/rpc/server/dispatcher.go b/services/wspr/internal/rpc/server/dispatcher.go
index 82efad2..6d8e2cc 100644
--- a/services/wspr/internal/rpc/server/dispatcher.go
+++ b/services/wspr/internal/rpc/server/dispatcher.go
@@ -7,7 +7,6 @@
import (
"bytes"
"encoding/json"
- "fmt"
"sync"
"v.io/v23/rpc"
@@ -60,6 +59,7 @@
invokerFactory invokerFactory
authFactory authFactory
outstandingLookups map[int32]chan lookupReply
+ closed bool
}
var _ rpc.Dispatcher = (*dispatcher)(nil)
@@ -78,17 +78,23 @@
func (d *dispatcher) Cleanup() {
d.mu.Lock()
defer d.mu.Unlock()
+ d.closed = true
for _, ch := range d.outstandingLookups {
- verr := verror.Convert(verror.ErrInternal, nil, fmt.Errorf("Cleaning up dispatcher")).(verror.E)
+ verr := NewErrServerStopped(nil).(verror.E)
ch <- lookupReply{Err: &verr}
}
}
// Lookup implements dispatcher interface Lookup.
func (d *dispatcher) Lookup(suffix string) (interface{}, security.Authorizer, error) {
- flow := d.flowFactory.createFlow()
+ // If the server has been closed, we immediately return a retryable error.
d.mu.Lock()
+ if d.closed {
+ d.mu.Unlock()
+ return nil, nil, NewErrServerStopped(nil)
+ }
+ flow := d.flowFactory.createFlow()
ch := make(chan lookupReply, 1)
d.outstandingLookups[flow.ID] = ch
d.mu.Unlock()
diff --git a/services/wspr/internal/rpc/server/server.go b/services/wspr/internal/rpc/server/server.go
index 3a97693..c9bf07c 100644
--- a/services/wspr/internal/rpc/server/server.go
+++ b/services/wspr/internal/rpc/server/server.go
@@ -101,16 +101,11 @@
id uint32
helper ServerHelper
- // outstandingRequestLock should be acquired only to update the
- // outstanding request maps below.
- outstandingRequestLock sync.Mutex
-
- // The set of outstanding server requests.
- outstandingServerRequests map[int32]chan *lib.ServerRpcReply
-
- outstandingAuthRequests map[int32]chan error
-
- outstandingValidationRequests map[int32]chan []error
+ // outstandingRequestLock should be acquired only to update the outstanding request maps below.
+ outstandingRequestLock sync.Mutex
+ outstandingServerRequests map[int32]chan *lib.ServerRpcReply // GUARDED_BY outstandingRequestLock
+ outstandingAuthRequests map[int32]chan error // GUARDED_BY outstandingRequestLock
+ outstandingValidationRequests map[int32]chan []error // GUARDED_BY outstandingRequestLock
// statusClose will be closed when the server is shutting down, this will
// cause the status poller to exit.
@@ -440,8 +435,6 @@
func (s *Server) convertSecurityCall(ctx *context.T, includeBlessingStrings bool) SecurityCall {
call := security.GetCall(ctx)
- // TODO(bprosnitz) Local/Remote Endpoint should always be non-nil, but isn't
- // due to a TODO in vc/auth.go
var localEndpoint string
if call.LocalEndpoint() != nil {
localEndpoint = call.LocalEndpoint().String()
@@ -606,7 +599,7 @@
if ch == nil {
vlog.Errorf("unexpected result from JavaScript. No channel "+
"for MessageId: %d exists. Ignoring the results(%s)", id, data)
- //Ignore unknown responses that don't belong to any channel
+ // Ignore unknown responses that don't belong to any channel
return
}
// Decode the result and send it through the channel
@@ -636,7 +629,7 @@
if ch == nil {
vlog.Errorf("unexpected result from JavaScript. No channel "+
"for validation response with MessageId: %d exists. Ignoring the results(%s)", id, data)
- //Ignore unknown responses that don't belong to any channel
+ // Ignore unknown responses that don't belong to any channel
return
}
@@ -692,7 +685,6 @@
for _, ch := range s.outstandingAuthRequests {
ch <- fmt.Errorf("Cleaning up server")
}
- s.outstandingAuthRequests = make(map[int32]chan error)
for _, ch := range s.outstandingServerRequests {
select {
@@ -707,10 +699,9 @@
s.serverStateLock.Unlock()
s.server.Stop()
- // Only clear the validation requests map after stopping. clearing
- // it can cause the publisher to get stuck waiting for a caveat validtion
- // that will never be answered, which in turn causes it to not be
- // able to stop which prevents the server from stopping.
+ // Only clear the validation requests map after stopping. Clearing them before
+ // can cause the publisher to get stuck waiting for a caveat validation that
+ // will never be answered, which prevents the server from stopping.
s.serverStateLock.Lock()
s.outstandingRequestLock.Lock()
s.outstandingValidationRequests = make(map[int32]chan []error)
diff --git a/services/wspr/internal/rpc/server/server.vdl b/services/wspr/internal/rpc/server/server.vdl
index 043a986..00fad2b 100644
--- a/services/wspr/internal/rpc/server/server.vdl
+++ b/services/wspr/internal/rpc/server/server.vdl
@@ -33,4 +33,5 @@
error (
CaveatValidationTimeout() {"en": "Caveat validation has timed out"}
InvalidValidationResponseFromJavascript() {"en": "Invalid validation response from javascript"}
+ ServerStopped() {RetryBackoff, "en": "Server has been stopped"}
)
diff --git a/services/wspr/internal/rpc/server/server.vdl.go b/services/wspr/internal/rpc/server/server.vdl.go
index 21c0041..eb13c27 100644
--- a/services/wspr/internal/rpc/server/server.vdl.go
+++ b/services/wspr/internal/rpc/server/server.vdl.go
@@ -64,11 +64,13 @@
var (
ErrCaveatValidationTimeout = verror.Register("v.io/x/ref/services/wspr/internal/rpc/server.CaveatValidationTimeout", verror.NoRetry, "{1:}{2:} Caveat validation has timed out")
ErrInvalidValidationResponseFromJavascript = verror.Register("v.io/x/ref/services/wspr/internal/rpc/server.InvalidValidationResponseFromJavascript", verror.NoRetry, "{1:}{2:} Invalid validation response from javascript")
+ ErrServerStopped = verror.Register("v.io/x/ref/services/wspr/internal/rpc/server.ServerStopped", verror.RetryBackoff, "{1:}{2:} Server has been stopped")
)
func init() {
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrCaveatValidationTimeout.ID), "{1:}{2:} Caveat validation has timed out")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrInvalidValidationResponseFromJavascript.ID), "{1:}{2:} Invalid validation response from javascript")
+ i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrServerStopped.ID), "{1:}{2:} Server has been stopped")
}
// NewErrCaveatValidationTimeout returns an error with the ErrCaveatValidationTimeout ID.
@@ -80,3 +82,8 @@
func NewErrInvalidValidationResponseFromJavascript(ctx *context.T) error {
return verror.New(ErrInvalidValidationResponseFromJavascript, ctx)
}
+
+// NewErrServerStopped returns an error with the ErrServerStopped ID.
+func NewErrServerStopped(ctx *context.T) error {
+ return verror.New(ErrServerStopped, ctx)
+}
