How to avoid reentrancy with async void event handlers?

Question

In a WPF application, I have a class that receives messages over the network. Whenever an object of said class has received a full message, an event is raised. In the MainWindow of the application I have an event handler subscribed to that event. The event handler is guaranteed to be called on the GUI thread of the application.

Whenever the event handler is called, the contents of the message needs to be applied to the model. Doing so can be quite costly (>200ms on current hardware). That's why applying the message is offloaded onto the thread pool with Task.Run.

Now, messages can be received in very close succession, so the event handler can be called while a previous change is still being processed. What is the simplest way to ensure that messages are only applied one at time? So far, I've come up with the following:

using System;
using System.Threading.Tasks;
using System.Windows;

public partial class MainWindow : Window
{
    private Model model = new Model();
    private Task pending = Task.FromResult<bool>(false);

    // Assume e carries a message received over the network.
    private void OnMessageReceived(object sender, EventArgs e)
    {
        this.pending = ApplyToModel(e);
    }

    private async Task ApplyToModel(EventArgs e)
    {
        await this.pending;
        await Task.Run(() => this.model.Apply(e)); // Assume this is an expensive call.
    }
}

This seems to work as expected, however it also appears this will inevitably produce a "memory leak", because the task to apply a message will always first wait on the task that applied the previous message. If so, then the following change should avoid the leak:

private async Task ApplyToModel(EventArgs e)
{
    if (!this.pending.IsCompleted)
    {
        await this.pending;
    }

    await Task.Run(() => this.model.Apply(e));
}

Is this a sensible way to avoid reentrancy with async void event handlers?

EDIT: Removed the unnecessary await this.pending; statement in OnMessageReceived.

EDIT 2: The messages must be applied to the model in the same order in which they have been received.

Answer 1

Given an eventhandler which uses async await we cannot use a lock outside the Task because the calling thread is the same for every event call so the lock will always let it pass.

var object m_LockObject = new Object();

private async void OnMessageReceived(object sender, EventArgs e)
{
    // Does not work
    Monitor.Enter(m_LockObject);

    await Task.Run(() => this.model.Apply(e));

    Monitor.Exit(m_LockObject);
}

But we can lock inside the Task because Task.Run always generates a new Task which is not run in parallel on the same thread

var object m_LockObject = new Object();

private async void OnMessageReceived(object sender, EventArgs e)
{
    await Task.Run(() => 
    {
        // Does work
        lock(m_LockObject)
        {
            this.model.Apply(e);
        }
    });
}

So when an event calls OnMessageReceived it returns immidiatly and model.Apply is only entered one after another.

Answer 2

We need to thank Stephen Toub here, as he has some very useful async locking constructs demonstrated in a blog series, including an async lock block.

Here is the code from that article (including some code from the previous article in the series):

public class AsyncLock
{
    private readonly AsyncSemaphore m_semaphore;
    private readonly Task<Releaser> m_releaser;

    public AsyncLock()
    {
        m_semaphore = new AsyncSemaphore(1);
        m_releaser = Task.FromResult(new Releaser(this));
    }

    public Task<Releaser> LockAsync()
    {
        var wait = m_semaphore.WaitAsync();
        return wait.IsCompleted ?
            m_releaser :
            wait.ContinueWith((_, state) => new Releaser((AsyncLock)state),
                this, CancellationToken.None,
                TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
    }

    public struct Releaser : IDisposable
    {
        private readonly AsyncLock m_toRelease;

        internal Releaser(AsyncLock toRelease) { m_toRelease = toRelease; }

        public void Dispose()
        {
            if (m_toRelease != null)
                m_toRelease.m_semaphore.Release();
        }
    }
}

public class AsyncSemaphore
{
    private readonly static Task s_completed = Task.FromResult(true);
    private readonly Queue<TaskCompletionSource<bool>> m_waiters = new Queue<TaskCompletionSource<bool>>();
    private int m_currentCount;

    public AsyncSemaphore(int initialCount)
    {
        if (initialCount < 0) throw new ArgumentOutOfRangeException("initialCount");
        m_currentCount = initialCount;
    }
    public Task WaitAsync()
    {
        lock (m_waiters)
        {
            if (m_currentCount > 0)
            {
                --m_currentCount;
                return s_completed;
            }
            else
            {
                var waiter = new TaskCompletionSource<bool>();
                m_waiters.Enqueue(waiter);
                return waiter.Task;
            }
        }
    }
    public void Release()
    {
        TaskCompletionSource<bool> toRelease = null;
        lock (m_waiters)
        {
            if (m_waiters.Count > 0)
                toRelease = m_waiters.Dequeue();
            else
                ++m_currentCount;
        }
        if (toRelease != null)
            toRelease.SetResult(true);
    }
}

Now applying it to your case:

private readonly AsyncLock m_lock = new AsyncLock();

private async void OnMessageReceived(object sender, EventArgs e)
{
    using(var releaser = await m_lock.LockAsync()) 
    {
        await Task.Run(() => this.model.Apply(e));
    }
}