add worker to bitswap for reproviding new blocks
Jeromy committed
Feb 18, 2015 at 08:18 UTC
d7eb57f48fb4f84d6474e4cfa1a3983120d3f7d8
3 files changed
+156
-91
blocks/blocks.go
+6
@@ -42,3 +42,9 @@ func (b *Block) Key() u.Key {
42
func (b *Block) String() string {
43
return fmt.Sprintf("[Block %s]", b.Key())
44
}
45
+
46
+func (b *Block) Loggable() map[string]interface{} {
47
+ return map[string]interface{}{
48
+ "block": b.Key().String(),
49
+ }
50
+}
exchange/bitswap/bitswap.go
+17
-91
@@ -8,7 +8,6 @@ import (
8
"time"
9
10
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
11
- inflect "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/briantigerchow/inflect"
11
process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
12
13
blocks "github.com/jbenet/go-ipfs/blocks"
@@ -37,9 +36,13 @@ const (
36
maxProvidersPerRequest = 3
37
providerRequestTimeout = time.Second * 10
38
hasBlockTimeout = time.Second * 15
39
+ provideTimeout = time.Second * 15
40
sizeBatchRequestChan = 32
41
// kMaxPriority is the max priority as defined by the bitswap protocol
42
kMaxPriority = math.MaxInt32
43
+
44
+ hasBlockBufferSize = 256
45
+ provideWorkers = 4
46
)
47
48
var (
@@ -86,18 +89,12 @@ func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
89
wantlist: wantlist.NewThreadSafe(),
90
batchRequests: make(chan *blockRequest, sizeBatchRequestChan),
91
process: px,
92
+ newBlocks: make(chan *blocks.Block, hasBlockBufferSize),
93
}
94
network.SetDelegate(bs)
91
- px.Go(func(px process.Process) {
92
- bs.clientWorker(ctx)
93
- })
94
- px.Go(func(px process.Process) {
95
- bs.taskWorker(ctx)
96
- })
97
- px.Go(func(px process.Process) {
98
- bs.rebroadcastWorker(ctx)
99
- })
95
96
+ // Start up bitswaps async worker routines
97
+ bs.startWorkers(px, ctx)
98
return bs
99
}
100
@@ -126,6 +123,8 @@ type bitswap struct {
123
wantlist *wantlist.ThreadSafe
124
125
process process.Process
126
+
127
+ newBlocks chan *blocks.Block
128
}
129
130
type blockRequest struct {
@@ -172,7 +171,6 @@ func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, err
171
case <-parent.Done():
172
return nil, parent.Err()
173
}
175
-
174
}
175
176
// GetBlocks returns a channel where the caller may receive blocks that
@@ -205,6 +203,7 @@ func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.
203
// HasBlock announces the existance of a block to this bitswap service. The
204
// service will potentially notify its peers.
205
func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
206
+ log.Event(ctx, "hasBlock", blk)
207
select {
208
case <-bs.process.Closing():
209
return errors.New("bitswap is closed")
@@ -215,7 +214,12 @@ func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
214
}
215
bs.wantlist.Remove(blk.Key())
216
bs.notifications.Publish(blk)
218
- return bs.network.Provide(ctx, blk.Key())
217
+ select {
218
+ case bs.newBlocks <- blk:
219
+ case <-ctx.Done():
220
+ return ctx.Err()
221
+ }
222
+ return nil
223
}
224
225
func (bs *bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMessage, peers <-chan peer.ID) error {
@@ -310,6 +314,7 @@ func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
314
log.Debug(err)
315
}
316
}
317
+
318
var keys []u.Key
319
for _, block := range incoming.Blocks() {
320
keys = append(keys, block.Key())
@@ -391,82 +396,3 @@ func (bs *bitswap) send(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage)
396
func (bs *bitswap) Close() error {
397
return bs.process.Close()
398
}
394
-
395
-func (bs *bitswap) taskWorker(ctx context.Context) {
396
- defer log.Info("bitswap task worker shutting down...")
397
- for {
398
- select {
399
- case <-ctx.Done():
400
- return
401
- case nextEnvelope := <-bs.engine.Outbox():
402
- select {
403
- case <-ctx.Done():
404
- return
405
- case envelope, ok := <-nextEnvelope:
406
- if !ok {
407
- continue
408
- }
409
- log.Event(ctx, "deliverBlocks", envelope.Message, envelope.Peer)
410
- bs.send(ctx, envelope.Peer, envelope.Message)
411
- }
412
- }
413
- }
414
-}
415
-
416
-// TODO ensure only one active request per key
417
-func (bs *bitswap) clientWorker(parent context.Context) {
418
- defer log.Info("bitswap client worker shutting down...")
419
-
420
- for {
421
- select {
422
- case req := <-bs.batchRequests:
423
- keys := req.keys
424
- if len(keys) == 0 {
425
- log.Warning("Received batch request for zero blocks")
426
- continue
427
- }
428
- for i, k := range keys {
429
- bs.wantlist.Add(k, kMaxPriority-i)
430
- }
431
-
432
- bs.wantNewBlocks(req.ctx, keys)
433
-
434
- // NB: Optimization. Assumes that providers of key[0] are likely to
435
- // be able to provide for all keys. This currently holds true in most
436
- // every situation. Later, this assumption may not hold as true.
437
- child, _ := context.WithTimeout(req.ctx, providerRequestTimeout)
438
- providers := bs.network.FindProvidersAsync(child, keys[0], maxProvidersPerRequest)
439
- err := bs.sendWantlistToPeers(req.ctx, providers)
440
- if err != nil {
441
- log.Debugf("error sending wantlist: %s", err)
442
- }
443
- case <-parent.Done():
444
- return
445
- }
446
- }
447
-}
448
-
449
-func (bs *bitswap) rebroadcastWorker(parent context.Context) {
450
- ctx, cancel := context.WithCancel(parent)
451
- defer cancel()
452
-
453
- broadcastSignal := time.After(rebroadcastDelay.Get())
454
-
455
- for {
456
- select {
457
- case <-time.Tick(10 * time.Second):
458
- n := bs.wantlist.Len()
459
- if n > 0 {
460
- log.Debug(n, inflect.FromNumber("keys", n), "in bitswap wantlist")
461
- }
462
- case <-broadcastSignal: // resend unfulfilled wantlist keys
463
- entries := bs.wantlist.Entries()
464
- if len(entries) > 0 {
465
- bs.sendWantlistToProviders(ctx, entries)
466
- }
467
- broadcastSignal = time.After(rebroadcastDelay.Get())
468
- case <-parent.Done():
469
- return
470
- }
471
- }
472
-}
exchange/bitswap/workers.go
new
+133
@@ -0,0 +1,133 @@
1
+package bitswap
2
+
3
+import (
4
+ "time"
5
+
6
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
7
+ inflect "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/briantigerchow/inflect"
8
+ process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
9
+)
10
+
11
+func (bs *bitswap) startWorkers(px process.Process, ctx context.Context) {
12
+ // Start up a worker to handle block requests this node is making
13
+ px.Go(func(px process.Process) {
14
+ bs.clientWorker(ctx)
15
+ })
16
+
17
+ // Start up a worker to handle requests from other nodes for the data on this node
18
+ px.Go(func(px process.Process) {
19
+ bs.taskWorker(ctx)
20
+ })
21
+
22
+ // Start up a worker to manage periodically resending our wantlist out to peers
23
+ px.Go(func(px process.Process) {
24
+ bs.rebroadcastWorker(ctx)
25
+ })
26
+
27
+ // Spawn up multiple workers to handle incoming blocks
28
+ // consider increasing number if providing blocks bottlenecks
29
+ // file transfers
30
+ for i := 0; i < provideWorkers; i++ {
31
+ px.Go(func(px process.Process) {
32
+ bs.blockReceiveWorker(ctx)
33
+ })
34
+ }
35
+}
36
+
37
+func (bs *bitswap) taskWorker(ctx context.Context) {
38
+ defer log.Info("bitswap task worker shutting down...")
39
+ for {
40
+ select {
41
+ case nextEnvelope := <-bs.engine.Outbox():
42
+ select {
43
+ case envelope, ok := <-nextEnvelope:
44
+ if !ok {
45
+ continue
46
+ }
47
+ log.Event(ctx, "deliverBlocks", envelope.Message, envelope.Peer)
48
+ bs.send(ctx, envelope.Peer, envelope.Message)
49
+ case <-ctx.Done():
50
+ return
51
+ }
52
+ case <-ctx.Done():
53
+ return
54
+ }
55
+ }
56
+}
57
+
58
+func (bs *bitswap) blockReceiveWorker(ctx context.Context) {
59
+ for {
60
+ select {
61
+ case blk, ok := <-bs.newBlocks:
62
+ if !ok {
63
+ log.Debug("newBlocks channel closed")
64
+ return
65
+ }
66
+ ctx, _ := context.WithTimeout(ctx, provideTimeout)
67
+ err := bs.network.Provide(ctx, blk.Key())
68
+ if err != nil {
69
+ log.Error(err)
70
+ }
71
+ case <-ctx.Done():
72
+ return
73
+ }
74
+ }
75
+}
76
+
77
+// TODO ensure only one active request per key
78
+func (bs *bitswap) clientWorker(parent context.Context) {
79
+ defer log.Info("bitswap client worker shutting down...")
80
+
81
+ for {
82
+ select {
83
+ case req := <-bs.batchRequests:
84
+ keys := req.keys
85
+ if len(keys) == 0 {
86
+ log.Warning("Received batch request for zero blocks")
87
+ continue
88
+ }
89
+ for i, k := range keys {
90
+ bs.wantlist.Add(k, kMaxPriority-i)
91
+ }
92
+
93
+ bs.wantNewBlocks(req.ctx, keys)
94
+
95
+ // NB: Optimization. Assumes that providers of key[0] are likely to
96
+ // be able to provide for all keys. This currently holds true in most
97
+ // every situation. Later, this assumption may not hold as true.
98
+ child, _ := context.WithTimeout(req.ctx, providerRequestTimeout)
99
+ providers := bs.network.FindProvidersAsync(child, keys[0], maxProvidersPerRequest)
100
+ err := bs.sendWantlistToPeers(req.ctx, providers)
101
+ if err != nil {
102
+ log.Debugf("error sending wantlist: %s", err)
103
+ }
104
+ case <-parent.Done():
105
+ return
106
+ }
107
+ }
108
+}
109
+
110
+func (bs *bitswap) rebroadcastWorker(parent context.Context) {
111
+ ctx, cancel := context.WithCancel(parent)
112
+ defer cancel()
113
+
114
+ broadcastSignal := time.After(rebroadcastDelay.Get())
115
+
116
+ for {
117
+ select {
118
+ case <-time.Tick(10 * time.Second):
119
+ n := bs.wantlist.Len()
120
+ if n > 0 {
121
+ log.Debug(n, inflect.FromNumber("keys", n), "in bitswap wantlist")
122
+ }
123
+ case <-broadcastSignal: // resend unfulfilled wantlist keys
124
+ entries := bs.wantlist.Entries()
125
+ if len(entries) > 0 {
126
+ bs.sendWantlistToProviders(ctx, entries)
127
+ }
128
+ broadcastSignal = time.After(rebroadcastDelay.Get())
129
+ case <-parent.Done():
130
+ return
131
+ }
132
+ }
133
+}