Una cosa che ho sempre trovato scomoda utilizzando il pattern asincrono Begin/End e il fatto di dover gestire il metodo di callback e pensare a come continuare l’esecuzione del codice una volta finita l’operazione asincrona. Esempio:

…
SqlCommand comm = new SqlCommand(…)  
conn.Open();       
comm.BeginExecuteNonQuery(IAsync =>
{
     int affectedRows = comm.EndExecuteNonQuery(IAsync);
     conn.Close();
     UICode(affectedRows);
}, ...);
…

Questa operazione diventa ancora più scomoda se come nell’esempio il modello di threading in cui callback deve finire ha constraint particolari(non possiamo modificare il codice di UI da un thread diverso dal creatore di quel controllo). Con l’arrivo delle TPL questo non è più un problema e il tutto diventa molto più elegante e chiaro. Ecco come possiamo fare attraverso l’utilizzo della API Task.Factory.FromAsync:

 Task.Factory.FromAsync<int>(command.BeginExecuteNonQuery(),
  (IAsync) => { return command.EndExecuteNonQuery(IAsync); })
      .ContinueWith(t =>
      {
         int affectedRows = t.Result;
      }, TaskScheduler.FromCurrentSynchronizationContext());

Carino no?Con la continuazione il codice diventa molto più facile e se non è ancora chiaro possiamo 
mettere l’API in un extension method e far diventare il codice più semplice:
…
 SqlCommand comm =  new SqlCommand(…)           
conn.Open();
comm.ExecuteAsyncWithResult().ContinueWith(task =>
{
  int id = task.Result;
    conn.Close();
},TaskScheduler.FromCurrentSynchronizationContext());
…

l’extension method:


public static class SqlCommandAsyncExtensionMethods
    {
        public static Task<int> ExecuteAsyncWithResult(this SqlCommand command)
        {
            return Task.Factory.FromAsync<int>(command.BeginExecuteNonQuery(),
                (IAsync) => { return command.EndExecuteNonQuery(IAsync); });
       }
}

Il SqlCommand è un esempio, FromAsync funziona con qualsiasi classe implementi il pattern
IAsync BeginXXX/EndXXX(IAsync)

Con il .Net Framework 4.0 molte modifiche sono state fatte al nuovo ThreadPool(le potete leggere in un mio precendente post), una di quelle più interessanti è quella sull’algoritmo di “thread injection”, ovvero come regolare il numero di thread attivi contemporaneamente dato un insieme di work item(work item di tipo “worker” ovvero di pura computazione/dispatch non di I/O il quale segue regole completamente diverse).
Fino ad oggi l’algoritmo di injection è stato regolato principalmente da due parametri fondamentali, l’utilizzo di CPU e il throughput(work item eseguiti nell’unità di tempo). Per farla semplice semplice diciamo che se il processore è scarico di lavoro, nuovi thread vengono “attivati”, succesivamente ne viene controllato l’effettivo aumento del throughput, in caso positivo nuovi thread possono essere aggiunti altrimenti il “tentativo” di migliorare le prestazioni è fallito e quindi i thread vengono “disattivati”(il bisogno di inserire il throughput come parametro oltre al consumo di CPU è dovuto al fatto che un basso utilizzo di processore non significa che il sistema è “sottoutilizzato”, magari sta semplicemente paginando e quindi inserire più thread peggiora la situazione dato l’aumento del consumo di memoria). Purtroppo questo algoritmo presenta comunque dei limiti dovuti al fatto che il throughput viene influenzato non solo dal numero di thread attivi, ma da molti altri fattori(si pensi ad esempio al tipo di work item, ne esistono di “lenti” e di “veloci”, ma ve ne sono molti altri, ricordiamoci che siamo in un ambiente in cui esistono altri processi altri thread e un sistema operativo preemptive).
Una delle nuove migliorie introdotte a questo algoritmo viene dalla teoria del controllo dei sistemi e si tratta dell’algoritmo di Hill Climbing (HC). Questo algoritmo effettua un “auto tuning” ad intervalli brevi controllando il numero di thread e il throughput, tuttavia questo algoritmo è soggetto ad errori dovuti ai “rumori”(disturbi dell’ambiente in cui il threadPool gira, altri processi,altri thread, sincronizzazione, SO etc…) che rendono il “tuning” di breve termine poco “credibile”. Per togliere i rumori dall’analisi viene utilizzata una tenica conosciuta nella “teoria dei segnali”, dove praticamente un segnale conosciuto viene aggiunto ad un’altro, questo permette di capire come il rumore incide sul segnale di uscita. Nel ThreadPool fate conto che il segnale di ingresso siano il numero thread più il nostro segnale conosciuto e che il segnale di uscita sia il throughput più il nostro segnale introdotto(il tutto “distorto” dal rumore). Queste tecniche vengono chiamate ad esempio “band pass filter” o “match filter” e servono per estrarre onde da altre onde(il ThreadPool utilizza una trasformata di Fourier discreta).

Questo algoritmo non è sicuramente “perfetto”, ma il team sostiene che funziona molto bene per “work item” corti dai 10ms ai 250ms.

