Cloud/remote storage & tiering: configurable multipart upload/download concurrency (#11319)

* pb: add multipart concurrency fields to RemoteConf and tier move requests

RemoteConf gains upload_concurrency/download_concurrency (0 = client
default); VolumeTierMoveDatToRemote/FromRemote requests gain a
concurrency field (0 = backend default).

* remote storage: honor RemoteConf upload/download concurrency in s3 and azure clients

s3 client: ReadFile passes conf download_concurrency to the downloader,
WriteFile uses upload_concurrency for the uploader; previously
hard-coded 1 upload / 5 download parts. 0 keeps defaults. Same for
azure client.

* storage: plumb concurrency through backend interface and tier upload/download

BackendStorage.CopyFile/DownloadFile take a concurrency hint (<=0 =
backend configured default); s3 backend reads
upload_concurrency/download_concurrency from scaffold config with
parseConcurrency fallback, rclone updated to the new signature. Tier
move gRPC handlers forward the request concurrency to the backend.

* shell: -upload_concurrency/-download_concurrency for remote.configure, -concurrent for volume.tier

remote.configure exposes upload/download concurrency persisted into
RemoteConf; volume.tier move/evict commands forward -concurrent to the
tier move requests. Documented in master-cloud.toml scaffold.

* test: cover concurrency propagation in remote tier integration test

* remote.configure: merge existing config on partial update

Load the stored RemoteConf before saving so a partial update (e.g. only
-upload_concurrency) preserves credentials, endpoints, and type instead
of replacing them with new-config defaults. Only treat a confirmed
ErrNotFound as a new configuration; propagate all other load errors so a
transient filer failure does not overwrite stored settings.

On a type transition, reset backend-specific fields to the destination
type's new-config defaults rather than inheriting the old backend's
empty values. Bound configured concurrency to a sane maximum.

* remote storage: honor configured download concurrency in S3 and Azure

ReadFileWithConcurrency now resolves a zero request override against the
client's configured download_concurrency (new downloadConcurrency()
helpers), so the remote-mount/cache read path honors
RemoteConf.DownloadConcurrency instead of the hard-coded default.

Azure also clamps the resolved value to math.MaxUint16 regardless of
whether the fallback was used, preventing uint16 wraparound when a
configured value exceeds 65535.

* shell: rename -concurrent to -concurrency and validate tier transfer bounds

Rename the -concurrent flag to -concurrency across volume.tier.upload,
volume.tier.download, and volume.tier.compact to match the proto field and
RemoteConf field names. Add validateTierConcurrency to reject values that
would wrap int32 or exceed a 1024 cap before constructing the request.

* server: clamp tier move concurrency in gRPC handlers

Add clampTierConcurrency to both VolumeTierMoveDatToRemote and
VolumeTierMoveDatFromRemote handlers so a direct gRPC caller cannot spawn
an unbounded number of network workers.

* trim verbose comments added with concurrency feature

Remove redundant doc comments on the backend interface, rclone backend,
s3_backend parseConcurrency, and test helpers that restated the obvious.

* remote.configure: apply type defaults before re-parse so explicit flags win

applyTypeDefaults ran after the second flag parse, overwriting explicit
destination flags (e.g. -s3.region=eu-west-1) with new-config defaults.
Move the type-transition default reset before the re-parse so user-supplied
flags override the destination defaults.

* remote.configure: only treat explicit -type as a type transition

The first parse defaults -type to s3, so a concurrency-only update on an
existing non-S3 config captured requestedType=s3 and wrongly triggered a
type transition, resetting the stored backend to S3. Use fs.Visit to
detect whether -type was explicitly supplied; an omitted -type keeps the
stored backend.

---------

Co-authored-by: Jack Meredith <9480542+jackusm@users.noreply.github.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

docker/compose/master-cloud.toml

@@ -30,3 +30,5 @@ sleep_minutes = 17          # sleep minutes between each script execution
 	bucket = "volume_bucket"    # an existing bucket
 	endpoint = "http://server2:8333"
 	storage_class = "STANDARD_IA"
+	# upload_concurrency = 5     # concurrent multipart part uploads per volume (volume.tier.upload -concurrent overrides)
+	# download_concurrency = 5   # concurrent multipart part downloads per volume (volume.tier.download -concurrent overrides)

weed/pb/remote.proto

