Parallel Library Task WaitAny Conception

Question

Je viens de commencer à explorer la PTL et une question de conception.

Mon scénario: J'ai une liste d'URL qui renvoient à une image. Je veux chaque image à télécharger en parallèle. Dès que au moins un image est téléchargée, je veux exécuter une méthode qui fait quelque chose avec l'image téléchargée. Cette méthode ne doit pas être parallélisé -. Il devrait être série

Je pense que ce qui suit fonctionne, mais je ne suis pas sûr que ce soit la bonne façon de le faire. Parce que j'ai des classes séparées pour recueillir les images et pour faire « quelque chose » avec les images recueillies, je finis par passer autour d'un éventail de tâches qui semble mal car il expose le fonctionnement interne de la façon dont les images sont récupérées. Mais je ne sais pas moyen de contourner cela. En réalité, il est plus à ces deux méthodes, mais ce n'est pas important pour cela. Il suffit de savoir qu'ils ne devraient pas être regroupés dans une seule grande méthode que les deux et récupère fait quelque chose avec l'image.

//From the Director class
Task<Image>[] downloadTasks = collector.RetrieveImages(listOfURLs);

for (int i = 0; i < listOfURLs.Count; i++)
{
    //Wait for any of the remaining downloads to complete
    int completedIndex = Task<Image>.WaitAny(downloadTasks);
    Image completedImage = downloadTasks[completedIndex].Result;

    //Now do something with the image (this "something" must happen serially)
    //Uses the "Formatter" class to accomplish this let's say
}

///////////////////////////////////////////////////

//From the Collector class
public Task<Image>[] RetrieveImages(List<string> urls)
{
    Task<Image>[] tasks = new Task<Image>[urls.Count];

    int index = 0;
    foreach (string url in urls)
    {
        string lambdaVar = url;  //Required... Bleh
        tasks[index] = Task<Image>.Factory.StartNew(() =>
            {
                using (WebClient client = new WebClient())
                {
                    //TODO: Replace with live image locations
                    string fileName = String.Format("{0}.png", i);
                    client.DownloadFile(lambdaVar, Path.Combine(Application.StartupPath, fileName));
                }

                return Image.FromFile(Path.Combine(Application.StartupPath, fileName));
            },
            TaskCreationOptions.LongRunning | TaskCreationOptions.AttachedToParent);

        index++;
    }

    return tasks;
}

Answer 1

En général, vous utilisez WaitAny attendre une tâche lorsque vous ne se soucient pas des résultats de l'un des autres. Par exemple, si vous venez soucié la première image qui est arrivé à obtenir de retour.

Qu'en est-ce à la place.

Cela crée deux tâches, l'une qui charge les images et les ajoute à une collection de blocage. La deuxième tâche attend sur la collecte et traite toutes les images ajoutées à la file d'attente. Lorsque toutes les images sont chargées la première tâche ferme la file d'attente vers le bas de sorte que la deuxième tâche peut arrêter.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Drawing;
using System.IO;
using System.Net;
using System.Threading.Tasks;

namespace ClassLibrary1
{
    public class Class1
    {
        readonly string _path = Directory.GetCurrentDirectory();

        public void Demo()
        {
            IList<string> listOfUrls = new List<string>();
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/editicon.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/favorite-star-on.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/arrow_dsc_green.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/editicon.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/favorite-star-on.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/arrow_dsc_green.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/editicon.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/favorite-star-on.gif");
            listOfUrls.Add("http://i3.codeplex.com/Images/v16821/arrow_dsc_green.gif");

            BlockingCollection<Image> images = new BlockingCollection<Image>();

            Parallel.Invoke(
                () =>                   // Task 1: load the images
                {
                    Parallel.For(0, listOfUrls.Count, (i) =>
                        {
                            Image img = RetrieveImages(listOfUrls[i], i);
                            img.Tag = i;
                            images.Add(img);    // Add each image to the queue
                        });
                    images.CompleteAdding();    // Done with images.
                },
                () =>                   // Task 2: Process images serially
                {
                    foreach (var img in images.GetConsumingEnumerable())
                    {
                        string newPath = Path.Combine(_path, String.Format("{0}_rot.png", img.Tag));
                        Console.WriteLine("Rotating image {0}", img.Tag);
                        img.RotateFlip(RotateFlipType.RotateNoneFlipXY);

                        img.Save(newPath);
                    }
                });
        }

