bitswap: serialize connect/disconnect notifications over one channel.
Otherwise, we could end up receiving a disconnect notification before a connect notification (and think we have a connection that we don't have). License: MIT Signed-off-by: Steven Allen <steven@stebalien.com>
Steven Allen committed
Jul 27, 2017 at 14:06 UTC
4ef73ee25d7e149e43464ae1f7c7237bd1e9c313
1 file changed
+19
-14
exchange/bitswap/wantmanager.go
+19
-14
@@ -17,10 +17,9 @@ import (
17
18
type WantManager struct {
19
// sync channels for Run loop
20
- incoming chan *wantSet
21
- connect chan peer.ID // notification channel for new peers connecting
22
- disconnect chan peer.ID // notification channel for peers disconnecting
23
- peerReqs chan chan []peer.ID // channel to request connected peers on
20
+ incoming chan *wantSet
21
+ connectEvent chan peerStatus // notification channel for peers connecting/disconnecting
22
+ peerReqs chan chan []peer.ID // channel to request connected peers on
23
24
// synchronized by Run loop, only touch inside there
25
peers map[peer.ID]*msgQueue
@@ -35,6 +34,11 @@ type WantManager struct {
34
sentHistogram metrics.Histogram
35
}
36
37
+type peerStatus struct {
38
+ connect bool
39
+ peer peer.ID
40
+}
41
+
42
func NewWantManager(ctx context.Context, network bsnet.BitSwapNetwork) *WantManager {
43
ctx, cancel := context.WithCancel(ctx)
44
wantlistGauge := metrics.NewCtx(ctx, "wantlist_total",
@@ -43,8 +47,7 @@ func NewWantManager(ctx context.Context, network bsnet.BitSwapNetwork) *WantMana
47
" this bitswap").Histogram(metricsBuckets)
48
return &WantManager{
49
incoming: make(chan *wantSet, 10),
46
- connect: make(chan peer.ID, 10),
47
- disconnect: make(chan peer.ID, 10),
50
+ connectEvent: make(chan peerStatus, 10),
51
peerReqs: make(chan chan []peer.ID),
52
peers: make(map[peer.ID]*msgQueue),
53
wl: wantlist.NewThreadSafe(),
@@ -270,22 +273,22 @@ func (mq *msgQueue) openSender(ctx context.Context) error {
273
274
func (pm *WantManager) Connected(p peer.ID) {
275
select {
273
- case pm.connect <- p:
276
+ case pm.connectEvent <- peerStatus{peer: p, connect: true}:
277
case <-pm.ctx.Done():
278
}
279
}
280
281
func (pm *WantManager) Disconnected(p peer.ID) {
282
select {
280
- case pm.disconnect <- p:
283
+ case pm.connectEvent <- peerStatus{peer: p, connect: false}:
284
case <-pm.ctx.Done():
285
}
286
}
287
288
// TODO: use goprocess here once i trust it
289
func (pm *WantManager) Run() {
287
- tock := time.NewTicker(rebroadcastDelay.Get())
288
- defer tock.Stop()
290
+ // NOTE: Do not open any streams or connections from anywhere in this
291
+ // event loop. Really, just don't do anything likely to block.
292
for {
293
select {
294
case ws := <-pm.incoming:
@@ -329,10 +332,12 @@ func (pm *WantManager) Run() {
332
}
333
}
334
332
- case p := <-pm.connect:
333
- pm.startPeerHandler(p)
334
- case p := <-pm.disconnect:
335
- pm.stopPeerHandler(p)
335
+ case p := <-pm.connectEvent:
336
+ if p.connect {
337
+ pm.startPeerHandler(p.peer)
338
+ } else {
339
+ pm.stopPeerHandler(p.peer)
340
+ }
341
case req := <-pm.peerReqs:
342
var peers []peer.ID
343
for p := range pm.peers {