[WIP] Extract provider logic from BitSwap

core/builder.go

@@ -14,6 +14,7 @@ import (
 	dag "github.com/ipfs/go-ipfs/merkledag"
 	resolver "github.com/ipfs/go-ipfs/path/resolver"
 	pin "github.com/ipfs/go-ipfs/pin"
+	providers "github.com/ipfs/go-ipfs/providers"
 	repo "github.com/ipfs/go-ipfs/repo"
 	cfg "github.com/ipfs/go-ipfs/repo/config"
 	"github.com/ipfs/go-ipfs/thirdparty/verifbs"
@@ -237,7 +238,9 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
 			return err
 		}
 	} else {
+		n.SetupOfflineRouting()
 		n.Exchange = offline.Exchange(n.Blockstore)
+		n.Providers = providers.NewProviders(n.ctx, n.Routing, nil)
 	}
 
 	n.Blocks = bserv.New(n.Blockstore, n.Exchange)

core/commands/add.go

@@ -313,7 +313,7 @@ You can now check what blocks have been created by:
 			}
 
 			// copy intermediary nodes from editor to our actual dagservice
-			_, err := fileAdder.Finalize()
+			nd, err := fileAdder.Finalize()
 			if err != nil {
 				return err
 			}
@@ -322,6 +322,12 @@ You can now check what blocks have been created by:
 				return nil
 			}
 
+			if !local {
+				if err := n.Providers.ProvideRecursive(req.Context, nd, dserv); err != nil {
+					return err
+				}
+			}
+
 			return fileAdder.PinRoot()
 		}
 

core/commands/bitswap.go

@@ -173,7 +173,6 @@ var bitswapStatCmd = &cmds.Command{
 			}
 
 			fmt.Fprintln(w, "bitswap status")
-			fmt.Fprintf(w, "\tprovides buffer: %d / %d\n", out.ProvideBufLen, bitswap.HasBlockBufferSize)
 			fmt.Fprintf(w, "\tblocks received: %d\n", out.BlocksReceived)
 			fmt.Fprintf(w, "\tblocks sent: %d\n", out.BlocksSent)
 			fmt.Fprintf(w, "\tdata received: %d\n", out.DataReceived)

core/commands/block.go

@@ -222,6 +222,11 @@ than 'sha2-256' or format to anything other than 'v0' will result in CIDv1.
 			return
 		}
 
+		if err := n.Providers.Provide(b.Cid()); err != nil {
+			log.Error("BlockPut key: '%q'", err)
+			return
+		}
+
 		err = cmds.EmitOnce(res, &BlockStat{
 			Key:  b.Cid().String(),
 			Size: len(data),

core/commands/dag/dag.go

@@ -154,7 +154,9 @@ into an object of the specified format.
 				}
 			}
 
-			return nil
+			return cids.ForEach(func(c *cid.Cid) error {
+				return n.Providers.Provide(c)
+			})
 		}
 
 		go func() {

core/commands/object/object.go

@@ -512,6 +512,12 @@ Available templates:
 			res.SetError(err, cmdkit.ErrNormal)
 			return
 		}
+
+		if err = n.Providers.Provide(node.Cid()); err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
 		res.SetOutput(&Object{Hash: node.Cid().String()})
 	},
 	Marshalers: oldcmds.MarshalerMap{
@@ -610,6 +616,11 @@ func objectPut(ctx context.Context, n *core.IpfsNode, input io.Reader, encoding
 		return nil, err
 	}
 
+	err = n.Providers.Provide(dagnode.Cid())
+	if err != nil {
+		return nil, err
+	}
+
 	return dagnode.Cid(), nil
 }
 

core/commands/object/patch.go

@@ -118,6 +118,12 @@ the limit will not be respected by the network.
 			return
 		}
 
+		err = nd.Providers.Provide(rtpb.Cid())
+		if err != nil {
+			re.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
 		cmds.EmitOnce(re, &Object{Hash: rtpb.Cid().String()})
 	},
 	Type: Object{},
