address comments from CR
Jeromy committed
Apr 3, 2015 at 15:37 UTC
8450a8d4d827219952c41f2115e0f42d38ecdcd1
4 files changed
+25
-11
exchange/bitswap/decision/engine.go
+7
@@ -55,6 +55,9 @@ type Envelope struct {
55
Peer peer.ID
56
// Message is the payload
57
Message bsmsg.BitSwapMessage
58
+
59
+ // A callback to notify the decision queue that the task is complete
60
+ Sent func()
61
}
62
63
type Engine struct {
@@ -132,6 +135,9 @@ func (e *Engine) nextEnvelope(ctx context.Context) (*Envelope, error) {
135
136
block, err := e.bs.Get(nextTask.Entry.Key)
137
if err != nil {
138
+ // If we don't have the block, don't hold that against the peer
139
+ // make sure to update that the task has been 'completed'
140
+ nextTask.Done()
141
continue
142
}
143
@@ -140,6 +146,7 @@ func (e *Engine) nextEnvelope(ctx context.Context) (*Envelope, error) {
146
return &Envelope{
147
Peer: nextTask.Target,
148
Message: m,
149
+ Sent: nextTask.Done,
150
}, nil
151
}
152
}
exchange/bitswap/decision/peer_request_queue.go
+7
-6
@@ -173,11 +173,11 @@ func wrapCmp(f func(a, b *peerRequestTask) bool) func(a, b pq.Elem) bool {
173
}
174
175
type activePartner struct {
176
- lk sync.Mutex
176
177
// Active is the number of blocks this peer is currently being sent
178
// active must be locked around as it will be updated externally
180
- active int
179
+ activelk sync.Mutex
180
+ active int
181
182
// requests is the number of blocks this peer is currently requesting
183
// request need not be locked around as it will only be modified under
@@ -197,6 +197,7 @@ func partnerCompare(a, b pq.Elem) bool {
197
pb := b.(*activePartner)
198
199
// having no blocks in their wantlist means lowest priority
200
+ // having both of these checks ensures stability of the sort
201
if pa.requests == 0 {
202
return false
203
}
@@ -208,19 +209,19 @@ func partnerCompare(a, b pq.Elem) bool {
209
210
// StartTask signals that a task was started for this partner
211
func (p *activePartner) StartTask() {
211
- p.lk.Lock()
212
+ p.activelk.Lock()
213
p.active++
213
- p.lk.Unlock()
214
+ p.activelk.Unlock()
215
}
216
217
// TaskDone signals that a task was completed for this partner
218
func (p *activePartner) TaskDone() {
218
- p.lk.Lock()
219
+ p.activelk.Lock()
220
p.active--
221
if p.active < 0 {
222
panic("more tasks finished than started!")
223
}
223
- p.lk.Unlock()
224
+ p.activelk.Unlock()
225
}
226
227
// Index implements pq.Elem
exchange/bitswap/decision/peer_request_queue_test.go
+10
-5
@@ -105,10 +105,15 @@ func TestPeerRepeats(t *testing.T) {
105
106
// Now, if one of the tasks gets finished, the next task off the queue should
107
// be for the same peer
108
- tasks[0].Done()
109
-
110
- ntask := prq.Pop()
111
- if ntask.Target != tasks[0].Target {
112
- t.Fatal("Expected task from peer with lowest active count")
108
+ for blockI := 0; blockI < 4; blockI++ {
109
+ for i := 0; i < 4; i++ {
110
+ // its okay to mark the same task done multiple times here (JUST FOR TESTING)
111
+ tasks[i].Done()
112
+
113
+ ntask := prq.Pop()
114
+ if ntask.Target != tasks[i].Target {
115
+ t.Fatal("Expected task from peer with lowest active count")
116
+ }
117
+ }
118
}
119
}
exchange/bitswap/workers.go
+1
@@ -51,6 +51,7 @@ func (bs *Bitswap) taskWorker(ctx context.Context) {
51
}
52
log.Event(ctx, "deliverBlocks", envelope.Message, envelope.Peer)
53
bs.send(ctx, envelope.Peer, envelope.Message)
54
+ envelope.Sent()
55
case <-ctx.Done():
56
return
57
}