track wantlists sent to peers individually
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Apr 10, 2017 at 22:05 UTC
47479b672a6040ae8463cf125d276a8a9b4e93d1
1 file changed
+40
-14
exchange/bitswap/wantmanager.go
+40
-14
@@ -17,7 +17,7 @@ import (
17
18
type WantManager struct {
19
// sync channels for Run loop
20
- incoming chan []*bsmsg.Entry
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
@@ -41,7 +41,7 @@ func NewWantManager(ctx context.Context, network bsnet.BitSwapNetwork) *WantMana
41
sentHistogram := metrics.NewCtx(ctx, "sent_all_blocks_bytes", "Histogram of blocks sent by"+
42
" this bitswap").Histogram(metricsBuckets)
43
return &WantManager{
44
- incoming: make(chan []*bsmsg.Entry, 10),
44
+ incoming: make(chan *wantSet, 10),
45
connect: make(chan peer.ID, 10),
46
disconnect: make(chan peer.ID, 10),
47
peerReqs: make(chan chan []peer.ID),
@@ -61,6 +61,7 @@ type msgQueue struct {
61
outlk sync.Mutex
62
out bsmsg.BitSwapMessage
63
network bsnet.BitSwapNetwork
64
+ wl *wantlist.Wantlist
65
66
sender bsnet.MessageSender
67
@@ -76,8 +77,12 @@ func (pm *WantManager) WantBlocks(ctx context.Context, ks []*cid.Cid) {
77
}
78
79
func (pm *WantManager) CancelWants(ks []*cid.Cid) {
79
- log.Infof("cancel wants: %s", ks)
80
- pm.addEntries(context.TODO(), ks, true)
80
+ pm.addEntries(context.Background(), ks, true)
81
+}
82
+
83
+type wantSet struct {
84
+ entries []*bsmsg.Entry
85
+ targets []peer.ID
86
}
87
88
func (pm *WantManager) addEntries(ctx context.Context, ks []*cid.Cid, cancel bool) {
@@ -93,7 +98,7 @@ func (pm *WantManager) addEntries(ctx context.Context, ks []*cid.Cid, cancel boo
98
})
99
}
100
select {
96
- case pm.incoming <- entries:
101
+ case pm.incoming <- &wantSet{entries: entries}:
102
case <-pm.ctx.Done():
103
case <-ctx.Done():
104
}
@@ -133,6 +138,8 @@ func (pm *WantManager) startPeerHandler(p peer.ID) *msgQueue {
138
// new peer, we will want to give them our full wantlist
139
fullwantlist := bsmsg.New(true)
140
for _, e := range pm.wl.Entries() {
141
+ ne := *e
142
+ mq.wl.AddEntry(&ne)
143
fullwantlist.AddEntry(e.Cid, e.Priority)
144
}
145
mq.out = fullwantlist
@@ -278,27 +285,35 @@ func (pm *WantManager) Run() {
285
defer tock.Stop()
286
for {
287
select {
281
- case entries := <-pm.incoming:
288
+ case ws := <-pm.incoming:
289
290
// add changes to our wantlist
284
- var filtered []*bsmsg.Entry
285
- for _, e := range entries {
291
+ for _, e := range ws.entries {
292
if e.Cancel {
293
if pm.wl.Remove(e.Cid) {
294
pm.wantlistGauge.Dec()
289
- filtered = append(filtered, e)
295
}
296
} else {
297
if pm.wl.AddEntry(e.Entry) {
298
pm.wantlistGauge.Inc()
294
- filtered = append(filtered, e)
299
}
300
}
301
}
302
303
// broadcast those wantlist changes
300
- for _, p := range pm.peers {
301
- p.addMessage(filtered)
304
+ if len(ws.targets) == 0 {
305
+ for _, p := range pm.peers {
306
+ p.addMessage(ws.entries)
307
+ }
308
+ } else {
309
+ for _, t := range ws.targets {
310
+ p, ok := pm.peers[t]
311
+ if !ok {
312
+ log.Warning("tried sending wantlist change to non-partner peer")
313
+ continue
314
+ }
315
+ p.addMessage(ws.entries)
316
+ }
317
}
318
319
case <-tock.C:
@@ -335,6 +350,7 @@ func (wm *WantManager) newMsgQueue(p peer.ID) *msgQueue {
350
return &msgQueue{
351
done: make(chan struct{}),
352
work: make(chan struct{}, 1),
353
+ wl: wantlist.New(),
354
network: wm.network,
355
p: p,
356
refcnt: 1,
@@ -342,9 +358,13 @@ func (wm *WantManager) newMsgQueue(p peer.ID) *msgQueue {
358
}
359
360
func (mq *msgQueue) addMessage(entries []*bsmsg.Entry) {
361
+ var work bool
362
mq.outlk.Lock()
363
defer func() {
364
mq.outlk.Unlock()
365
+ if !work {
366
+ return
367
+ }
368
select {
369
case mq.work <- struct{}{}:
370
default:
@@ -361,9 +381,15 @@ func (mq *msgQueue) addMessage(entries []*bsmsg.Entry) {
381
// one passed in
382
for _, e := range entries {
383
if e.Cancel {
364
- mq.out.Cancel(e.Cid)
384
+ if mq.wl.Remove(e.Cid) {
385
+ work = true
386
+ mq.out.Cancel(e.Cid)
387
+ }
388
} else {
366
- mq.out.AddEntry(e.Cid, e.Priority)
389
+ if mq.wl.Add(e.Cid, e.Priority) {
390
+ work = true
391
+ mq.out.AddEntry(e.Cid, e.Priority)
392
+ }
393
}
394
}
395
}