coreapi: dht: use shared set in provideKeysRec

License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
ipfs · Aug 15, 2018 · dc7f683 · dc7f683
1 parent 274f308
commit dc7f683
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 14 deletions.
diff --git a/core/coreapi/dht.go b/core/coreapi/dht.go
@@ -7,6 +7,7 @@ import (
 
 	coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
 	caopts "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
+	"github.com/ipfs/go-ipfs/thirdparty/streaming-cid-set"
 
 	dag "gx/ipfs/QmQzSpSjkdGHW6WFBhUG6P3t9K8yv7iucucT1cQaqJ6tgd/go-merkledag"
 	routing "gx/ipfs/QmSD6bSPcXaaR7LpQHjytLWQD7DrCsb415CWfpbd9Szemb/go-libp2p-routing"
@@ -98,25 +99,33 @@ func provideKeys(ctx context.Context, r routing.IpfsRouting, cids []*cid.Cid) er
 }
 
 func provideKeysRec(ctx context.Context, r routing.IpfsRouting, bs blockstore.Blockstore, cids []*cid.Cid) error {
-	provided := cid.NewSet()
-	for _, c := range cids {
-		dserv := dag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
+	provided := streamingset.NewStreamingSet()
 
-		err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, provided.Visit)
-		if err != nil {
-			return err
-		}
-	}
+	errCh := make(chan error)
+	go func() {
+		for _, c := range cids {
+			dserv := dag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
 
-	for _, k := range provided.Keys() {
-		err := r.Provide(ctx, k, true)
-		if err != nil {
+			err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, provided.Visitor(ctx))
+			if err != nil {
+				errCh <- err
+			}
+		}
+	}()
+
+	for {
+		select {
+		case k := <-provided.New:
+			err := r.Provide(ctx, k, true)
+			if err != nil {
+				return err
+			}
+		case err := <-errCh:
 			return err
+		case <-ctx.Done():
+			return ctx.Err()
 		}
-		provided.Add(k)
 	}
-
-	return nil
 }
 
 func (api *DhtAPI) core() coreiface.CoreAPI {

diff --git a/core/coreapi/unixfs_test.go b/core/coreapi/unixfs_test.go
@@ -23,7 +23,10 @@ import (
 	mdag "gx/ipfs/QmQzSpSjkdGHW6WFBhUG6P3t9K8yv7iucucT1cQaqJ6tgd/go-merkledag"
 	config "gx/ipfs/QmRwCaRYotCqXsVZAXwWhEJ8A74iAaKnY7MUe6sDgFjrE5/go-ipfs-config"
 	mocknet "gx/ipfs/QmUDzeFgYrRmHL2hUB6NZmqcBVQtUzETwmFRUc9onfSSHr/go-libp2p/p2p/net/mock"
+	datastore "gx/ipfs/QmVG5gxteQNEMhrS8prJSmU2C9rebtFuTd3SYZ5kE3YZ5k/go-datastore"
+	syncds "gx/ipfs/QmVG5gxteQNEMhrS8prJSmU2C9rebtFuTd3SYZ5kE3YZ5k/go-datastore/sync"
 	cbor "gx/ipfs/QmVhWKoxHMJNbTMEPhqLAjKg1Y65j9tvWNecYWAHwyguAZ/go-ipld-cbor"
+	unixfs "gx/ipfs/QmWv8MYwgPK4zXYv1et1snWJ6FWGqaL6xY2y9X1bRSKBxk/go-unixfs"
 	pstore "gx/ipfs/QmYLXCWN2myozZpx8Wx4UjrRuQuhY3YtWoMi6SHaXii6aM/go-libp2p-peerstore"
 	peer "gx/ipfs/QmcZSzKEM5yDfpZbeEEZaVmaZ1zXm6JWTbrQZSB8hCVPzk/go-libp2p-peer"
 )