a little more correctness on the new bitswap impl
Jeromy committed
Nov 21, 2014 at 23:03 UTC
9120d107c33e1e2f365997ea3bc328506be5ae30
6 files changed
+84
-29
blockservice/blockservice.go
+3
-1
@@ -98,7 +98,7 @@ func (s *BlockService) GetBlock(ctx context.Context, k u.Key) (*blocks.Block, er
98
// GetBlocks gets a list of blocks asynchronously and returns through
99
// the returned channel.
100
// NB: No guarantees are made about order.
101
-func (s *BlockService) GetBlocks(ctx context.Context, ks []u.Key) <-chan *blocks.Block {
101
+func (s *BlockService) GetBlocks(parent context.Context, ks []u.Key) <-chan *blocks.Block {
102
out := make(chan *blocks.Block, 32)
103
go func() {
104
var toFetch []u.Key
@@ -112,11 +112,13 @@ func (s *BlockService) GetBlocks(ctx context.Context, ks []u.Key) <-chan *blocks
112
out <- block
113
}
114
115
+ ctx, cancel := context.WithCancel(parent)
116
nblocks, err := s.Remote.GetBlocks(ctx, toFetch)
117
if err != nil {
118
log.Errorf("Error with GetBlocks: %s", err)
119
return
120
}
121
+
122
for blk := range nblocks {
123
out <- blk
124
}
exchange/bitswap/bitswap.go
+37
-10
@@ -128,12 +128,35 @@ func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.
128
promise := bs.notifications.Subscribe(ctx, keys...)
129
select {
130
case bs.batchRequests <- keys:
131
- return promise, nil
131
+ return pipeBlocks(ctx, promise, len(keys)), nil
132
case <-ctx.Done():
133
return nil, ctx.Err()
134
}
135
}
136
137
+func pipeBlocks(ctx context.Context, in <-chan *blocks.Block, count int) <-chan *blocks.Block {
138
+ out := make(chan *blocks.Block, 1)
139
+ go func() {
140
+ defer close(out)
141
+ for i := 0; i < count; i++ {
142
+ select {
143
+ case blk, ok := <-in:
144
+ if !ok {
145
+ return
146
+ }
147
+ select {
148
+ case out <- blk:
149
+ case <-ctx.Done():
150
+ return
151
+ }
152
+ case <-ctx.Done():
153
+ return
154
+ }
155
+ }
156
+ }()
157
+ return out
158
+}
159
+
160
func (bs *bitswap) sendWantListTo(ctx context.Context, peers <-chan peer.Peer) error {
161
if peers == nil {
162
panic("Cant send wantlist to nil peerchan")
@@ -220,7 +243,7 @@ func (bs *bitswap) loop(parent context.Context) {
243
// HasBlock announces the existance of a block to this bitswap service. The
244
// service will potentially notify its peers.
245
func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
223
- log.Debugf("Has Block %v", blk.Key())
246
+ log.Debugf("Has Block %s", blk.Key())
247
bs.wantlist.Remove(blk.Key())
248
bs.sendToPeersThatWant(ctx, blk)
249
return bs.routing.Provide(ctx, blk.Key())
@@ -262,10 +285,6 @@ func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.Peer, incoming bsm
285
}
286
}
287
265
- message := bsmsg.New()
266
- for _, wanted := range bs.wantlist.Keys() {
267
- message.AddWanted(wanted)
268
- }
288
for _, key := range incoming.Wantlist() {
289
// TODO: might be better to check if we have the block before checking
290
// if we should send it to someone
@@ -273,14 +292,22 @@ func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.Peer, incoming bsm
292
if block, errBlockNotFound := bs.blockstore.Get(key); errBlockNotFound != nil {
293
continue
294
} else {
276
- message.AddBlock(block)
295
+ // Create a separate message to send this block in
296
+ blkmsg := bsmsg.New()
297
+
298
+ // TODO: only send this the first time
299
+ for _, k := range bs.wantlist.Keys() {
300
+ blkmsg.AddWanted(k)
301
+ }
302
+
303
+ blkmsg.AddBlock(block)
304
+ bs.strategy.MessageSent(p, blkmsg)
305
+ bs.send(ctx, p, blkmsg)
306
}
307
}
308
}
309
281
- bs.strategy.MessageSent(p, message)
282
- log.Debug("Returning message.")
283
- return p, message
310
+ return nil, nil
311
}
312
313
func (bs *bitswap) ReceiveError(err error) {
exchange/bitswap/bitswap_test.go
+1
-1
@@ -106,7 +106,7 @@ func TestLargeSwarm(t *testing.T) {
106
t.SkipNow()
107
}
108
t.Parallel()
109
- numInstances := 500
109
+ numInstances := 5
110
numBlocks := 2
111
PerformDistributionTest(t, numInstances, numBlocks)
112
}
exchange/bitswap/strategy/ledger.go
+1
@@ -61,6 +61,7 @@ func (l *ledger) ReceivedBytes(n int) {
61
62
// TODO: this needs to be different. We need timeouts.
63
func (l *ledger) Wants(k u.Key) {
64
+ log.Debugf("peer %s wants %s", l.Partner, k)
65
l.wantList[k] = struct{}{}
66
}
67
exchange/bitswap/strategy/strategy.go
+2
@@ -10,6 +10,8 @@ import (
10
u "github.com/jbenet/go-ipfs/util"
11
)
12
13
+var log = u.Logger("strategy")
14
+
15
// TODO niceness should be on a per-peer basis. Use-case: Certain peers are
16
// "trusted" and/or controlled by a single human user. The user may want for
17
// these peers to exchange data freely
merkledag/merkledag.go
+40
-17
@@ -163,6 +163,17 @@ func (n *Node) Multihash() (mh.Multihash, error) {
163
return n.cached, nil
164
}
165
166
+// Searches this nodes links for one to the given key,
167
+// returns the index of said link
168
+func (n *Node) FindLink(k u.Key) (int, error) {
169
+ for i, lnk := range n.Links {
170
+ if u.Key(lnk.Hash) == k {
171
+ return i, nil
172
+ }
173
+ }
174
+ return -1, u.ErrNotFound
175
+}
176
+
177
// Key returns the Multihash as a key, for maps.
178
func (n *Node) Key() (u.Key, error) {
179
h, err := n.Multihash()
@@ -296,6 +307,10 @@ func (ds *dagService) BatchFetch(ctx context.Context, root *Node) <-chan *Node {
307
var keys []u.Key
308
nodes := make([]*Node, len(root.Links))
309
310
+ //temp
311
+ recvd := []int{}
312
+ //
313
+
314
//
315
next := 0
316
//
@@ -306,28 +321,36 @@ func (ds *dagService) BatchFetch(ctx context.Context, root *Node) <-chan *Node {
321
322
blkchan := ds.Blocks.GetBlocks(ctx, keys)
323
324
+ count := 0
325
for blk := range blkchan {
310
- for i, lnk := range root.Links {
311
- if u.Key(lnk.Hash) != blk.Key() {
312
- continue
313
- }
326
+ count++
327
+ i, err := root.FindLink(blk.Key())
328
+ if err != nil {
329
+ panic("Received block that wasnt in this nodes links!")
330
+ }
331
315
- nd, err := Decoded(blk.Data)
316
- if err != nil {
317
- log.Error("Got back bad block!")
318
- break
319
- }
320
- nodes[i] = nd
321
-
322
- if next == i {
323
- sig <- nd
324
- next++
325
- for ; next < len(nodes) && nodes[next] != nil; next++ {
326
- sig <- nodes[next]
327
- }
332
+ recvd = append(recvd, i)
333
+
334
+ nd, err := Decoded(blk.Data)
335
+ if err != nil {
336
+ log.Error("Got back bad block!")
337
+ break
338
+ }
339
+ nodes[i] = nd
340
+
341
+ if next == i {
342
+ sig <- nd
343
+ next++
344
+ for ; next < len(nodes) && nodes[next] != nil; next++ {
345
+ sig <- nodes[next]
346
}
347
}
348
}
349
+ if next < len(nodes) {
350
+ log.Errorf("count = %d, links = %d", count, len(nodes))
351
+ log.Error(recvd)
352
+ panic("didnt receive all requested blocks!")
353
+ }
354
close(sig)
355
}()
356