fix: parametrise add index parallelism (#1827)

* fix: parametrise add index parallelism

Add two new configuration parameters:
* AddIndexConcurrency for boostd that sets the number of concurrent tasks that each add index operation can be split into;
* insert-concurrency for boost-data that sets the number of concurrent inserts that each add index operation can be split into;

* change default add index concurrency to 1

* check max insert batch size

Author: ischasny

cmd/boostd/recover.go

@@ -18,6 +18,7 @@ import (
 	"github.com/filecoin-project/boost/db"
 	bdclient "github.com/filecoin-project/boost/extern/boostd-data/client"
 	"github.com/filecoin-project/boost/extern/boostd-data/model"
+	"github.com/filecoin-project/boost/node/config"
 	"github.com/filecoin-project/boost/piecedirectory"
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-commp-utils/writer"
@@ -103,6 +104,11 @@ var lidCmd = &cli.Command{
 			Usage: "",
 			Value: 4,
 		},
+		&cli.IntFlag{
+			Name:  "add-index-concurrency",
+			Usage: "the maximum number of parallel tasks that a single add index operation can be split into",
+			Value: config.DefaultAddIndexConcurrency,
+		},
 		&cli.BoolFlag{
 			Name:  "ignore-commp",
 			Usage: "whether we should ignore sanity check of local data vs chain data",
@@ -203,7 +209,7 @@ func action(cctx *cli.Context) error {
 			return fmt.Errorf("connecting to local index directory service: %w", err)
 		}
 		pr := &piecedirectory.SectorAccessorAsPieceReader{SectorAccessor: sa}
-		pd = piecedirectory.NewPieceDirectory(cl, pr, cctx.Int("add-index-throttle"))
+		pd = piecedirectory.NewPieceDirectory(cl, pr, cctx.Int("add-index-throttle"), piecedirectory.WithAddIndexConcurrency(cctx.Int("add-index-concurrency")))
 		pd.Start(ctx)
 	}
 

cmd/booster-bitswap/run.go

@@ -13,6 +13,7 @@ import (
 	bdclient "github.com/filecoin-project/boost/extern/boostd-data/client"
 	"github.com/filecoin-project/boost/extern/boostd-data/shared/tracing"
 	"github.com/filecoin-project/boost/metrics"
+	"github.com/filecoin-project/boost/node/config"
 	"github.com/filecoin-project/boost/piecedirectory"
 	lcli "github.com/filecoin-project/lotus/cli"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -67,6 +68,11 @@ var runCmd = &cli.Command{
 			Usage: "the maximum number of add index operations that can run in parallel",
 			Value: 4,
 		},
+		&cli.IntFlag{
+			Name:  "add-index-concurrency",
+			Usage: "the maximum number of parallel tasks that a single add index operation can be split into",
+			Value: config.DefaultAddIndexConcurrency,
+		},
 		&cli.StringFlag{
 			Name:  "proxy",
 			Usage: "the multiaddr of the libp2p proxy that this node connects through",
@@ -234,7 +240,8 @@ var runCmd = &cli.Command{
 		if err != nil {
 			return fmt.Errorf("starting block filter: %w", err)
 		}
-		pd := piecedirectory.NewPieceDirectory(cl, sa, cctx.Int("add-index-throttle"))
+		pd := piecedirectory.NewPieceDirectory(cl, sa, cctx.Int("add-index-throttle"),
+			piecedirectory.WithAddIndexConcurrency(cctx.Int("add-index-concurrency")))
 		remoteStore := remoteblockstore.NewRemoteBlockstore(pd, &bitswapBlockMetrics)
 		server := NewBitswapServer(remoteStore, host, multiFilter)
 

cmd/booster-http/run.go

@@ -18,6 +18,7 @@ import (
 	"github.com/filecoin-project/boost/extern/boostd-data/model"
 	"github.com/filecoin-project/boost/extern/boostd-data/shared/tracing"
 	"github.com/filecoin-project/boost/metrics"
+	"github.com/filecoin-project/boost/node/config"
 	"github.com/filecoin-project/boost/piecedirectory"
 	"github.com/filecoin-project/dagstore/mount"
 	"github.com/filecoin-project/go-address"
@@ -75,6 +76,11 @@ var runCmd = &cli.Command{
 			Usage: "the maximum number of add index operations that can run in parallel",
 			Value: 4,
 		},
+		&cli.IntFlag{
+			Name:  "add-index-concurrency",
+			Usage: "the maximum number of parallel tasks that a single add index operation can be split into",
+			Value: config.DefaultAddIndexConcurrency,
+		},
 		&cli.StringFlag{
 			Name:     "api-fullnode",
 			Usage:    "the endpoint for the full node API",
@@ -242,7 +248,8 @@ var runCmd = &cli.Command{
 		defer sa.Close()
 
 		// Create the server API
-		pd := piecedirectory.NewPieceDirectory(cl, sa, cctx.Int("add-index-throttle"))
+		pd := piecedirectory.NewPieceDirectory(cl, sa, cctx.Int("add-index-throttle"),
+			piecedirectory.WithAddIndexConcurrency(cctx.Int("add-index-concurrency")))
 
 		opts := &HttpServerOptions{
 			ServePieces:      servePieces,

extern/boostd-data/cmd/run.go

@@ -28,6 +28,10 @@ import (
 	"go.opencensus.io/tag"
 )
 
+const (
+	DefaultInsertConcurrency = 4
+)
+
 var runCmd = &cli.Command{
 	Name: "run",
 	Subcommands: []*cli.Command{
@@ -104,6 +108,11 @@ var yugabyteCmd = &cli.Command{
 			Usage:    "postgres connect string eg 'postgresql://postgres:postgres@localhost'",
 			Required: true,
 		},
+		&cli.IntFlag{
+			Name:  "insert-concurrency",
+			Usage: "the number of concurrent tasks that each add index operation is split into",
+			Value: DefaultInsertConcurrency,
+		},
 		&cli.IntFlag{
 			Name:     "CQLTimeout",
 			Usage:    "client timeout value in seconds for CQL queries",
@@ -115,9 +124,10 @@ var yugabyteCmd = &cli.Command{
 	Action: func(cctx *cli.Context) error {
 		// Create a yugabyte data service
 		settings := yugabyte.DBSettings{
-			Hosts:         cctx.StringSlice("hosts"),
-			ConnectString: cctx.String("connect-string"),
-			CQLTimeout:    cctx.Int("CQLTimeout"),
+			Hosts:             cctx.StringSlice("hosts"),
+			ConnectString:     cctx.String("connect-string"),
+			CQLTimeout:        cctx.Int("CQLTimeout"),
+			InsertConcurrency: cctx.Int("insert-concurrency"),
 		}
 
 		// One of the migrations requires a miner address. But we don't want to
@@ -223,6 +233,11 @@ var yugabyteMigrateCmd = &cli.Command{
 			Name:  "miner-address",
 			Usage: "default miner address eg f1234",
 		},
+		&cli.IntFlag{
+			Name:  "insert-concurrency",
+			Usage: "the number of concurrent tasks that each add index operation is split into",
+			Value: DefaultInsertConcurrency,
+		},
 		&cli.IntFlag{
 			Name:     "CQLTimeout",
 			Usage:    "client timeout value in seconds for CQL queries",
@@ -236,9 +251,10 @@ var yugabyteMigrateCmd = &cli.Command{
 
 		// Create a yugabyte data service
 		settings := yugabyte.DBSettings{
-			Hosts:         cctx.StringSlice("hosts"),
-			ConnectString: cctx.String("connect-string"),
-			CQLTimeout:    cctx.Int("CQLTimeout"),
+			Hosts:             cctx.StringSlice("hosts"),
+			ConnectString:     cctx.String("connect-string"),
+			CQLTimeout:        cctx.Int("CQLTimeout"),
+			InsertConcurrency: cctx.Int("insert-concurrency"),
 		}
 
 		maddr := migrations.DisabledMinerAddr
@@ -280,6 +296,11 @@ var yugabyteAddIndexCmd = &cli.Command{
 			Usage:    "filename must be same as pieceCID",
 			Required: true,
 		},
+		&cli.IntFlag{
+			Name:  "insert-concurrency",
+			Usage: "the number of concurrent tasks that each add index operation is split into",
+			Value: DefaultInsertConcurrency,
+		},
 		&cli.IntFlag{
 			Name:     "CQLTimeout",
 			Usage:    "client timeout value in seconds for CQL queries",
@@ -292,9 +313,10 @@ var yugabyteAddIndexCmd = &cli.Command{
 
 		// Create a yugabyte data service
 		settings := yugabyte.DBSettings{
-			Hosts:         cctx.StringSlice("hosts"),
-			ConnectString: cctx.String("connect-string"),
-			CQLTimeout:    cctx.Int("CQLTimeout"),
+			Hosts:             cctx.StringSlice("hosts"),
+			ConnectString:     cctx.String("connect-string"),
+			CQLTimeout:        cctx.Int("CQLTimeout"),
+			InsertConcurrency: cctx.Int("insert-concurrency"),
 		}
 
 		migrator := yugabyte.NewMigrator(settings, migrations.DisabledMinerAddr)

extern/boostd-data/yugabyte/service.go

@@ -34,7 +34,8 @@ const CqlTimeout = 60
 
 // The Cassandra driver has a 50k limit on batch statements. Keeping
 // batch size small makes sure we're under the limit.
-const InsertBatchSize = 10000
+const InsertBatchSize = 10_000
+const MaxInsertBatchSize = 50_000
 
 const InsertConcurrency = 4
 
@@ -81,6 +82,9 @@ func NewStore(settings DBSettings, migrator *Migrator, opts ...StoreOpt) *Store
 	if settings.InsertBatchSize == 0 {
 		settings.InsertBatchSize = InsertBatchSize
 	}
+	if settings.InsertBatchSize > MaxInsertBatchSize {
+		settings.InsertBatchSize = MaxInsertBatchSize
+	}
 	if settings.InsertConcurrency == 0 {
 		settings.InsertConcurrency = InsertConcurrency
 	}

node/config/def.go

@@ -23,6 +23,10 @@ const (
 // CommP and traversing a DAG with graphsync; invokes a budget on DAG depth and density.
 var MaxTraversalLinks uint64 = 32 * (1 << 20)
 
+const (
+	DefaultAddIndexConcurrency = 1
+)
+
 func init() {
 	if envMaxTraversal, err := strconv.ParseUint(os.Getenv("LOTUS_MAX_TRAVERSAL_LINKS"), 10, 64); err == nil {
 		MaxTraversalLinks = envMaxTraversal
@@ -87,6 +91,7 @@ func DefaultBoost() *Boost {
 				Enabled: false,
 			},
 			ParallelAddIndexLimit: 4,
+			AddIndexConcurrency:   DefaultAddIndexConcurrency,
 			EmbeddedServicePort:   8042,
 			ServiceApiInfo:        "",
 			ServiceRPCTimeout:     Duration(15 * time.Minute),

node/config/doc_gen.go

@@ -418,6 +418,9 @@ type LocalIndexDirectoryConfig struct {
 	// the piece from the sealing subsystem, creates an index of where each block
 	// is in the piece, and adds the index to the local index directory.
 	ParallelAddIndexLimit int
+	// AddIndexConcurrency sets the number of concurrent tasks that each add index operation is split into.
+	// This setting is usefull to better utilise bandwidth between boostd and boost-data. The default value is 4.
+	AddIndexConcurrency int
 	// The port that the embedded local index directory data service runs on.
 	// Set this value to zero to disable the embedded local index directory data service
 	// (in that case the local index directory data service must be running externally)

node/config/types.go

@@ -136,7 +136,9 @@ func NewPieceDirectory(cfg *config.Boost) func(lc fx.Lifecycle, maddr dtypes.Min
 
 		// Create the piece directory implementation
 		pdctx, cancel := context.WithCancel(context.Background())
-		pd := piecedirectory.NewPieceDirectory(store, sa, cfg.LocalIndexDirectory.ParallelAddIndexLimit)
+		pd := piecedirectory.NewPieceDirectory(store, sa,
+			cfg.LocalIndexDirectory.ParallelAddIndexLimit,
+			piecedirectory.WithAddIndexConcurrency(cfg.LocalIndexDirectory.AddIndexConcurrency))
 		lc.Append(fx.Hook{
 			OnStart: func(ctx context.Context) error {
 				err := sa.Start(ctx, log)

node/modules/piecedirectory.go

@@ -13,6 +13,7 @@ import (
 	"github.com/filecoin-project/boost/extern/boostd-data/model"
 	"github.com/filecoin-project/boost/extern/boostd-data/shared/tracing"
 	bdtypes "github.com/filecoin-project/boost/extern/boostd-data/svc/types"
+	"github.com/filecoin-project/boost/node/config"
 	"github.com/filecoin-project/boost/piecedirectory/types"
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
@@ -37,16 +38,23 @@ import (
 var log = logging.Logger("piecedirectory")
 
 const (
-	MaxCachedReaders    = 128
-	AddIndexConcurrency = 8
+	MaxCachedReaders = 128
 )
 
-type Settings struct {
+type settings struct {
 	addIndexConcurrency int
 }
 
+type Option func(*settings)
+
+func WithAddIndexConcurrency(c int) Option {
+	return func(s *settings) {
+		s.addIndexConcurrency = c
+	}
+}
+
 type PieceDirectory struct {
-	settings    *Settings
+	settings    *settings
 	store       *bdclient.Store
 	pieceReader types.PieceReader
 
@@ -60,13 +68,7 @@ type PieceDirectory struct {
 	addIdxOpByCid      sync.Map
 }
 
-func NewPieceDirectory(store *bdclient.Store, pr types.PieceReader, addIndexThrottleSize int) *PieceDirectory {
-	return NewPieceDirectoryWithSettings(store, pr, addIndexThrottleSize, &Settings{
-		addIndexConcurrency: AddIndexConcurrency,
-	})
-}
-
-func NewPieceDirectoryWithSettings(store *bdclient.Store, pr types.PieceReader, addIndexThrottleSize int, settings *Settings) *PieceDirectory {
+func NewPieceDirectory(store *bdclient.Store, pr types.PieceReader, addIndexThrottleSize int, opts ...Option) *PieceDirectory {
 	prCache := ttlcache.NewCache()
 	_ = prCache.SetTTL(30 * time.Second)
 	prCache.SetCacheSizeLimit(MaxCachedReaders)
@@ -77,7 +79,13 @@ func NewPieceDirectoryWithSettings(store *bdclient.Store, pr types.PieceReader,
 		pieceReaderCache:   prCache,
 		addIdxThrottleSize: addIndexThrottleSize,
 		addIdxThrottle:     make(chan struct{}, addIndexThrottleSize),
-		settings:           settings,
+		settings: &settings{
+			addIndexConcurrency: config.DefaultAddIndexConcurrency,
+		},
+	}
+
+	for _, opt := range opts {
+		opt(pd.settings)
 	}
 
 	expireCallback := func(key string, reason ttlcache.EvictionReason, value interface{}) {