add --remote-worker-count (#258)

- **ADDED** New option to control the number of worker thread in remote stores to avoid overflowing the network connection. - `--remote-worker-count` Set number of workers created for the remote store, defaults to match number of logical CPUs with upper limit of 8 for networked remote stores - **FIXED** Don't update store index if we failed to upload blocks
DanEngelbrecht · Nov 23, 2024 · 8fa740d · 8fa740d
1 parent af01f04
commit 8fa740d
Show file tree

Hide file tree

Showing 17 changed files with 131 additions and 41 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,11 +1,13 @@
 ##
 - **FIXED** fix(s3): use HeadObject for checking if blob exists [bergemalm](https://github.com/bergemalm)
- update longtaillib to 0.4.3 (#260)
+- **ADDED** New option to control the number of worker thread in remote stores to avoid overflowing the network connection.
+  - `--remote-worker-count` Set number of workers created for the remote store, defaults to match number of logical CPUs with upper limit of 8 for networked remote stores
+- **FIXED** Don't update store index if we failed to upload blocks
 - **UPDATED** Update to golang 1.23
 - **UPDATED** Updated all golang dependencies
 - **UPDATED** Update longtaillib to v0.4.3
 
-## 0.4.3
+## v0.4.3
 - **CHANGED** For multi-source downsync/get the separator for paths is changed to | to avoid problems with path that contains spaces
 - **CHANGED** Made multi-path options separate from single-path
   -- `--source-path` vs `--source-paths`

diff --git a/cmd/longtail/main.go b/cmd/longtail/main.go
@@ -96,6 +96,8 @@ func runCommand() error {
 		context.NumWorkerCount = commands.Cli.WorkerCount
 	}
 
+	context.NumRemoteWorkerCount = commands.Cli.RemoteWorkerCount
+
 	if commands.Cli.MemTrace || commands.Cli.MemTraceDetailed || commands.Cli.MemTraceCSV != "" {
 		longtaillib.EnableMemtrace()
 		defer func() {

diff --git a/commands/cmd_clonestore.go b/commands/cmd_clonestore.go
@@ -516,6 +516,7 @@ func cloneOneVersion(
 
 func cloneStore(
 	numWorkerCount int,
+	remoteStoreWorkerCount int,
 	sourceStoreURI string,
 	sourceEndpointResolverURI string,
 	targetStoreURI string,
@@ -539,6 +540,7 @@ func cloneStore(
 	log := logrus.WithFields(logrus.Fields{
 		"fname":                        fname,
 		"numWorkerCount":               numWorkerCount,
+		"remoteStoreWorkerCount":       remoteStoreWorkerCount,
 		"sourceStoreURI":               sourceStoreURI,
 		"sourceEndpointResolverURI":    sourceEndpointResolverURI,
 		"targetStoreURI":               targetStoreURI,
@@ -578,7 +580,7 @@ func cloneStore(
 	localFS := longtaillib.CreateFSStorageAPI()
 	defer localFS.Dispose()
 
-	sourceRemoteIndexStore, err := remotestore.CreateBlockStoreForURI(sourceStoreURI, nil, jobs, numWorkerCount, 8388608, 1024, remotestore.ReadOnly, enableFileMapping, longtailutils.WithS3EndpointResolverURI(sourceEndpointResolverURI))
+	sourceRemoteIndexStore, err := remotestore.CreateBlockStoreForURI(sourceStoreURI, nil, jobs, remoteStoreWorkerCount, 8388608, 1024, remotestore.ReadOnly, enableFileMapping, longtailutils.WithS3EndpointResolverURI(sourceEndpointResolverURI))
 	if err != nil {
 		return storeStats, timeStats, errors.Wrap(err, fname)
 	}
@@ -612,7 +614,7 @@ func cloneStore(
 	defer sourceLRUBlockStore.Dispose()
 	defer sourceStore.Dispose()
 
-	targetRemoteStore, err := remotestore.CreateBlockStoreForURI(targetStoreURI, nil, jobs, numWorkerCount, targetBlockSize, maxChunksPerBlock, remotestore.ReadWrite, enableFileMapping, longtailutils.WithS3EndpointResolverURI(targetEndpointResolverURI))
+	targetRemoteStore, err := remotestore.CreateBlockStoreForURI(targetStoreURI, nil, jobs, remoteStoreWorkerCount, targetBlockSize, maxChunksPerBlock, remotestore.ReadWrite, enableFileMapping, longtailutils.WithS3EndpointResolverURI(targetEndpointResolverURI))
 	if err != nil {
 		return storeStats, timeStats, errors.Wrap(err, fname)
 	}
@@ -736,6 +738,7 @@ type CloneStoreCmd struct {
 func (r *CloneStoreCmd) Run(ctx *Context) error {
 	storeStats, timeStats, err := cloneStore(
 		ctx.NumWorkerCount,
+		ctx.NumRemoteWorkerCount,
 		r.SourceStorageURI,
 		r.SourceS3EndpointResolverURL,
 		r.TargetStorageURI,

diff --git a/commands/cmd_cp.go b/commands/cmd_cp.go
@@ -14,6 +14,7 @@ import (
 
 func cpVersionIndex(
 	numWorkerCount int,
+	remoteStoreWorkerCount int,
 	blobStoreURI string,
 	s3EndpointResolverURI string,
 	versionIndexPath string,
@@ -23,15 +24,16 @@ func cpVersionIndex(
 	enableFileMapping bool) ([]longtailutils.StoreStat, []longtailutils.TimeStat, error) {
 	const fname = "cpVersionIndex"
 	log := logrus.WithContext(context.Background()).WithFields(logrus.Fields{
-		"fname":                 fname,
-		"numWorkerCount":        numWorkerCount,
-		"blobStoreURI":          blobStoreURI,
-		"s3EndpointResolverURI": s3EndpointResolverURI,
-		"versionIndexPath":      versionIndexPath,
-		"localCachePath":        localCachePath,
-		"sourcePath":            sourcePath,
-		"targetPath":            targetPath,
-		"enableFileMapping":     enableFileMapping,
+		"fname":                  fname,
+		"numWorkerCount":         numWorkerCount,
+		"remoteStoreWorkerCount": remoteStoreWorkerCount,
+		"blobStoreURI":           blobStoreURI,
+		"s3EndpointResolverURI":  s3EndpointResolverURI,
+		"versionIndexPath":       versionIndexPath,
+		"localCachePath":         localCachePath,
+		"sourcePath":             sourcePath,
+		"targetPath":             targetPath,
+		"enableFileMapping":      enableFileMapping,
 	})
 	log.Info(fname)
 
@@ -48,7 +50,7 @@ func cpVersionIndex(
 	defer hashRegistry.Dispose()
 
 	// MaxBlockSize and MaxChunksPerBlock are just temporary values until we get the remote index settings
-	remoteIndexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, nil, jobs, numWorkerCount, 8388608, 1024, remotestore.ReadOnly, enableFileMapping, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
+	remoteIndexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, nil, jobs, remoteStoreWorkerCount, 8388608, 1024, remotestore.ReadOnly, enableFileMapping, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
 	if err != nil {
 		return storeStats, timeStats, errors.Wrap(err, fname)
 	}
@@ -220,6 +222,7 @@ type CpCmd struct {
 func (r *CpCmd) Run(ctx *Context) error {
 	storeStats, timeStats, err := cpVersionIndex(
 		ctx.NumWorkerCount,
+		ctx.NumRemoteWorkerCount,
 		r.StorageURI,
 		r.S3EndpointResolverURL,
 		r.VersionIndexPath,

diff --git a/commands/cmd_createversionstoreindex.go b/commands/cmd_createversionstoreindex.go
@@ -12,6 +12,7 @@ import (
 
 func createVersionStoreIndex(
 	numWorkerCount int,
+	remoteStoreWorkerCount int,
 	blobStoreURI string,
 	s3EndpointResolverURI string,
 	sourceFilePath string,
@@ -20,6 +21,7 @@ func createVersionStoreIndex(
 	log := logrus.WithFields(logrus.Fields{
 		"fname":                      fname,
 		"numWorkerCount":             numWorkerCount,
+		"remoteStoreWorkerCount":     remoteStoreWorkerCount,
 		"blobStoreURI":               blobStoreURI,
 		"s3EndpointResolverURI":      s3EndpointResolverURI,
 		"sourceFilePath":             sourceFilePath,
@@ -35,7 +37,7 @@ func createVersionStoreIndex(
 	jobs := longtaillib.CreateBikeshedJobAPI(uint32(numWorkerCount), 0)
 	defer jobs.Dispose()
 
-	indexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, nil, jobs, numWorkerCount, 8388608, 1024, remotestore.ReadOnly, false, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
+	indexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, nil, jobs, remoteStoreWorkerCount, 8388608, 1024, remotestore.ReadOnly, false, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
 	if err != nil {
 		return storeStats, timeStats, errors.Wrap(err, fname)
 	}
@@ -96,6 +98,7 @@ type CreateVersionStoreIndexCmd struct {
 func (r *CreateVersionStoreIndexCmd) Run(ctx *Context) error {
 	storeStats, timeStats, err := createVersionStoreIndex(
 		ctx.NumWorkerCount,
+		ctx.NumRemoteWorkerCount,
 		r.StorageURI,
 		r.S3EndpointResolverURL,
 		r.SourcePath,

diff --git a/commands/cmd_downsync.go b/commands/cmd_downsync.go
@@ -29,6 +29,7 @@ func readVersionIndex(sourceFilePath string, opts ...longtailstorelib.BlobStoreO
 
 func downsync(
 	numWorkerCount int,
+	remoteStoreWorkerCount int,
 	blobStoreURI string,
 	s3EndpointResolverURI string,
 	sourceFilePath string,
@@ -50,6 +51,7 @@ func downsync(
 	log := logrus.WithFields(logrus.Fields{
 		"fname":                       fname,
 		"numWorkerCount":              numWorkerCount,
+		"remoteStoreWorkerCount":      remoteStoreWorkerCount,
 		"blobStoreURI":                blobStoreURI,
 		"s3EndpointResolverURI":       s3EndpointResolverURI,
 		"sourceFilePath":              sourceFilePath,
@@ -191,7 +193,7 @@ func downsync(
 	}
 
 	// MaxBlockSize and MaxChunksPerBlock are just temporary values until we get the remote index settings
-	remoteIndexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, versionLocalStoreIndexPaths, jobs, numWorkerCount, 8388608, 1024, remotestore.ReadOnly, enableFileMapping, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
+	remoteIndexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, versionLocalStoreIndexPaths, jobs, remoteStoreWorkerCount, 8388608, 1024, remotestore.ReadOnly, enableFileMapping, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
 	if err != nil {
 		return storeStats, timeStats, errors.Wrap(err, fname)
 	}
@@ -486,6 +488,7 @@ type DownsyncCmd struct {
 func (r *DownsyncCmd) Run(ctx *Context) error {
 	storeStats, timeStats, err := downsync(
 		ctx.NumWorkerCount,
+		ctx.NumRemoteWorkerCount,
 		r.StorageURI,
 		r.S3EndpointResolverURL,
 		r.SourcePath,

diff --git a/commands/cmd_get.go b/commands/cmd_get.go
@@ -13,6 +13,7 @@ import (
 
 func get(
 	numWorkerCount int,
+	numRemoteWorkerCount int,
 	getConfigPath string,
 	getConfigPaths []string,
 	s3EndpointResolverURI string,
@@ -31,6 +32,7 @@ func get(
 	log := logrus.WithFields(logrus.Fields{
 		"fname":                 fname,
 		"numWorkerCount":        numWorkerCount,
+		"numRemoteWorkerCount":  numRemoteWorkerCount,
 		"getConfigPath":         getConfigPath,
 		"getConfigPaths":        getConfigPaths,
 		"s3EndpointResolverURI": s3EndpointResolverURI,
@@ -113,6 +115,7 @@ func get(
 
 	downSyncStoreStats, downSyncTimeStats, err := downsync(
 		numWorkerCount,
+		numRemoteWorkerCount,
 		blobStoreURI,
 		s3EndpointResolverURI,
 		"",
@@ -158,6 +161,7 @@ type GetCmd struct {
 func (r *GetCmd) Run(ctx *Context) error {
 	storeStats, timeStats, err := get(
 		ctx.NumWorkerCount,
+		ctx.NumRemoteWorkerCount,
 		r.GetConfigURI,
 		r.GetConfigURIs,
 		r.S3EndpointResolverURL,

diff --git a/commands/cmd_initremotestore.go b/commands/cmd_initremotestore.go
@@ -12,16 +12,18 @@ import (
 
 func initRemoteStore(
 	numWorkerCount int,
+	remoteStoreWorkerCount int,
 	blobStoreURI string,
 	s3EndpointResolverURI string,
 	hashAlgorithm string) ([]longtailutils.StoreStat, []longtailutils.TimeStat, error) {
 	const fname = "initRemoteStore"
 	log := logrus.WithFields(logrus.Fields{
-		"fname":                 fname,
-		"numWorkerCount":        numWorkerCount,
-		"blobStoreURI":          blobStoreURI,
-		"s3EndpointResolverURI": s3EndpointResolverURI,
-		"hashAlgorithm":         hashAlgorithm,
+		"fname":                  fname,
+		"numWorkerCount":         numWorkerCount,
+		"remoteStoreWorkerCount": remoteStoreWorkerCount,
+		"blobStoreURI":           blobStoreURI,
+		"s3EndpointResolverURI":  s3EndpointResolverURI,
+		"hashAlgorithm":          hashAlgorithm,
 	})
 	log.Info(fname)
 
@@ -33,7 +35,7 @@ func initRemoteStore(
 	jobs := longtaillib.CreateBikeshedJobAPI(uint32(numWorkerCount), 0)
 	defer jobs.Dispose()
 
-	remoteIndexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, nil, jobs, numWorkerCount, 8388608, 1024, remotestore.Init, false, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
+	remoteIndexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, nil, jobs, remoteStoreWorkerCount, 8388608, 1024, remotestore.Init, false, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
 	if err != nil {
 		return storeStats, timeStats, errors.Wrap(err, fname)
 	}
@@ -78,6 +80,7 @@ type InitRemoteStoreCmd struct {
 func (r *InitRemoteStoreCmd) Run(ctx *Context) error {
 	storeStats, timeStats, err := initRemoteStore(
 		ctx.NumWorkerCount,
+		ctx.NumRemoteWorkerCount,
 		r.StorageURI,
 		r.S3EndpointResolverURL,
 		r.Hashing)

diff --git a/commands/cmd_printVersionUsage.go b/commands/cmd_printVersionUsage.go
@@ -2,6 +2,7 @@ package commands
 
 import (
 	"fmt"
+	"runtime"
 	"time"
 
 	"github.com/DanEngelbrecht/golongtail/longtaillib"
@@ -14,6 +15,7 @@ import (
 
 func printVersionUsage(
 	numWorkerCount int,
+	remoteStoreWorkerCount int,
 	blobStoreURI string,
 	s3EndpointResolverURI string,
 	versionIndexPath string,
@@ -41,7 +43,7 @@ func printVersionUsage(
 
 	var indexStore longtaillib.Longtail_BlockStoreAPI
 
-	remoteIndexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, nil, jobs, numWorkerCount, 8388608, 1024, remotestore.ReadOnly, false, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
+	remoteIndexStore, err := remotestore.CreateBlockStoreForURI(blobStoreURI, nil, jobs, remoteStoreWorkerCount, 8388608, 1024, remotestore.ReadOnly, false, longtailutils.WithS3EndpointResolverURI(s3EndpointResolverURI))
 	if err != nil {
 		return storeStats, timeStats, errors.Wrap(err, fname)
 	}
@@ -103,7 +105,10 @@ func printVersionUsage(
 	defer progress.Dispose()
 
 	blockHashes := existingStoreIndex.GetBlockHashes()
-	maxBatchSize := int(numWorkerCount)
+	maxBatchSize := int(remoteStoreWorkerCount)
+	if maxBatchSize == 0 {
+		maxBatchSize = runtime.NumCPU()
+	}
 	for i := 0; i < len(blockHashes); {
 		batchSize := len(blockHashes) - i
 		if batchSize > maxBatchSize {
@@ -216,6 +221,7 @@ type PrintVersionUsageCmd struct {
 func (r *PrintVersionUsageCmd) Run(ctx *Context) error {
 	storeStats, timeStats, err := printVersionUsage(
 		ctx.NumWorkerCount,
+		ctx.NumRemoteWorkerCount,
 		r.StorageURI,
 		r.S3EndpointResolverURL,
 		r.VersionIndexPath,