@@ -189,6 +195,12 @@ Example:
 			return
 		}
 
+		err = nd.Providers.Provide(rtpb.Cid())
+		if err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
 		res.SetOutput(&Object{Hash: rtpb.Cid().String()})
 	},
 	Type: Object{},
@@ -251,6 +263,12 @@ Removes a link by the given name from root.
 
 		nc := nnode.Cid()
 
+		err = nd.Providers.Provide(nc)
+		if err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
 		res.SetOutput(&Object{Hash: nc.String()})
 	},
 	Type: Object{},
@@ -348,6 +366,12 @@ to a file containing 'bar', and returns the hash of the new object.
 
 		nc := nnode.Cid()
 
+		err = nd.Providers.Provide(nc)
+		if err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
 		res.SetOutput(&Object{Hash: nc.String()})
 	},
 	Type: Object{},

core/commands/pubsub.go

@@ -107,6 +107,10 @@ This command outputs data in the following encodings:
 					log.Error("pubsub discovery: ", err)
 					return
 				}
+				if err := n.Providers.Provide(blk.Cid()); err != nil {
+					log.Error("pubsub discovery: ", err)
+					return
+				}
 
 				connectToPubSubPeers(req.Context, n, blk.Cid())
 			}()

core/commands/tar.go

@@ -59,6 +59,12 @@ represent it.
 
 		c := node.Cid()
 
