Should I be using a Parallel foreach when uploading to azure storage?

Question

I have code that grabs images from my mongodb database and then uploads it to my azure storage as seen below:

public async Task UploadAssetsAsync(Func<GridFSFileInfo, string> prefixSelector, List<GridFSFileInfo> files, Func<GridFSFileInfo, Task<Stream>> streamOpener, Func<string, Task> progressAction)
        {
            if (flyersContainerClient == null)
                throw new Exception("Container client not initialized. Please initialize before doing blob operations.");
            var q = new Queue<Task<Response<BlobContentInfo>>>();
            progressAction?.Invoke($"{files.Count}");
            foreach (var f in files)
            {
                var pathPrefix = prefixSelector(f);
                var blobClient = flyersContainerClient.GetBlobClient($"{pathPrefix}/{f.Filename.Replace("_copy", "")}");
                IDictionary<string, string> metadata = new Dictionary<string, string>();
                var blobhttpheader = new BlobHttpHeaders();
                if (f.Filename.EndsWith("svg"))
                {

                    blobhttpheader.ContentType = "image/svg+xml";
                }
                var stream = await streamOpener(f);
                if (pathPrefix == "thumbnails")
                {
                    var format = ImageFormat.Jpeg;
                    Bitmap cropped = null;
                    using (Image image = Image.FromStream(stream))
                    {
                        format = image.RawFormat;
                        Rectangle rect = new Rectangle(0, 0, image.Width, (image.Width * 3) / 4);
                        cropped = new Bitmap(image.Width, (image.Width * 3) / 4);
                        using (Graphics g = Graphics.FromImage(cropped))
                        {
                            g.DrawImage(image, new Rectangle(0, 0, cropped.Width, cropped.Height), rect, GraphicsUnit.Pixel);
                        }
                    }
                    stream.Dispose();                    
                    stream = new MemoryStream();                                     
                    cropped.Save(stream, format);
                    stream.Position = 0;
                }
                //await blobClient.SetMetadataAsync(metadata);
                q.Enqueue(blobClient.UploadAsync(stream, new BlobUploadOptions { HttpHeaders = blobhttpheader, TransferOptions = new Azure.Storage.StorageTransferOptions { MaximumConcurrency = 8, InitialTransferSize = 50 * 1024 * 1024 } }));
                //await blobClient.SetHttpHeadersAsync(blobHttpHeader);
            }
            await Task.WhenAll(q);
        }        

You can see what it is doing when the foreach starts with my list of files. I am also using an asyncronous task at the end that waits for all of my q variables to finish as seen in the WhenAll at the bottom. Would it benefit my code to use a Parallel.Foreach for uploading my files or is there a faster way to achieve what I am doing?

Thanks for the help!

Answer 1

Parallel.ForEach is meant for data parallelism - processing a lot of in-memory data by using all cores at 100%. That's not what's going on here.

The most appropriate classes would be the TPL Dataflow's TransformBlock and ActionBlock. Each DataFlow block processes messages in its input queue using one or more worker tasks and forwards them to the next block in the pipeline.

This also allows using separate methods to crawl files, process images and upload them. The pipeline code could be something like this:

record ResizedImage(FileInfo File,MemoryStream Stream);

...
var options=new ExecutionDataflowBlockOptions {
    MaxDegree 
var crawler=new TransformManyBlock<string,FileInfo>(CrawlFolder);
var resizer=new TransformBlock<FileInfo,ResizedImage>(ResizeImage,
    new ExecutionDataflowBlockOptions {
        MaxDegreeOfParallelism=2,
        BoundedCapacity=4,
    });
var uploader=new TransformBlock<ResizedImage,Response<BlobContentInfo>>(UploadImage,
    new ExecutionDataflowBlockOptions {
        MaxDegreeOfParallelism=2,
        BoundedCapacity=4
    });
...

var linkOptions=new DataflowLinkOptions { PropagateCompletion = true};
crawler.LinkTo(resizer,linkOptions);
resizer.LinkTo(uploader,linkOptions);
...

Each method only needs to accept an input message and return the output. Buffering and concurrency are handled by the Dataflow blocks. In this case 1 task is used to crawl, 2 tasks are used to resize images and 2 to upload. The DOP can be adjusted to achieve optimal performance.

The BoundedCapacity limits the input buffer of each block. This prevents a slow block from flooding RAM with waiting messages. When a block's input buffer is full, upstream blocks have to wait before they can process and send new messages.

Once the pipeline is constructed, the application can start posting data to the head block, in this case the crawler :

while (var folder in folders)
{
    crawler.Post(folder);
}

Messages will flow from one block to the next. When the head block completes it will also signal downstream blocks to complete. The application can wait for the entire pipeline to finish by awaiting the final block's Completion property

crawler.Complete();
await tail.Completion;

Now that the worker methods don't have to deal with concurrency, they become a lot simpler :

IEnumerable<FileInfo> CrawlFolder(string folder)
{
    var di=new DirectoryInfo(folder);
    return di.EnumerateFiles("*.*");
}

ResizedImage ResizeImage(FileInfo file)
{
    var format = ImageFormat.Jpeg;
    using var stream=file.Open();
    using (Image image = Image.FromStream(stream))
    {
        var format = image.RawFormat;
        Rectangle rect = new Rectangle(0, 0, image.Width, (image.Width * 3) / 4);
        using var cropped = new Bitmap(image.Width, (image.Width * 3) / 4);
        using (Graphics g = Graphics.FromImage(cropped))
        {
            g.DrawImage(image, new Rectangle(0, 0, cropped.Width, cropped.Height), rect, GraphicsUnit.Pixel);
        }

        var newStream = new MemoryStream();                                     
        cropped.Save(newStream, format);
        newStream.Position = 0;
        return new ResizedImage(file,newStream);
    }
}


async Task<Response<BlobContentInfo>> UploadImage(ResizedImage rsImage)
{
...
}