coreapi: dht: use shared set in provideKeysRec

License: MIT
Signed-off-by: Łukasz Magiera <magik6k@gmail.com>

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/QmNr4E8z9bGTztvHJktp7uQaMdx9p3r9Asrq6eYk7iCh4a/go-merkledag"
 	offline "gx/ipfs/QmPuLWvxK1vg6ckKUpT53Dow9VLCcQGdL5Trwxa8PTLp7r/go-ipfs-exchange-offline"
@@ -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 {