LOTUS_DAGSTORE_COPY_CONCURRENCY for controlling copy concurrency.

go.mod

@@ -26,7 +26,7 @@ require (
 	github.com/elastic/gosigar v0.12.0
 	github.com/etclabscore/go-openrpc-reflect v0.0.36
 	github.com/fatih/color v1.9.0
-	github.com/filecoin-project/dagstore v0.3.2-0.20210728152352-8e0b04b9b384
+	github.com/filecoin-project/dagstore v0.3.2
 	github.com/filecoin-project/filecoin-ffi v0.30.4-0.20200910194244-f640612a1a1f
 	github.com/filecoin-project/go-address v0.0.5
 	github.com/filecoin-project/go-bitfield v0.2.4

go.sum

@@ -257,8 +257,8 @@ github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/fd/go-nat v1.0.0/go.mod h1:BTBu/CKvMmOMUPkKVef1pngt2WFH/lg7E6yQnulfp6E=
 github.com/filecoin-project/dagstore v0.3.1/go.mod h1:WY5OoLfnwISCk6eASSF927KKPqLPIlTwmG1qHpA08KY=
-github.com/filecoin-project/dagstore v0.3.2-0.20210728152352-8e0b04b9b384 h1:0LrfJHR6YAa8t7VEZmbNECW7Deov8Di9CZHsZaamiM4=
+github.com/filecoin-project/dagstore v0.3.2 h1:YgLW+0VpbjsWuRDUaPmjaS/KRz1cRbi0zAjZV/xfIwY=
-github.com/filecoin-project/dagstore v0.3.2-0.20210728152352-8e0b04b9b384/go.mod h1:WY5OoLfnwISCk6eASSF927KKPqLPIlTwmG1qHpA08KY=
+github.com/filecoin-project/dagstore v0.3.2/go.mod h1:WY5OoLfnwISCk6eASSF927KKPqLPIlTwmG1qHpA08KY=
 github.com/filecoin-project/go-address v0.0.3/go.mod h1:jr8JxKsYx+lQlQZmF5i2U0Z+cGQ59wMIps/8YW/lDj8=
 github.com/filecoin-project/go-address v0.0.5 h1:SSaFT/5aLfPXycUlFyemoHYhRgdyXClXCyDdNJKPlDM=
 github.com/filecoin-project/go-address v0.0.5/go.mod h1:jr8JxKsYx+lQlQZmF5i2U0Z+cGQ59wMIps/8YW/lDj8=

markets/dagstore/wrapper.go

@@ -43,6 +43,7 @@ type DAGStore interface {
 	RecoverShard(ctx context.Context, key shard.Key, out chan dagstore.ShardResult, _ dagstore.RecoverOpts) error
 	GC(ctx context.Context) (*dagstore.GCResult, error)
 	Close() error
+	Start(ctx context.Context) error
 }
 
 type closableBlockstore struct {
@@ -108,7 +109,7 @@ func NewDagStoreWrapper(cfg MarketDAGStoreConfig, mountApi LotusAccessor) (*Wrap
 	}, nil
 }
 
-func (ds *Wrapper) Start(ctx context.Context) {
+func (ds *Wrapper) Start(ctx context.Context) error {
 	ds.ctx, ds.cancel = context.WithCancel(ctx)
 
 	// Run a go-routine to do DagStore GC.
@@ -124,6 +125,8 @@ func (ds *Wrapper) Start(ctx context.Context) {
 		ds.backgroundWg.Add(1)
 		go dagstore.RecoverImmediately(ds.ctx, dss, ds.failureCh, maxRecoverAttempts, ds.backgroundWg.Done)
 	}
+
+	return ds.dagStore.Start(ctx)
 }
 
 func (ds *Wrapper) traceLoop() {

node/modules/storageminer.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -615,13 +616,23 @@ func DagStoreWrapper(
 		return nil, err
 	}
 
+	var maxCopies = 2
+	// TODO replace env with config.toml attribute.
+	v, ok := os.LookupEnv("LOTUS_DAGSTORE_COPY_CONCURRENCY")
+	if ok {
+		concurrency, err := strconv.Atoi(v)
+		if err == nil {
+			maxCopies = concurrency
+		}
+	}
+
 	cfg := dagstore.MarketDAGStoreConfig{
 		TransientsDir:       filepath.Join(dagStoreDir, "transients"),
 		IndexDir:            filepath.Join(dagStoreDir, "index"),
 		Datastore:           dagStoreDS,
 		GCInterval:          1 * time.Minute,
 		MaxConcurrentIndex:  5,
-		MaxConcurrentCopies: 2,
+		MaxConcurrentCopies: maxCopies,
 	}
 
 	dsw, err := dagstore.NewDagStoreWrapper(cfg, lotusAccessor)
@@ -631,8 +642,7 @@ func DagStoreWrapper(
 
 	lc.Append(fx.Hook{
 		OnStart: func(ctx context.Context) error {
-			dsw.Start(ctx)
+			return dsw.Start(ctx)
-			return nil
 		},
 		OnStop: func(context.Context) error {
 			return dsw.Close()