Task Based Asyncronous Pattern (TAP) – Parte II

Il task-based asyncronous pattern, meglio noto come TAP, è il pattern di programmazione asincrona raccomandato dalla Microsoft per lo sviluppo di applicativi in .NET Framework. Questo articolo è il continuo della prima parte: se le va siete persa correte veloci! In questo articolo ci occuperemo di concetti più avanzati, che permetteranno di sfruttare al meglio tale pattern.

Combinatori basati su task predefiniti

Se non avete barato e davvero avete letto per intero il primo articolo, siete arrivati a questo punto che le basi di TAP sono abbastanza consolidate; siete perciò pronti per scaricare file da linea di comando per impressionare i vostri amici!
Ci sono però ancora alcune cose utili da sapere, infatti il namespace System.Threading.Tasks fornisce alcuni metodi per comporre e per lavorare con i task che tornano spesso comodi.

Task.Run

La classe Task include vari metodi Run che potete usare per prelevare in modo agevole un Task o TaskTask<TResult[]> dal pool dei thread, per esempio:

public async void print()
{
    Console.WriteLine(await Task.Run(() => {
            // il lavoro sporco va qui
            return answer;
        });
    );
}

Task.FromResult

Il metodo FromResult trova la sua utilità in quegli scenari in cui i dati potrebbero già essere disponibili ed è solo necessario ritornarli come parte di un TaskTask<TResult[]>:

ConcurrentDictionary<string, byte[]> cached = new ConcurrentDictionary<string, byte[]>();
public Task<byte[]> GetValueAsync(string address)
{
    byte[] content;
    if (TryGet (address, out content))
        return Task.FromResult(content);
    return Task.Run(async () =>
        {
            content = await new WebClient().DownloadDataTaskAsync(address);
            cached.TryAdd(address, content);
            return content;
        });
}

Task.WhenAll

Quando si parla di sincronizzazione tra processi, dopo i semafori, uno dei primi concetti visti sono le barriere. Questi costrutti servono a bloccare tutti i thread fino a che non arrivano ad un certo punto. Data la natura ad alto livello di C#, le barriere di per sé non esistono, in quanto la complessità di creazione e gestione di thread è nascosta dietro ai task, tuttavia esiste un costrutto simile, ovvero il metodo WhenAll, che resta in attesa su un insieme di thread, finché non sono tutti completati. Questo metodo ha diversi overload, volti a risolvere problemi in cui è necessario gestire situazioni dove si hanno task non generici, o insiemi di task con tipi di ritorno eterogenei. Altri overload overload invece, come quello nell’esempio sottostante, servono per il caso più semplice in cui i tipi di ritorno sono omogenei.

IEnumerable<Task<byte[]>> asyncOps = from url in urls select DownloadDataTaskAsync(url);
try
{
    byte[][] data = await Task.WhenAll(asyncOps);
    ...
}
catch(Exception exc)
{
    foreach(Task<byte[]> faulted in asyncOps.Where(t => t.IsFaulted))
    {
        // gestione di faulted e faulted.Exception
    }
}

Ovviamente l’array data andrebbe prima inizializzato, ma il concetto alla base di WhenAll resta chiaro.
Un particolare degno di nota è la gestione delle eccezioni. Nell’esempio sovrastante, viene gestito singolarmente ogni Task che lancia un’eccezione. Questo ovviamente non è l’unico approccio possibile; un’altra soluzione papabile è infatti quella in cui il ciclo foreach viene eliminato e tutte le eccezioni lanciate vengono aggregate in una singola AggregateException. Sarebbe altresì possibile rimuovere completamente il try-catch; in questo modo l’eccezione verrebbe propagata al blocco in attesa.

Task.WhenAny

WhenAny, differentemente da WhenAll, permette di attendere su un insieme di task, finché almeno uno non ha completato il suo lavoro. Questo metodo trova quattro applicazioni:

• Ridondanza: Eseguire una stessa operazione più volte e selezionare quella che viene completata più velocemente.
• Interleaving: Eseguire una moltitudine di operazioni ed attendere che tutte siano complente, ma processando i risultati non appena sono disponibili.
• Throttling: Permettere un insieme limitato di operazioni di proseguire contemporaneamente, quindi aggiungerne di nuove non appena le vecchie sono complete. Questo scenario può essere interpretato come un’estenzione dell’interlacciamento. Un esempio potrebbe essere il caso in cui si vuole scaricare un numero ingente di immagini, limitando tuttavia il numero massimo di download concorrenti.
• Bailout Anticipato: Un task t1 è raggruppato tramite WhenAny con un task t2. t2 potrebbe essere un timer o una richiesta di annullamento da parte dell’utente. In questo modo se il timer va in time-out o, analogamente, l’utente richiede l’annullamento prima che t1 sia completato, t2 sbloccherà WaitAny.

Per ognuno di questi scenari verrà ora fornito un piccolo codice d’esempio per rendere più chiara l’idea alla base:

Ridondanza

In questo esempio vediamo brevemente come gestire la ridondanza. È bene notare la presenza di un blocco catch che reitera l’operazione nel caso di fallimento di un task (ad esempio per un host non raggiungibile sulla rete).

