cleanup, use a workgroup over channels
Jeromy committed
Dec 1, 2014 at 02:15 UTC
3a6b6c697482fc5160584342a28f3c5d80c7d075
3 files changed
+14
-9
exchange/bitswap/bitswap.go
+6
-5
@@ -3,6 +3,7 @@
3
package bitswap
4
5
import (
6
+ "sync"
7
"time"
8
9
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
@@ -180,8 +181,9 @@ func (bs *bitswap) sendWantListTo(ctx context.Context, peers <-chan peer.Peer) e
181
}
182
183
func (bs *bitswap) sendWantlistToProviders(ctx context.Context, ks []u.Key) {
183
- done := make(chan struct{})
184
+ wg := sync.WaitGroup{}
185
for _, k := range ks {
186
+ wg.Add(1)
187
go func(k u.Key) {
188
providers := bs.routing.FindProvidersAsync(ctx, k, maxProvidersPerRequest)
189
@@ -189,12 +191,10 @@ func (bs *bitswap) sendWantlistToProviders(ctx context.Context, ks []u.Key) {
191
if err != nil {
192
log.Errorf("error sending wantlist: %s", err)
193
}
192
- done <- struct{}{}
194
+ wg.Done()
195
}(k)
196
}
195
- for _ = range ks {
196
- <-done
197
- }
197
+ wg.Wait()
198
}
199
200
// TODO ensure only one active request per key
@@ -255,6 +255,7 @@ func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
255
return bs.routing.Provide(ctx, blk.Key())
256
}
257
258
+// receiveBlock handles storing the block in the blockstore and calling HasBlock
259
func (bs *bitswap) receiveBlock(ctx context.Context, block *blocks.Block) {
260
// TODO verify blocks?
261
if err := bs.blockstore.Put(block); err != nil {
merkledag/merkledag.go
+6
-3
@@ -28,7 +28,10 @@ type DAGService interface {
28
AddRecursive(*Node) error
29
Get(u.Key) (*Node, error)
30
Remove(*Node) error
31
- GetKeysAsync(context.Context, *Node) <-chan *Node
31
+
32
+ // GetDAG returns, in order, all the single leve child
33
+ // nodes of the passed in node.
34
+ GetDAG(context.Context, *Node) <-chan *Node
35
}
36
37
func NewDAGService(bs *bserv.BlockService) DAGService {
@@ -298,10 +301,10 @@ func FindLink(n *Node, k u.Key, found []*Node) (int, error) {
301
return -1, u.ErrNotFound
302
}
303
301
-// GetKeysAsync will fill out all of the links of the given Node.
304
+// GetDAG will fill out all of the links of the given Node.
305
// It returns a channel of nodes, which the caller can receive
306
// all the child nodes of 'root' on, in proper order.
304
-func (ds *dagService) GetKeysAsync(ctx context.Context, root *Node) <-chan *Node {
307
+func (ds *dagService) GetDAG(ctx context.Context, root *Node) <-chan *Node {
308
sig := make(chan *Node)
309
go func() {
310
var keys []u.Key
unixfs/io/dagreader.go
+2
-1
@@ -40,7 +40,7 @@ func NewDagReader(n *mdag.Node, serv mdag.DAGService) (io.Reader, error) {
40
case ftpb.Data_File:
41
var fetchChan <-chan *mdag.Node
42
if serv != nil {
43
- fetchChan = serv.GetKeysAsync(context.TODO(), n)
43
+ fetchChan = serv.GetDAG(context.TODO(), n)
44
}
45
return &DagReader{
46
node: n,
@@ -62,6 +62,7 @@ func (dr *DagReader) precalcNextBuf() error {
62
var nxt *mdag.Node
63
var ok bool
64
65
+ // TODO: require non-nil dagservice, use offline bitswap exchange
66
if dr.serv == nil {
67
// Only used when fetchChan is nil,
68
// which only happens when passed in a nil dagservice