0.6.20rc

Author: giventocode

.gitignore

@@ -12,6 +12,7 @@ _build/linux_amd64
 _build/windows_amd64
 _wd
 _old
+_vendor*
 .vscode
 *_testdata
 

.travis.yml

@@ -1,13 +1,16 @@
 sudo: required
 language: go
 go:
-- 1.9
+- 1.11
 before_install:
 - sudo apt-get install zip
-- go get github.com/tools/godep
+# Download the binary to bin folder in $GOPATH
+- curl -L -s https://github.com/golang/dep/releases/download/v${DEP_VERSION}/dep-linux-amd64 -o $GOPATH/bin/dep
+# Make the binary executable
+- chmod +x $GOPATH/bin/dep
 - go get -d github.com/stretchr/testify/assert
 install:
-- godep restore
+# - dep ensure
 script:
 - go test -v ./...
 - mkdir linux_amd64

Gopkg.lock

@@ -0,0 +1,46 @@
+# Gopkg.toml example
+#
+# Refer to https://golang.github.io/dep/docs/Gopkg.toml.html
+# for detailed Gopkg.toml documentation.
+#
+# required = ["github.com/user/thing/cmd/thing"]
+# ignored = ["github.com/user/project/pkgX", "bitbucket.org/user/project/pkgA/pkgY"]
+#
+# [[constraint]]
+#   name = "github.com/user/project"
+#   version = "1.0.0"
+#
+# [[constraint]]
+#   name = "github.com/user/project2"
+#   branch = "dev"
+#   source = "github.com/myfork/project2"
+#
+# [[override]]
+#   name = "github.com/x/y"
+#   version = "2.4.0"
+#
+# [prune]
+#   non-go = false
+#   go-tests = true
+#   unused-packages = true
+
+
+[[constraint]]
+  name = "github.com/Azure/azure-pipeline-go"
+  version = "0.1.7"
+
+[[constraint]]
+  branch = "master"
+  name = "github.com/Azure/azure-storage-blob-go"
+
+[[constraint]]
+  name = "github.com/minio/minio-go"
+  version = "4.0.6"
+
+[[constraint]]
+  name = "github.com/stretchr/testify"
+  version = "1.2.2"
+
+[prune]
+  go-tests = true
+  unused-packages = true

Gopkg.toml

@@ -43,6 +43,12 @@ type paramParserValidator struct {
 	params        *validatedParameters
 	sourceSegment transfer.TransferSegment
 	targetSegment transfer.TransferSegment
+	defaultValues defaults
+}
+
+type defaults struct {
+	defaultNumberOfReaders int
+	defaultNumberOfWorkers int
 }
 
 //respresents raw (not validated) the list of options/flags as received from the user
