A method that calls multiple async tasks with error handling, done the right way?

Question

I've got a method, CheckForValue(), that uses named pipes to send a message to another process running locally, and then receive a message from that process, returning a bool indicating if it is equivalent with some value.

I'm concerned that I am not handling CancellationTokens and errors correctly. If there any errors (or timeouts), I just want this method to return false.

Again, I am not concerned with different methods of IPC, or using a single duplex named pipe, etc. My concern is the error handling and the usage of CancellationTokens here.

CheckForValue

public async Task<bool> CheckForValue()
{
    int timeout = 300; //300ms should be plenty of time
    try
    {
        using (var getValueCancellationTokenSource = new CancellationTokenSource())
        {
            var valueTask = GetValueUsingNamedPipe(getValueCancellationTokenSource);
            using (var timeoutCancellationTokenSource = new CancellationTokenSource())
            {
                var completedTask = await Task.WhenAny(valueTask, Task.Delay(timeout, timeoutCancellationTokenSource.Token));
                if (completedTask == valueTask)
                {
                    if (timeoutCancellationTokenSource.Token.CanBeCanceled)
                    {
                        timeoutCancellationTokenSource.Cancel();
                    }

                    var result = valueTask.Result;
                    return (result == "WhatIWant");
                }

                if (getValueCancellationTokenSource.Token.CanBeCanceled)
                {
                    getValueCancellationTokenSource.Cancel ();
                }

                return false;
            }
        }

    } 
    catch (Exception)
    {
        return false;
    }
}

GetValueUsingNamedPipe

public async Task<string> GetValueUsingNamedPipe(CancellationTokenSource ct)
{
    var response = "";
    try
    {
        Task sendMsg = SendMessage ("MyMessage", ct);
        await sendMsg;
        response = await Listen(ct);
    } 
    catch (Exception)
    {
        return "";
    }
    return response;
}

SendMessage

public async Task SendMessage(string message, CancellationTokenSource ct)
{
    try
    {
        using (var _pipeClientStream = new NamedPipeClientStream(".", "MyPipe", PipeDirection.Out, PipeOptions.Asynchronous))
        {
            await _pipeClientStream.ConnectAsync (1000, ct.Token);

            var writer = new StreamWriter (_pipeClientStream) { AutoFlush = true };
            await writer.WriteLineAsync (message);
            await writer.WriteLineAsync (MessageFooter);
        }
    } 
    catch (Exception)
    {
        await Task.FromCanceled(ct.Token);
    }
}

Listen

public async Task<string> Listen(CancellationTokenSource ct)
{
    try
    {
        if (ct.Token.IsCancellationRequested)
        {
            ct.Token.ThrowIfCancellationRequested ();
        }
        using (var _pipeClientStream = new NamedPipeClientStream(".", "MyListenPipe", PipeDirection.In, PipeOptions.Asynchronous, TokenImpersonationLevel.Impersonation))
        {
            await _pipeClientStream.ConnectAsync (ct.Token);
            if (!ct.IsCancellationRequested)
            {
                var sb = new StringBuilder ();
                var reader = new StreamReader (_pipeClientStream);
                do
                {
                    string line = await reader.ReadLineAsync ();

                    if (line == MessageFooter || line == null)
                    {
                        break;
                    }
                    sb.AppendLine (line);

                } while (true);

                return sb.ToString ();
            }
            return "";
        }
    } 
    catch (Exception e)
    {
        return "";
    }
}

Answer 1

According to my understanding your piece of software can finish in one of the following states:

• Succeeded
• Failed
• Timed out

You are not exposing the ability to cancel it on the top level. So, it is not cancelable.

Your explicit cancellation mechanism is not needed because:

1. If timeout occurs at the top level then this fact will be available for all async methods, which received the CancellationToken. The built-in BCL functions will check the token's validity so you don't have to worry about it.
2. SendMessage and Listen are called sequentially. So if former fails then latter won't be called.
3. After connectAsync your IsCancellationRequested is pointless. If connectAsync succeeded then this property won't be true. If it was not succeeded then you would not reach this line because your Listen function will be aborted.

On the other hand there is one place where it could make sense to check the cancellationToken. That's inside the do-while loop because the ReadLineAsync is not cancellable.

So let's put all this together with these in mind. I've converted your code into C# 8 if you don't mind.

Listen

public async Task<string> Listen(CancellationToken ct)
{
    try
    {
        await using var _pipeClientStream = new NamedPipeClientStream(".", "MyListenPipe", PipeDirection.In, PipeOptions.Asynchronous, TokenImpersonationLevel.Impersonation); 
        await _pipeClientStream.ConnectAsync(ct);

        var sb = new StringBuilder();
        using var reader = new StreamReader(_pipeClientStream);
        do
        {
            ct.ThrowIfCancellationRequested();
            string line = await reader.ReadLineAsync();

            if (line == MessageFooter || line == null)
            {
                break;
            }
            sb.AppendLine(line);

        } while (true);

        return sb.ToString();
    }
    catch
    {
        return "";
    }
}

• The NamedPipeClientStream is IAsyncDisposable that's why I used await using.
• The StreamReader is IDisposable that's why I used using.
• I've checked the CancellationToken before each ReadLineAsync because it is not cancellable.

SendMessage

public async Task SendMessage(string message, CancellationToken ct)
{
    await using var _pipeClientStream = new NamedPipeClientStream(".", "MyPipe", PipeDirection.Out, PipeOptions.Asynchronous);
    await _pipeClientStream.ConnectAsync(1000, ct);

    await using var writer = new StreamWriter(_pipeClientStream) { AutoFlush = true };
    await writer.WriteLineAsync(message.AsMemory(), ct);
    await writer.WriteLineAsync(MessageFooter.AsMemory(), ct);
}

• WriteLineAsync can accept CancellationToken only if it is called with a ReadOnlyMemory or with a StringBuilder. That's why I used AsMemory.
• StreamWriter is IAsyncDisposable that's why I used await using.

GetValueUsingNamedPipe

public async Task<string> GetValueUsingNamedPipe(CancellationToken ct)
{
    await SendMessage("MyMessage", ct);
    return await Listen(ct);
}

• The sendMsg variable is pointless because right after the assignment it is awaited then never used.
• I've removed the try-catch block because in that way the TaskCancelledException would be caught here.
• So basically if you wish you can inline this method into the CheckForValue

CheckForValue

public async Task<bool> CheckForValue()
{
    const int timeout = 300;
    try
    {
        using var timeoutTokenSource = new CancellationTokenSource(timeout);
        var result = await GetValueUsingNamedPipe(timeoutTokenSource.Token);
        return (result == "WhatIWant");
    }
    catch (TaskCanceledException) //The actual type will be TimedOutTaskCanceledException
    {
        return false;
    }
    catch (Exception) // GetValueUsingNamedPipe failed for some reason
    {
        return false;
    }
}

• All the explicit cancellation calls are gone, because they are not needed. The cancellation fact will be propagated all the related parties and it will manifest in a TaskCancelledException.
• So it is much cleaner now. :D

I hope it helped you a bit.