WIP: super awesome bitswap cleanup fixtime
Jeromy committed
May 16, 2015 at 12:30 UTC
6bf33ad62fc4d209932fb8b81b4f2dc28ca6123a
9 files changed
+191
-209
exchange/bitswap/bitswap.go
+26
-108
@@ -4,7 +4,6 @@ package bitswap
4
5
import (
6
"errors"
7
- "fmt"
7
"math"
8
"sync"
9
"time"
@@ -23,7 +22,6 @@ import (
22
"github.com/ipfs/go-ipfs/thirdparty/delay"
23
eventlog "github.com/ipfs/go-ipfs/thirdparty/eventlog"
24
u "github.com/ipfs/go-ipfs/util"
26
- pset "github.com/ipfs/go-ipfs/util/peerset" // TODO move this to peerstore
25
)
26
27
var log = eventlog.Logger("bitswap")
@@ -45,9 +43,7 @@ const (
43
provideWorkers = 4
44
)
45
48
-var (
49
- rebroadcastDelay = delay.Fixed(time.Second * 10)
50
-)
46
+var rebroadcastDelay = delay.Fixed(time.Second * 10)
47
48
// New initializes a BitSwap instance that communicates over the provided
49
// BitSwapNetwork. This function registers the returned instance as the network
@@ -86,14 +82,13 @@ func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
82
notifications: notif,
83
engine: decision.NewEngine(ctx, bstore), // TODO close the engine with Close() method
84
network: network,
89
- wantlist: wantlist.NewThreadSafe(),
85
batchRequests: make(chan *blockRequest, sizeBatchRequestChan),
86
process: px,
87
newBlocks: make(chan *blocks.Block, HasBlockBufferSize),
88
provideKeys: make(chan u.Key),
94
- pm: NewPeerManager(network),
89
+ wm: NewWantManager(network),
90
}
96
- go bs.pm.Run(ctx)
91
+ go bs.wm.Run(ctx)
92
network.SetDelegate(bs)
93
94
// Start up bitswaps async worker routines
@@ -112,7 +107,7 @@ type Bitswap struct {
107
108
// the peermanager manages sending messages to peers in a way that
109
// wont block bitswap operation
115
- pm *PeerManager
110
+ wm *WantManager
111
112
// blockstore is the local database
113
// NB: ensure threadsafety
@@ -127,8 +122,6 @@ type Bitswap struct {
122
123
engine *decision.Engine
124
130
- wantlist *wantlist.ThreadSafe
131
-
125
process process.Process
126
127
newBlocks chan *blocks.Block
@@ -233,60 +226,21 @@ func (bs *Bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
226
return err
227
}
228
236
- bs.wantlist.Remove(blk.Key())
229
bs.notifications.Publish(blk)
230
select {
231
case bs.newBlocks <- blk:
232
+ // send block off to be reprovided
233
case <-ctx.Done():
234
return ctx.Err()
235
}
236
return nil
237
}
238
246
-func (bs *Bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMessage, peers <-chan peer.ID) error {
247
- set := pset.New()
248
-
249
-loop:
250
- for {
251
- select {
252
- case peerToQuery, ok := <-peers:
253
- if !ok {
254
- break loop
255
- }
256
-
257
- if !set.TryAdd(peerToQuery) { //Do once per peer
258
- continue
259
- }
260
-
261
- bs.pm.Send(peerToQuery, m)
262
- case <-ctx.Done():
263
- return nil
264
- }
265
- }
266
- return nil
267
-}
268
-
269
-func (bs *Bitswap) sendWantlistToPeers(ctx context.Context, peers <-chan peer.ID) error {
270
- entries := bs.wantlist.Entries()
271
- if len(entries) == 0 {
272
- return nil
273
- }
274
- message := bsmsg.New()
275
- message.SetFull(true)
276
- for _, wanted := range entries {
277
- message.AddEntry(wanted.Key, wanted.Priority)
278
- }
279
- return bs.sendWantlistMsgToPeers(ctx, message, peers)
280
-}
281
-
282
-func (bs *Bitswap) sendWantlistToProviders(ctx context.Context, entries []wantlist.Entry) {
239
+func (bs *Bitswap) connectToProviders(ctx context.Context, entries []wantlist.Entry) {
240
241
ctx, cancel := context.WithCancel(ctx)
242
defer cancel()
243
287
- // prepare a channel to hand off to sendWantlistToPeers
288
- sendToPeers := make(chan peer.ID)
289
-
244
// Get providers for all entries in wantlist (could take a while)
245
wg := sync.WaitGroup{}
246
for _, e := range entries {
@@ -298,97 +252,61 @@ func (bs *Bitswap) sendWantlistToProviders(ctx context.Context, entries []wantli
252
defer cancel()
253
providers := bs.network.FindProvidersAsync(child, k, maxProvidersPerRequest)
254
for prov := range providers {
301
- sendToPeers <- prov
255
+ go func(p peer.ID) {
256
+ bs.network.ConnectTo(ctx, p)
257
+ }(prov)
258
}
259
}(e.Key)
260
}
261
306
- go func() {
307
- wg.Wait() // make sure all our children do finish.
308
- close(sendToPeers)
309
- }()
310
-
311
- err := bs.sendWantlistToPeers(ctx, sendToPeers)
312
- if err != nil {
313
- log.Debugf("sendWantlistToPeers error: %s", err)
314
- }
262
+ wg.Wait() // make sure all our children do finish.
263
}
264
317
-// TODO(brian): handle errors
318
-func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) error {
265
+func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) {
266
// This call records changes to wantlists, blocks received,
267
// and number of bytes transfered.
268
bs.engine.MessageReceived(p, incoming)
269
// TODO: this is bad, and could be easily abused.
270
// Should only track *useful* messages in ledger
271
272
+ if len(incoming.Blocks()) == 0 {
273
+ return
274
+ }
275
+
276
+ // quickly send out cancels, reduces chances of duplicate block receives
277
var keys []u.Key
278
+ for _, block := range incoming.Blocks() {
279
+ keys = append(keys, block.Key())
280
+ }
281
+ bs.wm.CancelWants(keys)
282
+
283
for _, block := range incoming.Blocks() {
284
bs.blocksRecvd++
285
if has, err := bs.blockstore.Has(block.Key()); err == nil && has {
286
bs.dupBlocksRecvd++
287
}
288
log.Debugf("got block %s from %s", block, p)
289
+
290
hasBlockCtx, cancel := context.WithTimeout(ctx, hasBlockTimeout)
291
if err := bs.HasBlock(hasBlockCtx, block); err != nil {
334
- return fmt.Errorf("ReceiveMessage HasBlock error: %s", err)
292
+ log.Warningf("ReceiveMessage HasBlock error: %s", err)
293
}
294
cancel()
337
- keys = append(keys, block.Key())
295
}
339
-
340
- bs.cancelBlocks(ctx, keys)
341
- return nil
296
}
297
298
// Connected/Disconnected warns bitswap about peer connections
299
func (bs *Bitswap) PeerConnected(p peer.ID) {
300
// TODO: add to clientWorker??
347
- bs.pm.Connected(p)
348
- peers := make(chan peer.ID, 1)
349
- peers <- p
350
- close(peers)
351
- err := bs.sendWantlistToPeers(context.TODO(), peers)
352
- if err != nil {
353
- log.Debugf("error sending wantlist: %s", err)
354
- }
301
+ bs.wm.Connected(p)
302
}
303
304
// Connected/Disconnected warns bitswap about peer connections
305
func (bs *Bitswap) PeerDisconnected(p peer.ID) {
359
- bs.pm.Disconnected(p)
306
+ bs.wm.Disconnected(p)
307
bs.engine.PeerDisconnected(p)
308
}
309
363
-func (bs *Bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) {
364
- if len(bkeys) < 1 {
365
- return
366
- }
367
- message := bsmsg.New()
368
- message.SetFull(false)
369
- for _, k := range bkeys {
370
- log.Debug("cancel block: %s", k)
371
- message.Cancel(k)
372
- }
373
-
374
- bs.pm.Broadcast(message)
375
- return
376
-}
377
-
378
-func (bs *Bitswap) wantNewBlocks(ctx context.Context, bkeys []u.Key) {
379
- if len(bkeys) < 1 {
380
- return
381
- }
382
-
383
- message := bsmsg.New()
384
- message.SetFull(false)
385
- for i, k := range bkeys {
386
- message.AddEntry(k, kMaxPriority-i)
387
- }
388
-
389
- bs.pm.Broadcast(message)
390
-}
391
-
310
func (bs *Bitswap) ReceiveError(err error) {
311
log.Debugf("Bitswap ReceiveError: %s", err)
312
// TODO log the network error
@@ -401,7 +319,7 @@ func (bs *Bitswap) Close() error {
319
320
func (bs *Bitswap) GetWantlist() []u.Key {
321
var out []u.Key
404
- for _, e := range bs.wantlist.Entries() {
322
+ for _, e := range bs.wm.wl.Entries() {
323
out = append(out, e.Key)
324
}
325
return out
exchange/bitswap/bitswap_test.go
+11
-3
@@ -120,6 +120,16 @@ func TestLargeFile(t *testing.T) {
120
PerformDistributionTest(t, numInstances, numBlocks)
121
}
122
123
+func TestLargeFileTwoPeers(t *testing.T) {
124
+ if testing.Short() {
125
+ t.SkipNow()
126
+ }
127
+ t.Parallel()
128
+ numInstances := 2
129
+ numBlocks := 100
130
+ PerformDistributionTest(t, numInstances, numBlocks)
131
+}
132
+
133
func PerformDistributionTest(t *testing.T, numInstances, numBlocks int) {
134
if testing.Short() {
135
t.SkipNow()
@@ -129,8 +139,6 @@ func PerformDistributionTest(t *testing.T, numInstances, numBlocks int) {
139
defer sg.Close()
140
bg := blocksutil.NewBlockGenerator()
141
132
- t.Log("Test a few nodes trying to get one file with a lot of blocks")
133
-
142
instances := sg.Instances(numInstances)
143
blocks := bg.Blocks(numBlocks)
144
@@ -238,7 +246,7 @@ func TestBasicBitswap(t *testing.T) {
246
defer sg.Close()
247
bg := blocksutil.NewBlockGenerator()
248
241
- t.Log("Test a few nodes trying to get one file with a lot of blocks")
249
+ t.Log("Test a one node trying to get one block from another")
250
251
instances := sg.Instances(2)
252
blocks := bg.Blocks(1)
exchange/bitswap/decision/engine.go
+12
-4
@@ -92,7 +92,7 @@ func NewEngine(ctx context.Context, bs bstore.Blockstore) *Engine {
92
bs: bs,
93
peerRequestQueue: newPRQ(),
94
outbox: make(chan (<-chan *Envelope), outboxChanBuffer),
95
- workSignal: make(chan struct{}),
95
+ workSignal: make(chan struct{}, 1),
96
}
97
go e.taskWorker(ctx)
98
return e
@@ -156,7 +156,15 @@ func (e *Engine) nextEnvelope(ctx context.Context) (*Envelope, error) {
156
return &Envelope{
157
Peer: nextTask.Target,
158
Block: block,
159
- Sent: nextTask.Done,
159
+ Sent: func() {
160
+ nextTask.Done()
161
+ select {
162
+ case e.workSignal <- struct{}{}:
163
+ // work completing may mean that our queue will provide new
164
+ // work to be done.
165
+ default:
166
+ }
167
+ },
168
}, nil
169
}
170
}
@@ -202,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 {
205
- log.Debugf("cancel %s", entry.Key)
213
+ log.Errorf("cancel %s", entry.Key)
214
l.CancelWant(entry.Key)
215
e.peerRequestQueue.Remove(entry.Key, p)
216
} else {
209
- log.Debugf("wants %s - %d", entry.Key, entry.Priority)
217
+ log.Errorf("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/decision/peer_request_queue.go
+12
-6
@@ -51,12 +51,6 @@ func (tl *prq) Push(entry wantlist.Entry, to peer.ID) {
51
tl.partners[to] = partner
52
}
53
54
- if task, ok := tl.taskMap[taskKey(to, entry.Key)]; ok {
55
- task.Entry.Priority = entry.Priority
56
- partner.taskQueue.Update(task.index)
57
- return
58
- }
59
-
54
partner.activelk.Lock()
55
defer partner.activelk.Unlock()
56
_, ok = partner.activeBlocks[entry.Key]
@@ -64,6 +58,12 @@ func (tl *prq) Push(entry wantlist.Entry, to peer.ID) {
58
return
59
}
60
61
+ if task, ok := tl.taskMap[taskKey(to, entry.Key)]; ok {
62
+ task.Entry.Priority = entry.Priority
63
+ partner.taskQueue.Update(task.index)
64
+ return
65
+ }
66
+
67
task := &peerRequestTask{
68
Entry: entry,
69
Target: to,
@@ -220,6 +220,12 @@ func partnerCompare(a, b pq.Elem) bool {
220
if pb.requests == 0 {
221
return true
222
}
223
+ if pa.active == pb.active {
224
+ // sorting by taskQueue.Len() aids in cleaning out trash entries faster
225
+ // if we sorted instead by requests, one peer could potentially build up
226
+ // a huge number of cancelled entries in the queue resulting in a memory leak
227
+ return pa.taskQueue.Len() > pb.taskQueue.Len()
228
+ }
229
return pa.active < pb.active
230
}
231
exchange/bitswap/network/interface.go
+1
-1
@@ -33,7 +33,7 @@ type Receiver interface {
33
ReceiveMessage(
34
ctx context.Context,
35
sender peer.ID,
36
- incoming bsmsg.BitSwapMessage) error
36
+ incoming bsmsg.BitSwapMessage)
37
38
ReceiveError(error)
39
exchange/bitswap/peermanager.go
+102
-50
@@ -7,28 +7,36 @@ import (
7
engine "github.com/ipfs/go-ipfs/exchange/bitswap/decision"
8
bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message"
9
bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
10
+ wantlist "github.com/ipfs/go-ipfs/exchange/bitswap/wantlist"
11
peer "github.com/ipfs/go-ipfs/p2p/peer"
12
u "github.com/ipfs/go-ipfs/util"
13
)
14
14
-type PeerManager struct {
15
+type WantManager struct {
16
receiver bsnet.Receiver
17
17
- incoming chan *msgPair
18
- connect chan peer.ID
18
+ incoming chan []*bsmsg.Entry
19
+
20
+ // notification channel for new peers connecting
21
+ connect chan peer.ID
22
+
23
+ // notification channel for peers disconnecting
24
disconnect chan peer.ID
25
26
peers map[peer.ID]*msgQueue
27
28
+ wl *wantlist.Wantlist
29
+
30
network bsnet.BitSwapNetwork
31
}
32
26
-func NewPeerManager(network bsnet.BitSwapNetwork) *PeerManager {
27
- return &PeerManager{
28
- incoming: make(chan *msgPair, 10),
33
+func NewWantManager(network bsnet.BitSwapNetwork) *WantManager {
34
+ return &WantManager{
35
+ incoming: make(chan []*bsmsg.Entry, 10),
36
connect: make(chan peer.ID, 10),
37
disconnect: make(chan peer.ID, 10),
38
peers: make(map[peer.ID]*msgQueue),
39
+ wl: wantlist.New(),
40
network: network,
41
}
42
}
@@ -53,37 +61,68 @@ type msgQueue struct {
61
done chan struct{}
62
}
63
56
-func (pm *PeerManager) SendBlock(ctx context.Context, env *engine.Envelope) {
64
+func (pm *WantManager) WantBlocks(ks []u.Key) {
65
+ log.Error("WANT: ", ks)
66
+ pm.addEntries(ks, false)
67
+}
68
+
69
+func (pm *WantManager) CancelWants(ks []u.Key) {
70
+ log.Error("CANCEL: ", ks)
71
+ pm.addEntries(ks, true)
72
+}
73
+
74
+func (pm *WantManager) addEntries(ks []u.Key, cancel bool) {
75
+ var entries []*bsmsg.Entry
76
+ for i, k := range ks {
77
+ entries = append(entries, &bsmsg.Entry{
78
+ Cancel: cancel,
79
+ Entry: wantlist.Entry{
80
+ Key: k,
81
+ Priority: kMaxPriority - i,
82
+ },
83
+ })
84
+ }
85
+ pm.incoming <- entries
86
+}
87
+
88
+func (pm *WantManager) SendBlock(ctx context.Context, env *engine.Envelope) {
89
// Blocks need to be sent synchronously to maintain proper backpressure
90
// throughout the network stack
91
defer env.Sent()
92
93
msg := bsmsg.New()
94
msg.AddBlock(env.Block)
95
+ msg.SetFull(false)
96
err := pm.network.SendMessage(ctx, env.Peer, msg)
97
if err != nil {
98
log.Error(err)
99
}
100
}
101
69
-func (pm *PeerManager) startPeerHandler(ctx context.Context, p peer.ID) *msgQueue {
102
+func (pm *WantManager) startPeerHandler(ctx context.Context, p peer.ID) *msgQueue {
103
_, ok := pm.peers[p]
104
if ok {
105
// TODO: log an error?
106
return nil
107
}
108
76
- mq := new(msgQueue)
77
- mq.done = make(chan struct{})
78
- mq.work = make(chan struct{}, 1)
79
- mq.p = p
109
+ mq := newMsgQueue(p)
110
+
111
+ // new peer, we will want to give them our full wantlist
112
+ fullwantlist := bsmsg.New()
113
+ for _, e := range pm.wl.Entries() {
114
+ fullwantlist.AddEntry(e.Key, e.Priority)
115
+ }
116
+ fullwantlist.SetFull(true)
117
+ mq.out = fullwantlist
118
+ mq.work <- struct{}{}
119
120
pm.peers[p] = mq
121
go pm.runQueue(ctx, mq)
122
return mq
123
}
124
86
-func (pm *PeerManager) stopPeerHandler(p peer.ID) {
125
+func (pm *WantManager) stopPeerHandler(p peer.ID) {
126
pq, ok := pm.peers[p]
127
if !ok {
128
// TODO: log error?
@@ -94,32 +133,38 @@ func (pm *PeerManager) stopPeerHandler(p peer.ID) {
133
delete(pm.peers, p)
134
}
135
97
-func (pm *PeerManager) runQueue(ctx context.Context, mq *msgQueue) {
136
+func (pm *WantManager) runQueue(ctx context.Context, mq *msgQueue) {
137
for {
138
select {
139
case <-mq.work: // there is work to be done
140
102
- // TODO: this might not need to be done every time, figure out
103
- // a good heuristic
141
err := pm.network.ConnectTo(ctx, mq.p)
142
if err != nil {
143
log.Error(err)
144
// TODO: cant connect, what now?
145
}
146
110
- // grab outgoin message
147
+ // grab outgoing message
148
mq.outlk.Lock()
149
wlm := mq.out
150
mq.out = nil
151
mq.outlk.Unlock()
152
116
- if wlm != nil && !wlm.Empty() {
117
- // send wantlist updates
118
- err = pm.network.SendMessage(ctx, mq.p, wlm)
119
- if err != nil {
120
- log.Error("bitswap send error: ", err)
121
- // TODO: what do we do if this fails?
122
- }
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")
160
+ continue
161
+ }
162
+
163
+ // send wantlist updates
164
+ err = pm.network.SendMessage(ctx, mq.p, wlm)
165
+ if err != nil {
166
+ log.Error("bitswap send error: ", err)
167
+ // TODO: what do we do if this fails?
168
}
169
case <-mq.done:
170
return
@@ -127,46 +172,38 @@ func (pm *PeerManager) runQueue(ctx context.Context, mq *msgQueue) {
172
}
173
}
174
130
-func (pm *PeerManager) Send(to peer.ID, msg bsmsg.BitSwapMessage) {
131
- if len(msg.Blocks()) > 0 {
132
- panic("no blocks here!")
133
- }
134
- pm.incoming <- &msgPair{to: to, msg: msg}
135
-}
136
-
137
-func (pm *PeerManager) Broadcast(msg bsmsg.BitSwapMessage) {
138
- pm.incoming <- &msgPair{msg: msg}
139
-}
140
-
141
-func (pm *PeerManager) Connected(p peer.ID) {
175
+func (pm *WantManager) Connected(p peer.ID) {
176
pm.connect <- p
177
}
178
145
-func (pm *PeerManager) Disconnected(p peer.ID) {
179
+func (pm *WantManager) Disconnected(p peer.ID) {
180
pm.disconnect <- p
181
}
182
183
// TODO: use goprocess here once i trust it
150
-func (pm *PeerManager) Run(ctx context.Context) {
184
+func (pm *WantManager) Run(ctx context.Context) {
185
for {
186
select {
153
- case msgp := <-pm.incoming:
154
-
155
- // Broadcast message to all if recipient not set
156
- if msgp.to == "" {
157
- for _, p := range pm.peers {
158
- p.addMessage(msgp.msg)
187
+ case entries := <-pm.incoming:
188
+
189
+ msg := bsmsg.New()
190
+ msg.SetFull(false)
191
+ // add changes to our wantlist
192
+ for _, e := range entries {
193
+ if e.Cancel {
194
+ pm.wl.Remove(e.Key)
195
+ msg.Cancel(e.Key)
196
+ } else {
197
+ pm.wl.Add(e.Key, e.Priority)
198
+ msg.AddEntry(e.Key, e.Priority)
199
}
160
- continue
200
}
201
163
- p, ok := pm.peers[msgp.to]
164
- if !ok {
165
- //TODO: decide, drop message? or dial?
166
- p = pm.startPeerHandler(ctx, msgp.to)
202
+ // broadcast those wantlist changes
203
+ for _, p := range pm.peers {
204
+ p.addMessage(msg)
205
}
206
169
- p.addMessage(msgp.msg)
207
case p := <-pm.connect:
208
pm.startPeerHandler(ctx, p)
209
case p := <-pm.disconnect:
@@ -177,6 +214,15 @@ func (pm *PeerManager) Run(ctx context.Context) {
214
}
215
}
216
217
+func newMsgQueue(p peer.ID) *msgQueue {
218
+ mq := new(msgQueue)
219
+ mq.done = make(chan struct{})
220
+ mq.work = make(chan struct{}, 1)
221
+ mq.p = p
222
+
223
+ return mq
224
+}
225
+
226
func (mq *msgQueue) addMessage(msg bsmsg.BitSwapMessage) {
227
mq.outlk.Lock()
228
defer func() {
@@ -187,6 +233,10 @@ func (mq *msgQueue) addMessage(msg bsmsg.BitSwapMessage) {
233
}
234
}()
235
236
+ if msg.Full() {
237
+ log.Error("GOt FULL MESSAGE")
238
+ }
239
+
240
// if we have no message held, or the one we are given is full
241
// overwrite the one we are holding
242
if mq.out == nil || msg.Full() {
@@ -199,8 +249,10 @@ func (mq *msgQueue) addMessage(msg bsmsg.BitSwapMessage) {
249
// one passed in
250
for _, e := range msg.Wantlist() {
251
if e.Cancel {
252
+ log.Error("add message cancel: ", e.Key, mq.p)
253
mq.out.Cancel(e.Key)
254
} else {
255
+ log.Error("add message want: ", e.Key, mq.p)
256
mq.out.AddEntry(e.Key, e.Priority)
257
}
258
}
exchange/bitswap/testnet/network_test.go
+6
-10
@@ -29,19 +29,17 @@ func TestSendMessageAsyncButWaitForResponse(t *testing.T) {
29
responder.SetDelegate(lambda(func(
30
ctx context.Context,
31
fromWaiter peer.ID,
32
- msgFromWaiter bsmsg.BitSwapMessage) error {
32
+ msgFromWaiter bsmsg.BitSwapMessage) {
33
34
msgToWaiter := bsmsg.New()
35
msgToWaiter.AddBlock(blocks.NewBlock([]byte(expectedStr)))
36
waiter.SendMessage(ctx, fromWaiter, msgToWaiter)
37
-
38
- return nil
37
}))
38
39
waiter.SetDelegate(lambda(func(
40
ctx context.Context,
41
fromResponder peer.ID,
44
- msgFromResponder bsmsg.BitSwapMessage) error {
42
+ msgFromResponder bsmsg.BitSwapMessage) {
43
44
// TODO assert that this came from the correct peer and that the message contents are as expected
45
ok := false
@@ -54,9 +52,7 @@ func TestSendMessageAsyncButWaitForResponse(t *testing.T) {
52
53
if !ok {
54
t.Fatal("Message not received from the responder")
57
-
55
}
59
- return nil
56
}))
57
58
messageSentAsync := bsmsg.New()
@@ -71,7 +67,7 @@ func TestSendMessageAsyncButWaitForResponse(t *testing.T) {
67
}
68
69
type receiverFunc func(ctx context.Context, p peer.ID,
74
- incoming bsmsg.BitSwapMessage) error
70
+ incoming bsmsg.BitSwapMessage)
71
72
// lambda returns a Receiver instance given a receiver function
73
func lambda(f receiverFunc) bsnet.Receiver {
@@ -81,12 +77,12 @@ func lambda(f receiverFunc) bsnet.Receiver {
77
}
78
79
type lambdaImpl struct {
84
- f func(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) error
80
+ f func(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage)
81
}
82
83
func (lam *lambdaImpl) ReceiveMessage(ctx context.Context,
88
- p peer.ID, incoming bsmsg.BitSwapMessage) error {
89
- return lam.f(ctx, p, incoming)
84
+ p peer.ID, incoming bsmsg.BitSwapMessage) {
85
+ lam.f(ctx, p, incoming)
86
}
87
88
func (lam *lambdaImpl) ReceiveError(err error) {
exchange/bitswap/testnet/virtual.go
+2
-1
@@ -72,7 +72,8 @@ func (n *network) deliver(
72
73
n.delay.Wait()
74
75
- return r.ReceiveMessage(context.TODO(), from, message)
75
+ r.ReceiveMessage(context.TODO(), from, message)
76
+ return nil
77
}
78
79
type networkClient struct {
exchange/bitswap/workers.go
+19
-26
@@ -42,9 +42,11 @@ func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
42
}
43
44
// Start up a worker to manage periodically resending our wantlist out to peers
45
- px.Go(func(px process.Process) {
46
- bs.rebroadcastWorker(ctx)
47
- })
45
+ /*
46
+ px.Go(func(px process.Process) {
47
+ bs.rebroadcastWorker(ctx)
48
+ })
49
+ */
50
51
// Start up a worker to manage sending out provides messages
52
px.Go(func(px process.Process) {
@@ -72,7 +74,7 @@ func (bs *Bitswap) taskWorker(ctx context.Context) {
74
continue
75
}
76
75
- bs.pm.SendBlock(ctx, envelope)
77
+ bs.wm.SendBlock(ctx, envelope)
78
case <-ctx.Done():
79
return
80
}
@@ -146,30 +148,19 @@ func (bs *Bitswap) clientWorker(parent context.Context) {
148
log.Warning("Received batch request for zero blocks")
149
continue
150
}
149
- for i, k := range keys {
150
- bs.wantlist.Add(k, kMaxPriority-i)
151
- }
151
153
- done := make(chan struct{})
154
- go func() {
155
- bs.wantNewBlocks(req.ctx, keys)
156
- close(done)
157
- }()
152
+ bs.wm.WantBlocks(keys)
153
154
// NB: Optimization. Assumes that providers of key[0] are likely to
155
// be able to provide for all keys. This currently holds true in most
156
// every situation. Later, this assumption may not hold as true.
157
child, cancel := context.WithTimeout(req.ctx, providerRequestTimeout)
158
providers := bs.network.FindProvidersAsync(child, keys[0], maxProvidersPerRequest)
164
- err := bs.sendWantlistToPeers(req.ctx, providers)
165
- if err != nil {
166
- log.Debugf("error sending wantlist: %s", err)
159
+ for p := range providers {
160
+ go bs.network.ConnectTo(req.ctx, p)
161
}
162
cancel()
163
170
- // Wait for wantNewBlocks to finish
171
- <-done
172
-
164
case <-parent.Done():
165
return
166
}
@@ -180,22 +171,24 @@ func (bs *Bitswap) rebroadcastWorker(parent context.Context) {
171
ctx, cancel := context.WithCancel(parent)
172
defer cancel()
173
183
- broadcastSignal := time.After(rebroadcastDelay.Get())
184
- tick := time.Tick(10 * time.Second)
174
+ broadcastSignal := time.NewTicker(rebroadcastDelay.Get())
175
+ defer broadcastSignal.Stop()
176
+
177
+ tick := time.NewTicker(10 * time.Second)
178
+ defer tick.Stop()
179
180
for {
181
select {
188
- case <-tick:
189
- n := bs.wantlist.Len()
182
+ case <-tick.C:
183
+ n := bs.wm.wl.Len()
184
if n > 0 {
185
log.Debug(n, "keys in bitswap wantlist")
186
}
193
- case <-broadcastSignal: // resend unfulfilled wantlist keys
194
- entries := bs.wantlist.Entries()
187
+ case <-broadcastSignal.C: // resend unfulfilled wantlist keys
188
+ entries := bs.wm.wl.Entries()
189
if len(entries) > 0 {
196
- bs.sendWantlistToProviders(ctx, entries)
190
+ bs.connectToProviders(ctx, entries)
191
}
198
- broadcastSignal = time.After(rebroadcastDelay.Get())
192
case <-parent.Done():
193
return
194
}