Potenza e controllocon le Parallel Libraries

Raffaele Rialdi

Twitter: @raffaeler

Email: malta@vevy.com

Articoli e codice: http://www.iamraf.net

Blog: http://blogs.ugidotnet.org/raffaele

Profilo MVP: https://mvp.support.microsoft.com/profile/raffaele

Massima performance possibile• La legge di Amdahls dice che si può ottenere al massimo un guadagno di

– 1/[percentuale spesa nell'esecuzione sequenziale]

• Tutto quindi dipende dalla suddivisione in task– In pratica ogni task è l'unità minima di lavoro sequenziale

• In assenza di I/O il mapping 1 task per ogni core permette usare al massimo le cpu

• Numero ottimale di thread = core / percentuale media di utilizzo per task– Se un task impegna la cpu al 50%– Se ho a disposizione 4 core– Il numero di thread ottimale è 8

Il Parallel Computing firmato Microsoft

Managed

Query PLinq

Data Partitioning

Standard Query Operators

Data Merging

Algoritmi

TPL(Task Parallel

Library)

Strutture dati di coordinazione

Nativo Concurrency Runtime

PPL(Parallel Patterns

Library)

Agents Library

I Task

• Disaccoppia il concetto di esecuzione dal thread– Il Task può essere mappato su Thread o su Fiber– Un Thread può eseguire più Task

• Ogni task è esattamente un'istanza della classe Task• Sono in System.Threading.Tasks (mscorlib)

• I delegate usati da Task– Se non tornano risultati: Action o Action<object>– Se tornano risultati: Func<T> o Func<object, T>

• La proprietà Result fornisce accesso al risultato

• Espone le proprietà ID e Status• Espone i metodi WaitAny, WaitAll

Lo scheduler

• Usano un 'managed user mode scheduler'– Integrato all'interno del ThreadPool standard

• Scalabilità automatica– Lo scheduler a runtime decide il numero di core da

utilizzare– Il developer ha comunque modo di limitarli

• È possibile fornire un custom Scheduler• È possibile avviare un task nel contesto del thread

corrente (sincrono)

Taskusing(ManualResetEvent mre1 = new ManualResetEvent(false))using(ManualResetEvent mre2 = new ManualResetEvent(false))using(ManualResetEvent mre3 = new ManualResetEvent(false)){

ThreadPool.QueueUserWorkItem(delegate { A(); mre1.Set(); } );ThreadPool.QueueUserWorkItem(delegate { B(); mre2.Set(); } );ThreadPool.QueueUserWorkItem(delegate { C(); mre3.Set(); } );

WaitHandle.WaitAll(new WaitHandle[]{mre1, mre2, mre3});}

Task t1 = Task.Factory.StartNew(delegate { A(); });Task t2 = Task.Factory.StartNew(delegate { B(); });Task t3 = Task.Factory.StartNew(delegate { C(); });

Task.WaitAll(t1, t2, t3);

Task t1 = Task.Factory.StartNew(() => A());Task t2 = Task.Factory.StartNew(() => B());Task t3 = Task.Factory.StartNew(() => C());

Task.WaitAll(t1, t2, t3);

==

Continuations

• Permettono di decidere l'ordine temporale di esecuzione di Task e Task<T>– Utile quando una esecuzione necessita il risultato di una

esecuzione precedente– Non viene garantito che il thread di esecuzione sia sempre

lo stesso

var res = Task<int>.Factory.StartNew(() => A(10)).ContinueWith<int>(a => B(a.Result)).ContinueWith<int>(b => C(b.Result));

AggregateException

• L'esecuzione parallela può implicare eccezioni contemporanee• La TPL aggrega le eccezioni in AggregateException

– Questo significa che la migrazione di codice esistente non è trasparente (le catch devono essere rivedute)

– L'esecuzione non continua e riporta solo le eccezioni contemporanee– La collection InnerExceptions contiene le eccezioni con lo Stack Trace

try{

Task.WaitAll(t1, t2);}catch(AggregateException err){

foreach(var e in err.InnerExceptions){

Console.WriteLine("Catched! " + e.Message);}throw err.Flatten();

}

try{

Task.WaitAll(t1, t2, t3);}catch(AggregateException err){

err.Handle(e =>{

Console.WriteLine("Catched {0}: {1}",e.GetType().FullName, e.Message);

return true; // handled});

}

Parallel.For, Parallel.ForEach, Parallel.Invoke

for (int i = 0; i < N; i++){

results[i] = Compute(i);}

Parallel.For(0, N, i =>{

results[i] = Compute(i);} );

foreach(MyClass c in data){

Compute(c);}

Parallel.ForEach(data, c =>{

Compute(c);} );

static void WalkTree<T>(Tree<T> tree, Action<T> func){

if (tree == null) return;WalkTree(tree.Left, func);WalkTree(tree.Right, func);func(tree.Data);

}

static void WalkTree<T>(Tree<T> tree, Action<T> func){

if (tree == null) return;Parallel.Invoke(

() => WalkTree(tree.Left, func) ,() => WalkTree(tree.Right, func) ,() => func(tree.Data) );

}

Parallel.For - break e stop

• ParallelLoopState permette di invocare Break e Stop– Break permette un'interruzione 'soft'

• Le operazioni in esecuzione dal task da cui viene eseguito Break (o i task child) verranno completate

• LowestBreakIteration è il minimo indice al di sotto del quale tutti i risultati sono validi

– LowestBreakIteration è un nullable type

– Stop interrompe totalmente l'esecuzione• A fine loop LowestBreakIteration.HasValue è false

Cancellation

• CancellationTokenSource e CancellationToken servono a informare i Task che devono interrompere le operazioni

• La segnalazione ad un Task parent viene propagata ad i Task Child per tutta la gerarchia

• Il Task deve essere collaborativo e controllare periodicamente la richiesta di annullamento

PLINQPARALLEL LINQ

PLinq

• Permette di eseguire query parallele le query verso oggetti e XML

• NON serve a parallelizzare Linq to SQL o l'entity framework– È comunque possibile eseguire query parallele sui risultati

ottenuti con queste tecnologie

• Il concetto base consiste nel "partizionare" porzioni di una query– Ogni thread calcola un pezzo e lo aggiunge al risultato

PLinq

• ParallelEnumerable è la classe che contiene gli extension methods con le implementazioni parallele di Linq– Count, First, Last, Distinct, Join, Average, Aggregate, …

Operatori specifici di PLinq• AsParallel

– Parallelizzare il resto della query, se possibile

• AsSequential– Il resto della query deve essere eseguita in modo sequenziale invece di parallelo

• AsOrdered, AsUnordered– La query deve rispettare / non rispettare l'ordine della sequenza sorgente

• WithCancellation– Si richiede di monitorare il flag di annullamento

• WithDegreeOfParallelism– Numero massimo di Core da usare nella query

• WithMergeOptions– Suggerimenti su come eseguire il merge dei risultati paralleli parziali

• WithExecutionMode– Per forzare l'esecuzione parallela anche quando verrebbe eseguita sequenzialmente dal runtime

• ForAll– Permette di fruire in modo parallelo il risultato senza necessità di eseguire il merge sul thread che ha avviato la

query

• Aggregate– Permette l'aggregazione parziale dei risultati mantenuti in partizioni specifiche al TLS

Esecuzione parallela

• L'uso di AsParallel non implica che l'esecuzione avvenga in modo parallelo

• PLinq analizza la query e decide di parallelizzarla solo se la ritiene "safe"

• WithExecutionMode e WithDegreeOfParallelism sono molto utili per testare il guadagno al crescere dei core– In produzione è preferibile lasciare a PLinq la decisione del

numero di core da usare

Query ordering in PLinq

• Le query parallele sono eseguite a pezzi in thread diversi– Ogni volta l'ordine del risultato può essere differente

Th

rea

d 3

Th

rea

d 2

Th

rea

d 1

Th

rea

d 4

3Risultato 2 1 4

linea temporale t

Query ordering in PLinq• Possibilmente non fare affidamento sull'ordine del

risultato– I database fanno la stessa cosa

• Se è richiesto un risultato ordinato– Se possibile, ordinarlo dopo la query e non prima

– Diversamente specificare la clausola AsOrdered

– Oppure usare la clausola OrderBy

Array.Sort(a);var q = from x in a.AsParallel() ...

var q = from x in a.AsParallel() ...Array.Sort(q);

Array.Sort(a);var q = from x in a.AsParallel().AsOrdered() ...

Array.Sort(a);var q = from x in a.AsParallel().OrderBy(...) ...

Cancellation

• Si usa WithCancellation• CancellationToken e CancellationTokenSource

funzionano in modo identico a quanto già visto nei Task

COLLECTION PARALLELE

Collection in PFXSystem.Collections.Concurrent

• PFX offre diverse collection di dati tra cui:– ConcurrentBag<T>

• Lista non ordinata di oggetti

– ConcurrentDictionary<TKey, TValue>• Dizionario

– BlockingCollection<T>• Ottimizzata per gli scenari Producer-Consumer• Implementa IEnumerable<T>, ICollection<T> e IDisposable

– ConcurrentQueue<T> (analoga a Queue<T>)• Una lista FIFO concorrente• bool TryDequeue(out T result) e bool TryPeek(out T result)

– ConcurrentStack<T> (analoga a Stack<T>)• Una list LIFO concorrente• bool TryPop(out T result) e bool TryPeek(out T result)

• In particolare le Queue sono fondamentali per smistare dei lavori su un numero fisso di worker thread

Classi 'Slim'

• Sono classi 'leggere' cioè che non fanno uso di risorse native, con grande beneficio in termini di performance– ManualResetSlim– SemaphoreSlim– CountDownEvent (usa ManualResetSlim internamente)

• In sostanza non devono eseguire la transizione User‒Mode Kernel‒Mode

TIPS & TRICKS

Task lunghi o corti?

• TaskCreationOptions.LongRunning– usato per dire allo scheduler di incrementare il numero di

thread (es: i task sono impegnati in I/O)

• Se un task dura più di 500ms entra in gioco il meccanismo di "starvation detection"– è una strategia per evitare potenziali deadlock

• Difendersi dallo starvation:– Task più piccoli possibile– Custom scheduler– ThreadPool.SetMaxThreads(Environment.ProcessorCount)

Parallelizzare: problemi e soluzioni

• Per la gran parte dei casi rendere parallelo è problematico (the free lunch ...)

• Problema 1: lo stato condiviso tra più thread è un freno alla scalabilità e frutto di errori da incubo

• Problema 2: molti algoritmi sono intrinsecamente seriali e la loro versione parallela (se esiste) spesso è radicalmente differente

• Problema 3: i tradizionali sistemi di unit testing sono totalmente inefficaci perché le modalità di esecuzione dipendono dall'hardware e da altri fattori

PARALLEL PATTERNS

Una linea guida

• Scomporre i problemi in piccoli task– La dimensione conta!– Se ogni 'task' è troppo piccolo, l'overhead di gestione del task

finisce per penalizzare la sua esecuzione

• Coordinare l'esecuzione dei task– Strettamente dipendente da come funziona l'algoritmo

• Condividere i dati necessari all'esecuzione– Minore dipendenza da dati condivisi implica maggiori

performance– La condivisione implica l'accesso in modo serializzato alla risorse

condivise– Alcune tecniche minimizzano le implicazioni di performance

Parallel Loop Pattern

• È applicabile quando ogni step del loop non richiede i dati precedenti

• Strumenti:– Parallel.For– Parallel.ForEach– Parallel (PLinq)

Parallel Task Pattern

• Si possono isolare delle unità di lavoro distinte e con dipendenze ben definite?

• Le dipendenze sono limitate?• Se si verificano queste condizioni è possibile risolvere

creando tanti task quante sono le unità di lavoro

• Strumenti:– Task, Task<…>

Parallel Aggregation Pattern

• Esiste qualche forma di aggregazione sul risultato dei dati?

• Ogni step dell'elaborazione produce una parte del risultato?

• Se si verificano queste condizioni si usano gli strumenti di aggregazione di PLinq

• Strumenti:– PLinq

• (from x in sequence.AsParallel() select f(x) ).Sum();

Futures Pattern / Task Graph Pattern

• L'ordine di esecuzione di ogni step di elaborazione dipende dal flusso dei dati?

• Si può immaginare il progresso dell'elaborazione con un grafo dipendente dai dati?

• Se si verificano queste condizioni, è opportuno usare le Continuations

• Strumenti:– Task e Continuations

Dynamic Task Parallelism Pattern

• La suddivisione del problema in step è dinamica?• L'algoritmo prevede l'uso di strutture ricorsive

durante l'elaborazione (es: grafi)?

• Se si tratta di percorrere un grafo, la soluzione si può trovare in Parallel.Invoke e ForEach

• Strumenti:– Parallel.*

Domande ?