A parte la “palla” teorica dovrebbe essere chiaro che scrivere un threadPool non è una cosa tanto facile, nella mia esperienza l’ho sempre usato e sempre lo userò quando avrò bisogno di eseguire un “lavoro” computazionale relativamente corto(se stiamo ad esempio in wait su una coda consumer/producer rubare un thread al threadPool non è il massimo data la natura “lunga” dell’attività, se lo fate con un Task delle TPL ricordatevi di usare il flag TaskCreationOptions.LongRunning che non va ad utilizzare il pool, ma crea un thread ad-hoc). Per concludere vi ricordo che le Task Parallel Library(e di conseguenza tutto quello costruito sopra come ad esempio PLINQ)  si avvalgono del nuovo ThreadPool per eseguire i Task(a parte nel caso di utilizzo del flag TaskCreationOptions.LongRunning come ho già spiegato).

A tutti coloro che sviluppano applicazioni ASP.NET o librerie che possono essere usate in applicazioni di questo tipo, consiglio la lettura di un post molto interessante scritto dal team delle TPL:

http://blogs.msdn.com/pfxteam/archive/2010/02/08/9960003.aspx

il succo del discorso è il fatto che frameworks come asp.net sono già “paralleli” e “scalabili” by-design e che l’introduzione di ulteriore parallelismo potrebbe portare ad un degrado delle performance. Questa affermazione nasce dal fatto che solitamente un’applicazione web è tanto più “perfomante” tante più richieste al secondo riesce a soddisfare. Il runtime di asp.net cerca di sfruttare al massimo le risorse che ha disposizione, come ad esempio utilizzare tutti i core dei nostri server. Chiaramente introdurre ulteriore parallelismo in un ambiente che fa già uso di tutte le risorse di “calcolo” possibili non fa altro che aggiungere latenza agli altri utilizzatori di tale risorse, nel caso di asp.net significa che il runtime dovrà “condividere” i core con altre attività che richiedono la loro parte di potenza di calcolo. Tutto questo significa che il parallelismo non è sempre positivo, ma va usato con “attenzione”, a seconda dell’ambiente in cui si sta operando. Nel caso di ASP.NET l’introduzione di codice parallelo/asincrono ha senso se l’attività che stiamo facendo comprende attività di I/O, in questo modo liberiamo risorse(il thread che ha effettuato la richiesta di I/O; tanto la testina dei dischi o il web service che abbiamo chiamato possono lavorare senza che li dobbiamo per forza aspettare, questo diminuisce il numero di thread richiesti e abbassa il context switch)che possono essere utilizzate per fare altro(soddisfare nuove richieste). Nel caso in cui l’introduzione del parallelismo sia semplicemente dovuta a bisogni computazionali, le performance per una applicazione web ad alto carico(tante richieste al secondo) potrebbero peggiorare vistosamente, dato che i worker thread che si occupano di soddisfare le richieste devono anche essere impiegati per la computazione parallela(questo sottrare risorse per le richieste in ingresso). In questi casi potrebbe risultare più perfomante delegare a qualche altra risorsa come a un server di calcolo dedicato il compito di rendere il calcolo più performante(pagando il prezzo della chiamata out-of-proccess, chiaramente dobbiamo misurare e capire se ne vale la pena). La problematica non resta isolata allo sviluppo di applicazioni web, ma pensiamo allo sviluppo di una libreria che fa uso calcolo parallelo che viene poi utilizzata dalle nostre applicazioni web, il problema si ripresenta. Sarebbe utile avere la possibilità di specificare prima dell’utilizzo della libreria il “grado di parallelismo” richiesto.

Vi lascio alla lettura del post che riporta un bel diagramma di flusso che tutti dovremmo consultare prima di agire!

Fino ad oggi siamo stati abituati a far interagire le nostre applicazioni con l’ambiente in cui queste vivono in maniera “interattiva”, nel senso che è il nostro programma che “interroga” lo stato dell’ambiente e agisce di conseguenza. Per fare un esempio pensiamo all’interfaccia IEnumerable, essa ci restituisce un IEnumerator che ci permette attraverso la chiamata al metodo MoveNext() di ottenere un elemento(nella proprietà Current); in tal senso siamo “interattivi” ovvero siamo noi che chiediamo il prossimo elemento all’enumeratore. Un’altro modo di approcciare è quello di essere “reattivi” ovvero metterci nelle mani dell’ “ambiente”(l’ambiente nel nostro caso potrebbe essere rappresentato da una semplice classe) e lasciare a lui il compito di “attivarci”. Questo modo di lavorare può così essere definito “basato su eventi” e “asincrono”(by design, è l’ambiente che ci attiva e non sappiamo quando). Dalla versione 4.0 del .NET Framework abbiamo a disposizione 2 nuove interfaccie nate proprio allo scopo di permetterci di creare ambienti che siano “reattivi”(per versioni più vecchie di framework le possiamo scaricare insieme all Rx Framework di cui parliamo tra un pò). Le interfaccie in questione sono System.IObservable<> e System.IObserver<>. Vediamole in dettaglio:

public interface IObservable<out T>
{
      IDisposable Subscribe(IObserver<T> observer);
}

public interface IObserver<in T>
{
      void OnCompleted();
      void OnError(Exception error);
      void OnNext(T value);
}

Se le guardiamo con attenzione ci accorgiamo subito che queste due interfacce sono matematicamente il duale di IEnumerable/IEnumerator e ci permettono di avere un funzionamento di tipo “push” invece che il classico “pull” dell’IEnumerable/IEnumerator. Quando dico duale intendo il fatto che:

- per quanto riguarda IObservable/IEnumerable al posto del metodo GetEnumerator() che ritorna IEnumerable abbiamo il metodo Subscrible che accetta un parametro di tipo IObserver

