check implementation of the HLC

Author: gas

README.md

@@ -11,3 +11,9 @@ To test the clocks implementations run `make test`.
 
 ### API  
 *TODO*  
+
+### References  
+ - [Lamport clock](https://lamport.azurewebsites.net/pubs/time-clocks.pdf)  
+ - [Vector clock](https://fileadmin.cs.lth.se/cs/Personal/Amr_Ergawy/dist-algos-papers/4.pdf)  
+ - [Hybrid logical clock](https://cse.buffalo.edu/tech-reports/2014-04.pdf)  
+ - [Martin Fowler blog](https://martinfowler.com/articles/patterns-of-distributed-systems/hybrid-clock.html)  

hlc/hlc.go

@@ -5,23 +5,27 @@ import (
 	"time"
 )
 
+func getWallTimeMs() int64 {
+	return time.Now().UnixNano() / 1e6
+}
+
 type Hybrid struct {
-	mu    sync.Mutex
-	Time  int64 // Usually time in ms
-	Ticks int64
+	mu              sync.RWMutex
+	WallClockTimeMs int64
+	Ticks           int64
 }
 
 func New(systemTime, ticks int64) *Hybrid {
 	return &Hybrid{
-		Time:  systemTime,
-		Ticks: ticks,
+		WallClockTimeMs: systemTime,
+		Ticks:           ticks,
 	}
 }
 
 func NewNow(ticks int64) *Hybrid {
 	return &Hybrid{
-		Time:  time.Now().UnixNano() / 1000,
-		Ticks: ticks,
+		WallClockTimeMs: getWallTimeMs(),
+		Ticks:           ticks,
 	}
 }
 
@@ -38,25 +42,26 @@ func (h *Hybrid) AddTicks(ticks int64) {
 func (h *Hybrid) Now() Hybrid {
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	currentTime := time.Now().UnixNano() / 1000
-	if h.Time >= currentTime {
+	currentTime := getWallTimeMs()
+	// NOTE: check in case clock goes backwards fue to synchronization
+	if h.WallClockTimeMs >= currentTime {
 		h.addTicks(1)
 	} else {
-		h.Time = currentTime
+		h.WallClockTimeMs = currentTime
 		h.Ticks = 0
 	}
 	return Hybrid{
-		Time:  h.Time,
-		Ticks: h.Ticks,
+		WallClockTimeMs: h.WallClockTimeMs,
+		Ticks:           h.Ticks,
 	}
 }
 
 func Compare(a, b *Hybrid) int {
-	if (a.Time == b.Time) && (a.Ticks == b.Ticks) {
+	if (a.WallClockTimeMs == b.WallClockTimeMs) && (a.Ticks == b.Ticks) {
 		return 0
 	}
-	if (a.Time == b.Time && a.Ticks > b.Ticks) ||
-		(a.Time > b.Time) {
+	if (a.WallClockTimeMs == b.WallClockTimeMs && a.Ticks > b.Ticks) ||
+		(a.WallClockTimeMs > b.WallClockTimeMs) {
 		return 1
 	}
 	return -1
@@ -65,7 +70,7 @@ func Compare(a, b *Hybrid) int {
 func max(times ...*Hybrid) *Hybrid {
 	maxTime := times[0]
 	for _, time := range times {
-		cmp := Compare(maxTime, time)
+		cmp := Compare(time, maxTime)
 		if cmp == 1 {
 			maxTime = time
 		}
@@ -79,6 +84,12 @@ func (h *Hybrid) Tick(requestTime *Hybrid) {
 	hybridNow := NewNow(-1)
 	latestTime := max(hybridNow, requestTime, h)
 	latestTime.addTicks(1)
-	h.Time = latestTime.Time
+	h.WallClockTimeMs = latestTime.WallClockTimeMs
 	h.Ticks = latestTime.Ticks
 }
+
+func (h *Hybrid) GetCompactTimestampMs() int64 {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return (h.WallClockTimeMs >> 16 << 16) | (h.Ticks << 48 >> 48)
+}

hlc/hlc_test.go

@@ -1,9 +1,9 @@
 package hlc
 
 import (
-	"time"
-	// "sync"
+	"sync"
 	"testing"
+	"time"
 )
 
 func TestCompare(t *testing.T) {
@@ -26,37 +26,38 @@ func TestCompare(t *testing.T) {
 	}
 }
 
-// func TestConcurrentAdd(t *testing.T) {
-// 	lt := New("a", 0)
-// 	wg := sync.WaitGroup{}
-// 	wg.Add(10)
-// 	for i := 0; i < 10; i++ {
-// 		go func() {
-// 			defer wg.Done()
-// 			lt.AddTicks(1)
-// 		}()
-// 	}
-// 	wg.Wait()
-// 	now := lt.Now()
-// 	if now.Ticks < 10 {
-// 		t.Fatal("There must be exactly 10 ticks")
-// 	}
-// }
+func TestConcurrentAdd(t *testing.T) {
+	currentWallTime := getWallTimeMs()
+	c1 := New(currentWallTime, 0)
+	c2 := New(currentWallTime, 1)
+	wg := sync.WaitGroup{}
+	wg.Add(10)
+	for i := 0; i < 10; i++ {
+		go func() {
+			defer wg.Done()
+			c1.AddTicks(1)
+		}()
+	}
+	wg.Wait()
+	cmp := Compare(c1, c2)
+	if cmp < 1 {
+		t.Fatal("Timestamp with more ticks must be greater")
+	}
+}
 
-// func TestConcurrentTick(t *testing.T) {
-// 	lt := New("a", 0)
-// 	req := New("b", 0)
-// 	wg := sync.WaitGroup{}
-// 	wg.Add(10)
-// 	for i := 0; i < 10; i++ {
-// 		go func() {
-// 			defer wg.Done()
-// 			lt.Tick(req)
-// 		}()
-// 	}
-// 	wg.Wait()
-// 	now := lt.Now()
-// 	if now.Ticks < 10 {
-// 		t.Fatal("There must be exactly 10 ticks")
-// 	}
-// }
+func TestConcurrentTick(t *testing.T) {
+	c := NewNow(0)
+	req := New(getWallTimeMs()+1000, 10)
+	wg := sync.WaitGroup{}
+	wg.Add(10)
+	for i := 0; i < 10; i++ {
+		go func() {
+			defer wg.Done()
+			c.Tick(req)
+		}()
+	}
+	wg.Wait()
+	if c.Ticks < 20 {
+		t.Fatal("There must be exactly 20 ticks")
+	}
+}

vector/vector.go

@@ -62,16 +62,18 @@ func (v *Vector) Tick(requestTime *Vector) error {
 }
 
 func Compare(a, b *Vector) int {
-	equal, less := 0, 0
+	equal, less, greater := 0, 0, 0
 	for i := range a.Ticks {
 		if a.Ticks[i] == b.Ticks[i] {
 			equal++
 		} else if a.Ticks[i] < b.Ticks[i] {
 			less++
+		} else {
+			greater++
 		}
 	}
-	if len(a.Ticks) == equal {
-		return 0
+	if len(a.Ticks) == equal || (greater > 0 && less > 0 && greater == less) {
+		return 0 // == concurrent
 	}
 	if len(a.Ticks) == (equal + less) {
 		return -1

vector/vector_test.go

@@ -17,6 +17,12 @@ func TestCompare(t *testing.T) {
 	if cmp < 1 {
 		t.Fatal("Original timestamp now must be higher than it's copy")
 	}
+	_, c1 := New(1, 3)
+	c1.AddTicks(1)
+	cmp = Compare(c, c1)
+	if cmp != 0 {
+		t.Fatalf("Timestamps must be concurrent: %v != %v", c, c1)
+	}
 }
 
 func TestConcurrentAdd(t *testing.T) {