core/bootstrap: cleaned up bootstrapping
Moved it to its own package to isolate scope.
Juan Batiz-Benet committed
Jan 20, 2015 at 07:38 UTC
d6ce837d720ffc9f542f4c63f83364372f219f27
3 files changed
+179
-112
core/bootstrap.go
+148
-55
@@ -2,6 +2,7 @@ package core
2
3
import (
4
"errors"
5
+ "fmt"
6
"math/rand"
7
"sync"
8
"time"
@@ -16,109 +17,187 @@ import (
17
18
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
19
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
20
+ goprocess "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
21
+ periodicproc "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/periodic"
22
)
23
24
+// ErrNotEnoughBootstrapPeers signals that we do not have enough bootstrap
25
+// peers to bootstrap correctly.
26
+var ErrNotEnoughBootstrapPeers = errors.New("not enough bootstrap peers to bootstrap")
27
+
28
const (
22
- period = 30 * time.Second // how often to check connection status
23
- connectiontimeout time.Duration = period / 3 // duration to wait when attempting to connect
24
- recoveryThreshold = 4 // attempt to bootstrap if connection count falls below this value
25
- numDHTBootstrapQueries = 15 // number of DHT queries to execute
29
+ // BootstrapPeriod governs the periodic interval at which the node will
30
+ // attempt to bootstrap. The bootstrap process is not very expensive, so
31
+ // this threshold can afford to be small (<=30s).
32
+ BootstrapPeriod = 30 * time.Second
33
+
34
+ // BootstrapPeerThreshold governs the node Bootstrap process. If the node
35
+ // has less open connections than this number, it will open connections
36
+ // to the bootstrap nodes. From there, the routing system should be able
37
+ // to use the connections to the bootstrap nodes to connect to even more
38
+ // peers. Routing systems like the IpfsDHT do so in their own Bootstrap
39
+ // process, which issues random queries to find more peers.
40
+ BootstrapPeerThreshold = 4
41
+
42
+ // BootstrapConnectionTimeout determines how long to wait for a bootstrap
43
+ // connection attempt before cancelling it.
44
+ BootstrapConnectionTimeout time.Duration = BootstrapPeriod / 3
45
)
46
28
-func superviseConnections(parent context.Context,
29
- h host.Host,
30
- route *dht.IpfsDHT, // TODO depend on abstract interface for testing purposes
31
- store peer.Peerstore,
32
- peers []peer.PeerInfo) error {
47
+// nodeBootstrapper is a small object used to bootstrap an IpfsNode.
48
+type nodeBootstrapper struct {
49
+ node *IpfsNode
50
+}
51
34
- var dhtAlreadyBootstrapping bool
52
+// TryToBootstrap starts IpfsNode bootstrapping. This function will run an
53
+// initial bootstrapping phase before exiting: connect to several bootstrap
54
+// nodes. This allows callers to call this function synchronously to:
55
+// - check if an error occurrs (bootstrapping unsuccessful)
56
+// - wait before starting services which require the node to be bootstrapped
57
+//
58
+// If bootstrapping initially fails, Bootstrap() will try again for a total of
59
+// three times, before giving up completely. Note that in environments where a
60
+// node may be initialized offline, as normal operation, BootstrapForever()
61
+// should be used instead.
62
+//
63
+// Note: this function could be much cleaner if we were to relax the constraint
64
+// that we want to exit **after** we have performed initial bootstrapping (and are
65
+// thus connected to nodes). The constraint may not be that useful in practice.
66
+// Consider cases when we initialize the node while disconnected from the internet.
67
+// We don't want this launch to fail... want to continue launching the node, hoping
68
+// that bootstrapping will work in the future if we get connected.
69
+func (nb *nodeBootstrapper) TryToBootstrap(ctx context.Context, peers []peer.PeerInfo) error {
70
+ n := nb.node
71
+
72
+ // TODO what bootstrapping should happen if there is no DHT? i.e. we could
73
+ // continue connecting to our bootstrap peers, but for what purpose? for now
74
+ // simply exit without connecting to any of them. When we introduce another
75
+ // routing system that uses bootstrap peers we can change this.
76
+ dht, ok := n.Routing.(*dht.IpfsDHT)
77
+ if !ok {
78
+ return nil
79
+ }
80
36
- for {
37
- ctx, _ := context.WithTimeout(parent, connectiontimeout)
38
- // TODO get config from disk so |peers| always reflects the latest
39
- // information
40
- if err := bootstrap(ctx, h, route, store, peers); err != nil {
41
- log.Error(err)
81
+ for i := 0; i < 3; i++ {
82
+ if err := bootstrapRound(ctx, n.PeerHost, dht, n.Peerstore, peers); err != nil {
83
+ return err
84
}
85
+ }
86
44
- if !dhtAlreadyBootstrapping {
45
- dhtAlreadyBootstrapping = true // only call dht.Bootstrap once.
46
- if _, err := route.Bootstrap(); err != nil {
47
- log.Error(err)
48
- }
87
+ // at this point we have done at least one round of initial bootstrap.
88
+ // we're ready to kick off dht bootstrapping.
89
+ dbproc, err := dht.Bootstrap(ctx)
90
+ if err != nil {
91
+ return err
92
+ }
93
+
94
+ // kick off the node's periodic bootstrapping
95
+ proc := periodicproc.Tick(BootstrapPeriod, func(worker goprocess.Process) {
96
+ if err := bootstrapRound(ctx, n.PeerHost, dht, n.Peerstore, peers); err != nil {
97
+ log.Error(err)
98
}
99
+ })
100
+
101
+ // add dht bootstrap proc as a child, so it is closed automatically when we are.
102
+ proc.AddChild(dbproc)
103
+
104
+ // we were given a context. instead of returning proc for the caller
105
+ // to manage, for now we just close the proc when context is done.
106
+ go func() {
107
+ <-ctx.Done()
108
+ proc.Close()
109
+ }()
110
+ return nil
111
+}
112
51
- select {
52
- case <-parent.Done():
53
- return parent.Err()
54
- case <-time.Tick(period):
113
+// BootstrapForever starts IpfsNode bootstrapping. Unlike TryToBootstrap(),
114
+// BootstrapForever() will run indefinitely (until its context is cancelled).
115
+// This is particularly useful for the daemon and other services, which may
116
+// be started offline and will come online at a future date.
117
+//
118
+// TODO: check offline --to--> online case works well and doesn't hurt perf.
119
+// We may still be dialing. We should check network config.
120
+func (nb *nodeBootstrapper) BootstrapForever(ctx context.Context, peers []peer.PeerInfo) error {
121
+ for {
122
+ if err := nb.TryToBootstrap(ctx, peers); err == nil {
123
+ return nil
124
}
125
}
57
- return nil
126
}
127
60
-func bootstrap(ctx context.Context,
61
- h host.Host,
62
- r *dht.IpfsDHT,
63
- ps peer.Peerstore,
128
+func bootstrapRound(ctx context.Context,
129
+ host host.Host,
130
+ route *dht.IpfsDHT,
131
+ peerstore peer.Peerstore,
132
bootstrapPeers []peer.PeerInfo) error {
133
66
- connectedPeers := h.Network().Peers()
67
- if len(connectedPeers) >= recoveryThreshold {
68
- log.Event(ctx, "bootstrapSkip", h.ID())
69
- log.Debugf("%s core bootstrap skipped -- connected to %d (> %d) nodes",
70
- h.ID(), len(connectedPeers), recoveryThreshold)
134
+ ctx, _ = context.WithTimeout(ctx, BootstrapConnectionTimeout)
135
136
+ // determine how many bootstrap connections to open
137
+ connectedPeers := host.Network().Peers()
138
+ if len(connectedPeers) >= BootstrapPeerThreshold {
139
+ log.Event(ctx, "bootstrapSkip", host.ID())
140
+ log.Debugf("%s core bootstrap skipped -- connected to %d (> %d) nodes",
141
+ host.ID(), len(connectedPeers), BootstrapPeerThreshold)
142
return nil
143
}
74
- numCxnsToCreate := recoveryThreshold - len(connectedPeers)
75
-
76
- log.Event(ctx, "bootstrapStart", h.ID())
77
- log.Debugf("%s core bootstrapping to %d more nodes", h.ID(), numCxnsToCreate)
144
+ numCxnsToCreate := BootstrapPeerThreshold - len(connectedPeers)
145
146
+ // filter out bootstrap nodes we are already connected to
147
var notConnected []peer.PeerInfo
148
for _, p := range bootstrapPeers {
81
- if h.Network().Connectedness(p.ID) != inet.Connected {
149
+ if host.Network().Connectedness(p.ID) != inet.Connected {
150
notConnected = append(notConnected, p)
151
}
152
}
153
86
- // if not connected to all bootstrap peer candidates
87
- if len(notConnected) > 0 {
88
- var randomSubset = randomSubsetOfPeers(notConnected, numCxnsToCreate)
89
- log.Debugf("%s bootstrapping to %d nodes: %s", h.ID(), numCxnsToCreate, randomSubset)
90
- if err := connect(ctx, ps, r, randomSubset); err != nil {
91
- log.Event(ctx, "bootstrapError", h.ID(), lgbl.Error(err))
92
- log.Errorf("%s bootstrap error: %s", h.ID(), err)
93
- return err
94
- }
154
+ // if connected to all bootstrap peer candidates, exit
155
+ if len(notConnected) < 1 {
156
+ log.Debugf("%s no more bootstrap peers to create %d connections", host.ID(), numCxnsToCreate)
157
+ return ErrNotEnoughBootstrapPeers
158
+ }
159
+
160
+ // connect to a random susbset of bootstrap candidates
161
+ var randomSubset = randomSubsetOfPeers(notConnected, numCxnsToCreate)
162
+ log.Event(ctx, "bootstrapStart", host.ID())
163
+ log.Debugf("%s bootstrapping to %d nodes: %s", host.ID(), numCxnsToCreate, randomSubset)
164
+ if err := bootstrapConnect(ctx, peerstore, route, randomSubset); err != nil {
165
+ log.Event(ctx, "bootstrapError", host.ID(), lgbl.Error(err))
166
+ log.Errorf("%s bootstrap error: %s", host.ID(), err)
167
+ return err
168
}
169
return nil
170
}
171
99
-func connect(ctx context.Context, ps peer.Peerstore, r *dht.IpfsDHT, peers []peer.PeerInfo) error {
172
+func bootstrapConnect(ctx context.Context,
173
+ ps peer.Peerstore,
174
+ route *dht.IpfsDHT,
175
+ peers []peer.PeerInfo) error {
176
if len(peers) < 1 {
101
- return errors.New("bootstrap set empty")
177
+ return ErrNotEnoughBootstrapPeers
178
}
179
180
+ errs := make(chan error, len(peers))
181
var wg sync.WaitGroup
182
for _, p := range peers {
183
184
// performed asynchronously because when performed synchronously, if
185
// one `Connect` call hangs, subsequent calls are more likely to
186
// fail/abort due to an expiring context.
187
+ // Also, performed asynchronously for dial speed.
188
189
wg.Add(1)
190
go func(p peer.PeerInfo) {
191
defer wg.Done()
114
- log.Event(ctx, "bootstrapDial", r.LocalPeer(), p.ID)
115
- log.Debugf("%s bootstrapping to %s", r.LocalPeer(), p.ID)
192
+ log.Event(ctx, "bootstrapDial", route.LocalPeer(), p.ID)
193
+ log.Debugf("%s bootstrapping to %s", route.LocalPeer(), p.ID)
194
195
ps.AddAddresses(p.ID, p.Addrs)
118
- err := r.Connect(ctx, p.ID)
196
+ err := route.Connect(ctx, p.ID)
197
if err != nil {
198
log.Event(ctx, "bootstrapFailed", p.ID)
121
- log.Criticalf("failed to bootstrap with %v: %s", p.ID, err)
199
+ log.Errorf("failed to bootstrap with %v: %s", p.ID, err)
200
+ errs <- err
201
return
202
}
203
log.Event(ctx, "bootstrapSuccess", p.ID)
@@ -126,6 +205,20 @@ func connect(ctx context.Context, ps peer.Peerstore, r *dht.IpfsDHT, peers []pee
205
}(p)
206
}
207
wg.Wait()
208
+
209
+ // our failure condition is when no connection attempt succeeded.
210
+ // So drain the errs channel, counting the results.
211
+ close(errs)
212
+ count := 0
213
+ var err error
214
+ for err = range errs {
215
+ if err != nil {
216
+ count++
217
+ }
218
+ }
219
+ if count == len(peers) {
220
+ return fmt.Errorf("failed to bootstrap. %s", err)
221
+ }
222
return nil
223
}
224
core/core.go
+2
-20
@@ -297,30 +297,12 @@ func (n *IpfsNode) Resolve(path string) (*merkledag.Node, error) {
297
func (n *IpfsNode) Bootstrap(ctx context.Context, peers []peer.PeerInfo) error {
298
299
// TODO what should return value be when in offlineMode?
300
-
300
if n.Routing == nil {
301
return nil
302
}
303
305
- // TODO what bootstrapping should happen if there is no DHT? i.e. we could
306
- // continue connecting to our bootstrap peers, but for what purpose?
307
- dhtRouting, ok := n.Routing.(*dht.IpfsDHT)
308
- if !ok {
309
- return nil
310
- }
311
-
312
- // TODO consider moving connection supervision into the Network. We've
313
- // discussed improvements to this Node constructor. One improvement
314
- // would be to make the node configurable, allowing clients to inject
315
- // an Exchange, Network, or Routing component and have the constructor
316
- // manage the wiring. In that scenario, this dangling function is a bit
317
- // awkward.
318
-
319
- // spin off the node's connection supervisor.
320
- // TODO, clean up how this thing works. Make the superviseConnections thing
321
- // work like the DHT.Bootstrap.
322
- go superviseConnections(ctx, n.PeerHost, dhtRouting, n.Peerstore, peers)
323
- return nil
304
+ nb := nodeBootstrapper{n}
305
+ return nb.TryToBootstrap(ctx, peers)
306
}
307
308
func (n *IpfsNode) loadID() error {
routing/dht/dht_bootstrap.go
+29
-37
@@ -14,6 +14,7 @@ import (
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
+ periodicproc "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/periodic"
18
)
19
20
// DefaultBootstrapQueries specifies how many queries to run,
@@ -54,9 +55,9 @@ const DefaultBootstrapTimeout = time.Duration(10 * time.Second)
55
// and connected to at least a few nodes.
56
//
57
// Like PeriodicBootstrap, Bootstrap returns a process, so the user can stop it.
57
-func (dht *IpfsDHT) Bootstrap() (goprocess.Process, error) {
58
+func (dht *IpfsDHT) Bootstrap(ctx context.Context) (goprocess.Process, error) {
59
59
- if err := dht.runBootstrap(dht.Context(), DefaultBootstrapQueries); err != nil {
60
+ if err := dht.runBootstrap(ctx, DefaultBootstrapQueries); err != nil {
61
return nil, err
62
}
63
@@ -79,41 +80,32 @@ func (dht *IpfsDHT) BootstrapOnSignal(queries int, signal <-chan time.Time) (gop
80
return nil, fmt.Errorf("invalid signal: %v", signal)
81
}
82
82
- proc := goprocess.Go(func(worker goprocess.Process) {
83
- defer log.Debug("dht bootstrapper shutting down")
84
- for {
85
- select {
86
- case <-worker.Closing():
87
- return
88
-
89
- case <-signal:
90
- // it would be useful to be able to send out signals of when we bootstrap, too...
91
- // maybe this is a good case for whole module event pub/sub?
92
-
93
- ctx := dht.Context()
94
- if err := dht.runBootstrap(ctx, queries); err != nil {
95
- log.Error(err)
96
- // A bootstrapping error is important to notice but not fatal.
97
- // maybe the client should be able to consume these errors,
98
- // though I dont have a clear use case in mind-- what **could**
99
- // the client do if one of the bootstrap calls fails?
100
- //
101
- // This is also related to the core's bootstrap failures.
102
- // superviseConnections should perhaps allow clients to detect
103
- // bootstrapping problems.
104
- //
105
- // Anyway, passing errors could be done with a bootstrapper object.
106
- // this would imply the client should be able to consume a lot of
107
- // other non-fatal dht errors too. providing this functionality
108
- // should be done correctly DHT-wide.
109
- // NB: whatever the design, clients must ensure they drain errors!
110
- // This pattern is common to many things, perhaps long-running services
111
- // should have something like an ErrStream that allows clients to consume
112
- // periodic errors and take action. It should allow the user to also
113
- // ignore all errors with something like an ErrStreamDiscard. We should
114
- // study what other systems do for ideas.
115
- }
116
- }
83
+ proc := periodicproc.Ticker(signal, func(worker goprocess.Process) {
84
+ // it would be useful to be able to send out signals of when we bootstrap, too...
85
+ // maybe this is a good case for whole module event pub/sub?
86
+
87
+ ctx := dht.Context()
88
+ if err := dht.runBootstrap(ctx, queries); err != nil {
89
+ log.Error(err)
90
+ // A bootstrapping error is important to notice but not fatal.
91
+ // maybe the client should be able to consume these errors,
92
+ // though I dont have a clear use case in mind-- what **could**
93
+ // the client do if one of the bootstrap calls fails?
94
+ //
95
+ // This is also related to the core's bootstrap failures.
96
+ // superviseConnections should perhaps allow clients to detect
97
+ // bootstrapping problems.
98
+ //
99
+ // Anyway, passing errors could be done with a bootstrapper object.
100
+ // this would imply the client should be able to consume a lot of
101
+ // other non-fatal dht errors too. providing this functionality
102
+ // should be done correctly DHT-wide.
103
+ // NB: whatever the design, clients must ensure they drain errors!
104
+ // This pattern is common to many things, perhaps long-running services
105
+ // should have something like an ErrStream that allows clients to consume
106
+ // periodic errors and take action. It should allow the user to also
107
+ // ignore all errors with something like an ErrStreamDiscard. We should
108
+ // study what other systems do for ideas.
109
}
110
})
111