fix doubleGet issue caused by hasblock not announcing
License: MIT Signed-off-by: Jeromy <why@ipfs.io>
Jeromy committed
Apr 27, 2016 at 20:45 UTC
01d1b69da2b6972eba2e7d6cf59c1bd7ccd05fc9
5 files changed
+57
-23
exchange/bitswap/bitswap.go
+2
@@ -264,6 +264,8 @@ func (bs *Bitswap) HasBlock(blk *blocks.Block) error {
264
265
bs.notifications.Publish(blk)
266
267
+ bs.engine.AddBlock(blk)
268
+
269
select {
270
case bs.newBlocks <- blk:
271
// send block off to be reprovided
exchange/bitswap/bitswap_test.go
+2
@@ -335,6 +335,8 @@ func TestDoubleGet(t *testing.T) {
335
t.Fatal(err)
336
}
337
338
+ // ensure both requests make it into the wantlist at the same time
339
+ time.Sleep(time.Millisecond * 100)
340
cancel1()
341
342
_, ok := <-blkch1
exchange/bitswap/decision/engine.go
+26
-10
@@ -83,7 +83,7 @@ type Engine struct {
83
84
bs bstore.Blockstore
85
86
- lock sync.RWMutex // protects the fields immediatly below
86
+ lock sync.Mutex // protects the fields immediatly below
87
// ledgerMap lists Ledgers by their Partner key.
88
ledgerMap map[peer.ID]*ledger
89
}
@@ -178,8 +178,8 @@ func (e *Engine) Outbox() <-chan (<-chan *Envelope) {
178
179
// Returns a slice of Peers with whom the local node has active sessions
180
func (e *Engine) Peers() []peer.ID {
181
- e.lock.RLock()
182
- defer e.lock.RUnlock()
181
+ e.lock.Lock()
182
+ defer e.lock.Unlock()
183
184
response := make([]peer.ID, 0)
185
for _, ledger := range e.ledgerMap {
@@ -217,7 +217,7 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {
217
e.peerRequestQueue.Remove(entry.Key, p)
218
} else {
219
log.Debugf("wants %s - %d", entry.Key, entry.Priority)
220
- l.Wants(entry.Ctx, entry.Key, entry.Priority)
220
+ l.Wants(entry.Key, entry.Priority)
221
if exists, err := e.bs.Has(entry.Key); err == nil && exists {
222
e.peerRequestQueue.Push(entry.Entry, p)
223
newWorkExists = true
@@ -228,16 +228,32 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {
228
for _, block := range m.Blocks() {
229
log.Debugf("got block %s %d bytes", block.Key(), len(block.Data))
230
l.ReceivedBytes(len(block.Data))
231
- for _, l := range e.ledgerMap {
232
- if entry, ok := l.WantListContains(block.Key()); ok {
233
- e.peerRequestQueue.Push(entry, l.Partner)
234
- newWorkExists = true
235
- }
236
- }
231
}
232
return nil
233
}
234
235
+func (e *Engine) addBlock(block *blocks.Block) {
236
+ work := false
237
+
238
+ for _, l := range e.ledgerMap {
239
+ if entry, ok := l.WantListContains(block.Key()); ok {
240
+ e.peerRequestQueue.Push(entry, l.Partner)
241
+ work = true
242
+ }
243
+ }
244
+
245
+ if work {
246
+ e.signalNewWork()
247
+ }
248
+}
249
+
250
+func (e *Engine) AddBlock(block *blocks.Block) {
251
+ e.lock.Lock()
252
+ defer e.lock.Unlock()
253
+
254
+ e.addBlock(block)
255
+}
256
+
257
// TODO add contents of m.WantList() to my local wantlist? NB: could introduce
258
// race conditions where I send a message, but MessageSent gets handled after
259
// MessageReceived. The information in the local wantlist could become
exchange/bitswap/decision/ledger.go
+2
-5
@@ -6,8 +6,6 @@ import (
6
key "github.com/ipfs/go-ipfs/blocks/key"
7
wl "github.com/ipfs/go-ipfs/exchange/bitswap/wantlist"
8
peer "gx/ipfs/QmZwZjMVGss5rqYsJVGy18gNbkTJffFyq2x1uJ4e4p3ZAt/go-libp2p-peer"
9
-
10
- "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
9
)
10
11
// keySet is just a convenient alias for maps of keys, where we only care
@@ -69,10 +67,9 @@ func (l *ledger) ReceivedBytes(n int) {
67
l.Accounting.BytesRecv += uint64(n)
68
}
69
72
-// TODO: this needs to be different. We need timeouts.
73
-func (l *ledger) Wants(ctx context.Context, k key.Key, priority int) {
70
+func (l *ledger) Wants(k key.Key, priority int) {
71
log.Debugf("peer %s wants %s", l.Partner, k)
75
- l.wantList.Add(ctx, k, priority)
72
+ l.wantList.Add(k, priority)
73
}
74
75
func (l *ledger) CancelWant(k key.Key) {
exchange/bitswap/wantlist/wantlist.go
+25
-8
@@ -22,11 +22,12 @@ type Wantlist struct {
22
}
23
24
type Entry struct {
25
- // TODO consider making entries immutable so they can be shared safely and
26
- // slices can be copied efficiently.
25
Key key.Key
26
Priority int
29
- Ctx context.Context
27
+
28
+ Ctx context.Context
29
+ cancel func()
30
+ RefCnt int
31
}
32
33
type entrySlice []Entry
@@ -47,10 +48,10 @@ func New() *Wantlist {
48
}
49
}
50
50
-func (w *ThreadSafe) Add(ctx context.Context, k key.Key, priority int) {
51
+func (w *ThreadSafe) Add(k key.Key, priority int) {
52
w.lk.Lock()
53
defer w.lk.Unlock()
53
- w.Wantlist.Add(ctx, k, priority)
54
+ w.Wantlist.Add(k, priority)
55
}
56
57
func (w *ThreadSafe) AddEntry(e Entry) {
@@ -93,14 +94,19 @@ func (w *Wantlist) Len() int {
94
return len(w.set)
95
}
96
96
-func (w *Wantlist) Add(ctx context.Context, k key.Key, priority int) {
97
- if _, ok := w.set[k]; ok {
97
+func (w *Wantlist) Add(k key.Key, priority int) {
98
+ if e, ok := w.set[k]; ok {
99
+ e.RefCnt++
100
return
101
}
102
+
103
+ ctx, cancel := context.WithCancel(context.Background())
104
w.set[k] = Entry{
105
Key: k,
106
Priority: priority,
107
Ctx: ctx,
108
+ cancel: cancel,
109
+ RefCnt: 1,
110
}
111
}
112
@@ -112,7 +118,18 @@ func (w *Wantlist) AddEntry(e Entry) {
118
}
119
120
func (w *Wantlist) Remove(k key.Key) {
115
- delete(w.set, k)
121
+ e, ok := w.set[k]
122
+ if !ok {
123
+ return
124
+ }
125
+
126
+ e.RefCnt--
127
+ if e.RefCnt <= 0 {
128
+ delete(w.set, k)
129
+ if e.cancel != nil {
130
+ e.cancel()
131
+ }
132
+ }
133
}
134
135
func (w *Wantlist) Contains(k key.Key) (Entry, bool) {