- per quanto riguarda IObserver/IEnumerator non abbiamo più il metodo MoveNext() che ci permette di spostarci sul “prossimo” elemento(che verrà messo in Current), ma abbiamo il metodo OnNext(void) che ci “passa”(quando l’”ambiente” vuole) l’elemento corrente

-nell' IEnumerator il metodo MoveNext() comunica con un booleano se ci sono ancora elementi invece nel caso dell’IObserver abbiamo il metodo  void OnCompleted()

-nel caso di eccezioni nell’ IEnumerator queste vengono sollevate alla chiamata del MoveNext() mentre nell IObserver abbiamo un metodo OnError che ci fornisce il dettaglio dell’errore

Detto questo; i campi di applicazione di questa metodologia basata su eventi asincroni sono innumerevoli e vanno dalla scrittura di “reazioni”(per rendere l’idea) ad eventi di una applicazione WPF, all’utilizzo nel mondo “asicrono” per natura del web e in qualsiasi parte del nostro codice dove non siamo noi che “possiamo” decidere quando “andare avanti”, ma dove è l’ “ambiente” che ce lo comunica.

Parallelamente a questo in casa Microsoft è nato un progetto dai DevLabs chiamato “Reactive Extensions for .NET (Rx)”, con lo scopo di creare degli extension method che permettano(come per IEnumerable) di effettuare delle LINQ query su oggetti di tipo IObservable come possono ad esempio diventare normali eventi .NET o librerie che utilizzano il pattern APM(Asynchronous Programming Design Patterns), ma non solo. In questa libreria abbiamo molti extension method che ci permettono di effettuare “composizioni” anche molto complesse di oggetti IObservable senza doverci preoccupare di problematiche come sincronizzazione, multithreading, coordinamento tra i vari “input streams”(alla fine se devo unire tra loro più IObservable in qualche modo devo sincronizzare i vari OnNext()).

Un’ esempio dell’applicazione di questo pattern potrebbe essere lo sviluppo di una classe che permetta di mettersi in ascolto di messaggi che arrivano in modo “asincrono” da un WebService WCF con un contratto di CallBack, ad esempio con il seguente contratto:

namespace WCFService
{   
    [ServiceContract(CallbackContract=typeof(IEvent))]
    public interface IEventService
    {
        [OperationContract]
        void Subscribe();
        [OperationContract]
        void Unsubscribe();
    }

    [ServiceContract]
    public interface IEvent
    {
        [OperationContract]
        void EventFired(XElement data);
    }
}

Subscribe/Unsubscribe servono per abbonare/disabbonare il client e il contratto di callback ha come argomento un XElement di questa forma:

<Order id="10" customerId="1" xmlns="urn:GestioneOrdini" />

l’xml porta un’ipotetico OrderId e CustomerId.
La natura “event based” del funzionamento del servizio si presta bene a costrure un oggetto di tipo IObservable che permetta lato client di mettersi in ascolto per determinati tipi di ordine, magari solo quelli fatti da un determinato customer(esempio customerId == 1). Lato client il codice “standard” WCF per poter gestire le callback è la costruzione di un oggetto che implementa l’interfaccia di callback  per passarne poi un’instanza in fase di creazione al proxy che effettua le chiamate, nel nostro caso:

class CallbackClass : WCFClientProxy.IEventServiceCallback
{
          public delegate void ServiceFire(object sender, ServiceEvent data);
          public event ServiceFire EventFireEvent;
          public void EventFired(XElement data)
          {
              if (this.EventFireEvent != null)
                  this.EventFireEvent(this, new ServiceEvent { Data = data });
          }
}

nella mia implementazione ho deciso di far sollevare un’evento al momento dell’arrivo di un messaggio dal servizio, l’evento in questione ha un EventArgs del tipo:

public class ServiceEvent : EventArgs
{
    public XElement Data { get; set; }
}

la proprietà “Data” porta chiaramente il payload Xml ricevuto. Grazie all Rx framework e alle nostre 2 nuove interfaccie possiamo permettere a chi utilizza il servizio di scrivere un codice del tipo:

class Program
{
       static void Main(string[] args)
       {
           WCFServiceListener listener = new WCFServiceListener();

          var xs = from ev in listener
                        where ((int)ev.EventArgs.Data.Attribute("customerId")) == 1
                        select ev;

           IDisposable subscription = 
                    xs.Subscribe(e => Console.WriteLine(e.EventArgs.Data));                         

           Console.ReadKey();

           subscription.Dispose();

           Console.ReadKey();
       }
}

WCFServiceListener è una classe che implementa IObservable<T> che al suo interno racchiude la logica di Subscribe/Unsubscribe; la parte più interessante è l’implementazione del metodo Subscribe dell’interfaccia:

public IDisposable Subscribe(IObserver<IEvent<ServiceEvent>> observer)
      {
          if (TraceDispose._subscriblerCount == 0)
                  _proxy.Subscribe();
          return
             new TraceDispose(Observable.FromEvent<ServiceEvent>(
                        _callBackClass,
                        "EventFireEvent")
                      .Subscribe(observer), _proxy);
      }

attraverso un metodo dell’ Rx framework(metodo statico FromEvent della classe Observable) creo un IObservable a partire da un’evento(nel nostro caso EventFireEvent della classe CallbackClass) e da un’instanza dell’oggetto(_callBackClass passata al proxy in fase di creazione) che può sollevare un evento con quel nome. La classe TraceDispose serve solo per la MIA implementazione interna a tenere il conto di quanti si sono “abbonati” al servizio per chiamare il metodo _proxy.Unsubscribe() in caso TraceDispose._subscriblerCount sia uguale a 0.

