dht/query: fix important panic
Withe queries (particularly providers), it was possible to exit the query runner's Run BEFORE all its children were done, because the runner itself only listened to the context. This introduced the possibility of a panic (you can go check it out by running the TestProvidersMany test on dht_test in commits before this one). Thankfully, ctxgroup saved the day with almost _zero_ changes to the sync flow, and now we have the guarantee that the query runner will only exit if all its children are done. :heart: Conflicts: routing/dht/query.go
Juan Batiz-Benet committed
Dec 24, 2014 at 02:13 UTC
61cdef3570b87fa78596405966159dc96db527c4
2 files changed
+45
-29
routing/dht/dht_net.go
+11
-5
@@ -7,6 +7,7 @@ import (
7
inet "github.com/jbenet/go-ipfs/net"
8
peer "github.com/jbenet/go-ipfs/peer"
9
pb "github.com/jbenet/go-ipfs/routing/dht/pb"
10
+ ctxutil "github.com/jbenet/go-ipfs/util/ctx"
11
12
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
13
ggio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/io"
@@ -21,8 +22,10 @@ func (dht *IpfsDHT) handleNewMessage(s inet.Stream) {
22
defer s.Close()
23
24
ctx := dht.Context()
24
- r := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
25
- w := ggio.NewDelimitedWriter(s)
25
+ cr := ctxutil.NewReader(ctx, s) // ok to use. we defer close stream in this func
26
+ cw := ctxutil.NewWriter(ctx, s) // ok to use. we defer close stream in this func
27
+ r := ggio.NewDelimitedReader(cr, inet.MessageSizeMax)
28
+ w := ggio.NewDelimitedWriter(cw)
29
mPeer := s.Conn().RemotePeer()
30
31
// receive msg
@@ -76,8 +79,10 @@ func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.ID, pmes *pb.Message
79
}
80
defer s.Close()
81
79
- r := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
80
- w := ggio.NewDelimitedWriter(s)
82
+ cr := ctxutil.NewReader(ctx, s) // ok to use. we defer close stream in this func
83
+ cw := ctxutil.NewWriter(ctx, s) // ok to use. we defer close stream in this func
84
+ r := ggio.NewDelimitedReader(cr, inet.MessageSizeMax)
85
+ w := ggio.NewDelimitedWriter(cw)
86
87
start := time.Now()
88
@@ -113,7 +118,8 @@ func (dht *IpfsDHT) sendMessage(ctx context.Context, p peer.ID, pmes *pb.Message
118
}
119
defer s.Close()
120
116
- w := ggio.NewDelimitedWriter(s)
121
+ cw := ctxutil.NewWriter(ctx, s) // ok to use. we defer close stream in this func
122
+ w := ggio.NewDelimitedWriter(cw)
123
124
log.Debugf("%s writing", dht.self)
125
if err := w.WriteMsg(pmes); err != nil {
routing/dht/query.go
+34
-24
@@ -12,6 +12,7 @@ import (
12
todoctr "github.com/jbenet/go-ipfs/util/todocounter"
13
14
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
15
+ ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
16
)
17
18
var maxQueryConcurrency = AlphaValue
@@ -78,9 +79,8 @@ type dhtQueryRunner struct {
79
// peersRemaining is a counter of peers remaining (toQuery + processing)
80
peersRemaining todoctr.Counter
81
81
- // context
82
- ctx context.Context
83
- cancel context.CancelFunc
82
+ // context group
83
+ cg ctxgroup.ContextGroup
84
85
// result
86
result *dhtQueryResult
@@ -93,16 +93,13 @@ type dhtQueryRunner struct {
93
}
94
95
func newQueryRunner(ctx context.Context, q *dhtQuery) *dhtQueryRunner {
96
- ctx, cancel := context.WithCancel(ctx)
97
-
96
return &dhtQueryRunner{
99
- ctx: ctx,
100
- cancel: cancel,
97
query: q,
98
peersToQuery: queue.NewChanQueue(ctx, queue.NewXORDistancePQ(q.key)),
99
peersRemaining: todoctr.NewSyncCounter(),
100
peersSeen: peer.Set{},
101
rateLimit: make(chan struct{}, q.concurrency),
102
+ cg: ctxgroup.WithContext(ctx),
103
}
104
}
105
@@ -120,11 +117,13 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
117
118
// add all the peers we got first.
119
for _, p := range peers {
123
- r.addPeerToQuery(p, "") // don't have access to self here...
120
+ r.addPeerToQuery(r.cg.Context(), p, "") // don't have access to self here...
121
}
122
123
// go do this thing.
127
- go r.spawnWorkers()
124
+ // do it as a child func to make sure Run exits
125
+ // ONLY AFTER spawn workers has exited.
126
+ r.cg.AddChildFunc(r.spawnWorkers)
127
128
// so workers are working.
129
@@ -133,7 +132,7 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
132
133
select {
134
case <-r.peersRemaining.Done():
136
- r.cancel() // ran all and nothing. cancel all outstanding workers.
135
+ r.cg.Close()
136
r.RLock()
137
defer r.RUnlock()
138
@@ -141,10 +140,10 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
140
err = r.errs[0]
141
}
142
144
- case <-r.ctx.Done():
143
+ case <-r.cg.Closed():
144
r.RLock()
145
defer r.RUnlock()
147
- err = r.ctx.Err()
146
+ err = r.cg.Context().Err() // collect the error.
147
}
148
149
if r.result != nil && r.result.success {
@@ -154,7 +153,7 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
153
return nil, err
154
}
155
157
-func (r *dhtQueryRunner) addPeerToQuery(next peer.ID, benchmark peer.ID) {
156
+func (r *dhtQueryRunner) addPeerToQuery(ctx context.Context, next peer.ID, benchmark peer.ID) {
157
// if new peer is ourselves...
158
if next == r.query.dialer.LocalPeer() {
159
return
@@ -186,37 +185,42 @@ func (r *dhtQueryRunner) addPeerToQuery(next peer.ID, benchmark peer.ID) {
185
r.peersRemaining.Increment(1)
186
select {
187
case r.peersToQuery.EnqChan <- next:
189
- case <-r.ctx.Done():
188
+ case <-ctx.Done():
189
}
190
}
191
193
-func (r *dhtQueryRunner) spawnWorkers() {
192
+func (r *dhtQueryRunner) spawnWorkers(parent ctxgroup.ContextGroup) {
193
for {
194
195
select {
196
case <-r.peersRemaining.Done():
197
return
198
200
- case <-r.ctx.Done():
199
+ case <-r.cg.Closing():
200
return
201
202
case p, more := <-r.peersToQuery.DeqChan:
203
if !more {
204
return // channel closed.
205
}
207
- log.Debugf("spawning worker for: %v\n", p)
208
- go r.queryPeer(p)
206
+ log.Debugf("spawning worker for: %v", p)
207
+
208
+ // do it as a child func to make sure Run exits
209
+ // ONLY AFTER spawn workers has exited.
210
+ parent.AddChildFunc(func(cg ctxgroup.ContextGroup) {
211
+ r.queryPeer(cg, p)
212
+ })
213
}
214
}
215
}
216
213
-func (r *dhtQueryRunner) queryPeer(p peer.ID) {
217
+func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
218
log.Debugf("spawned worker for: %v", p)
219
220
// make sure we rate limit concurrency.
221
select {
222
case <-r.rateLimit:
219
- case <-r.ctx.Done():
223
+ case <-cg.Closing():
224
r.peersRemaining.Decrement(1)
225
return
226
}
@@ -233,7 +237,7 @@ func (r *dhtQueryRunner) queryPeer(p peer.ID) {
237
}()
238
239
// make sure we're connected to the peer.
236
- err := r.query.dialer.DialPeer(r.ctx, p)
240
+ err := r.query.dialer.DialPeer(cg.Context(), p)
241
if err != nil {
242
log.Debugf("ERROR worker for: %v -- err connecting: %v", p, err)
243
r.Lock()
@@ -243,7 +247,7 @@ func (r *dhtQueryRunner) queryPeer(p peer.ID) {
247
}
248
249
// finally, run the query against this peer
246
- res, err := r.query.qfunc(r.ctx, p)
250
+ res, err := r.query.qfunc(cg.Context(), p)
251
252
if err != nil {
253
log.Debugf("ERROR worker for: %v %v", p, err)
@@ -256,14 +260,20 @@ func (r *dhtQueryRunner) queryPeer(p peer.ID) {
260
r.Lock()
261
r.result = res
262
r.Unlock()
259
- r.cancel() // signal to everyone that we're done.
263
+ go r.cg.Close() // signal to everyone that we're done.
264
+ // must be async, as we're one of the children, and Close blocks.
265
266
} else if len(res.closerPeers) > 0 {
267
log.Debugf("PEERS CLOSER -- worker for: %v (%d closer peers)", p, len(res.closerPeers))
268
for _, next := range res.closerPeers {
269
// add their addresses to the dialer's peerstore
270
+ conns := r.query.dialer.ConnsToPeer(next.ID)
271
+ if len(conns) == 0 {
272
+ log.Infof("PEERS CLOSER -- worker for %v FOUND NEW PEER: %s %s", p, next.ID, next.Addrs)
273
+ }
274
+
275
r.query.dialer.Peerstore().AddAddresses(next.ID, next.Addrs)
266
- r.addPeerToQuery(next.ID, p)
276
+ r.addPeerToQuery(cg.Context(), next.ID, p)
277
log.Debugf("PEERS CLOSER -- worker for: %v added %v (%v)", p, next.ID, next.Addrs)
278
}
279
} else {