C# Parallel e PLINQ

Na aula Threads, você criou threads uma a uma. Mas o caso mais comum é bem mais simples: você tem uma lista grande e quer fazer a mesma conta em cada item, usando todos os núcleos do processador. Para isso, o .NET tem o Parallel (laços paralelos) e o PLINQ (LINQ paralelo), que dividem o trabalho entre as threads sozinhos. Nesta aula você vai contar os números primos até 3 milhões em um terço do tempo, reduzir fotos ao mesmo tempo e ver por que uma List comum não aguenta várias threads.

Esta aula usa for, foreach, lambdas, LINQ e a condição de corrida da aula Threads. Os programas rodam com dotnet run no notebook da Ana, com 8 núcleos. Os tempos e a ordem de algumas linhas mudam a cada execução.

A ideia: a correção das provas

A professora tem 300 provas para corrigir. Sozinha, leva a tarde inteira. Ela chama mais 7 professores, divide a pilha em montinhos e cada um corrige o seu. A tarde vira meia hora. Ninguém precisa conversar com ninguém: cada prova é independente, e no fim basta juntar as notas.

O Parallel.For é essa divisão: ele reparte as voltas do laço entre as threads do pool (aula Task), uma por núcleo, mais ou menos, e só termina quando todas acabarem. Funciona bem quando as voltas são independentes, como as provas. Se cada volta depende da anterior (um saldo que vai sendo atualizado, por exemplo), o laço não pode ser dividido assim.

1. Parallel.For: um for dividido entre os núcleos

Parallel.For(0, 8, i =>
{
    Console.WriteLine($"Volta {i} na thread {Environment.CurrentManagedThreadId}");
});
Console.WriteLine("Todas as voltas terminaram");

$ dotnet run ordem.cs
Volta 4 na thread 4
Volta 6 na thread 10
Volta 7 na thread 11
Volta 5 na thread 7
Volta 1 na thread 8
Volta 0 na thread 1
Volta 2 na thread 9
Volta 3 na thread 6
Todas as voltas terminaram
$ dotnet run ordem.cs
Volta 2 na thread 5
Volta 4 na thread 8
Volta 1 na thread 1
Volta 3 na thread 7
Volta 0 na thread 4
Volta 5 na thread 9
Volta 6 na thread 10
Volta 7 na thread 11
Todas as voltas terminaram

Parallel.For(0, 8, i => { … })

O mesmo que for (int i = 0; i < 8; i++) { … }, mas com as voltas espalhadas entre threads. O primeiro número é o começo (incluído), o segundo é o fim (não incluído), e a lambda é o corpo do laço, que recebe o i. A classe Parallel fica em System.Threading.Tasks, já incluído.

a ordem das voltas

As 8 voltas rodaram em 8 threads, e a ordem da saída mudou de uma execução para a outra. Não há "volta 0 primeiro": todas acontecem ao mesmo tempo. A própria thread principal (a 1) também ajuda.

Todas as voltas terminaram

O Parallel.For só devolve o controle quando todas as voltas acabaram. Não precisa de Join nem de await.

Guardar os resultados: cada volta na sua posição

Para guardar o que cada volta calculou, o jeito mais simples e seguro é um array do tamanho do laço: a volta i escreve só na posição i. Como nenhuma volta mexe na posição de outra, não há corrida, e o array termina na ordem certa, mesmo que as voltas tenham rodado misturadas.

int[] quadrados = new int[100];
for (int i = 0; i < quadrados.Length; i++)
{
    quadrados[i] = (i + 1) * (i + 1);
}
Console.WriteLine($"Primeiros: {quadrados[0]}, {quadrados[1]}, {quadrados[2]}");
Console.WriteLine($"Soma dos quadrados: {quadrados.Sum()}");

int[] quadrados = new int[100];
Parallel.For(0, quadrados.Length, i =>
{
    quadrados[i] = (i + 1) * (i + 1);
});
Console.WriteLine($"Primeiros: {quadrados[0]}, {quadrados[1]}, {quadrados[2]}");
Console.WriteLine($"Soma dos quadrados: {quadrados.Sum()}");

2. Quanto fica mais rápido?

Contar os números primos até 3 milhões dá bastante trabalho: para cada número, o programa tenta dividi-lo por vários outros. É o caso ideal: muitas contas independentes. O mesmo trabalho de três jeitos:

using System.Diagnostics;

