Programmation asynchrone dans .NET: meilleures pratiques

Plan

• Problèmes avec les approches résolues avec async / wait.
• Exemples de conception controversée.
• Une tâche de la vie réelle que nous allons résoudre de manière asynchrone.

Async / attente et problèmes à résoudre

Chemin vers async / attente

Rappeler

Void Foo(params, Action callback) {…}
 

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

Plus de rappels

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
        }); 
      });
    });
}

Rappel: exceptions

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


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

Combiner les suites

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

Démarrer et continuer les tâches

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); 
});

Hiérarchie parent-enfant

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); 
}

• la poursuite générale se terminera avec succès; RunToCompletion se produira;
• la tâche sera erronée;
• l'annulation aura lieu;
• la tâche n'atteindra pas son terme du tout, elle sera dans une sorte de vide;
• option - "dépend".

Annulation

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

                                                      Failed

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

                                                      Canceled

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

Pile profonde

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(); 
         }
   } 
}

Complexité cognitive

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

Changer le contexte

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

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

Concurrence structurée

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

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

Comportement séquentiel

async Task MyAsync() {

  var task1 = StartTask1Async();
  await task1;

  var task2 = StartTask2Async();
  await task2; 
}

Comportement simultané

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

  await task1;
  await task2; 
}

Gestion des exceptions

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);
}

En pratique

Méthode synchrone

DataTable<File, ProcessedFile> sharedMemory;

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

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

Méthode asynchrone

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

  File f = blockingQueue.Dequeue(); 

  ProcessedFile p = ProcessInParallel(f);

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

Simplifiez le code parallèle

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);
  } 
}

Solutions squelettes

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


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

  return res; 
}

Async paresseux

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;
}

Débarrassez-vous des verrous

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(); 
}

Sommaire

• Évitez la synchronisation en code asynchrone.
• Le code série est plus simple que parallèle.
• Le code asynchrone peut être simple et utiliser un minimum de paramètres et un contexte implicite qui modifient son comportement.

Que lire

• Document TAP
• Guidage asynchrone
• Durée de vie
• medium.com/@elizarov/deadlocks-in-non-hierarchical-csp-e5910d137cc

-10 . DotNext .