implement bitswap sessions
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Apr 3, 2017 at 19:21 UTC
bda8c3a6873e35eb674ad17633f35e6720931f94
11 files changed
+528
-57
blocks/blocksutil/block_generator.go
+2
-2
@@ -25,8 +25,8 @@ func (bg *BlockGenerator) Next() *blocks.BasicBlock {
25
}
26
27
// Blocks generates as many BasicBlocks as specified by n.
28
-func (bg *BlockGenerator) Blocks(n int) []*blocks.BasicBlock {
29
- blocks := make([]*blocks.BasicBlock, 0)
28
+func (bg *BlockGenerator) Blocks(n int) []blocks.Block {
29
+ blocks := make([]blocks.Block, 0, n)
30
for i := 0; i < n; i++ {
31
b := bg.Next()
32
blocks = append(blocks, b)
core/corehttp/gateway_handler.go
+1
-1
@@ -27,7 +27,7 @@ import (
27
node "gx/ipfs/QmPAKbSsgEX5B6fpmxa61jXYnoWzZr5sNafd3qgPiSH8Uv/go-ipld-format"
28
humanize "gx/ipfs/QmPSBJL4momYnE7DcUyk2DVhD6rH488ZmHBGLbxNdhU44K/go-humanize"
29
cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
30
- multibase "gx/ipfs/QmcxkxTVuURV2Ptse8TvkqH5BQDwV62X1x19JqqvbBzwUM/go-multibase"
30
+ multibase "gx/ipfs/Qme4T6BE4sQxg7ZouamF5M7Tx1ZFTqzcns7BkyQPXpoT99/go-multibase"
31
)
32
33
const (
exchange/bitswap/bitswap.go
+30
-43
@@ -7,6 +7,7 @@ import (
7
"errors"
8
"math"
9
"sync"
10
+ "sync/atomic"
11
"time"
12
13
blockstore "github.com/ipfs/go-ipfs/blocks/blockstore"
@@ -17,13 +18,13 @@ import (
18
notifications "github.com/ipfs/go-ipfs/exchange/bitswap/notifications"
19
flags "github.com/ipfs/go-ipfs/flags"
20
"github.com/ipfs/go-ipfs/thirdparty/delay"
20
- blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
21
22
metrics "gx/ipfs/QmRg1gKTHzc3CZXSKzem8aR4E3TubFhbgXwfVuWnSK5CC5/go-metrics-interface"
23
process "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess"
24
procctx "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess/context"
25
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
26
loggables "gx/ipfs/QmVesPmqbPp7xRGyY96tnBwzDtVV1nqv4SCVxo5zCqKyH8/go-libp2p-loggables"
27
+ blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
28
cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
29
peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
30
)
@@ -159,10 +160,15 @@ type Bitswap struct {
160
blocksSent int
161
dataSent uint64
162
dataRecvd uint64
163
+ messagesRecvd uint64
164
165
// Metrics interface metrics
166
dupMetric metrics.Histogram
167
allMetric metrics.Histogram
168
+
169
+ // Sessions
170
+ sessions []*Session
171
+ sessLk sync.Mutex
172
}
173
174
type blockRequest struct {
@@ -173,45 +179,7 @@ type blockRequest struct {
179
// GetBlock attempts to retrieve a particular block from peers within the
180
// deadline enforced by the context.
181
func (bs *Bitswap) GetBlock(parent context.Context, k *cid.Cid) (blocks.Block, error) {
176
- if k == nil {
177
- log.Error("nil cid in GetBlock")
178
- return nil, blockstore.ErrNotFound
179
- }
180
-
181
- // Any async work initiated by this function must end when this function
182
- // returns. To ensure this, derive a new context. Note that it is okay to
183
- // listen on parent in this scope, but NOT okay to pass |parent| to
184
- // functions called by this one. Otherwise those functions won't return
185
- // when this context's cancel func is executed. This is difficult to
186
- // enforce. May this comment keep you safe.
187
- ctx, cancelFunc := context.WithCancel(parent)
188
-
189
- // TODO: this request ID should come in from a higher layer so we can track
190
- // across multiple 'GetBlock' invocations
191
- ctx = logging.ContextWithLoggable(ctx, loggables.Uuid("GetBlockRequest"))
192
- log.Event(ctx, "Bitswap.GetBlockRequest.Start", k)
193
- defer log.Event(ctx, "Bitswap.GetBlockRequest.End", k)
194
- defer cancelFunc()
195
-
196
- promise, err := bs.GetBlocks(ctx, []*cid.Cid{k})
197
- if err != nil {
198
- return nil, err
199
- }
200
-
201
- select {
202
- case block, ok := <-promise:
203
- if !ok {
204
- select {
205
- case <-ctx.Done():
206
- return nil, ctx.Err()
207
- default:
208
- return nil, errors.New("promise channel was closed")
209
- }
210
- }
211
- return block, nil
212
- case <-parent.Done():
213
- return nil, parent.Err()
214
- }
182
+ return getBlock(parent, k, bs.GetBlocks)
183
}
184
185
func (bs *Bitswap) WantlistForPeer(p peer.ID) []*cid.Cid {
@@ -251,7 +219,7 @@ func (bs *Bitswap) GetBlocks(ctx context.Context, keys []*cid.Cid) (<-chan block
219
log.Event(ctx, "Bitswap.GetBlockRequest.Start", k)
220
}
221
254
- bs.wm.WantBlocks(ctx, keys)
222
+ bs.wm.WantBlocks(ctx, keys, nil)
223
224
// NB: Optimization. Assumes that providers of key[0] are likely to
225
// be able to provide for all keys. This currently holds true in most
@@ -304,7 +272,7 @@ func (bs *Bitswap) GetBlocks(ctx context.Context, keys []*cid.Cid) (<-chan block
272
273
// CancelWant removes a given key from the wantlist
274
func (bs *Bitswap) CancelWants(cids []*cid.Cid) {
307
- bs.wm.CancelWants(cids)
275
+ bs.wm.CancelWants(context.Background(), cids, nil)
276
}
277
278
// HasBlock announces the existance of a block to this bitswap service. The
@@ -340,7 +308,22 @@ func (bs *Bitswap) HasBlock(blk blocks.Block) error {
308
return nil
309
}
310
311
+func (bs *Bitswap) SessionsForBlock(c *cid.Cid) []*Session {
312
+ bs.sessLk.Lock()
313
+ defer bs.sessLk.Unlock()
314
+
315
+ var out []*Session
316
+ for _, s := range bs.sessions {
317
+ if s.InterestedIn(c) {
318
+ out = append(out, s)
319
+ }
320
+ }
321
+ return out
322
+}
323
+
324
func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) {
325
+ atomic.AddUint64(&bs.messagesRecvd, 1)
326
+
327
// This call records changes to wantlists, blocks received,
328
// and number of bytes transfered.
329
bs.engine.MessageReceived(p, incoming)
@@ -362,7 +345,8 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
345
}
346
keys = append(keys, block.Cid())
347
}
365
- bs.wm.CancelWants(keys)
348
+
349
+ bs.wm.CancelWants(context.Background(), keys, nil)
350
351
wg := sync.WaitGroup{}
352
for _, block := range iblocks {
@@ -375,6 +359,9 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
359
k := b.Cid()
360
log.Event(ctx, "Bitswap.GetBlockRequest.End", k)
361
362
+ for _, ses := range bs.SessionsForBlock(k) {
363
+ ses.ReceiveBlock(p, b)
364
+ }
365
log.Debugf("got block %s from %s", b, p)
366
if err := bs.HasBlock(b); err != nil {
367
log.Warningf("ReceiveMessage HasBlock error: %s", err)
exchange/bitswap/bitswap_test.go
+12
-1
@@ -370,6 +370,9 @@ func TestDoubleGet(t *testing.T) {
370
instances := sg.Instances(2)
371
blocks := bg.Blocks(1)
372
373
+ // NOTE: A race condition can happen here where these GetBlocks requests go
374
+ // through before the peers even get connected. This is okay, bitswap
375
+ // *should* be able to handle this.
376
ctx1, cancel1 := context.WithCancel(context.Background())
377
blkch1, err := instances[1].Exchange.GetBlocks(ctx1, []*cid.Cid{blocks[0].Cid()})
378
if err != nil {
@@ -385,7 +388,7 @@ func TestDoubleGet(t *testing.T) {
388
}
389
390
// ensure both requests make it into the wantlist at the same time
388
- time.Sleep(time.Millisecond * 100)
391
+ time.Sleep(time.Millisecond * 20)
392
cancel1()
393
394
_, ok := <-blkch1
@@ -405,6 +408,14 @@ func TestDoubleGet(t *testing.T) {
408
}
409
t.Log(blk)
410
case <-time.After(time.Second * 5):
411
+ p1wl := instances[0].Exchange.WantlistForPeer(instances[1].Peer)
412
+ if len(p1wl) != 1 {
413
+ t.Logf("wantlist view didnt have 1 item (had %d)", len(p1wl))
414
+ } else if !p1wl[0].Equals(blocks[0].Cid()) {
415
+ t.Logf("had 1 item, it was wrong: %s %s", blocks[0].Cid(), p1wl[0])
416
+ } else {
417
+ t.Log("had correct wantlist, somehow")
418
+ }
419
t.Fatal("timed out waiting on block")
420
}
421
exchange/bitswap/decision/engine.go
+1
-1
@@ -2,10 +2,10 @@
2
package decision
3
4
import (
5
+ "context"
6
"sync"
7
"time"
8
8
- context "context"
9
bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
10
bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message"
11
wl "github.com/ipfs/go-ipfs/exchange/bitswap/wantlist"
exchange/bitswap/get.go
new
+100
@@ -0,0 +1,100 @@
1
+package bitswap
2
+
3
+import (
4
+ "context"
5
+ "errors"
6
+
7
+ blocks "github.com/ipfs/go-ipfs/blocks"
8
+ blockstore "github.com/ipfs/go-ipfs/blocks/blockstore"
9
+ notifications "github.com/ipfs/go-ipfs/exchange/bitswap/notifications"
10
+
11
+ cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
12
+)
13
+
14
+type getBlocksFunc func(context.Context, []*cid.Cid) (<-chan blocks.Block, error)
15
+
16
+func getBlock(p context.Context, k *cid.Cid, gb getBlocksFunc) (blocks.Block, error) {
17
+ if k == nil {
18
+ log.Error("nil cid in GetBlock")
19
+ return nil, blockstore.ErrNotFound
20
+ }
21
+
22
+ // Any async work initiated by this function must end when this function
23
+ // returns. To ensure this, derive a new context. Note that it is okay to
24
+ // listen on parent in this scope, but NOT okay to pass |parent| to
25
+ // functions called by this one. Otherwise those functions won't return
26
+ // when this context's cancel func is executed. This is difficult to
27
+ // enforce. May this comment keep you safe.
28
+ ctx, cancel := context.WithCancel(p)
29
+ defer cancel()
30
+
31
+ promise, err := gb(ctx, []*cid.Cid{k})
32
+ if err != nil {
33
+ return nil, err
34
+ }
35
+
36
+ select {
37
+ case block, ok := <-promise:
38
+ if !ok {
39
+ select {
40
+ case <-ctx.Done():
41
+ return nil, ctx.Err()
42
+ default:
43
+ return nil, errors.New("promise channel was closed")
44
+ }
45
+ }
46
+ return block, nil
47
+ case <-p.Done():
48
+ return nil, p.Err()
49
+ }
50
+}
51
+
52
+type wantFunc func(context.Context, []*cid.Cid)
53
+
54
+func getBlocksImpl(ctx context.Context, keys []*cid.Cid, notif notifications.PubSub, want wantFunc, cwants func([]*cid.Cid)) (<-chan blocks.Block, error) {
55
+ if len(keys) == 0 {
56
+ out := make(chan blocks.Block)
57
+ close(out)
58
+ return out, nil
59
+ }
60
+
61
+ remaining := cid.NewSet()
62
+ promise := notif.Subscribe(ctx, keys...)
63
+ for _, k := range keys {
64
+ log.Event(ctx, "Bitswap.GetBlockRequest.Start", k)
65
+ remaining.Add(k)
66
+ }
67
+
68
+ want(ctx, keys)
69
+
70
+ out := make(chan blocks.Block)
71
+ go handleIncoming(ctx, remaining, promise, out, cwants)
72
+ return out, nil
73
+}
74
+
75
+func handleIncoming(ctx context.Context, remaining *cid.Set, in <-chan blocks.Block, out chan blocks.Block, cfun func([]*cid.Cid)) {
76
+ ctx, cancel := context.WithCancel(ctx)
77
+ defer func() {
78
+ cancel()
79
+ close(out)
80
+ // can't just defer this call on its own, arguments are resolved *when* the defer is created
81
+ cfun(remaining.Keys())
82
+ }()
83
+ for {
84
+ select {
85
+ case blk, ok := <-in:
86
+ if !ok {
87
+ return
88
+ }
89
+
90
+ remaining.Remove(blk.Cid())
91
+ select {
92
+ case out <- blk:
93
+ case <-ctx.Done():
94
+ return
95
+ }
96
+ case <-ctx.Done():
97
+ return
98
+ }
99
+ }
100
+}
exchange/bitswap/session.go
new
+221
@@ -0,0 +1,221 @@
1
+package bitswap
2
+
3
+import (
4
+ "context"
5
+ "time"
6
+
7
+ blocks "github.com/ipfs/go-ipfs/blocks"
8
+ notifications "github.com/ipfs/go-ipfs/exchange/bitswap/notifications"
9
+
10
+ logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
11
+ lru "gx/ipfs/QmVYxfoJQiZijTgPNHCHgHELvQpbsJNTg6Crmc3dQkj3yy/golang-lru"
12
+ loggables "gx/ipfs/QmVesPmqbPp7xRGyY96tnBwzDtVV1nqv4SCVxo5zCqKyH8/go-libp2p-loggables"
13
+ cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
14
+ peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
15
+)
16
+
17
+const activeWantsLimit = 16
18
+
19
+type Session struct {
20
+ ctx context.Context
21
+ tofetch []*cid.Cid
22
+ activePeers map[peer.ID]struct{}
23
+ activePeersArr []peer.ID
24
+
25
+ bs *Bitswap
26
+ incoming chan blkRecv
27
+ newReqs chan []*cid.Cid
28
+ cancelKeys chan []*cid.Cid
29
+
30
+ interest *lru.Cache
31
+ liveWants map[string]time.Time
32
+ liveCnt int
33
+
34
+ tick *time.Timer
35
+ baseTickDelay time.Duration
36
+
37
+ latTotal time.Duration
38
+ fetchcnt int
39
+
40
+ notif notifications.PubSub
41
+
42
+ uuid logging.Loggable
43
+}
44
+
45
+func (bs *Bitswap) NewSession(ctx context.Context) *Session {
46
+ s := &Session{
47
+ activePeers: make(map[peer.ID]struct{}),
48
+ liveWants: make(map[string]time.Time),
49
+ newReqs: make(chan []*cid.Cid),
50
+ cancelKeys: make(chan []*cid.Cid),
51
+ ctx: ctx,
52
+ bs: bs,
53
+ incoming: make(chan blkRecv),
54
+ notif: notifications.New(),
55
+ uuid: loggables.Uuid("GetBlockRequest"),
56
+ baseTickDelay: time.Millisecond * 500,
57
+ }
58
+
59
+ cache, _ := lru.New(2048)
60
+ s.interest = cache
61
+
62
+ bs.sessLk.Lock()
63
+ bs.sessions = append(bs.sessions, s)
64
+ bs.sessLk.Unlock()
65
+
66
+ go s.run(ctx)
67
+
68
+ return s
69
+}
70
+
71
+type blkRecv struct {
72
+ from peer.ID
73
+ blk blocks.Block
74
+}
75
+
76
+func (s *Session) ReceiveBlock(from peer.ID, blk blocks.Block) {
77
+ s.incoming <- blkRecv{from: from, blk: blk}
78
+}
79
+
80
+func (s *Session) InterestedIn(c *cid.Cid) bool {
81
+ return s.interest.Contains(c.KeyString())
82
+}
83
+
84
+const provSearchDelay = time.Second * 10
85
+
86
+func (s *Session) addActivePeer(p peer.ID) {
87
+ if _, ok := s.activePeers[p]; !ok {
88
+ s.activePeers[p] = struct{}{}
89
+ s.activePeersArr = append(s.activePeersArr, p)
90
+ }
91
+}
92
+
93
+func (s *Session) resetTick() {
94
+ if s.latTotal == 0 {
95
+ s.tick.Reset(provSearchDelay)
96
+ } else {
97
+ avLat := s.latTotal / time.Duration(s.fetchcnt)
98
+ s.tick.Reset(s.baseTickDelay + (3 * avLat))
99
+ }
100
+}
101
+
102
+func (s *Session) run(ctx context.Context) {
103
+ s.tick = time.NewTimer(provSearchDelay)
104
+ newpeers := make(chan peer.ID, 16)
105
+ for {
106
+ select {
107
+ case blk := <-s.incoming:
108
+ s.tick.Stop()
109
+
110
+ s.addActivePeer(blk.from)
111
+
112
+ s.receiveBlock(ctx, blk.blk)
113
+
114
+ s.resetTick()
115
+ case keys := <-s.newReqs:
116
+ for _, k := range keys {
117
+ s.interest.Add(k.KeyString(), nil)
118
+ }
119
+ if s.liveCnt < activeWantsLimit {
120
+ toadd := activeWantsLimit - s.liveCnt
121
+ if toadd > len(keys) {
122
+ toadd = len(keys)
123
+ }
124
+ s.liveCnt += toadd
125
+
126
+ now := keys[:toadd]
127
+ keys = keys[toadd:]
128
+
129
+ s.wantBlocks(ctx, now)
130
+ }
131
+ s.tofetch = append(s.tofetch, keys...)
132
+ case keys := <-s.cancelKeys:
133
+ s.cancel(keys)
134
+
135
+ case <-s.tick.C:
136
+ var live []*cid.Cid
137
+ for c, _ := range s.liveWants {
138
+ cs, _ := cid.Cast([]byte(c))
139
+ live = append(live, cs)
140
+ s.liveWants[c] = time.Now()
141
+ }
142
+
143
+ // Broadcast these keys to everyone we're connected to
144
+ s.bs.wm.WantBlocks(ctx, live, nil)
145
+
146
+ if len(live) > 0 {
147
+ go func() {
148
+ for p := range s.bs.network.FindProvidersAsync(ctx, live[0], 10) {
149
+ newpeers <- p
150
+ }
151
+ }()
152
+ }
153
+ s.resetTick()
154
+ case p := <-newpeers:
155
+ s.addActivePeer(p)
156
+ case <-ctx.Done():
157
+ return
158
+ }
159
+ }
160
+}
161
+
162
+func (s *Session) receiveBlock(ctx context.Context, blk blocks.Block) {
163
+ ks := blk.Cid().KeyString()
164
+ if _, ok := s.liveWants[ks]; ok {
165
+ s.liveCnt--
166
+ tval := s.liveWants[ks]
167
+ s.latTotal += time.Since(tval)
168
+ s.fetchcnt++
169
+ delete(s.liveWants, ks)
170
+ s.notif.Publish(blk)
171
+
172
+ if len(s.tofetch) > 0 {
173
+ next := s.tofetch[0:1]
174
+ s.tofetch = s.tofetch[1:]
175
+ s.wantBlocks(ctx, next)
176
+ }
177
+ }
178
+}
179
+
180
+func (s *Session) wantBlocks(ctx context.Context, ks []*cid.Cid) {
181
+ for _, c := range ks {
182
+ s.liveWants[c.KeyString()] = time.Now()
183
+ }
184
+ s.bs.wm.WantBlocks(ctx, ks, s.activePeersArr)
185
+}
186
+
187
+func (s *Session) cancel(keys []*cid.Cid) {
188
+ sset := cid.NewSet()
189
+ for _, c := range keys {
190
+ sset.Add(c)
191
+ }
192
+ var i, j int
193
+ for ; j < len(s.tofetch); j++ {
194
+ if sset.Has(s.tofetch[j]) {
195
+ continue
196
+ }
197
+ s.tofetch[i] = s.tofetch[j]
198
+ i++
199
+ }
200
+ s.tofetch = s.tofetch[:i]
201
+}
202
+
203
+func (s *Session) cancelWants(keys []*cid.Cid) {
204
+ s.cancelKeys <- keys
205
+}
206
+
207
+func (s *Session) fetch(ctx context.Context, keys []*cid.Cid) {
208
+ select {
209
+ case s.newReqs <- keys:
210
+ case <-ctx.Done():
211
+ }
212
+}
213
+
214
+func (s *Session) GetBlocks(ctx context.Context, keys []*cid.Cid) (<-chan blocks.Block, error) {
215
+ ctx = logging.ContextWithLoggable(ctx, s.uuid)
216
+ return getBlocksImpl(ctx, keys, s.notif, s.fetch, s.cancelWants)
217
+}
218
+
219
+func (s *Session) GetBlock(parent context.Context, k *cid.Cid) (blocks.Block, error) {
220
+ return getBlock(parent, k, s.GetBlocks)
221
+}
exchange/bitswap/session_test.go
new
+152
@@ -0,0 +1,152 @@
1
+package bitswap
2
+
3
+import (
4
+ "context"
5
+ "fmt"
6
+ "testing"
7
+ "time"
8
+
9
+ blocks "github.com/ipfs/go-ipfs/blocks"
10
+ blocksutil "github.com/ipfs/go-ipfs/blocks/blocksutil"
11
+
12
+ cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
13
+)
14
+
15
+func TestBasicSessions(t *testing.T) {
16
+ ctx, cancel := context.WithCancel(context.Background())
17
+ defer cancel()
18
+
19
+ vnet := getVirtualNetwork()
20
+ sesgen := NewTestSessionGenerator(vnet)
21
+ defer sesgen.Close()
22
+ bgen := blocksutil.NewBlockGenerator()
23
+
24
+ block := bgen.Next()
25
+ inst := sesgen.Instances(2)
26
+
27
+ a := inst[0]
28
+ b := inst[1]
29
+
30
+ if err := b.Blockstore().Put(block); err != nil {
31
+ t.Fatal(err)
32
+ }
33
+
34
+ sesa := a.Exchange.NewSession(ctx)
35
+
36
+ blkout, err := sesa.GetBlock(ctx, block.Cid())
37
+ if err != nil {
38
+ t.Fatal(err)
39
+ }
40
+
41
+ if !blkout.Cid().Equals(block.Cid()) {
42
+ t.Fatal("got wrong block")
43
+ }
44
+}
45
+
46
+func assertBlockLists(got, exp []blocks.Block) error {
47
+ if len(got) != len(exp) {
48
+ return fmt.Errorf("got wrong number of blocks, %d != %d", len(got), len(exp))
49
+ }
50
+
51
+ h := cid.NewSet()
52
+ for _, b := range got {
53
+ h.Add(b.Cid())
54
+ }
55
+ for _, b := range exp {
56
+ if !h.Has(b.Cid()) {
57
+ return fmt.Errorf("didnt have: %s", b.Cid())
58
+ }
59
+ }
60
+ return nil
61
+}
62
+
63
+func TestSessionBetweenPeers(t *testing.T) {
64
+ ctx, cancel := context.WithCancel(context.Background())
65
+ defer cancel()
66
+
67
+ vnet := getVirtualNetwork()
68
+ sesgen := NewTestSessionGenerator(vnet)
69
+ defer sesgen.Close()
70
+ bgen := blocksutil.NewBlockGenerator()
71
+
72
+ inst := sesgen.Instances(10)
73
+
74
+ blks := bgen.Blocks(101)
75
+ if err := inst[0].Blockstore().PutMany(blks); err != nil {
76
+ t.Fatal(err)
77
+ }
78
+
79
+ var cids []*cid.Cid
80
+ for _, blk := range blks {
81
+ cids = append(cids, blk.Cid())
82
+ }
83
+
84
+ ses := inst[1].Exchange.NewSession(ctx)
85
+ if _, err := ses.GetBlock(ctx, cids[0]); err != nil {
86
+ t.Fatal(err)
87
+ }
88
+ blks = blks[1:]
89
+ cids = cids[1:]
90
+
91
+ for i := 0; i < 10; i++ {
92
+ ch, err := ses.GetBlocks(ctx, cids[i*10:(i+1)*10])
93
+ if err != nil {
94
+ t.Fatal(err)
95
+ }
96
+
97
+ var got []blocks.Block
98
+ for b := range ch {
99
+ got = append(got, b)
100
+ }
101
+ if err := assertBlockLists(got, blks[i*10:(i+1)*10]); err != nil {
102
+ t.Fatal(err)
103
+ }
104
+ }
105
+ for _, is := range inst[2:] {
106
+ if is.Exchange.messagesRecvd > 2 {
107
+ t.Fatal("uninvolved nodes should only receive two messages", is.Exchange.messagesRecvd)
108
+ }
109
+ }
110
+}
111
+
112
+func TestSessionSplitFetch(t *testing.T) {
113
+ ctx, cancel := context.WithCancel(context.Background())
114
+ defer cancel()
115
+
116
+ vnet := getVirtualNetwork()
117
+ sesgen := NewTestSessionGenerator(vnet)
118
+ defer sesgen.Close()
119
+ bgen := blocksutil.NewBlockGenerator()
120
+
121
+ inst := sesgen.Instances(11)
122
+
123
+ blks := bgen.Blocks(100)
124
+ for i := 0; i < 10; i++ {
125
+ if err := inst[i].Blockstore().PutMany(blks[i*10 : (i+1)*10]); err != nil {
126
+ t.Fatal(err)
127
+ }
128
+ }
129
+
130
+ var cids []*cid.Cid
131
+ for _, blk := range blks {
132
+ cids = append(cids, blk.Cid())
133
+ }
134
+
135
+ ses := inst[10].Exchange.NewSession(ctx)
136
+ ses.baseTickDelay = time.Millisecond * 10
137
+
138
+ for i := 0; i < 10; i++ {
139
+ ch, err := ses.GetBlocks(ctx, cids[i*10:(i+1)*10])
140
+ if err != nil {
141
+ t.Fatal(err)
142
+ }
143
+
144
+ var got []blocks.Block
145
+ for b := range ch {
146
+ got = append(got, b)
147
+ }
148
+ if err := assertBlockLists(got, blks[i*10:(i+1)*10]); err != nil {
149
+ t.Fatal(err)
150
+ }
151
+ }
152
+}
exchange/bitswap/testutils.go
+2
-2
@@ -47,7 +47,7 @@ func (g *SessionGenerator) Next() Instance {
47
if err != nil {
48
panic("FIXME") // TODO change signature
49
}
50
- return Session(g.ctx, g.net, p)
50
+ return MkSession(g.ctx, g.net, p)
51
}
52
53
func (g *SessionGenerator) Instances(n int) []Instance {
@@ -86,7 +86,7 @@ func (i *Instance) SetBlockstoreLatency(t time.Duration) time.Duration {
86
// NB: It's easy make mistakes by providing the same peer ID to two different
87
// sessions. To safeguard, use the SessionGenerator to generate sessions. It's
88
// just a much better idea.
89
-func Session(ctx context.Context, net tn.Network, p testutil.Identity) Instance {
89
+func MkSession(ctx context.Context, net tn.Network, p testutil.Identity) Instance {
90
bsdelay := delay.Fixed(0)
91
92
adapter := net.Adapter(p)
exchange/bitswap/wantmanager.go
+6
-6
@@ -71,13 +71,13 @@ type msgQueue struct {
71
done chan struct{}
72
}
73
74
-func (pm *WantManager) WantBlocks(ctx context.Context, ks []*cid.Cid) {
74
+func (pm *WantManager) WantBlocks(ctx context.Context, ks []*cid.Cid, peers []peer.ID) {
75
log.Infof("want blocks: %s", ks)
76
- pm.addEntries(ctx, ks, false)
76
+ pm.addEntries(ctx, ks, peers, false)
77
}
78
79
-func (pm *WantManager) CancelWants(ks []*cid.Cid) {
80
- pm.addEntries(context.Background(), ks, true)
79
+func (pm *WantManager) CancelWants(ctx context.Context, ks []*cid.Cid, peers []peer.ID) {
80
+ pm.addEntries(context.Background(), ks, peers, true)
81
}
82
83
type wantSet struct {
@@ -85,7 +85,7 @@ type wantSet struct {
85
targets []peer.ID
86
}
87
88
-func (pm *WantManager) addEntries(ctx context.Context, ks []*cid.Cid, cancel bool) {
88
+func (pm *WantManager) addEntries(ctx context.Context, ks []*cid.Cid, targets []peer.ID, cancel bool) {
89
var entries []*bsmsg.Entry
90
for i, k := range ks {
91
entries = append(entries, &bsmsg.Entry{
@@ -98,7 +98,7 @@ func (pm *WantManager) addEntries(ctx context.Context, ks []*cid.Cid, cancel boo
98
})
99
}
100
select {
101
- case pm.incoming <- &wantSet{entries: entries}:
101
+ case pm.incoming <- &wantSet{entries: entries, targets: targets}:
102
case <-pm.ctx.Done():
103
case <-ctx.Done():
104
}
exchange/bitswap/workers.go
+1
-1
@@ -49,7 +49,7 @@ func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
49
50
func (bs *Bitswap) taskWorker(ctx context.Context, id int) {
51
idmap := logging.LoggableMap{"ID": id}
52
- defer log.Info("bitswap task worker shutting down...")
52
+ defer log.Debug("bitswap task worker shutting down...")
53
for {
54
log.Event(ctx, "Bitswap.TaskWorker.Loop", idmap)
55
select {