Good approach for running Tasks synchronously with FIFO?

Question

Currently, I do my first steps with async/await and tasks in .NET and I am very excited about how easy it is to run things asynchronously! However, at the moment I have to communicate with devices through a SerialPort. Since only one connection is possible at the same time, I just wrote a few extension methods to run all those methods, coming from different tasks/threads, synchronously and in a first in first out order:

public static class Extensions
{
    private readonly static object LockObject = new object();

    public static Task<TResult> RunAfter<TResult>(this Task<TResult> task, ConcurrentQueue<Task> others)
        => (Task<TResult>)task.RunAllSynchronously(others);

    public static Task RunAfter(this Task task, ConcurrentQueue<Task> others)
        => task.RunAllSynchronously(others);

    private static Task RunAllSynchronously(this Task task, ConcurrentQueue<Task> others)
    {
        if (others == null) throw new ArgumentNullException("The value of " + nameof(others) + " is null!");
        lock (LockObject)
        {
            others.Enqueue(task);
            Task currentTask;
            while (others.TryDequeue(out currentTask))
            {
                currentTask.RunSynchronously();
                if (currentTask == task) break;
            }
        }
        return task;
    }
}

Does this approach seems to be a good way or should such a case be treated differently?

Answer 1

Why you run them synchronously?

You should run tasks asynchronously and use async and await to execute them one by one:

 Task currentTask;
 while (others.TryDequeue(out currentTask))
 {
      await currentTask;
      if (currentTask == task) break;
 }

In the other hand, looking at your code, I can't find a reason to use lock (thread synchronization) at all. You synchronize threads against some shared resource (i.e. some object that may or may not be expected to be read/modified by more than a thread). You method could be reworked to:

private static async Task RunAllAsync(this Task task, ConcurrentQueue<Task> others)
{
    // Design by contract rocks ;)
    // See:  https://msdn.microsoft.com/en-us/library/dd264808(v=vs.110).aspx
    Contracts.Requires(task != null);
    Contracts.Requires(others != null);

    others.Enqueue(task);

    // See how I've improved your loop. Since ConcurrentQueue.TryDequeue
    // will return false if other thread has called it already, your loop
    // should try to dequeue again until it returns true, and it should
    // break if dequeued task is the task against which the extension method
    // was called or the concurrent queue has no more items, to prevent a 
    // possible infinite loop
    do
    { 
       Task currentTask;
       if(others.TryDequeue(out currentTask))
          await currentTask;

    }
    while (currentTask == task || others.Count > 0);

    return task;
}

Update

OP said:

