Processando grandes arquivos com o System.IO.Pipelines

O System.IO.Pipelines é uma biblioteca absurdamente eficiente para gerenciamento de I/O no dotnet. Ela foi projetada especialmente para ser utilizada em processamento de grandes volumes de dados. Imagine um parser de um arquivo gigante, ou de um stream de dados de uma conexão TCP por exemplo, tudo isso oferecendo um desempenho elevado com baixo overhead de memória.

Essa biblioteca usa o padrão de pipeline de I/O para ler e escrever dados de maneira muito eficiente, ela evita a cópia desnecessária de dados, minimizando a alocação de memória.
Outra coisa interessante é que ela oferece suporte ao processamento por fragmentos, permitindo assim, a manipulação de dados em pequenas partes. Isso é extremamente útil para aplicativos que precisam lidar com grandes fluxos de dados. Imagine você receber um stream gigantesco de dados, porém, processar um pedacinho por vez. É disso que estamos falando :)

O System.IO.Pipelines também é altamente escalável e pode ser usado em ambientes com muito processamento paralelo - múltiplas threads, sem a necessidade de sincronização. Se você conhece threads sabe o quão complexo, chato e burocrático fazer isso, e essa biblioteca já resolve esse problema. Aliás, escrevi um post sobre Channels, vale a pena ler. Imagine as possibilidades que se abrem ao unir Channels com o Pipelines. É tipo queijo com goiabada.

E tudo isso é possível graças ao uso de uma arquitetura baseada em tarefas, que permite uma execução segura de operações de I/O de forma assíncrona e paralela.
Acredito que o melhor exemplo para mostrar o poder de uso do System IO Pipelines é a criação de um leitor de arquivos grandes que processa os dados em pequenos blocos, ao invés de ler os dados inteiros de uma só vez.

Vamos criar um console application e instalar o pacote System.IO.Pipelines. Estou usando dotnet 7.

dotnet new console -o LeitorDeArquivoGrande
cd LeitorDeArquivoGrande
dotnet add package System.IO.Pipelines

using System.Buffers;
using System.IO.Pipelines;
using System.IO;
using System.Threading.Tasks;
using System;
using System.Text;

var caminhoDoArquivo = "meu_arquivo_gigante.txt";
using var fileStream = new FileStream(caminhoDoArquivo, FileMode.Open, FileAccess.Read);

await LerArquivo(fileStream);

static async Task LerArquivo(Stream stream)
{
    // Cria o Pipe
    var pipe = new Pipe();

    // Lê os dados do arquivo e joga no pipe
    var preencheTask = EscreveDadosNoPipe(stream, pipe.Writer);

    // Processa os dados à medida que são lidos
    var processaTask = ProcessaDadosDoPipe(pipe.Reader);

    // Aguarda até que os dados sejam lidos e processados
    await Task.WhenAll(preencheTask, processaTask);
}

static async Task EscreveDadosNoPipe(Stream stream, PipeWriter writer)
{
    const int tamanhoDoBuffer = 512; //magic number, também conhecido como o tamanho do pedaço que você percisa processar

    while (true) // não me julgue, isso é apenas um exemplo ;p
    {
        // Aloca um buffer
        var buffer = new byte[tamanhoDoBuffer];

        // Lê os dados do arquivo
        var bytesLidos = await stream.ReadAsync(buffer);

        // Se não existe dados a serem lidos, sai do loop
        if (bytesLidos == 0)
            break;

        // Escreve os dados no pipe
        await writer.WriteAsync(buffer.AsMemory(0, bytesLidos));
    }

    // Finaliza o processo de escrita
    await writer.CompleteAsync();
}

static async Task ProcessaDadosDoPipe(PipeReader reader)
{
    while (true) // ok, me julgue...
    {
        // Lê os dados do pipe
        var result = await reader.ReadAsync();

        // Obtém a memória do pipe
        var buffer = result.Buffer;

        // Processa os dados
        ProcessaParteDoTexto(buffer);

        // Notifica que já foram processados os dados
        reader.AdvanceTo(buffer.End);

        // Se não há mais dados, sai do loop
        if (result.IsCompleted)
            break;
    }
}

static void ProcessaParteDoTexto(ReadOnlySequence<byte> buffer)
{
    // Processa os dados aqui

    // Apenas para efeito didático...
    var texto = Encoding.UTF8.GetString(buffer.ToArray());
    Console.WriteLine(texto);
}

Como podemos notar, é bem simples efetuar uma leitura de arquivo usando o Pipeline. Da mesma forma, podemos processar stream de dados TCP, por exemplo.

Mais detalhes, acesse a documentação oficial da Microsoft aqui.

Gostou desse artigo? Deixe seu comentário.
Até mais!