@@ -65,6 +65,9 @@ message RemoteConf {
   string contabo_secret_key = 69;
   string contabo_endpoint = 70;
   string contabo_region = 71;
+
+  uint32 upload_concurrency = 72; // multipart upload concurrency per file, 0 = client default (1 for S3, 16 for Azure)
+  uint32 download_concurrency = 73; // multipart download concurrency per read, 0 = client default (5 for S3, 16 for Azure)
 }
 
 message RemoteStorageMapping {

weed/pb/remote_pb/remote.pb.go

@@ -71,6 +71,8 @@ type RemoteConf struct {
 	ContaboSecretKey                string                 `protobuf:"bytes,69,opt,name=contabo_secret_key,json=contaboSecretKey,proto3" json:"contabo_secret_key,omitempty"`
 	ContaboEndpoint                 string                 `protobuf:"bytes,70,opt,name=contabo_endpoint,json=contaboEndpoint,proto3" json:"contabo_endpoint,omitempty"`
 	ContaboRegion                   string                 `protobuf:"bytes,71,opt,name=contabo_region,json=contaboRegion,proto3" json:"contabo_region,omitempty"`
+	UploadConcurrency               uint32                 `protobuf:"varint,72,opt,name=upload_concurrency,json=uploadConcurrency,proto3" json:"upload_concurrency,omitempty"`       // multipart upload concurrency per file, 0 = client default (1 for S3, 16 for Azure)
+	DownloadConcurrency             uint32                 `protobuf:"varint,73,opt,name=download_concurrency,json=downloadConcurrency,proto3" json:"download_concurrency,omitempty"` // multipart download concurrency per read, 0 = client default (5 for S3, 16 for Azure)
 	unknownFields                   protoimpl.UnknownFields
 	sizeCache                       protoimpl.SizeCache
 }
@@ -420,6 +422,20 @@ func (x *RemoteConf) GetContaboRegion() string {
 	return ""
 }
 
+func (x *RemoteConf) GetUploadConcurrency() uint32 {
+	if x != nil {
+		return x.UploadConcurrency
+	}
+	return 0
+}
+
+func (x *RemoteConf) GetDownloadConcurrency() uint32 {
+	if x != nil {
+		return x.DownloadConcurrency
+	}
+	return 0
+}
+
 type RemoteStorageMapping struct {
 	state                    protoimpl.MessageState            `protogen:"open.v1"`
 	Mappings                 map[string]*RemoteStorageLocation `protobuf:"bytes,1,rep,name=mappings,proto3" json:"mappings,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
@@ -552,7 +568,7 @@ var File_remote_proto protoreflect.FileDescriptor
 
 const file_remote_proto_rawDesc = "" +
 	"\n" +
-	"\fremote.proto\x12\tremote_pb\"\xea\x0e\n" +
+	"\fremote.proto\x12\tremote_pb\"\xcc\x0f\n" +
 	"\n" +
 	"RemoteConf\x12\x12\n" +
 	"\x04type\x18\x01 \x01(\tR\x04type\x12\x12\n" +
@@ -601,7 +617,9 @@ const file_remote_proto_rawDesc = "" +
 	"\x12contabo_access_key\x18D \x01(\tR\x10contaboAccessKey\x12,\n" +
 	"\x12contabo_secret_key\x18E \x01(\tR\x10contaboSecretKey\x12)\n" +
 	"\x10contabo_endpoint\x18F \x01(\tR\x0fcontaboEndpoint\x12%\n" +
-	"\x0econtabo_region\x18G \x01(\tR\rcontaboRegion\"\xff\x01\n" +
+	"\x0econtabo_region\x18G \x01(\tR\rcontaboRegion\x12-\n" +
+	"\x12upload_concurrency\x18H \x01(\rR\x11uploadConcurrency\x121\n" +
+	"\x14download_concurrency\x18I \x01(\rR\x13downloadConcurrency\"\xff\x01\n" +
 	"\x14RemoteStorageMapping\x12I\n" +
 	"\bmappings\x18\x01 \x03(\v2-.remote_pb.RemoteStorageMapping.MappingsEntryR\bmappings\x12=\n" +
 	"\x1bprimary_bucket_storage_name\x18\x02 \x01(\tR\x18primaryBucketStorageName\x1a]\n" +

weed/pb/volume_server.proto

@@ -659,6 +659,7 @@ message VolumeTierMoveDatToRemoteRequest {
     string collection = 2;
     string destination_backend_name = 3;
     bool keep_local_dat_file = 4;
+    int32 concurrency = 5; // multipart upload concurrency, 0 = backend default (5 for S3)
 }
 message VolumeTierMoveDatToRemoteResponse {
     int64 processed = 1;
@@ -669,6 +670,7 @@ message VolumeTierMoveDatFromRemoteRequest {
     uint32 volume_id = 1;
     string collection = 2;
     bool keep_remote_dat_file = 3;
+    int32 concurrency = 4; // multipart download concurrency, 0 = backend default (5 for S3)
 }
 message VolumeTierMoveDatFromRemoteResponse {
     int64 processed = 1;

weed/pb/volume_server_pb/volume_server.pb.go

@@ -5471,6 +5471,7 @@ type VolumeTierMoveDatToRemoteRequest struct {
 	Collection             string                 `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"`
 	DestinationBackendName string                 `protobuf:"bytes,3,opt,name=destination_backend_name,json=destinationBackendName,proto3" json:"destination_backend_name,omitempty"`
 	KeepLocalDatFile       bool                   `protobuf:"varint,4,opt,name=keep_local_dat_file,json=keepLocalDatFile,proto3" json:"keep_local_dat_file,omitempty"`
+	Concurrency            int32                  `protobuf:"varint,5,opt,name=concurrency,proto3" json:"concurrency,omitempty"` // multipart upload concurrency, 0 = backend default (5 for S3)
 	unknownFields          protoimpl.UnknownFields
 	sizeCache              protoimpl.SizeCache
 }
@@ -5533,6 +5534,13 @@ func (x *VolumeTierMoveDatToRemoteRequest) GetKeepLocalDatFile() bool {
 	return false
 }
 
+func (x *VolumeTierMoveDatToRemoteRequest) GetConcurrency() int32 {
+	if x != nil {
+		return x.Concurrency
+	}
+	return 0
+}
+
 type VolumeTierMoveDatToRemoteResponse struct {
 	state               protoimpl.MessageState `protogen:"open.v1"`
 	Processed           int64                  `protobuf:"varint,1,opt,name=processed,proto3" json:"processed,omitempty"`
@@ -5590,6 +5598,7 @@ type VolumeTierMoveDatFromRemoteRequest struct {
 	VolumeId          uint32                 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"`
 	Collection        string                 `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"`
 	KeepRemoteDatFile bool                   `protobuf:"varint,3,opt,name=keep_remote_dat_file,json=keepRemoteDatFile,proto3" json:"keep_remote_dat_file,omitempty"`
+	Concurrency       int32                  `protobuf:"varint,4,opt,name=concurrency,proto3" json:"concurrency,omitempty"` // multipart download concurrency, 0 = backend default (5 for S3)
 	unknownFields     protoimpl.UnknownFields
 	sizeCache         protoimpl.SizeCache
 }
@@ -5645,6 +5654,13 @@ func (x *VolumeTierMoveDatFromRemoteRequest) GetKeepRemoteDatFile() bool {
 	return false
 }
 
+func (x *VolumeTierMoveDatFromRemoteRequest) GetConcurrency() int32 {
+	if x != nil {
+		return x.Concurrency
+	}
+	return 0
+}
+
 type VolumeTierMoveDatFromRemoteResponse struct {
 	state               protoimpl.MessageState `protogen:"open.v1"`
 	Processed           int64                  `protobuf:"varint,1,opt,name=processed,proto3" json:"processed,omitempty"`
@@ -7644,23 +7660,25 @@ const file_volume_server_proto_rawDesc = "" +
 	"\vBytesOffset\x18\x04 \x01(\rR\vBytesOffset\x12\"\n" +
 	"\rdat_file_size\x18\x05 \x01(\x03R\vdatFileSize\x12 \n" +
 	"\vDestroyTime\x18\x06 \x01(\x04R\vDestroyTime\x12\x1b\n" +
-	"\tread_only\x18\a \x01(\bR\breadOnly\"\xc8\x01\n" +
+	"\tread_only\x18\a \x01(\bR\breadOnly\"\xea\x01\n" +
 	" VolumeTierMoveDatToRemoteRequest\x12\x1b\n" +
 	"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" +
 	"\n" +
 	"collection\x18\x02 \x01(\tR\n" +
 	"collection\x128\n" +
 	"\x18destination_backend_name\x18\x03 \x01(\tR\x16destinationBackendName\x12-\n" +
-	"\x13keep_local_dat_file\x18\x04 \x01(\bR\x10keepLocalDatFile\"s\n" +
+	"\x13keep_local_dat_file\x18\x04 \x01(\bR\x10keepLocalDatFile\x12 \n" +
+	"\vconcurrency\x18\x05 \x01(\x05R\vconcurrency\"s\n" +
 	"!VolumeTierMoveDatToRemoteResponse\x12\x1c\n" +
 	"\tprocessed\x18\x01 \x01(\x03R\tprocessed\x120\n" +
-	"\x13processedPercentage\x18\x02 \x01(\x02R\x13processedPercentage\"\x92\x01\n" +
+	"\x13processedPercentage\x18\x02 \x01(\x02R\x13processedPercentage\"\xb4\x01\n" +
 	"\"VolumeTierMoveDatFromRemoteRequest\x12\x1b\n" +
 	"\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x1e\n" +
 	"\n" +
 	"collection\x18\x02 \x01(\tR\n" +
 	"collection\x12/\n" +
-	"\x14keep_remote_dat_file\x18\x03 \x01(\bR\x11keepRemoteDatFile\"u\n" +
+	"\x14keep_remote_dat_file\x18\x03 \x01(\bR\x11keepRemoteDatFile\x12 \n" +
+	"\vconcurrency\x18\x04 \x01(\x05R\vconcurrency\"u\n" +
 	"#VolumeTierMoveDatFromRemoteResponse\x12\x1c\n" +
 	"\tprocessed\x18\x01 \x01(\x03R\tprocessed\x120\n" +
 	"\x13processedPercentage\x18\x02 \x01(\x02R\x13processedPercentage\"\x1b\n" +

weed/remote_storage/azure/azure_storage_client.go

@@ -153,6 +153,20 @@ type azureRemoteStorageClient struct {
 var _ = remote_storage.RemoteStorageClient(&azureRemoteStorageClient{})
 var _ = remote_storage.RemoteStorageConcurrentReader(&azureRemoteStorageClient{})
 
+func (az *azureRemoteStorageClient) uploadConcurrency() int {
+	if n := int(az.conf.GetUploadConcurrency()); n > 0 {
+		return n
+	}
+	return defaultConcurrency
+}
+
+func (az *azureRemoteStorageClient) downloadConcurrency() int {
+	if n := int(az.conf.GetDownloadConcurrency()); n > 0 {
+		return n
+	}
+	return defaultReadConcurrency
+}
+
 func (az *azureRemoteStorageClient) ListDirectory(ctx context.Context, loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) (err error) {
 	pathKey := loc.Path[1:]
 	if pathKey != "" && !strings.HasSuffix(pathKey, "/") {
@@ -303,7 +317,7 @@ func (az *azureRemoteStorageClient) Traverse(loc *remote_pb.RemoteStorageLocatio
 }
 
 func (az *azureRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (data []byte, err error) {
-	return az.ReadFileWithConcurrency(loc, offset, size, defaultReadConcurrency)
+	return az.ReadFileWithConcurrency(loc, offset, size, 0)
 }
 
 // ReadFileWithConcurrency fetches a byte range of a blob using the Azure SDK's
@@ -322,9 +336,9 @@ func (az *azureRemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.Remot
 	}
 
 	if concurrency <= 0 {
-		concurrency = defaultReadConcurrency
-	} else if concurrency > math.MaxUint16 {
-		// DownloadBufferOptions.Concurrency is uint16; clamp to avoid wraparound.
+		concurrency = az.downloadConcurrency()
+	}
+	if concurrency > math.MaxUint16 {
 		concurrency = math.MaxUint16
 	}
 
@@ -448,7 +462,7 @@ func (az *azureRemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocati
 
 	_, err = blobClient.UploadStream(context.Background(), reader, &blockblob.UploadStreamOptions{
 		BlockSize:   defaultBlockSize,
-		Concurrency: defaultConcurrency,
+		Concurrency: az.uploadConcurrency(),
 		HTTPHeaders: httpHeaders,
 		Metadata:    metadata,
 	})

weed/remote_storage/s3/s3_storage_client.go

@@ -229,6 +229,25 @@ type s3RemoteStorageClient struct {
 	conn s3iface.S3API
 }
 
+const (
+	defaultUploadConcurrency = 1
+	defaultReadConcurrency   = 5
+)
+
+func (s *s3RemoteStorageClient) uploadConcurrency() int {
+	if n := int(s.conf.GetUploadConcurrency()); n > 0 {
+		return n
+	}
+	return defaultUploadConcurrency
+}
+
+func (s *s3RemoteStorageClient) downloadConcurrency() int {
+	if n := int(s.conf.GetDownloadConcurrency()); n > 0 {
+		return n
+	}
+	return defaultReadConcurrency
+}
+
 var _ = remote_storage.RemoteStorageClient(&s3RemoteStorageClient{})
 
 func (s *s3RemoteStorageClient) Traverse(remote *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) (err error) {
@@ -374,12 +393,12 @@ func (s *s3RemoteStorageClient) StatFile(loc *remote_pb.RemoteStorageLocation) (
 }
 
 func (s *s3RemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (data []byte, err error) {
-	return s.ReadFileWithConcurrency(loc, offset, size, 5)
+	return s.ReadFileWithConcurrency(loc, offset, size, 0)
 }
 
 func (s *s3RemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.RemoteStorageLocation, offset int64, size int64, concurrency int) (data []byte, err error) {
 	if concurrency <= 0 {
-		concurrency = 5
+		concurrency = s.downloadConcurrency()
 	}
 	downloader := s3manager.NewDownloaderWithClient(s.conn, func(u *s3manager.Downloader) {
 		u.PartSize = int64(4 * 1024 * 1024)
@@ -493,7 +512,7 @@ func (s *s3RemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation,
 	// Create an uploader with the session and custom options
 	uploader := s3manager.NewUploaderWithClient(s.conn, func(u *s3manager.Uploader) {
 		u.PartSize = partSize
-		u.Concurrency = 1
+		u.Concurrency = s.uploadConcurrency()
 	})
 
 	// process tagging

weed/server/volume_grpc_tier_download.go

@@ -70,7 +70,7 @@ func (vs *VolumeServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.Volume
 	}
 	// copy the data file (DownloadFile opens, fsyncs, and closes the .dat internally)
 	datFileName := v.FileName(".dat")
-	_, err := backendStorage.DownloadFile(datFileName, storageKey, fn)
+	_, err := backendStorage.DownloadFile(datFileName, storageKey, fn, clampTierConcurrency(int(req.Concurrency)))
 	if err != nil {
 		return fmt.Errorf("backend %s copy file %s: %v", storageName, datFileName, err)
 	}

weed/server/volume_grpc_tier_download_test.go

@@ -32,6 +32,9 @@ type tierTestBackend struct {
 
 	mu      sync.Mutex
 	deletes []string
+
+	lastUploadConcurrency   int
+	lastDownloadConcurrency int
 }
 
 func (b *tierTestBackend) ToProperties() map[string]string { return map[string]string{"root": b.root} }
@@ -40,7 +43,10 @@ func (b *tierTestBackend) NewStorageFile(key string, tierInfo *volume_server_pb.
 	return &tierTestBackendFile{backend: b, key: key, tierInfo: tierInfo}
 }
 
-func (b *tierTestBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
+func (b *tierTestBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) {
+	b.mu.Lock()
+	b.lastUploadConcurrency = concurrency
+	b.mu.Unlock()
 	key = fmt.Sprintf("obj-%d", time.Now().UnixNano())
 	dst := filepath.Join(b.root, key)
 	out, err := os.Create(dst)
@@ -58,7 +64,10 @@ func (b *tierTestBackend) CopyFile(f *os.File, fn func(progressed int64, percent
 	return key, written, nil
 }
 
-func (b *tierTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) {
+func (b *tierTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) {
+	b.mu.Lock()
+	b.lastDownloadConcurrency = concurrency
+	b.mu.Unlock()
 	in, err := os.Open(filepath.Join(b.root, key))
 	if err != nil {
 		return 0, err
@@ -105,6 +114,12 @@ func (b *tierTestBackend) objectExists(key string) bool {
 	return err == nil
 }
 
+func (b *tierTestBackend) concurrencySeen() (upload, download int) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.lastUploadConcurrency, b.lastDownloadConcurrency
+}
+
 type tierTestBackendFile struct {
 	backend  *tierTestBackend
 	key      string
@@ -201,7 +216,7 @@ func tierUpVolumeOnDisk(t *testing.T, dir string, vid needle.VolumeId, b *tierTe
 		t.Fatalf("expected on-disk backend before tier-up, got %T", v.DataBackend)
 	}
 	datPath := v.FileName(".dat")
-	key, size, err := b.CopyFile(diskFile.File, nil)
+	key, size, err := b.CopyFile(diskFile.File, nil, 0)
 	if err != nil {
 		t.Fatalf("upload to fake backend: %v", err)
 	}
@@ -326,3 +341,44 @@ func TestTierMoveDatFromRemote_KeepRemote_LeavesReplicaLocal(t *testing.T) {
 		t.Fatalf("local .dat content mismatch: got %d bytes, want %d", len(gotDat), len(localDat))
 	}
 }
+
+// TestTierMoveConcurrencyPlumbing verifies the -concurrency value travels the
+// whole tier-move path to CopyFile and DownloadFile.
+func TestTierMoveConcurrencyPlumbing(t *testing.T) {
+	b := &tierTestBackend{root: t.TempDir()}
+	backend.BackendStorages[tierTestBackendName] = b
+	t.Cleanup(func() { delete(backend.BackendStorages, tierTestBackendName) })
+
+	dir := t.TempDir()
+	const vid = needle.VolumeId(72)
+	tierUpVolumeOnDisk(t, dir, vid, b)
+
+	// download path: request concurrency must reach DownloadFile
+	store := newTierTestStore(t, dir)
+	v := store.GetVolume(vid)
+	if v == nil {
+		t.Fatal("tiered volume not loaded by store")
+	}
+	vs := &VolumeServer{store: store}
+	if err := vs.VolumeTierMoveDatFromRemote(&volume_server_pb.VolumeTierMoveDatFromRemoteRequest{
+		VolumeId:    uint32(vid),
+		Concurrency: 4,
+	}, &fakeTierStream{}); err != nil {
+		t.Fatalf("VolumeTierMoveDatFromRemote: %v", err)
+	}
+	if up, down := b.concurrencySeen(); down != 4 {
+		t.Fatalf("DownloadFile concurrency = %d (upload=%d), want 4", down, up)
+	}
+
+	// upload path: the volume is local again after the download
+	if err := vs.VolumeTierMoveDatToRemote(&volume_server_pb.VolumeTierMoveDatToRemoteRequest{
+		VolumeId:               uint32(vid),
+		DestinationBackendName: tierTestBackendName,
+		Concurrency:            3,
+	}, &discardServerStream[volume_server_pb.VolumeTierMoveDatToRemoteResponse]{}); err != nil {
+		t.Fatalf("VolumeTierMoveDatToRemote: %v", err)
+	}
+	if up, down := b.concurrencySeen(); up != 3 || down != 4 {
+		t.Fatalf("CopyFile concurrency = %d (download=%d), want 3/4", up, down)
+	}
+}

weed/server/volume_grpc_tier_test.go

@@ -52,7 +52,7 @@ func (b *tierTimestampTestBackend) NewStorageFile(key string, volumeInfo *volume
 	}
 }
 
-func (b *tierTimestampTestBackend) CopyFile(file *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
+func (b *tierTimestampTestBackend) CopyFile(file *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) {
 	key = "remote.dat"
 	fileInfo, err := file.Stat()
 	if err != nil {
@@ -72,7 +72,7 @@ func (b *tierTimestampTestBackend) CopyFile(file *os.File, fn func(progressed in
 	return key, size, err
 }
 
-func (b *tierTimestampTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) {
+func (b *tierTimestampTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) {
 	input, err := os.Open(filepath.Join(b.root, key))
 	if err != nil {
 		return 0, err

weed/server/volume_grpc_tier_upload.go

@@ -10,6 +10,20 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 )
 
+// clampTierConcurrency bounds a per-request transfer concurrency so a direct
+// gRPC caller cannot spawn an unbounded number of network workers.
+const maxTierConcurrency = 1024
+
+func clampTierConcurrency(n int) int {
+	if n < 0 {
+		return 0
+	}
+	if n > maxTierConcurrency {
+		return maxTierConcurrency
+	}
+	return n
+}
+
 // VolumeTierMoveDatToRemote copy dat file to a remote tier
 func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTierMoveDatToRemoteRequest, stream volume_server_pb.VolumeServer_VolumeTierMoveDatToRemoteServer) error {
 	if err := vs.checkGrpcAdminAuth(stream.Context()); err != nil {
@@ -73,7 +87,7 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi
 	}
 
 	// copy the data file
-	key, size, err := backendStorage.CopyFile(diskFile.File, fn)
+	key, size, err := backendStorage.CopyFile(diskFile.File, fn, clampTierConcurrency(int(req.Concurrency)))
 	if err != nil {
 		return fmt.Errorf("backend %s copy file %s: %v", req.DestinationBackendName, diskFile.Name(), err)
 	}

weed/shell/command_remote_configure.go

@@ -2,6 +2,7 @@ package shell
 
 import (
 	"context"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -45,6 +46,9 @@ func (c *commandRemoteConfigure) Help() string {
 	remote.configure -name=cloud7 -type=storj -storj.access_key=xxx -storj.secret_key=yyy -storj.endpoint=https://gateway.us1.storjshare.io
 	remote.configure -name=cloud8 -type=filebase -filebase.access_key=xxx -filebase.secret_key=yyy -filebase.endpoint=https://s3.filebase.com
 
+	# tune transfer concurrency (applies to s3-compatible and azure storage)
+	remote.configure -name=cloud1 -upload_concurrency=4 -download_concurrency=8
+
 	# delete one configuration
 	remote.configure -delete -name=cloud1
 
@@ -63,71 +67,11 @@ func (c *commandRemoteConfigure) Do(args []string, commandEnv *CommandEnv, write
 
 	conf := &remote_pb.RemoteConf{}
 
-	remoteConfigureCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
-	isDelete := remoteConfigureCommand.Bool("delete", false, "delete one remote storage by its name")
-
-	remoteConfigureCommand.StringVar(&conf.Name, "name", "", "a short name to identify the remote storage")
-	remoteConfigureCommand.StringVar(&conf.Type, "type", "s3", fmt.Sprintf("[%s] storage type", remote_storage.GetAllRemoteStorageNames()))
-
-	remoteConfigureCommand.StringVar(&conf.S3AccessKey, "s3.access_key", "", "s3 access key")
-	remoteConfigureCommand.StringVar(&conf.S3SecretKey, "s3.secret_key", "", "s3 secret key")
-	remoteConfigureCommand.StringVar(&conf.S3Region, "s3.region", "us-east-2", "s3 region")
-	remoteConfigureCommand.StringVar(&conf.S3Endpoint, "s3.endpoint", "", "endpoint for s3-compatible local object store")
-	remoteConfigureCommand.StringVar(&conf.S3StorageClass, "s3.storage_class", "", "s3 storage class")
-	remoteConfigureCommand.BoolVar(&conf.S3ForcePathStyle, "s3.force_path_style", true, "s3 force path style")
-	remoteConfigureCommand.BoolVar(&conf.S3V4Signature, "s3.v4_signature", false, "s3 V4 signature")
-	remoteConfigureCommand.BoolVar(&conf.S3SupportTagging, "s3.support_tagging", true, "s3 supportTagging")
-
-	remoteConfigureCommand.StringVar(&conf.GcsGoogleApplicationCredentials, "gcs.appCredentialsFile", "", "google cloud storage credentials file, default to use env GOOGLE_APPLICATION_CREDENTIALS")
-	remoteConfigureCommand.StringVar(&conf.GcsProjectId, "gcs.projectId", "", "google cloud storage project id, default to use env GOOGLE_CLOUD_PROJECT")
-
-	remoteConfigureCommand.StringVar(&conf.AzureAccountName, "azure.account_name", "", "azure account name, default to use env AZURE_STORAGE_ACCOUNT")
-	remoteConfigureCommand.StringVar(&conf.AzureAccountKey, "azure.account_key", "", "azure account key, default to use env AZURE_STORAGE_ACCESS_KEY. Leave empty to authenticate with Entra ID")
-	remoteConfigureCommand.StringVar(&conf.AzureClientId, "azure.client_id", "", "azure user-assigned identity to authenticate, when no account key is given. Workload identity also reads env AZURE_TENANT_ID and AZURE_FEDERATED_TOKEN_FILE")
-	remoteConfigureCommand.StringVar(&conf.AzureEndpoint, "azure.endpoint", "", "azure blob service url, for accounts outside the public cloud, e.g. https://xxx.blob.core.usgovcloudapi.net/")
-
-	remoteConfigureCommand.StringVar(&conf.BackblazeKeyId, "b2.key_id", "", "backblaze keyID")
-	remoteConfigureCommand.StringVar(&conf.BackblazeApplicationKey, "b2.application_key", "", "backblaze applicationKey. Note that your Master Application Key will not work with the S3 Compatible API. You must create a new key that is eligible for use. For more information: https://help.backblaze.com/hc/en-us/articles/360047425453")
-	remoteConfigureCommand.StringVar(&conf.BackblazeEndpoint, "b2.endpoint", "", "backblaze endpoint")
-	remoteConfigureCommand.StringVar(&conf.BackblazeRegion, "b2.region", "us-west-002", "backblaze region")
-
-	remoteConfigureCommand.StringVar(&conf.AliyunAccessKey, "aliyun.access_key", "", "Aliyun access key, default to use env ALICLOUD_ACCESS_KEY_ID")
-	remoteConfigureCommand.StringVar(&conf.AliyunSecretKey, "aliyun.secret_key", "", "Aliyun secret key, default to use env ALICLOUD_ACCESS_KEY_SECRET")
-	remoteConfigureCommand.StringVar(&conf.AliyunEndpoint, "aliyun.endpoint", "", "Aliyun endpoint")
-	remoteConfigureCommand.StringVar(&conf.AliyunRegion, "aliyun.region", "", "Aliyun region")
-
-	remoteConfigureCommand.StringVar(&conf.TencentSecretId, "tencent.secret_id", "", "Tencent Secret Id, default to use env COS_SECRETID")
-	remoteConfigureCommand.StringVar(&conf.TencentSecretKey, "tencent.secret_key", "", "Tencent secret key, default to use env COS_SECRETKEY")
-	remoteConfigureCommand.StringVar(&conf.TencentEndpoint, "tencent.endpoint", "", "Tencent endpoint")
-
-	remoteConfigureCommand.StringVar(&conf.BaiduAccessKey, "baidu.access_key", "", "Baidu access key, default to use env BDCLOUD_ACCESS_KEY")
-	remoteConfigureCommand.StringVar(&conf.BaiduSecretKey, "baidu.secret_key", "", "Baidu secret key, default to use env BDCLOUD_SECRET_KEY")
-	remoteConfigureCommand.StringVar(&conf.BaiduEndpoint, "baidu.endpoint", "", "Baidu endpoint")
-	remoteConfigureCommand.StringVar(&conf.BaiduRegion, "baidu.region", "", "Baidu region")
-
-	remoteConfigureCommand.StringVar(&conf.WasabiAccessKey, "wasabi.access_key", "", "Wasabi access key")
-	remoteConfigureCommand.StringVar(&conf.WasabiSecretKey, "wasabi.secret_key", "", "Wasabi secret key")
-	remoteConfigureCommand.StringVar(&conf.WasabiEndpoint, "wasabi.endpoint", "", "Wasabi endpoint, see https://wasabi.com/wp-content/themes/wasabi/docs/API_Guide/index.html#t=topics%2Fapidiff-intro.htm")
-	remoteConfigureCommand.StringVar(&conf.WasabiRegion, "wasabi.region", "", "Wasabi region")
-
-	remoteConfigureCommand.StringVar(&conf.FilebaseAccessKey, "filebase.access_key", "", "Filebase access key")
-	remoteConfigureCommand.StringVar(&conf.FilebaseSecretKey, "filebase.secret_key", "", "Filebase secret key")
-	remoteConfigureCommand.StringVar(&conf.FilebaseEndpoint, "filebase.endpoint", "", "Filebase endpoint, https://s3.filebase.com")
-
-	remoteConfigureCommand.StringVar(&conf.StorjAccessKey, "storj.access_key", "", "Storj access key")
-	remoteConfigureCommand.StringVar(&conf.StorjSecretKey, "storj.secret_key", "", "Storj secret key")
-	remoteConfigureCommand.StringVar(&conf.StorjEndpoint, "storj.endpoint", "", "Storj endpoint")
-
-	if err = remoteConfigureCommand.Parse(args); err != nil {
+	fs, isDelete, uploadConcurrency, downloadConcurrency := c.configureFlagSet(conf, false)
+	if err = fs.Parse(args); err != nil {
 		return nil
 	}
 
-	if conf.Type != "s3" {
-		// clear out the default values
-		conf.S3Region = ""
-		conf.S3ForcePathStyle = false
-	}
-
 	if conf.Name == "" {
 		return c.listExistingRemoteStorages(commandEnv, writer)
 	}
@@ -140,10 +84,183 @@ func (c *commandRemoteConfigure) Do(args []string, commandEnv *CommandEnv, write
 		return c.deleteRemoteStorage(commandEnv, writer, conf.Name)
 	}
 
+	// Merge with an existing configuration so a partial update preserves
+	// previously stored credentials and endpoints. Only treat a confirmed
+	// missing entry as a new configuration; propagate all other load errors
+	// so a transient filer failure does not overwrite stored credentials.
+	typeExplicit := false
+	fs.Visit(func(f *flag.Flag) {
+		if f.Name == "type" {
+			typeExplicit = true
+		}
+	})
+	requestedType := conf.Type
+	existing, loadErr := c.loadRemoteStorageConf(commandEnv, conf.Name)
+	if loadErr != nil && !errors.Is(loadErr, filer_pb.ErrNotFound) {
+		return fmt.Errorf("load existing configuration %s: %v", conf.Name, loadErr)
+	}
+	if existing != nil {
+		conf = existing
+		// On an explicit type transition, reset backend-specific fields to
+		// the destination defaults before re-parsing so explicit flags
+		// override. An omitted -type keeps the stored backend.
+		if typeExplicit && requestedType != existing.Type {
+			conf.Type = requestedType
+			c.applyTypeDefaults(conf)
+		}
+		fs, isDelete, uploadConcurrency, downloadConcurrency = c.configureFlagSet(conf, true)
+		if err = fs.Parse(args); err != nil {
+			return nil
+		}
+	}
+
+	if err = applyConcurrency(conf, *uploadConcurrency, *downloadConcurrency); err != nil {
+		return err
+	}
+
+	if conf.Type != "s3" {
+		conf.S3Region = ""
+		conf.S3ForcePathStyle = false
+	}
+
 	return c.saveRemoteStorage(commandEnv, writer, conf)
 
 }
 
+// configureFlagSet builds the remote.configure flag set bound to conf. When
+// existing is true, flag defaults are taken from conf so omitted flags preserve
+// prior values instead of being reset to the hard-coded new-config defaults.
+func (c *commandRemoteConfigure) configureFlagSet(conf *remote_pb.RemoteConf, existing bool) (fs *flag.FlagSet, isDelete *bool, uploadConcurrency *int, downloadConcurrency *int) {
+	fs = flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	isDelete = fs.Bool("delete", false, "delete one remote storage by its name")
+
+	fs.StringVar(&conf.Name, "name", conf.Name, "a short name to identify the remote storage")
+	typeDefault := "s3"
+	if existing {
+		typeDefault = conf.Type
+	}
+	fs.StringVar(&conf.Type, "type", typeDefault, fmt.Sprintf("[%s] storage type", remote_storage.GetAllRemoteStorageNames()))
+
+	uploadConcurrency = fs.Int("upload_concurrency", int(conf.UploadConcurrency), "concurrent part uploads per file (0 = client default: s3 1, azure 16)")
+	downloadConcurrency = fs.Int("download_concurrency", int(conf.DownloadConcurrency), "concurrent part downloads per read (0 = client default: s3 5, azure 16)")
+
+	fs.StringVar(&conf.S3AccessKey, "s3.access_key", conf.S3AccessKey, "s3 access key")
+	fs.StringVar(&conf.S3SecretKey, "s3.secret_key", conf.S3SecretKey, "s3 secret key")
+	s3RegionDefault := "us-east-2"
+	if existing {
+		s3RegionDefault = conf.S3Region
+	}
+	fs.StringVar(&conf.S3Region, "s3.region", s3RegionDefault, "s3 region")
+	fs.StringVar(&conf.S3Endpoint, "s3.endpoint", conf.S3Endpoint, "endpoint for s3-compatible local object store")
+	fs.StringVar(&conf.S3StorageClass, "s3.storage_class", conf.S3StorageClass, "s3 storage class")
+	s3ForcePathStyleDefault := true
+	if existing {
+		s3ForcePathStyleDefault = conf.S3ForcePathStyle
+	}
+	fs.BoolVar(&conf.S3ForcePathStyle, "s3.force_path_style", s3ForcePathStyleDefault, "s3 force path style")
+	fs.BoolVar(&conf.S3V4Signature, "s3.v4_signature", conf.S3V4Signature, "s3 V4 signature")
+	s3SupportTaggingDefault := true
+	if existing {
+		s3SupportTaggingDefault = conf.S3SupportTagging
+	}
+	fs.BoolVar(&conf.S3SupportTagging, "s3.support_tagging", s3SupportTaggingDefault, "s3 supportTagging")
+
+	fs.StringVar(&conf.GcsGoogleApplicationCredentials, "gcs.appCredentialsFile", conf.GcsGoogleApplicationCredentials, "google cloud storage credentials file, default to use env GOOGLE_APPLICATION_CREDENTIALS")
+	fs.StringVar(&conf.GcsProjectId, "gcs.projectId", conf.GcsProjectId, "google cloud storage project id, default to use env GOOGLE_CLOUD_PROJECT")
+
+	fs.StringVar(&conf.AzureAccountName, "azure.account_name", conf.AzureAccountName, "azure account name, default to use env AZURE_STORAGE_ACCOUNT")
+	fs.StringVar(&conf.AzureAccountKey, "azure.account_key", conf.AzureAccountKey, "azure account key, default to use env AZURE_STORAGE_ACCESS_KEY. Leave empty to authenticate with Entra ID")
+	fs.StringVar(&conf.AzureClientId, "azure.client_id", conf.AzureClientId, "azure user-assigned identity to authenticate, when no account key is given. Workload identity also reads env AZURE_TENANT_ID and AZURE_FEDERATED_TOKEN_FILE")
+	fs.StringVar(&conf.AzureEndpoint, "azure.endpoint", conf.AzureEndpoint, "azure blob service url, for accounts outside the public cloud, e.g. https://xxx.blob.core.usgovcloudapi.net/")
+
+	fs.StringVar(&conf.BackblazeKeyId, "b2.key_id", conf.BackblazeKeyId, "backblaze keyID")
+	fs.StringVar(&conf.BackblazeApplicationKey, "b2.application_key", conf.BackblazeApplicationKey, "backblaze applicationKey. Note that your Master Application Key will not work with the S3 Compatible API. You must create a new key that is eligible for use. For more information: https://help.backblaze.com/hc/en-us/articles/360047425453")
+	fs.StringVar(&conf.BackblazeEndpoint, "b2.endpoint", conf.BackblazeEndpoint, "backblaze endpoint")
+	b2RegionDefault := "us-west-002"
+	if existing {
+		b2RegionDefault = conf.BackblazeRegion
+	}
+	fs.StringVar(&conf.BackblazeRegion, "b2.region", b2RegionDefault, "backblaze region")
+
+	fs.StringVar(&conf.AliyunAccessKey, "aliyun.access_key", conf.AliyunAccessKey, "Aliyun access key, default to use env ALICLOUD_ACCESS_KEY_ID")
+	fs.StringVar(&conf.AliyunSecretKey, "aliyun.secret_key", conf.AliyunSecretKey, "Aliyun secret key, default to use env ALICLOUD_ACCESS_KEY_SECRET")
+	fs.StringVar(&conf.AliyunEndpoint, "aliyun.endpoint", conf.AliyunEndpoint, "Aliyun endpoint")
+	fs.StringVar(&conf.AliyunRegion, "aliyun.region", conf.AliyunRegion, "Aliyun region")
+
+	fs.StringVar(&conf.TencentSecretId, "tencent.secret_id", conf.TencentSecretId, "Tencent Secret Id, default to use env COS_SECRETID")
+	fs.StringVar(&conf.TencentSecretKey, "tencent.secret_key", conf.TencentSecretKey, "Tencent secret key, default to use env COS_SECRETKEY")
+	fs.StringVar(&conf.TencentEndpoint, "tencent.endpoint", conf.TencentEndpoint, "Tencent endpoint")
+
+	fs.StringVar(&conf.BaiduAccessKey, "baidu.access_key", conf.BaiduAccessKey, "Baidu access key, default to use env BDCLOUD_ACCESS_KEY")
+	fs.StringVar(&conf.BaiduSecretKey, "baidu.secret_key", conf.BaiduSecretKey, "Baidu secret key, default to use env BDCLOUD_SECRET_KEY")
+	fs.StringVar(&conf.BaiduEndpoint, "baidu.endpoint", conf.BaiduEndpoint, "Baidu endpoint")
+	fs.StringVar(&conf.BaiduRegion, "baidu.region", conf.BaiduRegion, "Baidu region")
+
+	fs.StringVar(&conf.WasabiAccessKey, "wasabi.access_key", conf.WasabiAccessKey, "Wasabi access key")
+	fs.StringVar(&conf.WasabiSecretKey, "wasabi.secret_key", conf.WasabiSecretKey, "Wasabi secret key")
+	fs.StringVar(&conf.WasabiEndpoint, "wasabi.endpoint", conf.WasabiEndpoint, "Wasabi endpoint, see https://wasabi.com/wp-content/themes/wasabi/docs/API_Guide/index.html#t=topics%2Fapidiff-intro.htm")
+	fs.StringVar(&conf.WasabiRegion, "wasabi.region", conf.WasabiRegion, "Wasabi region")
+
+	fs.StringVar(&conf.FilebaseAccessKey, "filebase.access_key", conf.FilebaseAccessKey, "Filebase access key")
+	fs.StringVar(&conf.FilebaseSecretKey, "filebase.secret_key", conf.FilebaseSecretKey, "Filebase secret key")
+	fs.StringVar(&conf.FilebaseEndpoint, "filebase.endpoint", conf.FilebaseEndpoint, "Filebase endpoint, https://s3.filebase.com")
+
+	fs.StringVar(&conf.StorjAccessKey, "storj.access_key", conf.StorjAccessKey, "Storj access key")
+	fs.StringVar(&conf.StorjSecretKey, "storj.secret_key", conf.StorjSecretKey, "Storj secret key")
+	fs.StringVar(&conf.StorjEndpoint, "storj.endpoint", conf.StorjEndpoint, "Storj endpoint")
+
+	return
+}
+
+// maxRemoteConcurrency caps configured remote transfer concurrency so one
+// configuration cannot request an unbounded number of network workers.
+const maxRemoteConcurrency = 1024
+
+// applyTypeDefaults resets backend-specific fields to their new-config defaults
+// for conf.Type, used when the storage type changes on update.
+func (c *commandRemoteConfigure) applyTypeDefaults(conf *remote_pb.RemoteConf) {
+	conf.S3Region = ""
+	conf.S3ForcePathStyle = false
+	conf.S3SupportTagging = false
+	conf.S3V4Signature = false
+	conf.BackblazeRegion = ""
+	if conf.Type == "s3" {
+		conf.S3Region = "us-east-2"
+		conf.S3ForcePathStyle = true
+		conf.S3SupportTagging = true
+	} else if conf.Type == "b2" {
+		conf.BackblazeRegion = "us-west-002"
+	}
+}
+
+func applyConcurrency(conf *remote_pb.RemoteConf, upload, download int) error {
+	if upload < 0 || upload > maxRemoteConcurrency {
+		return fmt.Errorf("upload_concurrency must be between 0 and %d", maxRemoteConcurrency)
+	}
+	if download < 0 || download > maxRemoteConcurrency {
+		return fmt.Errorf("download_concurrency must be between 0 and %d", maxRemoteConcurrency)
+	}
+	conf.UploadConcurrency = uint32(upload)
+	conf.DownloadConcurrency = uint32(download)
+	return nil
+}
+
+func (c *commandRemoteConfigure) loadRemoteStorageConf(commandEnv *CommandEnv, name string) (*remote_pb.RemoteConf, error) {
+	var conf *remote_pb.RemoteConf
+	err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		content, readErr := filer.ReadInsideFiler(context.Background(), client, filer.DirectoryEtcRemote, name+filer.REMOTE_STORAGE_CONF_SUFFIX)
+		if readErr != nil {
+			return readErr
+		}
+		conf = &remote_pb.RemoteConf{}
+		if unmarshalErr := proto.Unmarshal(content, conf); unmarshalErr != nil {
+			return fmt.Errorf("unmarshal %s/%s: %v", filer.DirectoryEtcRemote, name, unmarshalErr)
+		}
+		return nil
+	})
+	return conf, err
+}
+
 func (c *commandRemoteConfigure) listExistingRemoteStorages(commandEnv *CommandEnv, writer io.Writer) error {
 
 	return filer_pb.ReadDirAllEntries(context.Background(), commandEnv, util.FullPath(filer.DirectoryEtcRemote), "", func(entry *filer_pb.Entry, isLast bool) error {

weed/shell/command_volume_tier_compact.go

@@ -29,8 +29,8 @@ func (c *commandVolumeTierCompact) Name() string {
 func (c *commandVolumeTierCompact) Help() string {
 	return `compact remote volumes to reclaim space on cloud storage
 
-	volume.tier.compact [-volumeId=<volume_id>]
-	volume.tier.compact [-collection=""] [-garbageThreshold=0.3]
+	volume.tier.compact [-volumeId=<volume_id>] [-concurrency=<n>]
+	volume.tier.compact [-collection=""] [-garbageThreshold=0.3] [-concurrency=<n>]
 
 	e.g.:
 	volume.tier.compact -volumeId=7
@@ -64,10 +64,15 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri
 	volumeId := tierCommand.Int("volumeId", 0, "the volume id")
 	collection := tierCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty matches the collection with no name")
 	garbageThreshold := tierCommand.Float64("garbageThreshold", 0.3, "compact when garbage ratio exceeds this value")
+	concurrency := tierCommand.Int("concurrency", 0, "multipart transfer concurrency (0 = backend default)")
 	if err = tierCommand.Parse(args); err != nil {
 		return nil
 	}
 
+	if err = validateTierConcurrency(*concurrency); err != nil {
+		return err
+	}
+
 	if err = commandEnv.confirmIsLocked(args); err != nil {
 		return
 	}
@@ -107,7 +112,7 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri
 
 	var failedCount int
 	for _, rv := range remoteVolumes {
-		if err = doVolumeTierCompact(commandEnv, writer, rv, *garbageThreshold); err != nil {
+		if err = doVolumeTierCompact(commandEnv, writer, rv, *garbageThreshold, *concurrency); err != nil {
 			fmt.Fprintf(writer, "error compacting volume %d: %v\n", rv.vid, err)
 			failedCount++
 		}
@@ -194,7 +199,7 @@ func collectRemoteVolumesWithInfo(topoInfo *master_pb.TopologyInfo, collectionPa
 	return result, nil
 }
 
-func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolumeInfo, garbageThreshold float64) error {
+func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolumeInfo, garbageThreshold float64, concurrency int) error {
 	grpcDialOption := commandEnv.option.GrpcDialOption
 
 	// step 1: check garbage level
@@ -214,7 +219,7 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu
 	// step 2: download .dat from remote to local
 	// this deletes the remote file and reloads the volume as local, then re-uploads below
 	fmt.Fprintf(writer, "  downloading volume %d from %s to local...\n", rv.vid, rv.remoteStorageName)
-	err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, false)
+	err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, false, concurrency)
 	if err != nil {
 		return fmt.Errorf("download volume %d from remote: %v", rv.vid, err)
 	}
@@ -227,7 +232,7 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu
 		// upload the uncompacted volume back to restore cloud tier state
 		fmt.Fprintf(writer, "  compaction failed: %v\n", err)
 		fmt.Fprintf(writer, "  re-uploading volume %d to %s without compaction...\n", rv.vid, rv.remoteStorageName)
-		uploadErr := uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false)
+		uploadErr := uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false, concurrency)
 		if uploadErr != nil {
 			return fmt.Errorf("compaction failed (%v) and re-upload also failed (%v), volume %d remains local",
 				err, uploadErr, rv.vid)
@@ -238,7 +243,7 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu
 
 	// step 4: upload compacted volume back to remote
 	fmt.Fprintf(writer, "  uploading compacted volume %d to %s...\n", rv.vid, rv.remoteStorageName)
-	err = uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false)
+	err = uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false, concurrency)
 	if err != nil {
 		return fmt.Errorf("upload compacted volume %d to %s: %v (volume remains local with compacted data)",
 			rv.vid, rv.remoteStorageName, err)

weed/shell/command_volume_tier_download.go

@@ -32,7 +32,7 @@ func (c *commandVolumeTierDownload) Help() string {
 	return `download the dat file of a volume from a remote tier
 
 	volume.tier.download [-collection=""]
-	volume.tier.download [-collection=""] -volumeId=<volume_id>
+	volume.tier.download [-collection=""] -volumeId=<volume_id> [-concurrency=<n>]
 
 	The -collection parameter supports regular expressions for pattern matching:
 	  - Use exact match: volume.tier.download -collection="^mybucket$"
@@ -41,6 +41,7 @@ func (c *commandVolumeTierDownload) Help() string {
 
 	e.g.:
 	volume.tier.download -volumeId=7
+	volume.tier.download -volumeId=7 -concurrency=1
 
 	This command will download the dat file of a volume from a remote tier to a volume server in local cluster.
 
@@ -56,10 +57,15 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr
 	tierCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	volumeId := tierCommand.Int("volumeId", 0, "the volume id")
 	collection := tierCommand.String("collection", "", "comma-separated collection names, wildcards, or regex patterns; empty matches the collection with no name")
+	concurrency := tierCommand.Int("concurrency", 0, "multipart download concurrency (0 = backend default)")
 	if err = tierCommand.Parse(args); err != nil {
 		return nil
 	}
 
+	if err = validateTierConcurrency(*concurrency); err != nil {
+		return err
+	}
+
 	if err = commandEnv.confirmIsLocked(args); err != nil {
 		return
 	}
@@ -74,7 +80,7 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr
 
 	// volumeId is provided
 	if vid != 0 {
-		return doVolumeTierDownload(commandEnv, writer, *collection, vid)
+		return doVolumeTierDownload(commandEnv, writer, *collection, vid, *concurrency)
 	}
 
 	// apply to all volumes in the collection
@@ -85,7 +91,7 @@ func (c *commandVolumeTierDownload) Do(args []string, commandEnv *CommandEnv, wr
 	}
 	fmt.Printf("tier download volumes: %v\n", volumeIds)
 	for _, vid := range volumeIds {
-		if err = doVolumeTierDownload(commandEnv, writer, *collection, vid); err != nil {
+		if err = doVolumeTierDownload(commandEnv, writer, *collection, vid, *concurrency); err != nil {
 			return err
 		}
 	}
@@ -118,7 +124,7 @@ func collectRemoteVolumes(topoInfo *master_pb.TopologyInfo, collectionPattern st
 	return
 }
 
-func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId) (err error) {
+func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, concurrency int) (err error) {
 	// find volume location
 	locations, found := commandEnv.MasterClient.GetLocationsClone(uint32(vid))
 	if !found {
@@ -131,7 +137,7 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection s
 	for i, loc := range locations {
 		keepRemote := i < len(locations)-1
 		// copy the .dat file from remote tier to local
-		err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress(), keepRemote)
+		err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress(), keepRemote, concurrency)
 		if err != nil {
 			// A replica already made local by a prior interrupted run is not a
 			// failure; skip it so the remaining remote replicas still download.
@@ -146,13 +152,14 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection s
 	return nil
 }
 
-func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress, keepRemote bool) error {
+func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress, keepRemote bool, concurrency int) error {
 
 	err := operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
 		stream, downloadErr := volumeServerClient.VolumeTierMoveDatFromRemote(context.Background(), &volume_server_pb.VolumeTierMoveDatFromRemoteRequest{
 			VolumeId:          uint32(volumeId),
 			Collection:        collection,
 			KeepRemoteDatFile: keepRemote,
+			Concurrency:       int32(concurrency),
 		})
 
 		var lastProcessed int64

weed/shell/command_volume_tier_upload.go

@@ -5,6 +5,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"math"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -21,6 +22,21 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/wdclient"
 )
 
+// maxTierConcurrency caps per-request multipart transfer concurrency.
+const maxTierConcurrency = 1024
+
+// validateTierConcurrency rejects values that would wrap when narrowed to the
+// int32 proto field or exhaust volume-server resources. 0 means backend default.
+func validateTierConcurrency(n int) error {
+	if n < 0 || n > math.MaxInt32 {
+		return fmt.Errorf("concurrency must be between 0 and %d, got %d", math.MaxInt32, n)
+	}
+	if n > maxTierConcurrency {
+		return fmt.Errorf("concurrency must be at most %d, got %d", maxTierConcurrency, n)
+	}
+	return nil
+}
+
 func init() {
 	Commands = append(Commands, &commandVolumeTierUpload{})
 }
@@ -36,11 +52,12 @@ func (c *commandVolumeTierUpload) Help() string {
 	return `upload the dat file of a volume to a remote tier
 
 	volume.tier.upload [-collection=""] [-fullPercent=95] [-quietFor=1h]
-	volume.tier.upload [-collection=""] -volumeId=<volume_id> -dest=<storage_backend> [-keepLocalDatFile]
+	volume.tier.upload [-collection=""] -volumeId=<volume_id> -dest=<storage_backend> [-keepLocalDatFile] [-concurrency=<n>]
 
 	e.g.:
 	volume.tier.upload -volumeId=7 -dest=s3
 	volume.tier.upload -volumeId=7 -dest=s3.default
+	volume.tier.upload -volumeId=7 -dest=s3.telegram -concurrency=1
 
 	The <storage_backend> is defined in master.toml.
 	For example, "s3.default" in [storage.backend.s3.default]
@@ -78,10 +95,15 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ
 	dest := tierCommand.String("dest", "", "the target tier name")
 	keepLocalDatFile := tierCommand.Bool("keepLocalDatFile", false, "whether keep local dat file")
 	disk := tierCommand.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
+	concurrency := tierCommand.Int("concurrency", 0, "multipart upload concurrency (0 = backend default)")
 	if err = tierCommand.Parse(args); err != nil {
 		return nil
 	}
 
+	if err = validateTierConcurrency(*concurrency); err != nil {
+		return err
+	}
+
 	if err = commandEnv.confirmIsLocked(args); err != nil {
 		return
 	}
@@ -90,7 +112,7 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ
 
 	// volumeId is provided
 	if vid != 0 {
-		return doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile)
+		return doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile, *concurrency)
 	}
 
 	var diskType *types.DiskType
@@ -107,7 +129,7 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ
 	}
 	fmt.Printf("tier upload volumes: %v\n", volumeIds)
 	for _, vid := range volumeIds {
-		if err = doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile); err != nil {
+		if err = doVolumeTierUpload(commandEnv, writer, *collection, vid, *dest, *keepLocalDatFile, *concurrency); err != nil {
 			return err
 		}
 	}
@@ -115,7 +137,7 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ
 	return nil
 }
 
-func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, dest string, keepLocalDatFile bool) (err error) {
+func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, dest string, keepLocalDatFile bool, concurrency int) (err error) {
 	// find volume location
 	topoInfo, _, err := collectTopologyInfo(commandEnv, 0)
 	if err != nil {
@@ -137,7 +159,7 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str
 	}
 
 	// copy the .dat file to remote tier
-	err = uploadDatToRemoteTier(commandEnv.option.GrpcDialOption, writer, vid, collection, existingLocations[0].ServerAddress(), dest, keepLocalDatFile)
+	err = uploadDatToRemoteTier(commandEnv.option.GrpcDialOption, writer, vid, collection, existingLocations[0].ServerAddress(), dest, keepLocalDatFile, concurrency)
 	if err != nil {
 		return fmt.Errorf("copy dat file for volume %d on %s to %s: %v", vid, existingLocations[0].Url, dest, err)
 	}
@@ -190,7 +212,7 @@ func collectVolumeTierUploadLocations(topoInfo *master_pb.TopologyInfo, vid need
 	return append(tiered, local...)
 }
 
-func uploadDatToRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, sourceVolumeServer pb.ServerAddress, dest string, keepLocalDatFile bool) error {
+func uploadDatToRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, sourceVolumeServer pb.ServerAddress, dest string, keepLocalDatFile bool, concurrency int) error {
 
 	err := operation.WithVolumeServerClient(true, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
 		stream, copyErr := volumeServerClient.VolumeTierMoveDatToRemote(context.Background(), &volume_server_pb.VolumeTierMoveDatToRemoteRequest{
@@ -198,6 +220,7 @@ func uploadDatToRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, vol
 			Collection:             collection,
 			DestinationBackendName: dest,
 			KeepLocalDatFile:       keepLocalDatFile,
+			Concurrency:            int32(concurrency),
 		})
 
 		if stream == nil {

weed/storage/backend/backend.go

@@ -25,8 +25,9 @@ type BackendStorageFile interface {
 type BackendStorage interface {
 	ToProperties() map[string]string
 	NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) BackendStorageFile
-	CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error)
-	DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error)
+	// concurrency > 0 caps concurrent network transfers; <= 0 uses the backend default.
+	CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error)
+	DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error)
 	DeleteFile(key string) (err error)
 }
 

weed/storage/backend/rclone_backend/rclone_backend.go

@@ -103,7 +103,7 @@ func (s *RcloneBackendStorage) NewStorageFile(key string, tierInfo *volume_serve
 	return f
 }
 
-func (s *RcloneBackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
+func (s *RcloneBackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) {
 	randomUuid, err := uuid.NewRandom()
 	if err != nil {
 		return key, 0, err
@@ -154,7 +154,7 @@ func uploadViaRclone(rfs fs.Fs, filename string, key string, fn func(progressed
 	return obj.Size(), err
 }
 
-func (s *RcloneBackendStorage) DownloadFile(filename string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) {
+func (s *RcloneBackendStorage) DownloadFile(filename string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) {
 	glog.V(1).Infof("download dat file of %s from remote rclone.%s as %s", filename, s.id, key)
 
 	util.Retry("download via Rclone", func() error {

weed/storage/backend/s3_backend/s3_backend.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -41,6 +42,8 @@ type S3BackendStorage struct {
 	endpoint              string
 	storageClass          string
 	forcePathStyle        bool
+	uploadConcurrency     int
+	downloadConcurrency   int
 	conn                  s3iface.S3API
 }
 
@@ -54,6 +57,8 @@ func newS3BackendStorage(configuration backend.StringProperties, configPrefix st
 	s.endpoint = configuration.GetString(configPrefix + "endpoint")
 	s.storageClass = configuration.GetString(configPrefix + "storage_class")
 	s.forcePathStyle = util.ParseBool(configuration.GetString(configPrefix+"force_path_style"), true)
+	s.uploadConcurrency = parseConcurrency(configuration.GetString(configPrefix+"upload_concurrency"), defaultUploadConcurrency)
+	s.downloadConcurrency = parseConcurrency(configuration.GetString(configPrefix+"download_concurrency"), defaultDownloadConcurrency)
 	if s.storageClass == "" {
 		s.storageClass = "STANDARD_IA"
 	}
@@ -73,9 +78,27 @@ func (s *S3BackendStorage) ToProperties() map[string]string {
 	m["endpoint"] = s.endpoint
 	m["storage_class"] = s.storageClass
 	m["force_path_style"] = util.BoolToString(s.forcePathStyle)
+	m["upload_concurrency"] = strconv.Itoa(s.uploadConcurrency)
+	m["download_concurrency"] = strconv.Itoa(s.downloadConcurrency)
 	return m
 }
 
+const (
+	defaultUploadConcurrency   = 5
+	defaultDownloadConcurrency = 5
+)
+
+func parseConcurrency(value string, def int) int {
+	if value == "" {
+		return def
+	}
+	if n, err := strconv.Atoi(value); err == nil && n > 0 {
+		return n
+	}
+	glog.Warningf("invalid concurrency value %q, using default %d", value, def)
+	return def
+}
+
 func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile {
 	if strings.HasPrefix(key, "/") {
 		key = key[1:]
@@ -90,25 +113,32 @@ func (s *S3BackendStorage) NewStorageFile(key string, tierInfo *volume_server_pb
 	return f
 }
 
-func (s *S3BackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
+func (s *S3BackendStorage) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) {
 	randomUuid, _ := uuid.NewRandom()
 	key = randomUuid.String()
 
+	if concurrency <= 0 {
+		concurrency = s.uploadConcurrency
+	}
+
 	glog.V(1).Infof("copying dat file of %s to remote s3.%s as %s", f.Name(), s.id, key)
 
 	util.Retry("upload to S3", func() error {
-		size, err = uploadToS3(s.conn, f.Name(), s.bucket, key, s.storageClass, fn)
+		size, err = uploadToS3(s.conn, f.Name(), s.bucket, key, s.storageClass, fn, concurrency)
 		return err
 	})
 
 	return
 }
 
-func (s *S3BackendStorage) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) {
+func (s *S3BackendStorage) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) {
+	if concurrency <= 0 {
+		concurrency = s.downloadConcurrency
+	}
 
 	glog.V(1).Infof("download dat file of %s from remote s3.%s as %s", fileName, s.id, key)
 
-	size, err = downloadFromS3(s.conn, fileName, s.bucket, key, fn)
+	size, err = downloadFromS3(s.conn, fileName, s.bucket, key, fn, concurrency)
 
 	return
 }

weed/storage/backend/s3_backend/s3_backend_test.go

@@ -12,6 +12,12 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
+type testProperties map[string]string
+
+func (m testProperties) GetString(key string) string {
+	return m[key]
+}
+
 type stubS3Client struct {
 	s3iface.S3API
 	getObject func(*s3.GetObjectInput) (*s3.GetObjectOutput, error)
@@ -101,3 +107,50 @@ func TestReadAtRejectsInvalidRequestsLocally(t *testing.T) {
 		})
 	}
 }
+
+func TestParseConcurrency(t *testing.T) {
+	cases := []struct {
+		value    string
+		def      int
+		expected int
+	}{
+		{"", 5, 5},    // unset -> default
+		{"0", 5, 5},   // explicit zero means default
+		{"-3", 5, 5},  // invalid -> default
+		{"abc", 5, 5}, // invalid -> default
+		{"1", 5, 1},   // override
+		{"64", 5, 64}, // override
+	}
+	for _, tt := range cases {
+		if got := parseConcurrency(tt.value, tt.def); got != tt.expected {
+			t.Errorf("parseConcurrency(%q, %d) = %d, want %d", tt.value, tt.def, got, tt.expected)
+		}
+	}
+}
+
+func TestS3BackendStorageConcurrencyConfigRoundTrip(t *testing.T) {
+	s, err := newS3BackendStorage(testProperties{}, "", "test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if s.uploadConcurrency != defaultUploadConcurrency || s.downloadConcurrency != defaultDownloadConcurrency {
+		t.Fatalf("defaults: upload=%d download=%d, want %d/%d",
+			s.uploadConcurrency, s.downloadConcurrency, defaultUploadConcurrency, defaultDownloadConcurrency)
+	}
+
+	s, err = newS3BackendStorage(testProperties{
+		"upload_concurrency":   "1",
+		"download_concurrency": "17",
+	}, "", "test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if s.uploadConcurrency != 1 || s.downloadConcurrency != 17 {
+		t.Fatalf("configured: upload=%d download=%d, want 1/17", s.uploadConcurrency, s.downloadConcurrency)
+	}
+
+	props := s.ToProperties()
+	if props["upload_concurrency"] != "1" || props["download_concurrency"] != "17" {
+		t.Fatalf("ToProperties: %v", props)
+	}
+}

weed/storage/backend/s3_backend/s3_download.go

@@ -14,7 +14,10 @@ import (
 )
 
 func downloadFromS3(sess s3iface.S3API, destFileName string, sourceBucket string, sourceKey string,
-	fn func(progressed int64, percentage float32) error) (fileSize int64, err error) {
+	fn func(progressed int64, percentage float32) error, concurrency int) (fileSize int64, err error) {
+	if concurrency <= 0 {
+		concurrency = defaultDownloadConcurrency
+	}
 
 	fileSize, err = getFileSize(sess, sourceBucket, sourceKey)
 	if err != nil {
@@ -31,7 +34,7 @@ func downloadFromS3(sess s3iface.S3API, destFileName string, sourceBucket string
 	// Create a downloader with the session and custom options
 	downloader := s3manager.NewDownloaderWithClient(sess, func(u *s3manager.Downloader) {
 		u.PartSize = int64(64 * 1024 * 1024)
-		u.Concurrency = 5
+		u.Concurrency = concurrency
 	})
 
 	fileWriter := &s3DownloadProgressedWriter{

weed/storage/backend/s3_backend/s3_upload.go

@@ -12,7 +12,10 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 )
 
-func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey string, storageClass string, fn func(progressed int64, percentage float32) error) (fileSize int64, err error) {
+func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey string, storageClass string, fn func(progressed int64, percentage float32) error, concurrency int) (fileSize int64, err error) {
+	if concurrency <= 0 {
+		concurrency = defaultUploadConcurrency
+	}
 
 	//open the file
 	f, err := os.Open(filename)
@@ -36,7 +39,7 @@ func uploadToS3(sess s3iface.S3API, filename string, destBucket string, destKey
 	// Create an uploader with the session and custom options
 	uploader := s3manager.NewUploaderWithClient(sess, func(u *s3manager.Uploader) {
 		u.PartSize = partSize
-		u.Concurrency = 5
+		u.Concurrency = concurrency
 	})
 
 	fileReader := &s3UploadProgressedReader{

weed/storage/remote_tier_integration_test.go

@@ -52,7 +52,7 @@ func (b *localDirBackend) NewStorageFile(key string, tierInfo *volume_server_pb.
 	return &localDirBackendFile{backend: b, key: key, tierInfo: tierInfo}
 }
 
-func (b *localDirBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) {
+func (b *localDirBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (key string, size int64, err error) {
 	key = fmt.Sprintf("obj-%d-%d", time.Now().UnixNano(), os.Getpid())
 	dst := filepath.Join(b.root, key)
 	out, err := os.Create(dst)
@@ -73,7 +73,7 @@ func (b *localDirBackend) CopyFile(f *os.File, fn func(progressed int64, percent
 	return key, written, nil
 }
 
-func (b *localDirBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) {
+func (b *localDirBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error, concurrency int) (size int64, err error) {
 	src := filepath.Join(b.root, key)
 	in, err := os.Open(src)
 	if err != nil {
@@ -193,7 +193,7 @@ func tierUpVolumeLive(t *testing.T, dir string, vid needle.VolumeId, b *localDir
 	diskFile, ok := v.DataBackend.(*backend.DiskFile)
 	require.True(t, ok, "expected on-disk backend before tier-up")
 
-	uploadKey, size, err := b.CopyFile(diskFile.File, nil)
+	uploadKey, size, err := b.CopyFile(diskFile.File, nil, 0)
 	require.NoError(t, err)
 
 	bType, bId := backend.BackendNameToTypeId(testBackendName)
@@ -422,7 +422,7 @@ func TestRemoteTier_ECEncodeDecode_AfterDownload(t *testing.T) {
 
 	baseFileName := filepath.Join(dir, fmt.Sprintf("%d", uint32(vid)))
 	datPath := baseFileName + ".dat"
-	_, err := b.DownloadFile(datPath, key, nil)
+	_, err := b.DownloadFile(datPath, key, nil, 0)
 	require.NoError(t, err)
 
 	require.NoError(t, erasure_coding.WriteSortedFileFromIdx(baseFileName, ".ecx"))