Building Task-based Combinators

Due to a task’s ability to completely represent an asynchronous operation and provide synchronous and asynchronous capabilities for joining with the operation, retrieving its results, and so forth, it becomes possible to build useful libraries of “combinators” that compose tasks to build larger patterns. As mentioned in the previous section of this document, the .NET Framework includes several built-in combinators, however it’s also possible and expected that developers will build their own. Here we provide several examples of potential combinator methods and types.

RetryOnFault

In many situations, it is desirable to retry an operation if a previous attempt at the operation fails. For synchronous code, we might build a helper method to accomplish this as follows:

public static T RetryOnFault<T>(

Func<T> function, int maxTries)

{

for(int i=0; i<maxTries; i++)

{

   try { return function(); }

   catch { if (i == maxTries-1) throw; }

}

return default(T);

}

 

We can build an almost identical helper method, but for asynchronous operations implemented with the TAP and thus returning tasks:

public static async Task<T> RetryOnFault<T>(

Func<Task<T>> function, int maxTries)

{

for(int i=0; i<maxTries; i++)

{

   try { return await function().ConfigureAwait(false); }

   catch { if (i == maxTries-1) throw; }

}

return default(T);

}

 

With our function in hand, we can now utilize this combinator to encode retries into our application’s logic, e.g.

// Download the URL, trying up to three times in case of failure

string pageContents = await RetryOnFault(

() => DownloadStringAsync(url), 3);

 

Our RetryOnFault function could be extended further, such as to accept another Func<Task> which will be invoked between retries in order to determine when it’s good to try again, e.g.

public static async Task<T> RetryOnFault<T>(

Func<Task<T>> function, int maxTries, Func<Task> retryWhen)

{

for(int i=0; i<maxTries; i++)

{

   try { return await function(); }

   catch { if (i == maxTries-1) throw; }

   await retryWhen().ConfigureAwait(false);

}

return default(T);

}

 

which could then be used like the following to wait for a second before retrying:

// Download the URL, trying up to three times in case of failure,

// and delaying for a second between retries

string pageContents = await RetryOnFault(

() => DownloadStringAsync(url), 3, () => Task.Delay(1000));

NeedOnlyOne

Sometimes redundancy is taken advantage of to improve an operation’s latency and chances for success. Consider multiple Web services that all provide stock quotes, but at various times of the day, each of the services may provide different levels of quality and response times. To deal with these, we may issues requests to all of the Web services, and as soon as we get any response, cancel the rest. We can implement a helper function to make easier this common pattern of launching multiple operations, waiting for any, and then canceling the rest:

public static async Task<T> NeedOnlyOne(

params Func<CancellationToken,Task<T>> [] functions)

{

var cts = new CancellationTokenSource();

var tasks = (from function in functions

            select function(cts.Token)).ToArray();

var completed = await Task.WhenAny(tasks).ConfigureAwait(false);

cts.Cancel();

foreach(var task in tasks)

{

   var ignored = task.ContinueWith(

       t => Log(t), TaskContinuationOptions.OnlyOnFaulted);

}

return completed;

}

 

This function can then be used to implement our example:

double currentPrice = await NeedOnlyOne(

ct => GetCurrentPriceFromServer1Async(“msft”, ct),

ct => GetCurrentPriceFromServer2Async(“msft”, ct),

ct => GetCurrentPriceFromServer3Async(“msft”, ct));

Interleaved

There is a potential performance problem with using Task.WhenAny to support an interleaving scenario when using very large sets of tasks. Every call to WhenAny will result in a continuation being registered with each task, which for N tasks will amount to O(N²) continuations created over the lifetime of the interleaving operation. To address that if working with a large set of tasks, one could use a combinatory dedicated to the goal:

static IEnumerable<Task<T>> Interleaved<T>(IEnumerable<Task<T>> tasks)

{

var inputTasks = tasks.ToList();

var sources = (from _ in Enumerable.Range(0, inputTasks.Count)

              select new TaskCompletionSource<T>()).ToList();

int nextTaskIndex = -1;

foreach (var inputTask in inputTasks)

{

   inputTask.ContinueWith(completed =>

   {

       var source = sources[Interlocked.Increment(ref nextTaskIndex)];

       if (completed.IsFaulted)

           source.TrySetException(completed.Exception.InnerExceptions);

       else if (completed.IsCanceled)

           source.TrySetCanceled();

       else

           source.TrySetResult(completed.Result);

   }, CancellationToken.None,

      TaskContinuationOptions.ExecuteSynchronously,

      TaskScheduler.Default);

}

return from source in sources

      select source.Task;

}

 

This could then be used to process the results of tasks as they complete, e.g.

IEnumerable<Task<int>> tasks = ...;

foreach(var task in tasks)

{

int result = await task;

…

}

 

WhenAllOrFirstException

In certain scatter/gather scenarios, you might want to wait for all tasks in a set, unless one of them faults, in which case you want to stop waiting as soon as the exception occurs. We can accomplish that with a combinator method as well, for example:

public static Task<T[]> WhenAllOrFirstException<T>(IEnumerable<Task<T>> tasks)
{

var inputs = tasks.ToList();

var ce = new CountdownEvent(inputs.Count);

var tcs = new TaskCompletionSource<T[]>();

Action<Task> onCompleted = (Task completed) =>
{
if (completed.IsFaulted)
tcs.TrySetException(completed.Exception.InnerExceptions);
if (ce.Signal() && !tcs.Task.IsCompleted)
tcs.TrySetResult(inputs.Select(t => t.Result).ToArray());
};
 
foreach (var t in inputs) t.ContinueWith(onCompleted);

return tcs.Task;
}

 

⇐ Предыдущая12345678Следующая ⇒

Поделиться: