- naming refactoring

Author: giventocode

README.md

@@ -208,6 +208,10 @@ By default files are downloaded to the same directory where you are running blob
 
 - `i`, `--remove_directories` *bool* if set blobs are downloaded or uploaded without keeping the directory structure of the source. Not applicable when the source is a HTTP endpoint.
 
+## Optimizing Transfers
+
+BlobPorter has several configuration options that can be used to miximize performance for specific use cases. 
+
 ## Performance Considerations
 
 By default, BlobPorter creates 5 readers and 8 workers for each core on the computer. You can overwrite these values by using the options -r (number of readers) and -g (number of workers). When overriding these options there are few considerations:

args.go

@@ -205,8 +205,7 @@ func (p *paramParserValidator) getTargetRules() ([]parseAndValidationRule, error
 			p.pvBlockSizeCheckForBlockBlobs,
 			p.pvTargetBlobAuthInfoIsReq}, nil
 	case transfer.Perf:
-		return []parseAndValidationRule{
-			p.pvPerfSourceIsReq}, nil
+		return []parseAndValidationRule{}, nil
 	}
 
 	return nil, fmt.Errorf("Invalid target segment type: %v ", p.targetSegment)

pipelinefactory.go

@@ -60,11 +60,11 @@ func (p *pipelinesFactory) newTargetPipeline() (pipeline.TargetPipeline, error)
 	switch p.target {
 	case transfer.File:
 		params := params.(targets.FileTargetParams)
-		return targets.NewMultiFile(params.Overwrite, params.NumberOfHandlesPerFile), nil
+		return targets.NewFileSystemTargetPipeline(params.Overwrite, params.NumberOfHandlesPerFile), nil
 	case transfer.BlockBlob:
-		return targets.NewAzureBlockPipeline(params.(targets.AzureTargetParams)), nil
+		return targets.NewAzureBlockTargetPipeline(params.(targets.AzureTargetParams)), nil
 	case transfer.PageBlob:
-		return targets.NewAzurePagePipeline(params.(targets.AzureTargetParams)), nil
+		return targets.NewAzurePageTargetPipeline(params.(targets.AzureTargetParams)), nil
 	case transfer.Perf:
 		return targets.NewPerfTargetPipeline(), nil
 	}