        public Image RetrieveImages(string url, int i)
        {
            using (WebClient client = new WebClient())
            {
                string fileName = Path.Combine(_path, String.Format("{0}.png", i));
                Console.WriteLine("Downloading {0}...", url);
                client.DownloadFile(url, Path.Combine(_path, fileName));
                Console.WriteLine("Saving {0} as {1}.", url, fileName);
                return Image.FromFile(Path.Combine(_path, fileName));
            }
        } 
    }
}

AVERTISSEMENT: Le code n'a pas de contrôle d'erreur ou d'annulation. Il est tard et vous avez besoin quelque chose à faire droit? :)

Ceci est un exemple du modèle de pipeline. Il suppose que l'obtention d'une image est assez lent et que le coût de blocage dans la collection de blocage ne va pas causer un problème, car il arrive relativement rarement par rapport au temps passé téléchargement des images.

Notre livre ... Vous pouvez en savoir plus sur à ce sujet et d'autres modèles pour la programmation parallèle http: //parallelpatterns.codeplex .com / Le chapitre 7 et les pipelines couvre les exemples ci-joints montrent des pipelines avec le traitement et l'annulation d'erreur.

Answer 2

TPL fournit déjà la fonction ContinueWith pour exécuter une tâche quand une autre finit. Enchaînement des tâches est l'un des modèles principaux utilisés dans TPL pour les opérations asynchrones.

La méthode suivante télécharge un ensemble d'images et continue en renommant chacun des fichiers

static void DownloadInParallel(string[] urls)
{
   var tempFolder = Path.GetTempPath();

   var downloads = from url in urls
                   select Task.Factory.StartNew<string>(() =>{
                       using (var client = new WebClient())
                       {
                           var uri = new Uri(url);
                           string file = Path.Combine(tempFolder,uri.Segments.Last());
                           client.DownloadFile(uri, file);
                           return file;
                       }
                   },TaskCreationOptions.LongRunning|TaskCreationOptions.AttachedToParent)
                  .ContinueWith(t=>{
                       var filePath = t.Result;
                       File.Move(filePath, filePath + ".test");
                  },TaskContinuationOptions.ExecuteSynchronously);

    var results = downloads.ToArray();
    Task.WaitAll(results);
}

Vous devriez également vérifier les WebClient Async Tâches à partir des échantillons ParallelExtensionsExtras. Les méthodes d'extension DownloadXXXTask offrent à la fois la création de tâches et le téléchargement asynchrone de fichiers.

La méthode suivante utilise l'extension DownloadDataTask pour obtenir les données de l'image et faites-le pivoter avant de l'enregistrer sur le disque

static void DownloadInParallel2(string[] urls)
{
    var tempFolder = Path.GetTempPath();

    var downloads = from url in urls
         let uri=new Uri(url)
         let filePath=Path.Combine(tempFolder,uri.Segments.Last())
         select new WebClient().DownloadDataTask(uri)                                                        
         .ContinueWith(t=>{
            var img = Image.FromStream(new MemoryStream(t.Result));
            img.RotateFlip(RotateFlipType.RotateNoneFlipY);
            img.Save(filePath);
         },TaskContinuationOptions.ExecuteSynchronously);

    var results = downloads.ToArray();
    Task.WaitAll(results);
}

Answer 3

La meilleure façon de le faire serait probablement en mettant en œuvre le modèle d'observateur: ont votre fonction RetreiveImages mettre en œuvre IObservable , mettre votre "terminé l'action d'image" dans un IObserver méthode OnNext de l'objet de, et abonnez-vous à RetreiveImages.

Je ne l'ai pas essayé moi-même encore (reste à jouer plus avec la bibliothèque de tâches) mais je pense que c'est la « bonne » pour le faire.

Answer 4

// télécharger toutes les images

private async void GetAllImages ()
{
    var downloadTasks = listOfURLs.Where(url =>   !string.IsNullOrEmpty(url)).Select(async url =>
            {
                var ret = await RetrieveImage(url);
                return ret;
        }).ToArray();

        var counts = await Task.WhenAll(downloadTasks);
}

//From the Collector class
public async Task<Image> RetrieveImage(string url)
{
    var lambdaVar = url;  //Required... Bleh
    using (WebClient client = new WebClient())
    {
        //TODO: Replace with live image locations
        var fileName = String.Format("{0}.png", i);
        await client.DownloadFile(lambdaVar, Path.Combine(Application.StartupPath, fileName));
    }
    return Image.FromFile(Path.Combine(Application.StartupPath, fileName));
}