Parallele Task Library Waitany Design

Question

Ich habe gerade angefangen, die PTL zu erkunden und eine Designfrage zu haben.

Mein Szenario: Ich habe eine Liste von URLs, die sich jeweils auf ein Bild beziehen. Ich möchte, dass jedes Bild parallel heruntergeladen wird. Sobald mindestens ein Bild wird heruntergeladen, ich möchte eine Methode ausführen, die etwas mit dem heruntergeladenen Bild macht. Diese Methode sollte nicht parallelisiert werden - es sollte seriell sein.

Ich denke, das Folgende wird funktionieren, aber ich bin mir nicht sicher, ob dies der richtige Weg ist, dies zu tun. Da ich separate Klassen für das Sammeln der Bilder und für das "etwas" mit den gesammelten Bildern habe, fiete ich am Ende eine Reihe von Aufgaben, was falsch zu sein scheint, da es die inneren Funktionsweise des Abholens von Bildern enthüllt. Aber ich kenne keinen Weg daran. In Wirklichkeit hat diese beiden Methoden mehr, aber das ist dafür nicht wichtig. Wisse nur, dass sie wirklich nicht zu einer großen Methode eingebunden werden sollten, die sowohl abrahlt als auch etwas mit dem Bild.

//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

Normalerweise verwenden Sie Waitany, um auf eine Aufgabe zu warten, wenn Sie sich nicht um die Ergebnisse eines der anderen kümmern. Zum Beispiel, wenn Sie sich nur um das erste Bild gekümmert haben, das zufällig zurückgegeben wurde.

Wie wäre es stattdessen.

Dies erzeugt zwei Aufgaben, eine, die Bilder lädt und sie zu einer blockierenden Sammlung hinzufügt. Die zweite Aufgabe wartet auf die Sammlung und verarbeitet alle Bilder, die der Warteschlange hinzugefügt wurden. Wenn alle Bilder geladen werden, schließt die erste Aufgabe die Warteschlange nach unten, damit die zweite Aufgabe heruntergefahren werden kann.

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));
            }
        } 
    }
}

WARNUNG: Der Code hat keine Fehlerprüfung oder Stornierung. Es ist spät und du brauchst etwas, um richtig zu tun? :)

Dies ist ein Beispiel für das Pipeline -Muster. Es wird davon ausgegangen, dass das Erhalten eines Bildes ziemlich langsam ist und dass die Kosten für die Einsperrung innerhalb der Blockierkollektion kein Problem verursachen, da es im Vergleich zu den zeitlichen Download -Bildern relativ selten auftritt.

Unser Buch ... Sie können mehr über dieses und andere Muster für parallele Programmierung lesen http://parallelpatterns.codeplex.com/Kapitel 7 deckt Pipelines ab und die begleitenden Beispiele zeigen Pipelines mit Fehlerbehandlung und Stornierung.

Answer 2

TPL bietet bereits die Funktion fort, eine Aufgabe auszuführen, wenn eine andere beendet ist. Aufgabenkettung ist eines der in TPL für asynchronen Operationen verwendeten Hauptmuster.

Die folgende Methode lädt eine Reihe von Bildern herunter und setzt sich fort, indem Sie die einzelnen Dateien umbenennen

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);
}

Sie sollten auch die überprüfen Webclient Async -Aufgaben Aus den paralleleltesionSextras -Proben. Die DownloadXXXXTask -Erweiterungsmethoden verarbeiten sowohl die Erstellung von Aufgaben als auch das asynchrone Herunterladen von Dateien.

Die folgende Methode verwendet die DownloadDatatask -Erweiterung, um die Daten des Bildes abzurufen und sie zu drehen, bevor Sie sie auf der Festplatte speichern

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

Der beste Weg, dies zu tun RetreiveImages Funktion implementieren Iobservabel, Stecken Sie Ihre "abgeschlossene Bildaktion" in eine Iobserver Objekt OnNext Methode und abonnieren RetreiveImages.

Ich habe das noch nicht selbst ausprobiert (muss noch mehr mit der Aufgabenbibliothek spielen), aber ich denke, dies ist der "richtige" Weg, dies zu tun.

Answer 4

// Alle Bilder herunterladen

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));
}