List<Task<byte[]>> dataTasks = …;  
while(dataTasks.Count > 0)  
{   
    Task<byte[]> dataTask = await Task.WhenAny(dataTask);      
    try  
    {  
        Task<byte[]> dataTask = await Task.WhenAny(dataTasks);  
        byte[] data = await dataTask;  
        // utilizzo dei dati 
        break;  
    }  
    catch(Exception exc)  
    {  
        dataTask.Remove(data);  
    }  
}

Interleaving

L’interleaving è in genere ottenuto ciclando sui task; quando un task completa la sua operazione lo notificherà a WaitAny. A questo punto, il task completato verrà rimosso dalla lista e, mentre una parte del codice si rimette in WaitAny sulla nuova lista, un’altra parte processerà il risultato.

List<Task<byte[]>> dataTasks =   
    (from imageUrl in urls select DownloadDataTaskAsync(url)).ToList();  
while(dataTasks.Count > 0)  
{
    try  
    {  
        Task<byte[]> dataTask = await Task.WhenAny(dataTasks);  
        dataTasks.Remove(dataTask);  

        byte[] data = await dataTask;  
        // utilizzo dei dati 
    }  
    catch (Exception exc) { Log(exc); }
}

Throttling

const int CONCURRENCY_LEVEL = …;  
Uri[] urls = …;  
int nextIndex = 0;  
var dataTasks = new List<Task<byte[]>>();  
while(nextIndex < CONCURRENCY_LEVEL && nextIndex < urls.Length)  
{  
    dataTasks.Add(DownloadDataTaskAsync(urls[nextIndex]));  
    nextIndex++;  
}  

while(dataTasks.Count > 0)  
{  
    try  
    {  
        Task<byte[]> dataTask = await Task.WhenAny(dataTasks);  
        dataTasks.Remove(dataTask);  

        byte[]data = await dataTask;  
        // utilizzo dei dati
    }  
    catch(Exception exc) { Log(exc); }

    if (nextIndex < urls.Length)  
    {  
        dataTasks.Add(DownloadDataTaskAsync(urls[nextIndex]));  
        nextIndex++;  
    }  
}

Bailout Anticipato

Nell’esempio sottostante vediamo uno scenario di bailout anticipato tramite token di annullamento.

public async void Download(string address, CancellationToken ct)  
{  
    try  
    {  
        var client = new WebClient();
        Task<byte[]> dataDownload = client.DownloadDataTaskAsync(address, ct);   
        await UntilCompletionOrCancellation(dataDownload, cts);  
        byte[] data = await dataDownload;  
        // utilizzo dei dati
    } catch(OperationCanceledException) {
        // gestione bailout anticipato
    }
}

private static async Task UntilCompletionOrCancellation(Task asyncOp, CancellationToken ct)  
{  
    var tcs = new TaskCompletionSource<bool>();  
    using(ct.Register(() => tcs.TrySetResult(true))) { 
        await Task.WhenAny(asyncOp, tcs.Task);  
    }
    return asyncOp;  
}

Il codice mostrato a titolo di esempio è suddiviso in due metodi: il primo è l’interfaccia mostrata all’utente che chiede in input l’indirizzo della risorsa da scaricare e un CancellationToken. Chiaramente andrebbe effettuato un controllo per verificare che tale token non sia nullo, ma per brevità e maggiore comprensione del codice, il controllo è stato omesso.
La riga byte[] data = await dataDownload è quella che effettivamente ritornerà il dato se disponibile o lancerà l’eccezione OperationCanceledException se è stato invocato il metodo Cancel() sul token.
Il resto del codice si limita ad avviare il download e chiamare il secondo metodo, che è quello più interessante.
In tale metodo vediamo infatti la creazione di un nuovo task che risulterà sempre non completato, finché TrySetResult non viene invocato. Tale invocazione avverrà però esclusivamente alla cancellazione del token, infatti il metoto Register del CancellationToken registra un delegato che verrà chiamato quanto il token viene annullato.
In questo modo la riga successiva uscirà dal WhenAny se asyncOp completa la sua esecuzione o se il metodo Cancel() viene chiamato (e di conseguenza la lambda definita in Register).

Task.Delay

Il metodo Task.Delay permette di introdurre delle pause nell’esecuzione dei metodi asincroni. Questa funzionalità trova particolare applicazione in casi come cicli di polling o il ritardo nella gestione dell’input dell’utente per un determinate periodo di tempo. Inoltre, il metodo Delay è particolarmente utile in combinazione conWhenAny nel caso in cui si voglia fare bailout anticipato con timer.
Poter inserire un time-out è fondamentale, infatti prendiamo in considerazione un task che è parte di una operazione asincrona lunga e complessa. Se questo task ci dovesse mettere troppo tempo a completarsi, specialmente se dovesse poi fallire, l’intera operazione ne verrebbe pesantemente penalizzata. I metodi sincroni Wait, WaitAll e WaitAny hanno un overload che accetta un time-out come parametro, mentre i corrispettivi asincroni non presentano tale overload. Tuttavia, similmente a quanto visto precedentemente con il bailout anticipato, è posibile usare Delay in congiunzione con WhenAny per implementare il timeout.

Sitografia

Task-based Asyncronous Pattern, Documentazione Microsoft