F miran seprate upload download worker

cmd/migration_cmd.go

@@ -26,7 +26,6 @@ var (
 	region                     string
 	migrateToPath              string
 	duplicateSuffix            string
-	concurrency                int
 	encrypt                    bool
 	resume                     bool
 	skip                       int // 0 --> Replace; 1 --> Skip; 2 --> Duplicate
@@ -59,7 +58,6 @@ func init() {
 	migrateCmd.Flags().BoolVar(&deleteSource, "delete-source", false, "Delete object in s3 that is migrated to dStorage")
 	migrateCmd.Flags().StringVar(&awsCredPath, "aws-cred-path", "", "File Path to aws credentials")
 	migrateCmd.Flags().StringVar(&workDir, "wd", filepath.Join(util.GetHomeDir(), ".s3migration"), "Working directory")
-	migrateCmd.Flags().IntVar(&concurrency, "concurrency", 10, "number of concurrent files to process concurrently during migration")
 	migrateCmd.Flags().BoolVar(&resume, "resume", false, "pass this option to resume migration from previous state")
 	migrateCmd.Flags().IntVar(&skip, "skip", 1, "0 --> Replace existing files; 1 --> Skip migration; 2 --> Duplicate")
 	migrateCmd.Flags().IntVar(&retryCount, "retry", 3, "retry count for upload to dstorage")
@@ -186,7 +184,6 @@ var migrateCmd = &cobra.Command{
 			AllocationID:    allocationId,
 			Region:          region,
 			Skip:            skip,
-			Concurrency:     concurrency,
 			Bucket:          bucket,
 			Prefix:          prefix,
 			MigrateToPath:   migrateToPath,

migration/migrate.go

@@ -50,9 +50,6 @@ type Migration struct {
 	skip       int
 	retryCount int
 
-	//Number of goroutines to run. So at most concurrency * Batch goroutines will run. i.e. for bucket level and object level
-	concurrency int
-
 	szCtMu               sync.Mutex //size and count mutex; used to update migratedSize and totalMigratedObjects
 	migratedSize         uint64
 	totalMigratedObjects uint64
@@ -100,7 +97,6 @@ func InitMigration(mConfig *MigrationConfig) error {
 		zStore:        dStorageService,
 		awsStore:      awsStorageService,
 		skip:          mConfig.Skip,
-		concurrency:   mConfig.Concurrency,
 		retryCount:    mConfig.RetryCount,
 		stateFilePath: mConfig.StateFilePath,
 		migrateTo:     mConfig.MigrateToPath,
@@ -199,6 +195,7 @@ func (m *Migration) DownloadWorker(ctx context.Context, migrator *MigrationWorke
 			DoneChan:  make(chan struct{}, 1),
 			ErrChan:   make(chan error, 1),
 		}
+		migrator.PushToDownloadQueue(downloadObjMeta)
 
 		go func() {
 			defer wg.Done()
@@ -209,6 +206,7 @@ func (m *Migration) DownloadWorker(ctx context.Context, migrator *MigrationWorke
 			}
 			if downloadObjMeta.IsFileAlreadyExist && migration.skip == Skip {
 				zlogger.Logger.Info("Skipping migration of object" + downloadObjMeta.ObjectKey)
+				downloadObjMeta.DoneChan <- struct{}{}
 				return
 			}
 			migrator.DownloadStart(downloadObjMeta)
@@ -241,6 +239,8 @@ func (m *Migration) UploadWorker(ctx context.Context, migrator *MigrationWorker)
 			ErrChan:   make(chan error, 1),
 			Size:      downloadObj.Size,
 		}
+		migrator.PushToUploadQueue(uploadObj)
+
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
@@ -249,6 +249,10 @@ func (m *Migration) UploadWorker(ctx context.Context, migrator *MigrationWorker)
 				migrator.SetMigrationError(err)
 				return
 			}
+			if downloadObj.LocalPath == "" {
+				uploadObj.DoneChan <- struct{}{}
+				return
+			}
 			defer func() {
 				_ = m.fs.Remove(downloadObj.LocalPath)
 			}()

migration/migrateConfig.go

@@ -5,7 +5,6 @@ import "time"
 type MigrationConfig struct {
 	AllocationID    string
 	Skip            int
-	Concurrency     int
 	Bucket          string
 	Region          string
 	Prefix          string

migration/migration_worker.go

@@ -84,10 +84,13 @@ func (m *MigrationWorker) PauseUpload() {
 	}
 }
 
+func (m *MigrationWorker) PushToUploadQueue(u *UploadObjectMeta) {
+	m.uploadQueue <- u
+}
+
 func (m *MigrationWorker) UploadStart(u *UploadObjectMeta) {
 	m.incrUploadConcurrency()
 	atomic.AddInt64(&m.currentUploadSize, u.Size)
-	m.uploadQueue <- u
 }
 
 func (m *MigrationWorker) UploadDone(u *UploadObjectMeta, err error) {
@@ -96,8 +99,9 @@ func (m *MigrationWorker) UploadDone(u *UploadObjectMeta, err error) {
 	atomic.AddInt64(&m.currentUploadSize, -u.Size)
 	if err != nil {
 		u.ErrChan <- err
+	} else {
+		u.DoneChan <- struct{}{}
 	}
-	u.DoneChan <- struct{}{}
 }
 
 func (m *MigrationWorker) CloseUploadQueue() {
@@ -126,9 +130,12 @@ func (m *MigrationWorker) PauseDownload() {
 	}
 }
 
+func (m *MigrationWorker) PushToDownloadQueue(d *DownloadObjectMeta) {
+	m.downloadQueue <- d
+}
+
 func (m *MigrationWorker) DownloadStart(d *DownloadObjectMeta) {
 	m.incrDownloadConcurrency()
-	m.downloadQueue <- d
 	m.updateFileSizeOnDisk(d.Size)
 	atomic.AddInt64(&m.currentDownloadSize, d.Size)
 }

util/util.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"time"
 
+	zlogger "github.com/0chain/s3migration/logger"
 	"github.com/mitchellh/go-homedir"
 	"github.com/spf13/viper"
 )
@@ -111,7 +112,7 @@ func ConvertGoSDKTimeToTime(in string) time.Time {
 func Retry(attempts int, sleep time.Duration, f func() error) (err error) {
 	for i := 0; i < attempts; i++ {
 		if err = f(); err != nil {
-			log.Println("retrying after error:", err)
+			zlogger.Logger.Error("retrying after error: ", err)
 			time.Sleep(sleep)
 			continue
 		}