remove context from HasBlock, use bitswap process instead
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Sep 2, 2015 at 14:44 UTC
63f72a5155de4f9dbb38abd2ea87029035be73b7
10 files changed
+14
-399
blockservice/blockservice.go
+2
-2
@@ -47,7 +47,7 @@ func (s *BlockService) AddBlock(b *blocks.Block) (key.Key, error) {
47
if err != nil {
48
return k, err
49
}
50
- if err := s.Exchange.HasBlock(context.TODO(), b); err != nil {
50
+ if err := s.Exchange.HasBlock(b); err != nil {
51
return "", errors.New("blockservice is closed")
52
}
53
return k, nil
@@ -61,7 +61,7 @@ func (s *BlockService) AddBlocks(bs []*blocks.Block) ([]key.Key, error) {
61
62
var ks []key.Key
63
for _, b := range bs {
64
- if err := s.Exchange.HasBlock(context.TODO(), b); err != nil {
64
+ if err := s.Exchange.HasBlock(b); err != nil {
65
return nil, errors.New("blockservice is closed")
66
}
67
ks = append(ks, b.Key())
blockservice/worker/bench/main.go
deleted
-91
@@ -1,91 +0,0 @@
1
-/*
2
-Benchmark github.com/ipfs/go-ipfs/blockservice/worker.
3
-
4
-Loop over a range of workers and buffer sizes and measure the time it
5
-per block-transfer operation for each value. Run with:
6
-
7
- $ go run "${GOPATH}/src/github.com/ipfs/go-ipfs/blockservice/worker/bench/main.go"
8
-*/
9
-
10
-package main
11
-
12
-import (
13
- "log"
14
- "math"
15
- "testing"
16
- "time"
17
-
18
- ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
19
- ds_sync "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
20
- blocks "github.com/ipfs/go-ipfs/blocks"
21
- blockstore "github.com/ipfs/go-ipfs/blocks/blockstore"
22
- worker "github.com/ipfs/go-ipfs/blockservice/worker"
23
- "github.com/ipfs/go-ipfs/exchange/offline"
24
- "github.com/ipfs/go-ipfs/thirdparty/delay"
25
- "github.com/ipfs/go-ipfs/util/datastore2"
26
-)
27
-
28
-const kEstRoutingDelay = time.Second
29
-
30
-const kBlocksPerOp = 100
31
-
32
-func main() {
33
- var bestConfig worker.Config
34
- var quickestNsPerOp int64 = math.MaxInt64
35
- for NumWorkers := 1; NumWorkers < 10; NumWorkers++ {
36
- for ClientBufferSize := 0; ClientBufferSize < 10; ClientBufferSize++ {
37
- for WorkerBufferSize := 0; WorkerBufferSize < 10; WorkerBufferSize++ {
38
- c := worker.Config{
39
- NumWorkers: NumWorkers,
40
- ClientBufferSize: ClientBufferSize,
41
- WorkerBufferSize: WorkerBufferSize,
42
- }
43
- result := testing.Benchmark(BenchmarkWithConfig(c))
44
- if result.NsPerOp() < quickestNsPerOp {
45
- bestConfig = c
46
- quickestNsPerOp = result.NsPerOp()
47
- }
48
- log.Printf("benched %+v \t result: %+v", c, result)
49
- }
50
- }
51
- }
52
- log.Println(bestConfig)
53
-}
54
-
55
-func BenchmarkWithConfig(c worker.Config) func(b *testing.B) {
56
- return func(b *testing.B) {
57
-
58
- routingDelay := delay.Fixed(0) // during setup
59
-
60
- dstore := ds_sync.MutexWrap(datastore2.WithDelay(ds.NewMapDatastore(), routingDelay))
61
- bstore := blockstore.NewBlockstore(dstore)
62
- var testdata []*blocks.Block
63
- var i int64
64
- for i = 0; i < kBlocksPerOp; i++ {
65
- testdata = append(testdata, blocks.NewBlock([]byte(string(i))))
66
- }
67
- b.ResetTimer()
68
- b.SetBytes(kBlocksPerOp)
69
- for i := 0; i < b.N; i++ {
70
-
71
- b.StopTimer()
72
- w := worker.NewWorker(offline.Exchange(bstore), c)
73
- b.StartTimer()
74
-
75
- prev := routingDelay.Set(kEstRoutingDelay) // during measured section
76
-
77
- for _, block := range testdata {
78
- if err := w.HasBlock(block); err != nil {
79
- b.Fatal(err)
80
- }
81
- }
82
-
83
- routingDelay.Set(prev) // to hasten the unmeasured close period
84
-
85
- b.StopTimer()
86
- w.Close()
87
- b.StartTimer()
88
-
89
- }
90
- }
91
-}
blockservice/worker/bench_worker_test.go
deleted
-42
@@ -1,42 +0,0 @@
1
-package worker
2
-
3
-import (
4
- "testing"
5
-
6
- ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
7
- dssync "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
8
- blocks "github.com/ipfs/go-ipfs/blocks"
9
- blockstore "github.com/ipfs/go-ipfs/blocks/blockstore"
10
- "github.com/ipfs/go-ipfs/exchange/offline"
11
-)
12
-
13
-func BenchmarkHandle10KBlocks(b *testing.B) {
14
- bstore := blockstore.NewBlockstore(dssync.MutexWrap(ds.NewMapDatastore()))
15
- var testdata []*blocks.Block
16
- for i := 0; i < 10000; i++ {
17
- testdata = append(testdata, blocks.NewBlock([]byte(string(i))))
18
- }
19
- b.ResetTimer()
20
- b.SetBytes(10000)
21
- for i := 0; i < b.N; i++ {
22
-
23
- b.StopTimer()
24
- w := NewWorker(offline.Exchange(bstore), Config{
25
- NumWorkers: 1,
26
- ClientBufferSize: 0,
27
- WorkerBufferSize: 0,
28
- })
29
- b.StartTimer()
30
-
31
- for _, block := range testdata {
32
- if err := w.HasBlock(block); err != nil {
33
- b.Fatal(err)
34
- }
35
- }
36
-
37
- b.StopTimer()
38
- w.Close()
39
- b.StartTimer()
40
-
41
- }
42
-}
blockservice/worker/worker.go
deleted
-184
@@ -1,184 +0,0 @@
1
-// TODO FIXME name me
2
-package worker
3
-
4
-import (
5
- "container/list"
6
- "errors"
7
- "time"
8
-
9
- process "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
10
- procctx "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
11
- ratelimit "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/ratelimit"
12
- blocks "github.com/ipfs/go-ipfs/blocks"
13
- key "github.com/ipfs/go-ipfs/blocks/key"
14
- exchange "github.com/ipfs/go-ipfs/exchange"
15
-
16
- logging "github.com/ipfs/go-ipfs/vendor/go-log-v1.0.0"
17
-)
18
-
19
-var log = logging.Logger("blockservice")
20
-
21
-var DefaultConfig = Config{
22
- NumWorkers: 1,
23
- ClientBufferSize: 0,
24
- WorkerBufferSize: 0,
25
-}
26
-
27
-type Config struct {
28
- // NumWorkers sets the number of background workers that provide blocks to
29
- // the exchange.
30
- NumWorkers int
31
-
32
- // ClientBufferSize allows clients of HasBlock to send up to
33
- // |ClientBufferSize| blocks without blocking.
34
- ClientBufferSize int
35
-
36
- // WorkerBufferSize can be used in conjunction with NumWorkers to reduce
37
- // communication-coordination within the worker.
38
- WorkerBufferSize int
39
-}
40
-
41
-// TODO FIXME name me
42
-type Worker struct {
43
- // added accepts blocks from client
44
- added chan *blocks.Block
45
- exchange exchange.Interface
46
-
47
- // workQueue is owned by the client worker
48
- // process manages life-cycle
49
- process process.Process
50
-}
51
-
52
-func NewWorker(e exchange.Interface, c Config) *Worker {
53
- if c.NumWorkers < 1 {
54
- c.NumWorkers = 1 // provide a sane default
55
- }
56
- w := &Worker{
57
- exchange: e,
58
- added: make(chan *blocks.Block, c.ClientBufferSize),
59
- process: process.WithParent(process.Background()), // internal management
60
- }
61
- w.start(c)
62
- return w
63
-}
64
-
65
-func (w *Worker) HasBlock(b *blocks.Block) error {
66
- select {
67
- case <-w.process.Closed():
68
- return errors.New("blockservice worker is closed")
69
- case w.added <- b:
70
- return nil
71
- }
72
-}
73
-
74
-func (w *Worker) Close() error {
75
- log.Debug("blockservice provide worker is shutting down...")
76
- return w.process.Close()
77
-}
78
-
79
-func (w *Worker) start(c Config) {
80
-
81
- workerChan := make(chan *blocks.Block, c.WorkerBufferSize)
82
-
83
- // clientWorker handles incoming blocks from |w.added| and sends to
84
- // |workerChan|. This will never block the client.
85
- w.process.Go(func(proc process.Process) {
86
- defer close(workerChan)
87
-
88
- var workQueue BlockList
89
- debugInfo := time.NewTicker(5 * time.Second)
90
- defer debugInfo.Stop()
91
- for {
92
-
93
- // take advantage of the fact that sending on nil channel always
94
- // blocks so that a message is only sent if a block exists
95
- sendToWorker := workerChan
96
- nextBlock := workQueue.Pop()
97
- if nextBlock == nil {
98
- sendToWorker = nil
99
- }
100
-
101
- select {
102
-
103
- // if worker is ready and there's a block to process, send the
104
- // block
105
- case sendToWorker <- nextBlock:
106
- case <-debugInfo.C:
107
- if workQueue.Len() > 0 {
108
- log.Debugf("%d blocks in blockservice provide queue...", workQueue.Len())
109
- }
110
- case block := <-w.added:
111
- if nextBlock != nil {
112
- workQueue.Push(nextBlock) // missed the chance to send it
113
- }
114
- // if the client sends another block, add it to the queue.
115
- workQueue.Push(block)
116
- case <-proc.Closing():
117
- return
118
- }
119
- }
120
- })
121
-
122
- // reads from |workerChan| until w.process closes
123
- limiter := ratelimit.NewRateLimiter(w.process, c.NumWorkers)
124
- limiter.Go(func(proc process.Process) {
125
- ctx := procctx.OnClosingContext(proc) // shut down in-progress HasBlock when time to die
126
- for {
127
- select {
128
- case <-proc.Closing():
129
- return
130
- case block, ok := <-workerChan:
131
- if !ok {
132
- return
133
- }
134
- limiter.LimitedGo(func(proc process.Process) {
135
- if err := w.exchange.HasBlock(ctx, block); err != nil {
136
- log.Infof("blockservice worker error: %s", err)
137
- }
138
- })
139
- }
140
- }
141
- })
142
-}
143
-
144
-type BlockList struct {
145
- list list.List
146
- uniques map[key.Key]*list.Element
147
-}
148
-
149
-func (s *BlockList) PushFront(b *blocks.Block) {
150
- if s.uniques == nil {
151
- s.uniques = make(map[key.Key]*list.Element)
152
- }
153
- _, ok := s.uniques[b.Key()]
154
- if !ok {
155
- e := s.list.PushFront(b)
156
- s.uniques[b.Key()] = e
157
- }
158
-}
159
-
160
-func (s *BlockList) Push(b *blocks.Block) {
161
- if s.uniques == nil {
162
- s.uniques = make(map[key.Key]*list.Element)
163
- }
164
- _, ok := s.uniques[b.Key()]
165
- if !ok {
166
- e := s.list.PushBack(b)
167
- s.uniques[b.Key()] = e
168
- }
169
-}
170
-
171
-func (s *BlockList) Pop() *blocks.Block {
172
- if s.list.Len() == 0 {
173
- return nil
174
- }
175
- e := s.list.Front()
176
- s.list.Remove(e)
177
- b := e.Value.(*blocks.Block)
178
- delete(s.uniques, b.Key())
179
- return b
180
-}
181
-
182
-func (s *BlockList) Len() int {
183
- return s.list.Len()
184
-}
blockservice/worker/worker_test.go
deleted
-63
@@ -1,63 +0,0 @@
1
-package worker
2
-
3
-import (
4
- blocks "github.com/ipfs/go-ipfs/blocks"
5
- "testing"
6
-)
7
-
8
-func TestStartClose(t *testing.T) {
9
- numRuns := 50
10
- if testing.Short() {
11
- numRuns = 5
12
- }
13
- for i := 0; i < numRuns; i++ {
14
- w := NewWorker(nil, DefaultConfig)
15
- w.Close()
16
- }
17
-}
18
-
19
-func TestQueueDeduplication(t *testing.T) {
20
- numUniqBlocks := 5 // arbitrary
21
-
22
- var firstBatch []*blocks.Block
23
- for i := 0; i < numUniqBlocks; i++ {
24
- firstBatch = append(firstBatch, blockFromInt(i))
25
- }
26
-
27
- // to get different pointer values and prevent the implementation from
28
- // cheating. The impl must check equality using Key.
29
- var secondBatch []*blocks.Block
30
- for i := 0; i < numUniqBlocks; i++ {
31
- secondBatch = append(secondBatch, blockFromInt(i))
32
- }
33
- var workQueue BlockList
34
-
35
- for _, b := range append(firstBatch, secondBatch...) {
36
- workQueue.Push(b)
37
- }
38
- for i := 0; i < numUniqBlocks; i++ {
39
- b := workQueue.Pop()
40
- if b.Key() != firstBatch[i].Key() {
41
- t.Fatal("list is not FIFO")
42
- }
43
- }
44
- if b := workQueue.Pop(); b != nil {
45
- t.Fatal("the workQueue did not de-duplicate the blocks")
46
- }
47
-}
48
-
49
-func TestPushPopPushPop(t *testing.T) {
50
- var workQueue BlockList
51
- orig := blockFromInt(1)
52
- dup := blockFromInt(1)
53
- workQueue.PushFront(orig)
54
- workQueue.Pop()
55
- workQueue.Push(dup)
56
- if workQueue.Len() != 1 {
57
- t.Fatal("the block list's internal state is corrupt")
58
- }
59
-}
60
-
61
-func blockFromInt(i int) *blocks.Block {
62
- return blocks.NewBlock([]byte(string(i)))
63
-}
exchange/bitswap/bitswap.go
+4
-6
@@ -228,7 +228,7 @@ func (bs *Bitswap) CancelWants(ks []key.Key) {
228
229
// HasBlock announces the existance of a block to this bitswap service. The
230
// service will potentially notify its peers.
231
-func (bs *Bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
231
+func (bs *Bitswap) HasBlock(blk *blocks.Block) error {
232
select {
233
case <-bs.process.Closing():
234
return errors.New("bitswap is closed")
@@ -246,8 +246,8 @@ func (bs *Bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
246
select {
247
case bs.newBlocks <- blk:
248
// send block off to be reprovided
249
- case <-ctx.Done():
250
- return ctx.Err()
249
+ case <-bs.process.Closing():
250
+ return bs.process.Close()
251
}
252
return nil
253
}
@@ -328,9 +328,7 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
328
log.Event(ctx, "Bitswap.GetBlockRequest.End", &k)
329
330
log.Debugf("got block %s from %s", b, p)
331
- hasBlockCtx, cancel := context.WithTimeout(ctx, hasBlockTimeout)
332
- defer cancel()
333
- if err := bs.HasBlock(hasBlockCtx, b); err != nil {
331
+ if err := bs.HasBlock(b); err != nil {
332
log.Warningf("ReceiveMessage HasBlock error: %s", err)
333
}
334
}(block)
exchange/bitswap/bitswap_test.go
+4
-7
@@ -70,7 +70,7 @@ func TestGetBlockFromPeerAfterPeerAnnounces(t *testing.T) {
70
hasBlock := peers[0]
71
defer hasBlock.Exchange.Close()
72
73
- if err := hasBlock.Exchange.HasBlock(context.Background(), block); err != nil {
73
+ if err := hasBlock.Exchange.HasBlock(block); err != nil {
74
t.Fatal(err)
75
}
76
@@ -162,7 +162,7 @@ func PerformDistributionTest(t *testing.T, numInstances, numBlocks int) {
162
first := instances[0]
163
for _, b := range blocks {
164
blkeys = append(blkeys, b.Key())
165
- first.Exchange.HasBlock(ctx, b)
165
+ first.Exchange.HasBlock(b)
166
}
167
168
t.Log("Distribute!")
@@ -224,7 +224,6 @@ func TestSendToWantingPeer(t *testing.T) {
224
t.Logf("Session %v\n", peerA.Peer)
225
t.Logf("Session %v\n", peerB.Peer)
226
227
- timeout := time.Second
227
waitTime := time.Second * 5
228
229
alpha := bg.Next()
@@ -237,9 +236,7 @@ func TestSendToWantingPeer(t *testing.T) {
236
}
237
238
// peerB announces to the network that he has block alpha
240
- ctx, cancel = context.WithTimeout(context.Background(), timeout)
241
- defer cancel()
242
- err = peerB.Exchange.HasBlock(ctx, alpha)
239
+ err = peerB.Exchange.HasBlock(alpha)
240
if err != nil {
241
t.Fatal(err)
242
}
@@ -266,7 +263,7 @@ func TestBasicBitswap(t *testing.T) {
263
264
instances := sg.Instances(2)
265
blocks := bg.Blocks(1)
269
- err := instances[0].Exchange.HasBlock(context.Background(), blocks[0])
266
+ err := instances[0].Exchange.HasBlock(blocks[0])
267
if err != nil {
268
t.Fatal(err)
269
}
exchange/interface.go
+1
-1
@@ -19,7 +19,7 @@ type Interface interface {
19
20
// TODO Should callers be concerned with whether the block was made
21
// available on the network?
22
- HasBlock(context.Context, *blocks.Block) error
22
+ HasBlock(*blocks.Block) error
23
24
io.Closer
25
}
exchange/offline/offline.go
+1
-1
@@ -28,7 +28,7 @@ func (e *offlineExchange) GetBlock(_ context.Context, k key.Key) (*blocks.Block,
28
}
29
30
// HasBlock always returns nil.
31
-func (e *offlineExchange) HasBlock(_ context.Context, b *blocks.Block) error {
31
+func (e *offlineExchange) HasBlock(b *blocks.Block) error {
32
return e.bs.Put(b)
33
}
34
exchange/offline/offline_test.go
+2
-2
@@ -26,7 +26,7 @@ func TestHasBlockReturnsNil(t *testing.T) {
26
ex := Exchange(store)
27
block := blocks.NewBlock([]byte("data"))
28
29
- err := ex.HasBlock(context.Background(), block)
29
+ err := ex.HasBlock(block)
30
if err != nil {
31
t.Fail()
32
}
@@ -44,7 +44,7 @@ func TestGetBlocks(t *testing.T) {
44
expected := g.Blocks(2)
45
46
for _, b := range expected {
47
- if err := ex.HasBlock(context.Background(), b); err != nil {
47
+ if err := ex.HasBlock(b); err != nil {
48
t.Fail()
49
}
50
}