PLINQ – sterowanie równoległością

Ostatnio opisywałem podstawowe informacje o PLINQ. Tym razem postaram się pokazać, jakie dodatkowe mechanizmy, oprócz tych znanych z “klasycznego” LINQ daje nam PLINQ.

Execution Modes

Przed uruchomieniem, PLINQ dokładnie analizuje zapytania, które będą wykonywane, aby sprawdzić, czy “opłaca” się “zrównoleglenie” zapytania. Ponieważ biblioteka nie jest w stanie sprawdzić, czy zapytanie szybciej działa równolegle czy szeregowo ani oszacować wielkości przetwarzanych danych, zapytanie sprawdzane jest pod kątem istnienia w nim metod, które z natury są sekwencyjne, takie jak: Skip, Take, TakeWhile, SkipWhile, czy indeksowane wersje Select i Where. Jeśli po sprawdzeniu wydajności wyszło nam, że lepiej jest jednak specyficzne zapytanie uruchomić na kilku wątkach, możemy wymusić użycie PLINQ, wywołując metodę WithExecutionMode z parametrem ParallelExecutionMode.ForceParallelism (jedyna sensowna opcja tego enuma), jak przykładzie poniżej:

var data = collection.AsParallel()
             .WithExecutionMode(
                 ParallelExecutionMode.ForceParallelism
              )
             .SkipWhile(el => ExpensiveCheck(el))
             .Select((el, idx) => Create(el, idx));

Przerywanie wywołania zapytań

Jak można się łatwo domyślić przerywanie wywołania PLINQ nie jest takie proste – przetwarzanie danych odbywa się w kilku wątkach, do których programista nie ma żadnego dostępu. Na szczęście projektanci biblioteki pomyśleli o i tym. Mechanizm stosowany do przerywania zapytań został wprowadzony w wersji 4 .NET Framework’a i jest wykorzystywany w całym przetwarzaniu równległym – nie tylko w bibliotece PLINQ (ale o tym kiedy indziej).
Przechodząc do szczegółów, klasa ParallelEnumerable oferuje metodę rozszerzającą WithCancellation, która jako parametr przyjmuje obiekt typu CancellationToken. Struktura ta służy do informowania o tym, że jakaś operacja ma być przerwana, choć sama nie ma żadnych metod, które pozwalają na anulowanie bieżącego przetwarzania. Do tego celu służy powiązana z nią klasa CancellationTokenSource. Aby zatem móc przerwać jakieś zapytanie PLINQ, musimy najpierw utworzyć instancję typu CancellationTokenSource, a następnie do metody WithCancellation przekazać property Token. Tak utworzone zapytanie możemy w każdej chwili przerwać wywołując metodę Cancel na wcześniej utworzonym obiekcie CancellationTokenSource. Spowoduje to przerwanie działania PLINQ i rzucenie wyjątku typu OperationCancelledException. Jak zwykle, przykład:

private CancellationTokenSource _cts = new CancellationTokenSource();

public List<T> OperationThread()
{
  
  try
  {
    var result = collection.AsParallel()
                   .WithCancellationToken(_cts.Token)
                   .Where(el => el.FulfilsCondition())
                   .Select(el => ExpensiveFunc(el));

    return result.ToList();
  }
  catch (OperationCanceledException e)
  {
    // Query was cancelled
  }
  return null;
}

public void CancelButtonClick()
{
  _cts.Cancel();
}

Kolejność elementów w kolekcji wynikowej

W “klasycznym” LINQ elementy w kolekcji wynikowej są w takiej samej kolejności jak zostały podane na wejściu. Takie podejście w przypadku stosowania “równoległości” wiązałoby się z dużymi kosztami – dlatego też, domyślnie PLINQ nie zachowuje kolejności elementów. Aby wymusić uporządkowanie elementów kolekcji na wyjściu, należy skorzystać z metody AsOrdered. Warto przy okazji wspomnieć, że metodę tę wolno używać tylko po wywołaniu jednej z 3 metod: AsParallel, Repeat oraz Range (w innym wypadku rzucony zostanie odpowiedni wyjątek). Jeśli w danym momencie chcemy przestać korzystać z uporządkowania elementów, wystarczy skorzystać z metody AsUnordered.

Inne możliwości

Biblioteka PLINQ oferuje nam jeszcze kilka innych opcji, o których warto tutaj wspomnieć:

• Ustawienie liczby wątków, na których będzie wykonywane zapytanie: WithDegreeOfParallelism. Domyślnie, liczba wątków odpowiada liczbie procesorów na maszynie.
• Ustalenie sposobu łączenia (merge) elementów w kolekcji wynikowej, czyli w jaki sposób zapytanie będzie buforowało wyniki z poszczególnych wątków – metoda WithMergeOptions przyjmująca jako parametr enumerację ParallelMergeOptions.

Linki

1. PLINQ
2. Order preservation in PLINQ
3. How to: Cancel a PLINQ Query