const int Limite = 3_000_000;
Stopwatch relogio = Stopwatch.StartNew();
int sequencial = 0;
for (int n = 2; n <= Limite; n++)
{
    if (EhPrimo(n)) sequencial++;
}
Console.WriteLine($"for:          {sequencial} primos em {relogio.Elapsed.TotalSeconds:F2} s");

relogio.Restart();
int paralelo = 0;
Parallel.For(2, Limite + 1, n =>
{
    if (EhPrimo(n)) Interlocked.Increment(ref paralelo);
});
Console.WriteLine($"Parallel.For: {paralelo} primos em {relogio.Elapsed.TotalSeconds:F2} s");

relogio.Restart();
int plinq = Enumerable.Range(2, Limite - 1).AsParallel().Count(EhPrimo);
Console.WriteLine($"PLINQ:        {plinq} primos em {relogio.Elapsed.TotalSeconds:F2} s");

bool EhPrimo(int n)
{
    if (n < 2) return false;
    for (int d = 2; d * d <= n; d++)
    {
        if (n % d == 0) return false;
    }
    return true;
}

$ dotnet run primos.cs
for:          216816 primos em 1,10 s
Parallel.For: 216816 primos em 0,33 s
PLINQ:        216816 primos em 0,32 s
$ dotnet run primos.cs
for:          216816 primos em 1,08 s
Parallel.For: 216816 primos em 0,23 s
PLINQ:        216816 primos em 0,35 s

EhPrimo(n)

"É primo?": tenta dividir n por 2, 3, 4… até a raiz quadrada de n (d * d <= n). Se alguma divisão for exata, não é primo.

Interlocked.Increment(ref paralelo)

Várias voltas somam no mesmo contador ao mesmo tempo. Com paralelo++, haveria a condição de corrida da aula Threads e o total sairia errado. O Interlocked soma de forma atômica.

Enumerable.Range(2, Limite - 1).AsParallel().Count(EhPrimo)

A mesma conta em PLINQ (seção 4): os números de 2 a 3 milhões, contados em paralelo. O PLINQ cuida de juntar as contagens das threads, sem Interlocked.

1,1 s → 0,3 s

Três a quatro vezes mais rápido em 8 núcleos, e o mesmo resultado: 216816 primos. Não chega a 8 vezes porque dividir e juntar o trabalho custa um pouco, e os núcleos dividem outras coisas (memória, outros programas).

Quando o paralelo fica mais lento

Dividir o trabalho tem um custo fixo: acordar threads, repartir a lista, juntar os resultados. Se cada conta é muito rápida e há poucas, o custo de dividir é maior que o ganho:

using System.Diagnostics;

int[] numeros = Enumerable.Range(1, 1000).ToArray();
Stopwatch r = Stopwatch.StartNew();
long a = numeros.Sum(n => (long)n * n);
double t1 = r.Elapsed.TotalMilliseconds;
r.Restart();
long b = numeros.AsParallel().Sum(n => (long)n * n);
double t2 = r.Elapsed.TotalMilliseconds;
Console.WriteLine($"LINQ:  {a} em {t1:F3} ms");
Console.WriteLine($"PLINQ: {b} em {t2:F3} ms");

$ dotnet run pequeno.cs
LINQ:  333833500 em 1,360 ms
PLINQ: 333833500 em 17,901 ms

Somar mil quadrados é instantâneo, e o PLINQ foi 13 vezes mais lento. A regra: paralelize quando houver muito trabalho (muitos itens, ou cada item demorado), e meça antes e depois com o Stopwatch.

3. Parallel.ForEach e Parallel.Invoke

O Parallel.ForEach é o foreach paralelo, para qualquer lista. A Ana precisa reduzir 16 fotos para mandar por mensagem; cada uma leva 0,25 s (aqui simulado com Thread.Sleep):

using System.Diagnostics;

string[] fotos = Enumerable.Range(1, 16).Select(i => $"foto{i:00}.jpg").ToArray();
Stopwatch relogio = Stopwatch.StartNew();
foreach (string foto in fotos)
{
    Reduzir(foto);
}
Console.WriteLine($"foreach: {relogio.Elapsed.TotalSeconds:F1} s");

relogio.Restart();
Parallel.ForEach(fotos, foto => Reduzir(foto));
Console.WriteLine($"Parallel.ForEach: {relogio.Elapsed.TotalSeconds:F1} s");

