Diseño de la biblioteca de tareas paralelas Waitany

Question

Acabo de comenzar a explorar el PTL y tener una pregunta de diseño.

Mi escenario: tengo una lista de URL que se refieren a una imagen. Quiero que cada imagen se descargue en paralelo. Tan pronto como al menos uno La imagen se descarga, quiero ejecutar un método que haga algo con la imagen descargada. Ese método no debe ser paralelo, debe ser en serie.

Creo que lo siguiente funcionará, pero no estoy seguro de si esta es la forma correcta de hacerlo. Debido a que tengo clases separadas para recopilar las imágenes y para hacer "algo" con las imágenes recopiladas, termino pasando una variedad de tareas que parece incorrecta ya que expone el funcionamiento interno de cómo se recuperan las imágenes. Pero no sé una forma de evitarlo. En realidad, hay más en ambos métodos, pero eso no es importante para esto. Solo sepa que realmente no deberían ser agrupados en un método grande que recupera y hace algo con la imagen.

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

Por lo general, usa Waitany para esperar una tarea cuando no le importan los resultados de ninguno de los otros. Por ejemplo, si te preocupaste por la primera imagen que se devolvió.

¿Qué tal esto en su lugar?

Esto crea dos tareas, una que carga imágenes y las agrega a una colección de bloqueo. La segunda tarea espera en la colección y procesa cualquier imagen agregada a la cola. Cuando se cargan todas las imágenes, la primera tarea cierra la cola hacia abajo para que la segunda tarea pueda apagarse.

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

ADVERTENCIA: El código no tiene ningún error de comprobación o cancelación. Es tarde y necesitas algo que hacer, ¿verdad? :)

Este es un ejemplo del patrón de tubería. Se supone que obtener una imagen es bastante lento y que el costo de bloquear dentro de la colección de bloqueo no causará un problema porque ocurre con poca frecuencia en comparación con el tiempo dedicado a descargar imágenes.

Nuestro libro ... puedes leer más sobre este y otros patrones para la programación paralela en http://parallelpatterns.codeplex.com/El Capítulo 7 cubre las tuberías y los ejemplos adjuntos muestran tuberías con manejo de errores y cancelación.

Answer 2

TPL ya proporciona la función continua para ejecutar una tarea cuando se termina otra. El encadenamiento de tareas es uno de los principales patrones utilizados en TPL para operaciones asincrónicas.

El siguiente método descarga un conjunto de imágenes y continúa cambiando el nombre de cada uno de los archivos

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

También debes consultar el Tareas de async de Client WebClient de las muestras de parallelextensionSextras. Los métodos de extensión DownloadXXXTask manejan tanto la creación de tareas como la descarga asincrónica de archivos.

El siguiente método utiliza la extensión DownloadDataTask para obtener los datos de la imagen y rotarlos antes de guardarlo en el 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

La mejor manera de hacer esto probablemente sería implementando el patrón de observador: tener su RetreiveImages función implementar Iobservable, ponga su "acción de imagen completa" en un Iobserver objetos OnNext método y suscríbalo a RetreiveImages.

Todavía no lo he probado yo mismo (todavía tengo que jugar más con la biblioteca de tareas) pero creo que esta es la forma "correcta" de hacerlo.

Answer 4

// Descargar todas las imágenes

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