Is it needed to make fields thread-safe when using async/await?

Question

Sometimes I encounter async/await code that accesses fields of an object. For example this snippet of code from the Stateless project:

private readonly Queue<QueuedTrigger> _eventQueue = new Queue<QueuedTrigger>();
private bool _firing;

async Task InternalFireQueuedAsync(TTrigger trigger, params object[] args)
{
    if (_firing)
    {
        _eventQueue.Enqueue(new QueuedTrigger { Trigger = trigger, Args = args });
        return;
    }

    try
    {
        _firing = true;

        await InternalFireOneAsync(trigger, args).ConfigureAwait(false);

        while (_eventQueue.Count != 0)
        {
            var queuedEvent = _eventQueue.Dequeue();
            await InternalFireOneAsync(queuedEvent.Trigger, queuedEvent.Args).ConfigureAwait(false);
        }
    }
    finally
    {
        _firing = false;
    }
}

If I understand correctly the await **.ConfigureAwait(false) indicates that the code that is executed after this await does not necessarily has to be executed on the same context. So the while loop here could be executed on a ThreadPool thread. I don't see what is making sure that the _firing and _eventQueue fields are synchronized, for example what is creating the a lock/memory-fence/barrier here? So my question is; do I need to make the fields thread-safe, or is something in the async/await structure taking care of this?

Edit: to clarify my question; in this case InternalFireQueuedAsync should always be called on the same thread. In that case only the continuation could run on a different thread, which makes me wonder, do I need synchronization-mechanisms(like an explicit barrier) to make sure the values are synchronized to avoid the issue described here: http://www.albahari.com/threading/part4.aspx

Edit 2: there is also a small discussion at stateless: https://github.com/dotnet-state-machine/stateless/issues/294

Answer 1

I don't see what is making sure that the _firing and _eventQueue fields are synchronized, for example what is creating the a lock/memory-fence/barrier here? So my question is; do I need to make the fields thread-safe, or is something in the async/await structure taking care of this?

await will ensure all necessary memory barriers are in place. However, that doesn't make them "thread-safe".

in this case InternalFireQueuedAsync should always be called on the same thread.

Then _firing is fine, and doesn't need volatile or anything like that.

However, the usage of _eventQueue is incorrect. Consider what happens when a thread pool thread has resumed the code after the await: it is entirely possible that Queue<T>.Count or Queue<T>.Dequeue() will be called by a thread pool thread at the same time Queue<T>.Enqueue is called by the main thread. This is not threadsafe.

If the main thread calling InternalFireQueuedAsync is a thread with a single-threaded context (such as a UI thread), then one simple fix is to remove all the instances of ConfigureAwait(false) in this method.

Answer 2

To be safe, you should mark field _firing as volatile - that will guarantee the memory barrier and be sure that the continuation part, which might run on a different thread, will read the correct value. Without volatile, the compiler, the CLR or the JIT compiler, or even the CPU may do some optimizations that cause the code to read a wrong value for it.

As for _eventQueue, you don't modify the field, so marking it as volatile is useless. If only one thread calls 'InternalFireQueuedAsync', you don't access it from multiple threads concurrently, so you are ok.

However, if multiple threads call InternalFireQueuedAsync, you will need to use a ConcurrentQueue instead, or lock your access to _eventQueue. You then better also lock your access to _firing, or access it using Interlocked, or replace it with a ManualResetEvent.

Answer 3

ConfigureAwait(false) means that the Context is not captured to run the continuation. Using the Thread Pool Context does not mean that continuations are run in parallel. Using await before and within the while loop ensures that the code (continuations) are run sequentially so no need to lock in this case. You may have however a race condition when checking the _firing value.

Answer 4

use lock or ConcurrentQueue.

solution with lock:

private readonly Queue<QueuedTrigger> _eventQueue = new Queue<QueuedTrigger>();
private bool _firing;
private object _eventQueueLock = new object();

async Task InternalFireQueuedAsync(TTrigger trigger, params object[] args)
{
if (_firing)
{
    lock(_eventQueueLock)
       _eventQueue.Enqueue(new QueuedTrigger { Trigger = trigger, Args = args });
    return;
}

try
{
    _firing = true;

    await InternalFireOneAsync(trigger, args).ConfigureAwait(false);

    lock(_eventQueueLock)
    while (_eventQueue.Count != 0)
    {
        var queuedEvent = _eventQueue.Dequeue();
        await InternalFireOneAsync(queuedEvent.Trigger, queuedEvent.Args).ConfigureAwait(false);
    }
}


finally
{
    _firing = false;
}

}

solution with ConcurrentQueue:

private readonly ConccurentQueue<QueuedTrigger> _eventQueue = new ConccurentQueue<QueuedTrigger>();
private bool _firing;

async Task InternalFireQueuedAsync(TTrigger trigger, params object[] args)
{
if (_firing)
{
    _eventQueue.Enqueue(new QueuedTrigger { Trigger = trigger, Args = args });
    return;
}

try
{
    _firing = true;

    await InternalFireOneAsync(trigger, args).ConfigureAwait(false);

    lock(_eventQueueLock)
    while (_eventQueue.Count != 0)
    {
        object queuedEvent; // change object > expected type
        if(!_eventQueue.TryDequeue())
           continue;
        await InternalFireOneAsync(queuedEvent.Trigger, queuedEvent.Args).ConfigureAwait(false);
    }
}


finally
{
    _firing = false;
}

}