relogio.Restart();
ParallelOptions opcoes = new ParallelOptions { MaxDegreeOfParallelism = 2 };
Parallel.ForEach(fotos, opcoes, foto => Reduzir(foto));
Console.WriteLine($"Parallel.ForEach com 2 de cada vez: {relogio.Elapsed.TotalSeconds:F1} s");

void Reduzir(string foto)
{
    Thread.Sleep(250);
}

$ dotnet run fotos.cs
foreach: 4,0 s
Parallel.ForEach: 0,5 s
Parallel.ForEach com 2 de cada vez: 2,0 s

Parallel.ForEach(fotos, foto => Reduzir(foto))

Cada foto da lista vai para uma volta, e as voltas se espalham entre as threads. 16 fotos em 8 núcleos: duas "rodadas" de 0,25 s.

new ParallelOptions { MaxDegreeOfParallelism = 2 }

"Grau máximo de paralelismo": no máximo 2 voltas ao mesmo tempo. Serve para não sobrecarregar algo que não aguenta muitos pedidos juntos (o disco, um servidor). 16 fotos, 2 de cada vez: 8 rodadas, 2 s.

Para fazer coisas diferentes ao mesmo tempo, e não a mesma coisa em vários itens, existe o Parallel.Invoke ("invocar"):

Parallel.Invoke(
    () => Console.WriteLine("Lavar a louça"),
    () => Console.WriteLine("Varrer a casa"),
    () => Console.WriteLine("Regar as plantas"));
Console.WriteLine("Faxina terminada");

$ dotnet run invoke.cs
Lavar a louça
Varrer a casa
Regar as plantas
Faxina terminada

Cada lambda é uma ação. Desta vez, a ordem saiu igual à do código, mas ela não é garantida. A única garantia é que Faxina terminada vem por último: o Invoke espera todas.

4. PLINQ: AsParallel

Se a conta já está escrita em LINQ, deixá-la paralela é uma palavra: AsParallel() ("como paralelo"). Tudo o que vem depois dele roda em várias threads:

int[] numeros = Enumerable.Range(1, 12).ToArray();
Console.WriteLine(string.Join(" ", numeros.AsParallel().Select(n => n * n)));
Console.WriteLine(string.Join(" ", numeros.AsParallel().AsOrdered().Select(n => n * n)));
Console.WriteLine(numeros.AsParallel().Sum(n => n * n));

$ dotnet run plinq.cs
1 4 9 25 49 81 100 121 144 16 36 64
1 4 9 16 25 36 49 64 81 100 121 144
650
$ dotnet run plinq.cs
1 9 25 49 81 100 121 144 4 16 36 64
1 4 9 16 25 36 49 64 81 100 121 144
650

numeros.AsParallel().Select(n => n * n)

Os quadrados, calculados em paralelo. A ordem do resultado embaralhou, e mudou entre as execuções: cada thread entregou o seu pedaço quando terminou.

.AsOrdered()

"Ordenado": as contas continuam em paralelo, mas o PLINQ devolve os resultados na ordem original. Custa um pouco de velocidade, por isso não é o padrão.

.Sum(…), .Count(…), .Max(…)

As agregações (aula Agregar) juntam os resultados das threads com segurança. Como somar não depende da ordem, o total é sempre o mesmo: 650.

O PLINQ também tem .WithDegreeOfParallelism(2), o equivalente do MaxDegreeOfParallelism, e .ToList() e .ToArray(), que juntam os resultados numa coleção com segurança (seção "Quando dá errado").

Qual usar

Quando dá errado

List<int> lista = [];
Parallel.For(0, 100_000, i =>
{
    lista.Add(i);
});
Console.WriteLine($"Itens na lista: {lista.Count}");