Tornando al codice del nostro main possiamo vedere come diventa estremamente chiaro e facile capire cosa stiamo facendo:

1)creo il mio IObservable

WCFServiceListener listener = new WCFServiceListener();

2)attraverso l’extension method “Where” dell’Rx filtro i messaggi che arrivano ricevendo solo quelli in cui il customerId è 1.

var xs = from ev in listener
                        where ((int)ev.EventArgs.Data.Attribute("customerId")) == 1
                        select ev; 

3) 
       IDisposable subscription = 
                    xs.Subscribe(e => Console.WriteLine(e.EventArgs.Data));                         

a questo punto mi “abbono” all’IObservable(questa versione di Subscribe fa parte sempre dell’ Rx framework, lo possiamo notare in quanto la firma dell’interfaccia non porta un metodo che accetta delle lambda, è l’overload che crea per noi un IObserver che in caso di chiamata OnNex() chiama il nostro delegate) e passo come delegate in caso di OnNext(quando arriva il messaggio dal server) la stampa in console del messaggio Xml.

Da questo punto in poi la mia applicazione è in ascolto in modo asincrono del servizio e appena arrivano gli ordini lo vediamo stampato in console, ad esempio:

possiamo notare il fatto che customrId è sempre 1 grazie alla nostra query LINQ sull’IObservable. Per disabbonarsi dobbiamo semplicemente chiamare Dispose sull’oggetto che ci viene ritornato in fase di Subscribe(Subscribe ritorna un oggetto IDisposable come da interfaccia) :

subscription.Dispose();

Naturale vero?A me piace un sacco, finalmente una sintassi “intuitiva” per tutte quelle applicazioni che “reagiscono” all’ambiente, supportato da un sacco di possibilità di “composition”(Rx framework ed extension method vari) e senza dover gestire problematiche come concorrenza e sincronizzazione.

Se volete approfondire questa tecnologia vi consiglio di farvi un giro sul portale ufficiale

Reactive Extensions for .NET (Rx)

Rx è utilizzabile con il .NET Framework 3.5 SP1,.NET Framework 4.0 e Silverlight 3

Se volete scaricare la demo del post

attenzione la demo non deve essere usata in linea…non è stata testata e potrebbe presentare problemi di sincronizzazione e di gestione delle eccezioni…ah serve VS2010(Beta2)

L’ IAsyncResult design pattern è un pattern per l’esecuzione asincrona di operazioni che viene largamente usato all’interno del framework in ogni sua parte; consiste nella scrittura di due metodi che permettono di iniziare un’operazione in modo asincrono e di un metodo che la completa, un esempio di una classe che implementa questo pattern potrebbe essere:

    public class AsyncLib
   {
       public int Foo(int parameter)
       {
           ...operazione
       }

       public IAsyncResult BeginFoo(int parameter, AsyncCallback callback, object state)
       {
          ...chiamata a Foo(int parameter) in thread separato e ritorno di un oggeto
          che implementa IAsyncResult
       }

       public int EndFoo(IAsyncResult asyncResult)
       {
           …se l’operazione è finita torno il risultato altrimenti blocco il thread che ha chiamato
       }
   }

il funzionamento è abbastanza semplice: chiamo BeginOperazione che ritorna immediatamente un’istanza di una classe che implementa l’interfaccia IAsyncResult che possiamo interrogare per sapere a che punto è la nostra operazione, o possiamo passare un delegate di tipo AsyncCallBack al metodo stesso per farci chiamare una volta che l’operazione è finita. Quando l’operazione asincrona è terminata(lo sappiamo grazie all’oggetto IAsyncResult o perchè ci viene chiamata la callBack) dobbiamo chiamare il metodo EndOperazione e passare l’istanza ritornata dal metodo BeginOperazione, a quel punto abbiamo in mano il risultato e possiamo farci quello che vogliamo. Questa spiegazione non è completa, ma non è lo scopo del post vi lascio il link alla documentazione ufficiale Asynchronous Programming Design Patterns. Lo stesso pattern viene implementato automaticamente se utilizziamo un delegate, ovvero viene compilata per noi una classe con i metodi Begin/End.

Se dobbiamo far implementare questo pattern alle nostre librerie la cosa più complessa da fare è quella di creare una classe che implementi IAsyncResult nel modo corretto e sincronizzare il tutto con il lavoro asincrono. Le cose da prendere in esame non sono poi così banali, dobbiamo accodare il lavoro da “qualche parte”(dipende da quale tipo di thread vogliamo utilizzare, thread pool, un nostro thread etc…),implementare ciò che l’interfaccia ci richiede(le cui cose più complesse sono come gestire l’oggetto di sincronizzazione e alcuni stati richiesti dall’oggetto in modo perfomante e corretto) e chiamare la callback in caso venga fornita alla fine dell’operazione asincrona. Problema ancora più subdolo è il fatto che la nostra libreria potrebbe essere chiamata in contesti di threading diversi, quindi in caso di callback dobbiamo assicurarci che tutto rispetti le regole di questo contesto. L’implementazione costa un pò di righe di codice che non sono così banali a chiunque.

