Biblioteca di attività parallele Waitany Design

Question

Ho appena iniziato a esplorare il PTL e avere una domanda di design.

Il mio scenario: ho un elenco di URL che ciascuno si riferisce a un'immagine. Voglio che ogni immagine venga scaricata in parallelo. Non appena almeno una L'immagine viene scaricata, voglio eseguire un metodo che fa qualcosa con l'immagine scaricata. Tale metodo non dovrebbe essere parallelizzato: dovrebbe essere seriale.

Penso che il seguente funzionerà, ma non sono sicuro che questo sia il modo giusto per farlo. Poiché ho classi separate per la raccolta delle immagini e per fare "qualcosa" con le immagini raccolte, finisco per passare una serie di compiti che sembrano sbagliati poiché espone i meccanismi interni di come vengono recuperate le immagini. Ma non conosco un modo per aggirarlo. In realtà c'è di più in entrambi questi metodi, ma questo non è importante per questo. Sappi solo che non dovrebbero davvero essere raggruppati in un grande metodo che recupera e fa qualcosa con l'immagine.

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

In genere usi Waitany per aspettare un'attività quando non ti interessa i risultati di nessuno degli altri. Ad esempio, se ti sei appena preoccupato della prima immagine che si è restituita.

Che ne dici invece.

Questo crea due attività, una che carica le immagini e le aggiunge a una raccolta di blocco. La seconda attività attende la raccolta ed elabora qualsiasi immagine aggiunta alla coda. Quando tutte le immagini vengono caricate, la prima attività chiude la coda in modo che la seconda attività possa chiudere.

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

ATTENZIONE: il codice non ha alcun controllo o annullamento degli errori. È tardi e hai bisogno di qualcosa da fare bene? :)

Questo è un esempio del modello della pipeline. Presume che ottenere un'immagine sia piuttosto lento e che il costo del blocco all'interno della collezione di blocco non causerà un problema perché accade relativamente raramente rispetto al tempo trascorso a scaricare immagini.

Il nostro libro ... puoi leggere di più su questo e altri schemi per la programmazione parallela a http://parallelpatterns.codeplex.com/Il capitolo 7 copre le condutture e gli esempi di accompagnamento mostrano condutture con gestione e cancellazione degli errori.

Answer 2

TPL fornisce già la funzione ContinueWith per eseguire un'attività quando un'altra termina. Il concatenamento delle attività è uno dei modelli principali utilizzati in TPL per le operazioni asincrone.

Il seguente metodo scarica una serie di immagini e continua a rinominare ciascuno dei file

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

Dovresti anche controllare il Compiti di asincroni WebClient Dai campioni ParallelextensionSextras. I metodi di estensione downloadxxxtask gestiscono sia la creazione di attività che il download asincrono dei file.

Il seguente metodo utilizza l'estensione DownloadDataSK per ottenere i dati dell'immagine e ruotarli prima di salvarli sul disco

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

Il modo migliore per farlo sarebbe probabilmente implementando il modello di osservatore: avere il tuo RetreiveImages implementazione della funzione Iobservable, metti la tua "azione immagine completata" in un Iobserver oggetto OnNext metodo e iscriverlo a RetreiveImages.

Non ci ho ancora provato (devo ancora giocare di più con la biblioteca delle attività) ma penso che questo sia il modo "giusto" per farlo.

Answer 4

// Scarica tutte le immagini

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