- v0.5.14:

- Fixes:
-- An error message is displayed insted of panic when when an invalid URL or protocol is used as source
-- Large page blobs (>200GB)  won't fail beacuse  to the block size check.

Author: giventocode

.gitignore

@@ -8,12 +8,13 @@ _obj
 _test
 _build/linux_amd64
 _build/windows_amd64
+.vscode
 
 # Architecture specific extensions/prefixes
 *.[568vq]
 [568vq].out
 debug
-.vscode
+launch.json
 *.cgo1.go
 *.cgo2.c
 _cgo_defun.c
@@ -30,4 +31,5 @@ _testmain.go
 *.out
 
 # external packages folder
-#vendor/
+#vendor/
+.vscode/launch.json

.vscode/launch.json

@@ -49,7 +49,7 @@ const (
 	// User can use environment variables to specify storage account information
 	storageAccountNameEnvVar = "ACCOUNT_NAME"
 	storageAccountKeyEnvVar  = "ACCOUNT_KEY"
-	programVersion           = "0.5.13" // version number to show in help
+	programVersion           = "0.5.14" // version number to show in help
 )
 
 const numOfWorkersFactor = 8
@@ -72,7 +72,7 @@ func init() {
 	)
 
 	const (
-		fileMsg                    = "URL, file or files (e.g. /data/*.gz) to upload."
+		fileMsg                    = "Source URL, file or files (e.g. /data/*.gz) to upload."
 		nameMsg                    = "Blob name (e.g. myblob.txt) or prefix for download scenarios."
 		containerNameMsg           = "Container name (e.g. mycontainer).\n\tIf the container does not exist, it will be created."
 		concurrentWorkersMsg       = "Number of workers for parallel upload."
@@ -93,7 +93,7 @@ func init() {
 	)
 
 	flag.Usage = func() {
-		util.PrintUsageDefaults("f", "file", "", fileMsg)
+		util.PrintUsageDefaults("f", "source_file", "", fileMsg)
 		util.PrintUsageDefaults("n", "name", "", nameMsg)
 		util.PrintUsageDefaults("c", "container_name", "", containerNameMsg)
 		util.PrintUsageDefaults("g", "concurrent_workers", strconv.Itoa(defaultNumberOfWorkers), concurrentWorkersMsg)
@@ -114,7 +114,7 @@ func init() {
 
 	}
 
-	util.StringListVarAlias(&sourceURIs, "f", "file", "", fileMsg)
+	util.StringListVarAlias(&sourceURIs, "f", "source_file", "", fileMsg)
 	util.StringListVarAlias(&blobNames, "n", "name", "", nameMsg)
 	util.StringVarAlias(&containerName, "c", "container_name", "", containerNameMsg)
 	util.IntVarAlias(&numberOfWorkers, "g", "concurrent_workers", defaultNumberOfWorkers, concurrentWorkersMsg)

blobporter.go

@@ -4,7 +4,6 @@ import (
 	"crypto/md5"
 	"encoding/base64"
 	"fmt"
-	"log"
 	"sync"
 	"time"
 
@@ -48,7 +47,7 @@ type SourceInfo struct {
 
 //TargetPipeline operations that abstract how parts a written and processed to a given target
 type TargetPipeline interface {
-	PreProcessSourceInfo(source *SourceInfo) (err error)
+	PreProcessSourceInfo(source *SourceInfo, blockSize uint64) (err error)
 	CommitList(listInfo *TargetCommittedListInfo, numberOfBlocks int, targetName string) (msg string, err error)
 	WritePart(part *Part) (duration time.Duration, startTime time.Time, numOfRetries int, err error)
 	ProcessWrittenPart(result *WorkerResult, listInfo *TargetCommittedListInfo) (requeue bool, err error)
@@ -154,11 +153,6 @@ func ConstructPartsPartition(numberOfPartitions int, size int64, blockSize int64
 	//bsib := uint64(blockSize)
 	numOfBlocks := int((size + blockSize - 1) / blockSize)
 
-	if numOfBlocks > util.MaxBlockCount { // more than 50,000 blocks needed, so can't work
-		var minBlkSize = (size + util.MaxBlockCount - 1) / util.MaxBlockCount
-		log.Fatalf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
-	}
-
 	Partitions := make([]PartsPartition, numberOfPartitions)
 	//the size of the partition needs to be a multiple (blockSize * int) to make sure all but the last part/block
 	//are the same size
@@ -191,11 +185,6 @@ func ConstructPartsQueue(size uint64, blockSize uint64, sourceURI string, target
 	var bsib = blockSize
 	numOfBlocks = int((size + (bsib - 1)) / bsib)
 
-	if numOfBlocks > util.MaxBlockCount { // more than 50,000 blocks needed, so can't work
-		var minBlkSize = (size + util.MaxBlockCount - 1) / util.MaxBlockCount
-		log.Fatalf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
-	}
-
 	parts = make([]Part, numOfBlocks)
 
 	var curFileOffset uint64

pipeline/pipeline.go

@@ -81,7 +81,14 @@ func convertToStorageBlockList(list interface{}, numOfBlocks int) []storage.Bloc
 
 //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) (err error) {
+func (t *AzureBlock) 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
+		var minBlkSize = (source.Size + util.MaxBlockCount - 1) / util.MaxBlockCount
+		return fmt.Errorf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
+	}
+
 	return util.CleanUncommittedBlocks(&t.StorageClient, t.Container, source.TargetAlias)
 }
 

targets/azureblock.go

@@ -28,20 +28,28 @@ func NewAzurePage(accountName string, accountKey string, container string) pipel
 	return &AzurePage{Creds: &creds, Container: container, StorageClient: &client}
 }
 
-//PageSize size of page in Azure Page Blob storage
-const PageSize int64 = 512
+//Page blobs limits and units
+
+//PageSize page size for page blobs
+const PageSize uint64 = 512
+const maxPageSize uint64 = 4 * util.MB
+const maxPageBlobSize uint64 = 8 * util.TB
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
 //initializes the page blob.
-func (t *AzurePage) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error) {
+func (t *AzurePage) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
 	size := int64(source.Size)
 
-	if size%PageSize != 0 {
-		return fmt.Errorf("Invalid size for a page blob. The size of the file %v (%v) is not a multiple of %v", source.SourceName, source.Size, PageSize)
+	if size%int64(PageSize) != 0 {
+		return fmt.Errorf(" invalid size for a page blob. The size of the file %v (%v) is not a multiple of %v ", source.SourceName, source.Size, PageSize)
+	}
+
+	if size > int64(maxPageBlobSize) {
+		return fmt.Errorf(" the file %v is too big (%v). Tha maximum size of a page blob is %v ", source.SourceName, source.Size, maxPageBlobSize)
 	}
 
-	if size > int64(util.TB) {
-		return fmt.Errorf("The file %v is too big (%v). Tha maximum size of a page blob is %v ", source.SourceName, source.Size, util.TB)
+	if blockSize > maxPageSize || blockSize < PageSize {
+		return fmt.Errorf(" invalid block size for page blob: %v. The value must be greater than %v and less than %v", PageSize, maxPageSize)
 	}
 
 	//if the max retries is exceeded, panic will happen, hence no error is returned.

targets/azurepage.go

@@ -32,7 +32,7 @@ func NewMultiFile(overwrite bool, numberOfHandles int) pipeline.TargetPipeline {
 
 //PreProcessSourceInfo implementation of PreProcessSourceInfo from the pipeline.TargetPipeline interface.
 //Passthrough no need to pre-process for a file target.
-func (t *MultiFile) PreProcessSourceInfo(source *pipeline.SourceInfo) (err error) {
+func (t *MultiFile) PreProcessSourceInfo(source *pipeline.SourceInfo, blockSize uint64) (err error) {
 	t.Lock()
 	defer t.Unlock()
 

targets/multifile.go

@@ -323,18 +323,18 @@ func (t *Transfer) StartTransfer(dupeLevel DupeCheckLevel, progressBarDelegate P
 	go t.processAndCommitResults(t.ControlChannels.Results, progressBarDelegate, t.TargetPipeline, &t.SyncWaitGroups.Commits)
 }
 
-//Sequentially calls the PreProcessSourceInfo implementation of the target pipeline for each source in the transfer.
+//Concurrely calls the PreProcessSourceInfo implementation of the target pipeline for each source in the transfer.
 func (t *Transfer) preprocessSources() {
 	sourcesInfo := (*t.SourcePipeline).GetSourcesInfo()
 	var wg sync.WaitGroup
 	wg.Add(len(sourcesInfo))
 	for i := 0; i < len(sourcesInfo); i++ {
-		go func(s *pipeline.SourceInfo) {
+		go func(s *pipeline.SourceInfo, b uint64) {
 			defer wg.Done()
-			if err := (*t.TargetPipeline).PreProcessSourceInfo(s); err != nil {
+			if err := (*t.TargetPipeline).PreProcessSourceInfo(s, b); err != nil {
 				log.Fatal(err)
 			}
-		}(&sourcesInfo[i])
+		}(&sourcesInfo[i], t.blockSize)
 	}
 
 	wg.Wait()

transfer/transfer.go

@@ -206,20 +206,6 @@ func RetriableOperation(operation func(r int) error) (duration time.Duration, st
 	}
 }
 
-///////////////////////////////////////////////////////////////////
-
-//GetNumberOfBlocks calculates the number of blocks from filesize and checks if the number is greater than what's allowed (MaxBlockCount).
-func GetNumberOfBlocks(size uint64, blockSize uint64) int {
-	numOfBlocks := int(size+(blockSize-1)) / int(blockSize)
-
-	if numOfBlocks > MaxBlockCount { // more than 50,000 blocks needed, so can't work
-		var minBlkSize = (size + MaxBlockCount - 1) / MaxBlockCount
-		log.Fatalf("Block size is too small, minimum block size for this file would be %d bytes", minBlkSize)
-	}
-
-	return numOfBlocks
-}
-
 //CreateContainerIfNotExists Creates a new container if doesn't exists. Validates the name of the container.
 func CreateContainerIfNotExists(container string, accountName string, accountKey string) {