PLinQ 2

En este nuevo post de PLinQ, nos vamos a centrar en los métodos más importantes de ParalallelEnumerable, decimos los más importantes, ya que nos ofrecen una funcionalidad extra para nuestras consultas parametrizadas.

Cabe destacar, como nombramos en el anterior post de PLinQ, que la clase ParallelEnumerable, tiene una definición para cada uno de los métodos extensores (operadores de consulta) de la clase System.LinQ, para hacer completamente transparente su uso, de modo que pensemos que estamos utilizando una consulta simple a un IEnumerable.

Recuerda que aquí tienes el indice de todos los posts del Curso de LinQ.

WithDegreeOfParallelism

Este método establece el nivel de paralelismo que queremos ofrecer a nuestras consultas parametrizadas, valga la redundancia. En un lenguaje un poco más cercano para todos, con este método, podemos indicar el nº máximo de tareas concurrentes, que podemos emplear en nuestra máquina al procesar la consulta.

Es importante señalar en este punto, que la decisión final del nº de cores/procesadores que se utilizan para realizar el proceso de la consulta, no es responsabilidad de PLinQ, ni siquiera del mismo Framework de .NET, es el Sistema Operativo el que toma esta decisión según la carga de trabajo que tenga la máquina en ese momento. También debemos recalcar, que un core de un procesador, solo puede hacer una tarea/hilo a la vez, es decir al mismo tiempo. Esto no quita que un core pueda hacerse cargo de ‘n’ tareas/hilos en ejecución, alternando la ejecución de cada uno de ellos.

Esta es la firma de su método:

public static ParallelQuery<TSource> WithDegreeOfParallelism<TSource>(this ParallelQuery<TSource> source, int degreeOfParallelism);

Nota:

·         El parámetro entero degreeOfParallelism, tiene que ser un valor numérico entre 1 y 511, en caso contrario se lanzará una excepción de tipo ArgumentOutRangeException.

·         Solo se podrá hacer una llamada a este método en cada consulta. Si se repite la llamada se lanzará una excepción de tipo InvalidOperationException.

Así se pondría en práctica:

var items = Enumerable.Range(1, 80000)
            .AsParallel().WithDegreeOfParallelism(2)
            .Select(a => new MiClase { Numero = a, TotalSum = CalcularTotalSum(a) }).ToList();

WithExecutionMode

Establece el modo de ejecución de una consulta.

Como explicamos en la entrada anterior, antes de realizar una consulta de forma paralela, PLinQ realiza una comprobación mediante el lanzamiento de un algoritmo que realiza una valoración sobre su modo de ejecución. Del resultado de su conclusión, dependerá que se la consulta se ejecute en paralelo o de forma simple. Con WithExecutionMode, obligamos a realizar la consulta según el parámetro introducido.

Su firma es la siguiente:

public static ParallelQuery<TSource> WithExecutionMode<TSource>(this ParallelQuery<TSource> source, ParallelExecutionMode executionMode);

La enumeración ParallelExecutionMode, tiene dos valores:

public enum ParallelExecutionMode
{
    Default          = 0,
    ForceParallelism = 1
}

Default, es el valor por defecto. Con este valor PLinQ, ejecutará el algoritmo de comprobación para ver su modo de aplicación.

Con ForceParallelism, forzaremos a que la consulta se realice en paralelo.

Nota:

·         Solo se podrá hacer una llamada a este método en cada consulta. Si se repite la llamada se lanzará una excepción de tipo InvalidOperationException.

Así se pondría en práctica:

var items = Enumerable.Range(1, 80000)
            .AsParallel().WithExecutionMode(ParallelExecutionMode.ForceParallelism)
            .Select(a => new MiClase { Numero = a, TotalSum = CalcularTotalSum(a) }).ToList();

WithMergeOptions

En el post anterior, pudimos verificar los diferentes pasos que se producen al ejecutar una consulta en paralelo. Durante este proceso, mencionamos que la consulta, era dividida en un número de partes según el número de cores/procesadores disponibles, para realizar este trabajo por cada uno de ellos de forma aislada y en paralelo. Cuando se finalizaba la tardea de todos los grupos, estos tenían que reunir sus resultados y combinarlos para gestar el producto final de la consulta.

El método WithMergeOptions, verifica el modo en que se procesaran en memoria la fusión de estos conjuntos de datos.

Vamos a ver su firma:

public static ParallelQuery<TSource> WithMergeOptions<TSource>(this ParallelQuery<TSource> source, ParallelMergeOptions mergeOptions);

La enumeración ParallelMergeOptions, marca el modo de fusión de los datos. Para entenderlo de una forma sencilla hay 2 aspectos importantes en esto, consumo de memoria vs velocidad. A una opción más rápida un consumo de memoria superior. 

