services/syncbase: Export the NeighborConnectionTimeout
So that it can be used as an option to the RPC Server that is created to
host the syncbase service.
Without this change, the following could happen:
- Syncbase A and syncbase B discover each other
- Because both have internet connectivity, they have
a connection to the proxy and an endpoint through it
- A and B invoke methods on each other through their
proxied endpoints
- B loses internet connectivity (and correctly connects
to A through some other means such as bluetooth)
- However, when A wants to invoke a method on B,
it ends up trying to re-use the (defunct) proxied
endpoint in its cache instead of the bluetooth
connection.
As a result, changes in B would be propagated to A,
but not the reverse.
With this channel timeout being provided to the server,
A will determine (within 5 seconds) that the proxied
connection is defunct and it should try some other
endpoint instead.
RPC requests to multiple endpoints cannot be sent in
parallel in order to maintain "at most once" semantics
of an RPC invocation. It is possible that this is not
needed for the syncbase/GetDeltas use-case, but that
larger investigation is left as an excercise for the future.
MultiPart: 2/2
Change-Id: I9325162244cf86556ad2675811dce1b670dfbb48
diff --git a/services/syncbase/syncbaselib/serve.go b/services/syncbase/syncbaselib/serve.go
index c67ec86..81b4842 100644
--- a/services/syncbase/syncbaselib/serve.go
+++ b/services/syncbase/syncbaselib/serve.go
@@ -11,9 +11,11 @@
"v.io/v23"
"v.io/v23/context"
+ "v.io/v23/options"
"v.io/v23/rpc"
"v.io/x/ref/lib/security/securityflag"
"v.io/x/ref/services/syncbase/server"
+ "v.io/x/ref/services/syncbase/vsync"
)
// Serve starts the Syncbase server. Returns rpc.Server and rpc.Dispatcher for
@@ -55,7 +57,7 @@
// Publish the service in the mount table.
ctx, cancel := context.WithCancel(ctx)
- ctx, s, err := v23.WithNewDispatchingServer(ctx, opts.Name, d)
+ ctx, s, err := v23.WithNewDispatchingServer(ctx, opts.Name, d, options.ChannelTimeout(vsync.NeighborConnectionTimeout))
if err != nil {
ctx.Fatal("v23.WithNewDispatchingServer() failed: ", err)
}
diff --git a/services/syncbase/vsync/parameters.go b/services/syncbase/vsync/parameters.go
index 4b20d4d..4b18430 100644
--- a/services/syncbase/vsync/parameters.go
+++ b/services/syncbase/vsync/parameters.go
@@ -16,12 +16,13 @@
// connections to peers.
channelTimeout = 2 * time.Second
+ // NeighborConnectionTimeout is the time duration we wait for a connection to be
+ // established with a peer discovered from the neighborhood.
+ //
// TODO(suharshs): Make the timeouts below more dynamic based on network. (i.e.
// bt vs tcp). Currently bt connection establishment takes ~3 seconds, so
// neighborhood connections get more time than cloud connections.
- // neighborConnectionTimeout is the time duration we wait for a connection to be
- // established with a peer discovered from the neighborhood.
- neighborConnectionTimeout = 5 * time.Second
+ NeighborConnectionTimeout = 5 * time.Second
// cloudConnectionTimeout is the duration we wait for a connection to be
// established with a cloud peer.
diff --git a/services/syncbase/vsync/peer_manager.go b/services/syncbase/vsync/peer_manager.go
index b8c695e..401ea06 100644
--- a/services/syncbase/vsync/peer_manager.go
+++ b/services/syncbase/vsync/peer_manager.go
@@ -456,7 +456,7 @@
vlog.VI(4).Infof("sync: pingPeers: sending names %v", names)
- res, err := ping.PingInParallel(ctx, names, neighborConnectionTimeout, channelTimeout)
+ res, err := ping.PingInParallel(ctx, names, NeighborConnectionTimeout, channelTimeout)
if err != nil {
return nil
}
diff --git a/services/syncbase/vsync/syncgroup.go b/services/syncbase/vsync/syncgroup.go
index 32f07c7..50256f3 100644
--- a/services/syncbase/vsync/syncgroup.go
+++ b/services/syncbase/vsync/syncgroup.go
@@ -1275,7 +1275,7 @@
neighbors := ss.filterSyncgroupAdmins(dbId, sgId)
for _, svc := range neighbors {
for _, addr := range svc.Addresses {
- ctx, cancel := context.WithTimeout(ctxIn, neighborConnectionTimeout)
+ ctx, cancel := context.WithTimeout(ctxIn, NeighborConnectionTimeout)
// TODO(fredq): check that the service at addr has the expectedSyncbaseBlessings.
c := interfaces.SyncClient(naming.Join(addr, common.SyncbaseSuffix))
sg, vers, gv, err := c.JoinSyncgroupAtAdmin(ctx, dbId, sgId, localSyncbaseName, myInfo)
