Enforce an Async Method to Be Called Once

Enforce an async method to be called once

I'd go with AsyncLazy<T> (slightly modified version):

public class AsyncLazy<T> : Lazy<Task<T>> 
{ 
    public AsyncLazy(Func<T> valueFactory) : 
        base(() => Task.Run(valueFactory)) { }

    public AsyncLazy(Func<Task<T>> taskFactory) : 
        base(() => Task.Run(() => taskFactory())) { } 

    public TaskAwaiter<T> GetAwaiter() { return Value.GetAwaiter(); } 
}

And consume it like this:

private AsyncLazy<bool> asyncLazy = new AsyncLazy<bool>(async () =>
                                    { 
                                        await DoStuffOnlyOnceAsync()
                                        return true;
                                    });

Note i'm using bool simply because you have no return type from DoStuffOnlyOnceAsync.

Edit:

Stephan Cleary (of course) also has an implementation of this here.

Enforce an async method to be called lazily on demand, and called again when the previous result has expired

Here is an implementation of an AsyncExpiringLazy<T> class, which is essentially an AsyncLazy<T> with added expiration functionality:

/// <summary>
/// Represents the result of an asynchronous operation that is invoked lazily
/// on demand, and is subject to an expiration policy. In case of failure the
/// operation is retried as many times as needed until it succeeds.
/// </summary>
/// <remarks>
/// Subsequent executions do not overlap under any circumstances. Concurrent
/// observers receive the result of the same operation.
/// </remarks>
public class AsyncExpiringLazy<TResult>
{
    private readonly Func<Task<TResult>> _taskFactory;
    private readonly Func<TResult, TimeSpan> _expirationSelector;
    private Tuple<Task<TResult>, TimeSpan, long> _state;

    public AsyncExpiringLazy(Func<Task<TResult>> taskFactory,
        Func<TResult, TimeSpan> expirationSelector)
    {
        ArgumentNullException.ThrowIfNull(taskFactory);
        ArgumentNullException.ThrowIfNull(expirationSelector);
        _taskFactory = taskFactory;
        _expirationSelector = expirationSelector;
    }

    public Task<TResult> Task
    {
        get
        {
            var capturedState = Volatile.Read(ref _state);
            while (true)
            {
                if (capturedState is not null)
                {
                    // Check if the task has expired, using static Stopwatch methods.
                    var (task, expiration, timestamp) = capturedState;
                    long elapsedTicks = Stopwatch.GetTimestamp() - timestamp;
                    TimeSpan elapsed = TimeSpan
                        .FromSeconds((double)elapsedTicks / Stopwatch.Frequency);
                    if (elapsed < expiration)
                        return task; // The task has not expired.
                }

                // Either this is the first call, or the task expired or failed.
                Tuple<Task<TResult>, TimeSpan, long> runningState = null;
                var newTaskTask = new Task<Task<TResult>>(_taskFactory);
                var newTask = newTaskTask.Unwrap().ContinueWith(task =>
                {
                    Tuple<Task<TResult>, TimeSpan, long> completedState = null;
                    try
                    {
                        if (task.IsCompletedSuccessfully)
                        {
                            var expiration = _expirationSelector(task.Result);
                            completedState = Tuple.Create(runningState.Item1,
                                expiration, Stopwatch.GetTimestamp());
                        }
                    }
                    finally
                    {
                        // In case the task or the selector failed, the _state will
                        // be updated to null, to trigger a retry later.
                        var original = Interlocked.CompareExchange(
                            ref _state, completedState, runningState);
                        Debug.Assert(ReferenceEquals(original, runningState));
                    }
                    return task;
                }, default, TaskContinuationOptions.DenyChildAttach |
                    TaskContinuationOptions.ExecuteSynchronously,
                    TaskScheduler.Default).Unwrap();

                // While the task is running, the expiration is set to never.
                runningState = Tuple.Create(newTask, TimeSpan.MaxValue, 0L);

                var originalState = Interlocked
                    .CompareExchange(ref _state, runningState, capturedState);
                if (ReferenceEquals(originalState, capturedState))
                {
                    // The state was updated successfully.
                    // Run the cold nested task on the current thread.
                    newTaskTask.RunSynchronously(TaskScheduler.Default);
                    return runningState.Item1;
                }

                // We lost the race to update the state.
                // It's extremely rare but not impossible that we must spin.
                capturedState = originalState;
                if (capturedState is null) continue;
                return capturedState.Item1;
            }
        }
    }

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

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