+		err = nd.Providers.Provide(c)
+		if err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
 		fi.FileName()
 		res.SetOutput(&coreunix.AddedObject{
 			Name: fi.FileName(),

core/core.go

@@ -33,6 +33,7 @@ import (
 	p2p "github.com/ipfs/go-ipfs/p2p"
 	"github.com/ipfs/go-ipfs/path/resolver"
 	pin "github.com/ipfs/go-ipfs/pin"
+	provider "github.com/ipfs/go-ipfs/providers"
 	repo "github.com/ipfs/go-ipfs/repo"
 	config "github.com/ipfs/go-ipfs/repo/config"
 	ft "github.com/ipfs/go-ipfs/unixfs"
@@ -129,6 +130,7 @@ type IpfsNode struct {
 	PeerHost     p2phost.Host        // the network host (server+client)
 	Bootstrapper io.Closer           // the periodic bootstrapper
 	Routing      routing.IpfsRouting // the routing system. recommend ipfs-dht
+	Providers    provider.Interface  // the content routing abstraction layer
 	Exchange     exchange.Interface  // the block exchange + strategy (bitswap)
 	Namesys      namesys.NameSystem  // the name system, resolves paths to hashes
 	Ping         *ping.PingService
@@ -493,8 +495,11 @@ func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost
 	// Wrap standard peer host with routing system to allow unknown peer lookups
 	n.PeerHost = rhost.Wrap(host, n.Routing)
 
+	// Wrap content routing with a buffering layer
+	n.Providers = provider.NewProviders(ctx, n.Routing, n.PeerHost)
+
 	// setup exchange service
-	bitswapNetwork := bsnet.NewFromIpfsHost(n.PeerHost, n.Routing)
+	bitswapNetwork := bsnet.NewFromIpfsHost(n.PeerHost, n.Providers)
 	n.Exchange = bitswap.New(ctx, bitswapNetwork, n.Blockstore)
 
 	size, err := n.getCacheSize()

core/corehttp/gateway_handler.go

@@ -504,6 +504,11 @@ func (i *gatewayHandler) putHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if err := i.node.Providers.Provide(newcid); err != nil {
+		webError(w, "putHandler: Could not provide newnode: ", err, http.StatusInternalServerError)
+		return
+	}
+
 	i.addUserHeaders(w) // ok, _now_ write user's headers.
 	w.Header().Set("IPFS-Hash", newcid.String())
 	http.Redirect(w, r, gopath.Join(ipfsPathPrefix, newcid.String(), newPath), http.StatusCreated)
@@ -560,6 +565,11 @@ func (i *gatewayHandler) deleteHandler(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
+		if err := i.node.Providers.Provide(c); err != nil {
+			webError(w, "Could not provide node", err, http.StatusInternalServerError)
+			return
+		}
+
 		pathpb, ok := pathNodes[j].(*dag.ProtoNode)
 		if !ok {
 			webError(w, "Cannot read non protobuf nodes through gateway", dag.ErrNotProtobuf, http.StatusBadRequest)
@@ -581,6 +591,11 @@ func (i *gatewayHandler) deleteHandler(w http.ResponseWriter, r *http.Request) {
 	// Redirect to new path
 	ncid := newnode.Cid()
 
+	if err := i.node.Providers.Provide(ncid); err != nil {
+		webError(w, "Could not provide node", err, http.StatusInternalServerError)
+		return
+	}
+
 	i.addUserHeaders(w) // ok, _now_ write user's headers.
 	w.Header().Set("IPFS-Hash", ncid.String())
 	http.Redirect(w, r, gopath.Join(ipfsPathPrefix+ncid.String(), path.Join(components[:len(components)-1])), http.StatusCreated)

exchange/bitswap/bitswap.go

@@ -16,7 +16,6 @@ import (
 	notifications "github.com/ipfs/go-ipfs/exchange/bitswap/notifications"
 
 	delay "gx/ipfs/QmRJVNatYJwTAHgdSM1Xef9QVQ1Ch3XHdmcrykjP5Y4soL/go-ipfs-delay"
-	flags "gx/ipfs/QmRMGdC6HKdLsPDABL9aXPDidrpmEHzJqFWSvshkbn9Hj8/go-ipfs-flags"
 	metrics "gx/ipfs/QmRg1gKTHzc3CZXSKzem8aR4E3TubFhbgXwfVuWnSK5CC5/go-metrics-interface"
 	process "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess"
 	procctx "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess/context"
@@ -31,35 +30,15 @@ import (
 var log = logging.Logger("bitswap")
 
 const (
-	// maxProvidersPerRequest specifies the maximum number of providers desired
-	// from the network. This value is specified because the network streams
-	// results.
-	// TODO: if a 'non-nice' strategy is implemented, consider increasing this value
-	maxProvidersPerRequest = 3
-	providerRequestTimeout = time.Second * 10
-	provideTimeout         = time.Second * 15
-	sizeBatchRequestChan   = 32
 	// kMaxPriority is the max priority as defined by the bitswap protocol
 	kMaxPriority = math.MaxInt32
 )
 
 var (
-	HasBlockBufferSize    = 256
-	provideKeysBufferSize = 2048
-	provideWorkerMax      = 512
-
 	// the 1<<18+15 is to observe old file chunks that are 1<<18 + 14 in size
 	metricsBuckets = []float64{1 << 6, 1 << 10, 1 << 14, 1 << 18, 1<<18 + 15, 1 << 22}
 )
 
-func init() {
-	if flags.LowMemMode {
-		HasBlockBufferSize = 64
-		provideKeysBufferSize = 512
-		provideWorkerMax = 16
-	}
-}
-
 var rebroadcastDelay = delay.Fixed(time.Minute)
 
 // New initializes a BitSwap instance that communicates over the provided
@@ -94,10 +73,7 @@ func New(parent context.Context, network bsnet.BitSwapNetwork,
 		notifications: notif,
 		engine:        decision.NewEngine(ctx, bstore), // TODO close the engine with Close() method
 		network:       network,
-		findKeys:      make(chan *blockRequest, sizeBatchRequestChan),
 		process:       px,
-		newBlocks:     make(chan *cid.Cid, HasBlockBufferSize),
-		provideKeys:   make(chan *cid.Cid, provideKeysBufferSize),
 		wm:            NewWantManager(ctx, network),
 		counters:      new(counters),
 
@@ -141,15 +117,6 @@ type Bitswap struct {
 	// appropriate user requests
 	notifications notifications.PubSub
 
-	// findKeys sends keys to a worker to find and connect to providers for them
-	findKeys chan *blockRequest
-	// newBlocks is a channel for newly added blocks to be provided to the
-	// network.  blocks pushed down this channel get buffered and fed to the
-	// provideKeys channel later on to avoid too much network activity
-	newBlocks chan *cid.Cid
-	// provideKeys directly feeds provide workers
-	provideKeys chan *cid.Cid
-
 	process process.Process
 
 	// Counters for various statistics
@@ -178,11 +145,6 @@ type counters struct {
 	messagesRecvd  uint64
 }
 
-type blockRequest struct {
-	Cid *cid.Cid
-	Ctx context.Context
-}
-
 // GetBlock attempts to retrieve a particular block from peers within the
 // deadline enforced by the context.
 func (bs *Bitswap) GetBlock(parent context.Context, k *cid.Cid) (blocks.Block, error) {
@@ -230,14 +192,6 @@ func (bs *Bitswap) GetBlocks(ctx context.Context, keys []*cid.Cid) (<-chan block
 
 	bs.wm.WantBlocks(ctx, keys, nil, mses)
 
-	// NB: Optimization. Assumes that providers of key[0] are likely to
-	// be able to provide for all keys. This currently holds true in most
-	// every situation. Later, this assumption may not hold as true.
-	req := &blockRequest{
-		Cid: keys[0],
-		Ctx: ctx,
-	}
-
 	remaining := cid.NewSet()
 	for _, k := range keys {
 		remaining.Add(k)
@@ -272,12 +226,13 @@ func (bs *Bitswap) GetBlocks(ctx context.Context, keys []*cid.Cid) (<-chan block
 		}
 	}()
 
-	select {
-	case bs.findKeys <- req:
-		return out, nil
-	case <-ctx.Done():
-		return nil, ctx.Err()
+	// NB: Optimization. Assumes that providers of key[0] are likely to
+	// be able to provide for all keys. This currently holds true in most
+	// every situation. Later, this assumption may not hold as true.
+	if err := bs.network.FindProviders(ctx, keys[0]); err != nil {
+		return nil, err
 	}
+	return out, nil
 }
 
 func (bs *Bitswap) getNextSessionID() uint64 {
@@ -298,6 +253,7 @@ func (bs *Bitswap) CancelWants(cids []*cid.Cid, ses uint64) {
 // HasBlock announces the existence of a block to this bitswap service. The
 // service will potentially notify its peers.
 func (bs *Bitswap) HasBlock(blk blocks.Block) error {
+	//TODO: call provide here?
 	return bs.receiveBlockFrom(blk, "")
 }
 
@@ -333,13 +289,6 @@ func (bs *Bitswap) receiveBlockFrom(blk blocks.Block, from peer.ID) error {
 	}
 
 	bs.engine.AddBlock(blk)
-
-	select {
-	case bs.newBlocks <- blk.Cid():
-		// send block off to be reprovided
-	case <-bs.process.Closing():
-		return bs.process.Close()
-	}
 	return nil
 }
 
@@ -385,6 +334,9 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
 			if err := bs.receiveBlockFrom(b, p); err != nil {
 				log.Warningf("ReceiveMessage recvBlockFrom error: %s", err)
 			}
+			if err := bs.network.Provide(ctx, b.Cid()); err != nil {
+				log.Warningf("ReceiveMessage Provide error: %s", err)
+			}
 			log.Event(ctx, "Bitswap.GetBlockRequest.End", b.Cid())
 		}(block)
 	}

exchange/bitswap/network/interface.go

@@ -65,6 +65,9 @@ type Routing interface {
 	// FindProvidersAsync returns a channel of providers for the given key
 	FindProvidersAsync(context.Context, *cid.Cid, int) <-chan peer.ID
 
+	// FindProvidersfinds providers for the given key and connects to them
+	FindProviders(ctx context.Context, c *cid.Cid) error
+
 	// Provide provides the key to the network
 	Provide(context.Context, *cid.Cid) error
 }