-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- there is more locking than seems necessary here, but I don't think this will be in contention, especially becuase these are inside SQL transactions that are happening one at a time - each bucket maintains it's own HLC, as it is stored inside the bucket db, and so needs to be re-initialized from the last possible clock that was serialized.
- Loading branch information
Showing
6 changed files
with
211 additions
and
34 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| // Copyright 2023-Present Couchbase, Inc. | ||
| // | ||
| // Use of this software is governed by the Business Source License included | ||
| // in the file licenses/BSL-Couchbase.txt. As of the Change Date specified | ||
| // in that file, in accordance with the Business Source License, use of this | ||
| // software will be governed by the Apache License, Version 2.0, included in | ||
| // the file licenses/APL2.txt. | ||
|
|
||
| package rosmar | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
| ) | ||
|
|
||
| type timestamp uint64 | ||
|
|
||
| // hybridLogicalClock is a hybrid logical clock implementation for rosmar that produces timestamps that will always be increasing regardless of clock changes. | ||
| type hybridLogicalClock struct { | ||
| clock clock | ||
| counter timestamp | ||
| highestTime timestamp | ||
| mutex sync.Mutex | ||
| } | ||
|
|
||
| // clock interface is used to abstract the system clock for testing purposes. | ||
| type clock interface { | ||
| // getTime returns the current time in nanoseconds. | ||
| getTime() timestamp | ||
| } | ||
|
|
||
| type systemClock struct{} | ||
|
|
||
| // getTime returns the current time in nanoseconds. | ||
| func (c *systemClock) getTime() timestamp { | ||
| return timestamp(time.Now().UnixNano()) | ||
| } | ||
|
|
||
| // NewHybridLogicalClock returns a new HLC from a previously initialized time. | ||
| func NewHybridLogicalClock(lastTime timestamp) *hybridLogicalClock { | ||
| return &hybridLogicalClock{ | ||
| highestTime: lastTime, | ||
| clock: &systemClock{}, | ||
| } | ||
| } | ||
|
|
||
| // Now returns the next time represented in nanoseconds. This can be the current timestamp, or if multiple occur in the same nanosecond, an increasing timestamp. | ||
| func (c *hybridLogicalClock) Now() timestamp { | ||
| c.mutex.Lock() | ||
| defer c.mutex.Unlock() | ||
| physicalTime := c.clock.getTime() | ||
| if c.highestTime >= physicalTime { | ||
| c.counter++ | ||
| } else { | ||
| c.counter = 0 | ||
| c.highestTime = physicalTime | ||
| } | ||
| return c.highestTime + c.counter | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| // Copyright 2023-Present Couchbase, Inc. | ||
| // | ||
| // Use of this software is governed by the Business Source License included | ||
| // in the file licenses/BSL-Couchbase.txt. As of the Change Date specified | ||
| // in that file, in accordance with the Business Source License, use of this | ||
| // software will be governed by the Apache License, Version 2.0, included in | ||
| // the file licenses/APL2.txt. | ||
|
|
||
| package rosmar | ||
|
|
||
| import ( | ||
| "sync" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestHybridLogicalClockNow(t *testing.T) { | ||
| clock := hybridLogicalClock{clock: &systemClock{}} | ||
| timestamp1 := clock.Now() | ||
| timestamp2 := clock.Now() | ||
| require.Greater(t, timestamp2, timestamp1) | ||
| } | ||
|
|
||
| func generateTimestamps(wg *sync.WaitGroup, clock *hybridLogicalClock, n int, result chan []timestamp) { | ||
| defer wg.Done() | ||
| timestamps := make([]timestamp, n) | ||
| for i := 0; i < n; i++ { | ||
| timestamps[i] = clock.Now() | ||
| } | ||
| result <- timestamps | ||
| } | ||
|
|
||
| func TestHLCNowConcurrent(t *testing.T) { | ||
| clock := hybridLogicalClock{clock: &systemClock{}} | ||
| goroutines := 100 | ||
| timestampCount := 100 | ||
|
|
||
| wg := sync.WaitGroup{} | ||
| results := make(chan []timestamp) | ||
| for i := 0; i < goroutines; i++ { | ||
| wg.Add(1) | ||
| go generateTimestamps(&wg, &clock, timestampCount, results) | ||
| } | ||
|
|
||
| doneChan := make(chan struct{}) | ||
| go func() { | ||
| wg.Wait() | ||
| doneChan <- struct{}{} | ||
| }() | ||
| allTimestamps := make([]timestamp, 0, goroutines*timestampCount) | ||
| loop: | ||
| for { | ||
| select { | ||
| case timestamps := <-results: | ||
| allTimestamps = append(allTimestamps, timestamps...) | ||
| case <-doneChan: | ||
| break loop | ||
| } | ||
| } | ||
| uniqueTimestamps := make(map[timestamp]struct{}) | ||
| for _, timestamp := range allTimestamps { | ||
| if _, ok := uniqueTimestamps[timestamp]; ok { | ||
| t.Errorf("Timestamp %d is not unique", timestamp) | ||
| } | ||
| uniqueTimestamps[timestamp] = struct{}{} | ||
| } | ||
| } | ||
|
|
||
| type fakeClock struct { | ||
| time timestamp | ||
| } | ||
|
|
||
| func (c *fakeClock) getTime() timestamp { | ||
| return c.time | ||
| } | ||
|
|
||
| func TestHLCReverseTime(t *testing.T) { | ||
| clock := &fakeClock{} | ||
| hlc := hybridLogicalClock{clock: clock} | ||
| require.Equal(t, timestamp(1), hlc.Now()) | ||
| require.Equal(t, timestamp(2), hlc.Now()) | ||
|
|
||
| // reverse time no counter | ||
| clock.time = timestamp(0) | ||
| require.Equal(t, timestamp(3), hlc.Now()) | ||
|
|
||
| // reset time to normal | ||
| clock.time = timestamp(6) | ||
| require.Equal(t, timestamp(6), hlc.Now()) | ||
|
|
||
| // reverse time again | ||
| clock.time = timestamp(1) | ||
| require.Equal(t, timestamp(7), hlc.Now()) | ||
|
|
||
| // jump to a value we had previously | ||
| clock.time = timestamp(6) | ||
| require.Equal(t, int(timestamp(8)), int(hlc.Now())) | ||
| require.Equal(t, int(timestamp(9)), int(hlc.Now())) | ||
| } |