blockservice: async HasBlock with ratelimit
Juan Batiz-Benet committed
Jan 6, 2015 at 08:45 UTC
238035c298c6f7b4e3b9ba311ec365e54b5b277b
2 files changed
+38
-10
blockservice/blockservice.go
+33
-9
@@ -8,6 +8,8 @@ import (
8
"fmt"
9
10
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
11
+ process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
12
+ procrl "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/ratelimit"
13
blocks "github.com/jbenet/go-ipfs/blocks"
14
"github.com/jbenet/go-ipfs/blocks/blockstore"
15
exchange "github.com/jbenet/go-ipfs/exchange"
@@ -17,6 +19,9 @@ import (
19
var log = u.Logger("blockservice")
20
var ErrNotFound = errors.New("blockservice: key not found")
21
22
+// MaxExchangeAddWorkers rate limits the number of exchange workers
23
+var MaxExchangeAddWorkers = 100
24
+
25
// BlockService is a hybrid block datastore. It stores data in a local
26
// datastore and may retrieve data from a remote Exchange.
27
// It uses an internal `datastore.Datastore` instance to store values.
@@ -24,6 +29,9 @@ type BlockService struct {
29
// TODO don't expose underlying impl details
30
Blockstore blockstore.Blockstore
31
Exchange exchange.Interface
32
+
33
+ rateLimiter *procrl.RateLimiter
34
+ exchangeAdd chan blocks.Block
35
}
36
37
// NewBlockService creates a BlockService with given datastore instance.
@@ -34,7 +42,17 @@ func New(bs blockstore.Blockstore, rem exchange.Interface) (*BlockService, error
42
if rem == nil {
43
log.Warning("blockservice running in local (offline) mode.")
44
}
37
- return &BlockService{Blockstore: bs, Exchange: rem}, nil
45
+
46
+ // exchangeAdd is a channel for async workers to add to the exchange.
47
+ // 100 blocks buffer. not clear what this number should be
48
+ exchangeAdd := make(chan blocks.Block, 100)
49
+
50
+ return &BlockService{
51
+ Blockstore: bs,
52
+ Exchange: rem,
53
+ exchangeAdd: exchangeAdd,
54
+ rateLimiter: procrl.NewRateLimiter(process.Background(), MaxExchangeAddWorkers),
55
+ }, nil
56
}
57
58
// AddBlock adds a particular block to the service, Putting it into the datastore.
@@ -46,15 +64,21 @@ func (s *BlockService) AddBlock(b *blocks.Block) (u.Key, error) {
64
return k, err
65
}
66
49
- // TODO this operation rate-limits blockservice operations, we should
50
- // consider moving this to an sync process.
67
+ // this operation rate-limits blockservice operations, so it is
68
+ // now an async process.
69
if s.Exchange != nil {
52
- ctx := context.TODO()
53
- if err := s.Exchange.HasBlock(ctx, b); err != nil {
54
- // suppress error, as the client shouldn't care about bitswap.
55
- // the client only cares about the blockstore.Put.
56
- log.Errorf("Exchange.HasBlock error: %s", err)
57
- }
70
+
71
+ // LimitedGo will spawn a goroutine but provide proper backpressure.
72
+ // it will not spawn the goroutine until the ratelimiter's work load
73
+ // is under the threshold.
74
+ s.rateLimiter.LimitedGo(func(worker process.Process) {
75
+ ctx := context.TODO()
76
+ if err := s.Exchange.HasBlock(ctx, b); err != nil {
77
+ // suppress error, as the client shouldn't care about bitswap.
78
+ // the client only cares about the blockstore.Put.
79
+ log.Errorf("Exchange.HasBlock error: %s", err)
80
+ }
81
+ })
82
}
83
return k, nil
84
}
epictest/core.go
+5
-1
@@ -74,7 +74,11 @@ func makeCore(ctx context.Context, rf RepoFactory) (*core, error) {
74
return nil, err
75
}
76
77
- bss := &blockservice.BlockService{repo.Blockstore(), repo.Exchange()}
77
+ bss, err := blockservice.New(repo.Blockstore(), repo.Exchange())
78
+ if err != nil {
79
+ return nil, err
80
+ }
81
+
82
dag := merkledag.NewDAGService(bss)
83
// to make sure nothing is omitted, init each individual field and assign
84
// all at once at the bottom.