some code cleanup and commenting
Jeromy committed
Apr 3, 2015 at 11:40 UTC
738db201d923a9931b31d9c7d6e4b4cee0f5cbff
2 files changed
+19
-11
exchange/bitswap/decision/engine.go
-4
@@ -55,9 +55,6 @@ 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()
58
}
59
60
type Engine struct {
@@ -143,7 +140,6 @@ func (e *Engine) nextEnvelope(ctx context.Context) (*Envelope, error) {
140
return &Envelope{
141
Peer: nextTask.Target,
142
Message: m,
146
- Sent: nextTask.Done,
143
}, nil
144
}
145
}
exchange/bitswap/decision/peer_request_queue.go
+19
-7
@@ -27,6 +27,7 @@ func newPRQ() peerRequestQueue {
27
}
28
}
29
30
+// verify interface implementation
31
var _ peerRequestQueue = &prq{}
32
33
// TODO: at some point, the strategy needs to plug in here
@@ -81,12 +82,7 @@ func (tl *prq) Pop() *peerRequestTask {
82
if tl.pQueue.Len() == 0 {
83
return nil
84
}
84
- pElem := tl.pQueue.Pop()
85
- if pElem == nil {
86
- return nil
87
- }
88
-
89
- partner := pElem.(*activePartner)
85
+ partner := tl.pQueue.Pop().(*activePartner)
86
87
var out *peerRequestTask
88
for partner.taskQueue.Len() > 0 {
@@ -97,6 +93,8 @@ func (tl *prq) Pop() *peerRequestTask {
93
}
94
break // and return |out|
95
}
96
+
97
+ // start the new task, and push the partner back onto the queue
98
partner.StartTask()
99
partner.requests--
100
tl.pQueue.Push(partner)
@@ -112,6 +110,8 @@ func (tl *prq) Remove(k u.Key, p peer.ID) {
110
// simply mark it as trash, so it'll be dropped when popped off the
111
// queue.
112
t.trash = true
113
+
114
+ // having canceled a block, we now account for that in the given partner
115
tl.partners[p].requests--
116
}
117
tl.lock.Unlock()
@@ -121,6 +121,7 @@ type peerRequestTask struct {
121
Entry wantlist.Entry
122
Target peer.ID
123
124
+ // A callback to signal that this task has been completed
125
Done func()
126
127
// trash in a book-keeping field
@@ -135,10 +136,12 @@ func (t *peerRequestTask) Key() string {
136
return taskKey(t.Target, t.Entry.Key)
137
}
138
139
+// Index implements pq.Elem
140
func (t *peerRequestTask) Index() int {
141
return t.index
142
}
143
144
+// SetIndex implements pq.Elem
145
func (t *peerRequestTask) SetIndex(i int) {
146
t.index = i
147
}
@@ -172,17 +175,22 @@ type activePartner struct {
175
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
179
active int
180
181
// requests is the number of blocks this peer is currently requesting
182
+ // request need not be locked around as it will only be modified under
183
+ // the peerRequestQueue's locks
184
requests int
185
186
+ // for the PQ interface
187
index int
188
182
- // priority queue of
189
+ // priority queue of tasks belonging to this peer
190
taskQueue pq.PQ
191
}
192
193
+// partnerCompare implements pq.ElemComparator
194
func partnerCompare(a, b pq.Elem) bool {
195
pa := a.(*activePartner)
196
pb := b.(*activePartner)
@@ -197,12 +205,14 @@ func partnerCompare(a, b pq.Elem) bool {
205
return pa.active < pb.active
206
}
207
208
+// StartTask signals that a task was started for this partner
209
func (p *activePartner) StartTask() {
210
p.lk.Lock()
211
p.active++
212
p.lk.Unlock()
213
}
214
215
+// TaskDone signals that a task was completed for this partner
216
func (p *activePartner) TaskDone() {
217
p.lk.Lock()
218
p.active--
@@ -212,10 +222,12 @@ func (p *activePartner) TaskDone() {
222
p.lk.Unlock()
223
}
224
225
+// Index implements pq.Elem
226
func (p *activePartner) Index() int {
227
return p.index
228
}
229
230
+// SetIndex implements pq.Elem
231
func (p *activePartner) SetIndex(i int) {
232
p.index = i
233
}