Invoking multiple tasks

Question

I've a C# application in which I need to invoke four asynchronous tasks(which internally call a third-party webservice). Each task returns a boolean value true/false depending upon success or failure.

I need to invoke another method say PostProcessing() once any of these 4 tasks return true. For e.g. if #2 method call returns true, I need to abort processing and invoke PostProcessing() method.

If all tasks return false, I dont want to invoke PostProcessing().

What's the best way to implement this approach please?Is it Task.ContinueWith()??

Thanks.

Answer 1

I would use Microsoft's Reactive Framework (Rx) for this - it becomes drop dead easy.

To start with I'll assume that the signatures of the methods you're working with are these:

public async Task<bool> WebService1()
public async Task<bool> WebService2()
public async Task<bool> WebService3()
public async Task<bool> WebService4()
public void PostProcessing()

Now you can set this use Rx like this:

var webservices = new Func<Task<bool>>[]
{
    WebService1, WebService2, WebService3, WebService4,
};

IObservable<bool> query =
    webservices
        .ToObservable()
        .SelectMany(ws => Observable.FromAsync(ws))
        .Where(b => b == true)
        .Take(1);

IDisposable subscription = query.Subscribe(b => PostProcessing());

This nicely calls all four web asynchronous services (Observable.FromAsync(ws)) asynchronously .ToObservable().SelectMany(...) and then filters the result to only those that return true (.Where(b => b == true)). It finally only wants one result .Take(1).

Then, if it does get one result - which must be true - it then calls PostProcessing when that occurs.

If you need to abort before any web service returns you can always call subscription.Dispose().

Simple.

Answer 2

Personally, I would probably use Reactive Extensions (like @Enigmativity) mentioned. If you want to avoid using Reactive Extensions, though, then I think you could combine Task.WhenAny with a while loop. If I were to go this route I would create a static method to keep things clean. So something like this:

public static class TaskExtensions
{
    public static async Task<Task<TResult>> WhenAnyWithPredicate<TResult>(Func<Task<TResult>, bool> predicate, params Task<TResult>[] tasks)
    {
        if (tasks == null)
        {
            throw new ArgumentNullException();
        }

        if (tasks.Length == 0)
        {
            return null;    
        }

        // Make a safe copy (in case the original array is modified while we are awaiting).
        tasks = tasks.ToArray();

        // Await the first task.
        var result = await Task.WhenAny(tasks);

        // Test the task and await the next task if necessary.
        while (tasks.Length > 0 && !predicate(result))
        {
            tasks = tasks.Where(x => x != result).ToArray();
            if (tasks.Length == 0)
            {
                result = null;
            }
            else
            {
                result = await Task.WhenAny(tasks);
            }
        }

        // Return the result.
        return result;
    }
}

You would use it like this.

CancellationTokenSource cts = new CancellationTokenSource();

// Start your four tasks.
var tasks = new Task<bool>[]
{
    CreateTask1WithCancellationToken(cts.Token),
    CreateTask2WithCancellationToken(cts.Token),
    CreateTask3WithCancellationToken(cts.Token),
    CreateTask4WithCancellationToken(cts.Token),
}

// Wait for the first task with a result of 'true' to complete.
var result = await TaskExtensions.WhenAnyWithPredicate(x => x.Status == TaskStatus.RanToCompletion && x.Result, tasks);

// Cancel the remaining tasks (if any).
cts.Cancel();

// If you have a nonnull task then success!
if (result != null)
{
    PostProcessing();
}