@cryptotaxi247 / kubo / commits / 1548c8aab

bitswap: clear wantlists when GetBlocks calls are cancelled

License: MIT Signed-off-by: Jeromy <why@ipfs.io>

Jeromy committed Sep 2, 2016 at 15:28 UTC 1548c8aabb20fe8e60bf09b25a6ae1cad64c9c0e
11 files changed +117 -89
exchange/bitswap/bitswap.go
+43 -7
@@ -22,7 +22,6 @@ import (
22 bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message"
23 bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
24 notifications "github.com/ipfs/go-ipfs/exchange/bitswap/notifications"
25 - wantlist "github.com/ipfs/go-ipfs/exchange/bitswap/wantlist"
25 flags "github.com/ipfs/go-ipfs/flags"
26 "github.com/ipfs/go-ipfs/thirdparty/delay"
27 loggables "github.com/ipfs/go-ipfs/thirdparty/loggables"
@@ -88,7 +87,7 @@ func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
87 notifications: notif,
88 engine: decision.NewEngine(ctx, bstore), // TODO close the engine with Close() method
89 network: network,
91 - findKeys: make(chan *wantlist.Entry, sizeBatchRequestChan),
90 + findKeys: make(chan *blockRequest, sizeBatchRequestChan),
91 process: px,
92 newBlocks: make(chan blocks.Block, HasBlockBufferSize),
93 provideKeys: make(chan key.Key, provideKeysBufferSize),
@@ -131,7 +130,7 @@ type Bitswap struct {
130 notifications notifications.PubSub
131
132 // send keys to a worker to find and connect to providers for them
134 - findKeys chan *wantlist.Entry
133 + findKeys chan *blockRequest
134
135 engine *decision.Engine
136
@@ -148,8 +147,8 @@ type Bitswap struct {
147 }
148
149 type blockRequest struct {
151 - key key.Key
152 - ctx context.Context
150 + Key key.Key
151 + Ctx context.Context
152 }
153
154 // GetBlock attempts to retrieve a particular block from peers within the
@@ -235,13 +234,50 @@ func (bs *Bitswap) GetBlocks(ctx context.Context, keys []key.Key) (<-chan blocks
234 // NB: Optimization. Assumes that providers of key[0] are likely to
235 // be able to provide for all keys. This currently holds true in most
236 // every situation. Later, this assumption may not hold as true.
238 - req := &wantlist.Entry{
237 + req := &blockRequest{
238 Key: keys[0],
239 Ctx: ctx,
240 }
241 +
242 + remaining := make(map[key.Key]struct{})
243 + for _, k := range keys {
244 + remaining[k] = struct{}{}
245 + }
246 +
247 + out := make(chan blocks.Block)
248 + go func() {
249 + ctx, cancel := context.WithCancel(ctx)
250 + defer cancel()
251 + defer close(out)
252 + defer func() {
253 + var toCancel []key.Key
254 + for k, _ := range remaining {
255 + toCancel = append(toCancel, k)
256 + }
257 + bs.CancelWants(toCancel)
258 + }()
259 + for {
260 + select {
261 + case blk, ok := <-promise:
262 + if !ok {
263 + return
264 + }
265 +
266 + delete(remaining, blk.Key())
267 + select {
268 + case out <- blk:
269 + case <-ctx.Done():
270 + return
271 + }
272 + case <-ctx.Done():
273 + return
274 + }
275 + }
276 + }()
277 +
278 select {
279 case bs.findKeys <- req:
244 - return promise, nil
280 + return out, nil
281 case <-ctx.Done():
282 return nil, ctx.Err()
283 }
exchange/bitswap/decision/bench_test.go
+1 -1
@@ -21,6 +21,6 @@ func BenchmarkTaskQueuePush(b *testing.B) {
21 }
22 b.ResetTimer()
23 for i := 0; i < b.N; i++ {
24 - q.Push(wantlist.Entry{Key: key.Key(i), Priority: math.MaxInt32}, peers[i%len(peers)])
24 + q.Push(&wantlist.Entry{Key: key.Key(i), Priority: math.MaxInt32}, peers[i%len(peers)])
25 }
26 }
exchange/bitswap/decision/engine.go
+2 -2
@@ -104,7 +104,7 @@ func NewEngine(ctx context.Context, bs bstore.Blockstore) *Engine {
104 return e
105 }
106
107 -func (e *Engine) WantlistForPeer(p peer.ID) (out []wl.Entry) {
107 +func (e *Engine) WantlistForPeer(p peer.ID) (out []*wl.Entry) {
108 e.lock.Lock()
109 partner, ok := e.ledgerMap[p]
110 if ok {
@@ -218,7 +218,7 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {
218
219 for _, entry := range m.Wantlist() {
220 if entry.Cancel {
221 - log.Debugf("cancel %s", entry.Key)
221 + log.Debugf("%s cancel %s", p, entry.Key)
222 l.CancelWant(entry.Key)
223 e.peerRequestQueue.Remove(entry.Key, p)
224 } else {
exchange/bitswap/decision/ledger.go
+1 -1
@@ -79,7 +79,7 @@ func (l *ledger) CancelWant(k key.Key) {
79 l.wantList.Remove(k)
80 }
81
82 -func (l *ledger) WantListContains(k key.Key) (wl.Entry, bool) {
82 +func (l *ledger) WantListContains(k key.Key) (*wl.Entry, bool) {
83 return l.wantList.Contains(k)
84 }
85
exchange/bitswap/decision/peer_request_queue.go
+3 -3
@@ -13,7 +13,7 @@ import (
13 type peerRequestQueue interface {
14 // Pop returns the next peerRequestTask. Returns nil if the peerRequestQueue is empty.
15 Pop() *peerRequestTask
16 - Push(entry wantlist.Entry, to peer.ID)
16 + Push(entry *wantlist.Entry, to peer.ID)
17 Remove(k key.Key, p peer.ID)
18
19 // NB: cannot expose simply expose taskQueue.Len because trashed elements
@@ -45,7 +45,7 @@ type prq struct {
45 }
46
47 // Push currently adds a new peerRequestTask to the end of the list
48 -func (tl *prq) Push(entry wantlist.Entry, to peer.ID) {
48 +func (tl *prq) Push(entry *wantlist.Entry, to peer.ID) {
49 tl.lock.Lock()
50 defer tl.lock.Unlock()
51 partner, ok := tl.partners[to]
@@ -166,7 +166,7 @@ func (tl *prq) thawRound() {
166 }
167
168 type peerRequestTask struct {
169 - Entry wantlist.Entry
169 + Entry *wantlist.Entry
170 Target peer.ID
171
172 // A callback to signal that this task has been completed
exchange/bitswap/decision/peer_request_queue_test.go
+5 -5
@@ -41,7 +41,7 @@ func TestPushPop(t *testing.T) {
41 for _, index := range rand.Perm(len(alphabet)) { // add blocks for all letters
42 letter := alphabet[index]
43 t.Log(partner.String())
44 - prq.Push(wantlist.Entry{Key: key.Key(letter), Priority: math.MaxInt32 - index}, partner)
44 + prq.Push(&wantlist.Entry{Key: key.Key(letter), Priority: math.MaxInt32 - index}, partner)
45 }
46 for _, consonant := range consonants {
47 prq.Remove(key.Key(consonant), partner)
@@ -78,10 +78,10 @@ func TestPeerRepeats(t *testing.T) {
78 // Have each push some blocks
79
80 for i := 0; i < 5; i++ {
81 - prq.Push(wantlist.Entry{Key: key.Key(i)}, a)
82 - prq.Push(wantlist.Entry{Key: key.Key(i)}, b)
83 - prq.Push(wantlist.Entry{Key: key.Key(i)}, c)
84 - prq.Push(wantlist.Entry{Key: key.Key(i)}, d)
81 + prq.Push(&wantlist.Entry{Key: key.Key(i)}, a)
82 + prq.Push(&wantlist.Entry{Key: key.Key(i)}, b)
83 + prq.Push(&wantlist.Entry{Key: key.Key(i)}, c)
84 + prq.Push(&wantlist.Entry{Key: key.Key(i)}, d)
85 }
86
87 // now, pop off four entries, there should be one from each
exchange/bitswap/message/message.go
+2 -2
@@ -64,7 +64,7 @@ func newMsg(full bool) *impl {
64 }
65
66 type Entry struct {
67 - wantlist.Entry
67 + *wantlist.Entry
68 Cancel bool
69 }
70
@@ -120,7 +120,7 @@ func (m *impl) addEntry(k key.Key, priority int, cancel bool) {
120 e.Cancel = cancel
121 } else {
122 m.wantlist[k] = Entry{
123 - Entry: wantlist.Entry{
123 + Entry: &wantlist.Entry{
124 Key: k,
125 Priority: priority,
126 },
exchange/bitswap/wantlist/wantlist.go
+29 -33
@@ -7,8 +7,6 @@ import (
7 "sync"
8
9 key "github.com/ipfs/go-ipfs/blocks/key"
10 -
11 - "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
10 )
11
12 type ThreadSafe struct {
@@ -18,19 +16,17 @@ type ThreadSafe struct {
16
17 // not threadsafe
18 type Wantlist struct {
21 - set map[key.Key]Entry
19 + set map[key.Key]*Entry
20 }
21
22 type Entry struct {
23 Key key.Key
24 Priority int
25
28 - Ctx context.Context
29 - cancel func()
26 RefCnt int
27 }
28
33 -type entrySlice []Entry
29 +type entrySlice []*Entry
30
31 func (es entrySlice) Len() int { return len(es) }
32 func (es entrySlice) Swap(i, j int) { es[i], es[j] = es[j], es[i] }
@@ -44,41 +40,41 @@ func NewThreadSafe() *ThreadSafe {
40
41 func New() *Wantlist {
42 return &Wantlist{
47 - set: make(map[key.Key]Entry),
43 + set: make(map[key.Key]*Entry),
44 }
45 }
46
51 -func (w *ThreadSafe) Add(k key.Key, priority int) {
47 +func (w *ThreadSafe) Add(k key.Key, priority int) bool {
48 w.lk.Lock()
49 defer w.lk.Unlock()
54 - w.Wantlist.Add(k, priority)
50 + return w.Wantlist.Add(k, priority)
51 }
52
57 -func (w *ThreadSafe) AddEntry(e Entry) {
53 +func (w *ThreadSafe) AddEntry(e *Entry) bool {
54 w.lk.Lock()
55 defer w.lk.Unlock()
60 - w.Wantlist.AddEntry(e)
56 + return w.Wantlist.AddEntry(e)
57 }
58
63 -func (w *ThreadSafe) Remove(k key.Key) {
59 +func (w *ThreadSafe) Remove(k key.Key) bool {
60 w.lk.Lock()
61 defer w.lk.Unlock()
66 - w.Wantlist.Remove(k)
62 + return w.Wantlist.Remove(k)
63 }
64
69 -func (w *ThreadSafe) Contains(k key.Key) (Entry, bool) {
65 +func (w *ThreadSafe) Contains(k key.Key) (*Entry, bool) {
66 w.lk.RLock()
67 defer w.lk.RUnlock()
68 return w.Wantlist.Contains(k)
69 }
70
75 -func (w *ThreadSafe) Entries() []Entry {
71 +func (w *ThreadSafe) Entries() []*Entry {
72 w.lk.RLock()
73 defer w.lk.RUnlock()
74 return w.Wantlist.Entries()
75 }
76
81 -func (w *ThreadSafe) SortedEntries() []Entry {
77 +func (w *ThreadSafe) SortedEntries() []*Entry {
78 w.lk.RLock()
79 defer w.lk.RUnlock()
80 return w.Wantlist.SortedEntries()
@@ -94,50 +90,50 @@ func (w *Wantlist) Len() int {
90 return len(w.set)
91 }
92
97 -func (w *Wantlist) Add(k key.Key, priority int) {
93 +func (w *Wantlist) Add(k key.Key, priority int) bool {
94 if e, ok := w.set[k]; ok {
95 e.RefCnt++
100 - return
96 + return false
97 }
98
103 - ctx, cancel := context.WithCancel(context.Background())
104 - w.set[k] = Entry{
99 + w.set[k] = &Entry{
100 Key: k,
101 Priority: priority,
107 - Ctx: ctx,
108 - cancel: cancel,
102 RefCnt: 1,
103 }
104 +
105 + return true
106 }
107
113 -func (w *Wantlist) AddEntry(e Entry) {
114 - if _, ok := w.set[e.Key]; ok {
115 - return
108 +func (w *Wantlist) AddEntry(e *Entry) bool {
109 + if ex, ok := w.set[e.Key]; ok {
110 + ex.RefCnt++
111 + return false
112 }
113 w.set[e.Key] = e
114 + return true
115 }
116
120 -func (w *Wantlist) Remove(k key.Key) {
117 +func (w *Wantlist) Remove(k key.Key) bool {
118 e, ok := w.set[k]
119 if !ok {
123 - return
120 + return false
121 }
122
123 e.RefCnt--
124 if e.RefCnt <= 0 {
125 delete(w.set, k)
129 - if e.cancel != nil {
130 - e.cancel()
131 - }
126 + return true
127 }
128 + return false
129 }
130
135 -func (w *Wantlist) Contains(k key.Key) (Entry, bool) {
131 +func (w *Wantlist) Contains(k key.Key) (*Entry, bool) {
132 e, ok := w.set[k]
133 return e, ok
134 }
135
140 -func (w *Wantlist) Entries() []Entry {
136 +func (w *Wantlist) Entries() []*Entry {
137 var es entrySlice
138 for _, e := range w.set {
139 es = append(es, e)
@@ -145,7 +141,7 @@ func (w *Wantlist) Entries() []Entry {
141 return es
142 }
143
148 -func (w *Wantlist) SortedEntries() []Entry {
144 +func (w *Wantlist) SortedEntries() []*Entry {
145 var es entrySlice
146 for _, e := range w.set {
147 es = append(es, e)
exchange/bitswap/wantmanager.go
+13 -13
@@ -75,6 +75,7 @@ func (pm *WantManager) WantBlocks(ctx context.Context, ks []key.Key) {
75 }
76
77 func (pm *WantManager) CancelWants(ks []key.Key) {
78 + log.Infof("cancel wants: %s", ks)
79 pm.addEntries(context.TODO(), ks, true)
80 }
81
@@ -83,16 +84,17 @@ func (pm *WantManager) addEntries(ctx context.Context, ks []key.Key, cancel bool
84 for i, k := range ks {
85 entries = append(entries, &bsmsg.Entry{
86 Cancel: cancel,
86 - Entry: wantlist.Entry{
87 + Entry: &wantlist.Entry{
88 Key: k,
89 Priority: kMaxPriority - i,
89 - Ctx: ctx,
90 + RefCnt: 1,
91 },
92 })
93 }
94 select {
95 case pm.incoming <- entries:
96 case <-pm.ctx.Done():
97 + case <-ctx.Done():
98 }
99 }
100
@@ -241,33 +243,31 @@ func (pm *WantManager) Run() {
243 case entries := <-pm.incoming:
244
245 // add changes to our wantlist
246 + var filtered []*bsmsg.Entry
247 for _, e := range entries {
248 if e.Cancel {
246 - pm.wl.Remove(e.Key)
249 + if pm.wl.Remove(e.Key) {
250 + filtered = append(filtered, e)
251 + }
252 } else {
248 - pm.wl.AddEntry(e.Entry)
253 + if pm.wl.AddEntry(e.Entry) {
254 + filtered = append(filtered, e)
255 + }
256 }
257 }
258
259 // broadcast those wantlist changes
260 for _, p := range pm.peers {
254 - p.addMessage(entries)
261 + p.addMessage(filtered)
262 }
263
264 case <-tock.C:
265 // resend entire wantlist every so often (REALLY SHOULDNT BE NECESSARY)
266 var es []*bsmsg.Entry
267 for _, e := range pm.wl.Entries() {
261 - select {
262 - case <-e.Ctx.Done():
263 - // entry has been cancelled
264 - // simply continue, the entry will be removed from the
265 - // wantlist soon enough
266 - continue
267 - default:
268 - }
268 es = append(es, &bsmsg.Entry{Entry: e})
269 }
270 +
271 for _, p := range pm.peers {
272 p.outlk.Lock()
273 p.out = bsmsg.New(true)
exchange/bitswap/workers.go
+15 -7
@@ -9,7 +9,6 @@ import (
9 context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
10
11 key "github.com/ipfs/go-ipfs/blocks/key"
12 - wantlist "github.com/ipfs/go-ipfs/exchange/bitswap/wantlist"
12 logging "gx/ipfs/QmNQynaz7qfriSUJkiEZUrm2Wen1u3Kj9goZzWtrPyu7XR/go-log"
13 peer "gx/ipfs/QmRBqJF7hb8ZSpRcMwUt8hNhydWcxGEhtk81HKq6oUwKvs/go-libp2p-peer"
14 )
@@ -172,10 +171,19 @@ func (bs *Bitswap) rebroadcastWorker(parent context.Context) {
171 }
172 case <-broadcastSignal.C: // resend unfulfilled wantlist keys
173 log.Event(ctx, "Bitswap.Rebroadcast.active")
174 + entries := bs.wm.wl.Entries()
175 + if len(entries) == 0 {
176 + continue
177 + }
178 + tctx, cancel := context.WithTimeout(ctx, providerRequestTimeout)
179 for _, e := range bs.wm.wl.Entries() {
180 e := e
177 - bs.findKeys <- &e
181 + bs.findKeys <- &blockRequest{
182 + Key: e.Key,
183 + Ctx: tctx,
184 + }
185 }
186 + cancel()
187 case <-parent.Done():
188 return
189 }
@@ -184,20 +192,20 @@ func (bs *Bitswap) rebroadcastWorker(parent context.Context) {
192
193 func (bs *Bitswap) providerQueryManager(ctx context.Context) {
194 var activeLk sync.Mutex
187 - active := make(map[key.Key]*wantlist.Entry)
195 + kset := key.NewKeySet()
196
197 for {
198 select {
199 case e := <-bs.findKeys:
200 activeLk.Lock()
193 - if _, ok := active[e.Key]; ok {
201 + if kset.Has(e.Key) {
202 activeLk.Unlock()
203 continue
204 }
197 - active[e.Key] = e
205 + kset.Add(e.Key)
206 activeLk.Unlock()
207
200 - go func(e *wantlist.Entry) {
208 + go func(e *blockRequest) {
209 child, cancel := context.WithTimeout(e.Ctx, providerRequestTimeout)
210 defer cancel()
211 providers := bs.network.FindProvidersAsync(child, e.Key, maxProvidersPerRequest)
@@ -210,7 +218,7 @@ func (bs *Bitswap) providerQueryManager(ctx context.Context) {
218 }(p)
219 }
220 activeLk.Lock()
213 - delete(active, e.Key)
221 + kset.Remove(e.Key)
222 activeLk.Unlock()
223 }(e)
224
test/sharness/t0220-bitswap.sh
+3 -15
@@ -11,13 +11,6 @@ test_description="test bitswap commands"
11 test_init_ipfs
12 test_launch_ipfs_daemon
13
14 -test_expect_success "'ipfs block get' adds hash to wantlist" '
15 - export NONEXIST=QmeXxaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa &&
16 - test_expect_code 1 ipfs block get $NONEXIST --timeout=10ms &&
17 - ipfs bitswap wantlist >wantlist_out &&
18 - grep $NONEXIST wantlist_out
19 -'
20 -
14 test_expect_success "'ipfs bitswap stat' succeeds" '
15 ipfs bitswap stat >stat_out
16 '
@@ -29,8 +22,7 @@ bitswap status
22 blocks received: 0
23 dup blocks received: 0
24 dup data received: 0 B
32 - wantlist [1 keys]
33 - $NONEXIST
25 + wantlist [0 keys]
26 partners [0]
27 EOF
28 test_cmp expected stat_out
@@ -45,12 +37,8 @@ test_expect_success "'ipfs bitswap wantlist -p' works" '
37 ipfs bitswap wantlist -p "$PEERID" >wantlist_p_out
38 '
39
48 -test_expect_failure "'ipfs bitswap wantlist -p' output looks good" '
49 - test_cmp wantlist_out wantlist_p_out
50 -'
51 -
52 -test_expect_success "'ipfs bitswap unwant' succeeds" '
53 - ipfs bitswap unwant $NONEXIST
40 +test_expect_success "'ipfs bitswap wantlist -p' output looks good" '
41 + test_must_be_empty wantlist_p_out
42 '
43
44 test_expect_success "hash was removed from wantlist" '