Add clock sync capability to syncbase server.
1) Adds a syncservice in clock package that deals with
clock data response from peer
2) Clock sync logic to vsync package with makes
rpc call to peer right before initiating getDeltas
3) Deamon to service to run clockservice and
ntpservice in the background
4) Unit tests for syncservice
Change-Id: I6434fb940a2b8c0360d6b71c213b0b0d5220960f
diff --git a/services/syncbase/clock/clockd.go b/services/syncbase/clock/clockd.go
new file mode 100644
index 0000000..6543406
--- /dev/null
+++ b/services/syncbase/clock/clockd.go
@@ -0,0 +1,95 @@
+// 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 clock
+
+// ClockD provides a daemon to run services provided within this package
+// like clockservice and ntpservice. These services are run regularly
+// in an independent go routine and can be exited by calling Close().
+import (
+ "sync"
+ "time"
+
+ "v.io/v23/context"
+ "v.io/x/lib/vlog"
+)
+
+const (
+ // clockCheckInterval is the duration between two consecutive runs of
+ // clockservice to catch any changes to the system clock.
+ clockCheckInterval = 1 * time.Second
+
+ // ntpCheckInterval is the duration between two consecutive clock syncs with
+ // NTP. It allows us to keep the syncbase clock in sync with NTP.
+ ntpCheckInterval = 1 * time.Hour
+)
+
+type ClockD struct {
+ vclock *VClock
+
+ // State to coordinate shutdown of spawned goroutines.
+ pending sync.WaitGroup
+ closed chan struct{}
+}
+
+func StartClockD(ctx *context.T, clock *VClock) *ClockD {
+ clockD := &ClockD{
+ vclock: clock,
+ }
+ clockD.closed = make(chan struct{})
+ clockD.pending.Add(1)
+
+ go clockD.runLoop()
+
+ return clockD
+}
+
+func (cd *ClockD) runLoop() {
+ vlog.VI(1).Infof("clockd: loop started")
+ defer vlog.VI(1).Infof("clockd: loop ended")
+ defer cd.pending.Done()
+
+ ticker := time.NewTicker(clockCheckInterval)
+ defer ticker.Stop()
+
+ var count int64 = 0
+ ntpCheckCount := ntpCheckInterval.Nanoseconds() / clockCheckInterval.Nanoseconds()
+ for {
+ // Run task first and then wait on ticker. This allows us to run the
+ // clockd tasks right away when syncbase starts regardless of the
+ // loop interval.
+ if count == 0 {
+ cd.vclock.runNtpCheck()
+ } else {
+ cd.vclock.runClockCheck()
+ }
+ count = (count + 1) % ntpCheckCount
+ vlog.VI(5).Infof("clockd: round of clockD run finished")
+
+ select {
+ case <-cd.closed:
+ vlog.VI(1).Infof("clockd: channel closed, stop work and exit")
+ return
+
+ case <-ticker.C:
+ }
+
+ // Give priority to close event if both ticker and closed are
+ // simultaneously triggered.
+ select {
+ case <-cd.closed:
+ vlog.VI(1).Info("clockd: channel closed, stop work and exit")
+ return
+
+ default:
+ }
+ }
+}
+
+// Close cleans up clock state.
+// TODO(jlodhia): Hook it up to server shutdown of syncbased.
+func (cd *ClockD) Close() {
+ close(cd.closed)
+ cd.pending.Wait()
+}
diff --git a/services/syncbase/clock/clockservice.go b/services/syncbase/clock/clockservice.go
index 575f83a..6343d1e 100644
--- a/services/syncbase/clock/clockservice.go
+++ b/services/syncbase/clock/clockservice.go
@@ -4,77 +4,102 @@
package clock
+// This file contains code related to checking current system clock to see
+// if it has been changed by any external action.
import (
"math"
"time"
- "v.io/v23/context"
"v.io/v23/verror"
"v.io/x/lib/vlog"
"v.io/x/ref/services/syncbase/server/util"
+ "v.io/x/ref/services/syncbase/store"
)
-// This file contains code related to checking current system clock to see
-// if it has been changed by any external action.
-
// runClockCheck estimates the current system time based on saved boottime
// and elapsed time since boot and checks if the system clock shows the same
// time. This involves the following steps:
-// 1) Check if system was rebooted since last run. If so update the saved
+// 1) Fetch stored ClockData. If none exists, this is the first time
+// runClockCheck has been run. Write new ClockData and exit.
+// 2) Check if system was rebooted since last run. If so update the saved
// ClockData.
-// 2) Fetch stored ClockData. If none exists, this is the first time
-// runClockCheck has been run. Write new ClockData.
// 3) Estimate current system clock time and check if the actual system clock
// agrees with the estimation. If not update the skew value appropriately.
// 4) Update saved elapsed time since boot. This is used to check if the system
// was rebooted or not. TODO(jlodhia): work with device manager to provide a
// way to notify syncbase if the system was just rebooted.
-func (c *VClock) runClockCheck(ctx *context.T) {
- checkSystemRebooted(ctx, c)
+func (c *VClock) runClockCheck() {
+ vlog.VI(2).Info("clock: runClockCheck: starting clock check")
+ defer vlog.VI(2).Info("clock: runClockCheck: clock check finished")
+ err := store.RunInTransaction(c.st, func(tx store.Transaction) error {
+ return c.runClockCheckInternal(tx)
+ })
+ if err != nil {
+ vlog.Errorf("clock: runClockCheck: error while commiting: %v", err)
+ }
+}
+
+func (c *VClock) runClockCheckInternal(tx store.Transaction) error {
clockData := &ClockData{}
- if err := c.sa.GetClockData(ctx, clockData); err != nil {
+ if err := c.GetClockData(tx, clockData); err != nil {
if verror.ErrorID(err) == verror.ErrNoExist.ID {
// VClock's cron job to setup UTC time at boot is being run for the
// first time. Skew is not known, hence assigning 0.
- writeNewClockData(ctx, c, 0)
- } else {
- vlog.Errorf("Error while fetching clock data: %v", err)
+ vlog.VI(2).Info("clock: runClockCheck: no clock data found, writing new data")
+ writeNewClockData(tx, c, 0, nil)
+ return nil
}
- return
+ vlog.Errorf("clock: runClockCheck: Error while fetching clock data: %v", err)
+ return err
+ }
+ if checkSystemRebooted(tx, c, clockData) {
+ // System was rebooted. checkSystemRebooted() has already written new
+ // data for vclock.
+ return nil
}
- systemTime := c.clock.Now()
- elapsedTime, err := c.clock.ElapsedTime()
+ systemTime := c.SysClock.Now()
+ elapsedTime, err := c.SysClock.ElapsedTime()
if err != nil {
- vlog.Errorf("Error while fetching elapsed time: %v", err)
- return
+ vlog.Errorf("clock: runClockCheck: Error while fetching elapsed time: %v", err)
+ return err
}
newClockData := &ClockData{
SystemTimeAtBoot: clockData.SystemTimeAtBoot,
Skew: clockData.Skew,
ElapsedTimeSinceBoot: elapsedTime.Nanoseconds(),
+ LastNtpTs: clockData.LastNtpTs,
+ NumReboots: clockData.NumReboots,
+ NumHops: clockData.NumHops,
}
estimatedClockTime := clockData.SystemBootTime().Add(elapsedTime)
diff := estimatedClockTime.Sub(systemTime)
if math.Abs(float64(diff.Nanoseconds())) > util.LocalClockDriftThreshold {
+ vlog.VI(2).Infof("clock: runClockCheck: clock drift of %v discovered.", diff)
newClockData.Skew = newClockData.Skew + diff.Nanoseconds()
newSystemTimeAtBoot := systemTime.Add(-elapsedTime)
newClockData.SystemTimeAtBoot = newSystemTimeAtBoot.UnixNano()
}
- if err := c.sa.SetClockData(ctx, newClockData); err != nil {
- vlog.Errorf("Error while setting clock data: %v", err)
+ if err := c.SetClockData(tx, newClockData); err != nil {
+ vlog.Errorf("clock: runClockCheck: Error while setting clock data: %v", err)
+ return err
}
+ return nil
}
-func writeNewClockData(ctx *context.T, c *VClock, skew time.Duration) {
- systemTime := c.clock.Now()
- elapsedTime, err := c.clock.ElapsedTime()
+// writeNewClockData overwrites existing clock data with the given params.
+// It recreates system boot time based on current system clock and elapsed time.
+// This method takes StoreWriter instead of Transaction to allow tests to pass
+// a store.Store object instead for simplicity.
+func writeNewClockData(tx store.Transaction, c *VClock, skew time.Duration, ntpTs *time.Time) {
+ systemTime := c.SysClock.Now()
+ elapsedTime, err := c.SysClock.ElapsedTime()
if err != nil {
- vlog.Errorf("Error while fetching elapsed time: %v", err)
+ vlog.Errorf("clock: Error while fetching elapsed time: %v", err)
return
}
systemTimeAtBoot := systemTime.Add(-elapsedTime)
@@ -82,9 +107,12 @@
SystemTimeAtBoot: systemTimeAtBoot.UnixNano(),
Skew: skew.Nanoseconds(),
ElapsedTimeSinceBoot: elapsedTime.Nanoseconds(),
+ LastNtpTs: ntpTs,
+ NumReboots: 0,
+ NumHops: 0,
}
- if err := c.sa.SetClockData(ctx, clockData); err != nil {
- vlog.Errorf("Error while setting clock data: %v", err)
+ if err := c.SetClockData(tx, clockData); err != nil {
+ vlog.Errorf("clock: Error while setting clock data: %v", err)
}
}
@@ -94,21 +122,11 @@
// be changed unless a reboot happens, if the current value is lower than the
// previous value then a reboot has happened since last run. If so, update
// the boot time and elapsed time since boot appropriately.
-func checkSystemRebooted(ctx *context.T, c *VClock) bool {
- currentSysTime := c.clock.Now()
- elapsedTime, err := c.clock.ElapsedTime()
+func checkSystemRebooted(tx store.Transaction, c *VClock, clockData *ClockData) bool {
+ currentSysTime := c.SysClock.Now()
+ elapsedTime, err := c.SysClock.ElapsedTime()
if err != nil {
- vlog.Errorf("Error while fetching elapsed time: %v", err)
- return false
- }
-
- clockData := &ClockData{}
- if err := c.sa.GetClockData(ctx, clockData); err != nil {
- if verror.ErrorID(err) != verror.ErrNoExist.ID {
- vlog.Errorf("Error while fetching clock delta: %v", err)
- }
- // In case of verror.ErrNoExist no clock data present. Nothing needed to
- // be done. writeNewClockData() will write new clock data to storage.
+ vlog.Errorf("clock: runClockCheck: Error while fetching elapsed time: %v", err)
return false
}
@@ -116,10 +134,12 @@
// Since the elapsed time since last boot provided by the system is
// less than the elapsed time since boot seen the last time clockservice
// ran, the system must have rebooted in between.
+ vlog.VI(2).Info("clock: runClockCheck: system reboot discovered")
clockData.SystemTimeAtBoot = currentSysTime.Add(-elapsedTime).UnixNano()
clockData.ElapsedTimeSinceBoot = elapsedTime.Nanoseconds()
- if err := c.sa.SetClockData(ctx, clockData); err != nil {
- vlog.Errorf("Error while setting clock data: %v", err)
+ clockData.NumReboots += 1
+ if err := c.SetClockData(tx, clockData); err != nil {
+ vlog.Errorf("clock: runClockCheck: Error while setting clock data: %v", err)
}
return true
}
diff --git a/services/syncbase/clock/clockservice_test.go b/services/syncbase/clock/clockservice_test.go
index 86650c1..72e7773 100644
--- a/services/syncbase/clock/clockservice_test.go
+++ b/services/syncbase/clock/clockservice_test.go
@@ -14,49 +14,55 @@
)
func TestWriteNewClockData(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
sysClock := MockSystemClock(sysTs, time.Duration(constElapsedTime))
- stAdapter := MockStorageAdapter()
- clock := NewVClockWithMockServices(stAdapter, sysClock, nil)
+ clock := NewVClockWithMockServices(testStore.st, sysClock, nil)
- writeNewClockData(nil, clock, 0)
+ tx := clock.St().NewTransaction()
+ writeNewClockData(tx, clock, 0, nil)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
expectedSystemTimeAtBoot := sysTs.UnixNano() - constElapsedTime
- verifyClockData(t, stAdapter, 0, expectedSystemTimeAtBoot, constElapsedTime)
+ expected := newClockData(expectedSystemTimeAtBoot, 0, constElapsedTime, nil, 0, 0)
+ VerifyClockData(t, clock, expected)
}
// This test runs the following scenarios
-// 1) Run checkSystemRebooted() with no ClockData stored
-// Result: no op.
-// 2) Run checkSystemRebooted() with ClockData that has SystemTimeAtBoot higher
+// 1) Run checkSystemRebooted() with ClockData that has SystemTimeAtBoot higher
// than the current elapsed time.
// Result: A new ClockData is written with updated SystemTimeAtBoot and
// elapsed time.
-// 3) Run checkSystemRebooted() again after moving the sysClock forward
+// 2) Run checkSystemRebooted() again after moving the sysClock forward
// Result: no op.
func TestCheckSystemRebooted(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
+ ntpTs := sysTs.Add(time.Duration(-10))
sysClock := MockSystemClock(sysTs, time.Duration(constElapsedTime))
- stAdapter := MockStorageAdapter()
-
- clock := NewVClockWithMockServices(stAdapter, sysClock, nil)
-
- // stAdapter will return ErrNoExist while fetching ClockData
- // checkSystemRebooted should return false.
- if checkSystemRebooted(nil, clock) {
- t.Error("Unexpected return value")
- }
+ clock := NewVClockWithMockServices(testStore.st, sysClock, nil)
// Set clock data with elapsed time greater than constElapsedTime
- clockData := &ClockData{25003, 25, 34569}
- stAdapter.SetClockData(nil, clockData)
-
- if !checkSystemRebooted(nil, clock) {
+ clockData := newClockData(25003, 25, 34569, &ntpTs, 1, 1)
+ tx := clock.St().NewTransaction()
+ clock.SetClockData(tx, clockData)
+ if !checkSystemRebooted(tx, clock, clockData) {
t.Error("Unexpected return value")
}
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
expectedSystemTimeAtBoot := sysTs.UnixNano() - constElapsedTime
- verifyClockData(t, stAdapter, 25, expectedSystemTimeAtBoot, constElapsedTime)
+ expected := newClockData(expectedSystemTimeAtBoot, 25, constElapsedTime, &ntpTs, 2 /*NumReboots incremented*/, 1)
+ VerifyClockData(t, clock, expected)
// move clock forward without reboot and run checkSystemRebooted again
var timePassed int64 = 200
@@ -64,25 +70,33 @@
sysClock.SetNow(newSysTs)
sysClock.SetElapsedTime(time.Duration(constElapsedTime + timePassed))
- if checkSystemRebooted(nil, clock) {
+ tx = clock.St().NewTransaction()
+ if checkSystemRebooted(tx, clock, clockData) {
t.Error("Unexpected return value")
}
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
expectedSystemTimeAtBoot = sysTs.UnixNano() - constElapsedTime
- verifyClockData(t, stAdapter, 25, expectedSystemTimeAtBoot, constElapsedTime)
+ expected = newClockData(expectedSystemTimeAtBoot, 25, constElapsedTime, &ntpTs, 2, 1)
+ VerifyClockData(t, clock, expected)
}
// Setup: No prior ClockData present.
// Result: A new ClockData value gets set.
func TestRunClockCheck1(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
sysClock := MockSystemClock(sysTs, time.Duration(constElapsedTime))
- stAdapter := MockStorageAdapter()
+ clock := NewVClockWithMockServices(testStore.st, sysClock, nil)
- clock := NewVClockWithMockServices(stAdapter, sysClock, nil)
-
- clock.runClockCheck(nil)
+ clock.runClockCheck()
expectedSystemTimeAtBoot := sysTs.UnixNano() - constElapsedTime
- verifyClockData(t, stAdapter, 0, expectedSystemTimeAtBoot, constElapsedTime)
+ expected := newClockData(expectedSystemTimeAtBoot, 0, constElapsedTime, nil, 0, 0)
+ VerifyClockData(t, clock, expected)
}
// Setup: ClockData present, system clock elapsed time is lower than whats
@@ -90,34 +104,49 @@
// Result: A new ClockData value gets set with new system boot time and elapsed
// time, skew remains the same.
func TestRunClockCheck2(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
+ ntpTs := sysTs.Add(time.Duration(-10))
sysClock := MockSystemClock(sysTs, time.Duration(constElapsedTime))
- stAdapter := MockStorageAdapter()
+ clock := NewVClockWithMockServices(testStore.st, sysClock, nil)
+
// Set clock data with elapsed time greater than constElapsedTime
- clockData := &ClockData{25003, 25, 34569}
- stAdapter.SetClockData(nil, clockData)
+ clockData := newClockData(25003, 25, 34569, &ntpTs, 1, 1)
+ tx := clock.St().NewTransaction()
+ clock.SetClockData(tx, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
- clock := NewVClockWithMockServices(stAdapter, sysClock, nil)
-
- clock.runClockCheck(nil)
+ clock.runClockCheck()
expectedSystemTimeAtBoot := sysTs.UnixNano() - constElapsedTime
- verifyClockData(t, stAdapter, 25, expectedSystemTimeAtBoot, constElapsedTime)
+ expected := newClockData(expectedSystemTimeAtBoot, 25, constElapsedTime, &ntpTs, 2 /*NumReboots incremented*/, 1)
+ VerifyClockData(t, clock, expected)
}
// Setup: ClockData present, system clock gets a skew of 10 seconds
// Result: A new ClockData value gets set with new elapsed time and skew,
// system boot time remains the same.
func TestRunClockCheck3(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
+ ntpTs := sysTs.Add(time.Duration(-20))
sysClock := MockSystemClock(sysTs, time.Duration(constElapsedTime))
- stAdapter := MockStorageAdapter()
+ clock := NewVClockWithMockServices(testStore.st, sysClock, nil)
bootTs := sysTs.Add(time.Duration(-constElapsedTime))
oldSkew := 25 * time.Second
- clockData := &ClockData{bootTs.UnixNano(), oldSkew.Nanoseconds(), 40}
- stAdapter.SetClockData(nil, clockData)
+ clockData := newClockData(bootTs.UnixNano(), oldSkew.Nanoseconds(), 40, &ntpTs, 1, 1)
- clock := NewVClockWithMockServices(stAdapter, sysClock, nil)
+ tx := clock.St().NewTransaction()
+ clock.SetClockData(tx, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
// introduce a change in sys clock
extraSkew := 10 * time.Second // moves clock closer to UTC
@@ -125,35 +154,47 @@
sysClock.SetNow(changedSysTs)
newSkew := 15 * time.Second
- clock.runClockCheck(nil)
+ clock.runClockCheck()
expectedSystemTimeAtBoot := bootTs.UnixNano() + extraSkew.Nanoseconds()
- verifyClockData(t, stAdapter, newSkew.Nanoseconds(), expectedSystemTimeAtBoot, constElapsedTime)
+ expected := newClockData(expectedSystemTimeAtBoot, newSkew.Nanoseconds(), constElapsedTime, &ntpTs, 1, 1)
+ VerifyClockData(t, clock, expected)
}
func TestWithRealSysClock(t *testing.T) {
- stAdapter := MockStorageAdapter()
- clock := NewVClockWithMockServices(stAdapter, nil, nil)
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
- writeNewClockData(nil, clock, 0)
+ clock := NewVClockWithMockServices(testStore.st, nil, nil)
+ ntpTs := time.Now().Add(time.Duration(-5000))
+
+ tx := clock.St().NewTransaction()
+ writeNewClockData(tx, clock, 0, &ntpTs)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
// Verify if clock data was written to StorageAdapter
clockData := &ClockData{}
- if err := stAdapter.GetClockData(nil, clockData); err != nil {
+ if err := clock.GetClockData(clock.St(), clockData); err != nil {
t.Errorf("Expected to find clockData, received error: %v", err)
}
// Verify that calling checkSystemRebooted() does nothing
- if checkSystemRebooted(nil, clock) {
+ tx = clock.St().NewTransaction()
+ if checkSystemRebooted(tx, clock, clockData) {
t.Error("Unexpected return value")
}
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
// sleep for 1 second more than the skew threshold
time.Sleep(1800 * time.Millisecond)
// Verify that calling runClockCheck() only updates elapsed time
- clock.runClockCheck(nil)
+ clock.runClockCheck()
newClockData := &ClockData{}
- if err := stAdapter.GetClockData(nil, newClockData); err != nil {
+ if err := clock.GetClockData(clock.St(), newClockData); err != nil {
t.Errorf("Expected to find clockData, received error: %v", err)
}
if newClockData.Skew != clockData.Skew {
@@ -167,25 +208,8 @@
t.Errorf("SystemTimeAtBoot expected: %d, found: %d",
clockData.SystemTimeAtBoot, newClockData.SystemTimeAtBoot)
}
-}
-
-func verifyClockData(t *testing.T, stAdapter StorageAdapter, skew int64,
- sysTimeAtBoot int64, elapsedTime int64) {
- // verify ClockData
- clockData := &ClockData{}
- if err := stAdapter.GetClockData(nil, clockData); err != nil {
- t.Errorf("Expected to find clockData, found error: %v", err)
- }
-
- if clockData.Skew != skew {
- t.Errorf("Expected value for skew: %d, found: %d", skew, clockData.Skew)
- }
- if clockData.ElapsedTimeSinceBoot != elapsedTime {
- t.Errorf("Expected value for elapsed time: %d, found: %d", elapsedTime,
- clockData.ElapsedTimeSinceBoot)
- }
- if clockData.SystemTimeAtBoot != sysTimeAtBoot {
- t.Errorf("Expected value for SystemTimeAtBoot: %d, found: %d",
- sysTimeAtBoot, clockData.SystemTimeAtBoot)
+ if !timeEquals(clockData.LastNtpTs, newClockData.LastNtpTs) {
+ t.Errorf("Expected value for LastNtpTs: %v, found: %v",
+ fmtTime(clockData.LastNtpTs), fmtTime(newClockData.LastNtpTs))
}
}
diff --git a/services/syncbase/clock/ntp.go b/services/syncbase/clock/ntp.go
index 090cc96..6f1d99e 100644
--- a/services/syncbase/clock/ntp.go
+++ b/services/syncbase/clock/ntp.go
@@ -9,6 +9,7 @@
"net"
"time"
+ "v.io/x/lib/vlog"
"v.io/x/ref/services/syncbase/server/util"
)
@@ -42,7 +43,7 @@
}
}
if canonicalSample == nil {
- err := fmt.Errorf("Failed to get any sample from NTP server: %s", ns.ntpHost)
+ err := fmt.Errorf("clock: NtpSync: Failed to get any sample from NTP server: %s", ns.ntpHost)
return nil, err
}
return canonicalSample, nil
@@ -72,6 +73,18 @@
}
defer con.Close()
+ // To make sure that the system clock does not change between fetching
+ // send and receive timestamps, we get the elapsed time since
+ // boot (which is immutable) before registering the send timestamp and
+ // after registering the receive timestamp and call HasSysClockChanged()
+ // to verify if the clock changed in between or not. If it did, we return
+ // ErrInternal as response.
+ elapsedOrig, err := ns.sc.ElapsedTime()
+ if err != nil {
+ vlog.Errorf("clock: NtpSync: error while fetching elapsed time: %v", err)
+ return nil, err
+ }
+
msg := ns.createRequest()
_, err = con.Write(msg)
if err != nil {
@@ -85,17 +98,29 @@
}
clientReceiveTs := ns.sc.Now()
+ elapsedEnd, err := ns.sc.ElapsedTime()
+ if err != nil {
+ vlog.Errorf("clock: NtpSync: error while fetching elapsed time: %v", err)
+ return nil, err
+ }
+
clientTransmitTs := extractTime(msg[24:32])
serverReceiveTs := extractTime(msg[32:40])
serverTransmitTs := extractTime(msg[40:48])
+ if HasSysClockChanged(clientTransmitTs, clientReceiveTs, elapsedOrig, elapsedEnd) {
+ err := fmt.Errorf("clock: NtpSync: system clock changed midway through syncing wih NTP.")
+ vlog.Errorf("%v", err)
+ return nil, err
+ }
+
// Following code extracts the clock offset and network delay based on the
// transmit and receive timestamps on the client and the server as per
// the formula explained at http://www.eecis.udel.edu/~mills/time.html
data := NtpData{}
data.offset = (serverReceiveTs.Sub(clientTransmitTs) + serverTransmitTs.Sub(clientReceiveTs)) / 2
data.delay = clientReceiveTs.Sub(clientTransmitTs) - serverTransmitTs.Sub(serverReceiveTs)
-
+ data.ntpTs = serverTransmitTs
return &data, nil
}
diff --git a/services/syncbase/clock/ntpservice.go b/services/syncbase/clock/ntpservice.go
index 7bf47c1..593e913 100644
--- a/services/syncbase/clock/ntpservice.go
+++ b/services/syncbase/clock/ntpservice.go
@@ -6,41 +6,62 @@
import (
"math"
+ "time"
- "v.io/v23/context"
"v.io/v23/verror"
"v.io/x/lib/vlog"
"v.io/x/ref/services/syncbase/server/util"
+ "v.io/x/ref/services/syncbase/store"
)
// runNtpCheck talks to an NTP server, fetches the current UTC time from it
// and corrects VClock time.
-func (c *VClock) runNtpCheck(ctx *context.T) error {
+func (c *VClock) runNtpCheck() {
+ vlog.VI(2).Info("clock: runNtpCheck: starting ntp check")
+ defer vlog.VI(2).Info("clock: runNtpCheck: ntp check finished")
+
+ err := store.RunInTransaction(c.st, func(tx store.Transaction) error {
+ return c.runNtpCheckInternal(tx)
+ })
+ if err != nil {
+ vlog.Errorf("clock: runNtpCheck: ntp check run failed: %v", err)
+ }
+}
+
+func (c *VClock) runNtpCheckInternal(tx store.Transaction) error {
ntpData, err := c.ntpSource.NtpSync(util.NtpSampleCount)
if err != nil {
- vlog.Errorf("Error while fetching ntp time: %v", err)
+ vlog.Errorf("clock: runNtpCheck: Error while fetching ntp time: %v", err)
return err
}
offset := ntpData.offset
+ vlog.VI(2).Infof("clock: runNtpCheck: offset is %v", offset)
data := &ClockData{}
- if err := c.sa.GetClockData(ctx, data); err != nil {
- if verror.ErrorID(err) == verror.ErrNoExist.ID {
- // No ClockData found, write a new one.
- writeNewClockData(ctx, c, offset)
- return nil
+ if err := c.GetClockData(tx, data); err != nil {
+ if verror.ErrorID(err) != verror.ErrNoExist.ID {
+ vlog.Infof("clock: runNtpCheck: Error while fetching clock data: %v", err)
+ vlog.Info("clock: runNtpCheck: Overwriting clock data with NTP")
}
- vlog.Info("Error while fetching clock data: %v", err)
- vlog.Info("Overwriting clock data with NTP")
- writeNewClockData(ctx, c, offset)
+ // If no ClockData found or error while reading, write a new one.
+ writeNewClockData(tx, c, offset, &ntpData.ntpTs)
return nil
}
// Update clock skew if the difference between offset and skew is larger
// than NtpDiffThreshold. NtpDiffThreshold helps avoid constant tweaking of
// the syncbase clock.
- if math.Abs(float64(offset.Nanoseconds()-data.Skew)) > util.NtpDiffThreshold {
- writeNewClockData(ctx, c, offset)
+ deviation := math.Abs(float64(offset.Nanoseconds() - data.Skew))
+ if deviation > util.NtpDiffThreshold {
+ vlog.VI(2).Infof("clock: runNtpCheck: clock found deviating from ntp by %v nsec. Updating clock.", deviation)
+ writeNewClockData(tx, c, offset, &ntpData.ntpTs)
+ return nil
}
+
+ // Update clock data with new LastNtpTs. We dont update the skew in db with
+ // the latest offset because there will always be minor errors introduced
+ // by asymmetrical network delays in measuring the offset leading to constant
+ // change in clock skew.
+ writeNewClockData(tx, c, time.Duration(data.Skew), &ntpData.ntpTs)
return nil
}
diff --git a/services/syncbase/clock/ntpservice_test.go b/services/syncbase/clock/ntpservice_test.go
index e505e01..8bbc93f 100644
--- a/services/syncbase/clock/ntpservice_test.go
+++ b/services/syncbase/clock/ntpservice_test.go
@@ -8,174 +8,179 @@
"net"
"testing"
"time"
+
+ "v.io/v23/verror"
)
func TestWithMockNtpForErr(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysClock := MockSystemClock(time.Now(), 0)
- stAdapter := MockStorageAdapter()
ntpSource := MockNtpSource()
ntpSource.Err = net.UnknownNetworkError("network err")
+ vclock := NewVClockWithMockServices(testStore.st, sysClock, ntpSource)
- vclock := NewVClockWithMockServices(stAdapter, sysClock, ntpSource)
-
- if err := vclock.runNtpCheck(nil); err == nil {
- t.Error("Network error expected but not found")
- }
-
- if stAdapter.clockData != nil {
- t.Error("Non-nil clock data found.")
+ vclock.runNtpCheck()
+ clockData := &ClockData{}
+ if err := vclock.GetClockData(vclock.St(), clockData); verror.ErrorID(err) != verror.ErrNoExist.ID {
+ t.Errorf("Non-nil clock data found: %v", clockData)
}
}
func TestWithMockNtpForDiffBelowThreshold(t *testing.T) {
- sysClock := MockSystemClock(time.Now(), 0) // not used
- stAdapter := MockStorageAdapter()
- originalData := NewClockData(0)
- stAdapter.SetClockData(nil, &originalData)
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
+ sysTs := time.Now()
+ elapsedTime := time.Duration(50)
+ var skew int64 = 0
+ sysClock := MockSystemClock(sysTs, elapsedTime)
+ originalData := NewClockData(skew)
ntpSource := MockNtpSource()
offset := 1800 * time.Millisecond // error threshold is 2 seconds
- ntpSource.Data = &NtpData{offset: offset, delay: 5 * time.Millisecond}
+ ntpSource.Data = &NtpData{
+ offset: offset,
+ delay: 5 * time.Millisecond,
+ ntpTs: sysTs.Add(offset),
+ }
- vclock := NewVClockWithMockServices(stAdapter, sysClock, ntpSource)
- if err := vclock.runNtpCheck(nil); err != nil {
- t.Errorf("Unexpected err: %v", err)
+ vclock := NewVClockWithMockServices(testStore.st, sysClock, ntpSource)
+ tx := vclock.St().NewTransaction()
+ vclock.SetClockData(tx, &originalData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
}
- if isClockDataChanged(stAdapter, &originalData) {
- t.Error("ClockData expected to be unchanged but found updated")
- }
+
+ vclock.runNtpCheck()
+ expectedSystemTimeAtBoot := sysTs.UnixNano() - elapsedTime.Nanoseconds()
+ expected := newClockData(expectedSystemTimeAtBoot, skew, elapsedTime.Nanoseconds(), &ntpSource.Data.ntpTs, 0, 0)
+ VerifyClockData(t, vclock, expected)
}
func TestWithMockNtpForDiffAboveThreshold(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
elapsedTime := 10 * time.Minute
sysClock := MockSystemClock(sysTs, elapsedTime)
-
- stAdapter := MockStorageAdapter()
originalData := NewClockData(0)
- stAdapter.SetClockData(nil, &originalData)
ntpSource := MockNtpSource()
skew := 2100 * time.Millisecond // error threshold is 2 seconds
- ntpSource.Data = &NtpData{offset: skew, delay: 5 * time.Millisecond}
+ ntpSource.Data = &NtpData{
+ offset: skew,
+ delay: 5 * time.Millisecond,
+ ntpTs: sysTs.Add(skew),
+ }
- vclock := NewVClockWithMockServices(stAdapter, sysClock, ntpSource)
- if err := vclock.runNtpCheck(nil); err != nil {
- t.Errorf("Unexpected err: %v", err)
+ vclock := NewVClockWithMockServices(testStore.st, sysClock, ntpSource)
+ tx := vclock.St().NewTransaction()
+ vclock.SetClockData(tx, &originalData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
}
- if !isClockDataChanged(stAdapter, &originalData) {
- t.Error("ClockData expected to be updated but found unchanged")
- }
+
+ vclock.runNtpCheck()
expectedBootTime := sysTs.Add(-elapsedTime).UnixNano()
- if stAdapter.clockData.Skew != skew.Nanoseconds() {
- t.Errorf("Skew expected to be %d but found %d",
- skew.Nanoseconds(), stAdapter.clockData.Skew)
- }
- if stAdapter.clockData.ElapsedTimeSinceBoot != elapsedTime.Nanoseconds() {
- t.Errorf("ElapsedTime expected to be %d but found %d",
- elapsedTime.Nanoseconds(), stAdapter.clockData.ElapsedTimeSinceBoot)
- }
- if stAdapter.clockData.SystemTimeAtBoot != expectedBootTime {
- t.Errorf("Skew expected to be %d but found %d",
- expectedBootTime, stAdapter.clockData.SystemTimeAtBoot)
- }
+ expected := newClockData(expectedBootTime, skew.Nanoseconds(), elapsedTime.Nanoseconds(), &ntpSource.Data.ntpTs, 0, 0)
+ VerifyClockData(t, vclock, expected)
}
func TestWithMockNtpForDiffBelowThresholdAndExistingLargeSkew(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
elapsedTime := 10 * time.Minute
sysClock := MockSystemClock(sysTs, elapsedTime)
- stAdapter := MockStorageAdapter()
originalData := NewClockData(2300 * time.Millisecond.Nanoseconds()) // large skew
- stAdapter.SetClockData(nil, &originalData)
ntpSource := MockNtpSource()
skew := 200 * time.Millisecond // error threshold is 2 seconds
- ntpSource.Data = &NtpData{offset: skew, delay: 5 * time.Millisecond}
+ ntpSource.Data = &NtpData{
+ offset: skew,
+ delay: 5 * time.Millisecond,
+ ntpTs: sysTs.Add(skew),
+ }
- vclock := NewVClockWithMockServices(stAdapter, sysClock, ntpSource)
- if err := vclock.runNtpCheck(nil); err != nil {
- t.Errorf("Unexpected err: %v", err)
+ vclock := NewVClockWithMockServices(testStore.st, sysClock, ntpSource)
+ tx := vclock.St().NewTransaction()
+ vclock.SetClockData(tx, &originalData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
}
- if !isClockDataChanged(stAdapter, &originalData) {
- t.Error("ClockData expected to be updated but found unchanged")
- }
+
+ vclock.runNtpCheck()
expectedBootTime := sysTs.Add(-elapsedTime).UnixNano()
- if stAdapter.clockData.Skew != skew.Nanoseconds() {
- t.Errorf("Skew expected to be %d but found %d",
- skew.Nanoseconds(), stAdapter.clockData.Skew)
- }
- if stAdapter.clockData.ElapsedTimeSinceBoot != elapsedTime.Nanoseconds() {
- t.Errorf("ElapsedTime expected to be %d but found %d",
- elapsedTime.Nanoseconds(), stAdapter.clockData.ElapsedTimeSinceBoot)
- }
- if stAdapter.clockData.SystemTimeAtBoot != expectedBootTime {
- t.Errorf("Skew expected to be %d but found %d",
- expectedBootTime, stAdapter.clockData.SystemTimeAtBoot)
- }
+ expected := newClockData(expectedBootTime, skew.Nanoseconds(), elapsedTime.Nanoseconds(), &ntpSource.Data.ntpTs, 0, 0)
+ VerifyClockData(t, vclock, expected)
}
func TestWithMockNtpForDiffBelowThresholdWithNoStoredClockData(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
elapsedTime := 10 * time.Minute
sysClock := MockSystemClock(sysTs, elapsedTime)
- stAdapter := MockStorageAdapter() // no skew data stored
-
ntpSource := MockNtpSource()
skew := 200 * time.Millisecond // error threshold is 2 seconds
- ntpSource.Data = &NtpData{offset: skew, delay: 5 * time.Millisecond}
+ ntpSource.Data = &NtpData{
+ offset: skew,
+ delay: 5 * time.Millisecond,
+ ntpTs: sysTs.Add(skew),
+ }
- vclock := NewVClockWithMockServices(stAdapter, sysClock, ntpSource)
- if err := vclock.runNtpCheck(nil); err != nil {
- t.Errorf("Unexpected err: %v", err)
- }
- if !isClockDataChanged(stAdapter, nil) {
- t.Error("ClockData expected to be updated but found unchanged")
- }
+ vclock := NewVClockWithMockServices(testStore.st, sysClock, ntpSource)
+ // no skew data stored
+ vclock.runNtpCheck()
expectedBootTime := sysTs.Add(-elapsedTime).UnixNano()
- if stAdapter.clockData.Skew != skew.Nanoseconds() {
- t.Errorf("Skew expected to be %d but found %d",
- skew.Nanoseconds(), stAdapter.clockData.Skew)
- }
- if stAdapter.clockData.ElapsedTimeSinceBoot != elapsedTime.Nanoseconds() {
- t.Errorf("ElapsedTime expected to be %d but found %d",
- elapsedTime.Nanoseconds(), stAdapter.clockData.ElapsedTimeSinceBoot)
- }
- if stAdapter.clockData.SystemTimeAtBoot != expectedBootTime {
- t.Errorf("Skew expected to be %d but found %d",
- expectedBootTime, stAdapter.clockData.SystemTimeAtBoot)
- }
+ expected := newClockData(expectedBootTime, skew.Nanoseconds(), elapsedTime.Nanoseconds(), &ntpSource.Data.ntpTs, 0, 0)
+ VerifyClockData(t, vclock, expected)
}
/*
-Following two tests are commented out as they hit the real NTP servers
-and can resut into being flaky if the clock of the machine running continuous
-test has a skew more than 2 seconds.
+// Following two tests are commented out as they hit the real NTP servers
+// and can resut into being flaky if the clock of the machine running continuous
+// test has a skew more than 2 seconds.
func TestWithRealNtp(t *testing.T) {
- stAdapter := MockStorageAdapter()
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
originalData := NewClockData(100 * time.Millisecond.Nanoseconds()) // small skew
- stAdapter.SetClockData(nil, &originalData)
- vclock := NewVClockWithMockServices(stAdapter, nil, nil)
- if err := vclock.runNtpCheck(nil); err != nil {
- t.Errorf("Unexpected err: %v", err)
+ vclock := NewVClockWithMockServices(testStore.st, nil, nil)
+ tx := vclock.St().NewTransaction()
+ vclock.SetClockData(tx, &originalData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
}
- if isClockDataChanged(stAdapter, &originalData) {
- t.Error("ClockData expected to be unchanged but found updated")
+ vclock.runNtpCheck()
+
+ clockData := ClockData{}
+ if err := vclock.GetClockData(vclock.St(), &clockData); err != nil {
+ t.Errorf("error looking up clock data: %v", err)
}
+ fmt.Printf("\nClockData old: %#v, new : %#v", originalData, clockData)
}
func TestWithRealNtpForNoClockData(t *testing.T) {
- stAdapter := MockStorageAdapter()
- vclock := NewVClockWithMockServices(stAdapter, nil, nil)
- if err := vclock.runNtpCheck(nil); err != nil {
- t.Errorf("Unexpected err: %v", err)
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
+ vclock := NewVClockWithMockServices(testStore.st, nil, nil)
+ vclock.runNtpCheck()
+
+ clockData := ClockData{}
+ if err := vclock.GetClockData(vclock.St(), &clockData); err != nil {
+ t.Errorf("error looking up clock data: %v", err)
}
- if !isClockDataChanged(stAdapter, nil) {
- t.Error("ClockData expected to be updated but found unchanged")
- }
+ fmt.Printf("\nClockData: %#v", clockData)
}
*/
@@ -186,7 +191,3 @@
ElapsedTimeSinceBoot: 0,
}
}
-
-func isClockDataChanged(stAdapter *storageAdapterMockImpl, originalData *ClockData) bool {
- return stAdapter.clockData != originalData // check for same pointer
-}
diff --git a/services/syncbase/clock/storage_adapter.go b/services/syncbase/clock/storage_adapter.go
deleted file mode 100644
index 11f6d6b..0000000
--- a/services/syncbase/clock/storage_adapter.go
+++ /dev/null
@@ -1,33 +0,0 @@
-// 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 clock
-
-import (
- "v.io/v23/context"
- "v.io/x/ref/services/syncbase/server/util"
- "v.io/x/ref/services/syncbase/store"
-)
-
-var _ StorageAdapter = (*storageAdapterImpl)(nil)
-
-func NewStorageAdapter(st store.Store) StorageAdapter {
- return &storageAdapterImpl{st}
-}
-
-type storageAdapterImpl struct {
- st store.Store
-}
-
-func (sa *storageAdapterImpl) GetClockData(ctx *context.T, data *ClockData) error {
- return util.Get(ctx, sa.st, clockDataKey(), data)
-}
-
-func (sa *storageAdapterImpl) SetClockData(ctx *context.T, data *ClockData) error {
- return util.Put(ctx, sa.st, clockDataKey(), data)
-}
-
-func clockDataKey() string {
- return util.ClockPrefix
-}
diff --git a/services/syncbase/clock/syncservice.go b/services/syncbase/clock/syncservice.go
new file mode 100644
index 0000000..ece678a
--- /dev/null
+++ b/services/syncbase/clock/syncservice.go
@@ -0,0 +1,96 @@
+// 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 clock
+
+import (
+ "math"
+ "time"
+
+ "v.io/x/lib/vlog"
+ "v.io/x/ref/services/syncbase/server/util"
+ "v.io/x/ref/services/syncbase/store"
+)
+
+// ProcessPeerClockData accepts PeerSyncData and local ClockData and updates
+// local ClockData and persists it if peer's clock is more accurate. Peer's
+// clock is more accurate if all of the following conditions are met:
+// 1) diff between the two clocks is > 2 seconds (avoids constant tweaking).
+// 2) peer synced with NTP and it was more recent than local.
+// 3) peer has not rebooted since NTP or the peer has rebooted since NTP but
+// the difference btw the two clocks is < 1 minute.
+// 4) num hops for peer's NTP sync data is < 2, i.e. either the peer has
+// synced with NTP itself or it synced with another peer that did NTP.
+func (c *VClock) ProcessPeerClockData(tx store.Transaction, resp *PeerSyncData, localData *ClockData) error {
+ offset := (resp.RecvTs.Sub(resp.MySendTs) + resp.SendTs.Sub(resp.MyRecvTs)) / 2
+ vlog.VI(2).Infof("clock: ProcessPeerClockData: offset between two clocks: %v", offset)
+ if math.Abs(float64(offset.Nanoseconds())) <= util.PeerSyncDiffThreshold {
+ vlog.VI(2).Info("clock: ProcessPeerClockData: the two clocks are synced within PeerSyncDiffThreshold.")
+ return nil
+ }
+ if resp.LastNtpTs == nil {
+ vlog.VI(2).Info("clock: ProcessPeerClockData: peer clock has not synced to NTP. Ignoring peer's clock.")
+ return nil
+ }
+ if (localData != nil) && !isPeerNtpSyncMoreRecent(localData.LastNtpTs, resp.LastNtpTs) {
+ vlog.VI(2).Info("clock: ProcessPeerClockData: peer NTP sync is less recent than local.")
+ return nil
+ }
+ if isOverRebootTolerance(offset, resp.NumReboots) {
+ vlog.VI(2).Info("clock: ProcessPeerClockData: peer clock is over reboot tolerance.")
+ return nil
+ }
+ if resp.NumHops >= util.HopTolerance {
+ vlog.VI(2).Info("clock: ProcessPeerClockData: peer clock is over hop tolerance.")
+ return nil
+ }
+ vlog.VI(2).Info("clock: ProcessPeerClockData: peer's clock is more accurate than local clock. Syncing to peer's clock.")
+ c.updateClockData(tx, resp, offset, localData)
+ return nil
+}
+
+func (c *VClock) updateClockData(tx store.Transaction, peerResp *PeerSyncData, offset time.Duration, localData *ClockData) {
+ systemTime := c.SysClock.Now()
+ elapsedTime, err := c.SysClock.ElapsedTime()
+ if err != nil {
+ vlog.Errorf("clock: ProcessPeerClockData: error while fetching elapsed time: %v", err)
+ return
+ }
+ systemTimeAtBoot := systemTime.Add(-elapsedTime)
+ var skew time.Duration
+ if localData == nil {
+ // localData nil means that the local estimated UTC time is equal to
+ // the system time. Hence the offset was in reality between the local
+ // system clock and the remote estimated UTC time.
+ skew = offset
+ } else {
+ skew = time.Duration(localData.Skew) + offset
+ }
+
+ newClockData := &ClockData{
+ SystemTimeAtBoot: systemTimeAtBoot.UnixNano(),
+ Skew: skew.Nanoseconds(),
+ ElapsedTimeSinceBoot: elapsedTime.Nanoseconds(),
+ LastNtpTs: peerResp.LastNtpTs,
+ NumReboots: peerResp.NumReboots,
+ NumHops: (peerResp.NumHops + 1),
+ }
+ if err := c.SetClockData(tx, newClockData); err != nil {
+ vlog.Errorf("clock: ProcessPeerClockData: error while setting new clock data: %v", err)
+ }
+}
+
+func isPeerNtpSyncMoreRecent(localNtpTs, peerNtpTs *time.Time) bool {
+ if localNtpTs == nil {
+ return true
+ }
+ if peerNtpTs == nil {
+ return false
+ }
+ return peerNtpTs.After(*localNtpTs)
+}
+
+func isOverRebootTolerance(offset time.Duration, numReboots uint16) bool {
+ return (math.Abs(float64(offset.Nanoseconds())) > util.RebootTolerance) && (numReboots > 0)
+}
diff --git a/services/syncbase/clock/syncservice_test.go b/services/syncbase/clock/syncservice_test.go
new file mode 100644
index 0000000..6fef2b4
--- /dev/null
+++ b/services/syncbase/clock/syncservice_test.go
@@ -0,0 +1,362 @@
+// 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 clock
+
+import (
+ "testing"
+ "time"
+
+ "v.io/v23/verror"
+)
+
+// ** Test Setup **
+// Local: No ClockData
+// Remote: Has ClockData but no NTP info
+// Result: Clock not synced
+func TestClockCheckLocalClock1(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ var remoteDelta time.Duration = 5 * time.Second
+ resp := createResp(remoteDelta, nil, 0, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, nil)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ err := vclock.GetClockData(vclock.St(), &ClockData{})
+ if verror.ErrorID(err) != verror.ErrNoExist.ID {
+ t.Errorf("Expected ErrNoExists but got: %v", err)
+ }
+}
+
+// ** Test Setup **
+// Local: No ClockData
+// Remote: Has ClockData with NTP info and acceptable {reboot,hop}
+// Result: Clock synced
+func TestClockCheckLocalClock2(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ var remoteDelta time.Duration = 5 * time.Second
+ remoteNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ resp := createResp(remoteDelta, &remoteNtpTs, 0, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, nil)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ verifyPartialClockData(t, vclock, remoteDelta, remoteNtpTs, 0, 1)
+}
+
+// ** Test Setup **
+// Local: Has ClockData but no NTP info
+// Remote: Has ClockData but no NTP info
+// Result: Clock not synced
+func TestClockCheckLocalClock3(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 3 * time.Second
+ clockData := newSyncClockData(localSkew.Nanoseconds(), nil, 0, 0)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 5 * time.Second
+ resp := createResp(remoteDelta, nil, 0, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ VerifyClockData(t, vclock, clockData)
+}
+
+// ** Test Setup **
+// Local: Has ClockData but no NTP info
+// Remote: Has ClockData with NTP info and acceptable {reboot,hop}
+// Result: Clock synced
+func TestClockCheckLocalClock4(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 3 * time.Second
+ clockData := newSyncClockData(localSkew.Nanoseconds(), nil, 0, 0)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 5 * time.Second
+ remoteNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ resp := createResp(remoteDelta, &remoteNtpTs, 0, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ verifyPartialClockData(t, vclock, remoteDelta+localSkew, remoteNtpTs, 0, 1)
+}
+
+// ** Test Setup **
+// Local: Has ClockData with NTP info
+// Remote: Has ClockData but no NTP info
+// Result: Clock not synced
+func TestClockCheckLocalClock5(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 3 * time.Second
+ localNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ clockData := newSyncClockData(localSkew.Nanoseconds(), &localNtpTs, 0, 0)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 5 * time.Second
+ resp := createResp(remoteDelta, nil, 0, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ VerifyClockData(t, vclock, clockData)
+}
+
+// ** Test Setup **
+// Local & Remote have ClockData with NTP info.
+// LocalNtp > RemoteNtp, acceptable values for {reboot,hop}
+// Result: Clock not synced
+func TestClockCheckLocalClock6(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 3 * time.Second
+ localNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ clockData := newSyncClockData(localSkew.Nanoseconds(), &localNtpTs, 1, 1)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 5 * time.Second
+ remoteNtpTs := localNtpTs.Add(-10 * time.Minute).UTC()
+ resp := createResp(remoteDelta, &remoteNtpTs, 0, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ VerifyClockData(t, vclock, clockData)
+}
+
+// ** Test Setup **
+// Local & Remote have ClockData with NTP info.
+// LocalNtp < RemoteNtp, Remote-reboot = 0, Remote-hop = 1
+// Result: Clock synced, skew = oldSkew + offset, numReboots = 0, numHops = 2
+func TestClockCheckLocalClock7(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 5 * time.Second
+ localNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ clockData := newSyncClockData(localSkew.Nanoseconds(), &localNtpTs, 0, 0)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 8 * time.Second
+ remoteNtpTs := localNtpTs.Add(10 * time.Minute).UTC()
+ resp := createResp(remoteDelta, &remoteNtpTs, 0, 1)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ verifyPartialClockData(t, vclock, remoteDelta+localSkew, remoteNtpTs, 0, 2)
+}
+
+// ** Test Setup **
+// Local & Remote have ClockData with NTP info.
+// LocalNtp < RemoteNtp, unacceptable value for hop
+// Result: Clock not synced
+func TestClockCheckLocalClock8(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 3 * time.Second
+ localNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ clockData := newSyncClockData(localSkew.Nanoseconds(), &localNtpTs, 1, 1)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 5 * time.Second
+ remoteNtpTs := localNtpTs.Add(10 * time.Minute).UTC()
+ resp := createResp(remoteDelta, &remoteNtpTs, 0, 2)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ VerifyClockData(t, vclock, clockData)
+}
+
+// ** Test Setup **
+// Local & Remote have ClockData with NTP info.
+// LocalNtp < RemoteNtp, Remote-reboots > 0 but Diff between two clocks < 1 minute
+// Result: Clock synced, skew = oldSkew + offset, numReboots = remote reboots, numHops = 1
+func TestClockCheckLocalClock9(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 3 * time.Second
+ localNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ clockData := newSyncClockData(localSkew.Nanoseconds(), &localNtpTs, 0, 0)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 5 * time.Second
+ remoteNtpTs := localNtpTs.Add(10 * time.Minute).UTC()
+ resp := createResp(remoteDelta, &remoteNtpTs, 3, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ verifyPartialClockData(t, vclock, remoteDelta+localSkew, remoteNtpTs, 3, 1)
+}
+
+// ** Test Setup **
+// Local & Remote have ClockData with NTP info.
+// LocalNtp < RemoteNtp, Remote-reboots > 0 and Diff between two clocks > 1 minute
+// Result: Clock not synced
+func TestClockCheckLocalClock10(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 3 * time.Second
+ localNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ clockData := newSyncClockData(localSkew.Nanoseconds(), &localNtpTs, 0, 0)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 5 * time.Hour
+ remoteNtpTs := localNtpTs.Add(10 * time.Minute).UTC()
+ resp := createResp(remoteDelta, &remoteNtpTs, 3, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ VerifyClockData(t, vclock, clockData)
+}
+
+// ** Test Setup **
+// Local & Remote have ClockData with NTP info.
+// LocalNtp < RemoteNtp, but the difference between the two clocks is too small
+// Result: Clock not synced
+func TestClockCheckLocalClock11(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+ vclock := NewVClock(testStore.st)
+
+ localSkew := 3 * time.Second
+ localNtpTs := time.Now().Add(-30 * time.Minute).UTC()
+ clockData := newSyncClockData(localSkew.Nanoseconds(), &localNtpTs, 1, 1)
+ storeClockData(t, vclock, clockData)
+
+ var remoteDelta time.Duration = 1 * time.Second
+ remoteNtpTs := localNtpTs.Add(10 * time.Minute).UTC()
+ resp := createResp(remoteDelta, &remoteNtpTs, 0, 0)
+
+ tx := vclock.St().NewTransaction()
+ vclock.ProcessPeerClockData(tx, resp, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ VerifyClockData(t, vclock, clockData)
+}
+
+func toRemoteTime(t time.Time, remoteDelta time.Duration) time.Time {
+ return t.Add(remoteDelta)
+}
+
+func toLocalTime(t time.Time, remoteDelta time.Duration) time.Time {
+ return t.Add(-remoteDelta)
+}
+
+func storeClockData(t *testing.T, c *VClock, clockData *ClockData) {
+ tx := c.St().NewTransaction()
+ if err := c.SetClockData(tx, clockData); err != nil {
+ t.Errorf("Failed to store clock data with error: %v", err)
+ }
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+}
+
+func fetchClockData(t *testing.T, c *VClock) *ClockData {
+ clockData := &ClockData{}
+ if err := c.GetClockData(c.St(), clockData); err != nil {
+ t.Errorf("Failed to store clock data with error: %v", err)
+ }
+ return clockData
+}
+
+func newSyncClockData(skew int64, ntp *time.Time, reboots, hops uint16) *ClockData {
+ return newClockData(0, skew, 0, ntp, reboots, hops)
+}
+
+func createResp(remoteDelta time.Duration, ntpTs *time.Time, reboots, hops uint16) *PeerSyncData {
+ oTs := time.Now()
+ rTs := toRemoteTime(oTs.Add(time.Second), remoteDelta)
+ sTs := rTs.Add(time.Millisecond)
+ clientRecvTs := toLocalTime(sTs, remoteDelta).Add(time.Second)
+ return &PeerSyncData{
+ MySendTs: oTs,
+ RecvTs: rTs,
+ SendTs: sTs,
+ MyRecvTs: clientRecvTs,
+ LastNtpTs: ntpTs,
+ NumReboots: reboots,
+ NumHops: hops,
+ }
+}
+
+func verifyPartialClockData(t *testing.T, c *VClock, skew time.Duration, remoteNtpTs time.Time, reboots, hops uint16) {
+ clockData := fetchClockData(t, c)
+ if time.Duration(clockData.Skew) != skew {
+ t.Errorf("Value for clock skew expected: %v, actual: %v", skew, time.Duration(clockData.Skew))
+ }
+ if clockData.LastNtpTs == nil {
+ t.Errorf("Clock data LastNtpTs should not be nil")
+ }
+ if !clockData.LastNtpTs.Equal(remoteNtpTs) {
+ t.Errorf("Clock data LastNtpTs expected: %v, actual: %v", remoteNtpTs, clockData.LastNtpTs)
+ }
+ if clockData.NumReboots != reboots {
+ t.Errorf("Clock data NumReboots expected: %v, actual: %v", reboots, clockData.NumReboots)
+ }
+ if clockData.NumHops != hops {
+ t.Errorf("Clock data NumHops expected: %v, actual: %v", hops, clockData.NumHops)
+ }
+}
diff --git a/services/syncbase/clock/test_util.go b/services/syncbase/clock/test_util.go
index 32673d1..1d1bca3 100644
--- a/services/syncbase/clock/test_util.go
+++ b/services/syncbase/clock/test_util.go
@@ -7,50 +7,17 @@
// Utilities for testing clock.
import (
+ "fmt"
+ "os"
+ "path"
+ "testing"
"time"
- "v.io/v23/context"
- "v.io/v23/verror"
+ "v.io/x/ref/services/syncbase/server/util"
+ "v.io/x/ref/services/syncbase/store"
)
/////////////////////////////////////////////////
-// Mock for StorageAdapter
-
-var _ StorageAdapter = (*storageAdapterMockImpl)(nil)
-
-func MockStorageAdapter() *storageAdapterMockImpl {
- return &storageAdapterMockImpl{}
-}
-
-type storageAdapterMockImpl struct {
- clockData *ClockData
- err error
-}
-
-func (sa *storageAdapterMockImpl) GetClockData(ctx *context.T, data *ClockData) error {
- if sa.err != nil {
- return sa.err
- }
- if sa.clockData == nil {
- return verror.NewErrNoExist(ctx)
- }
- *data = *sa.clockData
- return nil
-}
-
-func (sa *storageAdapterMockImpl) SetClockData(ctx *context.T, data *ClockData) error {
- if sa.err != nil {
- return sa.err
- }
- sa.clockData = data
- return nil
-}
-
-func (sa *storageAdapterMockImpl) SetError(err error) {
- sa.err = err
-}
-
-/////////////////////////////////////////////////
// Mock for SystemClock
var _ SystemClock = (*systemClockMockImpl)(nil)
@@ -104,7 +71,7 @@
return ns.Data, nil
}
-func NewVClockWithMockServices(sa StorageAdapter, sc SystemClock, ns NtpSource) *VClock {
+func NewVClockWithMockServices(st store.Store, sc SystemClock, ns NtpSource) *VClock {
if sc == nil {
sc = newSystemClock()
}
@@ -112,8 +79,97 @@
ns = NewNtpSource(sc)
}
return &VClock{
- clock: sc,
- sa: sa,
+ SysClock: sc,
ntpSource: ns,
+ st: st,
+ }
+}
+
+//////////////////////////////////////////////////
+// Utility functions
+
+func VerifyClockData(t *testing.T, vclock *VClock, expected *ClockData) {
+ // verify ClockData
+ clockData := &ClockData{}
+ if err := vclock.GetClockData(vclock.St(), clockData); err != nil {
+ t.Errorf("Expected to find clockData, found error: %v", err)
+ }
+
+ if clockData.Skew != expected.Skew {
+ t.Errorf("Expected value for skew: %d, found: %d", expected.Skew, clockData.Skew)
+ }
+ if clockData.ElapsedTimeSinceBoot != expected.ElapsedTimeSinceBoot {
+ t.Errorf("Expected value for elapsed time: %d, found: %d", expected.ElapsedTimeSinceBoot,
+ clockData.ElapsedTimeSinceBoot)
+ }
+ if clockData.SystemTimeAtBoot != expected.SystemTimeAtBoot {
+ t.Errorf("Expected value for SystemTimeAtBoot: %d, found: %d",
+ expected.SystemTimeAtBoot, clockData.SystemTimeAtBoot)
+ }
+ exNtp := expected.LastNtpTs
+ actNtp := clockData.LastNtpTs
+ if !timeEquals(exNtp, actNtp) {
+ t.Errorf("Expected value for LastNtpTs: %v, found: %v",
+ fmtTime(expected.LastNtpTs), fmtTime(clockData.LastNtpTs))
+ }
+ if clockData.NumReboots != expected.NumReboots {
+ t.Errorf("Expected value for NumReboots: %v, found %v",
+ expected.NumReboots, clockData.NumReboots)
+ }
+ if clockData.NumHops != expected.NumHops {
+ t.Errorf("Expected value for NumHops: %v, found %v",
+ expected.NumHops, clockData.NumHops)
+ }
+}
+
+func newClockData(sysBootTime, skew, elapsedTime int64, ntp *time.Time, reboots, hops uint16) *ClockData {
+ return &ClockData{
+ SystemTimeAtBoot: sysBootTime,
+ Skew: skew,
+ ElapsedTimeSinceBoot: elapsedTime,
+ LastNtpTs: ntp,
+ NumReboots: reboots,
+ NumHops: hops,
+ }
+}
+
+func timeEquals(expected, actual *time.Time) bool {
+ if expected == actual {
+ return true
+ }
+ if (expected == nil && actual != nil) || (expected != nil && actual == nil) {
+ return false
+ }
+ return expected.Equal(*actual)
+}
+
+func fmtTime(t *time.Time) string {
+ if t == nil {
+ return "nil"
+ }
+ return fmt.Sprintf("%v", *t)
+}
+
+type testStore struct {
+ st store.Store
+ engine string
+ dir string
+}
+
+func createStore(t *testing.T) *testStore {
+ engine := "memstore"
+ opts := util.OpenOptions{CreateIfMissing: true, ErrorIfExists: false}
+ dir := fmt.Sprintf("%s/vclock_test_%d_%d", os.TempDir(), os.Getpid(), time.Now().UnixNano())
+
+ st, err := util.OpenStore(engine, path.Join(dir, engine), opts)
+ if err != nil {
+ t.Fatalf("cannot create store %s (%s): %v", engine, dir, err)
+ }
+ return &testStore{st: st, engine: engine, dir: dir}
+}
+
+func destroyStore(t *testing.T, s *testStore) {
+ if err := util.DestroyStore(s.engine, s.dir); err != nil {
+ t.Fatalf("cannot destroy store %s (%s): %v", s.engine, s.dir, err)
}
}
diff --git a/services/syncbase/clock/types.go b/services/syncbase/clock/types.go
index 4642397..32f1fa1 100644
--- a/services/syncbase/clock/types.go
+++ b/services/syncbase/clock/types.go
@@ -6,8 +6,6 @@
import (
"time"
-
- "v.io/v23/context"
)
// This interface provides a wrapper over system clock to allow easy testing
@@ -23,11 +21,6 @@
ElapsedTime() (time.Duration, error)
}
-type StorageAdapter interface {
- GetClockData(ctx *context.T, data *ClockData) error
- SetClockData(ctx *context.T, data *ClockData) error
-}
-
type NtpSource interface {
// NtpSync obtains NtpData samples from an NTP server and returns the one
// which has the lowest network delay.
@@ -45,6 +38,31 @@
// Delay is the round trip network delay experienced while talking to NTP
// server. The smaller the delay, the more accurate the offset is.
delay time.Duration
+
+ // Timestamp from NTP server.
+ ntpTs time.Time
+}
+
+// PeerSyncData contains information about the peer's clock along with the
+// timestamps related to the roundtrip.
+type PeerSyncData struct {
+ // The send timestamp received in TimeReq from the originator.
+ MySendTs time.Time
+ // Time when the request was received by peer.
+ RecvTs time.Time
+ // Time when the response was sent by peer.
+ SendTs time.Time
+ // Time when the response was received by the originator.
+ MyRecvTs time.Time
+ // Timestamp received from NTP during last NTP sync. The last NTP sync could
+ // be done either by this device or some other device that this device
+ // synced its clock with.
+ LastNtpTs *time.Time
+ // Number of reboots since last NTP sync.
+ NumReboots uint16
+ // Number of hops between this device and the device that did the last
+ // NTP sync.
+ NumHops uint16
}
func (cd *ClockData) SystemBootTime() time.Time {
diff --git a/services/syncbase/clock/types.vdl b/services/syncbase/clock/types.vdl
index de02a5a..b4a4fa7 100644
--- a/services/syncbase/clock/types.vdl
+++ b/services/syncbase/clock/types.vdl
@@ -4,17 +4,36 @@
package clock
+import (
+ "time"
+)
+
// ClockData is the persistent state of syncbase clock used to estimate current
// NTP time and catch any unexpected changes to system clock.
type ClockData struct {
// UTC time in unix nano seconds obtained from system clock at boot.
SystemTimeAtBoot int64
- // Skew between the system clock and NTP time.
+ // Skew in nanoseconds between the system clock and NTP time.
Skew int64
- // The elapsed time since boot as last seen during a run of clockservice.
- // This is used to determine if the device rebooted since the last run of
- // clockservice.
+ // The elapsed time in nanoseconds since boot as last seen during a run of
+ // clockservice. This is used to determine if the device rebooted since the
+ // last run of clockservice.
ElapsedTimeSinceBoot int64
+
+ // Timestamp received from NTP during last NTP sync. The last NTP sync could
+ // be done either by this device or some other device that this device
+ // synced its clock with.
+ LastNtpTs ?time.Time
+
+ // Number of cumulative reboots since last NTP sync. If the LastNtpTs was
+ // received from a peer device then this field represents the number
+ // of reboots on that device since it did NTP and the number of reboots
+ // on this device since syncing with the peer device.
+ NumReboots uint16
+
+ // Number of hops between this device and the device that did the last
+ // NTP sync.
+ NumHops uint16
}
diff --git a/services/syncbase/clock/types.vdl.go b/services/syncbase/clock/types.vdl.go
index 1478e09..110dbc1 100644
--- a/services/syncbase/clock/types.vdl.go
+++ b/services/syncbase/clock/types.vdl.go
@@ -10,6 +10,10 @@
import (
// VDL system imports
"v.io/v23/vdl"
+
+ // VDL user imports
+ "time"
+ _ "v.io/v23/vdlroot/time"
)
// ClockData is the persistent state of syncbase clock used to estimate current
@@ -17,12 +21,24 @@
type ClockData struct {
// UTC time in unix nano seconds obtained from system clock at boot.
SystemTimeAtBoot int64
- // Skew between the system clock and NTP time.
+ // Skew in nanoseconds between the system clock and NTP time.
Skew int64
- // The elapsed time since boot as last seen during a run of clockservice.
- // This is used to determine if the device rebooted since the last run of
- // clockservice.
+ // The elapsed time in nanoseconds since boot as last seen during a run of
+ // clockservice. This is used to determine if the device rebooted since the
+ // last run of clockservice.
ElapsedTimeSinceBoot int64
+ // Timestamp received from NTP during last NTP sync. The last NTP sync could
+ // be done either by this device or some other device that this device
+ // synced its clock with.
+ LastNtpTs *time.Time
+ // Number of cumulative reboots since last NTP sync. If the LastNtpTs was
+ // received from a peer device then this field represents the number
+ // of reboots on that device since it did NTP and the number of reboots
+ // on this device since syncing with the peer device.
+ NumReboots uint16
+ // Number of hops between this device and the device that did the last
+ // NTP sync.
+ NumHops uint16
}
func (ClockData) __VDLReflect(struct {
diff --git a/services/syncbase/clock/util.go b/services/syncbase/clock/util.go
new file mode 100644
index 0000000..c779341
--- /dev/null
+++ b/services/syncbase/clock/util.go
@@ -0,0 +1,47 @@
+// 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 clock
+
+// Clock utility functions.
+
+import (
+ "time"
+
+ "v.io/x/lib/vlog"
+)
+
+// errorThreshold is kept at 2 seconds because there is an error margin of
+// half a second while fetching elapsed time. This means that the combined
+// error for fetching two elapsed timestamps can be up to 1 second.
+var errorThreshold time.Duration = 2 * time.Second
+
+// hasClockChanged checks if the system clock was changed between two
+// timestamp samples taken from system clock.
+// t1 is the first timestamp sampled from clock.
+// e1 is the elapsed time since boot sampled before t1 was sampled.
+// t2 is the second timestamp sampled from clock (t2 sampled after t1)
+// e2 is the elapsed time since boot sampled after t2 was sampled.
+// Note: e1 must be sampled before t1 and e2 must be sampled after t2.
+func HasSysClockChanged(t1, t2 time.Time, e1, e2 time.Duration) bool {
+ if t2.Before(t1) {
+ vlog.VI(2).Infof("clock: hasSysClockChanged: t2 is before t1, returning true")
+ return true
+ }
+
+ tsDiff := t2.Sub(t1)
+ elapsedDiff := e2 - e1
+
+ // Since elapsed time has an error margin of +/- half a second, elapsedDiff
+ // can end up being smaller than tsDiff sometimes. Hence we take abs() of
+ // the difference.
+ return abs(elapsedDiff-tsDiff) > errorThreshold
+}
+
+func abs(d time.Duration) time.Duration {
+ if d < 0 {
+ return -d
+ }
+ return d
+}
diff --git a/services/syncbase/clock/util_test.go b/services/syncbase/clock/util_test.go
new file mode 100644
index 0000000..b7dac71
--- /dev/null
+++ b/services/syncbase/clock/util_test.go
@@ -0,0 +1,81 @@
+// 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 clock
+
+import (
+ "math/rand"
+ "testing"
+ "time"
+)
+
+func TestHasSysClockChangedWithRealClock(t *testing.T) {
+ for i := 0; i < 10; i++ {
+ sysClock := newSystemClock()
+ e1, err := sysClock.ElapsedTime()
+ t1 := sysClock.Now()
+ if err != nil {
+ t.Errorf("Found error while fetching e1: %v", err)
+ }
+
+ // spend some time.
+ d := time.Duration(rand.Int63n(50)) * time.Millisecond
+ time.Sleep(d)
+
+ t2 := sysClock.Now()
+ e2, err := sysClock.ElapsedTime()
+ if err != nil {
+ t.Errorf("Found error while fetching e2: %v", err)
+ }
+
+ if HasSysClockChanged(t1, t2, e1, e2) {
+ t.Errorf("Clock found changed incorrectly. e1: %v, t1: %v, t2: %v, e2: %v", e1, t1, t2, e2)
+ }
+ }
+}
+
+func TestHasSysClockChangedFakeClock(t *testing.T) {
+ e1 := 2000 * time.Millisecond
+ t1 := time.Now()
+
+ // elapsed time diff slightly greater than clock diff.
+ t2 := t1.Add(200 * time.Millisecond)
+ e2 := e1 + 300*time.Millisecond
+
+ if HasSysClockChanged(t1, t2, e1, e2) {
+ t.Errorf("Clock found changed incorrectly. e1: %v, t1: %v, t2: %v, e2: %v", e1, t1, t2, e2)
+ }
+
+ // elapsed time diff slightly smaller than clock diff.
+ t2 = t1.Add(300 * time.Millisecond)
+ e2 = e1 + 200*time.Millisecond
+
+ if HasSysClockChanged(t1, t2, e1, e2) {
+ t.Errorf("Clock found changed incorrectly. e1: %v, t1: %v, t2: %v, e2: %v", e1, t1, t2, e2)
+ }
+
+ // elapsed time diff much greater than clock diff.
+ t2 = t1.Add(200 * time.Millisecond)
+ e2 = e1 + 3000*time.Millisecond
+
+ if !HasSysClockChanged(t1, t2, e1, e2) {
+ t.Errorf("Clock changed but not caught. e1: %v, t1: %v, t2: %v, e2: %v", e1, t1, t2, e2)
+ }
+
+ // elapsed time diff much smaller than clock diff.
+ t2 = t1.Add(4000 * time.Millisecond)
+ e2 = e1 + 300*time.Millisecond
+
+ if !HasSysClockChanged(t1, t2, e1, e2) {
+ t.Errorf("Clock changed but not caught. e1: %v, t1: %v, t2: %v, e2: %v", e1, t1, t2, e2)
+ }
+
+ // clock diff is negative
+ t2 = t1.Add(-200 * time.Millisecond)
+ e2 = e1 + 300*time.Millisecond
+
+ if !HasSysClockChanged(t1, t2, e1, e2) {
+ t.Errorf("Clock changed but not caught. e1: %v, t1: %v, t2: %v, e2: %v", e1, t1, t2, e2)
+ }
+}
diff --git a/services/syncbase/clock/vclock.go b/services/syncbase/clock/vclock.go
index cf67845..a8059e7 100644
--- a/services/syncbase/clock/vclock.go
+++ b/services/syncbase/clock/vclock.go
@@ -7,56 +7,89 @@
import (
"time"
- "v.io/v23/context"
"v.io/v23/verror"
"v.io/x/lib/vlog"
+ "v.io/x/ref/services/syncbase/server/util"
"v.io/x/ref/services/syncbase/store"
)
// VClock holds data required to provide an estimate of the UTC time at any
// given point. The fields contained in here are
-// - systemTimeAtBoot : the time shown by the system clock at boot.
-// - skew : the difference between the system clock and UTC time.
-// - clock : Instance of clock.SystemClock interface providing access
+// - SysClock : Instance of clock.SystemClock interface providing access
// to the system time.
-// - sa : adapter for storage of clock data.
// - ntpSource : source for fetching NTP data.
+// - st : store object representing syncbase (not a specific db).
type VClock struct {
- systemTimeAtBoot time.Time
- skew time.Duration
- clock SystemClock
- sa StorageAdapter
- ntpSource NtpSource
+ SysClock SystemClock
+ ntpSource NtpSource
+ st store.Store
}
func NewVClock(st store.Store) *VClock {
sysClock := newSystemClock()
return &VClock{
- clock: sysClock,
- sa: NewStorageAdapter(st),
+ SysClock: sysClock,
ntpSource: NewNtpSource(sysClock),
+ st: st,
}
}
// Now returns current UTC time based on the estimation of skew that
// the system clock has with respect to NTP time.
-func (c *VClock) Now(ctx *context.T) time.Time {
+// The timestamp is based on system timestamp taken after database lookup
+// for skew.
+func (c *VClock) Now() time.Time {
clockData := &ClockData{}
- if err := c.sa.GetClockData(ctx, clockData); err != nil {
+ if err := c.GetClockData(c.st, clockData); err != nil {
if verror.ErrorID(err) == verror.ErrNoExist.ID {
// VClock's cron job to setup UTC time at boot has not been run yet.
- // TODO(jlodhia): uncomment info messages once clock service
- // scheduling is enabled. In absence of clock service, no clock
- // data is present and hence these logs get printed all the time.
- // vlog.Info("No ClockData found while creating a timestamp")
+ vlog.VI(2).Info("clock: No ClockData found while creating a timestamp")
} else {
- vlog.Errorf("Error while fetching clock data: %v", err)
+ vlog.Errorf("clock: Error while fetching clock data: %v", err)
}
- // vlog.Info("Returning current system clock time")
- return c.clock.Now()
+ vlog.VI(2).Info("clock: Returning current system clock time")
+ return c.SysClock.Now()
}
skew := time.Duration(clockData.Skew)
- return c.clock.Now().Add(skew)
+ return c.SysClock.Now().Add(skew)
+}
+
+// NowNoLookup is similar to Now() but does not do a database lookup to get
+// the ClockData. Useful when clockData is already being fetched by calling
+// code.
+func (c *VClock) NowNoLookup(clockData ClockData) time.Time {
+ skew := time.Duration(clockData.Skew)
+ return c.SysClock.Now().Add(skew)
+}
+
+// VClockTs converts any system timestamp to a syncbase timestamp based on
+// skew provided in ClockData. Useful when the caller does not want the
+// time taken to fetch ClockData to be reflected in a timestamp.
+// example usage:
+// RpcMethod(req) {
+// recvSysTs := vclock.SysClock.Now()
+// clockData := vclock.GetClockData()
+// recvTs := vclock.VClockTs(recvSysTs, clockData)
+// ...
+// }
+func (c *VClock) VClockTs(sysTime time.Time, clockData ClockData) time.Time {
+ return sysTime.Add(time.Duration(clockData.Skew))
+}
+
+func (c *VClock) St() store.Store {
+ return c.st
+}
+
+func (c *VClock) GetClockData(st store.StoreReader, data *ClockData) error {
+ return util.Get(nil, st, clockDataKey(), data)
+}
+
+func (c *VClock) SetClockData(tx store.Transaction, data *ClockData) error {
+ return util.Put(nil, tx, clockDataKey(), data)
+}
+
+func clockDataKey() string {
+ return util.ClockPrefix
}
///////////////////////////////////////////////////
diff --git a/services/syncbase/clock/vclock_test.go b/services/syncbase/clock/vclock_test.go
index bf92f5c..198593c 100644
--- a/services/syncbase/clock/vclock_test.go
+++ b/services/syncbase/clock/vclock_test.go
@@ -7,18 +7,23 @@
import (
"testing"
"time"
-
- "v.io/v23/verror"
)
func TestVClock(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
sysClock := MockSystemClock(sysTs, 0)
- stAdapter := MockStorageAdapter()
- stAdapter.SetClockData(nil, &ClockData{0, 0, 0})
- clock := NewVClockWithMockServices(stAdapter, sysClock, nil)
+ clock := NewVClockWithMockServices(testStore.st, sysClock, nil)
- ts := clock.Now(nil)
+ tx := clock.St().NewTransaction()
+ clock.SetClockData(tx, newClockData(0, 0, 0, nil, 0, 0))
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ ts := clock.Now()
if ts != sysTs {
t.Errorf("timestamp expected to be %q but found to be %q", sysTs, ts)
}
@@ -32,18 +37,25 @@
}
func checkSkew(t *testing.T, skew int64) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
sysTs := time.Now()
+ ntpTs := sysTs.Add(time.Duration(-20))
sysClock := MockSystemClock(sysTs, 0)
var elapsedTime int64 = 100
- stAdapter := MockStorageAdapter()
bootTime := sysTs.UnixNano() - elapsedTime
- clockData := ClockData{bootTime, skew, elapsedTime}
- stAdapter.SetClockData(nil, &clockData)
+ clockData := newClockData(bootTime, skew, elapsedTime, &ntpTs, 1, 1)
+ clock := NewVClockWithMockServices(testStore.st, sysClock, nil)
- clock := NewVClockWithMockServices(stAdapter, sysClock, nil)
+ tx := clock.St().NewTransaction()
+ clock.SetClockData(tx, clockData)
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
- ts := clock.Now(nil)
+ ts := clock.Now()
if ts == sysTs {
t.Errorf("timestamp expected to be %q but found to be %q", sysTs, ts)
}
@@ -52,18 +64,18 @@
}
}
-func TestVClockWithInternalErr(t *testing.T) {
+func TestVClockTs(t *testing.T) {
+ testStore := createStore(t)
+ defer destroyStore(t, testStore)
+
+ vclock := NewVClock(testStore.st)
+ skew := time.Minute.Nanoseconds()
+ clockData := newClockData(0, skew, 0, nil, 0, 0)
+
sysTs := time.Now()
- sysClock := MockSystemClock(sysTs, 0)
+ adjTs := vclock.VClockTs(sysTs, *clockData)
- stAdapter := MockStorageAdapter()
- stAdapter.SetError(verror.NewErrInternal(nil))
-
- clock := NewVClockWithMockServices(stAdapter, sysClock, nil)
-
- // Internal err should result in vclock falling back to the system clock.
- ts := clock.Now(nil)
- if ts != sysTs {
- t.Errorf("timestamp expected to be %q but found to be %q", sysTs, ts)
+ if adjTs.Sub(sysTs) != time.Minute {
+ t.Errorf("Unexpected diff found. SysTs: %v, AdjTs: %v", sysTs, adjTs)
}
}
diff --git a/services/syncbase/server/interfaces/sync.vdl b/services/syncbase/server/interfaces/sync.vdl
index 80b46a0..4e11e3f 100644
--- a/services/syncbase/server/interfaces/sync.vdl
+++ b/services/syncbase/server/interfaces/sync.vdl
@@ -12,6 +12,10 @@
// Sync defines methods for data exchange between Syncbases.
// TODO(hpucha): Flesh this out further.
type Sync interface {
+ // GetTime returns metadata related to syncbase clock like syncbase clock
+ // timestamps, last NTP timestamp, num reboots, etc.
+ GetTime(req TimeReq, initiator string) (TimeResp | error)
+
// GetDeltas returns the responder's current generation vector and all
// the missing log records when compared to the initiator's generation
// vector for one Database for either SyncGroup metadata or data.
diff --git a/services/syncbase/server/interfaces/sync.vdl.go b/services/syncbase/server/interfaces/sync.vdl.go
index 102699b..ebaa912 100644
--- a/services/syncbase/server/interfaces/sync.vdl.go
+++ b/services/syncbase/server/interfaces/sync.vdl.go
@@ -41,6 +41,9 @@
// Sync defines methods for data exchange between Syncbases.
// TODO(hpucha): Flesh this out further.
type SyncClientMethods interface {
+ // GetTime returns metadata related to syncbase clock like syncbase clock
+ // timestamps, last NTP timestamp, num reboots, etc.
+ GetTime(ctx *context.T, req TimeReq, initiator string, opts ...rpc.CallOpt) (TimeResp, error)
// GetDeltas returns the responder's current generation vector and all
// the missing log records when compared to the initiator's generation
// vector for one Database for either SyncGroup metadata or data.
@@ -107,6 +110,11 @@
name string
}
+func (c implSyncClientStub) GetTime(ctx *context.T, i0 TimeReq, i1 string, opts ...rpc.CallOpt) (o0 TimeResp, err error) {
+ err = v23.GetClient(ctx).Call(ctx, c.name, "GetTime", []interface{}{i0, i1}, []interface{}{&o0}, opts...)
+ return
+}
+
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, i1}, opts...); err != nil {
@@ -472,6 +480,9 @@
// Sync defines methods for data exchange between Syncbases.
// TODO(hpucha): Flesh this out further.
type SyncServerMethods interface {
+ // GetTime returns metadata related to syncbase clock like syncbase clock
+ // timestamps, last NTP timestamp, num reboots, etc.
+ GetTime(ctx *context.T, call rpc.ServerCall, req TimeReq, initiator string) (TimeResp, error)
// GetDeltas returns the responder's current generation vector and all
// the missing log records when compared to the initiator's generation
// vector for one Database for either SyncGroup metadata or data.
@@ -528,6 +539,9 @@
// The only difference between this interface and SyncServerMethods
// is the streaming methods.
type SyncServerStubMethods interface {
+ // GetTime returns metadata related to syncbase clock like syncbase clock
+ // timestamps, last NTP timestamp, num reboots, etc.
+ GetTime(ctx *context.T, call rpc.ServerCall, req TimeReq, initiator string) (TimeResp, error)
// GetDeltas returns the responder's current generation vector and all
// the missing log records when compared to the initiator's generation
// vector for one Database for either SyncGroup metadata or data.
@@ -608,6 +622,10 @@
gs *rpc.GlobState
}
+func (s implSyncServerStub) GetTime(ctx *context.T, call rpc.ServerCall, i0 TimeReq, i1 string) (TimeResp, error) {
+ return s.impl.GetTime(ctx, call, i0, i1)
+}
+
func (s implSyncServerStub) GetDeltas(ctx *context.T, call *SyncGetDeltasServerCallStub, i0 DeltaReq, i1 string) error {
return s.impl.GetDeltas(ctx, call, i0, i1)
}
@@ -654,6 +672,17 @@
Doc: "// Sync defines methods for data exchange between Syncbases.\n// TODO(hpucha): Flesh this out further.",
Methods: []rpc.MethodDesc{
{
+ Name: "GetTime",
+ Doc: "// GetTime returns metadata related to syncbase clock like syncbase clock\n// timestamps, last NTP timestamp, num reboots, etc.",
+ InArgs: []rpc.ArgDesc{
+ {"req", ``}, // TimeReq
+ {"initiator", ``}, // string
+ },
+ OutArgs: []rpc.ArgDesc{
+ {"", ``}, // TimeResp
+ },
+ },
+ {
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 for one Database for either SyncGroup metadata or data.",
InArgs: []rpc.ArgDesc{
diff --git a/services/syncbase/server/interfaces/sync_types.vdl b/services/syncbase/server/interfaces/sync_types.vdl
index 849e8ec..acd58f6 100644
--- a/services/syncbase/server/interfaces/sync_types.vdl
+++ b/services/syncbase/server/interfaces/sync_types.vdl
@@ -146,3 +146,34 @@
type ChunkData struct {
Data []byte
}
+
+// TimeReq contains the send timestamp from the requester.
+type TimeReq struct {
+ SendTs time.Time
+}
+
+// TimeResp contains timestamps needed by the requester to correctly estimate
+// the difference between the two clocks along with clock metadata required
+// by the requester to decide which clock is more accurate.
+type TimeResp struct {
+ // The send timestamp received in TimeReq from the originator.
+ OrigTs time.Time
+
+ // Time when the request was received.
+ RecvTs time.Time
+
+ // Time when the response was sent.
+ SendTs time.Time
+
+ // Timestamp received from NTP during last NTP sync. The last NTP sync could
+ // be done either by this device or some other device that this device
+ // synced its clock with.
+ LastNtpTs ?time.Time
+
+ // Number of reboots since last NTP sync.
+ NumReboots uint16
+
+ // Number of hops between this device and the device that did the last
+ // NTP sync.
+ NumHops uint16
+}
diff --git a/services/syncbase/server/interfaces/sync_types.vdl.go b/services/syncbase/server/interfaces/sync_types.vdl.go
index a00591c..d63f32f 100644
--- a/services/syncbase/server/interfaces/sync_types.vdl.go
+++ b/services/syncbase/server/interfaces/sync_types.vdl.go
@@ -291,6 +291,42 @@
}) {
}
+// TimeReq contains the send timestamp from the requester.
+type TimeReq struct {
+ SendTs time.Time
+}
+
+func (TimeReq) __VDLReflect(struct {
+ Name string `vdl:"v.io/x/ref/services/syncbase/server/interfaces.TimeReq"`
+}) {
+}
+
+// TimeResp contains timestamps needed by the requester to correctly estimate
+// the difference between the two clocks along with clock metadata required
+// by the requester to decide which clock is more accurate.
+type TimeResp struct {
+ // The send timestamp received in TimeReq from the originator.
+ OrigTs time.Time
+ // Time when the request was received.
+ RecvTs time.Time
+ // Time when the response was sent.
+ SendTs time.Time
+ // Timestamp received from NTP during last NTP sync. The last NTP sync could
+ // be done either by this device or some other device that this device
+ // synced its clock with.
+ LastNtpTs *time.Time
+ // Number of reboots since last NTP sync.
+ NumReboots uint16
+ // Number of hops between this device and the device that did the last
+ // NTP sync.
+ NumHops uint16
+}
+
+func (TimeResp) __VDLReflect(struct {
+ Name string `vdl:"v.io/x/ref/services/syncbase/server/interfaces.TimeResp"`
+}) {
+}
+
func init() {
vdl.Register((*PrefixGenVector)(nil))
vdl.Register((*GenVector)(nil))
@@ -305,6 +341,8 @@
vdl.Register((*DeltaResp)(nil))
vdl.Register((*ChunkHash)(nil))
vdl.Register((*ChunkData)(nil))
+ vdl.Register((*TimeReq)(nil))
+ vdl.Register((*TimeResp)(nil))
}
const NoGroupId = GroupId(0)
diff --git a/services/syncbase/server/service.go b/services/syncbase/server/service.go
index 64d7547..e6e3787 100644
--- a/services/syncbase/server/service.go
+++ b/services/syncbase/server/service.go
@@ -19,6 +19,7 @@
wire "v.io/v23/services/syncbase"
"v.io/v23/verror"
"v.io/v23/vom"
+ "v.io/x/ref/services/syncbase/clock"
"v.io/x/ref/services/syncbase/server/interfaces"
"v.io/x/ref/services/syncbase/server/nosql"
"v.io/x/ref/services/syncbase/server/util"
@@ -28,9 +29,10 @@
// service is a singleton (i.e. not per-request) that handles Service RPCs.
type service struct {
- st store.Store // keeps track of which apps and databases exist, etc.
- sync interfaces.SyncServerMethods
- opts ServiceOptions
+ st store.Store // keeps track of which apps and databases exist, etc.
+ sync interfaces.SyncServerMethods
+ clockD *clock.ClockD
+ opts ServiceOptions
// Guards the fields below. Held during app Create, Delete, and
// SetPermissions.
mu sync.Mutex
@@ -125,11 +127,16 @@
return nil, err
}
}
+ vclock := clock.NewVClock(st)
// Note, vsync.New internally handles both first-time and subsequent
// invocations.
- if s.sync, err = vsync.New(ctx, call, s, opts.Engine, opts.RootDir); err != nil {
+ if s.sync, err = vsync.New(ctx, call, s, opts.Engine, opts.RootDir, vclock); err != nil {
return nil, err
}
+
+ // Create a new clock deamon and run clock check and ntp check in a
+ // scheduled loop.
+ s.clockD = clock.StartClockD(ctx, vclock)
return s, nil
}
diff --git a/services/syncbase/server/util/constants.go b/services/syncbase/server/util/constants.go
index 5841aa9..e38377e 100644
--- a/services/syncbase/server/util/constants.go
+++ b/services/syncbase/server/util/constants.go
@@ -55,7 +55,6 @@
// Object name component for Syncbase-to-Syncbase (sync) RPCs.
// Sync object names have the form:
// <syncbase>/%%sync/...
- // FIXME: update all apps, again...
SyncbaseSuffix = "%%sync"
// Separator for batch info in database names.
// Batch object names have the form:
@@ -72,4 +71,7 @@
NtpSampleCount = 15
LocalClockDriftThreshold = float64(time.Second)
NtpDiffThreshold = float64(2 * time.Second)
+ PeerSyncDiffThreshold = NtpDiffThreshold
+ RebootTolerance = float64(time.Minute)
+ HopTolerance = 2
)
diff --git a/services/syncbase/server/watchable/store_test.go b/services/syncbase/server/watchable/store_test.go
index d44cc08..53bbec1 100644
--- a/services/syncbase/server/watchable/store_test.go
+++ b/services/syncbase/server/watchable/store_test.go
@@ -65,7 +65,7 @@
func runTest(t *testing.T, mp []string, f func(t *testing.T, st store.Store)) {
st, destroy := createStore()
defer destroy()
- vClock := clock.NewVClockWithMockServices(clock.MockStorageAdapter(), nil, nil)
+ vClock := clock.NewVClockWithMockServices(st, nil, nil)
st, err := Wrap(st, vClock, &Options{ManagedPrefixes: mp})
if err != nil {
t.Fatal(err)
diff --git a/services/syncbase/server/watchable/transaction.go b/services/syncbase/server/watchable/transaction.go
index 3178cb7..b4556c3 100644
--- a/services/syncbase/server/watchable/transaction.go
+++ b/services/syncbase/server/watchable/transaction.go
@@ -140,7 +140,7 @@
return verror.New(verror.ErrInternal, nil, "seq maxed out")
}
// Write LogEntry records.
- timestamp := tx.st.clock.Now(nil).UnixNano()
+ timestamp := tx.st.clock.Now().UnixNano()
seq := tx.st.seq
for i, op := range tx.ops {
key := logEntryKey(seq)
@@ -177,7 +177,7 @@
// GetStoreTime returns the current time from the given transaction store.
func GetStoreTime(ctx *context.T, tx store.Transaction) time.Time {
wtx := tx.(*transaction)
- return wtx.st.clock.Now(ctx)
+ return wtx.st.clock.Now()
}
// AddSyncGroupOp injects a SyncGroup operation notification in the log entries
diff --git a/services/syncbase/server/watchable/transaction_test.go b/services/syncbase/server/watchable/transaction_test.go
index 681b3f7..518f7b3 100644
--- a/services/syncbase/server/watchable/transaction_test.go
+++ b/services/syncbase/server/watchable/transaction_test.go
@@ -71,9 +71,7 @@
t1 := time.Now()
inc := time.Duration(1) * time.Second
mockClock := newMockSystemClock(t1, inc)
- var mockAdapter clock.StorageAdapter = clock.MockStorageAdapter()
-
- vclock := clock.NewVClockWithMockServices(mockAdapter, mockClock, nil)
+ vclock := clock.NewVClockWithMockServices(ist, mockClock, nil)
wst1, err := Wrap(ist, vclock, &Options{ManagedPrefixes: nil})
if err != nil {
t.Fatalf("Wrap failed: %v", err)
@@ -143,9 +141,7 @@
t1 := time.Now()
inc := time.Duration(1) * time.Second
mockClock := newMockSystemClock(t1, inc)
- var mockAdapter clock.StorageAdapter = clock.MockStorageAdapter()
-
- vclock := clock.NewVClockWithMockServices(mockAdapter, mockClock, nil)
+ vclock := clock.NewVClockWithMockServices(ist, mockClock, nil)
wst, err := Wrap(ist, vclock, &Options{ManagedPrefixes: nil})
if err != nil {
t.Fatalf("Wrap failed: %v", err)
diff --git a/services/syncbase/server/watchable/util_test.go b/services/syncbase/server/watchable/util_test.go
index e0db4ac..b7b4412 100644
--- a/services/syncbase/server/watchable/util_test.go
+++ b/services/syncbase/server/watchable/util_test.go
@@ -14,8 +14,7 @@
func TestGetNextLogSeq(t *testing.T) {
st, destroy := createStore()
defer destroy()
- var mockAdapter clock.StorageAdapter = clock.MockStorageAdapter()
- vclock := clock.NewVClockWithMockServices(mockAdapter, nil, nil)
+ vclock := clock.NewVClockWithMockServices(st, nil, nil)
st, err := Wrap(st, vclock, &Options{})
if err != nil {
t.Fatal(err)
diff --git a/services/syncbase/vsync/clock.go b/services/syncbase/vsync/clock.go
new file mode 100644
index 0000000..13a9dda
--- /dev/null
+++ b/services/syncbase/vsync/clock.go
@@ -0,0 +1,184 @@
+// 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 (
+ "fmt"
+ "time"
+
+ "v.io/v23/context"
+ "v.io/v23/naming"
+ "v.io/v23/rpc"
+ "v.io/v23/verror"
+ "v.io/x/lib/vlog"
+ "v.io/x/ref/services/syncbase/clock"
+ "v.io/x/ref/services/syncbase/server/interfaces"
+ "v.io/x/ref/services/syncbase/server/util"
+)
+
+// GetTime implements the responder side of the GetTime RPC.
+func (s *syncService) GetTime(ctx *context.T, call rpc.ServerCall, req interfaces.TimeReq, initiator string) (interfaces.TimeResp, error) {
+ // To make sure that the system clock did not change between fetching
+ // receive timestamp and send timestamp, we get the elapsed time since
+ // boot (which is immutable) before registering the receive timestamp and
+ // after registering the send timestamp and call HasSysClockChanged()
+ // to verify if the clock changed in between or not. If it did, we return
+ // ErrInternal as response.
+ elapsedRecv, err := s.vclock.SysClock.ElapsedTime()
+
+ // In order to get the receive timestamp as close to the reciept of the req
+ // we need to first take a timestamp from the system clock and then lookup
+ // clock data to convert this timestamp into vclock timestamp to avoid
+ // adding the lookup time to receive timestamp.
+ sysTs := s.vclock.SysClock.Now()
+ if err != nil {
+ // We do this test after fetching sysTs to make sure that the sysTs is
+ // as close to the receipt time of request as possible.
+ vlog.Errorf("sync: GetTime: error while fetching elapsed time: %v", err)
+ return interfaces.TimeResp{}, err
+ }
+
+ // Lookup local clock data.
+ clockData := &clock.ClockData{}
+ if err := s.vclock.GetClockData(s.vclock.St(), clockData); err != nil {
+ vlog.Errorf("sync: GetTime: error while fetching clock data: %v", err)
+ return interfaces.TimeResp{}, err
+ }
+
+ // Convert the sysTs that was registered above into vclock ts.
+ recvTs := s.vclock.VClockTs(sysTs, *clockData)
+ sendTs := s.vclock.NowNoLookup(*clockData)
+
+ // Make sure that the system clock did not change since the request was
+ // sent.
+ if err := isClockChanged(s.vclock, recvTs, sendTs, elapsedRecv); err != nil {
+ vlog.Errorf("sync: GetTime: %v", err)
+ return interfaces.TimeResp{}, verror.NewErrInternal(ctx)
+ }
+
+ resp := interfaces.TimeResp{
+ OrigTs: req.SendTs,
+ RecvTs: recvTs,
+ SendTs: sendTs,
+ LastNtpTs: clockData.LastNtpTs,
+ NumReboots: clockData.NumReboots,
+ NumHops: clockData.NumHops,
+ }
+ return resp, nil
+}
+
+// syncClock syncs the syncbase clock with peer's syncbase clock.
+// TODO(jlodhia): Refactor the mount table entry search code based on the
+// unified solution for looking up peer once it exists.
+func (s *syncService) syncClock(ctx *context.T, peer string) {
+ vlog.VI(2).Infof("sync: syncClock: begin: contacting peer %s", peer)
+ defer vlog.VI(2).Infof("sync: syncClock: end: contacting peer %s", peer)
+
+ info := s.copyMemberInfo(ctx, peer)
+ if info == nil {
+ vlog.Fatalf("sync: syncClock: missing information in member view for %q", peer)
+ }
+
+ // Preferred mount tables for this peer.
+ if len(info.mtTables) < 1 {
+ vlog.Errorf("sync: syncClock: no mount tables found to connect to peer %s", peer)
+ return
+ }
+ for mt, _ := range info.mtTables {
+ absName := naming.Join(mt, peer, util.SyncbaseSuffix)
+ if err := syncWithPeer(ctx, s.vclock, absName, s.name); err == nil {
+ return
+ } else if (verror.ErrorID(err) == verror.ErrNoExist.ID) || (verror.ErrorID(err) == verror.ErrInternal.ID) {
+ vlog.Errorf("sync: syncClock: error returned by peer %s: %v", peer, err)
+ return
+ }
+ }
+ vlog.Errorf("sync: syncClock: couldn't connect to peer %s", peer)
+ return
+}
+
+// syncWithPeer tries to sync local clock with peer syncbase clock.
+// Returns error if the GetTime() rpc returns error.
+func syncWithPeer(ctx *context.T, vclock *clock.VClock, absPeerName string, myName string) error {
+ c := interfaces.SyncClient(absPeerName)
+
+ // Since syncing time with peer is a non critical task, its not necessary
+ // to retry it if the transaction fails. Peer time sync will be run
+ // frequently and so the next run will be the next retry.
+ tx := vclock.St().NewTransaction()
+
+ // Lookup clockData. If error received return ErrBadState to let the caller
+ // distinguish it from a network error so that it can stop looping through
+ // other mounttable entires.
+ localData := &clock.ClockData{}
+ err := vclock.GetClockData(tx, localData)
+ if err != nil {
+ // To avoid dealing with non existent clockdata during clock sync
+ // just abort clock sync and wait for the clockservice to create
+ // clockdata.
+ vlog.Errorf("sync: syncClock: error while fetching local clock data: %v", err)
+ tx.Abort()
+ return nil
+ }
+
+ // See comments in GetTime() related to catching system clock changing
+ // midway.
+ elapsedOrig, err := vclock.SysClock.ElapsedTime()
+ if err != nil {
+ vlog.Errorf("sync: syncClock: error while fetching elapsed time: %v", err)
+ return nil
+ }
+
+ timeResp, reqErr := c.GetTime(ctx, makeTimeReq(vclock, localData), myName)
+ if reqErr == nil {
+ recvTs := vclock.NowNoLookup(*localData)
+
+ // Make sure that the system clock did not change since the request was
+ // sent.
+ if err := isClockChanged(vclock, timeResp.OrigTs, recvTs, elapsedOrig); err != nil {
+ vlog.Errorf("sync: syncClock: %v", err)
+ return nil
+ }
+
+ vlog.VI(4).Infof("sync: syncClock: connection established on %s", absPeerName)
+ vclock.ProcessPeerClockData(tx, toPeerSyncData(&timeResp, recvTs), localData)
+ if commitErr := tx.Commit(); commitErr != nil {
+ vlog.Errorf("sync: syncClock: error while commiting tx: %v", commitErr)
+ }
+ } else {
+ vlog.Errorf("sync: syncClock: received error from peer: %v", reqErr)
+ }
+ // Return error received while making request if any to the caller.
+ return reqErr
+}
+
+func isClockChanged(vclock *clock.VClock, t1, t2 time.Time, e1 time.Duration) error {
+ e2, err := vclock.SysClock.ElapsedTime()
+ if err != nil {
+ return fmt.Errorf("error while fetching elapsed time: %v", err)
+ }
+ if clock.HasSysClockChanged(t1, t2, e1, e2) {
+ return fmt.Errorf("system clock changed midway through sycning clock with peer.")
+ }
+ return nil
+}
+
+func makeTimeReq(vclock *clock.VClock, clockData *clock.ClockData) interfaces.TimeReq {
+ return interfaces.TimeReq{
+ SendTs: vclock.NowNoLookup(*clockData),
+ }
+}
+
+func toPeerSyncData(resp *interfaces.TimeResp, recvTs time.Time) *clock.PeerSyncData {
+ return &clock.PeerSyncData{
+ MySendTs: resp.OrigTs,
+ RecvTs: resp.RecvTs,
+ SendTs: resp.SendTs,
+ MyRecvTs: recvTs,
+ LastNtpTs: resp.LastNtpTs,
+ NumReboots: resp.NumReboots,
+ NumHops: resp.NumHops,
+ }
+}
diff --git a/services/syncbase/vsync/clock_test.go b/services/syncbase/vsync/clock_test.go
new file mode 100644
index 0000000..e9f776e
--- /dev/null
+++ b/services/syncbase/vsync/clock_test.go
@@ -0,0 +1,75 @@
+// 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 (
+ "testing"
+ "time"
+
+ "v.io/x/ref/services/syncbase/clock"
+ "v.io/x/ref/services/syncbase/server/interfaces"
+)
+
+func TestClockGetTime(t *testing.T) {
+ service := createService(t)
+ defer destroyService(t, service)
+
+ ntpTs := time.Now().Add(-10 * time.Minute)
+ skew := 3 * time.Second
+
+ clockData := newClockData(skew.Nanoseconds(), &ntpTs, 0, 0)
+ vclock := clock.NewVClock(service.St())
+ tx := vclock.St().NewTransaction()
+ if err := vclock.SetClockData(tx, clockData); err != nil {
+ t.Errorf("Failed to store clock data with error: %v", err)
+ }
+ if err := tx.Commit(); err != nil {
+ t.Errorf("Error while commiting tx: %v", err)
+ }
+
+ req := createReq(time.Now())
+ resp, err := service.Sync().GetTime(nil, nil, req, "test")
+ if err != nil {
+ t.Errorf("GetTime rpc returned error: %v", err)
+ t.FailNow()
+ }
+ offset := getOffset(resp)
+
+ if abs(offset-skew) > time.Millisecond {
+ t.Errorf("GetTime returned a skew beyond error margin. expected: %v, actual: %v", skew, offset)
+ }
+}
+
+func abs(d time.Duration) time.Duration {
+ if d < 0 {
+ return -d
+ }
+ return d
+}
+
+func createReq(ts time.Time) interfaces.TimeReq {
+ return interfaces.TimeReq{
+ SendTs: ts,
+ }
+}
+
+func newClockData(skew int64, ntp *time.Time, reboots, hops uint16) *clock.ClockData {
+ return &clock.ClockData{
+ SystemTimeAtBoot: 0,
+ Skew: skew,
+ ElapsedTimeSinceBoot: 0,
+ LastNtpTs: ntp,
+ NumReboots: reboots,
+ NumHops: hops,
+ }
+}
+
+func getOffset(resp interfaces.TimeResp) time.Duration {
+ clientReceiveTs := time.Now()
+ clientTransmitTs := resp.OrigTs
+ serverReceiveTs := resp.RecvTs
+ serverTransmitTs := resp.SendTs
+ return (serverReceiveTs.Sub(clientTransmitTs) + serverTransmitTs.Sub(clientReceiveTs)) / 2
+}
diff --git a/services/syncbase/vsync/sync.go b/services/syncbase/vsync/sync.go
index 5db72ce..28f9263 100644
--- a/services/syncbase/vsync/sync.go
+++ b/services/syncbase/vsync/sync.go
@@ -23,6 +23,7 @@
"v.io/v23/rpc"
"v.io/v23/services/syncbase/nosql"
"v.io/v23/verror"
+ "v.io/x/ref/services/syncbase/clock"
blob "v.io/x/ref/services/syncbase/localblobstore"
fsblob "v.io/x/ref/services/syncbase/localblobstore/fs_cablobstore"
"v.io/x/ref/services/syncbase/server/interfaces"
@@ -91,6 +92,8 @@
blobDirectory map[nosql.BlobRef]*blobLocInfo // directory structure containing blob location information.
blobDirLock sync.RWMutex // lock to synchronize access to the blob directory information.
+ // Syncbase clock related variables.
+ vclock *clock.VClock
}
// syncDatabase contains the metadata for syncing a database. This struct is
@@ -128,11 +131,12 @@
// changes to its objects. The "initiator" thread is responsible for
// periodically contacting peers to fetch changes from them. In addition, the
// sync module responds to incoming RPCs from remote sync modules.
-func New(ctx *context.T, call rpc.ServerCall, sv interfaces.Service, blobStEngine, blobRootDir string) (*syncService, error) {
+func New(ctx *context.T, call rpc.ServerCall, sv interfaces.Service, blobStEngine, blobRootDir string, vclock *clock.VClock) (*syncService, error) {
s := &syncService{
sv: sv,
batches: make(batchSet),
sgPublishQueue: list.New(),
+ vclock: vclock,
}
data := &syncData{}
diff --git a/services/syncbase/vsync/syncer.go b/services/syncbase/vsync/syncer.go
index 22f485b..b3fb28f 100644
--- a/services/syncbase/vsync/syncer.go
+++ b/services/syncbase/vsync/syncer.go
@@ -71,6 +71,8 @@
continue
}
+ s.syncClock(ctx, peer)
+
// Sync Syncgroup metadata and data.
s.getDeltas(ctx, peer)
}
diff --git a/services/syncbase/vsync/test_util.go b/services/syncbase/vsync/test_util.go
index 7bef23c..4ec5f80 100644
--- a/services/syncbase/vsync/test_util.go
+++ b/services/syncbase/vsync/test_util.go
@@ -132,7 +132,7 @@
dir: dir,
shutdown: shutdown,
}
- if s.sync, err = New(ctx, nil, s, engine, dir); err != nil {
+ if s.sync, err = New(ctx, nil, s, engine, dir, vclock); err != nil {
util.DestroyStore(engine, dir)
t.Fatalf("cannot create sync service: %v", err)
}
