PROCESSING SEQUENCES OF ASYNCHRONOUS OPERATIONS WITH TASKS
Of late, I’ve seen binary folks asking most how to ingest tasks to asynchronously fulfil a ordering of operations. For example, presented threesome coetaneous functions: open progress DoA(string input); open progress DoB(string aResult); open progress DoC(string bResult); you could advert these functions with cipher like: progress aResult = DoA(input); progress bResult = DoB(aResult); progress cResult = DoC(bResult); Then, presented Task-based anachronic counterparts to these functions: open noise Task<string> DoAAsync(string input); open noise Task<string> DoBAsync(string aResult); open noise Task<string> DoCAsync(string bResult); how would you asynchronously do the equal of the coetaneous cipher previously shown? Async CTP and Language Support The Async CTP highlights a enthusiastic artefact to appendage this erst C# and Visual Basic hit built-in hold for awaiting tasks. With that module support, the anachronic edition looks nearly aforementioned to the synchronous, albeit with a whatever player keywords tangled in: progress aResult = await DoAAsync(input); progress bResult = await DoBAsync(aResult); progress cResult = await DoCAsync(bResult); Lovely! Of course, .NET 4 doesn’t currently hold that primary structure out-of-the-box, so what crapper we do in the meantime? It’s essential to actualise that the Async CTP meet builds on crowning of what’s acquirable in .NET 4, so patch the programme here is doing what compilers do prizewinning and essay lots of boilerplate so that you don’t hit to, the generated cipher is rattling ease meet using existing hold in the Task Parallel Library. We crapper ingest that hold directly, too. ContinueWith and Unwrap One move is to meet ingest what’s provided in TPL without utilizing whatever added helpers. The Task.ContinueWith method schedules cipher to separate when the prior duty completes, and returns a Task (or Task<TResult>) to equal that ensuant operation. So, we crapper essay to appendage the prototypal bound call with cipher aforementioned the following: var aResult = DoAAsync(input); var bResult = aResult.ContinueWith(t => DoBAsync(t.Result)); This isn’t quite right, however. This burden of ContinueWith creates a Task<TResult>, but the TResult here module be written as the termination of DoBAsync, which is Task<string>. Thus, the event returned from the ContinueWith call module be a Task<Task<string>>, which is not what we want. To appendage those nested tasks, we crapper apply the Unwrap method included in TPL to “unwrap” the intrinsic nested task, converting the Task<Task<string>> into the Task<string> we desire. And with that diminutive change, we crapper today rank our example: var aResult = DoAAsync(input); var bResult = aResult.ContinueWith(t => DoBAsync(t.Result)).Unwrap(); var cResult = bResult.ContinueWith(t => DoCAsync(t.Result)).Unwrap(); This move works, and it haw be every you need, but it does hit a pair of downsides. First, what happens if DoAAsync fails much that the returned duty is faulted kinda than streaming to completion? In that case, the postscript soured of aResult module ease execute, and accessing the t.Result exclusive that duty module move the exception. That module intend bResult to be faulted, and its postscript module fire, with that continuation’s t.Result propagating the exception, and so on. Net net, cResult module befittingly be faulted, but more impact module hit happened than was necessary (e.g. we invoked the continuations for B and C when we didn’t hit to, and we threw individual more exceptions than we necessary to), and apiece instance the omission was propagated it was enwrapped in added verify of AggregateException, so cResult module include an AggregateException containing an AggregateException containing… and so on (AggregateException’s Flatten method is enthusiastic for reaction these player levels of aggregation, by the way). Second, it’s a taste more cipher than you haw otherwise aforementioned to write. Can we do better? Then One of the rattling pleasant things most Tasks is that they enable beatific composition. Once you hit a azygos identify confident of representing whatever capricious anachronic operation, you crapper indite “combinators” over the identify that earmark you to combine/compose anachronic dealings in a myriad of ways. For example, we crapper nonfigurative absent this sequential compounding idea into a Then combinator, and at the aforementioned instance verify tending of the previously mentioned omission direction concerns. Our content here module be to physique the mass operator: open noise Task<T2> Then<T1, T2>(this Task<T1> first, Func<T1, Task<T2>> next); With that in place, we crapper then ingest it to re-implement our desirable functionality with cipher like: var aResult = DoAAsync(input); var bResult = aResult.Then(s => DoBAsync(s)); var cResult = bResult.Then(s => DoCAsync(s)); The activity of “Then” is that when the prototypal duty completes, the incoming duty is invoked to display a duty and is provided with the production from the preceding task. The duty returned from Then primarily represents the duty returned by the incoming function’s invocation; however, if the prototypal duty is canceled or faults, the incoming duty module not be invoked, and the returned duty module instead equal the prototypal task. With this approach, we crapper concern Then calls unitedly as was finished in the above example, and the test duty produced from the concern module equal every of the processing. To compel Then, we prototypal requirement the system to do discussion determination and to create the duty that Then module return: open noise Task<T2> Then<T1, T2>(this Task<T1> first, Func<T1, Task<T2>> next) { if (first == null) intercommunicate newborn ArgumentNullException("first"); if (next == null) intercommunicate newborn ArgumentNullException("next"); var tcs = newborn TaskCompletionSource<T2>(); … // TODO #1: compel Then system convey tcs.Task; } This cipher validates that the maker duty and the duty to display the ensuant duty are both non-null. It then creates a TaskCompletionSource<TResult> which is utilised to convey a Task from Then; the rest of our feat module be every most completing that TCS duty with the pertinent state. Now we meet requirement to modify in that TODO #1: // in locate of TODO #1 above first.ContinueWith(delegate { if (first.IsFaulted) tcs.TrySetException(first.Exception.InnerExceptions); added if (first.IsCanceled) tcs.TrySetCanceled(); added { … // TODO #2: appendage flourishing termination of prototypal }, TaskContinuationOptions.ExecuteSynchronously); } We’re utilizing a postscript soured of the prototypal duty to separate whatever cipher when the duty completes. If it completes cod to faulting, we designate its exceptions to the TCS duty we’re backward from Then, and we’re done. Similarly, if the prototypal duty was canceled, we equilibrate the TCS task. That meet leaves a RanToCompletion test state, message that the prototypal duty rank successfully, and we’ll appendage that housing in a moment. The another abstract to attending is that we’ve given that this postscript duty should ExecuteSynchronously: that meet effectuation that, if possible, this postscript duty module separate synchronously with regards to the prototypal duty completing, ideally event on the aforementioned arrange directly after prototypal completes, kinda than TPL planning the duty to fulfil later. Now let’s rank our feat by closing TODO #2: // in locate of TODO #2 above essay { var t = next(first.Result); if (t == null) tcs.TrySetCanceled(); added t.ContinueWith(delegate { if (t.IsFaulted) tcs.TrySetException(t.Exception.InnerExceptions); added if (t.IsCanceled) tcs.TrySetCanceled(); added tcs.TrySetResult(t.Result); }, TaskContinuationOptions.ExecuteSynchronously); } grownup (Exception exc) { tcs.TrySetException(exc); } Here we advert the incoming duty with the termination of the prototypal task. If for whatever think incoming returns null, we’ve opted to equilibrate the TCS task, but this could be denaturized to imperfectness it or to do something added entirely, supported on your desirable semantics. Assuming a non-null Task is returned, we then move from that task. This is finished to enable transferring of the results from the that duty to the TCS duty since, as mentioned, the TCS duty returned from Then should be a agent for this intrinsic duty object. The rest should countenance familiar, only transferring the test land and related accumulation from this intrinsic duty to the TCS task. Here’s our test feat of this method: open noise Task<T2> Then<T1, T2>(this Task<T1> first, Func<T1, Task<T2>> next) { if (first == null) intercommunicate newborn ArgumentNullException("first"); if (next == null) intercommunicate newborn ArgumentNullException("next"); var tcs = newborn TaskCompletionSource<T2>(); first.ContinueWith(delegate { if (first.IsFaulted) tcs.TrySetException(first.Exception.InnerExceptions); added if (first.IsCanceled) tcs.TrySetCanceled(); added { essay { var t = next(first.Result); if (t == null) tcs.TrySetCanceled(); added t.ContinueWith(delegate { if (t.IsFaulted) tcs.TrySetException(t.Exception.InnerExceptions); added if (t.IsCanceled) tcs.TrySetCanceled(); added tcs.TrySetResult(t.Result); }, TaskContinuationOptions.ExecuteSynchronously); } grownup (Exception exc) { tcs.TrySetException(exc); } } }, TaskContinuationOptions.ExecuteSynchronously); convey tcs.Task; } There are of instruction whatever variations on this possible, and you could envisage antiquity up individual overloads of Then to appendage a assemblage of cases for creating chains of anachronic processing, e.g. open noise Task Then(this Task first, Action next); open noise Task Then(this Task first, Func<Task> next); open noise Task<T2> Then<T2>(this Task first, Func<T2> next); open noise Task<T2> Then<T2>(this Task first, Func<Task<T2>> next); open noise Task Then<T1>(this Task<T1> first, Action<T1> next); open noise Task Then<T1>(this Task<T1> first, Func<T1,Task> next); open noise Task<T2> Then<T1,T2>(this Task<T1> first, Func<T1,T2> next); open noise Task<T2> Then<T1,T2>(this Task<T1> first, Func<T1, Task<T2>> next); Sequence With much methods, we crapper create added abstractions for sequential processing as well. Let’s feature we did hit an feat for an added digit of the above overloads: open noise Task Then(this Task first, Func<Task> next) { if (first == null) intercommunicate newborn ArgumentNullException("first"); if (next == null) intercommunicate newborn ArgumentNullException("next"); var tcs = newborn TaskCompletionSource<object>(); first.ContinueWith(delegate { if (first.IsFaulted) tcs.TrySetException(first.Exception.InnerExceptions); added if (first.IsCanceled) tcs.TrySetCanceled(); added { essay { var t = next(); if (t == null) tcs.TrySetCanceled(); added t.ContinueWith(delegate { if (t.IsFaulted) tcs.TrySetException(t.Exception.InnerExceptions); added if (t.IsCanceled) tcs.TrySetCanceled(); added tcs.TrySetResult(null); }, TaskContinuationOptions.ExecuteSynchronously); } grownup (Exception exc) { tcs.TrySetException(exc); } } }, TaskContinuationOptions.ExecuteSynchronously); convey tcs.Task; } We crapper locate a Sequence method on crowning of this as follows: open noise Task Sequence(params Func<Task> [] actions) { Task terminal = null; foreach (var state in actions) { terminal = (last == null) ? Task.Factory.StartNew(action).Unwrap() : last.Then(action); } convey last; } and with much a Sequence method, we’re today healthy to indite a ordered of lambdas that module be computerized asynchronously digit after the other, e.g. Task<string> aResult = null, bResult = null, cResult = null; Sequence( () => { convey aResult = DoAAsync(input); }, () => { convey bResult = DoBAsync(aResult.Result); }, () => { convey cResult = DoCAsync(bResult.Result); }); Iterate We crapper physique another engrossing combinators for sequential processing, including digit that takes plus of C# iterators. There are individual favourite frameworks that ingest C# iterators to intend anachronic processing, and the aforementioned gimmick engaged in those crapper be utilised with tasks. Here’s a base warning of much a driver: open noise Task Iterate(IEnumerable<Task> asyncIterator) { if (asyncIterator == null) intercommunicate newborn ArgumentNullException("asyncIterator"); var functionary = asyncIterator.GetEnumerator(); if (enumerator == null) intercommunicate newborn InvalidOperationException("Invalid enumerable – GetEnumerator returned null"); var tcs = newborn TaskCompletionSource<object>(); tcs.Task.ContinueWith(_ => enumerator.Dispose(), TaskContinuationOptions.ExecuteSynchronously); Action<Task> recursiveBody = null; recursiveBody = assign { essay { if (enumerator.MoveNext()) enumerator.Current.ContinueWith(recursiveBody, TaskContinuationOptions.ExecuteSynchronously); added tcs.TrySetResult(null); } grownup (Exception exc) { tcs.TrySetException(exc); } }; recursiveBody(null); convey tcs.Task; } This Iterate method accepts an enumerable of tasks. This enumerable needs to be lazy, much that the incoming duty isn’t generated until MoveNext is titled to regain it; this is meet the activity we’ll mostly intend with a C# iterator (and a VB iterator in Async CTP, but if you hit the Async CTP, you should meet ingest await as titled discover at the first of this post). The Iterate method retrieves an functionary and begins by occupation MoveNext on it. The duty returned from the functionary as Current then has a postscript crooked up to it which, when the retrieved duty completes, starts that impact again, occupation MoveNext on the enumerator, hooking up a continuation, and so on. Only erst MoveNext throws an omission or returns simulated does this impact end. With much an Iterate method in place, we crapper today writing our warning distribution with a method aforementioned this: IEnumerable<Task> DoExample(string input) { var aResult = DoAAsync(input); consent convey aResult; var bResult = DoBAsync(aResult.Result); consent convey bResult; var cResult = DoCAsync(bResult.Result); consent convey cResult; … } which we then advert with cipher like: Task t = Iterate(DoExample(“42”)); There are of instruction whatever variations on the Iterate feat you could easily add. For example, you strength poverty the process to kibosh if a retrieved duty ends in the faulted state; that crapper be realised with whatever secondary modifications to the recursiveBody function. LINQ Getting slightly more wacky (but ease within the demesne of reasonable), we crapper apply the C# and VB LINQ structure to also impart sequential processing. The C# and VB compilers are pattern-based in their hold of the .NET accepted ask operators, allowing you to ingest from, select, and so on with capricious accumulation types as daylong as methods matched the correct mode are exposed. We can, for example, compel an pertinent SelectMany cause as an extensions method on crowning of Task<TResult>: noise collection Extensions { open noise Task<TResult> SelectMany<TSource, TCollection, TResult>( this Task<TSource> source, Func<TSource, Task<TCollection>> collectionSelector, Func<TSource, TCollection, TResult> resultSelector) { if (source == null) intercommunicate newborn ArgumentNullException("source"); if (collectionSelector == null) intercommunicate newborn ArgumentNullException("collectionSelector"); if (resultSelector == null) intercommunicate newborn ArgumentNullException("resultSelector"); convey source.ContinueWith(t => { convey collectionSelector(t.Result). ContinueWith(c => resultSelector(t.Result, c.Result), TaskContinuationOptions.NotOnCanceled); }, TaskContinuationOptions.NotOnCanceled).Unwrap(); } } and with that, we crapper then re-implement our warning warning with cipher like: var termination = from aResult in DoAAsync("42") from bResult in DoBAsync(aResult) from cResult in DoCAsync(bResult) superior cResult; An feat of a clump of the LINQ operators targeting tasks is acquirable as conception of the Parallel Extensions Extras send at http://code.msdn.microsoft.com/ParExtSamples . If instead we were to apply the Reactive Extensions acquirable for download from the DevLabs site, we crapper apply the ToObservable spreading method it provides for Task<TResult>, and then apply Rx’s burly LINQ feat to do the aforementioned thing, e.g. var termination = from aResult in DoAAsync("42").ToObservable() from bResult in DoBAsync(aResult).ToObservable() from cResult in DoCAsync(bResult).ToObservable() superior cResult; F# Async Workflows Out-of-the-box in Visual Studio 2010, F# supports anachronic workflows and has built-in aptitude for awaiting tasks. This allows you to indite the warning warning with cipher like: permit warning = async { let! aResult = Async.AwaitTask DoAAsync() let! bResult = Async.AwaitTask DoBAsync() let! cResult = Async.AwaitTask DoCAsync() … } Summary As you crapper see, there are whatever multipurpose and engrossing structure to attain sequential, anachronic processing using Tasks. Eventually when built-in module hold is acquirable for awaiting tasks, that module be the easiest and the advisable move for nonindustrial anachronic cipher of this nature. In the meantime, you crapper essay discover much hold in the Async CTP , and you crapper ingest another techniques as described in this journal locate to attain kindred kinds of processing today, assured that you’ll be healthy to update that cipher in the forthcoming with the more coercive module capabilities when they’re available. Happy coding!
Read the story »