fix race bugs
Jeromy committed
May 16, 2015 at 14:26 UTC
32da687774427bdd44efd1686218448e8c834fd0
5 files changed
+17
-31
exchange/bitswap/bitswap.go
+3
@@ -128,6 +128,7 @@ type Bitswap struct {
128
129
provideKeys chan u.Key
130
131
+ counterLk sync.Mutex
132
blocksRecvd int
133
dupBlocksRecvd int
134
}
@@ -281,10 +282,12 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
282
bs.wm.CancelWants(keys)
283
284
for _, block := range incoming.Blocks() {
285
+ bs.counterLk.Lock()
286
bs.blocksRecvd++
287
if has, err := bs.blockstore.Has(block.Key()); err == nil && has {
288
bs.dupBlocksRecvd++
289
}
290
+ bs.counterLk.Unlock()
291
log.Debugf("got block %s from %s", block, p)
292
293
hasBlockCtx, cancel := context.WithTimeout(ctx, hasBlockTimeout)
exchange/bitswap/decision/engine.go
+2
-2
@@ -210,11 +210,11 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {
210
211
for _, entry := range m.Wantlist() {
212
if entry.Cancel {
213
- log.Errorf("cancel %s", entry.Key)
213
+ log.Debugf("cancel %s", entry.Key)
214
l.CancelWant(entry.Key)
215
e.peerRequestQueue.Remove(entry.Key, p)
216
} else {
217
- log.Errorf("wants %s - %d", entry.Key, entry.Priority)
217
+ log.Debugf("wants %s - %d", entry.Key, entry.Priority)
218
l.Wants(entry.Key, entry.Priority)
219
if exists, err := e.bs.Has(entry.Key); err == nil && exists {
220
e.peerRequestQueue.Push(entry.Entry, p)
exchange/bitswap/message/message.go
+1
-1
@@ -162,7 +162,7 @@ func (m *impl) ToProto() *pb.Message {
162
pbm.Wantlist.Entries = append(pbm.Wantlist.Entries, &pb.Message_Wantlist_Entry{
163
Block: proto.String(string(e.Key)),
164
Priority: proto.Int32(int32(e.Priority)),
165
- Cancel: &e.Cancel,
165
+ Cancel: proto.Bool(e.Cancel),
166
})
167
}
168
for _, b := range m.Blocks() {
exchange/bitswap/peermanager.go
+9
-28
@@ -62,12 +62,10 @@ type msgQueue struct {
62
}
63
64
func (pm *WantManager) WantBlocks(ks []u.Key) {
65
- log.Error("WANT: ", ks)
65
pm.addEntries(ks, false)
66
}
67
68
func (pm *WantManager) CancelWants(ks []u.Key) {
70
- log.Error("CANCEL: ", ks)
69
pm.addEntries(ks, true)
70
}
71
@@ -147,18 +145,12 @@ func (pm *WantManager) runQueue(ctx context.Context, mq *msgQueue) {
145
// grab outgoing message
146
mq.outlk.Lock()
147
wlm := mq.out
150
- mq.out = nil
151
- mq.outlk.Unlock()
152
-
153
- // no message or empty message, continue
154
- if wlm == nil {
155
- log.Error("nil wantlist")
156
- continue
157
- }
158
- if wlm.Empty() {
159
- log.Error("empty wantlist")
148
+ if wlm == nil || wlm.Empty() {
149
+ mq.outlk.Unlock()
150
continue
151
}
152
+ mq.out = nil
153
+ mq.outlk.Unlock()
154
155
// send wantlist updates
156
err = pm.network.SendMessage(ctx, mq.p, wlm)
@@ -186,22 +178,18 @@ func (pm *WantManager) Run(ctx context.Context) {
178
select {
179
case entries := <-pm.incoming:
180
189
- msg := bsmsg.New()
190
- msg.SetFull(false)
181
// add changes to our wantlist
182
for _, e := range entries {
183
if e.Cancel {
184
pm.wl.Remove(e.Key)
195
- msg.Cancel(e.Key)
185
} else {
186
pm.wl.Add(e.Key, e.Priority)
198
- msg.AddEntry(e.Key, e.Priority)
187
}
188
}
189
190
// broadcast those wantlist changes
191
for _, p := range pm.peers {
204
- p.addMessage(msg)
192
+ p.addMessage(entries)
193
}
194
195
case p := <-pm.connect:
@@ -223,7 +211,7 @@ func newMsgQueue(p peer.ID) *msgQueue {
211
return mq
212
}
213
226
-func (mq *msgQueue) addMessage(msg bsmsg.BitSwapMessage) {
214
+func (mq *msgQueue) addMessage(entries []*bsmsg.Entry) {
215
mq.outlk.Lock()
216
defer func() {
217
mq.outlk.Unlock()
@@ -233,26 +221,19 @@ func (mq *msgQueue) addMessage(msg bsmsg.BitSwapMessage) {
221
}
222
}()
223
236
- if msg.Full() {
237
- log.Error("GOt FULL MESSAGE")
238
- }
239
-
224
// if we have no message held, or the one we are given is full
225
// overwrite the one we are holding
242
- if mq.out == nil || msg.Full() {
243
- mq.out = msg
244
- return
226
+ if mq.out == nil {
227
+ mq.out = bsmsg.New()
228
}
229
230
// TODO: add a msg.Combine(...) method
231
// otherwise, combine the one we are holding with the
232
// one passed in
250
- for _, e := range msg.Wantlist() {
233
+ for _, e := range entries {
234
if e.Cancel {
252
- log.Error("add message cancel: ", e.Key, mq.p)
235
mq.out.Cancel(e.Key)
236
} else {
255
- log.Error("add message want: ", e.Key, mq.p)
237
mq.out.AddEntry(e.Key, e.Priority)
238
}
239
}
exchange/bitswap/stat.go
+2
@@ -17,8 +17,10 @@ func (bs *Bitswap) Stat() (*Stat, error) {
17
st := new(Stat)
18
st.ProvideBufLen = len(bs.newBlocks)
19
st.Wantlist = bs.GetWantlist()
20
+ bs.counterLk.Lock()
21
st.BlocksReceived = bs.blocksRecvd
22
st.DupBlksReceived = bs.dupBlocksRecvd
23
+ bs.counterLk.Unlock()
24
25
for _, p := range bs.engine.Peers() {
26
st.Peers = append(st.Peers, p.Pretty())