public enum ParallelMergeOptions
{
    Default       = 0,
    NotBuffered   = 1,
    AutoBuffered  = 2,
    FullyBuffered = 3
}

Default, tiene el mismo efecto en la actualidad que AutoBuffered, y no es más que la opción intermedia, velocidad media, consumo de memoria medio.

NotBuffered, es la opción más lenta y la que realiza menos gasto de memoria.

FullyBuffered es la opción contraria a NotBuffered, es la más veloz, pero la que más recursos consume a nivel de memoria.

Nota:

·         Solo se podrá hacer una llamada a este método en cada consulta. Si se repite la llamada se lanzará una excepción de tipo InvalidOperationException.

Así se pondría en práctica:

var items = Enumerable.Range(1, 80000)
            .AsParallel().WithMergeOptions(ParallelMergeOptions.NotBuffered)
            .Select(a => new MiClase { Numero = a, TotalSum = CalcularTotalSum(a) }).ToList();

En la práctica, lo más importante cuando lo que queremos es ganar el mayor rendimiento posible, es hacer pruebas con cada uno de ellos y ver cómo es su desempeño con cada una de las configuraciones.

WithCancellation

WithCancellation, nos da la posibilidad de cancelar nuestras consultas ejecutadas en paralelo. Para ello se precisa pasar por parámetros una referencia a un token de cancelación de tipo CancelationToken para asociarlo con su ejecución.

Esta es su firma

public static ParallelQuery<TSource> WithCancellation<TSource>(this ParallelQuery<TSource> source, CancellationToken cancellationToken);

Nota:

·         Solo se podrá hacer una llamada a este método en cada consulta. Si se repite la llamada se lanzará una excepción de tipo InvalidOperationException.

Para ilustrar el uso de este método, vamos a emplear un ejemplo de una básica aplicación de WPF. No aplicamos una app de consola, ya que su comprensión en este tipo de proyectos es mucho más complicada y no queda tan clara como cuando utilizamos los eventos de la UI.

La aplicación constará de una ventana con dos botones, uno que arrancará un proceso que realiza una consulta en paralelo (evidentemente lanzada de forma asíncrona async/await para que no congele la interfaz) y un segundo botón para cancelar el proceso. El mensaje o título de la pantalla, será utilizado para ir mostrando los diferentes estados de acción.

Esta sería la pantalla en ejecución:

La ejecución realizaría algo tan simple como al apretar el botón Arrancar, comenzaría la ejecución de una consulta en paralelo, para ello en el título de la ventana aparecerá la definición “Ejecutando”.

Esto puede finalizar de 2 formas posible, la primera esperando que finalice el trabajo, por lo que el título de la ventana mostrará “Hecho !!!”:

Y una segunda, pulsando el botón cancelar, mediante el cual, abortaríamos el trabajo y el título de la ventana saldría “Cancelado”.

Vamos a exponer con un pantallazo detallado, la parte más importante del código:

Vamos a aclarar cada uno de los puntos del código:

1.- Creamos un campo a nivel de clase de tipo CancellationTokenSource:

private CancellationTokenSource cts;

2.- Instanciamos el CancellationTokenSource, cada vez que pulsamos el botón ‘arrancar’.

cts = new CancellationTokenSource();

3.- Comenzamos la consulta en paralelo, insertando el CancellationTokenSource como parámetro:

var items = Enumerable.Range(1, 100000)
                            .AsParallel().WithDegreeOfParallelism(2).WithCancellation(cts.Token)
                            .Select(a => new MiClase { Numero = a, TotalSum = CalcularTotalSum(a) }).ToList();

4.- Llamamos al método Cancel, de nuestro CancellationTokenSource, para cancelar la consulta:

cts.Cancel(true);

Dejamos el código de esta mini app, para que podáis jugar con puntos de ruptura y ver su ejecución en directo. Aquí.

ForAll

Cuando queremos recorrer los resultados de una consulta de PLinQ, de forma paralela, no es recomendable, utilizar ninguna de las versiones disponibles de ForEach, ni la normal ni la disponible dentro de la clase Parallel, Parallel.ForEach. Estas están preparadas para enumerar colecciones fijas, y no colecciones en proceso de ejecución.

Por todo esto, el método ForAll, es la forma más óptima de recorrer los resultados dentro de una consulta lanzada en paralelo y se fusiona perfectamente con su resultado.

Enumerable.Range(1, 80000)
            .AsParallel().ForAll(a =>
            {
                /// Hacer Algo !!!!
            });

Bueno pues aquí finaliza PLinQ, ese toque de magia, que solo es mágico es unos escenarios muy determinados, y del que no se puede abusar en ninguno de los casos. Esto es extensible a TPL y a todo el engine de Task y ejecución en paralelo del .NET Framework.