routing/dht: periodic bootstrapping #572
Juan Batiz-Benet committed
Jan 16, 2015 at 12:52 UTC
82d38a269237895150a2c2207f189e440b85883f
4 files changed
+328
-102
core/bootstrap.go
+1
-1
@@ -86,7 +86,7 @@ func bootstrap(ctx context.Context,
86
87
// we can try running dht bootstrap even if we're connected to all bootstrap peers.
88
if len(h.Network().Conns()) > 0 {
89
- if err := r.Bootstrap(ctx, numDHTBootstrapQueries); err != nil {
89
+ if _, err := r.Bootstrap(); err != nil {
90
// log this as Info. later on, discern better between errors.
91
log.Infof("dht bootstrap err: %s", err)
92
return nil
routing/dht/dht.go
-63
@@ -370,66 +370,3 @@ func (dht *IpfsDHT) PingRoutine(t time.Duration) {
370
}
371
}
372
}
373
-
374
-// Bootstrap builds up list of peers by requesting random peer IDs
375
-func (dht *IpfsDHT) Bootstrap(ctx context.Context, queries int) error {
376
- var merr u.MultiErr
377
-
378
- randomID := func() peer.ID {
379
- // 16 random bytes is not a valid peer id. it may be fine becuase
380
- // the dht will rehash to its own keyspace anyway.
381
- id := make([]byte, 16)
382
- rand.Read(id)
383
- return peer.ID(id)
384
- }
385
-
386
- // bootstrap sequentially, as results will compound
387
- runQuery := func(ctx context.Context, id peer.ID) {
388
- p, err := dht.FindPeer(ctx, id)
389
- if err == routing.ErrNotFound {
390
- // this isn't an error. this is precisely what we expect.
391
- } else if err != nil {
392
- merr = append(merr, err)
393
- } else {
394
- // woah, actually found a peer with that ID? this shouldn't happen normally
395
- // (as the ID we use is not a real ID). this is an odd error worth logging.
396
- err := fmt.Errorf("Bootstrap peer error: Actually FOUND peer. (%s, %s)", id, p)
397
- log.Errorf("%s", err)
398
- merr = append(merr, err)
399
- }
400
- }
401
-
402
- sequential := true
403
- if sequential {
404
- // these should be parallel normally. but can make them sequential for debugging.
405
- // note that the core/bootstrap context deadline should be extended too for that.
406
- for i := 0; i < queries; i++ {
407
- id := randomID()
408
- log.Debugf("Bootstrapping query (%d/%d) to random ID: %s", i+1, queries, id)
409
- runQuery(ctx, id)
410
- }
411
-
412
- } else {
413
- // note on parallelism here: the context is passed in to the queries, so they
414
- // **should** exit when it exceeds, making this function exit on ctx cancel.
415
- // normally, we should be selecting on ctx.Done() here too, but this gets
416
- // complicated to do with WaitGroup, and doesnt wait for the children to exit.
417
- var wg sync.WaitGroup
418
- for i := 0; i < queries; i++ {
419
- wg.Add(1)
420
- go func() {
421
- defer wg.Done()
422
-
423
- id := randomID()
424
- log.Debugf("Bootstrapping query (%d/%d) to random ID: %s", i+1, queries, id)
425
- runQuery(ctx, id)
426
- }()
427
- }
428
- wg.Wait()
429
- }
430
-
431
- if len(merr) > 0 {
432
- return merr
433
- }
434
- return nil
435
-}
routing/dht/dht_bootstrap.go
new
+181
@@ -0,0 +1,181 @@
1
+// Package dht implements a distributed hash table that satisfies the ipfs routing
2
+// interface. This DHT is modeled after kademlia with Coral and S/Kademlia modifications.
3
+package dht
4
+
5
+import (
6
+ "crypto/rand"
7
+ "fmt"
8
+ "sync"
9
+ "time"
10
+
11
+ peer "github.com/jbenet/go-ipfs/p2p/peer"
12
+ routing "github.com/jbenet/go-ipfs/routing"
13
+ u "github.com/jbenet/go-ipfs/util"
14
+
15
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
16
+ goprocess "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
17
+)
18
+
19
+// DefaultBootstrapQueries specifies how many queries to run,
20
+// if the user does not specify a different number as an option.
21
+//
22
+// For now, this is set to 16 queries, which is an aggressive number.
23
+// We are currently more interested in ensuring we have a properly formed
24
+// DHT than making sure our dht minimizes traffic. Once we are more certain
25
+// of our implementation's robustness, we should lower this down to 8 or 4.
26
+//
27
+// Note there is also a tradeoff between the bootstrap period and the number
28
+// of queries. We could support a higher period with a smaller number of
29
+// queries
30
+const DefaultBootstrapQueries = 16
31
+
32
+// DefaultBootstrapPeriod specifies how often to periodically run bootstrap,
33
+// if the user does not specify a different number as an option.
34
+//
35
+// For now, this is set to 10 seconds, which is an aggressive period. We are
36
+// We are currently more interested in ensuring we have a properly formed
37
+// DHT than making sure our dht minimizes traffic. Once we are more certain
38
+// implementation's robustness, we should lower this down to 30s or 1m.
39
+//
40
+// Note there is also a tradeoff between the bootstrap period and the number
41
+// of queries. We could support a higher period with a smaller number of
42
+// queries
43
+const DefaultBootstrapPeriod = time.Duration(10 * time.Second)
44
+
45
+// Bootstrap runs bootstrapping once, then calls SignalBootstrap with default
46
+// parameters: DefaultBootstrapQueries and DefaultBootstrapPeriod. This allows
47
+// the user to catch an error off the bat if the connections are faulty. It also
48
+// allows BootstrapOnSignal not to run bootstrap at the beginning, which is useful
49
+// for instrumenting it on tests, or delaying bootstrap until the network is online
50
+// and connected to at least a few nodes.
51
+//
52
+// Like PeriodicBootstrap, Bootstrap returns a process, so the user can stop it.
53
+func (dht *IpfsDHT) Bootstrap() (goprocess.Process, error) {
54
+
55
+ if err := dht.runBootstrap(dht.Context(), DefaultBootstrapQueries); err != nil {
56
+ return nil, err
57
+ }
58
+
59
+ sig := time.Tick(DefaultBootstrapPeriod)
60
+ return dht.BootstrapOnSignal(DefaultBootstrapQueries, sig)
61
+}
62
+
63
+// SignalBootstrap ensures the dht routing table remains healthy as peers come and go.
64
+// it builds up a list of peers by requesting random peer IDs. The Bootstrap
65
+// process will run a number of queries each time, and run every time signal fires.
66
+// These parameters are configurable.
67
+//
68
+// SignalBootstrap returns a process, so the user can stop it.
69
+func (dht *IpfsDHT) BootstrapOnSignal(queries int, signal <-chan time.Time) (goprocess.Process, error) {
70
+ if queries <= 0 {
71
+ return nil, fmt.Errorf("invalid number of queries: %d", queries)
72
+ }
73
+
74
+ if signal == nil {
75
+ return nil, fmt.Errorf("invalid signal: %v", signal)
76
+ }
77
+
78
+ proc := goprocess.Go(func(worker goprocess.Process) {
79
+ for {
80
+ select {
81
+ case <-worker.Closing():
82
+ log.Debug("dht bootstrapper shutting down")
83
+ return
84
+
85
+ case <-signal:
86
+ // it would be useful to be able to send out signals of when we bootstrap, too...
87
+ // maybe this is a good case for whole module event pub/sub?
88
+
89
+ ctx := dht.Context()
90
+ if err := dht.runBootstrap(ctx, queries); err != nil {
91
+ log.Error(err)
92
+ // A bootstrapping error is important to notice but not fatal.
93
+ // maybe the client should be able to consume these errors,
94
+ // though I dont have a clear use case in mind-- what **could**
95
+ // the client do if one of the bootstrap calls fails?
96
+ //
97
+ // This is also related to the core's bootstrap failures.
98
+ // superviseConnections should perhaps allow clients to detect
99
+ // bootstrapping problems.
100
+ //
101
+ // Anyway, passing errors could be done with a bootstrapper object.
102
+ // this would imply the client should be able to consume a lot of
103
+ // other non-fatal dht errors too. providing this functionality
104
+ // should be done correctly DHT-wide.
105
+ // NB: whatever the design, clients must ensure they drain errors!
106
+ // This pattern is common to many things, perhaps long-running services
107
+ // should have something like an ErrStream that allows clients to consume
108
+ // periodic errors and take action. It should allow the user to also
109
+ // ignore all errors with something like an ErrStreamDiscard. We should
110
+ // study what other systems do for ideas.
111
+ }
112
+ }
113
+ }
114
+ })
115
+
116
+ return proc, nil
117
+}
118
+
119
+// runBootstrap builds up list of peers by requesting random peer IDs
120
+func (dht *IpfsDHT) runBootstrap(ctx context.Context, queries int) error {
121
+
122
+ var merr u.MultiErr
123
+
124
+ randomID := func() peer.ID {
125
+ // 16 random bytes is not a valid peer id. it may be fine becuase
126
+ // the dht will rehash to its own keyspace anyway.
127
+ id := make([]byte, 16)
128
+ rand.Read(id)
129
+ return peer.ID(id)
130
+ }
131
+
132
+ // bootstrap sequentially, as results will compound
133
+ runQuery := func(ctx context.Context, id peer.ID) {
134
+ p, err := dht.FindPeer(ctx, id)
135
+ if err == routing.ErrNotFound {
136
+ // this isn't an error. this is precisely what we expect.
137
+ } else if err != nil {
138
+ merr = append(merr, err)
139
+ } else {
140
+ // woah, actually found a peer with that ID? this shouldn't happen normally
141
+ // (as the ID we use is not a real ID). this is an odd error worth logging.
142
+ err := fmt.Errorf("Bootstrap peer error: Actually FOUND peer. (%s, %s)", id, p)
143
+ log.Errorf("%s", err)
144
+ merr = append(merr, err)
145
+ }
146
+ }
147
+
148
+ sequential := true
149
+ if sequential {
150
+ // these should be parallel normally. but can make them sequential for debugging.
151
+ // note that the core/bootstrap context deadline should be extended too for that.
152
+ for i := 0; i < queries; i++ {
153
+ id := randomID()
154
+ log.Debugf("Bootstrapping query (%d/%d) to random ID: %s", i+1, queries, id)
155
+ runQuery(ctx, id)
156
+ }
157
+
158
+ } else {
159
+ // note on parallelism here: the context is passed in to the queries, so they
160
+ // **should** exit when it exceeds, making this function exit on ctx cancel.
161
+ // normally, we should be selecting on ctx.Done() here too, but this gets
162
+ // complicated to do with WaitGroup, and doesnt wait for the children to exit.
163
+ var wg sync.WaitGroup
164
+ for i := 0; i < queries; i++ {
165
+ wg.Add(1)
166
+ go func() {
167
+ defer wg.Done()
168
+
169
+ id := randomID()
170
+ log.Debugf("Bootstrapping query (%d/%d) to random ID: %s", i+1, queries, id)
171
+ runQuery(ctx, id)
172
+ }()
173
+ }
174
+ wg.Wait()
175
+ }
176
+
177
+ if len(merr) > 0 {
178
+ return merr
179
+ }
180
+ return nil
181
+}
routing/dht/dht_test.go
+146
-38
@@ -75,25 +75,20 @@ func connect(t *testing.T, ctx context.Context, a, b *IpfsDHT) {
75
func bootstrap(t *testing.T, ctx context.Context, dhts []*IpfsDHT) {
76
77
ctx, cancel := context.WithCancel(ctx)
78
+ log.Error("hmm")
79
+ defer log.Error("hmm end")
80
+ log.Debugf("bootstrapping dhts...")
81
79
- rounds := 1
82
+ // tried async. sequential fares much better. compare:
83
+ // 100 async https://gist.github.com/jbenet/56d12f0578d5f34810b2
84
+ // 100 sync https://gist.github.com/jbenet/6c59e7c15426e48aaedd
85
+ // probably because results compound
86
81
- for i := 0; i < rounds; i++ {
82
- log.Debugf("bootstrapping round %d/%d\n", i, rounds)
83
-
84
- // tried async. sequential fares much better. compare:
85
- // 100 async https://gist.github.com/jbenet/56d12f0578d5f34810b2
86
- // 100 sync https://gist.github.com/jbenet/6c59e7c15426e48aaedd
87
- // probably because results compound
88
-
89
- start := rand.Intn(len(dhts)) // randomize to decrease bias.
90
- for i := range dhts {
91
- dht := dhts[(start+i)%len(dhts)]
92
- log.Debugf("bootstrapping round %d/%d -- %s\n", i, rounds, dht.self)
93
- dht.Bootstrap(ctx, 3)
94
- }
87
+ start := rand.Intn(len(dhts)) // randomize to decrease bias.
88
+ for i := range dhts {
89
+ dht := dhts[(start+i)%len(dhts)]
90
+ dht.runBootstrap(ctx, 3)
91
}
96
-
92
cancel()
93
}
94
@@ -235,6 +230,53 @@ func TestProvides(t *testing.T) {
230
}
231
}
232
233
+// if minPeers or avgPeers is 0, dont test for it.
234
+func waitForWellFormedTables(t *testing.T, dhts []*IpfsDHT, minPeers, avgPeers int, timeout time.Duration) bool {
235
+ // test "well-formed-ness" (>= minPeers peers in every routing table)
236
+
237
+ checkTables := func() bool {
238
+ totalPeers := 0
239
+ for _, dht := range dhts {
240
+ rtlen := dht.routingTable.Size()
241
+ totalPeers += rtlen
242
+ if minPeers > 0 && rtlen < minPeers {
243
+ t.Logf("routing table for %s only has %d peers (should have >%d)", dht.self, rtlen, minPeers)
244
+ return false
245
+ }
246
+ }
247
+ actualAvgPeers := totalPeers / len(dhts)
248
+ t.Logf("avg rt size: %d", actualAvgPeers)
249
+ if avgPeers > 0 && actualAvgPeers < avgPeers {
250
+ t.Logf("avg rt size: %d < %d", actualAvgPeers, avgPeers)
251
+ return false
252
+ }
253
+ return true
254
+ }
255
+
256
+ timeoutA := time.After(timeout)
257
+ for {
258
+ select {
259
+ case <-timeoutA:
260
+ log.Error("did not reach well-formed routing tables by %s", timeout)
261
+ return false // failed
262
+ case <-time.After(5 * time.Millisecond):
263
+ if checkTables() {
264
+ return true // succeeded
265
+ }
266
+ }
267
+ }
268
+}
269
+
270
+func printRoutingTables(dhts []*IpfsDHT) {
271
+ // the routing tables should be full now. let's inspect them.
272
+ fmt.Println("checking routing table of %d", len(dhts))
273
+ for _, dht := range dhts {
274
+ fmt.Printf("checking routing table of %s\n", dht.self)
275
+ dht.routingTable.Print()
276
+ fmt.Println("")
277
+ }
278
+}
279
+
280
func TestBootstrap(t *testing.T) {
281
// t.Skip("skipping test to debug another")
282
if testing.Short() {
@@ -258,38 +300,105 @@ func TestBootstrap(t *testing.T) {
300
}
301
302
<-time.After(100 * time.Millisecond)
261
- t.Logf("bootstrapping them so they find each other", nDHTs)
262
- ctxT, _ := context.WithTimeout(ctx, 5*time.Second)
263
- bootstrap(t, ctxT, dhts)
303
+ // bootstrap a few times until we get good tables.
304
+ stop := make(chan struct{})
305
+ go func() {
306
+ for {
307
+ t.Logf("bootstrapping them so they find each other", nDHTs)
308
+ ctxT, _ := context.WithTimeout(ctx, 5*time.Second)
309
+ bootstrap(t, ctxT, dhts)
310
+
311
+ select {
312
+ case <-time.After(50 * time.Millisecond):
313
+ continue // being explicit
314
+ case <-stop:
315
+ return
316
+ }
317
+ }
318
+ }()
319
+
320
+ waitForWellFormedTables(t, dhts, 7, 10, 5*time.Second)
321
+ close(stop)
322
323
if u.Debug {
324
// the routing tables should be full now. let's inspect them.
267
- <-time.After(5 * time.Second)
268
- t.Logf("checking routing table of %d", nDHTs)
269
- for _, dht := range dhts {
270
- fmt.Printf("checking routing table of %s\n", dht.self)
271
- dht.routingTable.Print()
272
- fmt.Println("")
325
+ printRoutingTables(dhts)
326
+ }
327
+}
328
+
329
+func TestPeriodicBootstrap(t *testing.T) {
330
+ // t.Skip("skipping test to debug another")
331
+ if testing.Short() {
332
+ t.SkipNow()
333
+ }
334
+
335
+ ctx := context.Background()
336
+
337
+ nDHTs := 30
338
+ _, _, dhts := setupDHTS(ctx, nDHTs, t)
339
+ defer func() {
340
+ for i := 0; i < nDHTs; i++ {
341
+ dhts[i].Close()
342
+ defer dhts[i].host.Close()
343
+ }
344
+ }()
345
+
346
+ // signal amplifier
347
+ amplify := func(signal chan time.Time, other []chan time.Time) {
348
+ for t := range signal {
349
+ for _, s := range other {
350
+ s <- t
351
+ }
352
+ }
353
+ for _, s := range other {
354
+ close(s)
355
}
356
}
357
276
- // test "well-formed-ness" (>= 3 peers in every routing table)
277
- avgsize := 0
358
+ signal := make(chan time.Time)
359
+ allSignals := []chan time.Time{}
360
+
361
+ // kick off periodic bootstrappers with instrumented signals.
362
+ for _, dht := range dhts {
363
+ s := make(chan time.Time)
364
+ allSignals = append(allSignals, s)
365
+ dht.BootstrapOnSignal(5, s)
366
+ }
367
+ go amplify(signal, allSignals)
368
+
369
+ t.Logf("dhts are not connected.", nDHTs)
370
+ for _, dht := range dhts {
371
+ rtlen := dht.routingTable.Size()
372
+ if rtlen > 0 {
373
+ t.Errorf("routing table for %s should have 0 peers. has %d", dht.self, rtlen)
374
+ }
375
+ }
376
+
377
+ for i := 0; i < nDHTs; i++ {
378
+ connect(t, ctx, dhts[i], dhts[(i+1)%len(dhts)])
379
+ }
380
+
381
+ t.Logf("dhts are now connected to 1-2 others.", nDHTs)
382
for _, dht := range dhts {
383
rtlen := dht.routingTable.Size()
280
- avgsize += rtlen
281
- t.Logf("routing table for %s has %d peers", dht.self, rtlen)
282
- if rtlen < 4 {
283
- // currently, we dont have good bootstrapping guarantees.
284
- // t.Errorf("routing table for %s only has %d peers", dht.self, rtlen)
384
+ if rtlen > 2 {
385
+ t.Errorf("routing table for %s should have at most 2 peers. has %d", dht.self, rtlen)
386
}
387
}
287
- avgsize = avgsize / len(dhts)
288
- avgsizeExpected := 6
388
290
- t.Logf("avg rt size: %d", avgsize)
291
- if avgsize < avgsizeExpected {
292
- t.Errorf("avg rt size: %d < %d", avgsize, avgsizeExpected)
389
+ if u.Debug {
390
+ printRoutingTables(dhts)
391
+ }
392
+
393
+ t.Logf("bootstrapping them so they find each other", nDHTs)
394
+ signal <- time.Now()
395
+
396
+ // this is async, and we dont know when it's finished with one cycle, so keep checking
397
+ // until the routing tables look better, or some long timeout for the failure case.
398
+ waitForWellFormedTables(t, dhts, 7, 10, 5*time.Second)
399
+
400
+ if u.Debug {
401
+ printRoutingTables(dhts)
402
}
403
}
404
@@ -319,7 +428,6 @@ func TestProvidesMany(t *testing.T) {
428
429
if u.Debug {
430
// the routing tables should be full now. let's inspect them.
322
- <-time.After(5 * time.Second)
431
t.Logf("checking routing table of %d", nDHTs)
432
for _, dht := range dhts {
433
fmt.Printf("checking routing table of %s\n", dht.self)