Merge "services/groups: Authorization fixes."
diff --git a/services/syncbase/server/interfaces/sync.vdl b/services/syncbase/server/interfaces/sync.vdl
index a3d7f7f..6dd6f25 100644
--- a/services/syncbase/server/interfaces/sync.vdl
+++ b/services/syncbase/server/interfaces/sync.vdl
@@ -14,15 +14,8 @@
type Sync interface {
// GetDeltas returns the responder's current generation vector and all
// the missing log records when compared to the initiator's generation
- // vector. This process happens one Database at a time encompassing all
- // the SyncGroups common to the initiator and the responder. For each
- // Database, the initiator sends a DeltaReq. In response, the
- // responder sends a "Start" DeltaResp record, all the missing log
- // records, the responder's genvector, and a "Finish" DeltaResp
- // record. The initiator parses the stream between a Start and a Finish
- // record as the response to its DeltaReq, and then moves on to the
- // next Database in common with this responder.
- GetDeltas(initiator string) stream<DeltaReq, DeltaResp> error {access.Read}
+ // vector for one Database for either SyncGroup metadata or data.
+ GetDeltas(req DeltaReq, initiator string) stream<_, DeltaResp> error {access.Read}
// SyncGroup-related methods.
diff --git a/services/syncbase/server/interfaces/sync.vdl.go b/services/syncbase/server/interfaces/sync.vdl.go
index a281543..3006f3c 100644
--- a/services/syncbase/server/interfaces/sync.vdl.go
+++ b/services/syncbase/server/interfaces/sync.vdl.go
@@ -28,15 +28,8 @@
type SyncClientMethods interface {
// GetDeltas returns the responder's current generation vector and all
// the missing log records when compared to the initiator's generation
- // vector. This process happens one Database at a time encompassing all
- // the SyncGroups common to the initiator and the responder. For each
- // Database, the initiator sends a DeltaReq. In response, the
- // responder sends a "Start" DeltaResp record, all the missing log
- // records, the responder's genvector, and a "Finish" DeltaResp
- // record. The initiator parses the stream between a Start and a Finish
- // record as the response to its DeltaReq, and then moves on to the
- // next Database in common with this responder.
- GetDeltas(ctx *context.T, initiator string, opts ...rpc.CallOpt) (SyncGetDeltasClientCall, error)
+ // vector for one Database for either SyncGroup metadata or data.
+ GetDeltas(ctx *context.T, req DeltaReq, initiator string, opts ...rpc.CallOpt) (SyncGetDeltasClientCall, error)
// PublishSyncGroup is typically invoked on a "central" peer to publish
// the SyncGroup.
PublishSyncGroup(ctx *context.T, sg SyncGroup, opts ...rpc.CallOpt) error
@@ -78,9 +71,9 @@
name string
}
-func (c implSyncClientStub) GetDeltas(ctx *context.T, i0 string, opts ...rpc.CallOpt) (ocall SyncGetDeltasClientCall, err error) {
+func (c implSyncClientStub) GetDeltas(ctx *context.T, i0 DeltaReq, i1 string, opts ...rpc.CallOpt) (ocall SyncGetDeltasClientCall, err error) {
var call rpc.ClientCall
- if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "GetDeltas", []interface{}{i0}, opts...); err != nil {
+ if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "GetDeltas", []interface{}{i0, i1}, opts...); err != nil {
return
}
ocall = &implSyncGetDeltasClientCall{ClientCall: call}
@@ -143,30 +136,13 @@
// Err returns any error encountered by Advance. Never blocks.
Err() error
}
- // SendStream returns the send side of the Sync.GetDeltas client stream.
- SendStream() interface {
- // Send places the item onto the output stream. Returns errors
- // encountered while sending, or if Send is called after Close or
- // the stream has been canceled. Blocks if there is no buffer
- // space; will unblock when buffer space is available or after
- // the stream has been canceled.
- Send(item DeltaReq) error
- // Close indicates to the server that no more items will be sent;
- // server Recv calls will receive io.EOF after all sent items.
- // This is an optional call - e.g. a client might call Close if it
- // needs to continue receiving items from the server after it's
- // done sending. Returns errors encountered while closing, or if
- // Close is called after the stream has been canceled. Like Send,
- // blocks if there is no buffer space available.
- Close() error
- }
}
// SyncGetDeltasClientCall represents the call returned from Sync.GetDeltas.
type SyncGetDeltasClientCall interface {
SyncGetDeltasClientStream
- // Finish performs the equivalent of SendStream().Close, then blocks until
- // the server is done, and returns the positional return values for the call.
+ // Finish blocks until the server is done, and returns the positional return
+ // values for call.
//
// Finish returns immediately if the call has been canceled; depending on the
// timing the output could either be an error signaling cancelation, or the
@@ -209,23 +185,6 @@
}
return c.c.errRecv
}
-func (c *implSyncGetDeltasClientCall) SendStream() interface {
- Send(item DeltaReq) error
- Close() error
-} {
- return implSyncGetDeltasClientCallSend{c}
-}
-
-type implSyncGetDeltasClientCallSend struct {
- c *implSyncGetDeltasClientCall
-}
-
-func (c implSyncGetDeltasClientCallSend) Send(item DeltaReq) error {
- return c.c.Send(item)
-}
-func (c implSyncGetDeltasClientCallSend) Close() error {
- return c.c.CloseSend()
-}
func (c *implSyncGetDeltasClientCall) Finish() (err error) {
err = c.ClientCall.Finish()
return
@@ -479,15 +438,8 @@
type SyncServerMethods interface {
// GetDeltas returns the responder's current generation vector and all
// the missing log records when compared to the initiator's generation
- // vector. This process happens one Database at a time encompassing all
- // the SyncGroups common to the initiator and the responder. For each
- // Database, the initiator sends a DeltaReq. In response, the
- // responder sends a "Start" DeltaResp record, all the missing log
- // records, the responder's genvector, and a "Finish" DeltaResp
- // record. The initiator parses the stream between a Start and a Finish
- // record as the response to its DeltaReq, and then moves on to the
- // next Database in common with this responder.
- GetDeltas(ctx *context.T, call SyncGetDeltasServerCall, initiator string) error
+ // vector for one Database for either SyncGroup metadata or data.
+ GetDeltas(ctx *context.T, call SyncGetDeltasServerCall, req DeltaReq, initiator string) error
// PublishSyncGroup is typically invoked on a "central" peer to publish
// the SyncGroup.
PublishSyncGroup(ctx *context.T, call rpc.ServerCall, sg SyncGroup) error
@@ -521,15 +473,8 @@
type SyncServerStubMethods interface {
// GetDeltas returns the responder's current generation vector and all
// the missing log records when compared to the initiator's generation
- // vector. This process happens one Database at a time encompassing all
- // the SyncGroups common to the initiator and the responder. For each
- // Database, the initiator sends a DeltaReq. In response, the
- // responder sends a "Start" DeltaResp record, all the missing log
- // records, the responder's genvector, and a "Finish" DeltaResp
- // record. The initiator parses the stream between a Start and a Finish
- // record as the response to its DeltaReq, and then moves on to the
- // next Database in common with this responder.
- GetDeltas(ctx *context.T, call *SyncGetDeltasServerCallStub, initiator string) error
+ // vector for one Database for either SyncGroup metadata or data.
+ GetDeltas(ctx *context.T, call *SyncGetDeltasServerCallStub, req DeltaReq, initiator string) error
// PublishSyncGroup is typically invoked on a "central" peer to publish
// the SyncGroup.
PublishSyncGroup(ctx *context.T, call rpc.ServerCall, sg SyncGroup) error
@@ -585,8 +530,8 @@
gs *rpc.GlobState
}
-func (s implSyncServerStub) GetDeltas(ctx *context.T, call *SyncGetDeltasServerCallStub, i0 string) error {
- return s.impl.GetDeltas(ctx, call, i0)
+func (s implSyncServerStub) GetDeltas(ctx *context.T, call *SyncGetDeltasServerCallStub, i0 DeltaReq, i1 string) error {
+ return s.impl.GetDeltas(ctx, call, i0, i1)
}
func (s implSyncServerStub) PublishSyncGroup(ctx *context.T, call rpc.ServerCall, i0 SyncGroup) error {
@@ -632,8 +577,9 @@
Methods: []rpc.MethodDesc{
{
Name: "GetDeltas",
- Doc: "// GetDeltas returns the responder's current generation vector and all\n// the missing log records when compared to the initiator's generation\n// vector. This process happens one Database at a time encompassing all\n// the SyncGroups common to the initiator and the responder. For each\n// Database, the initiator sends a DeltaReq. In response, the\n// responder sends a \"Start\" DeltaResp record, all the missing log\n// records, the responder's genvector, and a \"Finish\" DeltaResp\n// record. The initiator parses the stream between a Start and a Finish\n// record as the response to its DeltaReq, and then moves on to the\n// next Database in common with this responder.",
+ Doc: "// GetDeltas returns the responder's current generation vector and all\n// the missing log records when compared to the initiator's generation\n// vector for one Database for either SyncGroup metadata or data.",
InArgs: []rpc.ArgDesc{
+ {"req", ``}, // DeltaReq
{"initiator", ``}, // string
},
Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
@@ -691,18 +637,6 @@
// SyncGetDeltasServerStream is the server stream for Sync.GetDeltas.
type SyncGetDeltasServerStream interface {
- // RecvStream returns the receiver side of the Sync.GetDeltas server stream.
- RecvStream() interface {
- // Advance stages an item so that it may be retrieved via Value. Returns
- // true iff there is an item to retrieve. Advance must be called before
- // Value is called. May block if an item is not available.
- Advance() bool
- // Value returns the item that was staged by Advance. May panic if Advance
- // returned false or was not called. Never blocks.
- Value() DeltaReq
- // Err returns any error encountered by Advance. Never blocks.
- Err() error
- }
// SendStream returns the send side of the Sync.GetDeltas server stream.
SendStream() interface {
// Send places the item onto the output stream. Returns errors encountered
@@ -722,8 +656,6 @@
// a typesafe stub that implements SyncGetDeltasServerCall.
type SyncGetDeltasServerCallStub struct {
rpc.StreamServerCall
- valRecv DeltaReq
- errRecv error
}
// Init initializes SyncGetDeltasServerCallStub from rpc.StreamServerCall.
@@ -731,34 +663,6 @@
s.StreamServerCall = call
}
-// RecvStream returns the receiver side of the Sync.GetDeltas server stream.
-func (s *SyncGetDeltasServerCallStub) RecvStream() interface {
- Advance() bool
- Value() DeltaReq
- Err() error
-} {
- return implSyncGetDeltasServerCallRecv{s}
-}
-
-type implSyncGetDeltasServerCallRecv struct {
- s *SyncGetDeltasServerCallStub
-}
-
-func (s implSyncGetDeltasServerCallRecv) Advance() bool {
- s.s.valRecv = DeltaReq{}
- s.s.errRecv = s.s.Recv(&s.s.valRecv)
- return s.s.errRecv == nil
-}
-func (s implSyncGetDeltasServerCallRecv) Value() DeltaReq {
- return s.s.valRecv
-}
-func (s implSyncGetDeltasServerCallRecv) Err() error {
- if s.s.errRecv == io.EOF {
- return nil
- }
- return s.s.errRecv
-}
-
// SendStream returns the send side of the Sync.GetDeltas server stream.
func (s *SyncGetDeltasServerCallStub) SendStream() interface {
Send(item DeltaResp) error
diff --git a/services/syncbase/server/interfaces/sync_types.go b/services/syncbase/server/interfaces/sync_types.go
new file mode 100644
index 0000000..39732b1
--- /dev/null
+++ b/services/syncbase/server/interfaces/sync_types.go
@@ -0,0 +1,21 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package interfaces
+
+func (in GenVector) DeepCopy() GenVector {
+ out := make(GenVector)
+ for p, inpgv := range in {
+ out[p] = inpgv.DeepCopy()
+ }
+ return out
+}
+
+func (in PrefixGenVector) DeepCopy() PrefixGenVector {
+ out := make(PrefixGenVector)
+ for id, gen := range in {
+ out[id] = gen
+ }
+ return out
+}
diff --git a/services/syncbase/server/interfaces/sync_types.vdl b/services/syncbase/server/interfaces/sync_types.vdl
index 692a731..65a7f32 100644
--- a/services/syncbase/server/interfaces/sync_types.vdl
+++ b/services/syncbase/server/interfaces/sync_types.vdl
@@ -29,6 +29,7 @@
// PrefixGenVector is the generation vector for a data prefix, which maps each
// device id to its last locally known generation in the scope of that prefix.
+// TODO(hpucha): Rename this type.
type PrefixGenVector map[uint64]uint64
// GenVector is the generation vector for a Database, and maps prefixes to their
@@ -70,6 +71,7 @@
}
// GroupId is a globally unique SyncGroup ID.
+// TODO(hpucha): Make this a string since now the syncgroup id is an object id.
type GroupId uint64
// Possible states for a SyncGroup.
@@ -99,21 +101,35 @@
Joiners map[string]wire.SyncGroupMemberInfo // map of joiners to their metadata
}
-// DeltaReq contains the initiator's genvector and the set of SyncGroups it is
-// interested in within a Database (specified by the AppName/DbName) when
+// DeltaReq contains a request to sync either data or SyncGroup metadata for a
+// Database.
+type DeltaReq union {
+ Sgs SgDeltaReq
+ Data DataDeltaReq
+}
+
+// DataDeltaReq contains the initiator's genvector and the set of SyncGroups it
+// is interested in within a Database (specified by the AppName/DbName) when
// requesting deltas for that Database.
-type DeltaReq struct {
+type DataDeltaReq struct {
AppName string
DbName string
SgIds set[GroupId]
InitVec GenVector
}
+// SgDeltaReq contains the initiator's genvector for the SyncGroups it is
+// interested in within a Database (specified by the AppName/DbName) when
+// requesting deltas for those SyncGroups.
+type SgDeltaReq struct {
+ AppName string
+ DbName string
+ InitVec GenVector // Contains a genvector per SyncGroup.
+}
+
// DeltaResp contains the responder's genvector or the missing log records
// returned in response to an initiator's request for deltas for a Database.
type DeltaResp union {
- Start bool
- Finish bool
Rec LogRec
RespVec GenVector
}
diff --git a/services/syncbase/server/interfaces/sync_types.vdl.go b/services/syncbase/server/interfaces/sync_types.vdl.go
index 1fb1396..ca29abc 100644
--- a/services/syncbase/server/interfaces/sync_types.vdl.go
+++ b/services/syncbase/server/interfaces/sync_types.vdl.go
@@ -20,6 +20,7 @@
// PrefixGenVector is the generation vector for a data prefix, which maps each
// device id to its last locally known generation in the scope of that prefix.
+// TODO(hpucha): Rename this type.
type PrefixGenVector map[uint64]uint64
func (PrefixGenVector) __VDLReflect(struct {
@@ -78,6 +79,7 @@
}
// GroupId is a globally unique SyncGroup ID.
+// TODO(hpucha): Make this a string since now the syncgroup id is an object id.
type GroupId uint64
func (GroupId) __VDLReflect(struct {
@@ -157,18 +159,72 @@
}) {
}
-// DeltaReq contains the initiator's genvector and the set of SyncGroups it is
-// interested in within a Database (specified by the AppName/DbName) when
+type (
+ // DeltaReq represents any single field of the DeltaReq union type.
+ //
+ // DeltaReq contains a request to sync either data or SyncGroup metadata for a
+ // Database.
+ DeltaReq interface {
+ // Index returns the field index.
+ Index() int
+ // Interface returns the field value as an interface.
+ Interface() interface{}
+ // Name returns the field name.
+ Name() string
+ // __VDLReflect describes the DeltaReq union type.
+ __VDLReflect(__DeltaReqReflect)
+ }
+ // DeltaReqSgs represents field Sgs of the DeltaReq union type.
+ DeltaReqSgs struct{ Value SgDeltaReq }
+ // DeltaReqData represents field Data of the DeltaReq union type.
+ DeltaReqData struct{ Value DataDeltaReq }
+ // __DeltaReqReflect describes the DeltaReq union type.
+ __DeltaReqReflect struct {
+ Name string `vdl:"v.io/x/ref/services/syncbase/server/interfaces.DeltaReq"`
+ Type DeltaReq
+ Union struct {
+ Sgs DeltaReqSgs
+ Data DeltaReqData
+ }
+ }
+)
+
+func (x DeltaReqSgs) Index() int { return 0 }
+func (x DeltaReqSgs) Interface() interface{} { return x.Value }
+func (x DeltaReqSgs) Name() string { return "Sgs" }
+func (x DeltaReqSgs) __VDLReflect(__DeltaReqReflect) {}
+
+func (x DeltaReqData) Index() int { return 1 }
+func (x DeltaReqData) Interface() interface{} { return x.Value }
+func (x DeltaReqData) Name() string { return "Data" }
+func (x DeltaReqData) __VDLReflect(__DeltaReqReflect) {}
+
+// DataDeltaReq contains the initiator's genvector and the set of SyncGroups it
+// is interested in within a Database (specified by the AppName/DbName) when
// requesting deltas for that Database.
-type DeltaReq struct {
+type DataDeltaReq struct {
AppName string
DbName string
SgIds map[GroupId]struct{}
InitVec GenVector
}
-func (DeltaReq) __VDLReflect(struct {
- Name string `vdl:"v.io/x/ref/services/syncbase/server/interfaces.DeltaReq"`
+func (DataDeltaReq) __VDLReflect(struct {
+ Name string `vdl:"v.io/x/ref/services/syncbase/server/interfaces.DataDeltaReq"`
+}) {
+}
+
+// SgDeltaReq contains the initiator's genvector for the SyncGroups it is
+// interested in within a Database (specified by the AppName/DbName) when
+// requesting deltas for those SyncGroups.
+type SgDeltaReq struct {
+ AppName string
+ DbName string
+ InitVec GenVector // Contains a genvector per SyncGroup.
+}
+
+func (SgDeltaReq) __VDLReflect(struct {
+ Name string `vdl:"v.io/x/ref/services/syncbase/server/interfaces.SgDeltaReq"`
}) {
}
@@ -187,10 +243,6 @@
// __VDLReflect describes the DeltaResp union type.
__VDLReflect(__DeltaRespReflect)
}
- // DeltaRespStart represents field Start of the DeltaResp union type.
- DeltaRespStart struct{ Value bool }
- // DeltaRespFinish represents field Finish of the DeltaResp union type.
- DeltaRespFinish struct{ Value bool }
// DeltaRespRec represents field Rec of the DeltaResp union type.
DeltaRespRec struct{ Value LogRec }
// DeltaRespRespVec represents field RespVec of the DeltaResp union type.
@@ -200,30 +252,18 @@
Name string `vdl:"v.io/x/ref/services/syncbase/server/interfaces.DeltaResp"`
Type DeltaResp
Union struct {
- Start DeltaRespStart
- Finish DeltaRespFinish
Rec DeltaRespRec
RespVec DeltaRespRespVec
}
}
)
-func (x DeltaRespStart) Index() int { return 0 }
-func (x DeltaRespStart) Interface() interface{} { return x.Value }
-func (x DeltaRespStart) Name() string { return "Start" }
-func (x DeltaRespStart) __VDLReflect(__DeltaRespReflect) {}
-
-func (x DeltaRespFinish) Index() int { return 1 }
-func (x DeltaRespFinish) Interface() interface{} { return x.Value }
-func (x DeltaRespFinish) Name() string { return "Finish" }
-func (x DeltaRespFinish) __VDLReflect(__DeltaRespReflect) {}
-
-func (x DeltaRespRec) Index() int { return 2 }
+func (x DeltaRespRec) Index() int { return 0 }
func (x DeltaRespRec) Interface() interface{} { return x.Value }
func (x DeltaRespRec) Name() string { return "Rec" }
func (x DeltaRespRec) __VDLReflect(__DeltaRespReflect) {}
-func (x DeltaRespRespVec) Index() int { return 3 }
+func (x DeltaRespRespVec) Index() int { return 1 }
func (x DeltaRespRespVec) Interface() interface{} { return x.Value }
func (x DeltaRespRespVec) Name() string { return "RespVec" }
func (x DeltaRespRespVec) __VDLReflect(__DeltaRespReflect) {}
@@ -257,6 +297,8 @@
vdl.Register((*SyncGroupStatus)(nil))
vdl.Register((*SyncGroup)(nil))
vdl.Register((*DeltaReq)(nil))
+ vdl.Register((*DataDeltaReq)(nil))
+ vdl.Register((*SgDeltaReq)(nil))
vdl.Register((*DeltaResp)(nil))
vdl.Register((*ChunkHash)(nil))
vdl.Register((*ChunkData)(nil))
diff --git a/services/syncbase/vsync/blob.go b/services/syncbase/vsync/blob.go
index 51a8f7c..eb78d00 100644
--- a/services/syncbase/vsync/blob.go
+++ b/services/syncbase/vsync/blob.go
@@ -380,33 +380,7 @@
func (sd *syncDatabase) getMountTables(ctx *context.T, peer string) (map[string]struct{}, error) {
ss := sd.sync.(*syncService)
mInfo := ss.copyMemberInfo(ctx, peer)
-
- mtTables := make(map[string]struct{})
- for gdbName, sgInfo := range mInfo.db2sg {
- appName, dbName, err := splitAppDbName(ctx, gdbName)
- if err != nil {
- return nil, err
- }
- st, err := ss.getDbStore(ctx, nil, appName, dbName)
- if err != nil {
- return nil, err
- }
-
- for id := range sgInfo {
- sg, err := getSyncGroupById(ctx, st, id)
- if err != nil {
- continue
- }
- if _, ok := sg.Joiners[peer]; !ok {
- // Peer is no longer part of the SyncGroup.
- continue
- }
- for _, mt := range sg.Spec.MountTables {
- mtTables[mt] = struct{}{}
- }
- }
- }
- return mtTables, nil
+ return mInfo.mtTables, nil
}
// TODO(hpucha): Persist the blob directory periodically.
diff --git a/services/syncbase/vsync/conflict_resolution.go b/services/syncbase/vsync/conflict_resolution.go
index a8e41f6..689e2d3 100644
--- a/services/syncbase/vsync/conflict_resolution.go
+++ b/services/syncbase/vsync/conflict_resolution.go
@@ -131,11 +131,7 @@
if err != nil {
return nil, err
}
- dev, gen, err := splitLogRecKey(ctx, logKey)
- if err != nil {
- return nil, err
- }
- lrecs[p], err = getLogRec(ctx, iSt.tx, dev, gen)
+ lrecs[p], err = getLogRecByKey(ctx, iSt.tx, logKey)
if err != nil {
return nil, err
}
diff --git a/services/syncbase/vsync/dag_test.go b/services/syncbase/vsync/dag_test.go
index 2644655..c72b3a7 100644
--- a/services/syncbase/vsync/dag_test.go
+++ b/services/syncbase/vsync/dag_test.go
@@ -335,7 +335,7 @@
oid, isConflict, newHead, oldHead, ancestor, errConflict)
}
- if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:11:3" {
+ if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:data:11:3" {
t.Errorf("invalid logrec for newhead object %s:%s: %v", oid, newHead, logrec)
}
@@ -411,10 +411,10 @@
oid, isConflict, newHead, oldHead, ancestor, errConflict)
}
- if logrec, err := getLogRecKey(nil, st, oid, oldHead); err != nil || logrec != "$sync:log:10:3" {
+ if logrec, err := getLogRecKey(nil, st, oid, oldHead); err != nil || logrec != "$sync:log:data:10:3" {
t.Errorf("invalid logrec for oldhead object %s:%s: %v", oid, oldHead, logrec)
}
- if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:11:3" {
+ if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:data:11:3" {
t.Errorf("invalid logrec for newhead object %s:%s: %v", oid, newHead, logrec)
}
@@ -494,13 +494,13 @@
oid, isConflict, newHead, oldHead, ancestor, errConflict)
}
- if logrec, err := getLogRecKey(nil, st, oid, oldHead); err != nil || logrec != "$sync:log:10:3" {
+ if logrec, err := getLogRecKey(nil, st, oid, oldHead); err != nil || logrec != "$sync:log:data:10:3" {
t.Errorf("invalid logrec for oldhead object %s:%s: %s", oid, oldHead, logrec)
}
- if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:11:3" {
+ if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:data:11:3" {
t.Errorf("invalid logrec for newhead object %s:%s: %s", oid, newHead, logrec)
}
- if logrec, err := getLogRecKey(nil, st, oid, ancestor); err != nil || logrec != "$sync:log:10:2" {
+ if logrec, err := getLogRecKey(nil, st, oid, ancestor); err != nil || logrec != "$sync:log:data:10:2" {
t.Errorf("invalid logrec for ancestor object %s:%s: %s", oid, ancestor, logrec)
}
@@ -587,13 +587,13 @@
oid, isConflict, newHead, oldHead, ancestor, errConflict)
}
- if logrec, err := getLogRecKey(nil, st, oid, oldHead); err != nil || logrec != "$sync:log:10:3" {
+ if logrec, err := getLogRecKey(nil, st, oid, oldHead); err != nil || logrec != "$sync:log:data:10:3" {
t.Errorf("invalid logrec for oldhead object %s:%s: %s", oid, oldHead, logrec)
}
- if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:11:2" {
+ if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:data:11:2" {
t.Errorf("invalid logrec for newhead object %s:%s: %s", oid, newHead, logrec)
}
- if logrec, err := getLogRecKey(nil, st, oid, ancestor); err != nil || logrec != "$sync:log:10:2" {
+ if logrec, err := getLogRecKey(nil, st, oid, ancestor); err != nil || logrec != "$sync:log:data:10:2" {
t.Errorf("invalid logrec for ancestor object %s:%s: %s", oid, ancestor, logrec)
}
@@ -674,10 +674,10 @@
oid, isConflict, newHead, oldHead, ancestor, errConflict)
}
- if logrec, err := getLogRecKey(nil, st, oid, oldHead); err != nil || logrec != "$sync:log:10:3" {
+ if logrec, err := getLogRecKey(nil, st, oid, oldHead); err != nil || logrec != "$sync:log:data:10:3" {
t.Errorf("invalid logrec for oldhead object %s:%s: %s", oid, oldHead, logrec)
}
- if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:11:3" {
+ if logrec, err := getLogRecKey(nil, st, oid, newHead); err != nil || logrec != "$sync:log:data:11:3" {
t.Errorf("invalid logrec for newhead object %s:%s: %s", oid, newHead, logrec)
}
diff --git a/services/syncbase/vsync/initiator.go b/services/syncbase/vsync/initiator.go
index 9fa297f..5875e21 100644
--- a/services/syncbase/vsync/initiator.go
+++ b/services/syncbase/vsync/initiator.go
@@ -4,11 +4,10 @@
package vsync
-// Initiator is a goroutine that periodically picks a peer from all the known
-// remote peers, and requests deltas from that peer for all the SyncGroups in
-// common across all apps/databases. It then modifies the sync metadata (DAG and
-// local log records) based on the deltas, detects and resolves conflicts if
-// any, and suitably updates the local Databases.
+// Initiator requests deltas from a chosen peer for all the SyncGroups in common
+// across all apps/databases. It then modifies the sync metadata (DAG and local
+// log records) based on the deltas, detects and resolves conflicts if any, and
+// suitably updates the local Databases.
import (
"sort"
@@ -21,6 +20,7 @@
"v.io/v23/vdl"
"v.io/v23/verror"
"v.io/v23/vom"
+ "v.io/x/lib/set"
"v.io/x/lib/vlog"
"v.io/x/ref/services/syncbase/server/interfaces"
"v.io/x/ref/services/syncbase/server/util"
@@ -28,204 +28,167 @@
"v.io/x/ref/services/syncbase/store"
)
-// Policies to pick a peer to sync with.
-const (
- // Picks a peer at random from the available set.
- selectRandom = iota
-
- // TODO(hpucha): implement other policies.
- // Picks a peer with most differing generations.
- selectMostDiff
-
- // Picks a peer that was synced with the furthest in the past.
- selectOldest
-)
-
-var (
- // peerSyncInterval is the duration between two consecutive peer
- // contacts. During every peer contact, the initiator obtains any
- // pending updates from that peer.
- peerSyncInterval = 50 * time.Millisecond
-
- // peerSelectionPolicy is the policy used to select a peer when
- // the initiator gets a chance to sync.
- peerSelectionPolicy = selectRandom
-)
-
-// syncer wakes up every peerSyncInterval to do work: (1) Act as an initiator
-// for SyncGroup metadata by selecting a SyncGroup Admin, and syncing Syncgroup
-// metadata with it (getting updates from the remote peer, detecting and
-// resolving conflicts); (2) Refresh memberView if needed and act as an
-// initiator for data by selecting a peer, and syncing data corresponding to all
-// common SyncGroups across all Databases; (3) Act as a SyncGroup publisher to
-// publish pending SyncGroups; (4) Garbage collect older generations.
-//
-// TODO(hpucha): Currently only does initiation. Add rest.
-func (s *syncService) syncer(ctx *context.T) {
- defer s.pending.Done()
-
- ticker := time.NewTicker(peerSyncInterval)
- defer ticker.Stop()
-
- for {
- select {
- case <-s.closed:
- vlog.VI(1).Info("sync: syncer: channel closed, stop work and exit")
- return
-
- case <-ticker.C:
- }
-
- // TODO(hpucha): Cut a gen for the responder even if there is no
- // one to initiate to?
-
- // Do work.
- peer, err := s.pickPeer(ctx)
- if err != nil {
- continue
- }
- s.getDeltasFromPeer(ctx, peer)
- }
-}
-
-// getDeltasFromPeer performs an initiation round to the specified
-// peer. An initiation round consists of:
+// getDeltas performs an initiation round to the specified peer. An
+// initiation round consists of two sync rounds:
+// * Sync SyncGroup metadata.
+// * Sync data.
+// Each sync round involves:
// * Contacting the peer to receive all the deltas based on the local genvector.
// * Processing those deltas to discover objects which have been updated.
// * Processing updated objects to detect and resolve any conflicts if needed.
-// * Communicating relevant object updates to the Database, and updating local
-// genvector to catch up to the received remote genvector.
+// * Communicating relevant object updates to the Database in case of data.
+// * Updating local genvector to catch up to the received remote genvector.
//
-// The processing of the deltas is done one Database at a time. If a local error
-// is encountered during the processing of a Database, that Database is skipped
-// and the initiator continues on to the next one. If the connection to the peer
+// The processing of the deltas is done one Database at a time, encompassing all
+// the SyncGroups common to the initiator and the responder. If a local error is
+// encountered during the processing of a Database, that Database is skipped and
+// the initiator continues on to the next one. If the connection to the peer
// encounters an error, this initiation round is aborted. Note that until the
// local genvector is updated based on the received deltas (the last step in an
// initiation round), the work done by the initiator is idempotent.
//
// TODO(hpucha): Check the idempotence, esp in addNode in DAG.
-func (s *syncService) getDeltasFromPeer(ctxIn *context.T, peer string) {
- vlog.VI(2).Infof("sync: getDeltasFromPeer: begin: contacting peer %s", peer)
- defer vlog.VI(2).Infof("sync: getDeltasFromPeer: end: contacting peer %s", peer)
-
- ctx, cancel := context.WithRootCancel(ctxIn)
+func (s *syncService) getDeltas(ctx *context.T, peer string) {
+ vlog.VI(2).Infof("sync: getDeltas: begin: contacting peer %s", peer)
+ defer vlog.VI(2).Infof("sync: getDeltas: end: contacting peer %s", peer)
info := s.copyMemberInfo(ctx, peer)
if info == nil {
- vlog.Fatalf("sync: getDeltasFromPeer: missing information in member view for %q", peer)
+ vlog.Fatalf("sync: getDeltas: missing information in member view for %q", peer)
}
- connected := false
- var stream interfaces.SyncGetDeltasClientCall
+
+ // Preferred mount tables for this peer.
+ prfMtTbls := set.String.ToSlice(info.mtTables)
// Sync each Database that may have SyncGroups common with this peer,
// one at a time.
- for gdbName, sgInfo := range info.db2sg {
+ for gdbName := range info.db2sg {
+ vlog.VI(4).Infof("sync: getDeltas: started for peer %s db %s", peer, gdbName)
- // Initialize initiation state for syncing this Database.
- iSt, err := newInitiationState(ctx, s, peer, gdbName, sgInfo)
- if err != nil {
- vlog.Errorf("sync: getDeltasFromPeer: couldn't initialize initiator state for peer %s, gdb %s, err %v", peer, gdbName, err)
- continue
- }
-
- if len(iSt.sgIds) == 0 || len(iSt.sgPfxs) == 0 {
- vlog.Errorf("sync: getDeltasFromPeer: didn't find any SyncGroups for peer %s, gdb %s, err %v", peer, gdbName, err)
- continue
- }
-
- // Make contact with the peer once.
- if !connected {
- stream, connected = iSt.connectToPeer(ctx)
- if !connected {
- // Try a different Database. Perhaps there are
- // different mount tables.
- continue
- }
- }
-
- // Create local genvec so that it contains knowledge only about common prefixes.
- if err := iSt.createLocalGenVec(ctx); err != nil {
- vlog.Errorf("sync: getDeltasFromPeer: error creating local genvec for gdb %s, err %v", gdbName, err)
- continue
- }
-
- iSt.stream = stream
- req := interfaces.DeltaReq{
- AppName: iSt.appName,
- DbName: iSt.dbName,
- SgIds: iSt.sgIds,
- InitVec: iSt.local,
- }
-
- vlog.VI(3).Infof("sync: getDeltasFromPeer: send request: %v", req)
- sender := iSt.stream.SendStream()
- sender.Send(req)
-
- // Obtain deltas from the peer over the network.
- if err := iSt.recvAndProcessDeltas(ctx); err != nil {
- vlog.Errorf("sync: getDeltasFromPeer: error receiving deltas for gdb %s, err %v", gdbName, err)
- // Returning here since something could be wrong with
- // the connection, and no point in attempting the next
- // Database.
- cancel()
- stream.Finish()
+ if len(prfMtTbls) < 1 {
+ vlog.Errorf("sync: getDeltas: no mount tables found to connect to peer %s", peer)
return
}
- vlog.VI(3).Infof("sync: getDeltasFromPeer: got reply: %v", iSt.remote)
- if err := iSt.processUpdatedObjects(ctx); err != nil {
- vlog.Errorf("sync: getDeltasFromPeer: error processing objects for gdb %s, err %v", gdbName, err)
- // Move to the next Database even if processing updates
- // failed.
+ c, err := newInitiationConfig(ctx, s, peer, gdbName, info, prfMtTbls)
+ if err != nil {
+ vlog.Errorf("sync: getDeltas: couldn't initialize initiator config for peer %s, gdb %s, err %v", peer, gdbName, err)
continue
}
+
+ if err := s.getDBDeltas(ctx, peer, c, true); err == nil {
+ if err := s.getDBDeltas(ctx, peer, c, false); err != nil {
+ vlog.Errorf("sync: getDeltas: failed for data sync, err %v", err)
+ }
+ } else {
+ // If SyncGroup sync fails, abort data sync as well.
+ vlog.Errorf("sync: getDeltas: failed for SyncGroup sync, err %v", err)
+ }
+
+ // Cache the pruned mount table list for the next Database.
+ prfMtTbls = c.mtTables
+
+ vlog.VI(4).Infof("sync: getDeltas: done for peer %s db %s", peer, gdbName)
+ }
+}
+
+// getDBDeltas gets the deltas from the chosen peer. If sg flag is set to true,
+// it will sync SyncGroup metadata. If sg flag is false, it will sync data.
+func (s *syncService) getDBDeltas(ctxIn *context.T, peer string, c *initiationConfig, sg bool) error {
+ vlog.VI(2).Infof("sync: getDBDeltas: begin: contacting peer sg %v %s", sg, peer)
+ defer vlog.VI(2).Infof("sync: getDBDeltas: end: contacting peer sg %v %s", sg, peer)
+
+ ctx, cancel := context.WithRootCancel(ctxIn)
+ // cancel() is idempotent.
+ defer cancel()
+
+ // Initialize initiation state for syncing this Database.
+ iSt := newInitiationState(ctx, c, sg)
+
+ // Initialize SyncGroup prefixes for data syncing.
+ if !sg {
+ iSt.peerSgInfo(ctx)
+ if len(iSt.config.sgPfxs) == 0 {
+ return verror.New(verror.ErrInternal, ctx, "no SyncGroup prefixes found", peer, iSt.config.appName, iSt.config.dbName)
+ }
}
- if connected {
- stream.Finish()
+ if sg {
+ // Create local genvec so that it contains knowledge about
+ // common SyncGroups and then send the SyncGroup metadata sync
+ // request.
+ if err := iSt.prepareSGDeltaReq(ctx); err != nil {
+ return err
+ }
+ } else {
+ // Create local genvec so that it contains knowledge only about common
+ // prefixes and then send the data sync request.
+ if err := iSt.prepareDataDeltaReq(ctx); err != nil {
+ return err
+ }
}
- cancel()
+
+ // Make contact with the peer.
+ if !iSt.connectToPeer(ctx) {
+ return verror.New(verror.ErrInternal, ctx, "couldn't connect to peer", peer)
+ }
+
+ // Obtain deltas from the peer over the network.
+ if err := iSt.recvAndProcessDeltas(ctx); err != nil {
+ cancel()
+ iSt.stream.Finish()
+ return err
+ }
+
+ vlog.VI(4).Infof("sync: getDBDeltas: got reply: %v", iSt.remote)
+
+ if err := iSt.processUpdatedObjects(ctx); err != nil {
+ return err
+ }
+
+ return iSt.stream.Finish()
}
type sgSet map[interfaces.GroupId]struct{}
-// initiationState is accumulated for each Database during an initiation round.
-type initiationState struct {
- // Relative name of the peer to sync with.
- peer string
+// initiationConfig is the configuration information for a Database in an
+// initiation round.
+type initiationConfig struct {
+ peer string // relative name of the peer to sync with.
- // Collection of mount tables that this peer may have registered with.
- mtTables map[string]struct{}
+ // Mount tables that this peer may have registered with. The first entry
+ // in this array is the mount table where the peer was successfully
+ // reached the last time.
+ mtTables []string
- // SyncGroups being requested in the initiation round.
- sgIds sgSet
-
- // SyncGroup prefixes being requested in the initiation round, and their
- // corresponding SyncGroup ids.
- sgPfxs map[string]sgSet
-
- // Local generation vector.
- local interfaces.GenVector
-
- // Generation vector from the remote peer.
- remote interfaces.GenVector
-
- // Updated local generation vector at the end of the initiation round.
- updLocal interfaces.GenVector
-
- // State to track updated objects during a log replay.
- updObjects map[string]*objConflictState
-
- // DAG state that tracks conflicts and common ancestors.
- dagGraft graftMap
-
+ sgIds sgSet // SyncGroups being requested in the initiation round.
+ sgPfxs map[string]sgSet // SyncGroup prefixes and their ids being requested in the initiation round.
sync *syncService
appName string
dbName string
- st store.Store // Store handle to the Database.
- stream interfaces.SyncGetDeltasClientCall // Stream handle for the GetDeltas RPC.
+ st store.Store // Store handle to the Database.
+}
- // Transaction handle for the initiation round. Used during the update
+// initiationState is accumulated for a Database in each sync round in an
+// initiation round.
+type initiationState struct {
+ // Config information.
+ config *initiationConfig
+
+ // Accumulated sync state.
+ local interfaces.GenVector // local generation vector.
+ remote interfaces.GenVector // generation vector from the remote peer.
+ updLocal interfaces.GenVector // updated local generation vector at the end of sync round.
+ updObjects map[string]*objConflictState // tracks updated objects during a log replay.
+ dagGraft graftMap // DAG state that tracks conflicts and common ancestors.
+
+ req interfaces.DeltaReq // GetDeltas RPC request.
+ stream interfaces.SyncGetDeltasClientCall // stream handle for the GetDeltas RPC.
+
+ // Flag to indicate if this is SyncGroup metadata sync.
+ sg bool
+
+ // Transaction handle for the sync round. Used during the update
// of objects in the Database.
tx store.Transaction
}
@@ -240,93 +203,90 @@
res *conflictResolution
}
-// newInitiationState creates new initiation state.
-func newInitiationState(ctx *context.T, s *syncService, peer string, name string, sgInfo sgMemberInfo) (*initiationState, error) {
- iSt := &initiationState{}
- iSt.peer = peer
- iSt.updObjects = make(map[string]*objConflictState)
- iSt.dagGraft = newGraft()
- iSt.sync = s
+// newInitiatonConfig creates new initiation config. This will be shared between
+// the two sync rounds in the initiation round of a Database.
+func newInitiationConfig(ctx *context.T, s *syncService, peer string, name string, info *memberInfo, mtTables []string) (*initiationConfig, error) {
+ c := &initiationConfig{}
+ c.peer = peer
+ c.mtTables = mtTables
+ c.sgIds = make(sgSet)
+ for id := range info.db2sg[name] {
+ c.sgIds[id] = struct{}{}
+ }
+ if len(c.sgIds) == 0 {
+ return nil, verror.New(verror.ErrInternal, ctx, "no SyncGroups found", peer, name)
+ }
+ // Note: sgPfxs will be inited when needed by the data sync.
+
+ c.sync = s
// TODO(hpucha): Would be nice to standardize on the combined "app:db"
// name across sync (not syncbase) so we only join split/join them at
// the boundary with the store part.
var err error
- iSt.appName, iSt.dbName, err = splitAppDbName(ctx, name)
+ c.appName, c.dbName, err = splitAppDbName(ctx, name)
if err != nil {
return nil, err
}
// TODO(hpucha): nil rpc.ServerCall ok?
- iSt.st, err = s.getDbStore(ctx, nil, iSt.appName, iSt.dbName)
+ c.st, err = s.getDbStore(ctx, nil, c.appName, c.dbName)
if err != nil {
return nil, err
}
- iSt.peerMtTblsAndSgInfo(ctx, peer, sgInfo)
-
- return iSt, nil
+ return c, nil
}
-// peerMtTblsAndSgInfo computes the possible mount tables, the SyncGroup Ids and
-// prefixes common with a remote peer in a particular Database by consulting the
-// SyncGroups in the specified Database.
-func (iSt *initiationState) peerMtTblsAndSgInfo(ctx *context.T, peer string, info sgMemberInfo) {
- iSt.mtTables = make(map[string]struct{})
- iSt.sgIds = make(sgSet)
- iSt.sgPfxs = make(map[string]sgSet)
+// newInitiationState creates new initiation state.
+func newInitiationState(ctx *context.T, c *initiationConfig, sg bool) *initiationState {
+ iSt := &initiationState{}
+ iSt.config = c
+ iSt.updObjects = make(map[string]*objConflictState)
+ iSt.dagGraft = newGraft()
+ iSt.sg = sg
+ return iSt
+}
- for id := range info {
- sg, err := getSyncGroupById(ctx, iSt.st, id)
+// peerSgInfo computes the SyncGroup Ids and prefixes common with a remote peer
+// in a particular Database by consulting the SyncGroups in the specified
+// Database.
+func (iSt *initiationState) peerSgInfo(ctx *context.T) {
+ sgs := iSt.config.sgIds
+ iSt.config.sgIds = make(sgSet) // regenerate the sgids since we are going through the SyncGroups in any case.
+ iSt.config.sgPfxs = make(map[string]sgSet)
+
+ for id := range sgs {
+ sg, err := getSyncGroupById(ctx, iSt.config.st, id)
if err != nil {
continue
}
- if _, ok := sg.Joiners[peer]; !ok {
+ if _, ok := sg.Joiners[iSt.config.peer]; !ok {
// Peer is no longer part of the SyncGroup.
continue
}
- for _, mt := range sg.Spec.MountTables {
- iSt.mtTables[mt] = struct{}{}
- }
- iSt.sgIds[id] = struct{}{}
+
+ iSt.config.sgIds[id] = struct{}{}
for _, p := range sg.Spec.Prefixes {
- sgs, ok := iSt.sgPfxs[p]
+ sgs, ok := iSt.config.sgPfxs[p]
if !ok {
sgs = make(sgSet)
- iSt.sgPfxs[p] = sgs
+ iSt.config.sgPfxs[p] = sgs
}
sgs[id] = struct{}{}
}
}
}
-// connectToPeer attempts to connect to the remote peer using the mount tables
-// obtained from the SyncGroups being synced in the current Database.
-func (iSt *initiationState) connectToPeer(ctx *context.T) (interfaces.SyncGetDeltasClientCall, bool) {
- if len(iSt.mtTables) < 1 {
- vlog.Errorf("sync: connectToPeer: no mount tables found to connect to peer %s, app %s db %s", iSt.peer, iSt.appName, iSt.dbName)
- return nil, false
- }
- for mt := range iSt.mtTables {
- absName := naming.Join(mt, iSt.peer, util.SyncbaseSuffix)
- c := interfaces.SyncClient(absName)
- stream, err := c.GetDeltas(ctx, iSt.sync.name)
- if err == nil {
- vlog.VI(3).Infof("sync: connectToPeer: established on %s", absName)
- return stream, true
- }
- }
- return nil, false
-}
-
-// createLocalGenVec creates the generation vector with local knowledge for the
-// initiator to send to the responder.
+// prepareDataDeltaReq creates the generation vector with local knowledge for the
+// initiator to send to the responder, and creates the request to start the data
+// sync.
//
// TODO(hpucha): Refactor this code with computeDelta code in sync_state.go.
-func (iSt *initiationState) createLocalGenVec(ctx *context.T) error {
- iSt.sync.thLock.Lock()
- defer iSt.sync.thLock.Unlock()
+func (iSt *initiationState) prepareDataDeltaReq(ctx *context.T) error {
+ iSt.config.sync.thLock.Lock()
+ defer iSt.config.sync.thLock.Unlock()
// Freeze the most recent batch of local changes before fetching
// remote changes from a peer. This frozen state is used by the
@@ -341,19 +301,20 @@
// devices. These remote devices in turn can send these
// generations back to the initiator in progress which was
// started with older generation information.
- if err := iSt.sync.checkptLocalGen(ctx, iSt.appName, iSt.dbName); err != nil {
+ if err := iSt.config.sync.checkptLocalGen(ctx, iSt.config.appName, iSt.config.dbName, nil); err != nil {
return err
}
- local, lgen, err := iSt.sync.copyDbGenInfo(ctx, iSt.appName, iSt.dbName)
+ local, lgen, err := iSt.config.sync.copyDbGenInfo(ctx, iSt.config.appName, iSt.config.dbName, nil)
if err != nil {
return err
}
+
localPfxs := extractAndSortPrefixes(local)
- sgPfxs := make([]string, len(iSt.sgPfxs))
+ sgPfxs := make([]string, len(iSt.config.sgPfxs))
i := 0
- for p := range iSt.sgPfxs {
+ for p := range iSt.config.sgPfxs {
sgPfxs[i] = p
i++
}
@@ -390,14 +351,84 @@
if lpStart == "" {
// No matching prefixes for pfx were found.
iSt.local[pfx] = make(interfaces.PrefixGenVector)
- iSt.local[pfx][iSt.sync.id] = lgen
+ iSt.local[pfx][iSt.config.sync.id] = lgen
} else {
iSt.local[pfx] = local[lpStart]
}
}
+
+ // Send request.
+ req := interfaces.DataDeltaReq{
+ AppName: iSt.config.appName,
+ DbName: iSt.config.dbName,
+ SgIds: iSt.config.sgIds,
+ InitVec: iSt.local,
+ }
+
+ iSt.req = interfaces.DeltaReqData{req}
+
+ vlog.VI(4).Infof("sync: prepareDataDeltaReq: request: %v", req)
+
return nil
}
+// prepareSGDeltaReq creates the SyncGroup generation vector with local knowledge
+// for the initiator to send to the responder, and prepares the request to start
+// the SyncGroup sync.
+func (iSt *initiationState) prepareSGDeltaReq(ctx *context.T) error {
+ iSt.config.sync.thLock.Lock()
+ defer iSt.config.sync.thLock.Unlock()
+
+ if err := iSt.config.sync.checkptLocalGen(ctx, iSt.config.appName, iSt.config.dbName, iSt.config.sgIds); err != nil {
+ return err
+ }
+
+ var err error
+ iSt.local, _, err = iSt.config.sync.copyDbGenInfo(ctx, iSt.config.appName, iSt.config.dbName, iSt.config.sgIds)
+ if err != nil {
+ return err
+ }
+
+ // Send request.
+ req := interfaces.SgDeltaReq{
+ AppName: iSt.config.appName,
+ DbName: iSt.config.dbName,
+ InitVec: iSt.local,
+ }
+
+ iSt.req = interfaces.DeltaReqSgs{req}
+
+ vlog.VI(4).Infof("sync: prepareSGDeltaReq: request: %v", req)
+
+ return nil
+}
+
+// connectToPeer attempts to connect to the remote peer using the mount tables
+// obtained from all the common SyncGroups.
+func (iSt *initiationState) connectToPeer(ctx *context.T) bool {
+ if len(iSt.config.mtTables) < 1 {
+ vlog.Errorf("sync: connectToPeer: no mount tables found to connect to peer %s, app %s db %s", iSt.config.peer, iSt.config.appName, iSt.config.dbName)
+ return false
+ }
+
+ for i, mt := range iSt.config.mtTables {
+ absName := naming.Join(mt, iSt.config.peer, util.SyncbaseSuffix)
+ c := interfaces.SyncClient(absName)
+ var err error
+ iSt.stream, err = c.GetDeltas(ctx, iSt.req, iSt.config.sync.name)
+ if err == nil {
+ vlog.VI(4).Infof("sync: connectToPeer: established on %s", absName)
+
+ // Prune out the unsuccessful mount tables.
+ iSt.config.mtTables = iSt.config.mtTables[i:]
+ return true
+ }
+ }
+ iSt.config.mtTables = nil
+ vlog.Errorf("sync: connectToPeer: couldn't connect to peer %s", iSt.config.peer)
+ return false
+}
+
// recvAndProcessDeltas first receives the log records and generation vector
// from the GetDeltas RPC and puts them in the Database. It also replays the
// entire log stream as the log records arrive. These records span multiple
@@ -405,20 +436,28 @@
// resolution during replay. This avoids resolving conflicts that have already
// been resolved by other devices.
func (iSt *initiationState) recvAndProcessDeltas(ctx *context.T) error {
- iSt.sync.thLock.Lock()
- defer iSt.sync.thLock.Unlock()
+ // This is to handle issues with graftMap in DAG. Ideally, the
+ // transaction created to store all the deltas received over the network
+ // should not contend with any other store changes since this is all
+ // brand new information. However, as log records are received over the
+ // network, they are also incrementally processed. To enable incremental
+ // processing, the current head of each dirty object is read to populate
+ // the graftMap. This read can potentially contend with the watcher
+ // updating the head of an object. This lock prevents that contention in
+ // order to avoid retrying the whole transaction.
+ iSt.config.sync.thLock.Lock()
+ defer iSt.config.sync.thLock.Unlock()
// TODO(hpucha): This works for now, but figure out a long term solution
// as this may be implementation dependent. It currently works because
// the RecvStream call is stateless, and grabbing a handle to it
// repeatedly doesn't affect what data is seen next.
rcvr := iSt.stream.RecvStream()
- start, finish := false, false
// TODO(hpucha): See if we can avoid committing the entire delta stream
// as one batch. Currently the dependency is between the log records and
// the batch info.
- tx := iSt.st.NewTransaction()
+ tx := iSt.config.st.NewTransaction()
committed := false
defer func() {
@@ -433,18 +472,6 @@
for rcvr.Advance() {
resp := rcvr.Value()
switch v := resp.(type) {
- case interfaces.DeltaRespStart:
- if start {
- return verror.New(verror.ErrInternal, ctx, "received start followed by start in delta response stream")
- }
- start = true
-
- case interfaces.DeltaRespFinish:
- if finish {
- return verror.New(verror.ErrInternal, ctx, "received finish followed by finish in delta response stream")
- }
- finish = true
-
case interfaces.DeltaRespRespVec:
iSt.remote = v.Value
@@ -452,12 +479,18 @@
// Insert log record in Database.
// TODO(hpucha): Should we reserve more positions in a batch?
// TODO(hpucha): Handle if SyncGroup is left/destroyed while sync is in progress.
- pos := iSt.sync.reservePosInDbLog(ctx, iSt.appName, iSt.dbName, 1)
+ var pos uint64
+ if iSt.sg {
+ pos = iSt.config.sync.reservePosInDbLog(ctx, iSt.config.appName, iSt.config.dbName, v.Value.Metadata.ObjId, 1)
+ } else {
+ pos = iSt.config.sync.reservePosInDbLog(ctx, iSt.config.appName, iSt.config.dbName, "", 1)
+ }
+
rec := &localLogRec{Metadata: v.Value.Metadata, Pos: pos}
batchId := rec.Metadata.BatchId
if batchId != NoBatchId {
if cnt, ok := batchMap[batchId]; !ok {
- if iSt.sync.startBatch(ctx, tx, batchId) != batchId {
+ if iSt.config.sync.startBatch(ctx, tx, batchId) != batchId {
return verror.New(verror.ErrInternal, ctx, "failed to create batch info")
}
batchMap[batchId] = rec.Metadata.BatchCount
@@ -467,27 +500,25 @@
}
vlog.VI(4).Infof("sync: recvAndProcessDeltas: processing rec %v", rec)
- if err := iSt.insertRecInLogDagAndDb(ctx, rec, batchId, v.Value.Value, tx); err != nil {
+ if err := iSt.insertRecInLogAndDag(ctx, rec, batchId, tx); err != nil {
return err
}
- // Check for BlobRefs, and process them.
- if err := iSt.processBlobRefs(ctx, &rec.Metadata, v.Value.Value); err != nil {
- return err
+ if iSt.sg {
+ // Add the SyncGroup value to the Database.
+ } else {
+ if err := iSt.insertRecInDb(ctx, rec, v.Value.Value, tx); err != nil {
+ return err
+ }
+ // Check for BlobRefs, and process them.
+ if err := iSt.processBlobRefs(ctx, &rec.Metadata, v.Value.Value); err != nil {
+ return err
+ }
}
// Mark object dirty.
iSt.updObjects[rec.Metadata.ObjId] = &objConflictState{}
}
-
- // Break out of the stream.
- if finish {
- break
- }
- }
-
- if !(start && finish) {
- return verror.New(verror.ErrInternal, ctx, "didn't receive start/finish delimiters in delta response stream")
}
if err := rcvr.Err(); err != nil {
@@ -496,7 +527,7 @@
// End the started batches if any.
for bid, cnt := range batchMap {
- if err := iSt.sync.endBatch(ctx, tx, bid, cnt); err != nil {
+ if err := iSt.config.sync.endBatch(ctx, tx, bid, cnt); err != nil {
return err
}
}
@@ -516,6 +547,48 @@
return err
}
+// insertRecInLogAndDag adds a new log record to log and dag data structures.
+func (iSt *initiationState) insertRecInLogAndDag(ctx *context.T, rec *localLogRec, batchId uint64, tx store.Transaction) error {
+ var pfx string
+ m := rec.Metadata
+
+ if iSt.sg {
+ pfx = m.ObjId
+ } else {
+ pfx = logDataPrefix
+ }
+
+ if err := putLogRec(ctx, tx, pfx, rec); err != nil {
+ return err
+ }
+ logKey := logRecKey(pfx, m.Id, m.Gen)
+
+ var err error
+ switch m.RecType {
+ case interfaces.NodeRec:
+ err = iSt.config.sync.addNode(ctx, tx, m.ObjId, m.CurVers, logKey, m.Delete, m.Parents, m.BatchId, iSt.dagGraft)
+ case interfaces.LinkRec:
+ err = iSt.config.sync.addParent(ctx, tx, m.ObjId, m.CurVers, m.Parents[0], iSt.dagGraft)
+ default:
+ err = verror.New(verror.ErrInternal, ctx, "unknown log record type")
+ }
+
+ return err
+}
+
+// insertRecInDb inserts the versioned value in the Database.
+func (iSt *initiationState) insertRecInDb(ctx *context.T, rec *localLogRec, valbuf []byte, tx store.Transaction) error {
+ m := rec.Metadata
+ // TODO(hpucha): Hack right now. Need to change Database's handling of
+ // deleted objects. Currently, the initiator needs to treat deletions
+ // specially since deletions do not get a version number or a special
+ // value in the Database.
+ if !rec.Metadata.Delete && rec.Metadata.RecType == interfaces.NodeRec {
+ return watchable.PutAtVersion(ctx, tx, []byte(m.ObjId), valbuf, []byte(m.CurVers))
+ }
+ return nil
+}
+
func (iSt *initiationState) processBlobRefs(ctx *context.T, m *interfaces.LogRecMetadata, valbuf []byte) error {
objid := m.ObjId
srcPeer := syncbaseIdToName(m.Id)
@@ -538,16 +611,16 @@
}
sgIds := make(sgSet)
for br := range brs {
- for p, sgs := range iSt.sgPfxs {
+ for p, sgs := range iSt.config.sgPfxs {
if strings.HasPrefix(extractAppKey(objid), p) {
for sg := range sgs {
sgIds[sg] = struct{}{}
}
}
}
- vlog.VI(4).Infof("sync: processBlobRefs: Found blobref %v peer %v, source %v, sgs %v", br, iSt.peer, srcPeer, sgIds)
- info := &blobLocInfo{peer: iSt.peer, source: srcPeer, sgIds: sgIds}
- if err := iSt.sync.addBlobLocInfo(ctx, br, info); err != nil {
+ vlog.VI(4).Infof("sync: processBlobRefs: Found blobref %v peer %v, source %v, sgs %v", br, iSt.config.peer, srcPeer, sgIds)
+ info := &blobLocInfo{peer: iSt.config.peer, source: srcPeer, sgIds: sgIds}
+ if err := iSt.config.sync.addBlobLocInfo(ctx, br, info); err != nil {
return err
}
}
@@ -578,39 +651,6 @@
return nil
}
-// insertRecInLogDagAndDb adds a new log record to log and dag data structures,
-// and inserts the versioned value in the Database.
-func (iSt *initiationState) insertRecInLogDagAndDb(ctx *context.T, rec *localLogRec, batchId uint64, valbuf []byte, tx store.Transaction) error {
- if err := putLogRec(ctx, tx, rec); err != nil {
- return err
- }
-
- m := rec.Metadata
- logKey := logRecKey(m.Id, m.Gen)
-
- var err error
- switch m.RecType {
- case interfaces.NodeRec:
- err = iSt.sync.addNode(ctx, tx, m.ObjId, m.CurVers, logKey, m.Delete, m.Parents, m.BatchId, iSt.dagGraft)
- case interfaces.LinkRec:
- err = iSt.sync.addParent(ctx, tx, m.ObjId, m.CurVers, m.Parents[0], iSt.dagGraft)
- default:
- err = verror.New(verror.ErrInternal, ctx, "unknown log record type")
- }
-
- if err != nil {
- return err
- }
- // TODO(hpucha): Hack right now. Need to change Database's handling of
- // deleted objects. Currently, the initiator needs to treat deletions
- // specially since deletions do not get a version number or a special
- // value in the Database.
- if !rec.Metadata.Delete && rec.Metadata.RecType == interfaces.NodeRec {
- return watchable.PutAtVersion(ctx, tx, []byte(m.ObjId), valbuf, []byte(m.CurVers))
- }
- return nil
-}
-
// processUpdatedObjects processes all the updates received by the initiator,
// one object at a time. Conflict detection and resolution is carried out after
// the entire delta of log records is replayed, instead of incrementally after
@@ -660,15 +700,15 @@
}()
for {
- vlog.VI(3).Infof("sync: processUpdatedObjects: begin: %d objects updated", len(iSt.updObjects))
+ vlog.VI(4).Infof("sync: processUpdatedObjects: begin: %d objects updated", len(iSt.updObjects))
- iSt.tx = iSt.st.NewTransaction()
+ iSt.tx = iSt.config.st.NewTransaction()
watchable.SetTransactionFromSync(iSt.tx) // for echo-suppression
if count, err := iSt.detectConflicts(ctx); err != nil {
return err
} else {
- vlog.VI(3).Infof("sync: processUpdatedObjects: %d conflicts detected", count)
+ vlog.VI(4).Infof("sync: processUpdatedObjects: %d conflicts detected", count)
}
if err := iSt.resolveConflicts(ctx); err != nil {
@@ -682,10 +722,10 @@
if err == nil {
committed = true
// Update in-memory genvector since commit is successful.
- if err := iSt.sync.putDbGenInfoRemote(ctx, iSt.appName, iSt.dbName, iSt.updLocal); err != nil {
- vlog.Fatalf("sync: processUpdatedObjects: putting geninfo in memory failed for app %s db %s, err %v", iSt.appName, iSt.dbName, err)
+ if err := iSt.config.sync.putDbGenInfoRemote(ctx, iSt.config.appName, iSt.config.dbName, iSt.sg, iSt.updLocal); err != nil {
+ vlog.Fatalf("sync: processUpdatedObjects: putting geninfo in memory failed for app %s db %s, err %v", iSt.config.appName, iSt.config.dbName, err)
}
- vlog.VI(3).Info("sync: processUpdatedObjects: end: changes committed")
+ vlog.VI(4).Info("sync: processUpdatedObjects: end: changes committed")
return nil
}
if verror.ErrorID(err) != store.ErrConcurrentTransaction.ID {
@@ -699,7 +739,7 @@
// solution. Next iteration will have coordination with watch
// thread to intelligently retry. Hence this value is not a
// config param.
- vlog.VI(3).Info("sync: processUpdatedObjects: retry due to local mutations")
+ vlog.VI(4).Info("sync: processUpdatedObjects: retry due to local mutations")
iSt.tx.Abort()
time.Sleep(1 * time.Second)
}
@@ -736,7 +776,7 @@
// If the local version is picked, no further updates to the
// Database are needed. If the remote version is picked or if a
// new version is created, we put it in the Database.
- if confSt.res.ty != pickLocal {
+ if confSt.res.ty != pickLocal && !iSt.sg {
// TODO(hpucha): Hack right now. Need to change Database's
// handling of deleted objects.
@@ -815,10 +855,10 @@
}
// updateLogAndDag updates the log and dag data structures.
-func (iSt *initiationState) updateLogAndDag(ctx *context.T, obj string) error {
- confSt, ok := iSt.updObjects[obj]
+func (iSt *initiationState) updateLogAndDag(ctx *context.T, objid string) error {
+ confSt, ok := iSt.updObjects[objid]
if !ok {
- return verror.New(verror.ErrInternal, ctx, "object state not found", obj)
+ return verror.New(verror.ErrInternal, ctx, "object state not found", objid)
}
var newVersion string
@@ -831,11 +871,11 @@
switch {
case confSt.res.ty == pickLocal:
// Local version was picked as the conflict resolution.
- rec = iSt.createLocalLinkLogRec(ctx, obj, confSt.oldHead, confSt.newHead)
+ rec = iSt.createLocalLinkLogRec(ctx, objid, confSt.oldHead, confSt.newHead)
newVersion = confSt.oldHead
case confSt.res.ty == pickRemote:
// Remote version was picked as the conflict resolution.
- rec = iSt.createLocalLinkLogRec(ctx, obj, confSt.newHead, confSt.oldHead)
+ rec = iSt.createLocalLinkLogRec(ctx, objid, confSt.newHead, confSt.oldHead)
newVersion = confSt.newHead
default:
// New version was created to resolve the conflict.
@@ -843,7 +883,13 @@
newVersion = confSt.res.rec.Metadata.CurVers
}
- if err := putLogRec(ctx, iSt.tx, rec); err != nil {
+ var pfx string
+ if iSt.sg {
+ pfx = objid
+ } else {
+ pfx = logDataPrefix
+ }
+ if err := putLogRec(ctx, iSt.tx, pfx, rec); err != nil {
return err
}
@@ -852,9 +898,9 @@
m := rec.Metadata
switch m.RecType {
case interfaces.NodeRec:
- err = iSt.sync.addNode(ctx, iSt.tx, obj, m.CurVers, logRecKey(m.Id, m.Gen), m.Delete, m.Parents, NoBatchId, nil)
+ err = iSt.config.sync.addNode(ctx, iSt.tx, objid, m.CurVers, logRecKey(pfx, m.Id, m.Gen), m.Delete, m.Parents, NoBatchId, nil)
case interfaces.LinkRec:
- err = iSt.sync.addParent(ctx, iSt.tx, obj, m.CurVers, m.Parents[0], nil)
+ err = iSt.config.sync.addParent(ctx, iSt.tx, objid, m.CurVers, m.Parents[0], nil)
default:
return verror.New(verror.ErrInternal, ctx, "unknown log record type")
}
@@ -865,21 +911,26 @@
// Move the head. This should be idempotent. We may move head to the
// local head in some cases.
- return moveHead(ctx, iSt.tx, obj, newVersion)
+ return moveHead(ctx, iSt.tx, objid, newVersion)
}
-func (iSt *initiationState) createLocalLinkLogRec(ctx *context.T, obj, vers, par string) *localLogRec {
- gen, pos := iSt.sync.reserveGenAndPosInDbLog(ctx, iSt.appName, iSt.dbName, 1)
+func (iSt *initiationState) createLocalLinkLogRec(ctx *context.T, objid, vers, par string) *localLogRec {
+ vlog.VI(4).Infof("sync: createLocalLinkLogRec: obj %s vers %s par %s", objid, vers, par)
- vlog.VI(4).Infof("sync: createLocalLinkLogRec: obj %s vers %s par %s", obj, vers, par)
+ var gen, pos uint64
+ if iSt.sg {
+ gen, pos = iSt.config.sync.reserveGenAndPosInDbLog(ctx, iSt.config.appName, iSt.config.dbName, objid, 1)
+ } else {
+ gen, pos = iSt.config.sync.reserveGenAndPosInDbLog(ctx, iSt.config.appName, iSt.config.dbName, "", 1)
+ }
rec := &localLogRec{
Metadata: interfaces.LogRecMetadata{
- Id: iSt.sync.id,
+ Id: iSt.config.sync.id,
Gen: gen,
RecType: interfaces.LinkRec,
- ObjId: obj,
+ ObjId: objid,
CurVers: vers,
Parents: []string{par},
UpdTime: time.Now().UTC(),
@@ -895,35 +946,53 @@
// updateSyncSt updates local sync state at the end of an initiator cycle.
func (iSt *initiationState) updateSyncSt(ctx *context.T) error {
// Get the current local sync state.
- dsInMem, err := iSt.sync.copyDbSyncStateInMem(ctx, iSt.appName, iSt.dbName)
+ dsInMem, err := iSt.config.sync.copyDbSyncStateInMem(ctx, iSt.config.appName, iSt.config.dbName)
if err != nil {
return err
}
+ // Create the state to be persisted.
ds := &dbSyncState{
- Gen: dsInMem.gen,
- CheckptGen: dsInMem.checkptGen,
- GenVec: dsInMem.genvec,
+ Data: localGenInfo{
+ Gen: dsInMem.data.gen,
+ CheckptGen: dsInMem.data.checkptGen,
+ },
+ Sgs: make(map[interfaces.GroupId]localGenInfo),
+ GenVec: dsInMem.genvec,
+ SgGenVec: dsInMem.sggenvec,
}
+ for id, info := range dsInMem.sgs {
+ ds.Sgs[id] = localGenInfo{
+ Gen: info.gen,
+ CheckptGen: info.checkptGen,
+ }
+ }
+
+ genvec := ds.GenVec
+ if iSt.sg {
+ genvec = ds.SgGenVec
+ }
// remote can be a subset of local.
for rpfx, respgv := range iSt.remote {
- for lpfx, lpgv := range ds.GenVec {
+ for lpfx, lpgv := range genvec {
if strings.HasPrefix(lpfx, rpfx) {
mergePrefixGenVectors(lpgv, respgv)
}
}
- if _, ok := ds.GenVec[rpfx]; !ok {
- ds.GenVec[rpfx] = respgv
+ if _, ok := genvec[rpfx]; !ok {
+ genvec[rpfx] = respgv
}
}
- iSt.updLocal = ds.GenVec
+ iSt.updLocal = genvec
// Clean the genvector of any local state. Note that local state is held
// in gen/ckPtGen in sync state struct.
for _, pgv := range iSt.updLocal {
- delete(pgv, iSt.sync.id)
+ delete(pgv, iSt.config.sync.id)
}
+ // TODO(hpucha): Flip join pending if needed.
+
// TODO(hpucha): Add knowledge compaction.
return putDbSyncState(ctx, iSt.tx, ds)
@@ -938,31 +1007,3 @@
}
}
}
-
-////////////////////////////////////////
-// Peer selection policies.
-
-// pickPeer picks a Syncbase to sync with.
-func (s *syncService) pickPeer(ctx *context.T) (string, error) {
- switch peerSelectionPolicy {
- case selectRandom:
- members := s.getMembers(ctx)
- // Remove myself from the set.
- delete(members, s.name)
- if len(members) == 0 {
- return "", verror.New(verror.ErrInternal, ctx, "no useful peer")
- }
-
- // Pick a peer at random.
- ind := randIntn(len(members))
- for m := range members {
- if ind == 0 {
- return m, nil
- }
- ind--
- }
- return "", verror.New(verror.ErrInternal, ctx, "random selection didn't succeed")
- default:
- return "", verror.New(verror.ErrInternal, ctx, "unknown peer selection policy")
- }
-}
diff --git a/services/syncbase/vsync/initiator_test.go b/services/syncbase/vsync/initiator_test.go
index 01a63ad..935539b 100644
--- a/services/syncbase/vsync/initiator_test.go
+++ b/services/syncbase/vsync/initiator_test.go
@@ -15,12 +15,14 @@
import (
"fmt"
"reflect"
+ "sort"
"testing"
"time"
"v.io/v23/services/syncbase/nosql"
"v.io/v23/vdl"
"v.io/v23/vom"
+ "v.io/x/lib/set"
_ "v.io/x/ref/runtime/factories/generic"
"v.io/x/ref/services/syncbase/server/interfaces"
"v.io/x/ref/services/syncbase/server/util"
@@ -84,7 +86,7 @@
// TestLogStreamRemoteOnly tests processing of a remote log stream. Commands are
// in file testdata/remote-init-00.log.sync.
func TestLogStreamRemoteOnly(t *testing.T) {
- svc, iSt, cleanup := testInit(t, "", "remote-init-00.log.sync")
+ svc, iSt, cleanup := testInit(t, "", "remote-init-00.log.sync", false)
defer cleanup(t, svc)
// Check all log records.
@@ -92,9 +94,9 @@
var gen uint64
var parents []string
for gen = 1; gen < 4; gen++ {
- gotRec, err := getLogRec(nil, svc.St(), 11, gen)
+ gotRec, err := getLogRec(nil, svc.St(), logDataPrefix, 11, gen)
if err != nil || gotRec == nil {
- t.Fatalf("getLogRec can not find object 11 %d, err %v", gen, err)
+ t.Fatalf("getLogRec can not find object %s 11 %d, err %v", logDataPrefix, gen, err)
}
vers := fmt.Sprintf("%d", gen)
wantRec := &localLogRec{
@@ -174,7 +176,7 @@
// correctly applied (when there are no conflicts). Commands are in files
// testdata/<local-init-00.log.sync,remote-noconf-00.log.sync>.
func TestLogStreamNoConflict(t *testing.T) {
- svc, iSt, cleanup := testInit(t, "local-init-00.log.sync", "remote-noconf-00.log.sync")
+ svc, iSt, cleanup := testInit(t, "local-init-00.log.sync", "remote-noconf-00.log.sync", false)
defer cleanup(t, svc)
objid := util.JoinKeyParts(util.RowPrefix, "foo1")
@@ -185,10 +187,10 @@
for _, devid := range []uint64{10, 11} {
var gen uint64
for gen = 1; gen < 4; gen++ {
- gotRec, err := getLogRec(nil, svc.St(), devid, gen)
+ gotRec, err := getLogRec(nil, svc.St(), logDataPrefix, devid, gen)
if err != nil || gotRec == nil {
- t.Fatalf("getLogRec can not find object %d:%d, err %v",
- devid, gen, err)
+ t.Fatalf("getLogRec can not find object %s:%d:%d, err %v",
+ logDataPrefix, devid, gen, err)
}
vers := fmt.Sprintf("%d", version)
wantRec := &localLogRec{
@@ -271,7 +273,7 @@
// correctly applied when there are conflicts. Commands are in files
// testdata/<local-init-00.log.sync,remote-conf-00.log.sync>.
func TestLogStreamConflict(t *testing.T) {
- svc, iSt, cleanup := testInit(t, "local-init-00.log.sync", "remote-conf-00.log.sync")
+ svc, iSt, cleanup := testInit(t, "local-init-00.log.sync", "remote-conf-00.log.sync", false)
defer cleanup(t, svc)
objid := util.JoinKeyParts(util.RowPrefix, "foo1")
@@ -318,7 +320,7 @@
// two versions of an object have no common ancestor. Commands are in files
// testdata/<local-init-00.log.sync,remote-conf-03.log.sync>.
func TestLogStreamConflictNoAncestor(t *testing.T) {
- svc, iSt, cleanup := testInit(t, "local-init-00.log.sync", "remote-conf-03.log.sync")
+ svc, iSt, cleanup := testInit(t, "local-init-00.log.sync", "remote-conf-03.log.sync", false)
defer cleanup(t, svc)
objid := util.JoinKeyParts(util.RowPrefix, "foo1")
@@ -360,10 +362,59 @@
tx.Abort()
}
+// TestSgSync tests that a local and a remote log stream can be correctly
+// applied when there are conflicts on a SyncGroup update. Commands are in files
+// testdata/<local-sginit-00.log.sync,remote-sgconf-00.log.sync>. Note that
+// since data syncing tests are covering different scenarios in detail, a single
+// sniff test is added for SyncGroup sync.
+// TODO(hpucha): Add this test.
+/* func TestSgSync(t *testing.T) {
+ svc, iSt, cleanup := testInit(t, "local-sginit-00.log.sync", "remote-sgconf-00.log.sync", false)
+ defer cleanup(t, svc)
+
+ objid := util.JoinKeyParts(util.RowPrefix, "foo1")
+
+ // Verify conflict state.
+ if len(iSt.updObjects) != 1 {
+ t.Fatalf("Unexpected number of updated objects %d", len(iSt.updObjects))
+ }
+ st := iSt.updObjects[objid]
+ if !st.isConflict {
+ t.Fatalf("Didn't detect a conflict %v", st)
+ }
+ if st.newHead != "6" || st.oldHead != "3" || st.ancestor != "2" {
+ t.Fatalf("Conflict detection didn't succeed %v", st)
+ }
+ if st.res.ty != pickRemote {
+ t.Fatalf("Conflict resolution did not pick remote: %v", st.res.ty)
+ }
+
+ // Verify DAG state.
+ if head, err := getHead(nil, svc.St(), objid); err != nil || head != "6" {
+ t.Fatalf("Invalid object %s head in DAG %v, err %v", objid, head, err)
+ }
+
+ // Verify Database state.
+ valbuf, err := svc.St().Get([]byte(objid), nil)
+ var val string
+ if err := vom.Decode(valbuf, &val); err != nil {
+ t.Fatalf("Value decode failed, err %v", err)
+ }
+ if err != nil || val != "abc" {
+ t.Fatalf("Invalid object %s in Database %v, err %v", objid, string(valbuf), err)
+ }
+ tx := svc.St().NewTransaction()
+ versbuf, err := watchable.GetVersion(nil, tx, []byte(objid))
+ if err != nil || string(versbuf) != "6" {
+ t.Fatalf("Invalid object %s head in Database %v, err %v", objid, string(versbuf), err)
+ }
+ tx.Abort()
+} */
+
//////////////////////////////
// Helpers.
-func testInit(t *testing.T, lfile, rfile string) (*mockService, *initiationState, func(*testing.T, *mockService)) {
+func testInit(t *testing.T, lfile, rfile string, sg bool) (*mockService, *initiationState, func(*testing.T, *mockService)) {
// Set a large value to prevent the initiator from running.
peerSyncInterval = 1 * time.Hour
conflictResolutionPolicy = useTime
@@ -373,10 +424,20 @@
s.id = 10 // initiator
sgId1 := interfaces.GroupId(1234)
+ gdb := appDbName("mockapp", "mockdb")
nullInfo := nosql.SyncGroupMemberInfo{}
sgInfo := sgMemberInfo{
sgId1: nullInfo,
}
+ info := &memberInfo{
+ db2sg: map[string]sgMemberInfo{
+ gdb: sgInfo,
+ },
+ mtTables: map[string]struct{}{
+ "1/2/3/4": struct{}{},
+ "5/6/7/8": struct{}{},
+ },
+ }
sg1 := &interfaces.SyncGroup{
Name: "sg1",
@@ -411,32 +472,52 @@
return svc, nil, cleanup
}
- gdb := appDbName("mockapp", "mockdb")
- iSt, err := newInitiationState(nil, s, "b", gdb, sgInfo)
+ c, err := newInitiationConfig(nil, s, "b", gdb, info, set.String.ToSlice(info.mtTables))
if err != nil {
- t.Fatalf("newInitiationState failed with err %v", err)
+ t.Fatalf("newInitiationConfig failed with err %v", err)
}
- testIfSgPfxsEqual(t, iSt.sgPfxs, sg1.Spec.Prefixes)
- testIfMapArrEqual(t, iSt.mtTables, sg1.Spec.MountTables)
+ iSt := newInitiationState(nil, c, sg)
- s.initDbSyncStateInMem(nil, "mockapp", "mockdb")
-
- // Create local genvec so that it contains knowledge only about common prefixes.
- if err := iSt.createLocalGenVec(nil); err != nil {
- t.Fatalf("createLocalGenVec failed with err %v", err)
+ if !sg {
+ iSt.peerSgInfo(nil)
+ testIfSgPfxsEqual(t, iSt.config.sgPfxs, sg1.Spec.Prefixes)
}
- wantVec := interfaces.GenVector{
- "foo": interfaces.PrefixGenVector{10: 0},
- "bar": interfaces.PrefixGenVector{10: 0},
+ sort.Strings(iSt.config.mtTables)
+ sort.Strings(sg1.Spec.MountTables)
+ if !reflect.DeepEqual(iSt.config.mtTables, sg1.Spec.MountTables) {
+ t.Fatalf("Mount tables are not equal config %v, spec %v", iSt.config.mtTables, sg1.Spec.MountTables)
}
+
+ s.initSyncStateInMem(nil, "mockapp", "mockdb", sgId1)
+
+ iSt.stream = createReplayStream(t, rfile)
+
+ var wantVec interfaces.GenVector
+ if sg {
+ if err := iSt.prepareSGDeltaReq(nil); err != nil {
+ t.Fatalf("prepareSGDeltaReq failed with err %v", err)
+ }
+ sg := fmt.Sprintf("%d", sgId1)
+ wantVec = interfaces.GenVector{
+ sg: interfaces.PrefixGenVector{10: 0},
+ }
+ } else {
+ if err := iSt.prepareDataDeltaReq(nil); err != nil {
+ t.Fatalf("prepareDataDeltaReq failed with err %v", err)
+ }
+
+ wantVec = interfaces.GenVector{
+ "foo": interfaces.PrefixGenVector{10: 0},
+ "bar": interfaces.PrefixGenVector{10: 0},
+ }
+ }
+
if !reflect.DeepEqual(iSt.local, wantVec) {
t.Fatalf("createLocalGenVec failed got %v, want %v", iSt.local, wantVec)
}
- iSt.stream = createReplayStream(t, rfile)
-
if err := iSt.recvAndProcessDeltas(nil); err != nil {
t.Fatalf("recvAndProcessDeltas failed with err %v", err)
}
@@ -461,13 +542,6 @@
}
}
-func testIfMapArrEqual(t *testing.T, m map[string]struct{}, a []string) {
- aMap := arrToMap(a)
- if !reflect.DeepEqual(m, aMap) {
- t.Fatalf("testIfMapArrEqual failed map %v, arr %v", m, aMap)
- }
-}
-
func arrToMap(a []string) map[string]struct{} {
m := make(map[string]struct{})
for _, s := range a {
diff --git a/services/syncbase/vsync/replay_test.go b/services/syncbase/vsync/replay_test.go
index 1a4e2f1..e13eab6 100644
--- a/services/syncbase/vsync/replay_test.go
+++ b/services/syncbase/vsync/replay_test.go
@@ -278,21 +278,6 @@
func (ds *dummyStream) Cancel() {
}
-func (ds *dummyStream) SendStream() interface {
- Send(item interfaces.DeltaReq) error
- Close() error
-} {
- return ds
-}
-
-func (ds *dummyStream) Send(item interfaces.DeltaReq) error {
- return nil
-}
-
-func (ds *dummyStream) Close() error {
- return nil
-}
-
// replayLocalCommands replays local log records parsed from the input file.
func replayLocalCommands(t *testing.T, s *mockService, syncfile string) {
cmds, err := parseSyncCommands(syncfile)
@@ -342,9 +327,6 @@
}
stream := newStream()
- start := interfaces.DeltaRespStart{true}
- stream.add(start)
-
for _, cmd := range cmds {
var ty byte
switch cmd.cmd {
@@ -373,13 +355,11 @@
stream.add(rec)
}
- fin := interfaces.DeltaRespFinish{true}
- stream.add(fin)
return stream
}
func createMetadata(t *testing.T, ty byte, cmd syncCommand) interfaces.LogRecMetadata {
- id, gen, err := splitLogRecKey(nil, cmd.logrec)
+ _, id, gen, err := splitLogRecKey(nil, cmd.logrec)
if err != nil {
t.Fatalf("createReplayStream splitLogRecKey failed, key %s, err %v", cmd.logrec, gen)
}
diff --git a/services/syncbase/vsync/responder.go b/services/syncbase/vsync/responder.go
index 0290b41..46d03a1 100644
--- a/services/syncbase/vsync/responder.go
+++ b/services/syncbase/vsync/responder.go
@@ -6,7 +6,9 @@
import (
"container/heap"
+ "fmt"
"sort"
+ "strconv"
"strings"
"v.io/v23/context"
@@ -19,41 +21,62 @@
)
// GetDeltas implements the responder side of the GetDeltas RPC.
-func (s *syncService) GetDeltas(ctx *context.T, call interfaces.SyncGetDeltasServerCall, initiator string) error {
+func (s *syncService) GetDeltas(ctx *context.T, call interfaces.SyncGetDeltasServerCall, req interfaces.DeltaReq, initiator string) error {
vlog.VI(2).Infof("sync: GetDeltas: begin: from initiator %s", initiator)
defer vlog.VI(2).Infof("sync: GetDeltas: end: from initiator %s", initiator)
- recvr := call.RecvStream()
- for recvr.Advance() {
- req := recvr.Value()
- // Ignoring errors since if one Database fails for any reason,
- // it is fine to continue to the next one. In fact, sometimes
- // the failure might be genuine. For example, the responder is
- // no longer part of the requested SyncGroups, or the app/db is
- // locally deleted, or a permission change has denied access.
- rSt := newResponderState(ctx, call, s, req, initiator)
- rSt.sendDeltasPerDatabase(ctx)
- }
-
- // TODO(hpucha): Is there a need to call finish or some such?
- return recvr.Err()
+ rSt := newResponderState(ctx, call, s, req, initiator)
+ return rSt.sendDeltasPerDatabase(ctx)
}
// responderState is state accumulated per Database by the responder during an
// initiation round.
type responderState struct {
- req interfaces.DeltaReq
+ // Parameters from the request.
+ appName string
+ dbName string
+ sgIds sgSet
+ initVec interfaces.GenVector
+ sg bool
+
call interfaces.SyncGetDeltasServerCall // Stream handle for the GetDeltas RPC.
initiator string
- errState error // Captures the error from the first two phases of the responder.
sync *syncService
st store.Store // Store handle to the Database.
- diff genRangeVector
- outVec interfaces.GenVector
+
+ diff genRangeVector
+ outVec interfaces.GenVector
}
func newResponderState(ctx *context.T, call interfaces.SyncGetDeltasServerCall, sync *syncService, req interfaces.DeltaReq, initiator string) *responderState {
- rSt := &responderState{call: call, sync: sync, req: req, initiator: initiator}
+ rSt := &responderState{
+ call: call,
+ sync: sync,
+ initiator: initiator,
+ }
+
+ switch v := req.(type) {
+ case interfaces.DeltaReqData:
+ rSt.appName = v.Value.AppName
+ rSt.dbName = v.Value.DbName
+ rSt.sgIds = v.Value.SgIds
+ rSt.initVec = v.Value.InitVec
+
+ case interfaces.DeltaReqSgs:
+ rSt.sg = true
+ rSt.appName = v.Value.AppName
+ rSt.dbName = v.Value.DbName
+ rSt.initVec = v.Value.InitVec
+ rSt.sgIds = make(sgSet)
+ // Populate the sgids from the initvec.
+ for id := range rSt.initVec {
+ gid, err := strconv.ParseUint(id, 10, 64)
+ if err != nil {
+ vlog.Fatalf("sync: newResponderState: invalid syncgroup id", gid)
+ }
+ rSt.sgIds[interfaces.GroupId(gid)] = struct{}{}
+ }
+ }
return rSt
}
@@ -97,43 +120,60 @@
// (see http://goo.gl/mEa4L0) but only incur that overhead when the
// logging level specified is enabled.
vlog.VI(3).Infof("sync: sendDeltasPerDatabase: %s, %s: sgids %v, genvec %v",
- rSt.req.AppName, rSt.req.DbName, rSt.req.SgIds, rSt.req.InitVec)
+ rSt.appName, rSt.dbName, rSt.sgIds, rSt.initVec)
// Phase 1 of sendDeltas: Authorize the initiator and respond to the
// caller only for the SyncGroups that allow access.
- rSt.authorizeAndFilterSyncGroups(ctx)
+ err := rSt.authorizeAndFilterSyncGroups(ctx)
- // Phase 2 of sendDeltas: diff contains the bound on the
+ // Check error from phase 1.
+ if err != nil {
+ return err
+ }
+
+ if len(rSt.initVec) == 0 {
+ return verror.New(verror.ErrInternal, ctx, "empty initiator generation vector")
+ }
+
+ // Phase 2 and 3 of sendDeltas: diff contains the bound on the
// generations missing from the initiator per device.
- rSt.computeDeltaBound(ctx)
+ if rSt.sg {
+ err = rSt.sendSgDeltas(ctx)
+ } else {
+ err = rSt.sendDataDeltas(ctx)
+ }
- // Phase 3 of sendDeltas: Process the diff, filtering out records that
- // are not needed, and send the remainder on the wire ordered.
- return rSt.filterAndSendDeltas(ctx)
+ return err
}
// authorizeAndFilterSyncGroups authorizes the initiator against the requested
// SyncGroups and filters the initiator's prefixes to only include those from
// allowed SyncGroups (phase 1 of sendDeltas).
-func (rSt *responderState) authorizeAndFilterSyncGroups(ctx *context.T) {
- rSt.st, rSt.errState = rSt.sync.getDbStore(ctx, nil, rSt.req.AppName, rSt.req.DbName)
- if rSt.errState != nil {
- return
+func (rSt *responderState) authorizeAndFilterSyncGroups(ctx *context.T) error {
+ var err error
+ rSt.st, err = rSt.sync.getDbStore(ctx, nil, rSt.appName, rSt.dbName)
+ if err != nil {
+ return err
}
allowedPfxs := make(map[string]struct{})
- for sgid := range rSt.req.SgIds {
+ for sgid := range rSt.sgIds {
// Check permissions for the SyncGroup.
var sg *interfaces.SyncGroup
- sg, rSt.errState = getSyncGroupById(ctx, rSt.st, sgid)
- if rSt.errState != nil {
- return
- }
- rSt.errState = authorize(ctx, rSt.call.Security(), sg)
- if verror.ErrorID(rSt.errState) == verror.ErrNoAccess.ID {
+ sg, err = getSyncGroupById(ctx, rSt.st, sgid)
+ if err != nil {
+ vlog.Errorf("sync: authorizeAndFilterSyncGroups: accessing SyncGroup information failed %v, err %v", sgid, err)
continue
- } else if rSt.errState != nil {
- return
+ }
+ err = authorize(ctx, rSt.call.Security(), sg)
+ if verror.ErrorID(err) == verror.ErrNoAccess.ID {
+ if rSt.sg {
+ id := fmt.Sprintf("%d", sgid)
+ delete(rSt.initVec, id)
+ }
+ continue
+ } else if err != nil {
+ return err
}
for _, p := range sg.Spec.Prefixes {
@@ -147,8 +187,16 @@
rSt.addInitiatorToSyncGroup(ctx, sgid)
}
+ if err != nil {
+ return err
+ }
+
+ if rSt.sg {
+ return nil
+ }
+
// Filter the initiator's prefixes to what is allowed.
- for pfx := range rSt.req.InitVec {
+ for pfx := range rSt.initVec {
if _, ok := allowedPfxs[pfx]; ok {
continue
}
@@ -160,10 +208,10 @@
}
if !allowed {
- delete(rSt.req.InitVec, pfx)
+ delete(rSt.initVec, pfx)
}
}
- return
+ return nil
}
// addInitiatorToSyncGroup adds the request initiator to the membership of the
@@ -199,27 +247,58 @@
}
}
-// computeDeltaBound computes the bound on missing generations across all
-// requested prefixes (phase 2 of sendDeltas).
-func (rSt *responderState) computeDeltaBound(ctx *context.T) {
- // Check error from phase 1.
- if rSt.errState != nil {
- return
+// sendSgDeltas computes the bound on missing generations, and sends the missing
+// log records across all requested SyncGroups (phases 2 and 3 of sendDeltas).
+func (rSt *responderState) sendSgDeltas(ctx *context.T) error {
+ vlog.VI(3).Infof("sync: sendSgDeltas: %s, %s: sgids %v, genvec %v",
+ rSt.appName, rSt.dbName, rSt.sgIds, rSt.initVec)
+
+ respVec, _, err := rSt.sync.copyDbGenInfo(ctx, rSt.appName, rSt.dbName, rSt.sgIds)
+ if err != nil {
+ return err
}
- if len(rSt.req.InitVec) == 0 {
- rSt.errState = verror.New(verror.ErrInternal, ctx, "empty initiator generation vector")
- return
+ rSt.outVec = make(interfaces.GenVector)
+
+ for sg, initpgv := range rSt.initVec {
+ respgv, ok := respVec[sg]
+ if !ok {
+ continue
+ }
+ rSt.diff = make(genRangeVector)
+ rSt.diffPrefixGenVectors(respgv, initpgv)
+ rSt.outVec[sg] = respgv
+
+ if err := rSt.filterAndSendDeltas(ctx, sg); err != nil {
+ return err
+ }
+ }
+ return rSt.sendGenVec(ctx)
+}
+
+// sendDataDeltas computes the bound on missing generations across all requested
+// prefixes, and sends the missing log records (phases 2 and 3 of sendDeltas).
+func (rSt *responderState) sendDataDeltas(ctx *context.T) error {
+ // Phase 2 of sendDeltas: Compute the missing generations.
+ if err := rSt.computeDataDeltas(ctx); err != nil {
+ return err
}
- var respVec interfaces.GenVector
- var respGen uint64
- respVec, respGen, rSt.errState = rSt.sync.copyDbGenInfo(ctx, rSt.req.AppName, rSt.req.DbName)
- if rSt.errState != nil {
- return
+ // Phase 3 of sendDeltas: Process the diff, filtering out records that
+ // are not needed, and send the remainder on the wire ordered.
+ if err := rSt.filterAndSendDeltas(ctx, logDataPrefix); err != nil {
+ return err
+ }
+ return rSt.sendGenVec(ctx)
+}
+
+func (rSt *responderState) computeDataDeltas(ctx *context.T) error {
+ respVec, respGen, err := rSt.sync.copyDbGenInfo(ctx, rSt.appName, rSt.dbName, nil)
+ if err != nil {
+ return err
}
respPfxs := extractAndSortPrefixes(respVec)
- initPfxs := extractAndSortPrefixes(rSt.req.InitVec)
+ initPfxs := extractAndSortPrefixes(rSt.initVec)
rSt.outVec = make(interfaces.GenVector)
rSt.diff = make(genRangeVector)
@@ -235,7 +314,7 @@
pfx = p
// Lower bound on initiator's knowledge for this prefix set.
- initpgv := rSt.req.InitVec[pfx]
+ initpgv := rSt.initVec[pfx]
// Find the relevant responder prefixes and add the corresponding knowledge.
var respgv interfaces.PrefixGenVector
@@ -285,27 +364,17 @@
}
vlog.VI(3).Infof("sync: computeDeltaBound: %s, %s: diff %v, outvec %v",
- rSt.req.AppName, rSt.req.DbName, rSt.diff, rSt.outVec)
- return
+ rSt.appName, rSt.dbName, rSt.diff, rSt.outVec)
+ return nil
}
// filterAndSendDeltas filters the computed delta to remove records already
// known by the initiator, and sends the resulting records to the initiator
// (phase 3 of sendDeltas).
-func (rSt *responderState) filterAndSendDeltas(ctx *context.T) error {
- // Always send a start and finish response so that the initiator can
- // move on to the next Database.
- //
+func (rSt *responderState) filterAndSendDeltas(ctx *context.T, pfx string) error {
// TODO(hpucha): Although ok for now to call SendStream once per
// Database, would like to make this implementation agnostic.
sender := rSt.call.SendStream()
- sender.Send(interfaces.DeltaRespStart{true})
- defer sender.Send(interfaces.DeltaRespFinish{true})
-
- // Check error from phase 2.
- if rSt.errState != nil {
- return rSt.errState
- }
// First two phases were successful. So now on to phase 3. We now visit
// every log record in the generation range as obtained from phase 1 in
@@ -316,7 +385,7 @@
mh := make(minHeap, 0, len(rSt.diff))
for dev, r := range rSt.diff {
r.cur = r.min
- rec, err := getNextLogRec(ctx, rSt.st, dev, r)
+ rec, err := getNextLogRec(ctx, rSt.st, pfx, dev, r)
if err != nil {
return err
}
@@ -329,11 +398,14 @@
heap.Init(&mh)
// Process the log records in order.
- initPfxs := extractAndSortPrefixes(rSt.req.InitVec)
+ var initPfxs []string
+ if !rSt.sg {
+ initPfxs = extractAndSortPrefixes(rSt.initVec)
+ }
for mh.Len() > 0 {
rec := heap.Pop(&mh).(*localLogRec)
- if !filterLogRec(rec, rSt.req.InitVec, initPfxs) {
+ if rSt.sg || !filterLogRec(rec, rSt.initVec, initPfxs) {
// Send on the wire.
wireRec, err := makeWireLogRec(ctx, rSt.st, rec)
if err != nil {
@@ -344,7 +416,7 @@
// Add a new record from the same device if not done.
dev := rec.Metadata.Id
- rec, err := getNextLogRec(ctx, rSt.st, dev, rSt.diff[dev])
+ rec, err := getNextLogRec(ctx, rSt.st, pfx, dev, rSt.diff[dev])
if err != nil {
return err
}
@@ -354,7 +426,11 @@
delete(rSt.diff, dev)
}
}
+ return nil
+}
+func (rSt *responderState) sendGenVec(ctx *context.T) error {
+ sender := rSt.call.SendStream()
sender.Send(interfaces.DeltaRespRespVec{rSt.outVec})
return nil
}
@@ -426,9 +502,9 @@
// TODO(hpucha): This can be optimized using a scan instead of "gets" in a for
// loop.
-func getNextLogRec(ctx *context.T, st store.Store, dev uint64, r *genRange) (*localLogRec, error) {
+func getNextLogRec(ctx *context.T, st store.Store, pfx string, dev uint64, r *genRange) (*localLogRec, error) {
for i := r.cur; i <= r.max; i++ {
- rec, err := getLogRec(ctx, st, dev, i)
+ rec, err := getLogRec(ctx, st, pfx, dev, i)
if err == nil {
r.cur = i + 1
return rec, nil
diff --git a/services/syncbase/vsync/responder_test.go b/services/syncbase/vsync/responder_test.go
index dbd0fea..f1c42e0 100644
--- a/services/syncbase/vsync/responder_test.go
+++ b/services/syncbase/vsync/responder_test.go
@@ -111,7 +111,7 @@
for _, test := range tests {
want := test.genDiffWant
got := test.genDiffIn
- rSt := newResponderState(nil, nil, s, interfaces.DeltaReq{}, "fakeInitiator")
+ rSt := newResponderState(nil, nil, s, interfaces.DeltaReqData{}, "fakeInitiator")
rSt.diff = got
rSt.diffPrefixGenVectors(test.respPVec, test.initPVec)
checkEqualDevRanges(t, got, want)
@@ -121,7 +121,7 @@
// TestSendDeltas tests the computation of the delta bound (computeDeltaBound)
// and if the log records on the wire are correctly ordered (phases 2 and 3 of
// SendDeltas).
-func TestSendDeltas(t *testing.T) {
+func TestSendDataDeltas(t *testing.T) {
appName := "mockapp"
dbName := "mockdb"
@@ -350,16 +350,13 @@
s.id = 10 //responder.
wantDiff, wantVec := test.genDiff, test.outVec
- s.syncState[appDbName(appName, dbName)] = &dbSyncStateInMem{gen: test.respGen, checkptGen: test.respGen, genvec: test.respVec}
-
- req := interfaces.DeltaReq{AppName: appName, DbName: dbName, InitVec: test.initVec}
- rSt := newResponderState(nil, nil, s, req, "fakeInitiator")
-
- rSt.computeDeltaBound(nil)
- if rSt.errState != nil || !reflect.DeepEqual(rSt.outVec, wantVec) {
- t.Fatalf("computeDeltaBound failed (I: %v), (R: %v, %v), got %v, want %v err %v", test.initVec, test.respGen, test.respVec, rSt.outVec, wantVec, rSt.errState)
+ s.syncState[appDbName(appName, dbName)] = &dbSyncStateInMem{
+ data: &localGenInfoInMem{
+ gen: test.respGen,
+ checkptGen: test.respGen,
+ },
+ genvec: test.respVec,
}
- checkEqualDevRanges(t, rSt.diff, wantDiff)
////////////////////////////////////////
// Test sending deltas.
@@ -385,7 +382,7 @@
Metadata: interfaces.LogRecMetadata{Id: id, Gen: k, ObjId: okey, CurVers: vers, UpdTime: time.Now().UTC()},
Pos: pos + k,
}
- if err := putLogRec(nil, tx, rec); err != nil {
+ if err := putLogRec(nil, tx, logDataPrefix, rec); err != nil {
t.Fatalf("putLogRec(%d:%d) failed rec %v err %v", id, k, rec, err)
}
value := fmt.Sprintf("value_%s", okey)
@@ -404,16 +401,31 @@
t.Fatalf("cannot commit putting log rec, err %v", err)
}
+ req := interfaces.DataDeltaReq{
+ AppName: appName,
+ DbName: dbName,
+ InitVec: test.initVec,
+ }
+
+ rSt := newResponderState(nil, nil, s, interfaces.DeltaReqData{req}, "fakeInitiator")
d := &dummyResponder{}
rSt.call = d
- rSt.st, rSt.errState = rSt.sync.getDbStore(nil, nil, rSt.req.AppName, rSt.req.DbName)
- if rSt.errState != nil {
- t.Fatalf("filterAndSendDeltas failed to get store handle for app/db %v %v", rSt.req.AppName, rSt.req.DbName)
- }
- err := rSt.filterAndSendDeltas(nil)
+ var err error
+ rSt.st, err = rSt.sync.getDbStore(nil, nil, rSt.appName, rSt.dbName)
if err != nil {
- t.Fatalf("filterAndSendDeltas failed (I: %v), (R: %v, %v) err %v", test.initVec, test.respGen, test.respVec, err)
+ t.Fatalf("getDbStore failed to get store handle for app/db %v %v", rSt.appName, rSt.dbName)
}
+
+ err = rSt.computeDataDeltas(nil)
+ if err != nil || !reflect.DeepEqual(rSt.outVec, wantVec) {
+ t.Fatalf("computeDataDeltas failed (I: %v), (R: %v, %v), got %v, want %v err %v", test.initVec, test.respGen, test.respVec, rSt.outVec, wantVec, err)
+ }
+ checkEqualDevRanges(t, rSt.diff, wantDiff)
+
+ if err = rSt.sendDataDeltas(nil); err != nil {
+ t.Fatalf("sendDataDeltas failed, err %v", err)
+ }
+
d.diffLogRecs(t, wantRecs, wantVec)
destroyService(t, svc)
@@ -424,29 +436,10 @@
// Helpers
type dummyResponder struct {
- start, finish int
- gotRecs []*localLogRec
- outVec interfaces.GenVector
+ gotRecs []*localLogRec
+ outVec interfaces.GenVector
}
-func (d *dummyResponder) RecvStream() interface {
- Advance() bool
- Value() interfaces.DeltaReq
- Err() error
-} {
- return d
-}
-
-func (d *dummyResponder) Advance() bool {
- return false
-}
-
-func (d *dummyResponder) Value() interfaces.DeltaReq {
- return interfaces.DeltaReq{}
-}
-
-func (d *dummyResponder) Err() error { return nil }
-
func (d *dummyResponder) SendStream() interface {
Send(item interfaces.DeltaResp) error
} {
@@ -455,10 +448,6 @@
func (d *dummyResponder) Send(item interfaces.DeltaResp) error {
switch v := item.(type) {
- case interfaces.DeltaRespStart:
- d.start++
- case interfaces.DeltaRespFinish:
- d.finish++
case interfaces.DeltaRespRespVec:
d.outVec = v.Value
case interfaces.DeltaRespRec:
@@ -492,9 +481,6 @@
}
func (d *dummyResponder) diffLogRecs(t *testing.T, wantRecs []*localLogRec, wantVec interfaces.GenVector) {
- if d.start != 1 || d.finish != 1 {
- t.Fatalf("diffLogRecs incorrect start/finish records (%v, %v)", d.start, d.finish)
- }
if len(d.gotRecs) != len(wantRecs) {
t.Fatalf("diffLogRecs failed, gotLen %v, wantLen %v\n", len(d.gotRecs), len(wantRecs))
}
diff --git a/services/syncbase/vsync/sync_state.go b/services/syncbase/vsync/sync_state.go
index 94ab6eb..1eebf88 100644
--- a/services/syncbase/vsync/sync_state.go
+++ b/services/syncbase/vsync/sync_state.go
@@ -29,6 +29,26 @@
// generation vector for a Database. Log records are indexed such that they can
// be selectively retrieved from the store for any missing generation from any
// device.
+//
+// Sync also tracks the current generation number and the current local log
+// position for each mutation of a SyncGroup, created on a Database. Similar to
+// the data log records, these log records are used to sync SyncGroup metadata.
+//
+// The generations for the data mutations and mutations for each SyncGroup are
+// in separate spaces. Data mutations in a Database start at gen 1, and
+// grow. Mutations for each SyncGroup start at gen 1, and grow. Thus, for the
+// local data log records, the keys are of the form
+// $sync:log:data:<devid>:<gen>, and the keys for local SyncGroup log record are
+// of the form $sync:log:<sgid>:<devid>:<gen>.
+
+// TODO(hpucha): Should this space be separate from the data or not? If it is
+// not, it can provide consistency between data and SyncGroup metadata. For
+// example, lets say we mutate the data in a SyncGroup and soon after change the
+// SyncGroup ACL to prevent syncing with a device. This device may not get the
+// last batch of updates since the next time it will try to sync, it will be
+// rejected. However implementing consistency is not straightforward. Even if we
+// had SyncGroup updates in the same space as the data, we need to switch to the
+// right SyncGroup ACL at the responder based on the requested generations.
import (
"fmt"
@@ -36,24 +56,58 @@
"v.io/v23/context"
"v.io/v23/verror"
+ "v.io/x/lib/vlog"
"v.io/x/ref/services/syncbase/server/interfaces"
"v.io/x/ref/services/syncbase/server/util"
"v.io/x/ref/services/syncbase/store"
)
-// dbSyncStateInMem represents the in-memory sync state of a Database.
-type dbSyncStateInMem struct {
- gen uint64
- pos uint64
-
+// localGenInfoInMem represents the state corresponding to local generations.
+type localGenInfoInMem struct {
+ gen uint64
+ pos uint64
checkptGen uint64
- genvec interfaces.GenVector // Note: Generation vector contains state from remote devices only.
+}
+
+func (in *localGenInfoInMem) deepCopy() *localGenInfoInMem {
+ out := &localGenInfoInMem{
+ gen: in.gen,
+ pos: in.pos,
+ checkptGen: in.checkptGen,
+ }
+ return out
+}
+
+// dbSyncStateInMem represents the in-memory sync state of a Database and all
+// its SyncGroups.
+type dbSyncStateInMem struct {
+ data *localGenInfoInMem // info for data.
+ sgs map[interfaces.GroupId]*localGenInfoInMem // info for SyncGroups.
+
+ // Note: Generation vector contains state from remote devices only.
+ genvec interfaces.GenVector
+ sggenvec interfaces.GenVector
+}
+
+func (in *dbSyncStateInMem) deepCopy() *dbSyncStateInMem {
+ out := &dbSyncStateInMem{}
+ out.data = in.data.deepCopy()
+
+ out.sgs = make(map[interfaces.GroupId]*localGenInfoInMem)
+ for id, info := range in.sgs {
+ out.sgs[id] = info.deepCopy()
+ }
+
+ out.genvec = in.genvec.DeepCopy()
+ out.sggenvec = in.sggenvec.DeepCopy()
+
+ return out
}
// initSync initializes the sync module during startup. It scans all the
// databases across all apps to initialize the following:
-// a) in-memory sync state of a Database consisting of the current generation
-// number, log position and generation vector.
+// a) in-memory sync state of a Database and all its SyncGroups consisting of
+// the current generation number, log position and generation vector.
// b) watcher map of prefixes currently being synced.
// c) republish names in mount tables for all syncgroups.
//
@@ -94,14 +148,14 @@
if verror.ErrorID(err) == verror.ErrNoExist.ID {
scanStart, scanLimit = util.ScanPrefixArgs(logRecsPerDeviceScanPrefix(s.id), "")
} else {
- scanStart, scanLimit = util.ScanPrefixArgs(logRecKey(s.id, ds.Gen), "")
+ scanStart, scanLimit = util.ScanPrefixArgs(logRecKey(logDataPrefix, s.id, ds.Data.Gen), "")
}
var maxpos uint64
var dbName string
// Scan local log records to find the most recent one.
st.Scan(scanStart, scanLimit)
// Scan remote log records using the persisted GenVector.
- s.syncState[dbName] = &dbSyncStateInMem{pos: maxpos + 1}
+ s.syncState[dbName] = &dbSyncStateInMem{data: &localGenInfoInMem{pos: maxpos + 1}}
}
return false
@@ -110,42 +164,68 @@
return errFinal
}
+// Note: For all the utilities below, if the sgid parameter is non-nil, the
+// operation is performed in the SyncGroup space. If nil, it is performed in the
+// data space for the Database.
+//
+// TODO(hpucha): Once GroupId is changed to string, clean up these function
+// signatures.
+
// reserveGenAndPosInDbLog reserves a chunk of generation numbers and log
// positions in a Database's log. Used when local updates result in log
// entries.
-func (s *syncService) reserveGenAndPosInDbLog(ctx *context.T, appName, dbName string, count uint64) (uint64, uint64) {
- return s.reserveGenAndPosInternal(appName, dbName, count, count)
+func (s *syncService) reserveGenAndPosInDbLog(ctx *context.T, appName, dbName, sgid string, count uint64) (uint64, uint64) {
+ return s.reserveGenAndPosInternal(appName, dbName, sgid, count, count)
}
// reservePosInDbLog reserves a chunk of log positions in a Database's log. Used
// when remote log records are received.
-func (s *syncService) reservePosInDbLog(ctx *context.T, appName, dbName string, count uint64) uint64 {
- _, pos := s.reserveGenAndPosInternal(appName, dbName, 0, count)
+func (s *syncService) reservePosInDbLog(ctx *context.T, appName, dbName, sgid string, count uint64) uint64 {
+ _, pos := s.reserveGenAndPosInternal(appName, dbName, sgid, 0, count)
return pos
}
-func (s *syncService) reserveGenAndPosInternal(appName, dbName string, genCount, posCount uint64) (uint64, uint64) {
+func (s *syncService) reserveGenAndPosInternal(appName, dbName, sgid string, genCount, posCount uint64) (uint64, uint64) {
s.syncStateLock.Lock()
defer s.syncStateLock.Unlock()
name := appDbName(appName, dbName)
ds, ok := s.syncState[name]
if !ok {
- ds = &dbSyncStateInMem{gen: 1}
+ ds = &dbSyncStateInMem{
+ data: &localGenInfoInMem{gen: 1},
+ sgs: make(map[interfaces.GroupId]*localGenInfoInMem),
+ }
s.syncState[name] = ds
}
- gen := ds.gen
- pos := ds.pos
+ var info *localGenInfoInMem
+ if sgid != "" {
+ id, err := strconv.ParseUint(sgid, 10, 64)
+ if err != nil {
+ vlog.Fatalf("sync: reserveGenAndPosInternal: invalid syncgroup id", sgid)
+ }
- ds.gen += genCount
- ds.pos += posCount
+ var ok bool
+ info, ok = ds.sgs[interfaces.GroupId(id)]
+ if !ok {
+ info = &localGenInfoInMem{gen: 1}
+ ds.sgs[interfaces.GroupId(id)] = info
+ }
+ } else {
+ info = ds.data
+ }
+ gen := info.gen
+ pos := info.pos
+
+ info.gen += genCount
+ info.pos += posCount
return gen, pos
}
// checkptLocalGen freezes the local generation number for the responder's use.
-func (s *syncService) checkptLocalGen(ctx *context.T, appName, dbName string) error {
+func (s *syncService) checkptLocalGen(ctx *context.T, appName, dbName string, sgs sgSet) error {
s.syncStateLock.Lock()
defer s.syncStateLock.Unlock()
@@ -157,19 +237,40 @@
// The frozen generation is the last generation number used, i.e. one
// below the next available one to use.
- ds.checkptGen = ds.gen - 1
+ if len(sgs) > 0 {
+ // Checkpoint requested SyncGroups.
+ for id := range sgs {
+ info, ok := ds.sgs[id]
+ if !ok {
+ return verror.New(verror.ErrInternal, ctx, "sg state not found", name, id)
+ }
+ info.checkptGen = info.gen - 1
+ }
+ } else {
+ ds.data.checkptGen = ds.data.gen - 1
+ }
return nil
}
-// initDbSyncStateInMem initializes the in memory sync state of the Database if needed.
-func (s *syncService) initDbSyncStateInMem(ctx *context.T, appName, dbName string) {
+// initSyncStateInMem initializes the in memory sync state of the Database/SyncGroup if needed.
+func (s *syncService) initSyncStateInMem(ctx *context.T, appName, dbName string, sgid interfaces.GroupId) {
s.syncStateLock.Lock()
defer s.syncStateLock.Unlock()
name := appDbName(appName, dbName)
if s.syncState[name] == nil {
- s.syncState[name] = &dbSyncStateInMem{gen: 1}
+ s.syncState[name] = &dbSyncStateInMem{
+ data: &localGenInfoInMem{gen: 1},
+ sgs: make(map[interfaces.GroupId]*localGenInfoInMem),
+ }
}
+ if sgid != interfaces.NoGroupId {
+ ds := s.syncState[name]
+ if _, ok := ds.sgs[sgid]; !ok {
+ ds.sgs[sgid] = &localGenInfoInMem{gen: 1}
+ }
+ }
+ return
}
// copyDbSyncStateInMem returns a copy of the current in memory sync state of the Database.
@@ -182,20 +283,11 @@
if !ok {
return nil, verror.New(verror.ErrInternal, ctx, "db state not found", name)
}
-
- dsCopy := &dbSyncStateInMem{
- gen: ds.gen,
- pos: ds.pos,
- checkptGen: ds.checkptGen,
- }
-
- dsCopy.genvec = copyGenVec(ds.genvec)
-
- return dsCopy, nil
+ return ds.deepCopy(), nil
}
// copyDbGenInfo returns a copy of the current generation information of the Database.
-func (s *syncService) copyDbGenInfo(ctx *context.T, appName, dbName string) (interfaces.GenVector, uint64, error) {
+func (s *syncService) copyDbGenInfo(ctx *context.T, appName, dbName string, sgs sgSet) (interfaces.GenVector, uint64, error) {
s.syncStateLock.Lock()
defer s.syncStateLock.Unlock()
@@ -205,18 +297,30 @@
return nil, 0, verror.New(verror.ErrInternal, ctx, "db state not found", name)
}
- genvec := copyGenVec(ds.genvec)
+ var genvec interfaces.GenVector
+ var gen uint64
+ if len(sgs) > 0 {
+ genvec = make(interfaces.GenVector)
+ for id := range sgs {
+ sid := fmt.Sprintf("%d", id)
+ gv := ds.sggenvec[sid]
+ genvec[sid] = gv.DeepCopy()
+ genvec[sid][s.id] = ds.sgs[id].checkptGen
+ }
+ } else {
+ genvec = ds.genvec.DeepCopy()
- // Add local generation information to the genvec.
- for _, gv := range genvec {
- gv[s.id] = ds.checkptGen
+ // Add local generation information to the genvec.
+ for _, gv := range genvec {
+ gv[s.id] = ds.data.checkptGen
+ }
+ gen = ds.data.checkptGen
}
-
- return genvec, ds.checkptGen, nil
+ return genvec, gen, nil
}
// putDbGenInfoRemote puts the current remote generation information of the Database.
-func (s *syncService) putDbGenInfoRemote(ctx *context.T, appName, dbName string, genvec interfaces.GenVector) error {
+func (s *syncService) putDbGenInfoRemote(ctx *context.T, appName, dbName string, sg bool, genvec interfaces.GenVector) error {
s.syncStateLock.Lock()
defer s.syncStateLock.Unlock()
@@ -226,7 +330,11 @@
return verror.New(verror.ErrInternal, ctx, "db state not found", name)
}
- ds.genvec = copyGenVec(genvec)
+ if sg {
+ ds.sggenvec = genvec.DeepCopy()
+ } else {
+ ds.genvec = genvec.DeepCopy()
+ }
return nil
}
@@ -248,18 +356,6 @@
return parts[0], parts[1], nil
}
-func copyGenVec(in interfaces.GenVector) interfaces.GenVector {
- genvec := make(interfaces.GenVector)
- for p, inpgv := range in {
- pgv := make(interfaces.PrefixGenVector)
- for id, gen := range inpgv {
- pgv[id] = gen
- }
- genvec[p] = pgv
- }
- return genvec
-}
-
////////////////////////////////////////////////////////////
// Low-level utility functions to access sync state.
@@ -291,36 +387,42 @@
}
// logRecKey returns the key used to access a specific log record.
-func logRecKey(id, gen uint64) string {
- return util.JoinKeyParts(util.SyncPrefix, logPrefix, fmt.Sprintf("%d", id), fmt.Sprintf("%016x", gen))
+func logRecKey(pfx string, id, gen uint64) string {
+ return util.JoinKeyParts(util.SyncPrefix, logPrefix, pfx, fmt.Sprintf("%d", id), fmt.Sprintf("%016x", gen))
}
-// splitLogRecKey is the inverse of logRecKey and returns device id and generation number.
-func splitLogRecKey(ctx *context.T, key string) (uint64, uint64, error) {
+// splitLogRecKey is the inverse of logRecKey and returns the prefix, device id
+// and generation number.
+func splitLogRecKey(ctx *context.T, key string) (string, uint64, uint64, error) {
parts := util.SplitKeyParts(key)
verr := verror.New(verror.ErrInternal, ctx, "invalid logreckey", key)
- if len(parts) != 4 {
- return 0, 0, verr
+ if len(parts) != 5 {
+ return "", 0, 0, verr
}
if parts[0] != util.SyncPrefix || parts[1] != logPrefix {
- return 0, 0, verr
+ return "", 0, 0, verr
}
- id, err := strconv.ParseUint(parts[2], 10, 64)
+ if parts[2] != logDataPrefix {
+ if _, err := strconv.ParseUint(parts[2], 10, 64); err != nil {
+ return "", 0, 0, verr
+ }
+ }
+ id, err := strconv.ParseUint(parts[3], 10, 64)
if err != nil {
- return 0, 0, verr
+ return "", 0, 0, verr
}
- gen, err := strconv.ParseUint(parts[3], 16, 64)
+ gen, err := strconv.ParseUint(parts[4], 16, 64)
if err != nil {
- return 0, 0, verr
+ return "", 0, 0, verr
}
- return id, gen, nil
+ return parts[2], id, gen, nil
}
// hasLogRec returns true if the log record for (devid, gen) exists.
-func hasLogRec(st store.StoreReader, id, gen uint64) (bool, error) {
+func hasLogRec(st store.StoreReader, pfx string, id, gen uint64) (bool, error) {
// TODO(hpucha): optimize to avoid the unneeded fetch/decode of the data.
var rec localLogRec
- if err := util.Get(nil, st, logRecKey(id, gen), &rec); err != nil {
+ if err := util.Get(nil, st, logRecKey(pfx, id, gen), &rec); err != nil {
if verror.ErrorID(err) == verror.ErrNoExist.ID {
err = nil
}
@@ -330,20 +432,25 @@
}
// putLogRec stores the log record.
-func putLogRec(ctx *context.T, tx store.Transaction, rec *localLogRec) error {
- return util.Put(ctx, tx, logRecKey(rec.Metadata.Id, rec.Metadata.Gen), rec)
+func putLogRec(ctx *context.T, tx store.Transaction, pfx string, rec *localLogRec) error {
+ return util.Put(ctx, tx, logRecKey(pfx, rec.Metadata.Id, rec.Metadata.Gen), rec)
}
// getLogRec retrieves the log record for a given (devid, gen).
-func getLogRec(ctx *context.T, st store.StoreReader, id, gen uint64) (*localLogRec, error) {
+func getLogRec(ctx *context.T, st store.StoreReader, pfx string, id, gen uint64) (*localLogRec, error) {
+ return getLogRecByKey(ctx, st, logRecKey(pfx, id, gen))
+}
+
+// getLogRecByKey retrieves the log record for a given log record key.
+func getLogRecByKey(ctx *context.T, st store.StoreReader, key string) (*localLogRec, error) {
var rec localLogRec
- if err := util.Get(ctx, st, logRecKey(id, gen), &rec); err != nil {
+ if err := util.Get(ctx, st, key, &rec); err != nil {
return nil, err
}
return &rec, nil
}
// delLogRec deletes the log record for a given (devid, gen).
-func delLogRec(ctx *context.T, tx store.Transaction, id, gen uint64) error {
- return util.Delete(ctx, tx, logRecKey(id, gen))
+func delLogRec(ctx *context.T, tx store.Transaction, pfx string, id, gen uint64) error {
+ return util.Delete(ctx, tx, logRecKey(pfx, id, gen))
}
diff --git a/services/syncbase/vsync/sync_state_test.go b/services/syncbase/vsync/sync_state_test.go
index b621fdc..3036611 100644
--- a/services/syncbase/vsync/sync_state_test.go
+++ b/services/syncbase/vsync/sync_state_test.go
@@ -6,6 +6,7 @@
import (
"reflect"
+ "strconv"
"testing"
"time"
@@ -23,18 +24,31 @@
defer destroyService(t, svc)
s := svc.sync
- var wantGen, wantPos uint64 = 1, 0
- for i := 0; i < 5; i++ {
- gotGen, gotPos := s.reserveGenAndPosInternal("mockapp", "mockdb", 5, 10)
- if gotGen != wantGen || gotPos != wantPos {
- t.Fatalf("reserveGenAndPosInternal failed, gotGen %v wantGen %v, gotPos %v wantPos %v", gotGen, wantGen, gotPos, wantPos)
- }
- wantGen += 5
- wantPos += 10
+ sgids := []string{"", "100", "200"}
+ for _, sgid := range sgids {
+ var wantGen, wantPos uint64 = 1, 0
+ for i := 0; i < 5; i++ {
+ gotGen, gotPos := s.reserveGenAndPosInternal("mockapp", "mockdb", sgid, 5, 10)
+ if gotGen != wantGen || gotPos != wantPos {
+ t.Fatalf("reserveGenAndPosInternal failed, gotGen %v wantGen %v, gotPos %v wantPos %v", gotGen, wantGen, gotPos, wantPos)
+ }
+ wantGen += 5
+ wantPos += 10
- name := appDbName("mockapp", "mockdb")
- if s.syncState[name].gen != wantGen || s.syncState[name].pos != wantPos {
- t.Fatalf("reserveGenAndPosInternal failed, gotGen %v wantGen %v, gotPos %v wantPos %v", s.syncState[name].gen, wantGen, s.syncState[name].pos, wantPos)
+ name := appDbName("mockapp", "mockdb")
+ var info *localGenInfoInMem
+ if sgid == "" {
+ info = s.syncState[name].data
+ } else {
+ id, err := strconv.ParseUint(sgid, 10, 64)
+ if err != nil {
+ t.Fatalf("reserveGenAndPosInternal failed, invalid sgid %v", sgid)
+ }
+ info = s.syncState[name].sgs[interfaces.GroupId(id)]
+ }
+ if info.gen != wantGen || info.pos != wantPos {
+ t.Fatalf("reserveGenAndPosInternal failed, gotGen %v wantGen %v, gotPos %v wantPos %v", info.gen, wantGen, info.pos, wantPos)
+ }
}
}
}
@@ -52,9 +66,25 @@
1: 2, 3: 4, 5: 6,
},
}
+ sggv := interfaces.GenVector{
+ "mocksg1": interfaces.PrefixGenVector{
+ 10: 20, 30: 40, 50: 60,
+ },
+ "mocksg2": interfaces.PrefixGenVector{
+ 100: 200, 300: 400, 500: 600,
+ },
+ }
+ localsgs := make(map[interfaces.GroupId]localGenInfo)
+ localsgs[interfaces.GroupId(8888)] = localGenInfo{Gen: 56, CheckptGen: 2000}
+ localsgs[interfaces.GroupId(1008888)] = localGenInfo{Gen: 25890, CheckptGen: 100}
tx := st.NewTransaction()
- wantSt := &dbSyncState{Gen: 40, GenVec: gv}
+ wantSt := &dbSyncState{
+ Data: localGenInfo{Gen: 40},
+ Sgs: localsgs,
+ GenVec: gv,
+ SgGenVec: sggv,
+ }
if err := putDbSyncState(nil, tx, wantSt); err != nil {
t.Fatalf("putDbSyncState failed, err %v", err)
}
@@ -92,8 +122,8 @@
},
Pos: 10,
}
- if err := putLogRec(nil, tx, wantRec); err != nil {
- t.Fatalf("putLogRec(%d:%d) failed err %v", id, gen, err)
+ if err := putLogRec(nil, tx, logDataPrefix, wantRec); err != nil {
+ t.Fatalf("putLogRec(%s:%d:%d) failed err %v", logDataPrefix, id, gen, err)
}
if err := tx.Commit(); err != nil {
t.Fatalf("cannot commit putting log rec, err %v", err)
@@ -102,8 +132,8 @@
checkLogRec(t, st, id, gen, true, wantRec)
tx = st.NewTransaction()
- if err := delLogRec(nil, tx, id, gen); err != nil {
- t.Fatalf("delLogRec(%d:%d) failed err %v", id, gen, err)
+ if err := delLogRec(nil, tx, logDataPrefix, id, gen); err != nil {
+ t.Fatalf("delLogRec(%s:%d:%d) failed err %v", logDataPrefix, id, gen, err)
}
if err := tx.Commit(); err != nil {
t.Fatalf("cannot commit deleting log rec, err %v", err)
@@ -113,27 +143,28 @@
}
func TestLogRecKeyUtils(t *testing.T) {
- invalid := []string{"$sync:aa:bb", "log:aa:bb", "$sync:log:aa:xx", "$sync:log:x:bb"}
+ invalid := []string{"$sync:100:bb", "log:100:bb", "$sync:log:data:100:xx", "$sync:log:data:aa:bb", "$sync:log:xx:100:bb"}
for _, k := range invalid {
- if _, _, err := splitLogRecKey(nil, k); err == nil {
+ if _, _, _, err := splitLogRecKey(nil, k); err == nil {
t.Fatalf("splitting log rec key didn't fail %q", k)
}
}
valid := []struct {
+ pfx string
id uint64
gen uint64
}{
- {10, 20},
- {190, 540},
- {9999, 999999},
+ {logDataPrefix, 10, 20},
+ {"2500", 190, 540},
+ {"4200", 9999, 999999},
}
for _, v := range valid {
- gotId, gotGen, err := splitLogRecKey(nil, logRecKey(v.id, v.gen))
- if gotId != v.id || gotGen != v.gen || err != nil {
- t.Fatalf("failed key conversion id got %v want %v, gen got %v want %v, err %v", gotId, v.id, gotGen, v.gen, err)
+ gotPfx, gotId, gotGen, err := splitLogRecKey(nil, logRecKey(v.pfx, v.id, v.gen))
+ if gotPfx != v.pfx || gotId != v.id || gotGen != v.gen || err != nil {
+ t.Fatalf("failed key conversion pfx got %v want %v, id got %v want %v, gen got %v want %v, err %v", gotPfx, v.pfx, gotId, v.id, gotGen, v.gen, err)
}
}
}
@@ -158,7 +189,7 @@
}
func checkLogRec(t *testing.T, st store.StoreReader, id, gen uint64, exists bool, wantRec *localLogRec) {
- gotRec, err := getLogRec(nil, st, id, gen)
+ gotRec, err := getLogRec(nil, st, logDataPrefix, id, gen)
if (!exists && err == nil) || (exists && err != nil) {
t.Fatalf("getLogRec(%d:%d) failed, exists %v err %v", id, gen, exists, err)
@@ -168,7 +199,7 @@
t.Fatalf("getLogRec(%d:%d) failed, got %v, want %v", id, gen, gotRec, wantRec)
}
- if ok, err := hasLogRec(st, id, gen); err != nil || ok != exists {
+ if ok, err := hasLogRec(st, logDataPrefix, id, gen); err != nil || ok != exists {
t.Fatalf("hasLogRec(%d:%d) failed, want %v", id, gen, exists)
}
}
diff --git a/services/syncbase/vsync/syncer.go b/services/syncbase/vsync/syncer.go
new file mode 100644
index 0000000..22f485b
--- /dev/null
+++ b/services/syncbase/vsync/syncer.go
@@ -0,0 +1,105 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package vsync
+
+import (
+ "time"
+
+ "v.io/v23/context"
+ "v.io/v23/verror"
+ "v.io/x/lib/vlog"
+)
+
+// Policies to pick a peer to sync with.
+const (
+ // Picks a peer at random from the available set.
+ selectRandom = iota
+
+ // TODO(hpucha): implement other policies.
+ // Picks a peer with most differing generations.
+ selectMostDiff
+
+ // Picks a peer that was synced with the furthest in the past.
+ selectOldest
+)
+
+var (
+ // peerSyncInterval is the duration between two consecutive peer
+ // contacts. During every peer contact, the initiator obtains any
+ // pending updates from that peer.
+ peerSyncInterval = 50 * time.Millisecond
+
+ // peerSelectionPolicy is the policy used to select a peer when
+ // the initiator gets a chance to sync.
+ peerSelectionPolicy = selectRandom
+)
+
+// syncer wakes up every peerSyncInterval to do work: (1) Refresh memberView if
+// needed and pick a peer from all the known remote peers to sync with. (2) Act
+// as an initiator and sync Syncgroup metadata for all common SyncGroups with
+// the chosen peer (getting updates from the remote peer, detecting and
+// resolving conflicts) (3) Act as an initiator and sync data corresponding to
+// all common SyncGroups across all Apps/Databases with the chosen peer; (4)
+// Fetch any queued blobs. (5) Transfer ownership of blobs if needed. (6) Act as
+// a SyncGroup publisher to publish pending SyncGroups; (6) Garbage collect
+// older generations.
+//
+// TODO(hpucha): Currently only does initiation. Add rest.
+func (s *syncService) syncer(ctx *context.T) {
+ defer s.pending.Done()
+
+ ticker := time.NewTicker(peerSyncInterval)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-s.closed:
+ vlog.VI(1).Info("sync: syncer: channel closed, stop work and exit")
+ return
+
+ case <-ticker.C:
+ }
+
+ // TODO(hpucha): Cut a gen for the responder even if there is no
+ // one to initiate to?
+
+ // Do work.
+ peer, err := s.pickPeer(ctx)
+ if err != nil {
+ continue
+ }
+
+ // Sync Syncgroup metadata and data.
+ s.getDeltas(ctx, peer)
+ }
+}
+
+////////////////////////////////////////
+// Peer selection policies.
+
+// pickPeer picks a Syncbase to sync with.
+func (s *syncService) pickPeer(ctx *context.T) (string, error) {
+ switch peerSelectionPolicy {
+ case selectRandom:
+ members := s.getMembers(ctx)
+ // Remove myself from the set.
+ delete(members, s.name)
+ if len(members) == 0 {
+ return "", verror.New(verror.ErrInternal, ctx, "no useful peer")
+ }
+
+ // Pick a peer at random.
+ ind := randIntn(len(members))
+ for m := range members {
+ if ind == 0 {
+ return m, nil
+ }
+ ind--
+ }
+ return "", verror.New(verror.ErrInternal, ctx, "random selection didn't succeed")
+ default:
+ return "", verror.New(verror.ErrInternal, ctx, "unknown peer selection policy")
+ }
+}
diff --git a/services/syncbase/vsync/syncgroup.go b/services/syncbase/vsync/syncgroup.go
index 7225404..91ba499 100644
--- a/services/syncbase/vsync/syncgroup.go
+++ b/services/syncbase/vsync/syncgroup.go
@@ -54,10 +54,13 @@
}
// memberInfo holds the member metadata for each SyncGroup this member belongs
-// to within each App/Database (i.e. global database name). It's a mapping of
-// global DB names to sets of SyncGroup member information.
+// to within each App/Database (i.e. global database name). It's a mapping of
+// global DB names to sets of SyncGroup member information. It also maintains
+// all the mount table candidates that could be used to reach this peer, learned
+// from the SyncGroup metadata.
type memberInfo struct {
- db2sg map[string]sgMemberInfo
+ db2sg map[string]sgMemberInfo
+ mtTables map[string]struct{}
}
// sgMemberInfo maps SyncGroups to their member metadata.
@@ -220,12 +223,20 @@
// A member's info is different across SyncGroups, so gather all of them.
for member, info := range sg.Joiners {
if _, ok := newMembers[member]; !ok {
- newMembers[member] = &memberInfo{db2sg: make(map[string]sgMemberInfo)}
+ newMembers[member] = &memberInfo{
+ db2sg: make(map[string]sgMemberInfo),
+ mtTables: make(map[string]struct{}),
+ }
}
if _, ok := newMembers[member].db2sg[name]; !ok {
newMembers[member].db2sg[name] = make(sgMemberInfo)
}
newMembers[member].db2sg[name][sg.Id] = info
+
+ // Collect mount tables.
+ for _, mt := range sg.Spec.MountTables {
+ newMembers[member].mtTables[mt] = struct{}{}
+ }
}
return false
})
@@ -291,13 +302,19 @@
}
// Make a copy.
- infoCopy := &memberInfo{make(map[string]sgMemberInfo)}
+ infoCopy := &memberInfo{
+ db2sg: make(map[string]sgMemberInfo),
+ mtTables: make(map[string]struct{}),
+ }
for gdbName, sgInfo := range info.db2sg {
infoCopy.db2sg[gdbName] = make(sgMemberInfo)
for gid, mi := range sgInfo {
infoCopy.db2sg[gdbName][gid] = mi
}
}
+ for mt := range info.mtTables {
+ infoCopy.mtTables[mt] = struct{}{}
+ }
return infoCopy
}
@@ -407,6 +424,10 @@
vlog.VI(2).Infof("sync: CreateSyncGroup: begin: %s", sgName)
defer vlog.VI(2).Infof("sync: CreateSyncGroup: end: %s", sgName)
+ // Get this Syncbase's sync module handle.
+ ss := sd.sync.(*syncService)
+ var sg *interfaces.SyncGroup
+
err := store.RunInTransaction(sd.db.St(), func(tx store.Transaction) error {
// Check permissions on Database.
if err := sd.db.CheckPermsInternal(ctx, call, tx); err != nil {
@@ -419,11 +440,8 @@
// TODO(hpucha): Do some SG ACL checking. Check creator
// has Admin privilege.
- // Get this Syncbase's sync module handle.
- ss := sd.sync.(*syncService)
-
// Instantiate sg. Add self as joiner.
- sg := &interfaces.SyncGroup{
+ sg = &interfaces.SyncGroup{
Id: newSyncGroupId(),
Name: sgName,
SpecVersion: newSyncGroupVersion(),
@@ -449,6 +467,7 @@
return err
}
+ ss.initSyncStateInMem(ctx, sg.AppName, sg.DbName, sg.Id)
// Local SG create succeeded. Publish the SG at the chosen server.
sd.publishSyncGroup(ctx, call, sgName)
@@ -542,6 +561,8 @@
return nullSpec, err
}
+ ss.initSyncStateInMem(ctx, sg.AppName, sg.DbName, sg.Id)
+
// Publish at the chosen mount table and in the neighborhood.
sd.publishInMountTables(ctx, call, sg.Spec)
@@ -841,7 +862,11 @@
return addSyncGroup(ctx, tx, &sg)
})
- return err
+ if err != nil {
+ return err
+ }
+ s.initSyncStateInMem(ctx, sg.AppName, sg.DbName, sg.Id)
+ return nil
}
func (s *syncService) JoinSyncGroupAtAdmin(ctx *context.T, call rpc.ServerCall, sgName, joinerName string, joinerInfo wire.SyncGroupMemberInfo) (interfaces.SyncGroup, error) {
diff --git a/services/syncbase/vsync/syncgroup_test.go b/services/syncbase/vsync/syncgroup_test.go
index e48e1f6..4dc5845 100644
--- a/services/syncbase/vsync/syncgroup_test.go
+++ b/services/syncbase/vsync/syncgroup_test.go
@@ -329,7 +329,8 @@
Creator: "mockCreator",
SpecVersion: "etag-1",
Spec: nosql.SyncGroupSpec{
- Prefixes: []string{"foo"},
+ MountTables: []string{"mt1"},
+ Prefixes: []string{"foo"},
},
Joiners: map[string]nosql.SyncGroupMemberInfo{
"phone": nosql.SyncGroupMemberInfo{SyncPriority: 10},
@@ -345,7 +346,8 @@
Creator: "mockCreator",
SpecVersion: "etag-2",
Spec: nosql.SyncGroupSpec{
- Prefixes: []string{"bar"},
+ MountTables: []string{"mt2", "mt3"},
+ Prefixes: []string{"bar"},
},
Joiners: map[string]nosql.SyncGroupMemberInfo{
"tablet": nosql.SyncGroupMemberInfo{SyncPriority: 111},
@@ -383,6 +385,11 @@
t.Errorf("invalid SyncGroup members: got %v instead of %v", members, expMembers)
}
+ mt2and3 := map[string]struct{}{
+ "mt2": struct{}{},
+ "mt3": struct{}{},
+ }
+
expMemberInfo := map[string]*memberInfo{
"phone": &memberInfo{
db2sg: map[string]sgMemberInfo{
@@ -390,6 +397,7 @@
sgId1: sg1.Joiners["phone"],
},
},
+ mtTables: map[string]struct{}{"mt1": struct{}{}},
},
"tablet": &memberInfo{
db2sg: map[string]sgMemberInfo{
@@ -398,6 +406,11 @@
sgId2: sg2.Joiners["tablet"],
},
},
+ mtTables: map[string]struct{}{
+ "mt1": struct{}{},
+ "mt2": struct{}{},
+ "mt3": struct{}{},
+ },
},
"cloud": &memberInfo{
db2sg: map[string]sgMemberInfo{
@@ -405,6 +418,7 @@
sgId1: sg1.Joiners["cloud"],
},
},
+ mtTables: map[string]struct{}{"mt1": struct{}{}},
},
"door": &memberInfo{
db2sg: map[string]sgMemberInfo{
@@ -412,6 +426,7 @@
sgId2: sg2.Joiners["door"],
},
},
+ mtTables: mt2and3,
},
"lamp": &memberInfo{
db2sg: map[string]sgMemberInfo{
@@ -419,6 +434,7 @@
sgId2: sg2.Joiners["lamp"],
},
},
+ mtTables: mt2and3,
},
}
@@ -430,7 +446,7 @@
}
expInfo := expMemberInfo[mm]
if !reflect.DeepEqual(mi, expInfo) {
- t.Errorf("invalid Info for SyncGroup member %s: got %v instead of %v", mm, mi, expInfo)
+ t.Errorf("invalid Info for SyncGroup member %s: got %#v instead of %#v", mm, mi, expInfo)
}
}
@@ -462,6 +478,7 @@
sgId2: sg2.Joiners["tablet"],
},
},
+ mtTables: mt2and3,
},
"door": &memberInfo{
db2sg: map[string]sgMemberInfo{
@@ -469,6 +486,7 @@
sgId2: sg2.Joiners["door"],
},
},
+ mtTables: mt2and3,
},
"lamp": &memberInfo{
db2sg: map[string]sgMemberInfo{
@@ -476,6 +494,7 @@
sgId2: sg2.Joiners["lamp"],
},
},
+ mtTables: mt2and3,
},
}
diff --git a/services/syncbase/vsync/testdata/local-init-00.log.sync b/services/syncbase/vsync/testdata/local-init-00.log.sync
index 7435348..7f655eb 100644
--- a/services/syncbase/vsync/testdata/local-init-00.log.sync
+++ b/services/syncbase/vsync/testdata/local-init-00.log.sync
@@ -1,6 +1,6 @@
# Create an object locally and update it twice (linked-list).
# The format is: <cmd>|<objid>|<version>|<parent1>|<parent2>|<logrec>|<txid>|<txcount>|<deleted>
-addl|foo1|1|||$sync:log:10:1|0|1|false
-addl|foo1|2|1||$sync:log:10:2|0|1|false
-addl|foo1|3|2||$sync:log:10:3|0|1|false
+addl|foo1|1|||$sync:log:data:10:1|0|1|false
+addl|foo1|2|1||$sync:log:data:10:2|0|1|false
+addl|foo1|3|2||$sync:log:data:10:3|0|1|false
diff --git a/services/syncbase/vsync/testdata/remote-conf-00.log.sync b/services/syncbase/vsync/testdata/remote-conf-00.log.sync
index 060cf0c..66cdce3 100644
--- a/services/syncbase/vsync/testdata/remote-conf-00.log.sync
+++ b/services/syncbase/vsync/testdata/remote-conf-00.log.sync
@@ -3,6 +3,6 @@
# it from the local sync at v2, then updated separately).
# The format is: <cmd>|<objid>|<version>|<parent1>|<parent2>|<logrec>|<txid>|<txcount>|<deleted>
-addr|foo1|4|2||$sync:log:11:1|0|1|false
-addr|foo1|5|4||$sync:log:11:2|0|1|false
-addr|foo1|6|5||$sync:log:11:3|0|1|false
+addr|foo1|4|2||$sync:log:data:11:1|0|1|false
+addr|foo1|5|4||$sync:log:data:11:2|0|1|false
+addr|foo1|6|5||$sync:log:data:11:3|0|1|false
diff --git a/services/syncbase/vsync/testdata/remote-conf-01.log.sync b/services/syncbase/vsync/testdata/remote-conf-01.log.sync
index 2053157..5f440b5 100644
--- a/services/syncbase/vsync/testdata/remote-conf-01.log.sync
+++ b/services/syncbase/vsync/testdata/remote-conf-01.log.sync
@@ -5,6 +5,6 @@
# sees 2 graft points: v1-v4 and v2-v5.
# The format is: <cmd>|<objid>|<version>|<parent1>|<parent2>|<logrec>|<txid>|<txcount>|<deleted>
-addr|foo1|4|1||$sync:log:12:1|0|1|false
-addr|foo1|5|2|4|$sync:log:11:1|0|1|false
-addr|foo1|6|5||$sync:log:11:2|0|1|false
+addr|foo1|4|1||$sync:log:data:12:1|0|1|false
+addr|foo1|5|2|4|$sync:log:data:11:1|0|1|false
+addr|foo1|6|5||$sync:log:data:11:2|0|1|false
diff --git a/services/syncbase/vsync/testdata/remote-conf-03.log.sync b/services/syncbase/vsync/testdata/remote-conf-03.log.sync
index 673405e..edf91a0 100644
--- a/services/syncbase/vsync/testdata/remote-conf-03.log.sync
+++ b/services/syncbase/vsync/testdata/remote-conf-03.log.sync
@@ -1,6 +1,6 @@
# Create the same object remotely from scratch and update it twice (linked-list).
# The format is: <cmd>|<objid>|<version>|<parent1>|<parent2>|<logrec>|<txid>|<txcount>|<deleted>
-addr|foo1|4|||$sync:log:11:1|0|1|false
-addr|foo1|5|4||$sync:log:11:2|0|1|false
-addr|foo1|6|5||$sync:log:11:3|0|1|false
+addr|foo1|4|||$sync:log:data:11:1|0|1|false
+addr|foo1|5|4||$sync:log:data:11:2|0|1|false
+addr|foo1|6|5||$sync:log:data:11:3|0|1|false
diff --git a/services/syncbase/vsync/testdata/remote-init-00.log.sync b/services/syncbase/vsync/testdata/remote-init-00.log.sync
index 35b1511..a5a4864 100644
--- a/services/syncbase/vsync/testdata/remote-init-00.log.sync
+++ b/services/syncbase/vsync/testdata/remote-init-00.log.sync
@@ -1,7 +1,7 @@
# Create an object remotely and update it twice (linked-list).
# The format is: <cmd>|<objid>|<version>|<parent1>|<parent2>|<logrec>|<txid>|<txcount>|<deleted>
-addr|foo1|1|||$sync:log:11:1|0|1|false
-addr|foo1|2|1||$sync:log:11:2|0|1|false
-addr|foo1|3|2||$sync:log:11:3|0|1|false
+addr|foo1|1|||$sync:log:data:11:1|0|1|false
+addr|foo1|2|1||$sync:log:data:11:2|0|1|false
+addr|foo1|3|2||$sync:log:data:11:3|0|1|false
genvec|foo1|10:0,11:3|bar|11:0
diff --git a/services/syncbase/vsync/testdata/remote-noconf-00.log.sync b/services/syncbase/vsync/testdata/remote-noconf-00.log.sync
index ef52b0a..94e144e 100644
--- a/services/syncbase/vsync/testdata/remote-noconf-00.log.sync
+++ b/services/syncbase/vsync/testdata/remote-noconf-00.log.sync
@@ -2,8 +2,7 @@
# after it was created locally up to v3 (i.e. assume the remote sync
# received it from the local sync first, then updated it).
# The format is: <cmd>|<objid>|<version>|<parent1>|<parent2>|<logrec>|<txid>|<txcount>|<deleted>
-
-addr|foo1|4|3||$sync:log:11:1|0|1|false
-addr|foo1|5|4||$sync:log:11:2|0|1|false
-addr|foo1|6|5||$sync:log:11:3|0|1|false
+addr|foo1|4|3||$sync:log:data:11:1|0|1|false
+addr|foo1|5|4||$sync:log:data:11:2|0|1|false
+addr|foo1|6|5||$sync:log:data:11:3|0|1|false
genvec|foo1|10:0,11:3|bar|11:0
diff --git a/services/syncbase/vsync/types.vdl b/services/syncbase/vsync/types.vdl
index f34d415..9cf53d3 100644
--- a/services/syncbase/vsync/types.vdl
+++ b/services/syncbase/vsync/types.vdl
@@ -11,10 +11,11 @@
// Key prefixes for sync data structures. All these prefixes are prepended with
// util.SyncPrefix.
const (
- logPrefix = "log"
- dbssPrefix = "dbss"
- dagPrefix = "dag"
- sgPrefix = "sg"
+ logPrefix = "log" // log state.
+ logDataPrefix = "data" // data log state.
+ dbssPrefix = "dbss" // database sync state.
+ dagPrefix = "dag" // dag state.
+ sgPrefix = "sg" // syncgroup state.
)
// syncData represents the persistent state of the sync module.
@@ -22,11 +23,18 @@
Id uint64
}
+// localGenInfo represents the persistent state corresponding to local generations.
+type localGenInfo struct {
+ Gen uint64 // local generation number incremented on every local update.
+ CheckptGen uint64 // local generation number advertised to remote peers (used by the responder).
+}
+
// dbSyncState represents the persistent sync state of a Database.
type dbSyncState struct {
- Gen uint64 // local generation number incremented on every local update.
- CheckptGen uint64 // local generation number advertised to remote peers (used by the responder).
- GenVec interfaces.GenVector // generation vector capturing the locally-known generations of remote peers.
+ Data localGenInfo
+ Sgs map[interfaces.GroupId]localGenInfo
+ GenVec interfaces.GenVector // generation vector capturing the locally-known generations of remote peers for data in Database.
+ SgGenVec interfaces.GenVector // generation vector capturing the locally-known generations of remote peers for SyncGroups in Database.
}
// localLogRec represents the persistent local state of a log record. Metadata
diff --git a/services/syncbase/vsync/types.vdl.go b/services/syncbase/vsync/types.vdl.go
index f305083..c02d2d4 100644
--- a/services/syncbase/vsync/types.vdl.go
+++ b/services/syncbase/vsync/types.vdl.go
@@ -25,11 +25,23 @@
}) {
}
+// localGenInfo represents the persistent state corresponding to local generations.
+type localGenInfo struct {
+ Gen uint64 // local generation number incremented on every local update.
+ CheckptGen uint64 // local generation number advertised to remote peers (used by the responder).
+}
+
+func (localGenInfo) __VDLReflect(struct {
+ Name string `vdl:"v.io/x/ref/services/syncbase/vsync.localGenInfo"`
+}) {
+}
+
// dbSyncState represents the persistent sync state of a Database.
type dbSyncState struct {
- Gen uint64 // local generation number incremented on every local update.
- CheckptGen uint64 // local generation number advertised to remote peers (used by the responder).
- GenVec interfaces.GenVector // generation vector capturing the locally-known generations of remote peers.
+ Data localGenInfo
+ Sgs map[interfaces.GroupId]localGenInfo
+ GenVec interfaces.GenVector // generation vector capturing the locally-known generations of remote peers for data in Database.
+ SgGenVec interfaces.GenVector // generation vector capturing the locally-known generations of remote peers for SyncGroups in Database.
}
func (dbSyncState) __VDLReflect(struct {
@@ -51,14 +63,17 @@
func init() {
vdl.Register((*syncData)(nil))
+ vdl.Register((*localGenInfo)(nil))
vdl.Register((*dbSyncState)(nil))
vdl.Register((*localLogRec)(nil))
}
-const logPrefix = "log"
+const logPrefix = "log" // log state.
-const dbssPrefix = "dbss"
+const logDataPrefix = "data" // data log state.
-const dagPrefix = "dag"
+const dbssPrefix = "dbss" // database sync state.
-const sgPrefix = "sg"
+const dagPrefix = "dag" // dag state.
+
+const sgPrefix = "sg" // syncgroup state.
diff --git a/services/syncbase/vsync/watcher.go b/services/syncbase/vsync/watcher.go
index 2e9e6fc..9c16294 100644
--- a/services/syncbase/vsync/watcher.go
+++ b/services/syncbase/vsync/watcher.go
@@ -120,7 +120,7 @@
}
// Initialize Database sync state if needed.
- s.initDbSyncStateInMem(ctx, appName, dbName)
+ s.initSyncStateInMem(ctx, appName, dbName, interfaces.NoGroupId)
// Get a batch of watch log entries, if any, after this resume marker.
logs, nextResmark, err := watchable.ReadBatchFromLog(st, resMark)
@@ -212,7 +212,7 @@
}
}
- gen, pos := s.reserveGenAndPosInDbLog(ctx, appName, dbName, count)
+ gen, pos := s.reserveGenAndPosInDbLog(ctx, appName, dbName, "", count)
vlog.VI(3).Infof("sync: processBatch: %s, %s: len %d, total %d, btid %x, gen %d, pos %d",
appName, dbName, count, totalCount, batchId, gen, pos)
@@ -251,12 +251,12 @@
// suitably updating the DAG metadata.
func (s *syncService) processLocalLogRec(ctx *context.T, tx store.Transaction, rec *localLogRec) error {
// Insert the new log record into the log.
- if err := putLogRec(ctx, tx, rec); err != nil {
+ if err := putLogRec(ctx, tx, logDataPrefix, rec); err != nil {
return err
}
m := rec.Metadata
- logKey := logRecKey(m.Id, m.Gen)
+ logKey := logRecKey(logDataPrefix, m.Id, m.Gen)
// Insert the new log record into dag.
if err := s.addNode(ctx, tx, m.ObjId, m.CurVers, logKey, m.Delete, m.Parents, m.BatchId, nil); err != nil {
