- Listing of source objects no longer blocks the transfer.

- S3 folders are not counted in the batch totals.
- Dial connection errors are retried.
- TCP connections are closed for small requests.

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.6.09/bp_linux.tar.gz
+wget -O bp_linux.tar.gz https://github.com/Azure/blobporter/releases/download/v0.6.10/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.6.09/bp_windows.zip)
+Download [BlobPorter.exe](https://github.com/Azure/blobporter/releases/download/v0.6.10/bp_windows.zip)
 
 Set environment variables (if using the command prompt):
 

blobporter.go

@@ -10,13 +10,12 @@ import (
 	"strconv"
 	"sync/atomic"
 
+	"github.com/Azure/blobporter/internal"
 	"github.com/Azure/blobporter/pipeline"
 	"github.com/Azure/blobporter/transfer"
 	"github.com/Azure/blobporter/util"
-	"github.com/Azure/blobporter/internal"
 )
 
-
 var argsUtil paramParserValidator
 
 func init() {
@@ -95,33 +94,29 @@ func init() {
 var dataTransferred uint64
 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)
-		var totalSize uint64
-		summary := ""
-
-		for _, source := range sourcesInfo {
-			//if the source is URL, remove the QS
-			display := source.SourceName
-			if u, err := url.Parse(source.SourceName); err == nil {
-				display = fmt.Sprintf("%v%v", u.Hostname(), u.Path)
-			}
-			summary = summary + fmt.Sprintf("Source: %v Size:%v \n", display, source.Size)
-			totalSize = totalSize + source.Size
-		}
+func displayFilesToTransfer(sourcesInfo []pipeline.SourceInfo) {
+	fmt.Printf("\nFiles to Transfer (%v) :\n", argsUtil.params.transferType)
+	var totalSize uint64
+	summary := ""
 
-		if len(sourcesInfo) < 10 {
-			fmt.Printf(summary)
-			return
+	for _, source := range sourcesInfo {
+		//if the source is URL, remove the QS
+		display := source.SourceName
+		if u, err := url.Parse(source.SourceName); err == nil {
+			display = fmt.Sprintf("%v%v", u.Hostname(), u.Path)
 		}
+		summary = summary + fmt.Sprintf("Source: %v Size:%v \n", display, source.Size)
+		totalSize = totalSize + source.Size
+	}
 
-		fmt.Printf("%v files. Total size:%v\n", len(sourcesInfo), totalSize)
-
+	if len(sourcesInfo) < 10 {
+		fmt.Printf(summary)
 		return
 	}
 
-	fmt.Printf("\nBatch transfer (%v).\nFiles per Batch: %v.\nBatch: %v of %v\n ", argsUtil.params.transferType, len(sourcesInfo), batchNumber+1, numOfBatches)
+	fmt.Printf("%v files. Total size:%v\n", len(sourcesInfo), totalSize)
+
+	return
 }
 
 func main() {
@@ -141,12 +136,17 @@ func main() {
 
 	stats := transfer.NewStats(argsUtil.params.numberOfWorkers, argsUtil.params.numberOfReaders)
 
-	for b, sourcePipeline := range sourcePipelines {
-		sourcesInfo := sourcePipeline.GetSourcesInfo()
+	for sourcePipeline := range sourcePipelines {
+
+		if sourcePipeline.Err != nil {
+			log.Fatal(sourcePipeline.Err)
+		}
+
+		sourcesInfo := sourcePipeline.Source.GetSourcesInfo()
 
-		tfer := transfer.NewTransfer(&sourcePipeline, &targetPipeline, argsUtil.params.numberOfReaders, argsUtil.params.numberOfWorkers, argsUtil.params.blockSize)
+		tfer := transfer.NewTransfer(&sourcePipeline.Source, &targetPipeline, argsUtil.params.numberOfReaders, argsUtil.params.numberOfWorkers, argsUtil.params.blockSize)
 
-		displayFilesToTransfer(sourcesInfo, len(sourcePipelines), b)
+		displayFilesToTransfer(sourcesInfo)
 		pb := getProgressBarDelegate(tfer.TotalSize, argsUtil.params.quietMode)
 
 		tfer.StartTransfer(argsUtil.params.dedupeLevel, pb)

internal/azutil.go

@@ -11,7 +11,7 @@ import (
 	"os"
 	"syscall"
 	"time"
-
+	"github.com/Azure/blobporter/util"
 	"github.com/Azure/azure-pipeline-go/pipeline"
 
 	"github.com/Azure/azure-storage-blob-go/2016-05-31/azblob"
@@ -190,6 +190,7 @@ func (p *AzUtil) PutBlockBlob(blobName string, body io.ReadSeeker, md5 []byte) e
 
 	h := azblob.BlobHTTPHeaders{}
 
+
 	//16 is md5.Size
 	if len(md5) != 16 {
 		var md5bytes [16]byte
@@ -375,6 +376,24 @@ func isWinsockTimeOutError(err error) net.Error {
 	return nil
 }
 
+func isDialConnectError(err error) net.Error {
+	if uerr, ok := err.(*url.Error); ok {
+		if derr, ok := uerr.Err.(*net.OpError); ok {
+			if serr, ok := derr.Err.(*os.SyscallError); ok && serr.Syscall == "connect" {
+				return &retriableError{error: err}
+			}
+		}
+	}
+	return nil
+}
+
+func isRetriableDialError(err error) net.Error {
+	if derr := isWinsockTimeOutError(err); derr != nil {
+		return derr
+	}
+	return isDialConnectError(err)
+}
+
 type retriableError struct {
 	error
 }
@@ -387,11 +406,19 @@ func (*retriableError) Temporary() bool {
 	return true
 }
 
+const tcpKeepOpenMinLength = 8 * int64(util.MB) 
+
 func (p *clientPolicy) Do(ctx context.Context, request pipeline.Request) (pipeline.Response, error) {
-	r, err := pipelineHTTPClient.Do(request.WithContext(ctx))
+	req := request.WithContext(ctx)
+	
+	if req.ContentLength < tcpKeepOpenMinLength {
+		req.Close=true
+	}
+
+	r, err := pipelineHTTPClient.Do(req)
 	pipresp := pipeline.NewHTTPResponse(r)
 	if err != nil {
-		if derr := isWinsockTimeOutError(err); derr != nil {
+		if derr := isRetriableDialError(err); derr != nil {
 			return pipresp, derr
 		}
 		err = pipeline.NewError(err, "HTTP request failed")
@@ -411,9 +438,9 @@ func newpipelineHTTPClient() *http.Client {
 				KeepAlive: 30 * time.Second,
 				DualStack: true,
 			}).Dial,
-			MaxIdleConns:           0,
+			MaxIdleConns:           100,
 			MaxIdleConnsPerHost:    100,
-			IdleConnTimeout:        90 * time.Second,
+			IdleConnTimeout:        60 * time.Second,
 			TLSHandshakeTimeout:    10 * time.Second,
 			ExpectContinueTimeout:  1 * time.Second,
 			DisableKeepAlives:      false,

internal/const.go

@@ -6,7 +6,7 @@ import (
 )
 
 //ProgramVersion blobporter version
-const ProgramVersion = "0.6.09"
+const ProgramVersion = "0.6.10"
 
 //HTTPClientTimeout  HTTP client timeout when reading from HTTP sources and try timeout for blob storage operations.
 var HTTPClientTimeout = 90

pipelinefactory.go

@@ -9,11 +9,11 @@ import (
 	"github.com/Azure/blobporter/transfer"
 )
 
-func newTransferPipelines(params *validatedParameters) ([]pipeline.SourcePipeline, pipeline.TargetPipeline, error) {
+func newTransferPipelines(params *validatedParameters) (<-chan sources.FactoryResult, pipeline.TargetPipeline, error) {
 
 	fact := newPipelinesFactory(params)
 
-	var sourcesp []pipeline.SourcePipeline
+	var sourcesp <-chan sources.FactoryResult
 	var targetp pipeline.TargetPipeline
 	var err error
 
@@ -72,7 +72,7 @@ func (p *pipelinesFactory) newTargetPipeline() (pipeline.TargetPipeline, error)
 	return nil, fmt.Errorf("Invalid target segment:%v", p.target)
 }
 
-func (p *pipelinesFactory) newSourcePipelines() ([]pipeline.SourcePipeline, error) {
+func (p *pipelinesFactory) newSourcePipelines() (<-chan sources.FactoryResult, error) {
 
 	params, err := p.newSourceParams()
 
@@ -83,18 +83,19 @@ func (p *pipelinesFactory) newSourcePipelines() ([]pipeline.SourcePipeline, erro
 	switch p.source {
 	case transfer.File:
 		params := params.(sources.FileSystemSourceParams)
-		return sources.NewFileSystemSourcePipeline(&params), nil
+		return sources.NewFileSystemSourcePipelineFactory(&params), nil
 	case transfer.HTTP:
 		params := params.(sources.HTTPSourceParams)
-		return []pipeline.SourcePipeline{sources.NewHTTPSourcePipeline(params.SourceURIs, params.TargetAliases, params.SourceParams.CalculateMD5)}, nil
+		return sources.NewHTTPSourcePipelineFactory(params), nil
 	case transfer.S3:
 		params := params.(sources.S3Params)
-		return sources.NewS3SourcePipeline(&params), nil
+		return sources.NewS3SourcePipelineFactory(&params), nil
 	case transfer.Blob:
 		params := params.(sources.AzureBlobParams)
-		return sources.NewAzureBlobSourcePipeline(&params), nil
+		return sources.NewAzBlobSourcePipelineFactory(&params), nil
 	case transfer.Perf:
-		return sources.NewPerfSourcePipeline(params.(sources.PerfSourceParams)), nil
+		params := params.(sources.PerfSourceParams)
+		return sources.NewPerfSourcePipelineFactory(params), nil
 	}
 
 	return nil, fmt.Errorf("Invalid source segment:%v", p.source)

sources/azblobinfo.go

@@ -1,7 +1,6 @@
 package sources
 
 import (
-	"fmt"
 	"log"
 	"path"
 	"time"
@@ -39,35 +38,52 @@ func newazBlobInfoProvider(params *AzureBlobParams) *azBlobInfoProvider {
 	return &azBlobInfoProvider{params: params, azUtil: azutil}
 }
 
-//getSourceInfo gets a list of SourceInfo that represent the list of azure blobs returned by the service
-// based on the provided criteria (container/prefix). If the exact match flag is set, then a specific match is
-// performed instead of the prefix. Marker semantics are also honored so a complete list is expected
-func (b *azBlobInfoProvider) getSourceInfo() ([]pipeline.SourceInfo, error) {
-	var err error
+func (b *azBlobInfoProvider) toSourceInfo(obj *azblob.Blob) (*pipeline.SourceInfo, error) {
 	exp := b.params.SasExp
 	if exp == 0 {
 		exp = defaultSasExpHours
 	}
 	date := time.Now().Add(time.Duration(exp) * time.Minute).UTC()
-	sourceURIs := make([]pipeline.SourceInfo, 0)
+
+	sourceURLWithSAS := b.azUtil.GetBlobURLWithReadOnlySASToken(obj.Name, date)
+
+	targetAlias := obj.Name
+	if !b.params.KeepDirStructure {
+		targetAlias = path.Base(obj.Name)
+	}
+
+	return &pipeline.SourceInfo{
+		SourceName:  sourceURLWithSAS.String(),
+		Size:        uint64(*obj.Properties.ContentLength),
+		TargetAlias: targetAlias}, nil
+}
+
+func (b *azBlobInfoProvider) listObjects(filter SourceFilter) <-chan ObjectListingResult {
+	sources := make(chan ObjectListingResult, 2)
+	list := make([]pipeline.SourceInfo, 0)
+	bsize := 0
 
 	blobCallback := func(blob *azblob.Blob, prefix string) (bool, error) {
 		include := true
 		if b.params.UseExactNameMatch {
 			include = blob.Name == prefix
 		}
-		if include {
-			sourceURLWithSAS := b.azUtil.GetBlobURLWithReadOnlySASToken(blob.Name, date)
+		if include && filter.IsIncluded(blob.Name) {
+
+			si, err := b.toSourceInfo(blob)
+
+			if err != nil {
+				return true, err
+			}
+
+			list = append(list, *si)
 
-			targetAlias := blob.Name
-			if !b.params.KeepDirStructure {
-				targetAlias = path.Base(blob.Name)
+			if bsize++; bsize == b.params.FilesPerPipeline {
+				sources <- ObjectListingResult{Sources: list}
+				list = make([]pipeline.SourceInfo, 0)
+				bsize = 0
 			}
 
-			sourceURIs = append(sourceURIs, pipeline.SourceInfo{
-				SourceName:  sourceURLWithSAS.String(),
-				Size:        uint64(*blob.Properties.ContentLength),
-				TargetAlias: targetAlias})
 			if b.params.UseExactNameMatch {
 				//true, stops iteration
 				return true, nil
@@ -78,19 +94,21 @@ func (b *azBlobInfoProvider) getSourceInfo() ([]pipeline.SourceInfo, error) {
 		return false, nil
 	}
 
-	for _, blobName := range b.params.BlobNames {
-		if err = b.azUtil.IterateBlobList(blobName, blobCallback); err != nil {
-			return nil, err
+	go func() {
+		for _, blobName := range b.params.BlobNames {
+			if err := b.azUtil.IterateBlobList(blobName, blobCallback); err != nil {
+				sources <- ObjectListingResult{Err: err}
+				return
+			}
+			if bsize > 0 {
+				sources <- ObjectListingResult{Sources: list}
+				list = make([]pipeline.SourceInfo, 0)
+				bsize = 0
+			}
 		}
-	}
+		close(sources)
 
-	if len(sourceURIs) == 0 {
-		nameMatchMode := "prefix"
-		if b.params.UseExactNameMatch {
-			nameMatchMode = "name"
-		}
-		return nil, fmt.Errorf(" the %v %s did not match any blob names ", nameMatchMode, b.params.BlobNames)
-	}
+	}()
 
-	return sourceURIs, nil
+	return sources
 }