@@ -82,17 +82,17 @@ func (p *pipelinesFactory) newSourcePipelines() ([]pipeline.SourcePipeline, erro
 
 	switch p.source {
 	case transfer.File:
-		params := params.(sources.MultiFileParams)
-		return sources.NewMultiFile(&params), nil
+		params := params.(sources.FileSystemSourceParams)
+		return sources.NewFileSystemSourcePipeline(&params), nil
 	case transfer.HTTP:
 		params := params.(sources.HTTPSourceParams)
-		return []pipeline.SourcePipeline{sources.NewHTTP(params.SourceURIs, params.TargetAliases, params.SourceParams.CalculateMD5)}, nil
+		return []pipeline.SourcePipeline{sources.NewHTTPSourcePipeline(params.SourceURIs, params.TargetAliases, params.SourceParams.CalculateMD5)}, nil
 	case transfer.S3:
 		params := params.(sources.S3Params)
-		return sources.NewS3Pipeline(&params), nil
+		return sources.NewS3SourcePipeline(&params), nil
 	case transfer.Blob:
 		params := params.(sources.AzureBlobParams)
-		return sources.NewAzureBlob(&params), nil
+		return sources.NewAzureBlobSourcePipeline(&params), nil
 	case transfer.Perf:
 		return sources.NewPerfSourcePipeline(params.(sources.PerfSourceParams)), nil
 	}
@@ -103,7 +103,7 @@ func (p *pipelinesFactory) newSourceParams() (interface{}, error) {
 
 	switch p.source {
 	case transfer.File:
-		return sources.MultiFileParams{
+		return sources.FileSystemSourceParams{
 			SourcePatterns:   p.valParams.sourceURIs,
 			BlockSize:        p.valParams.blockSize,
 			TargetAliases:    p.valParams.targetAliases,

sources/http.go

@@ -22,14 +22,14 @@ import (
 
 const sasTokenNumberOfHours = 4
 
-// HTTPPipeline  constructs parts  channel and implements data readers for file exposed via HTTP
-type HTTPPipeline struct {
+// HTTPSource  constructs parts  channel and implements data readers for file exposed via HTTP
+type HTTPSource struct {
 	Sources    []pipeline.SourceInfo
 	HTTPClient *http.Client
 	includeMD5 bool
 }
 
-type sourceHTTPPipelineFactory func(httpSource HTTPPipeline) (pipeline.SourcePipeline, error)
+type sourceHTTPPipelineFactory func(httpSource HTTPSource) (pipeline.SourcePipeline, error)
 
 func newHTTPSource(sourceListManager objectListManager, pipelineFactory sourceHTTPPipelineFactory, numOfFilePerPipeline int, includeMD5 bool) ([]pipeline.SourcePipeline, error) {
 	var err error
@@ -57,7 +57,7 @@ func newHTTPSource(sourceListManager objectListManager, pipelineFactory sourceHT
 			numOfFilesInBatch = filesSent
 		}
 
-		httpSource := HTTPPipeline{Sources: sourceInfos[start : start+numOfFilesInBatch], HTTPClient: httpSourceHTTPClient, includeMD5: includeMD5}
+		httpSource := HTTPSource{Sources: sourceInfos[start : start+numOfFilesInBatch], HTTPClient: httpSourceHTTPClient, includeMD5: includeMD5}
 
 		pipelines[b], err = pipelineFactory(httpSource)
 
@@ -71,9 +71,9 @@ func newHTTPSource(sourceListManager objectListManager, pipelineFactory sourceHT
 	return pipelines, err
 }
 
-//NewHTTP creates a new instance of an HTTP source
+//NewHTTPSourcePipeline creates a new instance of an HTTP source
 //To get the file size, a HTTP HEAD request is issued and the Content-Length header is inspected.
-func NewHTTP(sourceURIs []string, targetAliases []string, md5 bool) pipeline.SourcePipeline {
+func NewHTTPSourcePipeline(sourceURIs []string, targetAliases []string, md5 bool) pipeline.SourcePipeline {
 	setTargetAlias := len(sourceURIs) == len(targetAliases)
 	sources := make([]pipeline.SourceInfo, len(sourceURIs))
 	for i := 0; i < len(sourceURIs); i++ {
@@ -94,7 +94,7 @@ func NewHTTP(sourceURIs []string, targetAliases []string, md5 bool) pipeline.Sou
 			TargetAlias: targetAlias,
 			SourceName:  sourceURIs[i]}
 	}
-	return &HTTPPipeline{Sources: sources, HTTPClient: httpSourceHTTPClient, includeMD5: md5}
+	return &HTTPSource{Sources: sources, HTTPClient: httpSourceHTTPClient, includeMD5: md5}
 }
 
 func getSourceSize(sourceURI string) (size int) {
@@ -163,14 +163,14 @@ func getSourceSizeFromByteRangeHeader(sourceURI string) (size int) {
 
 //GetSourcesInfo implements GetSourcesInfo from the pipeline.SourcePipeline Interface.
 //Returns an array of pipeline.SourceInfo[] with the files URL, alias and size.
-func (f *HTTPPipeline) GetSourcesInfo() []pipeline.SourceInfo {
+func (f *HTTPSource) GetSourcesInfo() []pipeline.SourceInfo {
 	return f.Sources
 }
 
 //ExecuteReader implements ExecuteReader from the pipeline.SourcePipeline Interface.
 //For each part the reader makes a byte range request to the source
 // starting from the part's Offset to BytesToRead - 1 (zero based).
-func (f *HTTPPipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, readPartsQ chan pipeline.Part, id int, wg *sync.WaitGroup) {
+func (f *HTTPSource) ExecuteReader(partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, readPartsQ chan pipeline.Part, id int, wg *sync.WaitGroup) {
 	var blocksHandled = 0
 	var err error
 	var req *http.Request
@@ -239,7 +239,7 @@ func (f *HTTPPipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartition, p
 
 //ConstructBlockInfoQueue implements GetSourcesInfo from the pipeline.SourcePipeline Interface.
 //Constructs the Part's channel arithmetically from the size of the sources.
-func (f *HTTPPipeline) ConstructBlockInfoQueue(blockSize uint64) (partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, numOfBlocks int, size uint64) {
+func (f *HTTPSource) ConstructBlockInfoQueue(blockSize uint64) (partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, numOfBlocks int, size uint64) {
 	allParts := make([][]pipeline.Part, len(f.Sources))
 	//disable memory buffer for parts (bufferQ == nil)
 	var bufferQ chan []byte
@@ -274,23 +274,23 @@ const (
 	maxIdleConnsPerHost = 50
 )
 
-var httpSourceHTTPClient = newSourceHTTPClient() 
+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}}
+	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}}
 }

sources/multifile.go

@@ -17,11 +17,11 @@ import (
 )
 
 ////////////////////////////////////////////////////////////
-///// MultiFilePipeline
+///// FileSystemSource
 ////////////////////////////////////////////////////////////
 
-// MultiFilePipeline  Contructs blocks queue and implements data readers
-type MultiFilePipeline struct {
+// FileSystemSource  Contructs blocks queue and implements data readers
+type FileSystemSource struct {
 	filesInfo           map[string]FileInfo
 	totalNumberOfBlocks int
 	totalSize           uint64
@@ -39,8 +39,8 @@ type FileInfo struct {
 	NumOfBlocks int
 }
 
-//MultiFileParams parameters used to create a new instance of multi-file source pipeline
-type MultiFileParams struct {
+//FileSystemSourceParams parameters used to create a new instance of multi-file source pipeline
+type FileSystemSourceParams struct {
 	SourcePatterns   []string
 	BlockSize        uint64
 	TargetAliases    []string
@@ -50,11 +50,10 @@ type MultiFileParams struct {
 	KeepDirStructure bool
 }
 
-// NewMultiFile creates a new MultiFilePipeline.
-// If the sourcePattern results in a single file, the targetAlias, if set, will be used as the target name.
-// Otherwise the full original file name will be used instead.
-//sourcePatterns []string, blockSize uint64, targetAliases []string, numOfPartitions int, md5 bool
-func NewMultiFile(params *MultiFileParams) []pipeline.SourcePipeline {
+// NewFileSystemSourcePipeline creates a new MultiFilePipeline.
+// If the sourcePattern results in a single file and the targetAlias is set, the alias will be used as the target name.
+// Otherwise the full original file name will be used.
+func NewFileSystemSourcePipeline(params *FileSystemSourceParams) []pipeline.SourcePipeline {
 	var files []string
 	var err error
 	//get files from patterns
@@ -148,7 +147,7 @@ func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint
 
 	handlePool := internal.NewFileHandlePool(maxNumOfHandlesPerFile, internal.Read, false)
 
-	return &MultiFilePipeline{filesInfo: fileInfos,
+	return &FileSystemSource{filesInfo: fileInfos,
 		totalNumberOfBlocks: totalNumberOfBlocks,
 		blockSize:           blockSize,
 		totalSize:           totalSize,
@@ -161,7 +160,7 @@ func newMultiFilePipeline(files []string, targetAliases []string, blockSize uint
 //ExecuteReader implements ExecuteReader from the pipeline.SourcePipeline Interface.
 //For each file the reader will maintain a open handle from which data will be read.
 // This implementation uses partitions (group of parts that can be read sequentially).
-func (f *MultiFilePipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, readPartsQ chan pipeline.Part, id int, wg *sync.WaitGroup) {
+func (f *FileSystemSource) ExecuteReader(partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, readPartsQ chan pipeline.Part, id int, wg *sync.WaitGroup) {
 	var err error
 	var partition pipeline.PartsPartition
 
@@ -230,7 +229,7 @@ func (f *MultiFilePipeline) ExecuteReader(partitionsQ chan pipeline.PartsPartiti
 
 //GetSourcesInfo implements GetSourcesInfo from the pipeline.SourcePipeline Interface.
 //Returns an an array of SourceInfo with the name, alias and size of the files to be transferred.
-func (f *MultiFilePipeline) GetSourcesInfo() []pipeline.SourceInfo {
+func (f *FileSystemSource) GetSourcesInfo() []pipeline.SourceInfo {
 
 	sources := make([]pipeline.SourceInfo, len(f.filesInfo))
 	var i = 0
@@ -272,7 +271,7 @@ func createPartsFromSource(size uint64, sourceNumOfBlocks int, blockSize uint64,
 //ConstructBlockInfoQueue implements ConstructBlockInfoQueue from the pipeline.SourcePipeline Interface.
 // this implementation uses partitions to group parts into a set that can be read sequentially.
 // This is to avoid Window's memory pressure when calling SetFilePointer numerous times on the same handle
-func (f *MultiFilePipeline) ConstructBlockInfoQueue(blockSize uint64) (partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, numOfBlocks int, size uint64) {
+func (f *FileSystemSource) ConstructBlockInfoQueue(blockSize uint64) (partitionsQ chan pipeline.PartsPartition, partsQ chan pipeline.Part, numOfBlocks int, size uint64) {
 	numOfBlocks = f.totalNumberOfBlocks
 	size = f.totalSize
 	allPartitions := make([][]pipeline.PartsPartition, len(f.filesInfo))

sources/ostorefactory.go

@@ -7,16 +7,16 @@ import (
 	"github.com/Azure/blobporter/pipeline"
 )
 
-//AzureBlob constructs parts channel and implements data readers for Azure Blobs exposed via HTTP
-type AzureBlob struct {
-	HTTPPipeline
+//AzureBlobSource constructs parts channel and implements data readers for Azure Blobs exposed via HTTP
+type AzureBlobSource struct {
+	HTTPSource
 	Container      string
 	BlobNames      []string
 	exactNameMatch bool
 }
 
-//NewAzureBlob creates a new instance of the HTTPPipeline for Azure Blobs
-func NewAzureBlob(params *AzureBlobParams) []pipeline.SourcePipeline {
+//NewAzureBlobSourcePipeline creates a new instance of the HTTPPipeline for Azure Blobs
+func NewAzureBlobSourcePipeline(params *AzureBlobParams) []pipeline.SourcePipeline {
 	var err error
 	var azObjStorage objectListManager
 	azObjStorage = newazBlobInfoProvider(params)
@@ -25,10 +25,10 @@ func NewAzureBlob(params *AzureBlobParams) []pipeline.SourcePipeline {
 		log.Fatal(fmt.Errorf("Invalid operation. The number of files per batch must be greater than zero"))
 	}
 
-	factory := func(httpSource HTTPPipeline) (pipeline.SourcePipeline, error) {
-		return &AzureBlob{Container: params.Container,
+	factory := func(httpSource HTTPSource) (pipeline.SourcePipeline, error) {
+		return &AzureBlobSource{Container: params.Container,
 			BlobNames:      params.BlobNames,
-			HTTPPipeline:   httpSource,
+			HTTPSource:     httpSource,
 			exactNameMatch: params.SourceParams.UseExactNameMatch}, nil
 	}
 
@@ -40,14 +40,14 @@ func NewAzureBlob(params *AzureBlobParams) []pipeline.SourcePipeline {
 	return pipelines
 }
 
-//S3Pipeline S3 source HTTP based pipeline
-type S3Pipeline struct {
-	HTTPPipeline
+//S3Source S3 source HTTP based pipeline
+type S3Source struct {
+	HTTPSource
 	exactNameMatch bool
 }
 
-//NewS3Pipeline creates a new instance of the HTTPPipeline for S3
-func NewS3Pipeline(params *S3Params) []pipeline.SourcePipeline {
+//NewS3SourcePipeline creates a new instance of the HTTPPipeline for S3
+func NewS3SourcePipeline(params *S3Params) []pipeline.SourcePipeline {
 	var err error
 	var s3ObjStorage objectListManager
 	s3ObjStorage, err = newS3InfoProvider(params)
@@ -60,9 +60,9 @@ func NewS3Pipeline(params *S3Params) []pipeline.SourcePipeline {
 		log.Fatal(fmt.Errorf("Invalid operation. The number of files per batch must be greater than zero"))
 	}
 
-	factory := func(httpSource HTTPPipeline) (pipeline.SourcePipeline, error) {
-		return &S3Pipeline{
-			HTTPPipeline:   httpSource,
+	factory := func(httpSource HTTPSource) (pipeline.SourcePipeline, error) {
+		return &S3Source{
+			HTTPSource:     httpSource,
 			exactNameMatch: params.SourceParams.UseExactNameMatch}, nil
 	}
 

targets/azureblock.go

@@ -14,14 +14,14 @@ import (
 ///// AzureBlock Target
 ////////////////////////////////////////////////////////////
 
-//AzureBlock represents an Azure Block target
-type AzureBlock struct {
+//AzureBlockTarget represents an Azure Block target
+type AzureBlockTarget struct {
 	container string
 	azutil    *util.AzUtil
 }
 
-//NewAzureBlockPipeline TODO
-func NewAzureBlockPipeline(params AzureTargetParams) pipeline.TargetPipeline {
+//NewAzureBlockTargetPipeline TODO
+func NewAzureBlockTargetPipeline(params AzureTargetParams) pipeline.TargetPipeline {
 
 	azutil, err := util.NewAzUtil(params.AccountName, params.AccountKey, params.Container, params.BaseBlobURL)
 
@@ -40,12 +40,12 @@ func NewAzureBlockPipeline(params AzureTargetParams) pipeline.TargetPipeline {
 		log.Fatal(err)
 	}
 
-	return &AzureBlock{azutil: azutil, container: params.Container}
+	return &AzureBlockTarget{azutil: azutil, container: params.Container}
 }
 
 //CommitList implements CommitList from the pipeline.TargetPipeline interface.
 //Commits the list of blocks to Azure Storage to finalize the transfer.
-func (t *AzureBlock) CommitList(listInfo *pipeline.TargetCommittedListInfo, numberOfBlocks int, targetName string) (msg string, err error) {
+func (t *AzureBlockTarget) CommitList(listInfo *pipeline.TargetCommittedListInfo, numberOfBlocks int, targetName string) (msg string, err error) {
 	startTime := time.Now()
 
 	//Only commit if the number blocks is greater than one.
@@ -80,7 +80,7 @@ func convertToBase64EncodedList(list interface{}, numOfBlocks int) []string {
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
 //Checks if uncommitted blocks are present and cleans them by creating an empty blob.
-func (t *AzureBlock) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
+func (t *AzureBlockTarget) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
 	numOfBlocks := int(source.Size+(blockSize-1)) / int(blockSize)
 
 	if numOfBlocks > util.MaxBlockCount { // more than 50,000 blocks needed, so can't work
@@ -94,7 +94,7 @@ func (t *AzureBlock) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize
 //ProcessWrittenPart implements ProcessWrittenPart from the pipeline.TargetPipeline interface.
 //Appends the written part to a list. If the part is duplicated the list is updated with a reference, to the first occurrence of the block.
 //If the first occurrence has not yet being processed, the part is requested to be placed back in the results channel (requeue == true).
-func (t *AzureBlock) ProcessWrittenPart(result *pipeline.WorkerResult, listInfo *pipeline.TargetCommittedListInfo) (requeue bool, err error) {
+func (t *AzureBlockTarget) ProcessWrittenPart(result *pipeline.WorkerResult, listInfo *pipeline.TargetCommittedListInfo) (requeue bool, err error) {
 	requeue = false
 	blockList := convertToBase64EncodedList((*listInfo).List, result.NumberOfBlocks)
 
@@ -116,7 +116,7 @@ func (t *AzureBlock) ProcessWrittenPart(result *pipeline.WorkerResult, listInfo
 
 //WritePart implements WritePart from the pipeline.TargetPipeline interface.
 //Performs a PUT block operation with the data contained in the part.
-func (t *AzureBlock) WritePart(part *pipeline.Part) (duration time.Duration, startTime time.Time, numOfRetries int, err error) {
+func (t *AzureBlockTarget) WritePart(part *pipeline.Part) (duration time.Duration, startTime time.Time, numOfRetries int, err error) {
 	startTime = time.Now()
 	defer func() { duration = time.Now().Sub(startTime) }()
 	util.PrintfIfDebug("WritePart -> blockid:%v read:%v name:%v err:%v", part.BlockID, len(part.Data), part.TargetAlias, err)