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

PLINQ

Jak wiadomo, kilka tygodni temu wydano kolejną wersję .NET Framework’a a wraz z nią nowe Visual Studio. Zapoznając się z kolejnymi “feature’ami” czwartej już wersji Framework’a, postaram się napisać chociaż kilka zdań o tych najciekawszych.

Na pierwszy “ogień” pójdzie PLINQ. Cóż to zatem jest takiego? Otóż jest to nowa funkcjonalność polegająca na uruchamianiu zapytań LINQ dla obiektów równolegle (PLINQ = Parallel LINQ). Jak wiadomo obecnie producenci procesorów podążają w kierunku upakowania coraz większej ilości rdzeni na pojedynczym chipsecie, więc można przypuszczać, że w/w mechanizm szybko stanie popularny. Zwłaszcza, że użycie PLINQ (w najprostszym przypadku) jest banalne – sprowadza się ono do dopisania przed wywołaniem dowolnej metody “klasycznego” LINQ funkcji AsParallel().

Poniżej przykład:

var query = collection.AsParallel()
                .Select(el => el.ExpensiveMethod());

A teraz kilka słów jak to działa. Mianowicie do przestrzeni nazw System.Linq dodano nową klasę z metodami rozszerzającymi – ParallelEnumerable. Interfejs IEnumerable<T> rozszerza tylko o jedną metodę – wyżej już wspomnianą AsParallel(). Co ważne, metoda ta zwraca obiekt typu ParallelQuery<T>. I właśnie dla tej klasy metody rozszerzające znajdują się w klasie ParallelEnumerable – są tam wszystkie metody znane z “klasycznego” LINQ – jedyna różnica to zwracany obiekt – zamiast IEnumerable<T> mamy ParallelQuery<T>. Dzięki takiemu podejściu zrównoleglanie zapytań jest dziecinnie proste. Przykładowo mając zapytanie postaci:

var sum = collection.Where(el => el.IsCorrect())
                .Select(el => el.ExpensiveMethod())
                .Sum(el => el.Value);

zmiamy je w zapytanie wykonywane równoległe dopisując tylko np. na początku wywołania AsParallel():

var sum = collection.AsParallel()
                .Where(el => el.IsCorrect())
                .Select(el => el.ExpensiveMethod())
                .Sum(el => el.Value);

Często jest jednak tak, że nie całe zapytanie powinno być wykonywane równolegle – pewien fragment (ze względu np. na narzut na utworzenie wątków, itp.) może się wykonywać szybciej szeregowo. Aby zatem zrezygnować z dalszego wykonywania zapytania równolegle – ParallelEnumerable posiada specjalną metodę – AsSequential – po jej wywołaniu dalsza część zapytania będzie wykonywana szeregowo (metoda zwraca po prostu IEnumerable<T>, co powoduje, że dalsza część zapytania LINQ jest wykonywana “starym” sposobem). Przykładowo, mając następujące zapytanie:

var sum = collection.AsParallel()
                .Where(el => el.IsCorrect())
                .Select(el => el.ExpensiveMethod())
                .AsSequential()
                .FirstOrDefault(el => el.Value > 20);

Metody Where i Select zostaną wykonane równolegle, natomiast metoda FirstOrDefault będzie już wykonana sekwencyjnie.

To by było tyle wstępu o PLINQ, następnym razem postaram się opisać bardziej zaawansowane mechanizmy, czyli w jaki sposób sterować równoległym wykonywaniem zapytań.

Pierwszy post

No to zaczynamy. Postaram się (o ile oczywiście pozwoli na to czas) umieszczać na tym blogu  spostrzeżenia, artykuły, a także pewnie i różnorakie “potyczki” związane z technologiami M$, z którymi stykam się w czasie mojej pracy jako programista .NET. Blog ma głównie służyć jako zbiór notatek, do których w każdej chwili można wrócić, jeśli pojawi się jakiś już wcześniej rozwiązany problem, lub jeśli trzeba będzie ponownie skorzystać z dawno już nieużywanej technologii, czy biblioteki. Poza tym jak wiadomo, zapisanie czegoś zazwyczaj powoduje, że zostaje to dłużej w pamięci. 

Czy coś więcej z tego wyniknie, zobaczymy…