@@ -96,6 +102,7 @@ type validatedParameters struct {
 	blobTarget             blobParams
 	perfSourceDefinitions  []sources.SourceDefinition
 	tracker                *internal.TransferTracker
+	referenceMode          bool
 }
 
 type s3Source struct {
@@ -108,12 +115,13 @@ type s3Source struct {
 }
 
 type blobParams struct {
-	accountName string
-	accountKey  string
-	container   string
-	prefixes    []string
-	baseBlobURL string
-	sasExpMin   int
+	accountName   string
+	accountKey    string
+	container     string
+	prefixes      []string
+	baseBlobURL   string
+	sasExpMin     int
+	useServerSide bool
 }
 
 func (b blobParams) isSourceAuthAndContainerInfoSet() bool {
@@ -143,21 +151,28 @@ func newParamParserValidator() paramParserValidator {
 		transferDefStr:         string(transfer.FileToBlock),
 		numberOfHandlesPerFile: defaultNumberOfHandlesPerFile,
 		hTTPClientTimeout:      defaultHTTPClientTimeout,
-		numberOfFilesInBatch:   defaultNumberOfFilesInBatch}
+		numberOfFilesInBatch:   defaultNumberOfFilesInBatch,
+	}
 	params := &validatedParameters{
 		s3Source: s3Source{
 			preSignedExpMin: defaultReadTokenExp},
 		blobSource: blobParams{
 			sasExpMin: defaultReadTokenExp},
 		blobTarget: blobParams{}}
 
-	p := paramParserValidator{args: args, params: params}
+	p := paramParserValidator{args: args,
+		params: params,
+		defaultValues: defaults{
+			defaultNumberOfReaders: defaultNumberOfReaders,
+			defaultNumberOfWorkers: defaultNumberOfWorkers,
+		},
+	}
 
 	return p
 }
 
 func (p *paramParserValidator) parseAndValidate() error {
-	//Run global rules.. this will set the transfer type
+	//Run global rules.. this will set the transfer type which is needed in some of the global rules.
 	err := p.pvgTransferType()
 	if err != nil {
 		return err
@@ -166,6 +181,7 @@ func (p *paramParserValidator) parseAndValidate() error {
 	p.sourceSegment = s
 	p.targetSegment = t
 	err = p.runParseAndValidationRules(
+		p.pvgSetReferenceModeAndServerSide,
 		p.pvgCalculateReadersAndWorkers,
 		p.pvgTransferStatusPathIsPresent,
 		p.pvgBatchLimits,
@@ -180,7 +196,7 @@ func (p *paramParserValidator) parseAndValidate() error {
 		return err
 	}
 
-	//get and run target rules...
+	//get and run source and target rules...
 	var rules []parseAndValidationRule
 	if rules, err = p.getSourceRules(); err != nil {
 		return err
@@ -260,6 +276,16 @@ func (p *paramParserValidator) getSourceRules() ([]parseAndValidationRule, error
 //**************************
 
 //Global rules....
+func (p *paramParserValidator) pvgSetReferenceModeAndServerSide() error {
+	if p.targetSegment == transfer.BlockBlob {
+		if p.sourceSegment == transfer.Blob {
+			p.params.blobTarget.useServerSide = true
+			p.params.referenceMode = true
+		}
+	}
+	return nil
+}
+
 func (p *paramParserValidator) pvgUseExactMatch() error {
 	p.params.useExactMatch = p.args.exactNameMatch
 	return nil
@@ -302,6 +328,23 @@ func (p *paramParserValidator) pvgCalculateReadersAndWorkers() error {
 	p.params.numberOfWorkers = p.args.numberOfWorkers
 	p.params.numberOfReaders = p.args.numberOfReaders
 
+	//if using the default parameters and the transfer is blob-blockblob change
+	// params to maximes the transfer considering that this will be server-side sync
+	if p.defaultValues.defaultNumberOfReaders == p.params.numberOfReaders &&
+		p.defaultValues.defaultNumberOfWorkers == p.params.numberOfWorkers {
+
+		if p.params.transferType == transfer.BlobToBlock {
+			workers := 75 * runtime.NumCPU()
+
+			if workers > 200 {
+				workers = 200
+			}
+			p.params.numberOfWorkers = workers
+			p.params.numberOfReaders = 4
+		}
+
+	}
+
 	return nil
 
 }

args.go

@@ -131,34 +131,29 @@ func main() {
 	if err := argsUtil.parseAndValidate(); err != nil {
 		log.Fatal(err)
 	}
-
 	//Create pipelines
 	sourcePipelines, targetPipeline, err := newTransferPipelines(argsUtil.params)
 
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	stats := transfer.NewStats(argsUtil.params.numberOfWorkers, argsUtil.params.numberOfReaders)
-
+	prog := newProgressState(argsUtil.params.quietMode, argsUtil.params.numberOfReaders, argsUtil.params.numberOfWorkers)
 	for sourcePipeline := range sourcePipelines {
 
 		if sourcePipeline.Err != nil {
 			log.Fatal(sourcePipeline.Err)
 		}
 
 		sourcesInfo := sourcePipeline.Source.GetSourcesInfo()
-
 		tfer := transfer.NewTransfer(sourcePipeline.Source, targetPipeline, argsUtil.params.numberOfReaders, argsUtil.params.numberOfWorkers, argsUtil.params.blockSize)
 		tfer.SetTransferTracker(argsUtil.params.tracker)
-
-		displayFilesToTransfer(sourcesInfo)
-		pb := getProgressBarDelegate(tfer.TotalSize, argsUtil.params.quietMode)
-
-		tfer.StartTransfer(argsUtil.params.dedupeLevel, pb)
+		prog.newTransfer(float64(tfer.TotalSize), sourcesInfo, argsUtil.params.transferType)
+	
+		tfer.StartTransfer(argsUtil.params.dedupeLevel)
 		tfer.WaitForCompletion()
 
-		stats.AddTransferInfo(tfer.GetStats())
+	
 	}
 
 	if argsUtil.params.tracker != nil {
@@ -167,24 +162,25 @@ func main() {
 		}
 	}
 
-	stats.DisplaySummary()
+	prog.displayGlobalSummary()
 
+	
 }
 
 func getProgressBarDelegate(totalSize uint64, quietMode bool) func(r pipeline.WorkerResult, committedCount int, bufferLevel int) {
 	dataTransferred = 0
 	targetRetries = 0
 	if quietMode || totalSize == 0 {
 		return func(r pipeline.WorkerResult, committedCount int, bufferLevel int) {
-			atomic.AddInt32(&targetRetries, int32(r.Stats.Retries))
 			dataTransferred = dataTransferred + uint64(r.BlockSize)
 		}
 
 	}
 	return func(r pipeline.WorkerResult, committedCount int, bufferLevel int) {
 
-		atomic.AddInt32(&targetRetries, int32(r.Stats.Retries))
-
+		if r.Stats != nil {
+			atomic.AddInt32(&targetRetries, int32(r.Stats.Retries))
+		}
 		dataTransferred = dataTransferred + uint64(r.BlockSize)
 		p := int(math.Ceil((float64(dataTransferred) / float64(totalSize)) * 100))
 		var ind string