I have possibly forgotten to say, that the ConcurrentQueue is the resource that should be shared among the threads. I.e. Task.RunAllSynchronously() is called on every new Task (access to SerialPort) and this call could be come from a different thread. Also, I cannot ensure that RunAllSynchronously() is just called, when all currently running (or queued) tasks are finished (I could, but therefore i had to use something like lock outside the extension method, which is not really that nice of having an extension method.

This is why you're using ConcurrentQueue<T>. Thread safety is managed internally. If you call ConcurrentQueue<T>.TryDequeue and more than one thread calls it at once, only one will win and others will receive false as return value and out parameter won't be assigned. See what MSDN says for this:

ConcurrentQueue handles all synchronization internally. If two threads call TryDequeue at precisely the same moment, neither operation is blocked. When a conflict is detected between two threads, one thread has to try again to retrieve the next element, and the synchronization is handled internally.

TryDequeue tries to remove an element from the queue. If the method is successful, the item is removed and the method returns true; otherwise, it returns false. That happens atomically with respect to other operations on the queue. If the queue was populated with code such as q.Enqueue("a"); q.Enqueue("b"); q.Enqueue("c"); and two threads concurrently try to dequeue an element, one thread will dequeue a and the other thread will dequeue b. Both calls to TryDequeue will return true, because they were both able to dequeue an element. If each thread goes back to dequeue an additional element, one of the threads will dequeue c and return true, whereas the other thread will find the queue empty and will return false.

Answer 2

First of all:

You only benefit from async-await if your program has something else to do while your tasks are running.

If your main thread would start a task, and do nothing but wait for this task to finish, your main thread could do the work himself. That would even be faster.

In your example, I can imagine that sending over the serial line is significantly slower than your processing. So I can imagine that while one thread is busy sending data over the serial line, your thread can be busy creating the next data that is to be sent. Or maybe 10 threads are creating data that is to be sent one after another. Of course in the latter case it is not guaranteed in which order the data will be sent.

Buf let's see it simpler: one thread is creating data in its own speed, while another thread is sending data independently over the serial line.

This screams for a producer - consumer pattern: one thread is the producer, it produces items that the consumer reads and processes. After a while the producer tells the consumer that no data is to be expected anymore.

The key object in this is System.Threading.Tasks.Dataflow.BufferBlock. See MSDN. The remarks section says that it is distributed via nuget.

The bufferBlock implements two interfaces:

• ITargetBlock<T> for the producer to send its output to
• ISourceBlock<T> for the consumer to read the input from.

Let's assume you use System.IO.Ports.SerialPort to send your data. Alas this class has no async support, so we have to create it ourselves. Let's assume you want to convert objects of type T into a format that can be sent over the serial line. Code would look like follows:

private void Write(T t)
{
    var dataToSend = ConvertToData(t);
    serialPort.Write(dataToSend);
}

Not very async is it. So let's make an async function ofit:

private async Task WriteAsync(T t)
{
    return await Task.Run ( () =>
    {
        var dataToSend = ConvertToData(t);
        serialPort.Write(dataToSend);
    }
}

Or you could just call the other write function:

return await Task.Run ( () => Write(t));

Note: if you make sure there is only one thread that will use this function, you don't have to lock it.

Now that we do have an async function to send objects of type T over the serial line, let's create a producer that will create objects of type T and send them to the bufferblock.

I'll make it async, so the calling thread can do other things while data is being produced:

private BufferBlock<T> bufferBlock = new BufferBlock<T>();

private async Task ProduceAsync()
{
    while (objectsToProcessAvailable())
    {
        T nextObject = GetNextObjectToProcess()
        await bufferBlock.SendAsync(nextObject);
    }
    // nothing to process anymore: mark complete:
    bufferBlock.Complete();
}

The receiving side will be done by a different thread:

private Task ConsumeAsync()
{
    // as long as there is something to process: fetch it and process it
    while (await bufferBlock.OutputAvailableAsync())
    {
        T nextToProcess = await bufferBlock.ReceiveAsync();
        // use WriteAsync to send to the serial port:
        await WriteAsync(nextToProcess);
    }
    // if here: no more data to process. Return
}

Now all we need is one procedure that creates the two threads and waits until both tasks are finished:

private async Task ProduceConsumeAsync()
{
    var taskProducer = ProduceAsync();
    // while the producer is busy producing, you can start the consumer:
    var taskConsumer = ConsumeAsync();
    // while both tasks are busy, you can do other things,
    // like keep the UI responsive
    // after a while you need to be sure the tasks are finished:
    await Task.WhenAll(new Task[] {taskProducer, taskConsumer});
}

Note: because of the bufferBlock it is no problem that the producer is already producing while the consumer is not started yet.

All we need is a function that starts the async, if you have an event handler just declare it async:

private async void OnButton1_clicked(object sender, ...)
{
    await ProduceConsumeAsync()
}

If you have no async function, you have to create a task yourself:

private void MyFunction()
{
    // start produce consume:
    var myTask = Task.Run( () => ProduceConsumeAsync());
    // while the task is running, do other things.
    // when you need the task to finish:
    await myTask;
 }

More information about the consumer - producer pattern. See MSDN

How to: Implement a Producer-Consumer Dataflow Pattern

Answer 3

After playing with various things I just found a simple solution, which should be sufficient for me and is somewhat similar to the solution of Matías Fidemraizer:

private static ConcurrentQueue<Task> Tasks { get; } = new ConcurrentQueue<Task>();

public async static Task RunAlone(this Task task)
{
    Tasks.Enqueue(task);

    do
    {
        var nextTask = Tasks.First();

        if (nextTask == task)
        {
            nextTask.Start();
            await nextTask;
            Task deletingTask;
            Tasks.TryDequeue(out deletingTask);
            break;
        }
        else
        {
            nextTask.Wait();
        }
    } while (Tasks.Any());
}

public async static Task<TResult> RunAlone<TResult>(this Task<TResult> task)
{
    TResult result = default(TResult);
    Tasks.Enqueue(task);

    do
    {
        var nextTask = Tasks.First();

        if (nextTask == task)
        {
            nextTask.Start();
            result = await (Task<TResult>)nextTask;
            Task deletingTask;
            Tasks.TryDequeue(out deletingTask);
            break;
        }
        else
        {
            nextTask.Wait();
        }
    } while (Tasks.Any());

    return result;
}