Il team delle Task Parallel Library ha fortunatamente pensato anche a questo e ha gentilmente fatto implementare alla classe Task l’interfaccia IAsyncResult. Questo fa si che scrivere con .NET 4.0 un custom IAsyncResult diventi un gioco da ragazzi, di seguito un esempio:

  public class AsyncLib
   {
       public int Foo(int millisecond)
       {
           //simulo un carico di lavoro
           Thread.Sleep(millisecond);
           //simulo un risultato
           return new Random().Next(int.MaxValue);
       }

       public IAsyncResult BeginFoo(int i, AsyncCallback callback,
           object state)
       {
          

          Task<int> task = new Task<int>(o => Foo(i), state);

           task.ContinueWith(t =>
           {
               if (callback != null)
                   callback(t);
           },
           SynchronizationContext.Current == null ?
           TaskScheduler.Default :
           TaskScheduler.FromCurrentSynchronizationContext());

           task.Start();

           return task;

         }

       public int EndFoo(IAsyncResult asyncResult)
       {
           Task<int> result = asyncResult as Task<int>;

           if (result == null)
               throw new ArgumentException("Bad async result.");

           return result.Result;
       }
   }

cosa ne dite?Semplicemente la nostra Begin ritorna un task che esegue prima la chiamata a Foo in modo asincrono e poi attraverso ContinueWith esegue la callback(in caso venga passata) nel constesto di threading adeguato
(TaskScheduler.FromCurrentSynchronizationContext() o TaskScheduler.Default). Se contiamo le righe di codice sono 4 per la Begin(sono dovuto andare a capo per farci stare tutto)e 3/4 per la End. Con così poco semplice codice utilizzare questo pattern sarà alla portata di tutti, in quanto l’unica cosa non chiarissima è forse il check del contesto per scegliere il TaskScheduler adeguato, ma siamo in Beta 2 magari qualcosa di meglio lo pensano, per il resto non si vede l’ombra di XXXResetEvent, lock, volatile, ThreadPool, Thread,WaitXXX etc….sono solo 2 Task uno “attaccato” alla fine dell’altro…semplice no?

…ah dimenticavo, chiaramente funziona ovunque, Console app, Windows Form app, WPF app etc…

Se volete una demo contattatemi

Ho già parlato in passato di questo “incubation project” e vi consiglio di guardare questa sessione di Niklas Gustafsson che illustra in modo molto chiaro il motivo per cui è necessario avere un linguaggio “ad hoc” per poter sviluppare in modo corretto/produttivo applicazioni parallele altamente scalabili sulle piattaforme multicore di oggi e domani.

Certamente non tutto il codice che scriviamo deve per forza essere parallelo, una GUI, una logica di business, il codice di accesso ai dati, non hanno sempre questi requisiti, ma vi sono alcune parti di un programma in cui questa possibilità può rendere l’applicazione molto più soddisfaciente.

Axum è un linguaggio che implementa il parallelismo attraverso il paradigma di comunicazione denominato “Message-Passing” dove l’idea principale è quella di avere degli agenti(concettualmente delle classi che fanno qualcosa) che operano su dei dati che definiscono il dominio(ad esempio le classi che contengono i dati su cui fare qualcosa). La chiave di tutto è il fatto che gli agenti possono scambiare tra loro dati solo attraverso messaggi che trasportano dati che possono essere usati in un dominio. Questa constraint da la possibilità di specificare che un agente ha la possibilità di modificare i dati di un dominio, operazione preclusa agli altri agenti che scambiano dati attraverso “copie” e non attraverso riferimenti ai dati stessi.

Gli elementi principali di Axum sono:

-Agenti: gli utilizzatori dei dati del dominio(possono stare in-process o out-of-process, permettendo così di avere un "distributed programming model").

-Dominio: i dati che vengono usati dagli agenti(in modo esclusivo ad esempio)

-Channel: l’interfaccia di comunicazione tra un agente e gli altri agenti

-Schema: definiscono le constraints sui channels e sui messaggi che vengono scambiati dagli agenti

Questo stile di programmazione permette di superare un bel po' di problemi a cui dobbiamo far fronte quando scriviamo un’applicazione che deve “andare in parallelo” ed essere diciamo “asincrona” e “scalabile”. I problemi fondamentali sono quelli dell’isolamento degli stati(evitare lo sharing di parti di memoria attraverso copie e non riferimenti diretti a questa), la sincronizzazione e locking(è un task complesso e pieno di insidie lasciamolo fare ad un runtime) , consumo di risorse(ogni thread in .NET consuma 1MB minimo), scalabilità(sfruttamento di multicore, minimizzare il context switch tra i thread etc…), ma ce ne sono anche altri meno macroscopici.

Il progetto è ancora nei Labs di Microsoft…speriamo non uccidano il bambino nella culla.

Buona visione http://microsoftpdc.com/Sessions/VTL02

P.S.: buon 2010 a tutti.

Se avete “seriamente” bisogno di “performance” nelle vostre applicazioni vi consiglio di guardare questa sessione del pdc 2009 su Windows HPC Server 2008.

Praticamente questa estensione al sistema operativo Windows Server 2008(x64) permette di creare un cluster di server su cui distribuire calcoli. Nella sessione si vede come è possibile semplicemente aggiungendo qualche riga di codice .NET, permettere ad un servizio WCF(attraverso le HPC SOA Api) di essere utilizzato sfruttando il parallelismo del cluster, senza che il client debba sapere l’effettivo endpoint, ma usandone uno solo fornito dall’infrastruttura(il sistema prevede un broker che coordina le chiamate verso il cluster).

Come spiegato nel video questa tecnologia permette impostazioni e possibilità più “fini” confronto le classiche tecnologie di “load balancing” e il supporto a livello di Visual Studio e .NET framework rende il tutto molto appetibile e a basso impatto per uno sviluppatore “medio”.

