feat: support parallel upload of files in dir.

s3/s3.go

@@ -12,6 +12,7 @@ import (
 	"path"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
@@ -31,7 +32,7 @@ type S3Client interface {
 
 	// PutDirectory puts a complete directory into a bucket key prefix, with each file in the directory
 	// a separate key in the bucket.
-	PutDirectory(bucket, key, path string) error
+	PutDirectory(bucket, key, path string, maxParallel ...int) error
 
 	// GetFile downloads a file to a local file path
 	GetFile(bucket, key, path string) error
@@ -279,12 +280,34 @@ func generatePutTasks(keyPrefix, rootPath string) chan uploadTask {
 
 // PutDirectory puts a complete directory into a bucket key prefix, with each file in the directory
 // a separate key in the bucket.
-func (s *s3client) PutDirectory(bucket, key, path string) error {
-	for putTask := range generatePutTasks(key, path) {
-		err := s.PutFile(bucket, putTask.key, putTask.path)
-		if err != nil {
-			return err
-		}
+func (s *s3client) PutDirectory(bucket, key, path string, maxParallel ...int) error {
+	parallel := 10 // Default number of concurrencies
+	if len(maxParallel) > 0 {
+		parallel = maxParallel[0]
+	}
+	parallelNum := make(chan string, parallel)
+	tasks := generatePutTasks(key, path)
+	errCh := make(chan error, len(tasks))
+	var wg sync.WaitGroup
+	for putTask := range tasks {
+		parallelNum <- putTask.key
+		wg.Add(1)
+		go func(putTask uploadTask) {
+			defer wg.Done()
+			err := s.PutFile(bucket, putTask.key, putTask.path)
+			if err != nil {
+				log.WithFields(log.Fields{"endpoint": s.Endpoint, "bucket": bucket, "key": putTask.key, "path": putTask.path}).Error("Error uploading file to s3")
+				errCh <- err
+			}
+			<-parallelNum
+		}(putTask)
+	}
+	close(parallelNum)
+	wg.Wait()
+	close(errCh)
+	err := errorFromChannel(errCh)
+	if err != nil {
+		return err
 	}
 	return nil
 }
@@ -523,3 +546,12 @@ func parseKMSEncCntx(kmsEncCntx string) (*string, error) {
 
 	return &parsedKMSEncryptionContext, nil
 }
+
+func errorFromChannel(errCh <-chan error) error {
+	select {
+	case err := <-errCh:
+		return err
+	default:
+	}
+	return nil
+}