try harder to not send duplicate blocks
Jeromy committed
Apr 29, 2015 at 01:36 UTC
86fb07aed7d32ccd1a33f8bab0ea47aa3adbb359
2 files changed
+24
-6
exchange/bitswap/decision/peer_request_queue.go
+23
-5
@@ -46,7 +46,7 @@ func (tl *prq) Push(entry wantlist.Entry, to peer.ID) {
46
defer tl.lock.Unlock()
47
partner, ok := tl.partners[to]
48
if !ok {
49
- partner = &activePartner{taskQueue: pq.New(wrapCmp(V1))}
49
+ partner = newActivePartner()
50
tl.pQueue.Push(partner)
51
tl.partners[to] = partner
52
}
@@ -57,12 +57,19 @@ func (tl *prq) Push(entry wantlist.Entry, to peer.ID) {
57
return
58
}
59
60
+ partner.activelk.Lock()
61
+ defer partner.activelk.Unlock()
62
+ _, ok = partner.activeBlocks[entry.Key]
63
+ if ok {
64
+ return
65
+ }
66
+
67
task := &peerRequestTask{
68
Entry: entry,
69
Target: to,
70
created: time.Now(),
71
Done: func() {
65
- partner.TaskDone()
72
+ partner.TaskDone(entry.Key)
73
tl.lock.Lock()
74
tl.pQueue.Update(partner.Index())
75
tl.lock.Unlock()
@@ -93,7 +100,7 @@ func (tl *prq) Pop() *peerRequestTask {
100
continue // discarding tasks that have been removed
101
}
102
96
- partner.StartTask()
103
+ partner.StartTask(out.Entry.Key)
104
partner.requests--
105
break // and return |out|
106
}
@@ -179,6 +186,8 @@ type activePartner struct {
186
activelk sync.Mutex
187
active int
188
189
+ activeBlocks map[u.Key]struct{}
190
+
191
// requests is the number of blocks this peer is currently requesting
192
// request need not be locked around as it will only be modified under
193
// the peerRequestQueue's locks
@@ -191,6 +200,13 @@ type activePartner struct {
200
taskQueue pq.PQ
201
}
202
203
+func newActivePartner() *activePartner {
204
+ return &activePartner{
205
+ taskQueue: pq.New(wrapCmp(V1)),
206
+ activeBlocks: make(map[u.Key]struct{}),
207
+ }
208
+}
209
+
210
// partnerCompare implements pq.ElemComparator
211
func partnerCompare(a, b pq.Elem) bool {
212
pa := a.(*activePartner)
@@ -208,15 +224,17 @@ func partnerCompare(a, b pq.Elem) bool {
224
}
225
226
// StartTask signals that a task was started for this partner
211
-func (p *activePartner) StartTask() {
227
+func (p *activePartner) StartTask(k u.Key) {
228
p.activelk.Lock()
229
+ p.activeBlocks[k] = struct{}{}
230
p.active++
231
p.activelk.Unlock()
232
}
233
234
// TaskDone signals that a task was completed for this partner
218
-func (p *activePartner) TaskDone() {
235
+func (p *activePartner) TaskDone(k u.Key) {
236
p.activelk.Lock()
237
+ delete(p.activeBlocks, k)
238
p.active--
239
if p.active < 0 {
240
panic("more tasks finished than started!")
exchange/bitswap/workers.go
+1
-1
@@ -11,7 +11,7 @@ import (
11
u "github.com/ipfs/go-ipfs/util"
12
)
13
14
-var TaskWorkerCount = 16
14
+var TaskWorkerCount = 8
15
16
func init() {
17
twc := os.Getenv("IPFS_BITSWAP_TASK_WORKERS")