implement a worker to consolidate HasBlock provide calls into one to alieviate memory pressure
Jeromy committed
Mar 5, 2015 at 15:18 UTC
8937f5fbda7eb780dd41e42938d31ad0729cedde
2 files changed
+56
-3
exchange/bitswap/bitswap.go
+3
@@ -89,6 +89,7 @@ func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
89
batchRequests: make(chan *blockRequest, sizeBatchRequestChan),
90
process: px,
91
newBlocks: make(chan *blocks.Block, HasBlockBufferSize),
92
+ provideKeys: make(chan u.Key),
93
}
94
network.SetDelegate(bs)
95
@@ -124,6 +125,8 @@ type Bitswap struct {
125
process process.Process
126
127
newBlocks chan *blocks.Block
128
+
129
+ provideKeys chan u.Key
130
}
131
132
type blockRequest struct {
exchange/bitswap/workers.go
+53
-3
@@ -6,6 +6,7 @@ import (
6
inflect "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/chuckpreslar/inflect"
7
process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
9
+ u "github.com/jbenet/go-ipfs/util"
10
)
11
12
func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
@@ -24,6 +25,10 @@ func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
25
bs.rebroadcastWorker(ctx)
26
})
27
28
+ px.Go(func(px process.Process) {
29
+ bs.provideCollector(ctx)
30
+ })
31
+
32
// Spawn up multiple workers to handle incoming blocks
33
// consider increasing number if providing blocks bottlenecks
34
// file transfers
@@ -58,13 +63,13 @@ func (bs *Bitswap) taskWorker(ctx context.Context) {
63
func (bs *Bitswap) provideWorker(ctx context.Context) {
64
for {
65
select {
61
- case blk, ok := <-bs.newBlocks:
66
+ case k, ok := <-bs.provideKeys:
67
if !ok {
63
- log.Debug("newBlocks channel closed")
68
+ log.Debug("provideKeys channel closed")
69
return
70
}
71
ctx, _ := context.WithTimeout(ctx, provideTimeout)
67
- err := bs.network.Provide(ctx, blk.Key())
72
+ err := bs.network.Provide(ctx, k)
73
if err != nil {
74
log.Error(err)
75
}
@@ -74,6 +79,51 @@ func (bs *Bitswap) provideWorker(ctx context.Context) {
79
}
80
}
81
82
+func (bs *Bitswap) provideCollector(ctx context.Context) {
83
+ defer close(bs.provideKeys)
84
+ var toprovide []u.Key
85
+ var nextKey u.Key
86
+
87
+ select {
88
+ case blk, ok := <-bs.newBlocks:
89
+ if !ok {
90
+ log.Debug("newBlocks channel closed")
91
+ return
92
+ }
93
+ nextKey = blk.Key()
94
+ case <-ctx.Done():
95
+ return
96
+ }
97
+
98
+ for {
99
+ select {
100
+ case blk, ok := <-bs.newBlocks:
101
+ if !ok {
102
+ log.Debug("newBlocks channel closed")
103
+ return
104
+ }
105
+ toprovide = append(toprovide, blk.Key())
106
+ case bs.provideKeys <- nextKey:
107
+ if len(toprovide) > 0 {
108
+ nextKey = toprovide[0]
109
+ toprovide = toprovide[1:]
110
+ } else {
111
+ select {
112
+ case blk, ok := <-bs.newBlocks:
113
+ if !ok {
114
+ return
115
+ }
116
+ nextKey = blk.Key()
117
+ case <-ctx.Done():
118
+ return
119
+ }
120
+ }
121
+ case <-ctx.Done():
122
+ return
123
+ }
124
+ }
125
+}
126
+
127
// TODO ensure only one active request per key
128
func (bs *Bitswap) clientWorker(parent context.Context) {
129
defer log.Info("bitswap client worker shutting down...")