Skip to content

Commit 478b360

Browse files
author
Jordan Halterman
committed
Add package for operating on timestamps
1 parent f6c6cd2 commit 478b360

5 files changed

Lines changed: 759 additions & 0 deletions

File tree

‎pkg/time/composite.go‎

Lines changed: 230 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,230 @@
1+
// SPDX-FileCopyrightText: 2022-present Intel Corporation
2+
//
3+
// SPDX-License-Identifier: Apache-2.0
4+
5+
package time
6+
7+
import (
8+
runtimev1 "github.com/atomix/runtime/api/atomix/runtime/v1"
9+
"sync"
10+
)
11+
12+
const compositeSchemeName = "Composite"
13+
14+
// newCompositeScheme creates a new composite scheme
15+
func newCompositeScheme(schemes ...Scheme) Scheme {
16+
codecs := make([]Codec, len(schemes))
17+
for i, scheme := range schemes {
18+
codecs[i] = scheme.Codec()
19+
}
20+
return compositeScheme{
21+
schemes: schemes,
22+
codec: CompositeTimestampCodec{codecs},
23+
}
24+
}
25+
26+
type compositeScheme struct {
27+
schemes []Scheme
28+
codec Codec
29+
}
30+
31+
func (s compositeScheme) Name() string {
32+
return compositeSchemeName
33+
}
34+
35+
func (s compositeScheme) Codec() Codec {
36+
return s.codec
37+
}
38+
39+
func (s compositeScheme) NewClock() Clock {
40+
return NewCompositeClock(s.schemes...)
41+
}
42+
43+
// NewCompositeClock creates a new composite clock
44+
func NewCompositeClock(schemes ...Scheme) Clock {
45+
scheme := newCompositeScheme(schemes...)
46+
clocks := make([]Clock, len(schemes))
47+
timestamps := make([]Timestamp, len(schemes))
48+
for i, scheme := range schemes {
49+
clock := scheme.NewClock()
50+
clocks[i] = clock
51+
timestamps[i] = clock.Get()
52+
}
53+
return &CompositeClock{
54+
scheme: scheme,
55+
clocks: clocks,
56+
timestamp: NewCompositeTimestamp(timestamps...),
57+
}
58+
}
59+
60+
// CompositeClock is a clock that produces CompositeTimestamps
61+
type CompositeClock struct {
62+
scheme Scheme
63+
clocks []Clock
64+
timestamp Timestamp
65+
mu sync.RWMutex
66+
}
67+
68+
func (c *CompositeClock) Scheme() Scheme {
69+
return c.scheme
70+
}
71+
72+
func (c *CompositeClock) Get() Timestamp {
73+
c.mu.RLock()
74+
defer c.mu.RUnlock()
75+
return c.timestamp
76+
}
77+
78+
func (c *CompositeClock) Increment() Timestamp {
79+
timestamps := make([]Timestamp, len(c.clocks))
80+
for i, clock := range c.clocks {
81+
timestamps[i] = clock.Increment()
82+
}
83+
return NewCompositeTimestamp(timestamps...)
84+
}
85+
86+
func (c *CompositeClock) Update(t Timestamp) Timestamp {
87+
update, ok := t.(CompositeTimestamp)
88+
if !ok {
89+
panic("not a composite timestamp")
90+
}
91+
92+
timestamps := make([]Timestamp, len(c.clocks))
93+
for i, clock := range c.clocks {
94+
timestamps[i] = clock.Update(update.Timestamps[i])
95+
}
96+
97+
c.mu.Lock()
98+
defer c.mu.Unlock()
99+
c.timestamp = NewCompositeTimestamp(timestamps...)
100+
return c.timestamp
101+
}
102+
103+
// NewCompositeTimestamp creates a new composite Timestamp from the given set of Timestamps
104+
func NewCompositeTimestamp(timestamps ...Timestamp) Timestamp {
105+
schemes := make([]Scheme, len(timestamps))
106+
for i, timestamp := range timestamps {
107+
schemes[i] = timestamp.Scheme()
108+
}
109+
return CompositeTimestamp{
110+
scheme: newCompositeScheme(schemes...),
111+
Timestamps: timestamps,
112+
}
113+
}
114+
115+
// CompositeTimestamp is a composite Timestamp implementation
116+
type CompositeTimestamp struct {
117+
scheme Scheme
118+
Timestamps []Timestamp
119+
}
120+
121+
func (t CompositeTimestamp) Scheme() Scheme {
122+
return t.scheme
123+
}
124+
125+
func (t CompositeTimestamp) Before(u Timestamp) bool {
126+
v, ok := u.(CompositeTimestamp)
127+
if !ok {
128+
panic("not a composite timestamp")
129+
}
130+
if len(t.Timestamps) != len(v.Timestamps) {
131+
panic("incompatible composite timestamps")
132+
}
133+
for i := 0; i < len(t.Timestamps); i++ {
134+
t1 := t.Timestamps[i]
135+
t2 := v.Timestamps[i]
136+
if t1.Before(t2) {
137+
return true
138+
} else if i > 0 {
139+
for j := 0; j < i; j++ {
140+
v1 := t.Timestamps[j]
141+
v2 := v.Timestamps[j]
142+
if !v1.Equal(v2) {
143+
return false
144+
}
145+
}
146+
if !t1.Before(t2) {
147+
return false
148+
}
149+
}
150+
}
151+
return true
152+
}
153+
154+
func (t CompositeTimestamp) After(u Timestamp) bool {
155+
v, ok := u.(CompositeTimestamp)
156+
if !ok {
157+
panic("not a composite timestamp")
158+
}
159+
if len(t.Timestamps) != len(v.Timestamps) {
160+
panic("incompatible composite timestamps")
161+
}
162+
for i := 0; i < len(t.Timestamps); i++ {
163+
t1 := t.Timestamps[i]
164+
t2 := v.Timestamps[i]
165+
if t1.After(t2) {
166+
return true
167+
} else if i > 0 {
168+
for j := 0; j < i; j++ {
169+
v1 := t.Timestamps[j]
170+
v2 := v.Timestamps[j]
171+
if !v1.Equal(v2) {
172+
return false
173+
}
174+
}
175+
if !t1.After(t2) {
176+
return false
177+
}
178+
}
179+
}
180+
return true
181+
}
182+
183+
func (t CompositeTimestamp) Equal(u Timestamp) bool {
184+
v, ok := u.(CompositeTimestamp)
185+
if !ok {
186+
panic("not a composite timestamp")
187+
}
188+
if len(t.Timestamps) != len(v.Timestamps) {
189+
panic("incompatible composite timestamps")
190+
}
191+
for i := 0; i < len(t.Timestamps); i++ {
192+
t1 := t.Timestamps[i]
193+
t2 := v.Timestamps[i]
194+
if !t1.Equal(t2) {
195+
return false
196+
}
197+
}
198+
return true
199+
}
200+
201+
// CompositeTimestampCodec is a codec for Composite timestamps
202+
type CompositeTimestampCodec struct {
203+
codecs []Codec
204+
}
205+
206+
func (c CompositeTimestampCodec) EncodeTimestamp(timestamp Timestamp) runtimev1.Timestamp {
207+
t, ok := timestamp.(CompositeTimestamp)
208+
if !ok {
209+
panic("expected CompositeTimestamp")
210+
}
211+
timestamps := make([]runtimev1.Timestamp, 0, len(t.Timestamps))
212+
for _, timestamp := range t.Timestamps {
213+
timestamps = append(timestamps, timestamp.Scheme().Codec().EncodeTimestamp(timestamp))
214+
}
215+
return runtimev1.Timestamp{
216+
Timestamp: &runtimev1.Timestamp_CompositeTimestamp{
217+
CompositeTimestamp: &runtimev1.CompositeTimestamp{
218+
Timestamps: timestamps,
219+
},
220+
},
221+
}
222+
}
223+
224+
func (c CompositeTimestampCodec) DecodeTimestamp(timestamp runtimev1.Timestamp) (Timestamp, error) {
225+
timestamps := make([]Timestamp, 0, len(timestamp.GetCompositeTimestamp().Timestamps))
226+
for _, timestamp := range timestamp.GetCompositeTimestamp().Timestamps {
227+
timestamps = append(timestamps, NewTimestamp(timestamp))
228+
}
229+
return NewCompositeTimestamp(timestamps...), nil
230+
}

