Because a task is able to completely represent an asynchronous operation and provide synchronous and asynchronous capabilities for joining with the operation, retrieving its results, and so on, you can build useful libraries of combinators that compose tasks to build larger patterns.
// ----------------------------------------------------------------------------
// Building Task-based Combinators
//
// Because a task is able to completely represent an asynchronous operation and
// provide synchronous and asynchronous capabilities for joining with the
// operation, retrieving its results, and so on, you can build useful libraries of
// combinators that compose tasks to build larger patterns. As discussed in the
// previous section, the .NET Framework includes several built-in combinators, but
// you can also build your own. The following sections provide several examples of
// potential combinator methods and types.
//
// https://msdn.microsoft.com/en-us/library/hh873173(v=vs.110).aspx
// ----------------------------------------------------------------------------
/// <summary>
/// Retry On Fault Sync
///
/// In many situations, you may want to retry an operation if a previous
/// attempt fails. For synchronous code, you might build a helper method
/// such as RetryOnFault in the following example to accomplish this:
/// </summary>
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);
}
/// <summary>
/// Retry On Fault Async
///
/// You can build an almost identical helper method for asynchronous operations
/// that are implemented with TAP and thus return tasks:
/// </summary>
/// <example>
/// Download the URL, trying up to three times in case of failure.
/// <code>
/// string pageContents = await RetryOnFault(
/// () => DownloadStringAsync(url), 3);
/// </code>
/// </example>
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);
}
/// <summary>
/// You could extend the RetryOnFault function further. For example, the function
/// could accept another Func<Task> that will be invoked between retries to
/// determine when to try the operation again; for example:
/// </summary>
/// <example>
/// You could then use the function as follows to wait for a second before
/// retrying the operation:
/// <code>
/// // 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));
/// </code>
/// </example>
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().ConfigureAwait(false);
}
catch
{
if (i == maxTries - 1)
{
throw;
}
}
await retryWhen().ConfigureAwait(false);
}
return default(T);
}
/// <summary>
/// Need Only One
//
/// Sometimes, you can take advantage of redundancy to improve an operation's
/// latency and chances for success. Consider multiple web services that
/// provide stock quotes, but at various times of the day, each service may
/// provide different levels of quality and response times. To deal with
/// these fluctuations, you may issue requests to all the web services, and
/// as soon as you get a response from one, cancel the remaining requests.
/// You can implement a helper function to make it easier to implement this
/// common pattern of launching multiple operations, waiting for any, and
/// then canceling the rest. The NeedOnlyOne function in the following example
/// illustrates this scenario:
/// </summary>
/// <example>
/// <code>
/// double currentPrice = await NeedOnlyOne(
/// ct => GetCurrentPriceFromServer1Async("msft", ct),
/// ct => GetCurrentPriceFromServer2Async("msft", ct),
/// ct => GetCurrentPriceFromServer3Async("msft", ct));
/// </code>
/// </example>
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;
}
/// <summary>
/// Interleaved Operations
///
/// There is a potential performance problem with using the WhenAny method to
/// support an interleaving scenario when you're working with very large sets of
/// tasks. Every call to WhenAny results in a continuation being registered with
/// each task. For N number of tasks, this results in O(N2) continuations created
/// over the lifetime of the interleaving operation. If you're working with a large
/// set of tasks, you can use a combinator (Interleaved in the following example) to
/// address the performance issue:
/// </summary>
/// <example>
/// You can then use the combinator to process the results of tasks as they
/// complete; for example:
///
/// <code>
/// IEnumerable<Task<int>> tasks = ...;
/// foreach(var task in Interleaved(tasks))
/// {
/// int result = await task;
/// ...
/// }
/// </code>
///
/// </example>
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;
}
/// <summary>
/// When All Or First Exception
///
/// 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. You can accomplish that with a combinator method such
/// as WhenAllOrFirstException in the following example:
/// </summary>
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;
}
/// <summary>
/// AsyncCache
///
/// One important aspect of a task is that it may be handed out to multiple
/// consumers, all of whom may await it, register continuations with it, get its
/// result or exceptions (in the case of Task<TResult>), and so on. This makes Task
/// and Task<TResult> perfectly suited to be used in an asynchronous caching
/// infrastructure. Here's an example of a small but powerful asynchronous cache
/// built on top of Task<TResult>:
/// </summary>
/// <example>
/// The AsyncCache<TKey,TValue> class accepts as a delegate to its constructor a
/// function that takes a TKey and returns a Task<TResult>. Any previously accessed
/// values from the cache are stored in the internal dictionary, and the AsyncCache
/// ensures that only one task is generated per key, even if the cache is accessed
/// concurrently.
///
/// For example, you can build a cache for downloaded web pages:
///
/// <code>
/// private AsyncCache<string,string> m_webPages =
/// new AsyncCache<string,string>(DownloadStringAsync);
/// </code>
///
/// You can then use this cache in asynchronous methods whenever you need the
/// contents of a web page. The AsyncCache class ensures that you’re downloading as
/// few pages as possible, and caches the results.
///
/// <code>
/// private async void btnDownload_Click(object sender, RoutedEventArgs e)
/// {
/// btnDownload.IsEnabled = false;
/// try
/// {
/// txtContents.Text = await m_webPages["http://www.microsoft.com"];
/// }
/// finally { btnDownload.IsEnabled = true; }
/// }
/// </code>
/// </example>
public class AsyncCache<TKey, TValue>
{
private readonly Func<TKey, Task<TValue>> _valueFactory;
private readonly ConcurrentDictionary<TKey, Lazy<Task<TValue>>> _map;
public AsyncCache(Func<TKey, Task<TValue>> valueFactory)
{
if (valueFactory == null)
{
throw new ArgumentNullException("loader");
}
_valueFactory = valueFactory;
_map = new ConcurrentDictionary<TKey, Lazy<Task<TValue>>>();
}
public Task<TValue> this[TKey key]
{
get
{
if (key == null)
{
throw new ArgumentNullException("key");
}
return _map.GetOrAdd(key, toAdd =>
new Lazy<Task<TValue>>(() => _valueFactory(toAdd))).Value;
}
}
}