query: fix race condition: redux
this time just move to goprocess
Juan Batiz-Benet committed
Mar 7, 2015 at 03:20 UTC
c2bbca05963e4c63a152eefe7ea648ded0208a0b
2 files changed
+42
-38
routing/dht/ext_test.go
+1
-1
@@ -52,7 +52,7 @@ func TestGetFailures(t *testing.T) {
52
err = merr[0]
53
}
54
55
- if err != context.DeadlineExceeded {
55
+ if err != context.DeadlineExceeded && err != context.Canceled {
56
t.Fatal("Got different error than we expected", err)
57
}
58
} else {
routing/dht/query.go
+41
-37
@@ -12,7 +12,8 @@ import (
12
pset "github.com/jbenet/go-ipfs/util/peerset"
13
todoctr "github.com/jbenet/go-ipfs/util/todocounter"
14
15
- ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
15
+ process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
16
+ ctxproc "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
17
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
18
)
19
@@ -61,8 +62,8 @@ func (q *dhtQuery) Run(ctx context.Context, peers []peer.ID) (*dhtQueryResult, e
62
ctx, cancel := context.WithCancel(ctx)
63
defer cancel()
64
64
- runner := newQueryRunner(ctx, q)
65
- return runner.Run(peers)
65
+ runner := newQueryRunner(q)
66
+ return runner.Run(ctx, peers)
67
}
68
69
type dhtQueryRunner struct {
@@ -77,22 +78,24 @@ type dhtQueryRunner struct {
78
rateLimit chan struct{} // processing semaphore
79
log eventlog.EventLogger
80
80
- cg ctxgroup.ContextGroup
81
+ proc process.Process
82
sync.RWMutex
83
}
84
84
-func newQueryRunner(ctx context.Context, q *dhtQuery) *dhtQueryRunner {
85
+func newQueryRunner(q *dhtQuery) *dhtQueryRunner {
86
+ proc := process.WithParent(process.Background())
87
+ ctx := ctxproc.WithProcessClosing(context.Background(), proc)
88
return &dhtQueryRunner{
89
query: q,
90
peersToQuery: queue.NewChanQueue(ctx, queue.NewXORDistancePQ(q.key)),
91
peersRemaining: todoctr.NewSyncCounter(),
92
peersSeen: pset.New(),
93
rateLimit: make(chan struct{}, q.concurrency),
91
- cg: ctxgroup.WithContext(ctx),
94
+ proc: proc,
95
}
96
}
97
95
-func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
98
+func (r *dhtQueryRunner) Run(ctx context.Context, peers []peer.ID) (*dhtQueryResult, error) {
99
r.log = log
100
101
if len(peers) == 0 {
@@ -107,31 +110,30 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
110
111
// add all the peers we got first.
112
for _, p := range peers {
110
- r.addPeerToQuery(r.cg.Context(), p)
111
- }
112
-
113
- // may be closed already. this caused an odd race (where we attempt to
114
- // add a child to an already closed ctxgroup). this is a temp workaround
115
- // as we'll switch to using a proc here soon.
116
- select {
117
- case <-r.cg.Closed():
118
- return nil, r.cg.Context().Err()
119
- default:
113
+ r.addPeerToQuery(p)
114
}
115
116
// go do this thing.
123
- // do it as a child func to make sure Run exits
117
+ // do it as a child proc to make sure Run exits
118
// ONLY AFTER spawn workers has exited.
125
- r.cg.AddChildFunc(r.spawnWorkers)
119
+ r.proc.Go(r.spawnWorkers)
120
121
// so workers are working.
122
123
// wait until they're done.
124
err := routing.ErrNotFound
125
126
+ // now, if the context finishes, close the proc.
127
+ // we have to do it here because the logic before is setup, which
128
+ // should run without closing the proc.
129
+ go func() {
130
+ <-ctx.Done()
131
+ r.proc.Close()
132
+ }()
133
+
134
select {
135
case <-r.peersRemaining.Done():
134
- r.cg.Close()
136
+ r.proc.Close()
137
r.RLock()
138
defer r.RUnlock()
139
@@ -143,12 +145,10 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
145
err = r.errs[0]
146
}
147
146
- case <-r.cg.Closed():
147
- log.Debug("r.cg.Closed()")
148
-
148
+ case <-r.proc.Closed():
149
r.RLock()
150
defer r.RUnlock()
151
- err = r.cg.Context().Err() // collect the error.
151
+ err = context.DeadlineExceeded
152
}
153
154
if r.result != nil && r.result.success {
@@ -158,7 +158,7 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
158
return nil, err
159
}
160
161
-func (r *dhtQueryRunner) addPeerToQuery(ctx context.Context, next peer.ID) {
161
+func (r *dhtQueryRunner) addPeerToQuery(next peer.ID) {
162
// if new peer is ourselves...
163
if next == r.query.dht.self {
164
r.log.Debug("addPeerToQuery skip self")
@@ -172,18 +172,18 @@ func (r *dhtQueryRunner) addPeerToQuery(ctx context.Context, next peer.ID) {
172
r.peersRemaining.Increment(1)
173
select {
174
case r.peersToQuery.EnqChan <- next:
175
- case <-ctx.Done():
175
+ case <-r.proc.Closing():
176
}
177
}
178
179
-func (r *dhtQueryRunner) spawnWorkers(parent ctxgroup.ContextGroup) {
179
+func (r *dhtQueryRunner) spawnWorkers(proc process.Process) {
180
for {
181
182
select {
183
case <-r.peersRemaining.Done():
184
return
185
186
- case <-r.cg.Closing():
186
+ case <-r.proc.Closing():
187
return
188
189
case p, more := <-r.peersToQuery.DeqChan:
@@ -193,24 +193,27 @@ func (r *dhtQueryRunner) spawnWorkers(parent ctxgroup.ContextGroup) {
193
194
// do it as a child func to make sure Run exits
195
// ONLY AFTER spawn workers has exited.
196
- parent.AddChildFunc(func(cg ctxgroup.ContextGroup) {
197
- r.queryPeer(cg, p)
196
+ proc.Go(func(proc process.Process) {
197
+ r.queryPeer(proc, p)
198
})
199
}
200
}
201
}
202
203
-func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
203
+func (r *dhtQueryRunner) queryPeer(proc process.Process, p peer.ID) {
204
// make sure we rate limit concurrency.
205
select {
206
case <-r.rateLimit:
207
- case <-cg.Closing():
207
+ case <-proc.Closing():
208
r.peersRemaining.Decrement(1)
209
return
210
}
211
212
// ok let's do this!
213
214
+ // create a context from our proc.
215
+ ctx := ctxproc.WithProcessClosing(context.Background(), proc)
216
+
217
// make sure we do this when we exit
218
defer func() {
219
// signal we're done proccessing peer p
@@ -227,10 +230,11 @@ func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
230
r.rateLimit <- struct{}{}
231
232
pi := peer.PeerInfo{ID: p}
230
- if err := r.query.dht.host.Connect(cg.Context(), pi); err != nil {
233
+
234
+ if err := r.query.dht.host.Connect(ctx, pi); err != nil {
235
log.Debugf("Error connecting: %s", err)
236
233
- notif.PublishQueryEvent(cg.Context(), ¬if.QueryEvent{
237
+ notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
238
Type: notif.QueryError,
239
Extra: err.Error(),
240
})
@@ -246,7 +250,7 @@ func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
250
}
251
252
// finally, run the query against this peer
249
- res, err := r.query.qfunc(cg.Context(), p)
253
+ res, err := r.query.qfunc(ctx, p)
254
255
if err != nil {
256
log.Debugf("ERROR worker for: %v %v", p, err)
@@ -259,7 +263,7 @@ func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
263
r.Lock()
264
r.result = res
265
r.Unlock()
262
- go r.cg.Close() // signal to everyone that we're done.
266
+ go r.proc.Close() // signal to everyone that we're done.
267
// must be async, as we're one of the children, and Close blocks.
268
269
} else if len(res.closerPeers) > 0 {
@@ -272,7 +276,7 @@ func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
276
277
// add their addresses to the dialer's peerstore
278
r.query.dht.peerstore.AddAddrs(next.ID, next.Addrs, peer.TempAddrTTL)
275
- r.addPeerToQuery(cg.Context(), next.ID)
279
+ r.addPeerToQuery(next.ID)
280
log.Debugf("PEERS CLOSER -- worker for: %v added %v (%v)", p, next.ID, next.Addrs)
281
}
282
} else {