forked from etcd-io/etcd
-
Notifications
You must be signed in to change notification settings - Fork 0
/
event_history.go
112 lines (89 loc) · 2.48 KB
/
event_history.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
package store
import (
"fmt"
"strings"
"sync"
etcdErr "github.com/coreos/etcd/error"
)
type EventHistory struct {
Queue eventQueue
StartIndex uint64
LastIndex uint64
LastTerm uint64
DupCnt uint64 // help to compute the watching point with duplicated indexes in the queue
rwl sync.RWMutex
}
func newEventHistory(capacity int) *EventHistory {
return &EventHistory{
Queue: eventQueue{
Capacity: capacity,
Events: make([]*Event, capacity),
},
}
}
// addEvent function adds event into the eventHistory
func (eh *EventHistory) addEvent(e *Event) *Event {
eh.rwl.Lock()
defer eh.rwl.Unlock()
var duped uint64
if e.Index == UndefIndex {
e.Index = eh.LastIndex
e.Term = eh.LastTerm
duped = 1
}
eh.Queue.insert(e)
eh.LastIndex = e.Index
eh.LastTerm = e.Term
eh.DupCnt += duped
eh.StartIndex = eh.Queue.Events[eh.Queue.Front].Index
return e
}
// scan function is enumerating events from the index in history and
// stops till the first point where the key has identified prefix
func (eh *EventHistory) scan(prefix string, index uint64) (*Event, *etcdErr.Error) {
eh.rwl.RLock()
defer eh.rwl.RUnlock()
start := index - eh.StartIndex
// the index should locate after the event history's StartIndex
if start < 0 {
return nil,
etcdErr.NewError(etcdErr.EcodeEventIndexCleared,
fmt.Sprintf("the requested history has been cleared [%v/%v]",
eh.StartIndex, index), UndefIndex, UndefTerm)
}
// the index should locate before the size of the queue minus the duplicate count
if start >= (uint64(eh.Queue.Size) - eh.DupCnt) { // future index
return nil, nil
}
i := int((start + uint64(eh.Queue.Front)) % uint64(eh.Queue.Capacity))
for {
e := eh.Queue.Events[i]
if strings.HasPrefix(e.Key, prefix) && index <= e.Index { // make sure we bypass the smaller one
return e, nil
}
i = (i + 1) % eh.Queue.Capacity
if i == eh.Queue.back() { // find nothing, return and watch from current index
return nil, nil
}
}
}
// clone will be protected by a stop-world lock
// do not need to obtain internal lock
func (eh *EventHistory) clone() *EventHistory {
clonedQueue := eventQueue{
Capacity: eh.Queue.Capacity,
Events: make([]*Event, eh.Queue.Capacity),
Size: eh.Queue.Size,
Front: eh.Queue.Front,
}
for i, e := range eh.Queue.Events {
clonedQueue.Events[i] = e
}
return &EventHistory{
StartIndex: eh.StartIndex,
Queue: clonedQueue,
LastIndex: eh.LastIndex,
LastTerm: eh.LastTerm,
DupCnt: eh.DupCnt,
}
}