How to make multiple concurrent calls to async function collapse into 1 task?

Question

I have a C# async function like SlowRewriteFolder(), and I have multiple calls of this function coming in asynchronously.

If a call to this function is already processing, I want subsequent callers to not kick off the work that this function does again and instead wait on the same result (especially while the first one is still in progress).

How can I make it so that the Task created for the first call is shared among subsequent callers while it is still in progress?

I have considered caching the Task instance and returning that if it is available and clearing it when the work is complete, but is that the best approach?

Answer 1

I have considered caching the Task instance and returning that if it is available and clearing it when the work is complete, but is that the best approach?

-ish. You'll need to ensure that "return or restart" logic is thread-safe.

Something like this...

• My simplistic approach uses a basic lock() inside a non-async-but-Task-returning method to do the job of swapping/resetting Task instances stored in a class field.

  • Note that the volatile keyword isn't needed here at all, and reference-type assignment is atomic in the CLR.
  • Note the usual lock() complications:
    • The lock() target is a private readonly Object which is not locked by anything else - this prevents deadlocks or bugs caused by misbehaving consumers.
    • Shared state (i.e. the this.cachedTask) is always first captured in a local (the task variable) before being inspected or compared.
    • An outer optimistic if( task is null || task.IsComplete ) guard means most invocations will never encounter the lock() at all.
    • Inside the lock another guard if prevents a consecutive thread from repeating the first thread's work.

• I'm sure my code can be improved though: I'm sure there's probably a lock-free way to do this (and by "lock-free" I don't mean that the lock is still there but buried in library code, like in ConcurrentDictionary) but I mean completely lock-free.

• BTW, the Interlocked.Increment( ref _invokeCount ) in SlowRewriteFolderImplAsync is just for demonstration purposes: it's to prove that SlowRewriteFolderImplAsync is only ever entered twice (in this example). It's unrelated to the logic in SlowRewriteFolderAsync.

• Also, don't use Task< List<T> > in situations like these... or more generally: don't use a mutable TResult with Task<TResult> when a single Task<TResult> may (or will) be awaited by multiple callers - this is because the current _{(albeit loose)} convention in .NET is that _{(sans documentation to the contrary)} the end-receiver of a Task<TResult> becomes the owner of the TResult contained within - this isn't a problem when TResult is immutable (or at least, is a read-only interface), as there's no risk of multiple end-receivers mutating TResult and breaking each other - but when TResult is a mutable type (e.g. some POCO with get; set; auto-properties) - or worse: a non-thread-safe mutable type, like List<T>) then that's when programs fall apart.

  • This is why my code below returns IReadOnlyList<String> instead of List<String>. If you want to be extra-robust in your own code then you should consider returning ImmutableArray<T> or ImmutableList<T> as those provide stronger-yet guarantees about immutablility that IReadOnlyList<T> does not.

public class FolderRewriter
{
    private readonly Object lockObj = new Object();
    private Task< IReadOnlyList<String> >? cachedTask;

    public Task< IReadOnlyList<String> > SlowRewriteFolderAsync()
    {
        Task< IReadOnlyList<String> >? task = this.cachedTask;

        if( task is null || task.IsCompleted )
        {
           lock( this.lockObj )
           {
                if( this.cachedTask is null || this.cachedTask.IsCompleted )
                {
                    this.cachedTask = this.SlowRewriteFolderImplAsync();
                    task = this.cachedTask;
                }
                else
                {
                    task = this.cachedTask;
                }
            }
        }

        return task;
    }
    
    private static Int32 _invokeCount = 0;

    private async Task< IReadOnlyList<String> > SlowRewriteFolderImplAsync()
    {
        Int32 c = Interlocked.Increment( ref _invokeCount );
        
        await Task.Yield();
        
        const String FMT = nameof(SlowRewriteFolderImplAsync) + "(), UtcNow: {0:o} Invoke Count: {1:N0}";
        FMT.FmtInv( DateTime.UtcNow, c ).Dump();
        
        return Array.Empty<String>();
    }
}

Usage:

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

