Wprowadzenie do PLINQ

Artykuł
03/05/2024

Parallel LINQ (PLINQ) to równoległa implementacja wzorca zapytania zintegrowanego z językiem (LINQ ). PLINQ implementuje pełny zestaw standardowych operatorów zapytań LINQ jako metod rozszerzenia dla System.Linq przestrzeni nazw i ma dodatkowe operatory dla operacji równoległych. PLINQ łączy prostotę i czytelność składni LINQ z mocą programowania równoległego.

Napiwek

Jeśli nie znasz linQ, zawiera ujednolicony model do wykonywania zapytań dotyczących dowolnego wyliczalnego źródła danych w bezpieczny sposób. LINQ to Objects to nazwa zapytań LINQ, które są uruchamiane względem kolekcji w pamięci, takich jak List<T> i tablice. W tym artykule założono, że masz podstawową wiedzę na temat LINQ. Aby uzyskać więcej informacji, zobacz Zapytanie zintegrowane ze zintegrowanym językiem (LINQ).

Co to jest zapytanie równoległe?

Zapytanie PLINQ na wiele sposobów przypomina równoległe zapytanie LINQ to Objects. Zapytania PLINQ, podobnie jak sekwencyjne zapytania LINQ, działają w dowolnej pamięci IEnumerable lub IEnumerable<T> źródle danych i mają odroczone wykonywanie, co oznacza, że nie rozpoczynają wykonywania, dopóki zapytanie nie zostanie wyliczone. Podstawową różnicą jest to, że PLINQ próbuje w pełni wykorzystać wszystkie procesory w systemie. Robi to przez podzielenie źródła danych na segmenty, a następnie wykonanie zapytania w poszczególnych segmentach w osobnych wątkach roboczych równolegle na wielu procesorach. W wielu przypadkach wykonywanie równoległe oznacza, że zapytanie działa znacznie szybciej.

Dzięki wykonaniu równoległym plINQ może osiągnąć znaczną poprawę wydajności w porównaniu ze starszym kodem dla niektórych rodzajów zapytań, często tylko przez dodanie AsParallel operacji zapytania do źródła danych. Jednak równoległość może wprowadzać własne złożoności, a nie wszystkie operacje zapytań działają szybciej w PLINQ. W rzeczywistości równoległe spowolnienie niektórych zapytań. W związku z tym należy zrozumieć, w jaki sposób problemy, takie jak porządkowanie, wpływają na zapytania równoległe. Aby uzyskać więcej informacji, zobacz Understanding Speedup in PLINQ (Opis szybkości w PLINQ).

Uwaga

Ta dokumentacja używa wyrażeń lambda do definiowania delegatów w PLINQ. Jeśli nie znasz wyrażeń lambda w języku C# lub Visual Basic, zobacz Wyrażenia lambda w plINQ i TPL.

W pozostałej części tego artykułu omówiono główne klasy PLINQ i omówiono sposób tworzenia zapytań PLINQ. Każda sekcja zawiera linki do bardziej szczegółowych informacji i przykładów kodu.

Klasa ParallelEnumerable

Klasa System.Linq.ParallelEnumerable uwidacznia prawie wszystkie funkcje PLINQ. System.Linq Pozostałe typy przestrzeni nazw są kompilowane w zestawie System.Core.dll. Domyślne projekty języka C# i Visual Basic w programie Visual Studio odwołują się do zestawu i importują przestrzeń nazw.