$ dotnet run lista.cs; echo "código: $?"
Itens na lista: 85308
código: 0
$ dotnet run lista.cs; echo "código: $?"
Unhandled exception. System.AggregateException: One or more errors occurred. (Source array was not long enough. Check the source index, length, and the array's lower bounds. (Parameter 'sourceArray')) (Source array was not long enough. Check the source index, length, and the array's lower bounds. (Parameter 'sourceArray')) (Source array was not long enough. Check the source index, length, and the array's lower bounds. (Parameter 'sourceArray'))
 ---> System.ArgumentException: Source array was not long enough. Check the source index, length, and the array's lower bounds. (Parameter 'sourceArray')
   at System.Array.CopyImpl(Array sourceArray, Int32 sourceIndex, Array destinationArray, Int32 destinationIndex, Int32 length, Boolean reliable)
   at System.Collections.Generic.List`1.set_Capacity(Int32 value)
   at System.Collections.Generic.List`1.AddWithResize(T item)
   at Program.<>c__DisplayClass0_0.<<Main>$>b__0(Int32 i) in /home/ana/cs/lista.cs:line 4
   …                                                « mais 34 linhas, com as outras duas exceções
código: 134
$ dotnet run lista.cs; echo "código: $?"
Itens na lista: 16403
código: 0

using System.Collections.Concurrent;

ConcurrentBag<int> saco = [];
Parallel.For(0, 100_000, i => saco.Add(i));
Console.WriteLine($"ConcurrentBag: {saco.Count}");

List<int> lista = Enumerable.Range(0, 100_000).AsParallel().Where(i => i % 2 == 0).ToList();
Console.WriteLine($"PLINQ ToList: {lista.Count}");

$ dotnet run lista2.cs
ConcurrentBag: 100000
PLINQ ToList: 50000

$ sed -i "s/Interlocked.Increment(ref paralelo)/paralelo++/" primos.cs
$ grep -n "paralelo++" primos.cs
16:    if (EhPrimo(n)) paralelo++;
$ dotnet run primos.cs
for:          216816 primos em 1,07 s
Parallel.For: 215244 primos em 0,22 s
PLINQ:        216816 primos em 0,26 s
$ dotnet run primos.cs
for:          216816 primos em 1,11 s
Parallel.For: 200636 primos em 0,19 s
PLINQ:        216816 primos em 0,33 s

try
{
    Parallel.For(0, 10, i =>
    {
        if (i == 7) throw new InvalidOperationException($"Falhou na volta {i}");
    });
}
catch (AggregateException e)
{
    Console.WriteLine($"{e.InnerExceptions.Count} erro(s):");
    foreach (Exception erro in e.InnerExceptions)
    {
        Console.WriteLine($" - {erro.Message}");
    }
}

$ dotnet run exc.cs
1 erro(s):
 - Falhou na volta 7

Referência rápida

Exercícios

Missão final: os primos em paralelo

9592 primos até 100000
Os 10 primeiros: 2 3 5 7 11 13 17 19 23 29

IEnumerable<int> numeros = Enumerable.Range(2, 99_999);

int total = numeros.Count(EhPrimo);
Console.WriteLine($"{total} primos até 100000");
// os 10 primeiros, em ordem

bool EhPrimo(int n)
{
    for (int d = 2; d * d <= n; d++)
    {
        if (n % d == 0) return false;
    }
    return n >= 2;
}

IEnumerable<int> numeros = Enumerable.Range(2, 99_999);

int total = numeros.AsParallel().Count(EhPrimo);
Console.WriteLine($"{total} primos até 100000");
IEnumerable<int> primeiros = numeros.AsParallel().AsOrdered().Where(EhPrimo).Take(10);
Console.WriteLine($"Os 10 primeiros: {string.Join(" ", primeiros)}");

bool EhPrimo(int n)
{
    for (int d = 2; d * d <= n; d++)
    {
        if (n % d == 0) return false;
    }
    return n >= 2;
}

O Enumerable.Range(2, 99_999) são os 99.999 números de 2 a 100.000. Sem o AsOrdered, num computador com vários núcleos, os "10 primeiros" poderiam vir fora de ordem, ou nem ser os 10 menores, porque cada thread acha os seus primos no seu pedaço da lista. Se quiser ir além: ponha um Stopwatch e aumente o limite para 3 milhões, como na seção 2, e compare o tempo com e sem AsParallel no seu computador.

Resumo

• Parallel.For e Parallel.ForEach dividem as voltas de um laço entre os núcleos e só terminam quando todas acabam. As voltas precisam ser independentes, e a ordem entre elas não é garantida.
• Parallel.Invoke roda ações diferentes ao mesmo tempo; MaxDegreeOfParallelism limita quantas rodam juntas.
• AsParallel() deixa uma consulta LINQ paralela; AsOrdered() mantém a ordem. Count, Sum e ToList juntam os resultados com segurança.
• Com muito trabalho, o ganho é grande (1,1 s → 0,3 s em 8 núcleos); com pouco, o paralelo fica mais lento. Meça.
• Nada de List.Add ou total++ compartilhados: use posições de um array, Interlocked, coleções Concurrent… ou as agregações do PLINQ.
• Exceções das voltas chegam juntas numa AggregateException. Para esperas (rede, disco), o caminho é async/await, e não o Parallel.