Pour les tâches parallèles, l'accès cnocurrentiel à des ressources et structures de données partagées est un aspect crucial. On doit souvent verrouiller-déverrouiller des files d'attente et effectuer des opérations mutuellement exclusives comme des insertions ou suppressions. .NET Framework 4.0 offre des conteneurs de données encapsulés pour des opérations parallèles, simplifiant ainsi la programmation concurrente.
Bibliothèque de base
L'espace de noms pour les collections parallèles dans le .NET Framework est *System.Collections.Concurrent*. Voici quelques-uns des conteneurs concurrents disponibles:
- *ConcurrentQueue*
- *ConcurrentStack*
- *ConcurrentDictionary*
- *ConcurrentBag*: une collection non ordonnée utile lorsque l'ordre n'est pas important.
- *BlockingCollection*: similairee aux files d'attente bloquantes classiques.
Ces collections utilisent en partie des techniques sans verrous (*Compare-and-Swap (CAS)* et *barrières mémoire (Memory Barrier)*) pour améliorer les performances par rapport aux verrous mutuels. Toutefois, leur utilisation dans des programmes séquentiels n'est pas recommandée car elles peuvent dégrader les performances.
ConcurrentQueue
Cette file d'attente est entièrement sans verrous, mais peut se retrouver dans une boucle de réessai (spin) si une compétition sur les ressources échoue avec CAS.
- *Enqueue*: ajoute un élément à la fin de la file d'attente.
- *TryDequeue*: tente de supprimer l'élément au début de la file d'attente et le retourne via un paramètre de sortie.
- *TryPeek*: retourne l'élément au début de la file d'attente sans le supprimer.
class Programme
{
private static ConcurrentQueue<int> _fileAttenteTest;
class TravailThread1 // producteur
{
public void Executer()
{
Console.WriteLine("TravailThread1 Exécution { ");
for (int i = 0; i < 30; i++)
{
Console.WriteLine("TravailThread1 producteur: " + i);
_fileAttenteTest.Enqueue(i);
}
Console.WriteLine("TravailThread1 Exécution } ");
}
}
class TravailThread2 // consommateur
{
public void Executer()
{
int valeur = 0;
bool estSupprime = false;
Console.WriteLine("TravailThread2 Exécution { ");
for (;;)
{
estSupprime = _fileAttenteTest.TryDequeue(out valeur);
if (estSupprime)
Console.WriteLine("TravailThread2 consommateur: " + valeur * valeur + " =====");
if (valeur == 29)
break;
}
Console.WriteLine("TravailThread2 Exécution } ");
}
}
static void LancerT1()
{
TravailThread1 travail1 = new TravailThread1();
travail1.Executer();
}
static void LancerT2()
{
TravailThread2 travail2 = new TravailThread2();
travail2.Executer();
}
public static void Main()
{
Task tache1 = new Task(() => LancerT1());
Task tache2 = new Task(() => LancerT2());
_fileAttenteTest = new ConcurrentQueue<int>();
Console.WriteLine("Main {");
Console.WriteLine("Main tache1 tache2 démarrées {");
tache1.Start();
tache2.Start();
Console.WriteLine("Main tache1 tache2 démarrées }");
Console.WriteLine("Main attend tache1 tache2 finies {");
Task.WaitAll(tache1, tache2);
Console.WriteLine("Main attend tache1 tache2 finies }");
Console.WriteLine("Main }");
Console.ReadKey();
}
}
ConcurrentStack
Cette pile est également entièrement sans verrous, mais peut aussi se retrouver dans une boucle de réessai si CAS échoue lors d'une compétition sur les ressources.
- *Push*: ajoute un élément au sommet de la pile.
- *TryPop*: retire un élément du sommet de la pile et le retourne via un paramètre de sortie.
- *TryPeek*: retourne l'élément au sommet de la pile sans le retirer.
ConcurrentBag
C'est une collection non ordonnée où les éléments peuvent être ajoutés ou supprimés.
- *Add*: ajoute un élément à la collection.
- *TryTake*: retire un élément de la collection et le retourne via un paramètre de sortie.
- *TryPeek*: retourne un élément de la collection sans le retirer.
BlockingCollection
C'est un conteneur qui prend en charge les limites et les blocages.
- *Add*: ajoute un élément au conteneur.
- *TryTake*: retire un élément du conteneur et le retourne via un paramètre de sortie.
- *TryPeek*: retourne un élément du conteneur sans le retirer.
- *CompleteAdding*: signale au conteneur que l'ajout d'éléments est terminé. Toute tentative d'ajout ultérieure entraînera une exception.
- *IsCompleted*: indique aux threads consommateurs que le thread producteur continue de s'exécuter et que la tâche n'est pas encore terminée.
Point clé 1: les threads consommateurs utilisent généralement une condition while (!\_TestBCollection.IsCompleted) pour déterminer quand arrêter leur exécution.
Point clé 2: une fois que BlockingCollection a été signalé comme complet (.CompleteAdding()), toute tentative d'ajout supplémentaire entraînera une expection.
class Programme
{
private static BlockingCollection<int> _collectionBloquanteTest;
class TravailThread1 // producteur
{
public void Executer()
{
Console.WriteLine("TravailThread1 Exécution { ");
for (int i = 0; i < 100; i++)
{
Console.WriteLine("TravailThread1 producteur: " + i);
_collectionBloquanteTest.Add(i);
}
_collectionBloquanteTest.CompleteAdding();
Console.WriteLine("TravailThread1 Exécution } ");
}
}
class TravailThread2 // consommateur
{
public void Executer()
{
int valeur = 0;
int compteur = 0;
bool estSupprime = false;
Console.WriteLine("TravailThread2 Exécution { ");
while (!_collectionBloquanteTest.IsCompleted)
{
estSupprime = _collectionBloquanteTest.TryTake(out valeur);
if (estSupprime)
{
Console.WriteLine("TravailThread2 consommateur: " + valeur * valeur + " =====" + valeur);
compteur++;
}
}
Console.WriteLine("TravailThread2 Exécution } ");
}
}
static void LancerT1()
{
TravailThread1 travail1 = new TravailThread1();
travail1.Executer();
}
static void LancerT2()
{
TravailThread2 travail2 = new TravailThread2();
travail2.Executer();
}
static void Main(string[] arguments)
{
Task tache1 = new Task(() => LancerT1());
Task tache2 = new Task(() => LancerT2());
_collectionBloquanteTest = new BlockingCollection<int>();
Console.WriteLine("Main {");
Console.WriteLine("Main tache1 tache2 démarrées {");
tache1.Start();
tache2.Start();
Console.WriteLine("Main tache1 tache2 démarrées }");
Console.WriteLine("Main attend tache1 tache2 finies {");
Task.WaitAll(tache1, tache2);
Console.WriteLine("Main attend tache1 tache2 finies }");
Console.WriteLine("Main }");
Console.ReadKey();
}
}
ConcurrentDictionary
Les opérations de lecture sont entièrement sans verrous, tandis que les modifications par plusieurs threads utilisent des verrous à fine granularité.
- *AddOrUpdate*: ajoute une nouvelle clé/valeur si la clé n'existe pas, sinon met à jour la valeur existante.
- *GetOrAdd*: ajoute une nouvelle clé/valeur si la clé n'existe pas, sinon retourne la valeur existante sans ajouter une nouvelle clé/valeur.
- *TryAdd*: tente d'ajouter une nouvelle clé/valeur.
- *TryGetValue*: tente d'obtenir une valeur selon une clé spécifiée.
- *TryRemove*: tente de supprimer une clé spécifiée.
- *TryUpdate*: met à jour la valeur d'une clé spécifiée de manière conditionnelle.
- *GetEnumerator*: retourne un énumérateur capable de parcourir tout le conteneur.
class Programme
{
private static ConcurrentDictionary<int, int> _dictionnaireTest;
class TravailThread1 // producteur
{
public void Executer()
{
Console.WriteLine("TravailThread1 Exécution { ");
for (int i = 0; i < 100; i++)
{
Console.WriteLine("TravailThread1 producteur: " + i);
_dictionnaireTest.TryAdd(i, i);
}
Console.WriteLine("TravailThread1 Exécution } ");
}
}
class TravailThread2 // consommateur
{
public void Executer()
{
int cle = 0, compteur = 0;
int valeur = 0;
bool estOk = false;
Console.WriteLine("TravailThread2 Exécution { ");
while (compteur < 100)
{
estOk = _dictionnaireTest.TryGetValue(cle, out valeur);
if (estOk)
{
Console.WriteLine("TravailThread2 consommateur: " + cle * cle + " =====" + cle);
valeur = valeur * valeur;
_dictionnaireTest.AddOrUpdate(cle, valeur, (cleActuelle, valeurActuelle) => { return valeurActuelle = valeur; });
compteur++;
cle++;
}
}
Console.WriteLine("TravailThread2 Exécution } ");
}
}
static void LancerT1()
{
TravailThread1 travail1 = new TravailThread1();
travail1.Executer();
}
static void LancerT2()
{
TravailThread2 travail2 = new TravailThread2();
travail2.Executer();
}
static void Main(string[] arguments)
{
Task tache1 = new Task(() => LancerT1());
Task tache2 = new Task(() => LancerT2());
bool estSuivant = true;
int valeur = 0;
_dictionnaireTest = new ConcurrentDictionary<int, int>();
Console.WriteLine("Main {");
Console.WriteLine("Main tache1 tache2 démarrées {");
tache1.Start();
tache2.Start();
Console.WriteLine("Main tache1 tache2 démarrées }");
Console.WriteLine("Main attend tache1 tache2 finies {");
Task.WaitAll(tache1, tache2);
Console.WriteLine("Main attend tache1 tache2 finies }");
foreach (var paire in _dictionnaireTest)
{
Console.WriteLine(paire.Key + " : " + paire.Value);
}
IEnumerator<keyvaluepair int="">
enumerateur = _dictionnaireTest.GetEnumerator();
while (estSuivant)
{
estSuivant = enumerateur.MoveNext();
Console.WriteLine("Clé: " + enumerateur.Current.Key +
" Valeur: " + enumerateur.Current.Valeur);
_dictionnaireTest.TryRemove(enumerateur.Current.Key, out valeur);
}
Console.WriteLine("\n\nNombre d'entrées dans le dictionnaire: " + _dictionnaireTest.Count);
Console.WriteLine("Main }");
Console.ReadKey();
}
}
</keyvaluepair>