PageBlob.HighConcurrency.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.WindowsAzure.Storage.Blob;

namespace ConsoleApp
{
    public class Program
    {
        private const string ContainerSas = "";

        private const int PageSizeBytes = 512;
        private const int PageCount = 1000;

        public static void Main(string[] args)
        {
            CloudBlobContainer container = new CloudBlobContainer(new Uri(ContainerSas));
            CloudPageBlob blob = container.GetPageBlobReference("page_text");
            blob.Create(PageCount * PageSizeBytes);


            IEnumerable<byte[]> pages = Enumerable.Range(0, PageCount)
                .Select(i => GenerateBytes(PageSizeBytes));

            long offset = 0;

            pages.ForEachAsync(async buffer =>
            {
                using (var ms = new MemoryStream(buffer))
                {
                    long nextOffset = Interlocked.Add(ref offset, PageSizeBytes);
                    long currentOffset = nextOffset - PageSizeBytes;
                    await blob.WritePagesAsync(ms, currentOffset, null);
                }
            }).Wait();
        }

        private static byte[] GenerateBytes(int length)
        {
            return Enumerable.Range(0, length)
                .Select(i => (byte) i)
                .ToArray();
        }
    }

    public static class Extensions
    {
        public static Task ForEachAsync<T>(this IEnumerable<T> enumerable, Func<T, Task> func,
            int concurrencyLevel = 20)
        {
            concurrencyLevel = concurrencyLevel > 1 ? concurrencyLevel : 1;
            ConcurrentQueue<T> queue = new ConcurrentQueue<T>(enumerable);

            IEnumerable<Task> tasks = Enumerable
                .Range(0, concurrencyLevel)
                .Select(async i =>
                {
                    while (queue.TryDequeue(out T obj))
                    {
                        await func(obj);
                    }
                });

            return Task.WhenAll(tasks);
        }
    }
}