    public bool ExpireImmediately()
    {
        var capturedState = Volatile.Read(ref _state);
        if (capturedState is null || !capturedState.Item1.IsCompleted) return false;
        return ReferenceEquals(Interlocked
            .CompareExchange(ref _state, null, capturedState), capturedState);
    }
}

Usage example:

_webApi = new WebApi();
_accessToken = new AsyncExpiringLazy<AccessToken>(
    async () => await _webApi.Authenticate("xxx", "yyy"), _ => TimeSpan.FromMinutes(15));

await _webApi.PurchaseItem(await _accessToken, itemId, quantity);

This implementation is a modified version of the AsyncLazy<T> class that can be found in this answer. The AsyncExpiringLazy<T> constructor accepts two delegates. The taskFactory is the asynchronous method that produces the result, and it is invoked on the calling thread (the thread that calls the await _accessToken in the example above). The expirationSelector is a selector of the expiration period, which is a TimeSpan, and takes the produced result as argument. This delegate is invoked on an unknown thread (usually on the ThreadPool), immediately after a result has been asynchronously produced.

The ExpireImmediately method causes the immediate expiration of the previously completed task. In case a task is currently running, or in case the previous task failed, this method has no effect.

This implementation propagates all the exceptions that might be thrown by the taskFactory delegate, not just the first one.

An online demonstration of the AsyncExpiringLazy<T> class can be found here. It demonstrates the behavior of the class when used by multiple concurrent workers, and when the taskFactory fails.

How to call asynchronous method from synchronous method in C#?

Asynchronous programming does "grow" through the code base. It has been compared to a zombie virus. The best solution is to allow it to grow, but sometimes that's not possible.

I have written a few types in my Nito.AsyncEx library for dealing with a partially-asynchronous code base. There's no solution that works in every situation, though.

Solution A

If you have a simple asynchronous method that doesn't need to synchronize back to its context, then you can use Task.WaitAndUnwrapException:

var task = MyAsyncMethod();
var result = task.WaitAndUnwrapException();

You do not want to use Task.Wait or Task.Result because they wrap exceptions in AggregateException.

This solution is only appropriate if MyAsyncMethod does not synchronize back to its context. In other words, every await in MyAsyncMethod should end with ConfigureAwait(false). This means it can't update any UI elements or access the ASP.NET request context.

Solution B

If MyAsyncMethod does need to synchronize back to its context, then you may be able to use AsyncContext.RunTask to provide a nested context:

var result = AsyncContext.RunTask(MyAsyncMethod).Result;

---

*Update 4/14/2014: In more recent versions of the library the API is as follows:

var result = AsyncContext.Run(MyAsyncMethod);

---

