TPL FromAsync Task keeps getting cancelled

Question

I'm sending 10 messages from a client to a server in a loop, and then responses back from the server to the client using TPL's FromAsync (targeting .NET 4.0)

After the last message is sent from the client, I'm trying to measure the total time taken until the final response is received. To do this, I call Task.WaitAll but I get an AggregateException who's inner exception says "A task was cancelled" for all the tasks. The server is receiving the requests.

Code:

private static double MeasureAsyncTPL(List<string> stringList)
    {
        var stopwatch = new Stopwatch();
        stopwatch.Start();

        var tasks = new Task[stringList.Count];
        for (int i = 0; i < stringList.Count; i++)
        {
            string source = stringList[i];
            tasks[i] = SendMessageToServerAsync(source);
            //SendMessageToServerAsync(source);
        }
        Task.WaitAll(tasks);


        stopwatch.Stop();            
        return stopwatch.Elapsed.TotalMilliseconds; 
    }

        static Task SendMessageToServerAsync(string message)
    {
        Task task;
        var byteArray = MessageToByteArray(message, Encoding.UTF8);
        using (var tcpClient = new TcpClient())
        {
            tcpClient.Connect("127.0.0.1", 5000);
            using (var networkStream = tcpClient.GetStream())
            {
                task = Write(networkStream,byteArray,0);
                Task continuation = task.ContinueWith(ant => Read(ant, networkStream), TaskContinuationOptions.NotOnFaulted | TaskContinuationOptions.AttachedToParent);

                return continuation;
            }
        }
    }

        private static Task Write(NetworkStream stream, byte[] buffer, int offset)
    {
        return Task.Factory.FromAsync(stream.BeginWrite, stream.EndWrite, buffer, offset, buffer.Length, null);
    }

    private static Task Read(Task write, NetworkStream stream)
    {

        byte[] data = new byte[50];
        return Task<int>.Factory.FromAsync(stream.BeginRead, stream.EndRead, data, 0, data.Length, null);

    }

The exception is being picked up inside MeasureAsyncTPL on the line Task.WaitAll(tasks). How can I fix this exception?

Second question - is FromAsync inside my Read method guaranteed to return the entire message from the server? Or do I need to call it multiple times while keeping track of a the size of the byte array that the server sent?

You are disposing of the networkStream before writing or reading. — Yacoub Massad
The cancellation is automatic for a continuation when the antecedent task is faulted and you specify the NotOnFaulted option, pointing to an exception in your Write task (as others have pointed, likely due to premature stream disposal). Fix that and you'll fix the TaskCanceledException. — Kirill Shlenskiy
Also, your TaskContinuationOptions.AttachedToParent does absolutely nothing as your task is not scheduled inside another task - it's just a continuation and has no parent to attach to. — Kirill Shlenskiy
If you use a continuation style task, you cant run it inside a using block without synchronously blocking, otherwise the element will get disposed and you'll end up with a race condition, like you're seeing now. — Yuval Itzchakov
@YuvalItzchakov, solid advice, although I would extend it to any task which relies on the IDisposable being alive, not just continuations (i.e. you shouldn't return a task asynchronously reading from a Stream inside the using block regardless of whether it has a continuation or not). — Kirill Shlenskiy

Yacoub Massad Yacoub Massad · Accepted Answer · 2015-12-16T13:04:22

You are disposing of the networkStream and tcpClient before the write and read operations start or have time to work.

With .NET 4.0, you need to do something as complex as this (this is something to start with, it needs more work):

var byteArray = MessageToByteArray(message, Encoding.UTF8);

var tcpClient = new TcpClient();

NetworkStream networkStream = null;

try
{
    tcpClient.Connect("127.0.0.1", 5000);

    networkStream = tcpClient.GetStream();
}
catch
{
    ((IDisposable)tcpClient).Dispose();
    throw;
}

byte[] read_buffer = new byte[50];

Action dispose_action = () =>
{
    ((IDisposable)tcpClient).Dispose();
    networkStream.Dispose();
};

var write_task = networkStream.WriteAsync(byteArray, 0, byteArray.Length);

write_task
    .ContinueWith(wt =>
    {
        var read_task = networkStream.ReadAsync(read_buffer, 0, read_buffer.Length);


        read_task.ContinueWith(rt =>
        {
            //See if rt task is successful and if so, process read data here

            dispose_action();
        });

    },
    TaskContinuationOptions.OnlyOnRanToCompletion);

write_task.ContinueWith(wt => dispose_action, TaskContinuationOptions.NotOnRanToCompletion);

The main idea of the code above is that you don't dispose of the TCP client and the network stream object until the read/write operations are done.

Imagine how complex this code would become if you want to read multiple times in some kind of loop.

With .NET 4.5, you can use async and await to do something (simpler) like this:

static async Task SendMessageToServerAsync(string message)
{
    var byteArray = MessageToByteArray(message, Encoding.UTF8);
    using (var tcpClient = new TcpClient())
    {
        tcpClient.Connect("127.0.0.1", 5000);
        using (var networkStream = tcpClient.GetStream())
        {
            await networkStream.WriteAsync(byteArray, 0, byteArray.Length);

            byte[] read_buffer = new byte[50];

            int read = await networkStream.ReadAsync(read_buffer, 0, read_buffer.Length);

            // do something with the read_buffer
        }
    }
}

TPL FromAsync Task keeps getting cancelled

1 Answers