async Task Main()
{
    FolderRewriter rewriter = new FolderRewriter();
    
    // First:
    {
        Task< IReadOnlyList<String> > task0 = rewriter.SlowRewriteFolderAsync();
        Task< IReadOnlyList<String> > task1 = rewriter.SlowRewriteFolderAsync();
        Task< IReadOnlyList<String> > task2 = rewriter.SlowRewriteFolderAsync();

        Debug.Assert( Object.ReferenceEquals( task0, task1 ) );
        Debug.Assert( Object.ReferenceEquals( task1, task2 ) );

        IReadOnlyList<String> result0 = await task0.ConfigureAwait(false);
        IReadOnlyList<String> result1 = await task1.ConfigureAwait(false);
        IReadOnlyList<String> result2 = await task2.ConfigureAwait(false);
        
        Debug.Assert( Object.ReferenceEquals( result0, result1 ) );
        Debug.Assert( Object.ReferenceEquals( result0, result2 ) );
    }
    
    // Second:
    {
        Task< IReadOnlyList<String> > task0 = rewriter.SlowRewriteFolderAsync();
        Task< IReadOnlyList<String> > task1 = rewriter.SlowRewriteFolderAsync();
        Task< IReadOnlyList<String> > task2 = rewriter.SlowRewriteFolderAsync();

        Debug.Assert( Object.ReferenceEquals( task0, task1 ) );
        Debug.Assert( Object.ReferenceEquals( task1, task2 ) );

        IReadOnlyList<String> result0 = await task0.ConfigureAwait(false);
        IReadOnlyList<String> result1 = await task1.ConfigureAwait(false);
        IReadOnlyList<String> result2 = await task2.ConfigureAwait(false);
        
        Debug.Assert( Object.ReferenceEquals( result0, result1 ) );
        Debug.Assert( Object.ReferenceEquals( result0, result2 ) );
    }
}

Gives me this output in Linqpad - as you can see, the SlowRewriteFolderImplAsync method is only invoked twice, not 6 times:

Note that it gets far more gnarly if you want to use CancellationToken with SlowRewriteFolderAsync as the SlowRewriteFolderImplAsync will only have access to the CancellationToken of the first invocation, so subsequent invocations cannot be canceled.

Answer 2

Here is a component similar in shape with the AsyncLazy<T> type (also available in the Nito.AsyncEx library by Stephen Cleary), that has a behavior tailored to your needs:

/// <summary>
/// Represents an asynchronous operation that is invoked lazily on demand, can be
/// invoked multiple times, and is subject to a non-concurrent execution policy.
/// Concurrent observers receive the result of the same operation.
/// </summary>
public class AsyncCollapseConcurrent
{
    private readonly Func<Task> _taskFactory;
    private volatile Task _task;

    public AsyncCollapseConcurrent(Func<Task> taskFactory)
    {
        ArgumentNullException.ThrowIfNull(taskFactory);
        _taskFactory = taskFactory;
    }

    public Task Task
    {
        get
        {
            Task capturedTask = _task;
            if (capturedTask is not null) return capturedTask;
            Task<Task> newTaskTask = new(_taskFactory);
            Task newTask = newTaskTask.Unwrap().ContinueWith(t =>
            {
                _task = null;
                return t;
            }, default, TaskContinuationOptions.DenyChildAttach |
                TaskContinuationOptions.ExecuteSynchronously,
                TaskScheduler.Default).Unwrap();
            capturedTask = Interlocked
                .CompareExchange(ref _task, newTask, null) ?? newTask;
            if (ReferenceEquals(capturedTask, newTask))
                newTaskTask.RunSynchronously(TaskScheduler.Default);
            return capturedTask;
        }
    }

    public TaskAwaiter GetAwaiter() => Task.GetAwaiter();

    public ConfiguredTaskAwaitable ConfigureAwait(bool continueOnCapturedContext)
        => Task.ConfigureAwait(continueOnCapturedContext);
}

Usage example:

private readonly AsyncCollapseConcurrent _asyncLazy;

//...

_asyncLazy = new(() => SlowRewriteFolderAsync());

//...

await _asyncLazy;

The AsyncCollapseConcurrent ensures that the taskFactory will not be invoked concurrently, by creating a cold nested Task<Task> using the Task<T> constructor, and starting this task only in case the atomic Interlocked.CompareExchange operation succeeds. Otherwise, in case the race to update the _task field is won by another thread, the current thread discards the cold Task<Task> without starting it.

I have used this technique for implementing various AsyncLazy<T> variants, like this (with retry) or this (with expiration).

In case your SlowRewriteFolderAsync method returns a generic Task<TResult>, you can find a compatible generic AsyncCollapseConcurrent<TResult> class here.

Answer 3

If you only ever want the task to run once with multiple callers then the easy way is with Lazy<T>.

Try this:

public Lazy<Task<List<String>>> SlowRewriteFolderAsyncLazy =>
    new Lazy<Task<List<String>>>(() => SlowRewriteFolderAsync());

You then call it like this:

Lazy<Task<List<String>>> lazy = SlowRewriteFolderAsyncLazy;
Task<List<String>> task = lazy.Value;
List<String> value = await task;

The task within the Lazy<> type doesn't begin to run until the first caller invokes the .Value property, so this is safe to define SlowRewriteFolderAsyncLazy as a property.

All subsequent callers get the same completed task.