E’ ancora in beta…ma non vedo l’ora di provarlo…buona visione.

Codice(C#,VS2008,.NET 3.5):

public class LoopInvariant
   {
       int count = 0;
       public void Cicla()
       {
           while (this.count == 0) {}
       }
       public void Cambia()
       {
           this.count = 1;
       }
   }

fino a qua, nulla di particolare, questa classe non fa un bel niente, semplicemente abbiamo 2 metodi, un metodo Cicla() che esegue una while verificando che count sia uguale a 0, poi abbiamo Cambia() un metodo che cambia il contatore e lo porta a 1.

Ora usiamo questa classe(console application):

static void Main(string[] args)
       {
           LoopInvariant var = new LoopInvariant();
           ThreadPool.QueueUserWorkItem(o =>
           {
               System.Threading.Thread.Sleep(5000);
               ((LoopInvariant)o).Cambia();
               Console.WriteLine("Valore modificato");
           }, var);
           var.Cicla();
           Console.WriteLine("Fine test");
           Console.ReadKey();
       }

qua cosa succede; niente di particolare: istanzio la classe LoopInvariant, faccio partire in un thread separato(attraverso il threadpool di .NET) un pezzo di codice che si blocca per 5 secondi(Sleep(5000)) dopo di che chiama il metodo Cambia() sulla mia istanza e stampa che lo ha fatto.

Bene, compiliamo(in release). Lanciamo(CTRL+F5)…passano i 5 secondi…6…7….8….9….?!?!?!?!?

cavolo succede?perchè non stampa “Fine test”?ho cambiato il valore di count…dovrebbe stampare “Fine test” (visto che la condizione count == 0 è falsa) e fermarsi li ad aspettare la pressione di un tasto…

Il problema è che il Jit compiler analizzando il metodo Cicla() applica una ottimizzazione chiamata Loop-invariant code motion, ovvero dal fatto che la variabile count non viene “toccata” all’interno del corpo del while, il compilatore può concludere che count è “loop invariant” e così metterlo in una variabile temporanea prima di iniziare il loop(in un registro del processore magari, così evitiamo di dover risolvere l’indirizzo in ram della variabile ad ogni ciclo migliorando le prestazioni) , questo comporta che quanto l’altro thread chiama Cambia() sulla stessa istanza la modifica non viene “vista” dal nostro metodo Cicla() portanto ad un loop infinito. Cosa che genera ancora più confusione è che se facciamo il debug(F5) di questo codice tutto funziona correttamente in quanto le ottimizzazioni sono disabilitate in configurazione DEBUG.

Per disabilitare questa ottimizzazione occorre dichiarare la nostra variabile count come volatile, questo rende il nostro count esplicitamente non “loop invariant” , come Joe Duffy spiega nel suo libro:

“Load and stores of volatile variables can never be introduced or removed, both in .NET and VC++, because they are assumed to be constantly changing. As such, they aren’t eligible for being considered loop invariant and hoisted outside of loops…”

quindi basta cambiare il nostro codice in questo modo:

public class LoopInvariant
   {
       volatile int count = 0;
       public void Cicla()
       {
           while (this.count == 0) {}
       }
       public void Cambia()
       {
           this.count = 1;
       }
   }

a questo punto se compiliamo(sempre in release altrimenti le ottimizzazioni vengono soppresse per supporto al debug) e lanciamo(CTRL+F5)…1…2….3..4..5…e

adesso il programma funziona come ci aspettavamo…

Attenzione questo non è il solo effetto della keyword volatile, ricordo che leggere da un campo volatile(o usare Thread.VolatileRead) equivale logicamente ad una acquire fence, mentre scrivere un campo volatile(o usare Thread.VolatileWrite) equivale logicamente ad una release fence.

La classe LoopInvariant chiaramente non serve ad un piffero ed è li solo per didattica, ma questo ci fa capire che la programmazione multithread “aggiunge” una dimensione che oggi non è gestita in modo automatico dai compilatori, nel caso della loop invariant fare quella ottimizzazione non cambiava la semantica del programma, ma il danno lo abbiamo visto tutti.

Occhio…

Fonti: Concurrent Programming on Windows

Dopo aver letto questo mio precedente post , dovrebbe essere chiaro come sia possibile che stores/loads possano essere riordinati nel rispetto delle regole del memory model in cui il nostro codice gira.

Se ci pensiamo un attimo, la cosa non dovrebbe toccarci, nel senso che se il compilatore/processore aggiunge/toglie/sposta istruzioni per rendere tutto più performante a noi dovrebbe andare più che bene. Questo è effettivamente vero finchè non facciamo uso di tecniche di programmazione “lock free”, ovvero quando non viene fatto uso di primitive di sincronizzazione(lock(…),Monitor,Events,Mutex) per sincronizzare l’accesso a zone di memoria condivise da più thread. Questa tecnica, rende molto più scalabili e performanti le applicazioni multithread su più processori, in quanto il fatto di non mettere mai il thread in “wait” permette di avere un “throughput” maggiore a confronto di tecniche che utilizzano i lock(nessun thread “blocca” nessun altro thread, se non a basso livello, ma è trasparente a noi e viene fatto dal processore, niente context switch insomma). 

Perchè dovrebbero essere un problema quindi queste ottimizzazioni?
Perchè, mentre muovere istruzioni dal punto di vista di un solo thread quindi di un solo percorso di esecuzione non da nessuna inconsistenza, quando sono più di uno i thread che devono eseguire questo percorso in modo parallelo/contemporaneo il riordinamento di una store/load potrebbe inficiare il corretto funzionamento dell’algoritmo portando a inconsistenze e a difficili bugs da scovare/risolvere.

Per poter avere il controllo sul riordinamento(ottimizzazione) fatto da processore/compilatore, possiamo usare delle istruzioni cosiddette fences/barriers(recinto/barriera), attraverso le quali possiamo indicare al processore/compilatore quali tipi ottimizzazioni(spostamenti) sono possibili.

Ricordiamoci che le ottimizzazioni possono essere fatte a qualsiasi livello dello stack software, quindi sia a livello di compilazione che a livello di esecuzione(processore), mettere delle fence a livello di codice(compilatore) non significa metterle anche a livello di processore, tutto questo dipende dal runtime su cui state sviluppando, es:.NET, VC++, C etc..

Detto questo esistono vari tipi di fence(da ora in poi li chiamerò così e non barrier, ma sono la stessa cosa, dipende dal runtime/ambiente in cui state lavorando):

Full fence: nessuna stores o loads possono essere spostate al di là della fence, in tutte e due le direzioni(molte architetture mettono ad disposizione delle istruzioni come ad esempio MFENCE) quindi(pseudo codice):

load A
load B
store A

Fence

store B
store C
load C

le istruzioni stores/loads ABA e BCC non possono essere spostate al di là della fence in nessuna delle due direzioni, le istruzioni dopo la fence non possono muoversi prima, quelle prima non possono spostarsi dopo. Le full fence sono quelle più “rigide” e tutti gli altri tipi vengono usati per non sacrificare troppo le ottimizzazioni.

Store fence: simili alle full fence eccetto che si applicano solo alle istruzioni di stores e permettono lo spostamento delle loads. Quindi(pseudo codice):

load A
load B
store A

Store fence

store B
store C
load C

Le istruzioni loads A,B,C possono essere spostate prima/dopo della fence(questa fence è disponibile su x86 e x64 e viene generalmente esposta con l’istruzione SFENCE)

Load fence: sono il perfetto contrario delle store fence(e vengono comunemente espresse con l’istruzione LFENCE)

load A
load B
store A

Load fence

store B
store C
load C

Le istruzioni di stores A,B,C possono essere spostate prima/dopo della fence.

Come abbiamo detto prima le fence possono essere utilizzate a livello di processore(full fence, store fence, load fence), ma anche a livello di compilazione:

Acquire fence: assicurano che ne le stores ne le loads che ci sono dopo la fence possano essere spostate prima di questa. Le istruzioni che vengono prima possono essere spostate dopo.

load A
load B
store A

Acquire fence

store B
store C
load C

store B,c e load C non possono essere spostate prima della fence.

Release fence: assicurano che ne le stores ne le loads che ci sono prima della fence possano muoversi dopo della stessa. Le istruzioni dopo la fence possono essere spostate prima

load A
load B
store A

Release fence

store B
store C
load C

load A,B e store A non possono essere spostate dopo la fence.

Queste ultime due fence vengono usate dai complilatori e dall’architettura IA64, e hanno la peculiarità di essere applicate solo alla direzione e non al tipo di istruzione.

Dobbiamo assicurarci di applicare la fence sia a livello di compilatore che a livello di processore, in quanto sono 2 layer separati che possono in modo indipendente effettuare dei riordinamenti(in gergo si dice “move” o meglio “move after..” “move before..”).

Esistono vari modi di introdurre delle fence nel nostro codice e questo varia dalla piattaforma che stiamo usando(.NET,VC++, C), come ad esempio il marcare “volatile” una variabile con .NET o usare delle macro in VC++, ma ce ne sono altri e questo è argomento per un’altro post.

Per concludere è bene sottolineare che il riordinamento che viene effettuato a livello di compilatore/processore, rispetta sempre la “data dependence”(dipendenza dei dati) per non provocare errori logici durante le stores/loads(altro argomento un pò complesso che è bene trattare separatamente).

Fonti: Concurrent programming on Windows(Joe Duffy)

Nel precedente post sui memory model ho parlato spesso di loads e stores.

Si ok, ma cosa vuol dire?

Allora quando parliamo di loads si intendono quelle istruzioni che spostano i dati da una locazione di memoria in un registro del processore, mentre le stores sono quelle istruzioni che spostano i dati da un registro del processore ad una locazione di memoria.

In una architettura load/store, le istruzioni di load e store solo le uniche istruzioni che accedono ai dati in memoria.

Esempio ad alto livello:

x = 1  Store, il valore 1 verrà spostato da un registro del processore all’indirizzo in memoria x
y = x Load, il valore in memoria x verrà spostato in un registro  del processore per essere utilizzato

Così…per chiarezza.

Fonti: Concurrent Programming on Windows( Joe Duffy)

Se vi è per caso capitato di approfondire, per qualche motivo, tematiche legate all’implementazione di soluzioni che fanno uso di codice parallelo/multithread/sincronizzazione probabilmente avete finito leggendo che vi è la possibilità di aumentare le performance usando tecniche di sincronizzazione “lock free”, ovvero senza ricorrere alle primitive di sincronizzazione messe a disposizione dal sistema operativo(che consumano risorse e vanno usate con attenzione).

Molto probabilmente qualche riga sotto l’espressione “lock free” avrete sicuramente trovato che tutto questo è possibile grazie al “Memory consistency model” (anche detto semplicemente memory model) implementato dall’hardware su cui il vostro codice gira.

Ma cos’è il memory model ?Perchè è importante conoscerlo ?

Per spiegarlo facciamo un passo indietro. Quando noi scriviamo codice sorgente ci immaginiamo che l’esecuzione dello stesso sarà così come l’abbiamo scritta. Purtroppo nei moderni processori non è così, nel senso che l’esecuzione del codice macchina potrebbe risultare diversa da quella da noi pensata anche senza inficiare il corretto funzionamento dei nostri algoritmi.

Le motivazioni principali sono:

1)I compilatori ottimizzano (Jit compiler compreso) il codice, spostando (code motion),cancellando, aggiungendo istruzioni per renderne più performante l’esecuzione.
2)I processori impiegano tecniche di “instruction level parallelism” (ILP) permettendo la parallelizzazione delle istruzione così da ridurre i cicli di clock totali.
3)I processori fanno uso di cache locali per velocizzare la scrittura (stores) e lettura (loads) delle locazioni di memoria, in sistemi multiprocessore la sincronizzazione tra le varie cache può portare ad uno spostamento delle istruzioni(cache coherency).