ParallelEnumerable Zawiera implementacje wszystkich standardowych operatorów zapytań, które obsługuje LINQ to Objects, chociaż nie próbuje zrównać każdego z nich. Jeśli nie znasz linQ, zobacz Wprowadzenie do LINQ (C#) i Wprowadzenie do LINQ (Visual Basic).

Oprócz standardowych operatorów ParallelEnumerable zapytań klasa zawiera zestaw metod, które umożliwiają zachowanie specyficzne dla wykonywania równoległego. Te metody specyficzne dla PLINQ są wymienione w poniższej tabeli.

ParallelEnumerable Operator	opis
AsParallel	Punkt wejścia dla PLINQ. Określa, że reszta zapytania powinna być zrównana, jeśli jest to możliwe.
AsSequential	Określa, że reszta zapytania powinna być uruchamiana sekwencyjnie, jako zapytanie LINQ, które nie jest równoległe.
AsOrdered	Określa, że PLINQ powinna zachować kolejność sekwencji źródłowej dla pozostałej części zapytania lub dopóki kolejność nie zostanie zmieniona, na przykład przez użycie klauzuli orderby (Order By w Visual Basic).
AsUnordered	Określa, że PLINQ dla pozostałej części zapytania nie jest wymagane do zachowania kolejności sekwencji źródłowej.
WithCancellation	Określa, że PLINQ powinien okresowo monitorować stan podanego tokenu anulowania i anulować wykonywanie, jeśli jest to wymagane.
WithDegreeOfParallelism	Określa maksymalną liczbę procesorów, które PLINQ powinny być używane do równoległości zapytania.
WithMergeOptions	Zawiera wskazówkę dotyczącą sposobu, w jaki plINQ powinien, jeśli jest to możliwe, scal równoległe wyniki z powrotem do tylko jednej sekwencji w wątku zużywających.
WithExecutionMode	Określa, czy PLINQ ma być równoległe zapytanie nawet wtedy, gdy domyślne zachowanie będzie uruchamiać je sekwencyjnie.
ForAll	Wielowątkowa metoda wyliczania, która w przeciwieństwie do iteracji wyników zapytania, umożliwia przetwarzanie wyników równolegle bez uprzedniego scalania z powrotem do wątku odbiorcy.
Aggregate Przeciążenie	Przeciążenie, które jest unikatowe dla PLINQ i umożliwia agregację pośrednią w partycjach lokalnych wątków oraz ostateczną funkcję agregacji w celu połączenia wyników wszystkich partycji.

Model zgody

Podczas pisania zapytania wybierz opcję PLINQ, wywołując ParallelEnumerable.AsParallel metodę rozszerzenia w źródle danych, jak pokazano w poniższym przykładzie.

var source = Enumerable.Range(1, 10000);

// Opt in to PLINQ with AsParallel.
var evenNums = from num in source.AsParallel()
               where num % 2 == 0
               select num;
Console.WriteLine("{0} even numbers out of {1} total",
                  evenNums.Count(), source.Count());
// The example displays the following output:
//       5000 even numbers out of 10000 total

Dim source = Enumerable.Range(1, 10000)

' Opt in to PLINQ with AsParallel
Dim evenNums = From num In source.AsParallel()
               Where num Mod 2 = 0
               Select num
Console.WriteLine("{0} even numbers out of {1} total",
                  evenNums.Count(), source.Count())
' The example displays the following output:
'       5000 even numbers out of 10000 total

Metoda AsParallel rozszerzenia wiąże kolejne operatory zapytań, w tym przypadku where i select, z System.Linq.ParallelEnumerable implementacjami.

Tryby wykonywania

Domyślnie PLINQ jest konserwatywny. W czasie wykonywania infrastruktura PLINQ analizuje ogólną strukturę zapytania. Jeśli zapytanie może przyspieszyć przez równoległość, plINQ dzieli sekwencję źródłową na zadania, które mogą być uruchamiane współbieżnie. Jeśli nie można bezpiecznie zrównoleglić zapytania, plINQ po prostu uruchamia zapytanie sekwencyjnie. Jeśli plINQ ma wybór między potencjalnie kosztownym algorytmem równoległym lub niedrogim algorytmem sekwencyjnym, domyślnie wybiera algorytm sekwencyjny. Możesz użyć WithExecutionMode metody i System.Linq.ParallelExecutionMode wyliczenia, aby poinstruować PLINQ, aby wybrać algorytm równoległy. Jest to przydatne w przypadku testowania i pomiaru, że określone zapytanie wykonuje się szybciej równolegle. Aby uzyskać więcej informacji, zobacz How to: Specify the Execution Mode in PLINQ (Instrukcje: określanie trybu wykonywania w plINQ).

Stopień równoległości

Domyślnie PLINQ używa wszystkich procesorów na komputerze hosta. Można poinstruować PLINQ, aby używać nie więcej niż określonej liczby procesorów przy użyciu WithDegreeOfParallelism metody . Jest to przydatne, gdy chcesz upewnić się, że inne procesy uruchomione na komputerze otrzymują określony czas procesora CPU. Poniższy fragment kodu ogranicza zapytanie do użycia maksymalnie dwóch procesorów.

var query = from item in source.AsParallel().WithDegreeOfParallelism(2)
            where Compute(item) > 42
            select item;

Dim query = From item In source.AsParallel().WithDegreeOfParallelism(2)
            Where Compute(item) > 42
            Select item

W przypadkach, gdy zapytanie wykonuje znaczną ilość pracy niezwiązanej z obliczeniami, takich jak we/wy pliku, korzystne może być określenie stopnia równoległości większego niż liczba rdzeni na maszynie.

Zapytania równoległe uporządkowane i nieurządzane

W niektórych zapytaniach operator zapytania musi wygenerować wyniki, które zachowują kolejność sekwencji źródłowej. PlINQ udostępnia AsOrdered operator w tym celu. AsOrdered różni się od AsSequential. Sekwencja AsOrdered jest nadal przetwarzana równolegle, ale jej wyniki są buforowane i sortowane. Ponieważ zachowywanie kolejności zwykle wiąże się z dodatkową pracą, AsOrdered sekwencja może być przetwarzana wolniej niż sekwencja domyślna AsUnordered . To, czy określona uporządkowana operacja równoległa jest szybsza niż sekwencyjna wersja operacji, zależy od wielu czynników.

Poniższy przykład kodu pokazuje, jak wyrazić zgodę na zachowanie kolejności.

var evenNums =
    from num in numbers.AsParallel().AsOrdered()
    where num % 2 == 0
    select num;

Dim evenNums = From num In numbers.AsParallel().AsOrdered()
               Where num Mod 2 = 0
               Select num

Aby uzyskać więcej informacji, zobacz Zachowywanie kolejności w PLINQ.

Zapytania równoległe a sekwencyjne

Niektóre operacje wymagają dostarczenia danych źródłowych w sekwencyjny sposób. Operatory ParallelEnumerable zapytań są przywracane do trybu sekwencyjnego automatycznie, gdy jest to wymagane. W przypadku operatorów zapytań zdefiniowanych przez użytkownika i delegatów użytkownika, które wymagają wykonywania sekwencyjnego, plINQ udostępnia metodę AsSequential . Gdy używasz metody AsSequential, wszystkie kolejne operatory w zapytaniu są wykonywane sekwencyjnie do momentu AsParallel ponownego wywołania. Aby uzyskać więcej informacji, zobacz How to: Combine Parallel and Sequential LINQ Queries (Instrukcje: łączenie równoległych i sekwencyjnych zapytań LINQ).

Opcje scalania wyników zapytania

Gdy zapytanie PLINQ jest wykonywane równolegle, jego wyniki z każdego wątku roboczego muszą zostać scalone z powrotem na główny wątek do użycia przez pętlę foreach (For Each w Visual Basic) lub wstawienie do listy lub tablicy. W niektórych przypadkach korzystne może być określenie konkretnego rodzaju operacji scalania, na przykład w celu szybszego tworzenia wyników. W tym celu plINQ obsługuje metodę WithMergeOptions i ParallelMergeOptions wyliczenie. Aby uzyskać więcej informacji, zobacz Opcje scalania w PLINQ.

The ForAll Operator

W sekwencyjnych zapytaniach LINQ wykonywanie jest odroczone, dopóki zapytanie nie zostanie wyliczone w pętli (For Eachw foreach visual basic) lub przez wywołanie metody, takiej jak ToList , ToArray lub ToDictionary. W plINQ można również użyć foreach polecenia , aby wykonać zapytanie i wykonać iterowanie wyników. Jednak foreach sam w sobie nie działa równolegle i dlatego wymaga scalenia danych wyjściowych ze wszystkich zadań równoległych z powrotem do wątku, na którym jest uruchomiona pętla. W PLINQ można użyć foreach , gdy musisz zachować ostateczną kolejność wyników zapytania, a także za każdym razem, gdy przetwarzasz wyniki w sposób szeregowy, na przykład podczas wywoływania Console.WriteLine dla każdego elementu. Aby przyspieszyć wykonywanie zapytań, gdy zachowanie kolejności nie jest wymagane, a przetwarzanie wyników może zostać zrównoleglizowane, użyj ForAll metody w celu wykonania zapytania PLINQ. ForAll nie wykonuje tego ostatniego kroku scalania. W poniższym przykładzie kodu pokazano, jak używać ForAll metody . System.Collections.Concurrent.ConcurrentBag<T> Jest używany w tym miejscu, ponieważ jest zoptymalizowany pod kątem wielu wątków dodających jednocześnie bez próby usunięcia żadnych elementów.

var nums = Enumerable.Range(10, 10000);
var query =
    from num in nums.AsParallel()
    where num % 10 == 0
    select num;

// Process the results as each thread completes
// and add them to a System.Collections.Concurrent.ConcurrentBag(Of Int)
// which can safely accept concurrent add operations
query.ForAll(e => concurrentBag.Add(Compute(e)));

Dim nums = Enumerable.Range(10, 10000)
Dim query = From num In nums.AsParallel()
            Where num Mod 10 = 0
            Select num

' Process the results as each thread completes
' and add them to a System.Collections.Concurrent.ConcurrentBag(Of Int)
' which can safely accept concurrent add operations
query.ForAll(Sub(e) concurrentBag.Add(Compute(e)))

Na poniższej ilustracji przedstawiono różnicę między operacjami foreach i ForAll w odniesieniu do wykonywania zapytań.

ForAll vs. ForEach

Anulowanie

PLINQ jest zintegrowany z typami anulowania na platformie .NET. (Aby uzyskać więcej informacji, zobacz Anulowanie w zarządzanych wątkach). W związku z tym, w przeciwieństwie do sekwencyjnych zapytań LINQ to Objects, zapytania PLINQ można anulować. Aby utworzyć anulowane zapytanie PLINQ, użyj WithCancellation operatora w zapytaniu i podaj CancellationToken wystąpienie jako argument. IsCancellationRequested Gdy właściwość tokenu jest ustawiona na wartość true, funkcja PLINQ zauważy ją, zatrzyma przetwarzanie we wszystkich wątkach i zgłosi wyjątek OperationCanceledException.

Istnieje możliwość, że zapytanie PLINQ może nadal przetwarzać niektóre elementy po ustawieniu tokenu anulowania.

Aby uzyskać większą szybkość reakcji, możesz również odpowiadać na żądania anulowania w długotrwałych delegatach użytkowników. Aby uzyskać więcej informacji, zobacz How to: Cancel a PLINQ Query (Instrukcje: anulowanie zapytania PLINQ).

Wyjątki

Gdy zapytanie PLINQ jest wykonywane, wiele wyjątków może być zgłaszanych jednocześnie z różnych wątków. Ponadto kod do obsługi wyjątku może znajdować się w innym wątku niż kod, który zgłosił wyjątek. PlINQ używa AggregateException typu do hermetyzacji wszystkich wyjątków, które zostały zgłoszone przez zapytanie, i marshaling tych wyjątków z powrotem do wątku wywołującego. W wątku wywołującym wymagany jest tylko jeden blok try-catch. Można jednak iterować przez wszystkie wyjątki, które są hermetyzowane w obiekcie AggregateException i przechwytywać dowolne, z których można bezpiecznie odzyskać. W rzadkich przypadkach niektóre wyjątki mogą być zgłaszane, które nie są opakowane w AggregateExceptionelement , a ThreadAbortExceptions również nie są opakowane.

Gdy wyjątki mogą być bąbelkowe z powrotem do wątku przyłączania, możliwe jest, że zapytanie może nadal przetwarzać niektóre elementy po wystąpieniu wyjątku.

Aby uzyskać więcej informacji, zobacz How to: Handle Exceptions in a PLINQ Query (Instrukcje: obsługa wyjątków w zapytaniu PLINQ).

Niestandardowe partycjonatory

W niektórych przypadkach można zwiększyć wydajność zapytań, pisząc niestandardowy partycjonator, który korzysta z niektórych cech danych źródłowych. W zapytaniu sam partycjonator niestandardowy jest obiektem wyliczalnym, którego dotyczy zapytanie.

int[] arr = new int[9999];
Partitioner<int> partitioner = new MyArrayPartitioner<int>(arr);
var query = partitioner.AsParallel().Select(SomeFunction);

Dim arr(10000) As Integer
Dim partitioner As Partitioner(Of Integer) = New MyArrayPartitioner(Of Integer)(arr)
Dim query = partitioner.AsParallel().Select(Function(x) SomeFunction(x))

PlINQ obsługuje stałą liczbę partycji (chociaż dane mogą być dynamicznie ponownie przypisywane do tych partycji w czasie wykonywania na potrzeby równoważenia obciążenia). For obsługują ForEach tylko partycjonowanie dynamiczne, co oznacza, że liczba partycji zmienia się w czasie wykonywania. Aby uzyskać więcej informacji, zobacz Custom Partitioners for PLINQ and TPL (Niestandardowe partycjonatory dla PLINQ i TPL).

Mierzenie wydajności PLINQ

W wielu przypadkach zapytanie może być równoległe, ale obciążenie związane z konfigurowaniem zapytania równoległego przewyższa korzyści wynikające z wydajności. Jeśli zapytanie nie wykonuje dużej ilości obliczeń lub jeśli źródło danych jest małe, zapytanie PLINQ może być wolniejsze niż sekwencyjne zapytanie LINQ to Objects. Możesz użyć równoległej Analizator wydajności w programie Visual Studio Team Server, aby porównać wydajność różnych zapytań, zlokalizować wąskie gardła przetwarzania i określić, czy zapytanie działa równolegle, czy sekwencyjnie. Aby uzyskać więcej informacji, zobacz Concurrency Visualizer (Wizualizator współbieżności) i How to: Measure PLINQ Query Performance (Jak mierzyć wydajność zapytań PLINQ).