{"id":493,"date":"2012-08-01T16:19:19","date_gmt":"2012-08-01T14:19:19","guid":{"rendered":"http:\/\/blog.khaledtannir.net\/?p=493"},"modified":"2012-08-01T16:19:19","modified_gmt":"2012-08-01T14:19:19","slug":"implementer-mapreduce-avec-tpl-task-parallel-library","status":"publish","type":"post","link":"https:\/\/khaledtannir.net\/en\/2012\/08\/01\/implementer-mapreduce-avec-tpl-task-parallel-library\/","title":{"rendered":"Impl\u00e9menter MapReduce avec TPL (Task Parallel Library)"},"content":{"rendered":"

Dans cet article je vous propose une impl\u00e9mentation de MapReduce avec TPL (Task Parallel Library<\/em>) du framework .NET 4.0. Si vous ne connaissez pas MapReduce, vous avez toujours la possibilit\u00e9 de consulter mon article (partie 1<\/a> et partie 2<\/a>) d\u00e9di\u00e9 \u00e0 sa pr\u00e9sentation et \u00e0 son mode op\u00e9ratoire.
\n<\/p>\n

Introduction :<\/span><\/h2>\n

J’ai voulu dans cet article proposer une impl\u00e9mentation possible du mod\u00e8le de traitement parall\u00e8le MapReduce. Bien s\u00fbr il ne s’agit pas d’impl\u00e9menter toutes les fonctionnalit\u00e9s du mod\u00e8le cr\u00e9\u00e9 par Google mais plut\u00f4t la fonction Map qui effectue des traitements en parall\u00e8le et la fonction Reduce qui, quant \u00e0 elle, agr\u00e8ge les r\u00e9sultats avant de nous les retourner.
\nConcevoir une application qui g\u00e8re et synchronise plusieurs dizaines de threads reste une t\u00e2che difficile. Faire de la programmation parall\u00e8le est aussi une t\u00e2che difficile et les API de d\u00e9veloppement sont g\u00e9n\u00e9ralement complexes et demandent des efforts d\u2019architecture\/programmation\/d\u00e9bug afin de rendre le tout utilisable.
\nPartant de ce principe j’ai fait le choix d’utiliser\u00a0TPL (Task Parallel Library<\/em>) du framework .NET 4.0. Cette librairie regroupe des classes permettant\u00a0de parall\u00e9liser des traitements.\u00a0Ces classes permettant une ex\u00e9cution des traitements en parall\u00e8le,\u00a0g\u00e8rent \u00e9galement la pr\u00e9sence de coeurs multiples de la machine ex\u00e9cutant le code\u00a0de l’application.
\nTPL\u00a0est compos\u00e9 de 8 classes et de 3 namespaces principaux (.NET 4) :<\/span><\/p>\n

    \n
  1. System.Threading.Tasks<\/span> : Ce namespace contient la base de Task Parallel Library (TPL), il contient la classe principale de la technologie : la classe Tasks (une sorte de \u201csuper Thread\u201d)<\/li>\n
  2. System.Collections.Concurrent<\/span> : Ce namespace contient les collections pouvant \u00eatre utilis\u00e9 dans des contextes multithreads (thread safe).<\/li>\n
  3. System.Linq<\/span> : LINQ a \u00e9t\u00e9 am\u00e9lior\u00e9 pour int\u00e9grer des aides au multithreading. Connu sous le nom de PLINQ (Parallel LINQ), il permet de parall\u00e9liser certaines \u00e9tapes des requ\u00eates LINQ afin de les rendre plus rapide.<\/li>\n<\/ol>\n

    Le namespace System.Threading.Task<\/span>\u00a0contient une classe Task<\/span><\/em>. Cette classe repr\u00e9sente une op\u00e9ration asynchrone. Ainsi on d\u00e9fini notre op\u00e9ration (t\u00e2che) et, lorsqu\u2019on la d\u00e9marre, celle-ci est assign\u00e9 par le TaskScheduler<\/span> <\/em>\u00e0 un des coeurs disponibles de la machine. Le fonctionnement de ce syst\u00e8me est bas\u00e9 sur\u00a0une\u00a0 file de type FIFO (first in \/ first out) prenant en compte un niveau de\u00a0priorit\u00e9. Le Taskscheduler<\/em> int\u00e8gre un m\u00e9canisme de r\u00e9partition de charge et, est capable de distribuer un grand nombre de t\u00e2ches aux diff\u00e9rentes c\u0153urs et de “balancer” la charge si un c\u0153ur est plus occup\u00e9 qu\u2019un autre. L’utilisation de TPL reste simple,\u00a0 la cr\u00e9ation et l\u2019utilisation de la classe Task<\/span> <\/em>est \u00e9galement simple.
    \nL’impl\u00e9mentation que je propose se base sur la classe Task<\/span> <\/em>et utilise ses m\u00e9canismes de synchronisations pour\u00a0synchroniser les diff\u00e9rentes op\u00e9rations. Pour sa mise en oeuvre l’application, tr\u00e8s simplissime, TPLMapReduce\u00a0\u00a0<\/em>impl\u00e9mente une classe qui contient les m\u00e9thodes n\u00e9cessaires \u00e0 l\u2019ex\u00e9cution du code.<\/p>\n

    L’application TPLMapReduce :<\/span><\/h2>\n

    Pour illustrer le fonctionnement de MapReduce, l’application TPLMapReduce aura comme objectif de compter les mots trouv\u00e9s dans 4 documents de type texte. Deux impl\u00e9mentations sont propos\u00e9es pour effectuer cette t\u00e2che ; la premi\u00e8re impl\u00e9mentation consiste \u00e0 cr\u00e9er plusieurs op\u00e9rations de type Map puis agr\u00e9ger le tous avec une seule op\u00e9ration de type Reduce. La deuxi\u00e8me impl\u00e9mentation consiste \u00e0 attribuer \u00e0 chaque docuement un ensemble de t\u00e2ches Map puis une fonction Reduce d\u00e9di\u00e9e afin d\u2019agr\u00e9ger les donn\u00e9es de celui-ci. Au final le r\u00e9sultat global est le m\u00eame sauf que je voulais garder les portes de l’imagination et de la cr\u00e9ativit\u00e9 ouvertes.
    \nPour commencer, la classe MapReduce est une classe absrtaite et ne peut pas \u00eatre instanci\u00e9e directement, ceci dans le but de permettre son extension ind\u00e9pendamment des classes qui pourront l’utiliser. Elle impl\u00e9mente deux propri\u00e9t\u00e9s Map et Reduce. Ces propri\u00e9t\u00e9s d\u00e9finissent les fonctions Map et Reduce respectivement. Ces fonctions seront d\u00e9finies dans les classes filles de la classe MapReduce.
    \nLa fonction Map acc\u00e8pte un type de donn\u00e9es en entr\u00e9e (par exemple string) et retourne un type de donn\u00e9es interm\u00e9diaire qui peut \u00eatre du m\u00eame type que les donn\u00e9es en entr\u00e9es ou un type diff\u00e9rent. Ceci est la signature de notre fonction Map.<\/p>\n

    \/\/\/ <summary>\n\/\/\/ Obtient ou d\u00e9finit la fonction Map\n\/\/\/ <\/summary>\npublic virtual Func<TInput, TPartialResult> Map { get; protected set; }<\/code><\/pre>\n
    La fonction Reduce quant \u00e0 elle doit prendre en entr\u00e9e l’ensemble des r\u00e9sultats interm\u00e9diaires des t\u00e2ches Map et nous retourner\u00a0 un r\u00e9sultat. Pour cela, sa signature est la suivante :<\/p>\n
    \/\/\/ <summary>\n\/\/\/ Obtient ou d\u00e9finit la fonction Reduce\n\/\/\/ <\/summary>\npublic virtual Func<TPartialResult[], TOutput> Reduce { get; protected set; }<\/code><\/pre>\n
    Remarquez que la modification des deux\u00a0propri\u00e9t\u00e9s Map et Reduce ont \u00e9t\u00e9 restrintes\u00a0aux classes filles uniquement.<\/p>\n
    Les op\u00e9rations Map :<\/span><\/h3>\n
    Les cr\u00e9ation des t\u00e2ches Map est de la responsabilit\u00e9 de la m\u00e9thode CreateMapTasks<\/em>. Cette m\u00e9thode accepte un tableau contenant les donn\u00e9s en entr\u00e9es auxquelles le traitement doit \u00eatre appliqu\u00e9.<\/p>\n
    \/\/\/ <summary>\n\/\/\/ Cr\u00e9\u00e9 les diff\u00e9rentes taches Map\n\/\/\/ <\/summary>\n\/\/\/ <param name=\"map\">La fonction Map<\/param>\n\/\/\/ <param name=\"inputs\">Tableau des donn\u00e9es en entr\u00e9e du type TInput<\/param>\n\/\/\/ <returns>Retourne un tableau contenant le r\u00e9sultats interm\u00e9diaires du traitement<\/returns>\nprotected virtual Task<TPartialResult>[] CreateMapTasks(TInput[] inputs)\n{\n   var tasks = new Task<TPartialResult>[inputs.Length];\n       for (int i = 0; i < inputs.Length; ++i)\n       {\n           tasks[i] = Task.Factory.StartNew(() => Map(inputs[i]));\n       }\n   return tasks;\n}<\/code><\/pre>\n
    Le code de cette m\u00e9thode est relativement simple. Ici je parcours le tableau des donn\u00e9es en entr\u00e9es et pour chacune des\u00a0lignes du\u00a0tableau je cr\u00e9\u00e9 une nouvelle instance de la classe Task\u00a0<\/span><\/em>en appelant la m\u00e9thode Factory<\/span> <\/em>de cette classe.\u00a0 Une op\u00e9ration de type Task<\/span> <\/em>est un\u00a0conteneur pour une op\u00e9ration qui sera ex\u00e9cut\u00e9e de fa\u00e7on asynchrone. Il suffit donc de cr\u00e9er une instance de la classe Task<\/span><\/em>, de lui assigner un traitement \u00e0 r\u00e9aliser et de l’ex\u00e9cuter. Le runtime .Net d\u00e9termine automatiquement combien d’instances Task<\/span> <\/em>(ou t\u00e2ches) il est possible d’ex\u00e9cuter simultan\u00e9ment et ceci en fonction du nombre de coeurs disponibles, puis\u00a0les g\u00e8re au travers du ThreadPool<\/span> <\/em>qui est le \u00ab scheduler \u00bb par d\u00e9faut de le la TPL. Il est possible d’ex\u00e9cuter les Tasks dans un ordre donn\u00e9, d’attendre la fin d’une Task pour en ex\u00e9cuter une autre, une Task peut retourner ou non un r\u00e9sultat, etc.
    \nDans l’extrait de\u00a0code ci-dessus, j’ai fait appel \u00e0 la m\u00e9thode StartNew()<\/span> <\/em>afin de d\u00e9marrer imm\u00e9diatement la t\u00e2che sans \u00eatre oblig\u00e9 par la suite de boucler sur l’ensemble des t\u00e2ches et d\u2019appeler la m\u00e9thode Start()<\/span> <\/em>sur chacune des t\u00e2ches.
    \nA la fin du traitement de chacune des t\u00e2ches, la propri\u00e9t\u00e9 Result<\/span> <\/em>contiendra le r\u00e9sultat du traitement appliqu\u00e9 et son\u00a0type sera celui\u00a0d\u00e9finit par la fonction Map.<\/p>\n
    Les op\u00e9rations Reduce :<\/span><\/h3>\n
    Les op\u00e9rations Reduce quant \u00e0 elles doivent \u00eatre ex\u00e9cut\u00e9es une fois toutes les op\u00e9rations Map ont termin\u00e9 leur travail. Pour assurer cette synchronisation, j’ai utilis\u00e9 la m\u00e9thode ContinueWhenAll() de la Classe Factory imbriqu\u00e9e \u00e0 la classe Task. En effet cette m\u00e9thode va attendre que la totalit\u00e9 des taches du tableau mapTasks en entr\u00e9e\u00a0aient termin\u00e9 leurs traitements avant d’appeler l’op\u00e9ration Reduce.<\/p>\n
    \/\/\/ <summary>\n\/\/\/ Cr\u00e9\u00e9 les diff\u00e9rentes taches Reduce\n\/\/\/ <\/summary>\n\/\/\/ <param name=\"mapTasks\">Tableau des taches Map ayant \u00e9t\u00e9 ex\u00e9cut\u00e9es<\/param>\n\/\/\/ <returns>Retourne le r\u00e9sultat de l'application de la fonction Reduce typ\u00e9 selon TOutput<\/returns>\nprotected virtual Task<TOutput> CreateReduceTask(Task<TPartialResult>[] mapTasks)\n{\n    return Task.Factory.ContinueWhenAll(mapTasks, tasks => PerformReduce(tasks));\n}<\/code><\/pre>\n
    <\/h3>\n
    Exemple d’utilisation :<\/span><\/h3>\n
    Les fichiers textes (format UTF-8) utilis\u00e9s dans cet exemple sont disponibles sur le site du Projet Gutenberg<\/a>.
    \nAfin de compter tous les mots qui apparaissent dans ces documents, ceux-ci n\u00e9cessitent une petite phase de pr\u00e9traitement en vue de leur pr\u00e9paration \u00e0 cette t\u00e2che. Pour cela, j’ai choisi de les lire en parall\u00e8le puis de les d\u00e9couper en lignes. A la fin de ce pr\u00e9traitement, chaque ligne du tableau contiendra une ligne extraite du fichier texte. Bien s\u00fbr qu’il s’agit d’un traitement sommaire pour les besoins de l’illustration et non d’un traitement en profondeur.<\/p>\n
    \/\/ D\u00e9finition des fichiers dont les mots devront \u00eatre compt\u00e9s\n            string[] files = {\n                                \"pg1041.txt\",\n                                \"pg1105.txt\",\n                                \"pg1745.txt\",\n                                \"pg25979.txt\"\n                             };\n            char[] delimiters = { ' ', ',', ';', '.' };\n            string[][] inputs = new string[files.Length][];\n            \/\/ Lecture en parall\u00e8le des fichiers, chaque fichiers est d\u00e9coup\u00e9 en un tableau de lignes\n            for (int i = 0; i < files.Length; i++)\n            {\n                inputs[i] = File.ReadLines(files[i]).AsParallel().SelectMany(line => line.Split(delimiters)).ToArray();\n            }<\/code><\/pre>\n
    La classe Exemple 1<\/span> h\u00e9rite de la classe MapReduce et permet de la concr\u00e9tiser avec la d\u00e9finition des deux fonctions Map et Reduce. L’impl\u00e9mentation de la classe MapReduce implique la d\u00e9finition de trois types g\u00e9n\u00e9riques :<\/p>\n