TPL Thread Leak and Memory Leak

Question

I'm trying to track down what I suppose to be memory/thread leak in one of my programs.

My program uses a function to upload a file to a Windows Azure Storage Blob. In order to make this function resilient to various transient error conditions (such as intermittent network failures, etc.), I want to make use of the Transient Fault Handling Application Block for Windows Azure "topaz", part of Enterprise Library, which includes a configurable retry mechanism.

Also, in order to circumvent timeout conditions when dealing with large file and improve and reduce the scope for portions of the upload that should be repeated in case of failure, I try to upload the specified file stream in chunks, that each get uploaded independently.

Finally, in order to adhere to best practices using Windows Azure Storage, I make use of asynchronous operations as much as possible when dealing with Windows Azure, in an attempt to improve the scalability of my apps.

Here is the resulting snippet function :

    public static Task ChunkedUploadStreamAsync(CloudBlockBlob blob, Stream source, BlobRequestOptions options, int chunkSize, RetryPolicy policy)
    {
        var blockids = new List<string>();
        var blockid = 0;

        var count = 0;
        var bytes = new byte[chunkSize];

        // first create a list of TPL Tasks for uploading blocks asynchronously

        var tasks = new List<Task>();

        while ((count = source.Read(bytes, 0, bytes.Length)) != 0)
        {
            var id = Convert.ToBase64String(BitConverter.GetBytes(++blockid));

            Func<Task> uploadTaskFunc = () => new TaskFactory()
                .FromAsync(
                    (asyncCallback, state) => blob.BeginPutBlock(id, new MemoryStream(bytes, 0, count), null, null, null, null, asyncCallback, state)
                    , blob.EndPutBlock
                    , null
                )

                .ContinueWith(antecedent => blockids.Add(id), TaskContinuationOptions.NotOnFaulted);

            tasks.Add(policy.ExecuteAsync(uploadTaskFunc));
        }

        return new TaskFactory().ContinueWhenAll(
            tasks.ToArray(),
            array =>
                {
                    // propagate exceptions and make all faulted Tasks as observed
                    Task.WaitAll(array);

                    // create continuation task for committing uploaded blocks
                    Func<Task> commitTaskFunc = () => new TaskFactory()
                        .FromAsync(
                            (asyncCallback, state) => blob.BeginPutBlockList(blockids, asyncCallback, state)
                            , blob.EndPutBlockList
                            , null);

                    policy
                        .ExecuteAsync(commitTaskFunc)
                        .Wait();
                });
    }

With Performance Monitor, I can observe that after calling this function, the number of threads and amount of memory used by my program is increasing significantly. For instance, here is a snapshot of just one call of this function :

Please, can someone advise as to where I'm doing something wrong. Please, suggest a better design if necessary.

Answer 1

There are several issues in your code that make it error-prone:

• you are using a single byte array to feed all memory streams, thus the same data may be sent in different chunks. Same stands for count.
• the order of chunks in blockids can be different to the one you've sent so resulting BLOB will have incorrect order of blocks.
• there is a potential thread contention on blockids as different threads may add a new element at the same time causing all sorts of problems like missing blocks in resulting list, incorrect block IDs etc.
• use Task.Factory instead of new TaskFactory()
• if you create a task and immediately wait for it then you don't need that task, so you can replace the last BeginPutBlockList with synchronous version

Other than that I don't see memory leak from the graphs you provided. Threads may still be alive in a thread pool, and memory may not be collected by GC yet.

Here is what I've got as a result (note that it may consume more memory as you need to store all the chunks that are currently being sent in memory):

    private static Task PutBlockAsync(CloudBlockBlob blob, string id, Stream stream, RetryPolicy policy)
    {
        Func<Task> uploadTaskFunc = () => Task.Factory
            .FromAsync(
                (asyncCallback, state) => blob.BeginPutBlock(id, stream, null, null, null, null, asyncCallback, state)
                , blob.EndPutBlock
                , null
            );
        return policy.ExecuteAsync(uploadTaskFunc);
    }

    public static Task ChunkedUploadStreamAsync(CloudBlockBlob blob, Stream source, BlobRequestOptions options, int chunkSize, RetryPolicy policy)
    {
        var blockids = new List<string>();
        var blockid = 0;

        int count;

        // first create a list of TPL Tasks for uploading blocks asynchronously
        var tasks = new List<Task>();

        var bytes = new byte[chunkSize];
        while ((count = source.Read(bytes, 0, bytes.Length)) != 0)
        {
            var id = Convert.ToBase64String(BitConverter.GetBytes(++blockid));
            blockids.Add(id);
            tasks.Add(PutBlockAsync(blob, id, new MemoryStream(bytes, 0, count), policy));
            bytes = new byte[chunkSize]; //need a new buffer to avoid overriding previous one
        }

        return Task.Factory.ContinueWhenAll(
            tasks.ToArray(),
            array =>
            {
                // propagate exceptions and make all faulted Tasks as observed
                Task.WaitAll(array);
                policy.ExecuteAction(() => blob.PutBlockList(blockids));
            });
    }