Tutto questo passa sotto il nome di instruction reordering.

Chi scrive codice a basso livello o compilatori o codice lock free, deve conoscere molto bene questi comportamenti per poter effettuare ottimizzazioni e far funzionare come si deve il codice (cioè come noi lo abbiamo scritto o senza provocare race condition).

Per concludere il memory model specifica precisamente quali tipi di scritture (stores) e letture (loads) possono essere spostate (o riordinate), in quali circostanze e come vengono spostate l’una rispetto all’altra.

I memory model si dividono tra weak (deboli) o strong (forti). Un “weak memory model” permette il riordinamento delle letture/scritture, un “strong memory model” (sequential consistency) proibisce tale riordinamento. Molti sistemi come quelli dal quale sto scrivendo questo post (architettura Intel x86) e la maggior parte dei sistemi sul quale Windows gira utilizzano un “weak memory model”.

Fonti: Concurrent Programming on Windows (Joe Duffy)

Segnalo un nuovo interessante progetto pubblicato dai DevLabs chiamato Software Transactional Memory.

Il progetto è nato con l’intento di fornire un modo “semplice” per permettere ad uno sviluppatore di dichiarare che l’esecuzione di una parte di codice debba essere eseguita in modo “atomico” e “transazionale”.

I problemi che questa nuova features risolverebbe non sono pochi e di semplice soluzione per uno sviluppatore che si appresta a scrivere codice parallelo.

