Async programming in .NET: best practices

Plan

• Problems with approaches that are solved with async / await.
• Examples of controversial design.
• A task from real life that we will solve asynchronously.

Async / await and issues to be resolved

Path to async / await

Callback

Void Foo(params, Action callback) {…}
 

Void OurMethod() {
    …//synchronous code
 
    Foo(params,() =>{
       …//asynchronous code;continuation
    });
}

More callbacks

void Foo(params, Action callback) {...} 
void Bar(Action callback) {...}
void Baz(Action callback) {...}

void OurMethod() {
    ... //synchronous code
    
    Foo(params, () => { 
      ... //continuation 1 
      Bar(() => {
        //continuation 2
        Baz(() => {
          //continuation 3
        }); 
      });
    });
}

Callback: exceptions

void Foo(params, Action onSuccess, Action onFailure) {...}


void OurMethod() {
    ... //synchronous code 
    Foo(params, () => {
      ... //asynchronous code on success 
    },
    () => {
        ... //asynchronous code on failure
    }); 
}

Combining continuations

Task ourMethod() {
  return Task.RunSynchronously(() =>{
    ... //synchronous code
  })
  .ContinueWith(_ =>{
    Foo(); //continuation 1
  })
  .ContinueWith(_ =>{
    Bar(); //continuation 2
  })
  .ContinueWith(_ =>{
    Baz(); //continuation 3
  })
}

Start & continue tasks

Task.Factory.StartNew(Action, 
  TaskCreationOptions, 
  TaskScheduler, 
  CancellationToken
)
Task.ContinueWith(Action<Task>, 
  TaskContinuationOptions, 
  TaskScheduler, 
  CancellationToken
)

TaskCompletionSource

var  tcs  =  new   TaskCompletionSource<int>(
       TaskContinuationsOptions.RunContinuationsAsynchronously  
) ;
lock(mylock)
{  
    tcs.SetResult(O); 
});

Parent-child hierarchy

Task.Factory.StartNew(() => 
{
  //... some parent activity

   Task.Factory.StartNew(() => {
      //... some child activity
   })

})
.ContinueWith(...) // don’t wait for child

Task.Factory.StartNew(() => 
{
  //... some parent activity
  Foo(); 

})
.ContinueWith(...) // still wait for child

void Foo() { 
   Task.Factory.StartNew(() => {
      //... parent task to attach is in ThreadStatic
   }, TaskCreationOptions.AttachedToParent); 
}

Task.Factory.StartNew(() => 
{
  //... some parent activity

  Foo();
}, TaskCreationOptions.DenyChildAttach)
.ContinueWith(...) // don’t wait for child

void Foo() { 
   Task.Factory.StartNew(() => {
      //... some child activity
   }, TaskCreationOptions.AttachedToParent); 
}

Task.Run(() => 
{
  //... some parent activity

  Foo(); 

})
.ContinueWith(...) //don’t wait for child

void Foo() { 
   Task.Factory.StartNew(() => {
      //... some child activity
    }, TaskCreationOptions.AttachedToParent); 
}

• general continuation will be completed successfully; RunToCompletion will occur;
• the task will be in error;
• cancellation will occur;
• the task will not reach completion at all, it will be in some kind of limbo;
• option - “depends”.

Cancellation

Task.Factory.StartNew(() => 
{
    throw new OperationCanceledException(); 
});

                                                      Failed

Task.Factory.StartNew(() => 
{
    throw new OperationCanceledException(cancellationToken); 
}, cancellationToken);

                                                      Canceled

Task.Factory.StartNew(() => 
{
    cancellationToken.ThrowIfCancellationRequested(); 
}, cancellationToken);
                                                       Canceled

Deep stack

Task.Factory.StartNew(() => 
{
    Foo();
}, cancellationToken);

  void Foo() { 
     Bar() {
       ...
          Baz() {
             //how to get cancellation token?
          } 
    }
}

static AsyncLocal<Cancelation> asyncLocalCancellation;

Task.Factory.StartNew(() => 
{
     asyncLocalCancellation.Set(cancellationToken) 
    Foo();
}, cancellationToken); // use AsyncLocal to put cancellation int

  void Foo() { 
     async Bar() {
      ...
         Baz() {
             asyncLocalCancellation.Value.CheckForInterrupt(); 
         }
   } 
}

Cognitive complexity

Task.Factory.StartNew(Action, 
    TaskCreationOptions, 
    TaskScheduler, 
    CancellationToken
)
                                                   JetBrains.Lifetimes

lifetime.Start(TaskScheduler, Action) //puts lifetime in AsyncLocal

lifetime.StartMainRead(Action) 
lifetime.StartMainWrite(TaskScheduler, Action) 
lifetime.StartBackgroundRead(TaskScheduler, Action)

async Task MyFuncAsync() { 
  synchronousCode();
   await InnerAsync();
   await Task.Yield(); //guaranteed !IsCompleted 
   continuationCode();
}

async Task MyFuncAsync() { 
  synchronousCode();

    await InnerAsync().ConfigureAwait(false);
    continuationCode(); 
}

async Task MyFuncAsync() { 
  synchronousCode();

    await InnerAsync().ConfigureAwait(continueOnCapturedContext: false); 
    continuationCode(); //code must be absolutely context-agnostic
}