‎pkg/time/epoch.go‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
// SPDX-FileCopyrightText: 2022-present Intel Corporation
2+
//
3+
// SPDX-License-Identifier: Apache-2.0
4+
5+
package time
6+
7+
import (
8+
runtimev1 "github.com/atomix/runtime/api/atomix/runtime/v1"
9+
"sync"
10+
)
11+
12+
const epochSchemeName = "Epoch"
13+
14+
var EpochScheme = newEpochScheme()
15+
16+
// newEpochScheme creates a new Epoch scheme
17+
func newEpochScheme() Scheme {
18+
return epochScheme{
19+
codec: EpochTimestampCodec{},
20+
}
21+
}
22+
23+
type epochScheme struct {
24+
codec Codec
25+
}
26+
27+
func (s epochScheme) Name() string {
28+
return epochSchemeName
29+
}
30+
31+
func (s epochScheme) Codec() Codec {
32+
return s.codec
33+
}
34+
35+
func (s epochScheme) NewClock() Clock {
36+
return NewEpochClock()
37+
}
38+
39+
// NewEpochClock creates a new epoch clock
40+
func NewEpochClock() Clock {
41+
return &EpochClock{
42+
timestamp: NewEpochTimestamp(Epoch(0), LogicalTime(0)).(EpochTimestamp),
43+
}
44+
}
45+
46+
// EpochClock is a clock that produces EpochTimestamps
47+
type EpochClock struct {
48+
timestamp EpochTimestamp
49+
mu sync.RWMutex
50+
}
51+
52+
func (c *EpochClock) Scheme() Scheme {
53+
return EpochScheme
54+
}
55+
56+
func (c *EpochClock) Get() Timestamp {
57+
c.mu.RLock()
58+
defer c.mu.RUnlock()
59+
return c.timestamp
60+
}
61+
62+
func (c *EpochClock) Increment() Timestamp {
63+
c.mu.Lock()
64+
defer c.mu.Unlock()
65+
c.timestamp = NewEpochTimestamp(c.timestamp.Epoch, c.timestamp.Time+1).(EpochTimestamp)
66+
return c.timestamp
67+
}
68+
69+
func (c *EpochClock) Update(t Timestamp) Timestamp {
70+
update, ok := t.(EpochTimestamp)
71+
if !ok {
72+
panic("not a logical timestamp")
73+
}
74+
75+
c.mu.RLock()
76+
current := c.timestamp
77+
c.mu.RUnlock()
78+
if !update.After(current) {
79+
return current
80+
}
81+
82+
c.mu.Lock()
83+
defer c.mu.Unlock()
84+
if update.After(c.timestamp) {
85+
c.timestamp = update
86+
}
87+
return c.timestamp
88+
}
89+
90+
type Epoch uint64
91+
92+
// NewEpochTimestamp creates a new Epoch based Timestamp
93+
func NewEpochTimestamp(epoch Epoch, time LogicalTime) Timestamp {
94+
return EpochTimestamp{
95+
Epoch: epoch,
96+
Time: time,
97+
}
98+
}
99+
100+
// EpochTimestamp is a Timestamp based on an Epoch and LogicalTime
101+
type EpochTimestamp struct {
102+
Epoch Epoch
103+
Time LogicalTime
104+
}
105+
106+
func (t EpochTimestamp) Scheme() Scheme {
107+
return EpochScheme
108+
}
109+
110+
func (t EpochTimestamp) Before(u Timestamp) bool {
111+
v, ok := u.(EpochTimestamp)
112+
if !ok {
113+
panic("not an epoch timestamp")
114+
}
115+
return t.Epoch < v.Epoch || (t.Epoch == v.Epoch && t.Time < v.Time)
116+
}
117+
118+
func (t EpochTimestamp) After(u Timestamp) bool {
119+
v, ok := u.(EpochTimestamp)
120+
if !ok {
121+
panic("not an epoch timestamp")
122+
}
123+
return t.Epoch > v.Epoch || (t.Epoch == v.Epoch && t.Time > v.Time)
124+
}
125+
126+
func (t EpochTimestamp) Equal(u Timestamp) bool {
127+
v, ok := u.(EpochTimestamp)
128+
if !ok {
129+
panic("not an epoch timestamp")
130+
}
131+
return t.Epoch == v.Epoch && t.Time == v.Time
132+
}
133+
134+
// EpochTimestampCodec is a codec for epoch timestamps
135+
type EpochTimestampCodec struct{}
136+
137+
func (c EpochTimestampCodec) EncodeTimestamp(timestamp Timestamp) runtimev1.Timestamp {
138+
t, ok := timestamp.(EpochTimestamp)
139+
if !ok {
140+
panic("expected EpochTimestamp")
141+
}
142+
return runtimev1.Timestamp{
143+
Timestamp: &runtimev1.Timestamp_EpochTimestamp{
144+
EpochTimestamp: &runtimev1.EpochTimestamp{
145+
Epoch: runtimev1.Epoch(t.Epoch),
146+
Time: runtimev1.LogicalTime(t.Time),
147+
},
148+
},
149+
}
150+
}
151+
152+
func (c EpochTimestampCodec) DecodeTimestamp(timestamp runtimev1.Timestamp) (Timestamp, error) {
153+
return NewEpochTimestamp(Epoch(timestamp.GetEpochTimestamp().Epoch), LogicalTime(timestamp.GetEpochTimestamp().Time)), nil
154+
}

0 commit comments

Comments
 (0)