(It's OK to use Task.Result in this example because RunTask will propagate Task exceptions).

The reason you may need AsyncContext.RunTask instead of Task.WaitAndUnwrapException is because of a rather subtle deadlock possibility that happens on WinForms/WPF/SL/ASP.NET:

1. A synchronous method calls an async method, obtaining a Task.
2. The synchronous method does a blocking wait on the Task.
3. The async method uses await without ConfigureAwait.
4. The Task cannot complete in this situation because it only completes when the async method is finished; the async method cannot complete because it is attempting to schedule its continuation to the SynchronizationContext, and WinForms/WPF/SL/ASP.NET will not allow the continuation to run because the synchronous method is already running in that context.

This is one reason why it's a good idea to use ConfigureAwait(false) within every async method as much as possible.

Solution C

AsyncContext.RunTask won't work in every scenario. For example, if the async method awaits something that requires a UI event to complete, then you'll deadlock even with the nested context. In that case, you could start the async method on the thread pool:

var task = Task.Run(async () => await MyAsyncMethod());
var result = task.WaitAndUnwrapException();

However, this solution requires a MyAsyncMethod that will work in the thread pool context. So it can't update UI elements or access the ASP.NET request context. And in that case, you may as well add ConfigureAwait(false) to its await statements, and use solution A.

Update, 2019-05-01: The current "least-worst practices" are in an MSDN article here.

I need not wait until executing async method call()

The main thread is blocked on the call to future.get() until the task completes. Remove this call and your second log statement will print immediately.

That addresses what you asked exactly. But I doubt it is what you want.

The purpose of submitting an asynchronous task is to permit the main thread to carry on immediately with other work. If the main thread requires the result of the task to proceed, the task is not a candidate for performing asynchronously.

Instead, add the processing of the result to the asynchronous task itself. Then the main thread does not have to wait for the result.

What exactly happens when you call an async method without the await keyword?

Then my periodic job and PutRecordBatchAsync are processed concurrently?

Using Task API you can ensure that they are executed concurrently (using Thread pool), but you need to understand the difference between in memory concurrent operations Vs IO based concurrency.

While In memory concurrency does benefit using Tasks, IO call once executed doesn't need thread at all, as it relies on hardware concurrency, if it ever use the thread, all that it would do it wait for the IO call to return, thus wasting the precious system resources and reducing system scalability

You case is that IO based concurrency, as you call a remote / network based API, how does async-await helps here ?

Well true Async operation will free up the thread context, on windows it would use IO completion port (queuing mechanism) to execute the Async call, while the calling thread is used to dispatch other similar calls, it would just need thread context on return of the IO call for serving the response and for that too, if its not a UI call, then use ConfigureAwait(false), so that any thread context can be used to deliver the response.

What if you don't use await with async ?

The call meant to be Asynchronous becomes Synchronous and would immediately impact the system scalability, as threads are now blocked, even worse for a long running IO operations. Have you seen how JavaScript frameworks always make a AJAX (Async) call to the server API, thus much more work is possible W/o blocking the Browser threads.

In general for In memory processing you would create certain number of Tasks and process them using Task.WaitAll or Parallel.ForEach for a collection, for Async processing, ideally recommendation is not to have a Task.Run anywhere, its preferred to have to Async from the entry point, like its possible in case of MVC, controllers can be async. Multiple calls are grouped together using Task.WhenAll representative Task and then awaited upon. even if you use Task.Run as in your code, then use async lambda to execute an asynchronous call

Summary :

Its mandatory to use await for asynchronous calls, else they async keyword is useless in that context and yes await will wait for the IO call to return before continuation is executed, though no thread is blocked in the process

Allow async method to be called only one instance at a time

You need some sort of async lock. Stephen Toub has a whole series of articles about building async synchronization primitives (including AsyncLock). A version of AsyncLock is also contained in Stephen Cleary's AsyncEx library.

But probably a simpler solution would be to use the built-in SemaphoreSlim, which does support asynchronous waiting:

private static SemaphoreSlim SlowStuffSemaphore = new SemaphoreSlim(1, 1);

private static async void CallSlowStuff () {
    await SlowStuffSemaphore.WaitAsync();
    try {
        await DoSlowStuff();
        Console.WriteLine("Done!");
    }
    catch (Exception e) {
        Console.WriteLine(e.Message);
    }
    finally {
        SlowStuffSemaphore.Release();
    }
}

Ensure an async method is never executed in parallel

There are no problems with this approach, using a promise queue is a standard pattern that will solve your problem.

While your code does work fine, I would recommend to avoid the Promise constructor antipattern:

let workQueue: Promise<void> = Promise.resolve();
const ignore = _ => {};
function executeSequentially<T>(action: () => Promise<T>): Promise<T> {
    const result = workQueue.then(action);
    workQueue = result.then(ignore, ignore);
    return result;
}

Also you say that you'd be "calling the method" like executeSequentially(() => asyncThreadUnsafeMethod(123));, but this does leave the potential for mistakes where you forget the executeSequentially wrapper and still call the unsafe method directly. Instead I'd recommend to introduce an indirection where you wrap the entire safe call in a function and then export only that, having your codebase only interact with that facade and never with the library itself. To help with the creation of such a facade, I'd curry the executeSequentially function:

const ignore = _ => {};
function makeSequential<T, A extends unknown[]>(fn: (...args: A) => Promise<T>) => (...args: A) => Promise<T> {
    let workQueue: Promise<void> = Promise.resolve();
    return (...args) => {
        const result = workQueue.then(() => fn(...args));
        workQueue = result.then(ignore, ignore);
        return result;
    };
}

export const asyncSafeMethod = makeSequential(asyncThreadUnsafeMethod);

or just directly implement it for that one case

const ignore = _ => {};
let workQueue: Promise<void> = Promise.resolve();

export function asyncSafeMethod(x: number) {
    const result = workQueue.then(() => asyncThreadUnsafeMethod(x));
    workQueue = result.then(ignore, ignore);
    return result;
}

import { asyncSafeMethod } from …;

asyncSafeMethod(123);

---

Related Topics