Deadlocks

async Task MyFuncAsync() { //UI thread 
  synchronousCode();

    await Task.Delay(10).ConfigureAwait(continueOnCapturedContext: false); 
    continuationCode();
}
myFuncAsync().Wait() //on UI thread

async Task OnBluttionClick() { //UI thread 
  int v = Button.Text.ParseInt();

    await Task.Delay(10).ConfigureAwait(continueOnCapturedContext: false); 
  Button.Text.Set((v+1).ToString());
}
myFuncAsync().Wait() //on UI thread

async Task MyFuncAsync() { //UI thread 
  synchronousCode();

    await Task.Delay(10).ConfigureAwait(continueOnCapturedContext: false); 
    continuationCode(); //The same UI context
}
PumpUntil(() => task.IsCompleted);
//VS synchronization contexts always pump on any Wait

Change context

async Task MyFuncAsync() { 
  synchronousCode(); // on initial context

    await myTaskScheduler;
    continuationCode(); //on scheduler context 
}

Structured concurrency

async Task MyFuncAsync() { 
  synchronousCode(); // on initial context

    await Task.Factory.StartNew(() => {...}, myTaskScheduler);
    continuationCode(); //on initial context 
}

Sequential behavior

async Task MyAsync() {

  var task1 = StartTask1Async();
  await task1;

  var task2 = StartTask2Async();
  await task2; 
}

Concurrent behavior

async Task MyAsync() {
  var task1 = StartTask1Async();
  var task2 = StartTask2Async();

  await task1;
  await task2; 
}

Exception handling

async Task MyAsync() {
  var task1 = StartTask1Async();
  var task2 = StartTask2Async(); 

  await task1;
  await task2;

  // if task1 throws exception and task2 throws exception we only throw and
  // handle task1’s exception

  //4.0 -> 4.5 framework: unhandled exceptions now don’t crush process
  //still visible in UnobservedExceptionHandler
}

async  Task  MyAsync(CancellationToken cancellationToken)  {  

  await  SomeTask1  Async(cancellationToken); 
 
  await  Some Task2Async( cancellation  Token); 
  //you should always pass use async API with cancelationToken  if possible 
} 
  
try { 
    await  MyAsync( cancellation  Token); 
} catch (OperationException e) { // do nothing: OCE happened
} catch (Exception e) { 
    log.Error(e);
}

In practice

Synchronous method

DataTable<File, ProcessedFile> sharedMemory;

// in any thread
void SynchronousWorker(...) {
  File f = blockingQueue.Dequeue(); 
  ProcessedFile p = ProcessInParallel(f);

  lock (_lock) { 
    sharedMemory.add(f, p);
  } 
}

Asynchronous method

DataTable<File, ProcessedFile> sharedMemory;
// in any thread
async Task WorkerAsync(...) {

  File f = blockingQueue.Dequeue(); 

  ProcessedFile p = ProcessInParallel(f);

  lock (_lock) { 
    sharedMemory.add(f, p);
  } 
}

Simplify Parallel Code

DataTable<File, ProcessedFile> sharedMemory;

// in any thread
async Task WorkerAsync(...) {

  File f = await channel.ReadAsync();

  ProcessedFile p = await ProcessInParallelAsync(f);

  lock (_lock) { 
    sharedMemory.add(f, p);
  } 
}

Skeleton solutions

Dictionary<Action<ProcessedFile>, Action<ProcessedFile>[]> ExecuteBefore; async Task<ProcessedFile> ProcessInParallelAsync() {
  var res = new ProcessedFile();


  // lots of work with toposort, locks, etc.

  return res; 
}

Async lazy

Dictionary<Action<ProcessedFile>, Action<ProcessedFile>[]> ExecuteBefore;
async Task<ProcessedFile> ProcessInParallelAsync() {
  var res = new ProcessedFile();
  var lazy = new Dictionary<Action<ProcessedFile>, Lazy<Task>>(); 
  foreach ((action, beforeList) in ExecuteBefore)
    lazy[action] = new Lazy<Task>(async () => 
    {
      await Task.WhenAll(beforeList.Select(b => lazy[b].Value)) 
      await Task.Yield();
      action(res);
}
  await Task.WhenAll(lazy.Values.Select(l => l.Value)) 
  return res;
}

Get rid of locks

DataTable<File, ProcessedFile> sharedMemory;

// in any thread
async Task WorkerAsync(...) {

  File f = await channel.ReadAsync();

  ProcessedFile p = await ProcessInParallelAsync(f);

    lock (_lock) {
      sharedMemory.add(f, p);
   }
 }

Channel<Pair<File, ProcessedFile>> output;
// in any thread
async Task WorkerAsync(...) {

  File f = await channel.ReadAsync();

  ProcessedFile p = await ProcessInParallelAsync(f);
  
  await output.WriteAsync(); 
}

Summary

• Avoid synchronization in asynchronous code.
• Serial code is simpler than parallel.
• Asynchronous code can be simple and use a minimum of parameters and an implicit context that change its behavior.

What to read

• TAP document
• Asyncs guidance
• Lifetimes
• medium.com/@elizarov/deadlocks-in-non-hierarchical-csp-e5910d137cc

-10 . DotNext .