forked from armon/go-chord
-
Notifications
You must be signed in to change notification settings - Fork 2
/
vnode.go
356 lines (305 loc) · 8.31 KB
/
vnode.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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
package chord
import (
"encoding/binary"
"fmt"
"log"
"time"
)
type Vnode struct {
Id []byte
Host string
}
func (vn *Vnode) String() string {
return fmt.Sprintf("%x", vn.Id)
}
// Initializes a local vnode
func (vn *localVnode) init(idx int) {
// Generate an ID
vn.genId(uint16(idx))
// Set our host
vn.Host = vn.ring.config.Hostname
// Initialize all state
vn.successors = make([]*Vnode, vn.ring.config.NumSuccessors)
vn.finger = make([]*Vnode, vn.ring.config.hashBits)
// Register with the RPC mechanism
vn.ring.transport.Register(&vn.Vnode, vn)
}
// Schedules the Vnode to do regular maintenence
func (vn *localVnode) schedule() {
// Setup our stabilize timer
vn.timer = time.AfterFunc(randStabilize(vn.ring.config), vn.stabilize)
}
// Generates an ID for the node
func (vn *localVnode) genId(idx uint16) {
// Use the hash funciton
conf := vn.ring.config
hash := conf.HashFunc()
hash.Write([]byte(conf.Hostname))
binary.Write(hash, binary.BigEndian, idx)
// Use the hash as the ID
vn.Id = hash.Sum(nil)
}
// Called to periodically stabilize the vnode
func (vn *localVnode) stabilize() {
// Clear the timer
vn.timer = nil
// Check for shutdown
if vn.ring.shutdown != nil {
vn.ring.shutdown <- true
return
}
// Setup the next stabilize timer
defer vn.schedule()
// Check for new successor
if err := vn.checkNewSuccessor(); err != nil {
log.Printf("[ERR] Error checking for new successor: %s", err)
}
// Notify the successor
if err := vn.notifySuccessor(); err != nil {
log.Printf("[ERR] Error notifying successor: %s", err)
}
// Finger table fix up
if err := vn.fixFingerTable(); err != nil {
log.Printf("[ERR] Error fixing finger table: %s", err)
}
// Check the predecessor
if err := vn.checkPredecessor(); err != nil {
log.Printf("[ERR] Error checking predecessor: %s", err)
}
// Set the last stabilized time
vn.stabilized = time.Now()
}
// Checks for a new successor
func (vn *localVnode) checkNewSuccessor() error {
// Ask our successor for it's predecessor
trans := vn.ring.transport
CHECK_NEW_SUC:
succ := vn.successors[0]
if succ == nil {
panic("Node has no successor!")
}
maybe_suc, err := trans.GetPredecessor(succ)
if err != nil {
// Check if we have succ list, try to contact next live succ
known := vn.knownSuccessors()
if known > 1 {
for i := 0; i < known; i++ {
if alive, _ := trans.Ping(vn.successors[0]); !alive {
// Don't eliminate the last successor we know of
if i+1 == known {
return fmt.Errorf("All known successors dead!")
}
// Advance the successors list past the dead one
copy(vn.successors[0:], vn.successors[1:])
vn.successors[known-1-i] = nil
} else {
// Found live successor, check for new one
goto CHECK_NEW_SUC
}
}
}
return err
}
// Check if we should replace our successor
if maybe_suc != nil && between(vn.Id, succ.Id, maybe_suc.Id) {
// Check if new successor is alive before switching
alive, err := trans.Ping(maybe_suc)
if alive && err == nil {
copy(vn.successors[1:], vn.successors[0:len(vn.successors)-1])
vn.successors[0] = maybe_suc
} else {
return err
}
}
return nil
}
// RPC: Invoked to return out predecessor
func (vn *localVnode) GetPredecessor() (*Vnode, error) {
return vn.predecessor, nil
}
// Notifies our successor of us, updates successor list
func (vn *localVnode) notifySuccessor() error {
// Notify successor
succ := vn.successors[0]
succ_list, err := vn.ring.transport.Notify(succ, &vn.Vnode)
if err != nil {
return err
}
// Trim the successors list if too long
max_succ := vn.ring.config.NumSuccessors
if len(succ_list) > max_succ-1 {
succ_list = succ_list[:max_succ-1]
}
// Update local successors list
for idx, s := range succ_list {
if s == nil {
break
}
// Ensure we don't set ourselves as a successor!
if s == nil || s.String() == vn.String() {
break
}
vn.successors[idx+1] = s
}
return nil
}
// RPC: Notify is invoked when a Vnode gets notified
func (vn *localVnode) Notify(maybe_pred *Vnode) ([]*Vnode, error) {
// Check if we should update our predecessor
if vn.predecessor == nil || between(vn.predecessor.Id, vn.Id, maybe_pred.Id) {
// Inform the delegate
conf := vn.ring.config
old := vn.predecessor
vn.ring.invokeDelegate(func() {
conf.Delegate.NewPredecessor(&vn.Vnode, maybe_pred, old)
})
vn.predecessor = maybe_pred
}
// Return our successors list
return vn.successors, nil
}
// Fixes up the finger table
func (vn *localVnode) fixFingerTable() error {
// Determine the offset
hb := vn.ring.config.hashBits
offset := powerOffset(vn.Id, vn.lastFinger, hb)
// Find the successor
nodes, err := vn.FindSuccessors(1, offset)
if nodes == nil || len(nodes) == 0 || err != nil {
return err
}
node := nodes[0]
// Update the finger table
vn.finger[vn.lastFinger] = node
// Try to skip as many finger entries as possible
for {
next := vn.lastFinger + 1
if next >= hb {
break
}
offset := powerOffset(vn.Id, next, hb)
// While the node is the successor, update the finger entries
if betweenRightIncl(vn.Id, node.Id, offset) {
vn.finger[next] = node
vn.lastFinger = next
} else {
break
}
}
// Increment to the index to repair
if vn.lastFinger+1 == hb {
vn.lastFinger = 0
} else {
vn.lastFinger++
}
return nil
}
// Checks the health of our predecessor
func (vn *localVnode) checkPredecessor() error {
// Check predecessor
if vn.predecessor != nil {
res, err := vn.ring.transport.Ping(vn.predecessor)
if err != nil {
return err
}
// Predecessor is dead
if !res {
vn.predecessor = nil
}
}
return nil
}
// Finds next N successors. N must be <= NumSuccessors
func (vn *localVnode) FindSuccessors(n int, key []byte) ([]*Vnode, error) {
// Check if we are the immediate predecessor
if betweenRightIncl(vn.Id, vn.successors[0].Id, key) {
return vn.successors[:n], nil
}
// Try the closest preceeding nodes
cp := closestPreceedingVnodeIterator{}
cp.init(vn, key)
for {
// Get the next closest node
closest := cp.Next()
if closest == nil {
break
}
// Try that node, break on success
res, err := vn.ring.transport.FindSuccessors(closest, n, key)
if err == nil {
return res, nil
}
log.Printf("[ERR] Failed to contact %s. Got %s", closest.String(), err)
}
// Determine how many successors we know of
successors := vn.knownSuccessors()
// Check if the ID is between us and any non-immediate successors
for i := 1; i <= successors-n; i++ {
if betweenRightIncl(vn.Id, vn.successors[i].Id, key) {
remain := vn.successors[i:]
if len(remain) > n {
remain = remain[:n]
}
return remain, nil
}
}
// Checked all closer nodes and our successors!
return nil, fmt.Errorf("Exhausted all preceeding nodes!")
}
// Instructs the vnode to leave
func (vn *localVnode) leave() error {
// Inform the delegate we are leaving
conf := vn.ring.config
pred := vn.predecessor
succ := vn.successors[0]
vn.ring.invokeDelegate(func() {
conf.Delegate.Leaving(&vn.Vnode, pred, succ)
})
// Notify predecessor to advance to their next successor
var err error
trans := vn.ring.transport
if vn.predecessor != nil {
err = trans.SkipSuccessor(vn.predecessor, &vn.Vnode)
}
// Notify successor to clear old predecessor
err = mergeErrors(err, trans.ClearPredecessor(vn.successors[0], &vn.Vnode))
return err
}
// Used to clear our predecessor when a node is leaving
func (vn *localVnode) ClearPredecessor(p *Vnode) error {
if vn.predecessor != nil && vn.predecessor.String() == p.String() {
// Inform the delegate
conf := vn.ring.config
old := vn.predecessor
vn.ring.invokeDelegate(func() {
conf.Delegate.PredecessorLeaving(&vn.Vnode, old)
})
vn.predecessor = nil
}
return nil
}
// Used to skip a successor when a node is leaving
func (vn *localVnode) SkipSuccessor(s *Vnode) error {
// Skip if we have a match
if vn.successors[0].String() == s.String() {
// Inform the delegate
conf := vn.ring.config
old := vn.successors[0]
vn.ring.invokeDelegate(func() {
conf.Delegate.SuccessorLeaving(&vn.Vnode, old)
})
known := vn.knownSuccessors()
copy(vn.successors[0:], vn.successors[1:])
vn.successors[known-1] = nil
}
return nil
}
// Determine how many successors we know of
func (vn *localVnode) knownSuccessors() (successors int) {
for i := 0; i < len(vn.successors); i++ {
if vn.successors[i] != nil {
successors = i + 1
}
}
return
}