add parallelism to azure downloads

Signed-off-by: Avi Deitcher <avi@deitcher.net>
(cherry picked from commit 3ea9020)

libs/zedUpload/azureutil/azure.go

@@ -23,6 +23,7 @@ const (
 	SingleMB       int64 = 1024 * 1024
 	blobURLPattern       = "https://%s.blob.core.windows.net/%s"
 	maxRetries           = 20
+	parallelism          = 128
 )
 
 // UpdateStats contains the information for the progress of an update
@@ -133,12 +134,6 @@ func DownloadAzureBlob(accountName, accountKey, containerName, remoteFile, local
 	containerURL := azblob.NewContainerURL(*URL, p)
 	blobURL := containerURL.NewBlockBlobURL(remoteFile)
 	ctx := context.Background()
-	downloadResponse, err := blobURL.Download(ctx, 0, azblob.CountToEnd, azblob.BlobAccessConditions{}, false, azblob.ClientProvidedKeyOptions{})
-	if err != nil {
-		return fmt.Errorf("could not start download: %v", err)
-	}
-
-	readCloser := downloadResponse.Body(azblob.RetryReaderOptions{MaxRetryRequests: maxRetries})
 
 	tempLocalFile := localFile
 	index := strings.LastIndex(tempLocalFile, "/")
@@ -152,27 +147,33 @@ func DownloadAzureBlob(accountName, accountKey, containerName, remoteFile, local
 		return err
 	}
 	defer file.Close()
-	defer readCloser.Close()
-	chunkSize := SingleMB
-	var written, copiedSize int64
-	var copyErr error
+
 	stats.Size = objSize
-	for {
-		if written, copyErr = io.CopyN(file, readCloser, chunkSize); copyErr != nil && copyErr != io.EOF {
-			return copyErr
-		}
-		copiedSize += written
-		if written != chunkSize {
-			break
-		}
-		stats.Asize = copiedSize
-		if prgNotify != nil {
+	var progressReceiver pipeline.ProgressReceiver
+	if prgNotify != nil {
+		progressReceiver = func(bytesTransferred int64) {
+			stats.Asize = bytesTransferred
 			select {
 			case prgNotify <- stats:
 			default: //ignore we cannot write
 			}
 		}
 	}
+	// we could just return the error that comes from this function, but in the case of
+	// a nil error, we want to be sure we sent the total
+	if err := azblob.DownloadBlobToFile(ctx, blobURL.BlobURL, 0, 0, file, azblob.DownloadFromBlobOptions{
+		BlockSize:                  SingleMB,
+		Parallelism:                uint16(parallelism),
+		RetryReaderOptionsPerBlock: azblob.RetryReaderOptions{MaxRetryRequests: maxRetries},
+		Progress:                   progressReceiver,
+	}); err != nil {
+		return err
+	}
+	// ensure we send the total; it is theoretically possible that it downloaded without error
+	// but the progress receiver did not get invoked at the end
+	if stats.Asize < stats.Size {
+		progressReceiver(stats.Size)
+	}
 	return nil
 }
 

pkg/pillar/go.mod

@@ -59,7 +59,7 @@ require (
 	github.com/yvasiyarov/newrelic_platform_go v0.0.0-20160601141957-9c099fbc30e9 // indirect
 	go.opencensus.io v0.23.0 // indirect
 	golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a
-	golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d // indirect
+	golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d
 	golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e
 	google.golang.org/genproto v0.0.0-20210224155714-063164c882e6 // indirect
 	google.golang.org/grpc v1.36.0