support bitswap configurability (#8268)

* feat: extract Bitswap fx initialization to its own file
* chore: bump go-bitswap dependency
* feat: bump go-ipfs-config dependency and utilize the new Internal.Bitswap configuration options. Add documentation around the new OptionalInteger config type as well as the Internal.Bitswap options.
* docs(docs/config.md): move the table of contents towards the top of the document and update it

Co-authored-by: Petar Maymounkov <[email protected]>
Co-authored-by: Marcin Rataj <[email protected]>
Co-authored-by: Gus Eggert <[email protected]>

core/node/bitswap.go

@@ -0,0 +1,52 @@
+package node
+
+import (
+	"context"
+
+	"github.com/ipfs/go-bitswap"
+	"github.com/ipfs/go-bitswap/network"
+	blockstore "github.com/ipfs/go-ipfs-blockstore"
+	config "github.com/ipfs/go-ipfs-config"
+	exchange "github.com/ipfs/go-ipfs-exchange-interface"
+	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/routing"
+	"go.uber.org/fx"
+
+	"github.com/ipfs/go-ipfs/core/node/helpers"
+)
+
+const (
+	// Docs: https://github.com/ipfs/go-ipfs/blob/master/docs/config.md#internalbitswap
+	DefaultEngineBlockstoreWorkerCount = 128
+	DefaultTaskWorkerCount             = 8
+	DefaultEngineTaskWorkerCount       = 8
+	DefaultMaxOutstandingBytesPerPeer  = 1 << 20
+)
+
+// OnlineExchange creates new LibP2P backed block exchange (BitSwap)
+func OnlineExchange(cfg *config.Config, provide bool) interface{} {
+	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt routing.Routing, bs blockstore.GCBlockstore) exchange.Interface {
+		bitswapNetwork := network.NewFromIpfsHost(host, rt)
+
+		var internalBsCfg config.InternalBitswap
+		if cfg.Internal.Bitswap != nil {
+			internalBsCfg = *cfg.Internal.Bitswap
+		}
+
+		opts := []bitswap.Option{
+			bitswap.ProvideEnabled(provide),
+			bitswap.EngineBlockstoreWorkerCount(int(internalBsCfg.EngineBlockstoreWorkerCount.WithDefault(DefaultEngineBlockstoreWorkerCount))),
+			bitswap.TaskWorkerCount(int(internalBsCfg.TaskWorkerCount.WithDefault(DefaultTaskWorkerCount))),
+			bitswap.EngineTaskWorkerCount(int(internalBsCfg.EngineTaskWorkerCount.WithDefault(DefaultEngineTaskWorkerCount))),
+			bitswap.MaxOutstandingBytesPerPeer(int(internalBsCfg.MaxOutstandingBytesPerPeer.WithDefault(DefaultMaxOutstandingBytesPerPeer))),
+		}
+		exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, bs, opts...)
+		lc.Append(fx.Hook{
+			OnStop: func(ctx context.Context) error {
+				return exch.Close()
+			},
+		})
+		return exch
+
+	}
+}

core/node/core.go

@@ -3,8 +3,7 @@ package node
 import (
 	"context"
 	"fmt"
-	"github.com/ipfs/go-bitswap"
-	"github.com/ipfs/go-bitswap/network"
+
 	"github.com/ipfs/go-blockservice"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
@@ -24,8 +23,6 @@ import (
 	"github.com/ipld/go-ipld-prime"
 	basicnode "github.com/ipld/go-ipld-prime/node/basic"
 	"github.com/ipld/go-ipld-prime/schema"
-	"github.com/libp2p/go-libp2p-core/host"
-	"github.com/libp2p/go-libp2p-core/routing"
 	"go.uber.org/fx"
 
 	"github.com/ipfs/go-ipfs/core/node/helpers"
@@ -111,21 +108,6 @@ func Dag(bs blockservice.BlockService) format.DAGService {
 	return merkledag.NewDAGService(bs)
 }
 
-// OnlineExchange creates new LibP2P backed block exchange (BitSwap)
-func OnlineExchange(provide bool) interface{} {
-	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt routing.Routing, bs blockstore.GCBlockstore) exchange.Interface {
-		bitswapNetwork := network.NewFromIpfsHost(host, rt)
-		exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, bs, bitswap.ProvideEnabled(provide))
-		lc.Append(fx.Hook{
-			OnStop: func(ctx context.Context) error {
-				return exch.Close()
-			},
-		})
-		return exch
-
-	}
-}
-
 // Files loads persisted MFS root
 func Files(mctx helpers.MetricsCtx, lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, error) {
 	dsk := datastore.NewKey("/local/filesroot")

core/node/groups.go

@@ -262,7 +262,7 @@ func Online(bcfg *BuildCfg, cfg *config.Config) fx.Option {
 	shouldBitswapProvide := !cfg.Experimental.StrategicProviding
 
 	return fx.Options(
-		fx.Provide(OnlineExchange(shouldBitswapProvide)),
+		fx.Provide(OnlineExchange(cfg, shouldBitswapProvide)),
 		maybeProvide(Graphsync, cfg.Experimental.GraphsyncEnabled),
 		fx.Provide(DNSResolver),
 		fx.Provide(Namesys(ipnsCacheSize)),