Fino ad oggi se noi vogliamo sfruttare il parallelismo nel nostro codice, dobbiamo fare molta attenzione agli “stati condivisi” e all’ ”invarianza” che precluderebbero il buon funzionamento dei nostri algoritmi.

Oggi per fare questo ci possiamo affidare a delle tecniche di locking come ad esempio un bel Monitor, es(il codice è al minimo per chiarezza e non tiene conto di eventuali stati di errore che potrebbero portare a lock orfani, deadlock etc…):

Monitor.Enter(obj);

…//codice atomico

Monitor.Exit(obj);

Il codice non è complesso, ma come ho indicato tra parentesi i problemi che possono nascere sono molti e non così semplici e scontati da anticipare. Inoltre il nostro bel monitor(l’oggetto obj) deve essere trattato con cura onde evitare ulteriori danni(deadlock per esempio).

Come se non bastasse, il secondo problema non è legato al parallelismo di per sè, ma dalla parte di codice che sta tra l’Enter e l’Exit(critical region). In caso il codice al suo interno vada in errore potrebbe succedere che venga meno l’invarianza, cioè che il sucessivo thread che entra in quella regione atomica trovi stati logici inconsistenti precludento il corretto funzionamento del codice, es(il codice non è completo…fate finta che sia correttamente inserito in un try…catch):

Monitor.Enter(obj);

…

Decimal saldoConto = corrente – addebito;

…

//quì ad esempio potrebbe scatenarsi un’eccezzione che invalida le righe di codice sopra

…

Monitor.Exit(obj);

Come potete vedere questo errore è di una gravità mostruosa(sempre che non siate molto ricchi :-D) e solitamente prima di effettuare le operazioni servirebbe del codice che controlla che tutto sia coerente, ma questo non è sempre facile e dipende dalla problematica, che può essere molto complessa con molte casistiche da verificare.

Quello che STM potrebbe risolvere in modo stra-elegante sono proprio queste due problematiche, tutto con un nuovo semplice metodo(con il proprio statement shortcut) e una modifica al funzionamento del Jitter(quindi un framework con delle modifiche interne) es:

…

Atomic.Do(()=> { <statememts> });

…

o

…

atomic { <statememts> }

…

In questo modo non dobbiamo preoccuparti di trattare bene il nostro monitor(il rifermento all’oggetto obj del metodo Enter), ma la cosa più furba è il fatto che un errore nel blocco causa il rollback di tutte le modifiche fatte in quello scope mantenendo così l’invarianza senza dover fare controlli.

Vi lascio con un esempio preso dalla guida che rende tutto molto più chiaro:

class BankAccount
{
        private int m_balance;
        public void ModifyBalance(int amount) {
       atomic {
                       m_balance = m_balance + amount;
                       if (m_balance < 0)
                               throw new OverdraftException(m_balance, amount);
                  }
        }
}

Spettacolo!!!

Spero vivamente che il progetto non vada nel dimenticatoio, sarebbe una manna per tutti noi.

Qui trovate il portale del progetto.