- handleman for target pipeline

- keep dir structure is now true by default
- remove -p, and added -i option
- debug mode for handleman

Author: giventocode

README.md

@@ -29,7 +29,7 @@ Sources and targets are decoupled, this design enables the composition of variou
 Download, extract and set permissions:
 
 ```bash
-wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.5.25/bp_linux.tar.gz
+wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.6.03/bp_linux.tar.gz
 tar -xvf bp_linux.tar.gz linux_amd64/blobporter
 chmod +x ~/linux_amd64/blobporter
 cd ~/linux_amd64
@@ -46,7 +46,7 @@ export ACCOUNT_KEY=<STORAGE_ACCOUNT_KEY>
 
 ### Windows
 
-Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.5.25/bp_windows.zip)
+Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.6.03/bp_windows.zip)
 
 Set environment variables (if using the command prompt):
 

args.go

@@ -65,7 +65,7 @@ type arguments struct {
 	quietMode                bool
 	calculateMD5             bool
 	exactNameMatch           bool
-	keepDirStructure         bool
+	removeDirStructure       bool
 	hTTPClientTimeout        int
 	numberOfHandlesPerFile   int //numberOfHandlesPerFile = defaultNumberOfHandlesPerFile
 	numberOfFilesInBatch     int //numberOfFilesInBatch = defaultNumberOfFilesInBatch
@@ -121,15 +121,15 @@ func newParamParserValidator() paramParserValidator {
 	var defaultNumberOfWorkers = runtime.NumCPU() * numOfWorkersFactor
 	var defaultNumberOfReaders = runtime.NumCPU() * numOfReadersFactor
 
-	if defaultNumberOfWorkers > 50 {
-		defaultNumberOfWorkers = 50
+	if defaultNumberOfWorkers > 60 {
+		defaultNumberOfWorkers = 60
 	}
-	if defaultNumberOfReaders > 80 {
-		defaultNumberOfReaders = 80
+	if defaultNumberOfReaders > 50 {
+		defaultNumberOfReaders = 50
 	}
 
 	args := &arguments{
-		keepDirStructure: 		true,
+		removeDirStructure:     false,
 		numberOfReaders:        defaultNumberOfReaders,
 		numberOfWorkers:        defaultNumberOfWorkers,
 		blockSizeStr:           defaultBlockSizeStr,
@@ -192,7 +192,8 @@ func (p *paramParserValidator) getTargetRules() ([]parseAndValidationRule, error
 	switch p.targetSegment {
 	case transfer.File:
 		return []parseAndValidationRule{
-			p.pvNumOfHandlesPerFile}, nil
+			p.pvNumOfHandlesPerFile,
+			p.pvKeepDirStructueIsFalseWarning}, nil
 	case transfer.PageBlob:
 		return []parseAndValidationRule{
 			p.pvTargetContainerIsReq,
@@ -249,7 +250,7 @@ func (p *paramParserValidator) getSourceRules() ([]parseAndValidationRule, error
 
 //Global rules....
 func (p *paramParserValidator) pvgKeepDirectoryStructure() error {
-	p.params.keepDirStructure = p.args.keepDirStructure
+	p.params.keepDirStructure = !p.args.removeDirStructure
 	return nil
 }
 func (p *paramParserValidator) pvgQuietMode() error {
@@ -544,6 +545,14 @@ func (p *paramParserValidator) pvTargetBlobAuthInfoIsReq() error {
 	return nil
 }
 
+func (p *paramParserValidator) pvKeepDirStructueIsFalseWarning() error {
+	if !p.params.keepDirStructure && !p.params.quietMode {
+		fmt.Printf("Warning!\nThe directory structure from the source won't be created.\nFiles with the same file name will be overwritten.\n")
+	}
+
+	return nil
+}
+
 func (p *paramParserValidator) runParseAndValidationRules(rules ...parseAndValidationRule) error {
 	for i := 0; i < len(rules); i++ {
 		val := rules[i]

blobporter.go

@@ -9,14 +9,13 @@ import (
 	"os"
 	"strconv"
 	"sync/atomic"
-	"time"
 
 	"github.com/Azure/blobporter/pipeline"
 	"github.com/Azure/blobporter/transfer"
 	"github.com/Azure/blobporter/util"
 )
 
-const programVersion = "0.6.03"
+const programVersion = "0.6.05"
 
 var argsUtil paramParserValidator
 
@@ -46,7 +45,7 @@ func init() {
 		dupcheckLevelMsg           = "Desired level of effort to detect duplicate data to minimize upload size.\n\tMust be one of " + transfer.DupeCheckLevelStr
 		transferDefMsg             = "Defines the type of source and target in the transfer.\n\tMust be one of:\n\tfile-blockblob, file-pageblob, http-blockblob, http-pageblob, blob-file,\n\tpageblock-file (alias of blob-file), blockblob-file (alias of blob-file)\n\tor http-file."
 		exactNameMatchMsg          = "If set or true only blobs that match the name exactly will be downloaded."
-		keepDirStructureMsg        = "Keeps the directory structure from the source.\n\tNot applicable when the source is a HTTP endpoint."
+		removeDirStructureMsg      = "If set the directory structure from the source won't be kept.\n\tNot applicable when the source is a HTTP endpoint."
 		numberOfHandlersPerFileMsg = "Number of open handles for concurrent reads and writes per file."
 		numberOfFilesInBatchMsg    = "Maximum number of files in a transfer.\n\tIf the number is exceeded new transfers are created"
 	)
@@ -67,7 +66,7 @@ func init() {
 		util.PrintUsageDefaults("d", "dup_check_level", argsUtil.args.dedupeLevelOptStr, dupcheckLevelMsg)
 		util.PrintUsageDefaults("t", "transfer_definition", argsUtil.args.transferDefStr, transferDefMsg)
 		util.PrintUsageDefaults("e", "exact_name", "false", exactNameMatchMsg)
-		util.PrintUsageDefaults("p", "keep_directories", "true", keepDirStructureMsg)
+		util.PrintUsageDefaults("i", "remove_directories", "false", removeDirStructureMsg)
 		util.PrintUsageDefaults("h", "handles_per_file", strconv.Itoa(argsUtil.args.numberOfHandlesPerFile), numberOfHandlersPerFileMsg)
 		util.PrintUsageDefaults("x", "files_per_transfer", strconv.Itoa(argsUtil.args.numberOfFilesInBatch), numberOfFilesInBatchMsg)
 
@@ -88,7 +87,7 @@ func init() {
 	util.StringVarAlias(&argsUtil.args.dedupeLevelOptStr, "d", "dup_check_level", argsUtil.args.dedupeLevelOptStr, dupcheckLevelMsg)
 	util.StringVarAlias(&argsUtil.args.transferDefStr, "t", "transfer_definition", argsUtil.args.transferDefStr, transferDefMsg)
 	util.BoolVarAlias(&argsUtil.args.exactNameMatch, "e", "exact_name", false, exactNameMatchMsg)
-	util.BoolVarAlias(&argsUtil.args.keepDirStructure, "p", "keep_directories", true, keepDirStructureMsg)
+	util.BoolVarAlias(&argsUtil.args.removeDirStructure, "i", "remove_directories", false, removeDirStructureMsg)
 	util.IntVarAlias(&argsUtil.args.numberOfHandlesPerFile, "h", "handles_per_file", defaultNumberOfHandlesPerFile, numberOfHandlersPerFileMsg)
 	util.IntVarAlias(&argsUtil.args.numberOfFilesInBatch, "x", "files_per_transfer", defaultNumberOfFilesInBatch, numberOfFilesInBatchMsg)
 }
@@ -98,7 +97,7 @@ var targetRetries int32
 
 func displayFilesToTransfer(sourcesInfo []pipeline.SourceInfo, numOfBatches int, batchNumber int) {
 	if numOfBatches == 1 {
-		fmt.Printf("Files to Transfer (%v) :\n ", argsUtil.params.transferType)
+		fmt.Printf("Files to Transfer (%v) :\n", argsUtil.params.transferType)
 		var totalSize uint64
 		summary := ""
 
@@ -112,7 +111,7 @@ func displayFilesToTransfer(sourcesInfo []pipeline.SourceInfo, numOfBatches int,
 			totalSize = totalSize + source.Size
 		}
 
-		if len(sourcesInfo) < 50 {
+		if len(sourcesInfo) < 20 {
 			fmt.Printf(summary)
 			return
 		}
@@ -125,18 +124,6 @@ func displayFilesToTransfer(sourcesInfo []pipeline.SourceInfo, numOfBatches int,
 	fmt.Printf("\nBatch transfer (%v).\nFiles per Batch: %v.\nBatch: %v of %v\n ", argsUtil.params.transferType, len(sourcesInfo), batchNumber+1, numOfBatches)
 }
 
-func displayFinalWrapUpSummary(duration time.Duration, targetRetries int32, threadTarget int, totalNumberOfBlocks int, totalSize uint64, cumWriteDuration time.Duration) {
-	var netMB float64 = 1000000
-	fmt.Printf("\nThe transfer took %v to run.\n", duration)
-	MBs := float64(totalSize) / netMB / duration.Seconds()
-	fmt.Printf("Throughput: %1.2f MB/s (%1.2f Mb/s) \n", MBs, MBs*8)
-	fmt.Printf("Configuration: R=%d, W=%d, DataSize=%s, Blocks=%d\n",
-		argsUtil.params.numberOfReaders, argsUtil.params.numberOfWorkers, util.PrintSize(totalSize), totalNumberOfBlocks)
-	fmt.Printf("Cumulative Writes Duration: Total=%v, Avg Per Worker=%v\n",
-		cumWriteDuration, time.Duration(cumWriteDuration.Nanoseconds()/int64(argsUtil.params.numberOfWorkers)))
-	fmt.Printf("Retries: Avg=%v Total=%v\n", float32(targetRetries)/float32(totalNumberOfBlocks), targetRetries)
-}
-
 func main() {
 
 	flag.Parse()

files.sh

@@ -0,0 +1,4 @@
+#! /bin/bash
+for n in {51..200}; do
+    dd if=/dev/urandom of=file$( printf %03d "$n" ).bin bs=1 count=$(( 50000 * 1024 )) &
+done

sources/http.go

@@ -3,9 +3,11 @@ package sources
 import (
 	"io/ioutil"
 	"log"
+	"net"
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"fmt"
 	"net/http"
@@ -55,7 +57,7 @@ func newHTTPSource(sourceListManager objectListManager, pipelineFactory sourceHT
 			numOfFilesInBatch = filesSent
 		}
 
-		httpSource := HTTPPipeline{Sources: sourceInfos[start : start+numOfFilesInBatch], HTTPClient: util.NewHTTPClient(), includeMD5: includeMD5}
+		httpSource := HTTPPipeline{Sources: sourceInfos[start : start+numOfFilesInBatch], HTTPClient: httpSourceHTTPClient, includeMD5: includeMD5}
 
 		pipelines[b], err = pipelineFactory(httpSource)
 
@@ -92,7 +94,7 @@ func NewHTTP(sourceURIs []string, targetAliases []string, md5 bool) pipeline.Sou
 			TargetAlias: targetAlias,
 			SourceName:  sourceURIs[i]}
 	}
-	return &HTTPPipeline{Sources: sources, HTTPClient: util.NewHTTPClient(), includeMD5: md5}
+	return &HTTPPipeline{Sources: sources, HTTPClient: httpSourceHTTPClient, includeMD5: md5}
 }
 
 func getSourceSize(sourceURI string) (size int) {
@@ -207,7 +209,6 @@ func (f *HTTPPipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartition, p
 				if res != nil && res.Body != nil {
 					res.Body.Close()
 				}
-				f.HTTPClient = util.NewHTTPClient()
 
 				util.PrintfIfDebug("ExecuteReader -> blockid:%v toread:%v status:%v err:%v head:%v", p.BlockID, p.BytesToRead, status, err, header)
 
@@ -267,3 +268,29 @@ func (f *HTTPPipeline) ConstructBlockInfoQueue(blockSize uint64) (partitionsQ ch
 
 	return
 }
+
+const (
+	maxIdleConns        = 50
+	maxIdleConnsPerHost = 50
+)
+
+var httpSourceHTTPClient = newSourceHTTPClient() 
+
+func newSourceHTTPClient() *http.Client {
+	return  &http.Client{
+			Timeout: time.Duration(util.HTTPClientTimeout) * time.Second,
+			Transport: &http.Transport{
+				Dial: (&net.Dialer{
+					Timeout:   30 * time.Second, // dial timeout
+					KeepAlive: 30 * time.Second,
+					DualStack: true,
+				}).Dial,
+				MaxIdleConns:           maxIdleConns,
+				MaxIdleConnsPerHost:    maxIdleConnsPerHost,
+				IdleConnTimeout:        90 * time.Second,
+				TLSHandshakeTimeout:    30 * time.Second,
+				ExpectContinueTimeout:  1 * time.Second,
+				DisableKeepAlives:      false,
+				DisableCompression:     false,
+				MaxResponseHeaderBytes: 0}}
+}