Escalando processamento com Sockets utilizando ZeroMQ e dotnet

Este artigo segue abordando padrões de distribuição/processamento de requisições, utilizando como principal meio de comunicação sockets. No artigo anterior abordamos dois padrões de comunicação, que foram:

• Request-Reply
• Pub-Sub

Hoje vamos explorar mais um padrão, o Push-Pull.

Digamos que temos um conjunto de ações que devem ser executadas em uma requisição. Temos validação de logs, escrita em banco, regras de negócio, uma infinidade de operações que devem ser executadas. Tudo vai muito bem, até que determinadas partes do sistema começam a sofrer com tempo de execução, trazendo lentidão para a aplicação.

Apenas refatorar o código nem sempre reduz o tempo de execução, sempre podemos estar cometendo algum erro, por exemplo, mapeando entidades demais, buscando um volume de dados que não faz sentido para aquele contexto e etc.

Mas vamos imaginar que o nosso problema aqui não é exatamente um código zoado, a questão é que muitas atividades devem ser executadas para processar e por motivos diversos não conseguimos isolar melhor, ou simplificar as ações.

O que fazer?

Arquitetando a solução com sockets (zeromq)

Vamos imaginar que nosso servidor, que recebe a requisição, irá produzir "trabalhos" a serem executados. O servidor então irá gerar cargas de trabalho e enviar para que outros processos executem. Neste exemplo vamos chamar estes processos de workers.

Temos então um "Manager" que distribui requisições a serem executadas por "Workers", assim que finalizado o worker notifica algum agente sobre a conclusão do processamento.
O host responsável por receber o resultado do processamento pode tomar alguma ação com aquilo, seja persistir em banco, logar o resultado e etc.

Dessa forma, podemos crescer o número de workers conforme cresce a demanda, no exemplo temos apenas 2, mas poderia ser apenas 1 ou 10, tudo depende do volume de requisições que queremos atender.

Cenário

Temos produtos que precisam ser exibidos em um módulo de vendas, e associamos produtos do estoque para serem exibidos.

Algumas validações são necessárias, como processo, descrição e nome do produto.

Simples né? mas é uma situação que facilmente podemos nos deparar no dia a dia 🙂

Zmq Push Pull Socket com NetMQ

Quando falamos de Push-Pull, em geral, estamos falando exatamente sobre algum tipo de distribuição de tarefas entre diversos agentes. Na documentação do NetMQ temos:

The idea is that you have something that generates work, and then distributes the work out to n-many workers

Um socket do tipo Push vai enviar as requisições, quando o socket do tipo Pull espera por requisições para serem processadas.
Podemos atualizar o nosso exemplo anterior adicionando os sockets que usaremos.

Como demonstra a imagem, o manager possui apenas um socket do tipo Push que vai enviar request para serem executados pelos workers.
O worker possui dois sockets, um do tipo Pull que recebe as requisições do manager para processamento, e outro socket do tipo Push para notificar o finisher da conclusão.

Por sua vez, o finisher tem apenas um socket do tipo Pull para receber a notificação de conclusão.

Code time! Implementando com dotnet

Apesar do código abaixo ser escrito em dotnet, o zeromq já foi portado para mais de 28 linguagens, então provavelmente a sua de preferência vai ter uma biblioteca compatível 🙂

Vamos começar implementando o manager, ele deve produzir alguma carga de trabalho e enviar para os workers.

using (var manager = new PushSocket("@tcp://*:10000"))
{
    var command = string.Empty;

    while (true)
    {
        Console.WriteLine("Type 'yes' to start");
        command = Console.ReadLine();

        if (command.Equals("yes"))
        {
            var remoteControl = new Product(new Guid("1EDB76B0-F6C6-41B5-BCAA-CE8D1A18D361"), "Remove Control", "A remote control to rule the world.", 50, 5);
            var tv = new Product(new Guid("7691B9A6-6B0E-41C7-BA9B-FB2DA66710D6"),"TV", "The best televison in the universe ", 2000, 3, remoteControl);
            var productToBeValidated = Newtonsoft.Json.JsonConvert.SerializeObject(tv);

            manager.SendFrame(productToBeValidated);
        }
    }
}

Começamos criando o socket, e fazendo um bind para a porta 10000, em seguida criamos o produto a ser validado e enviamos para o worker.

manager.SendFrame(productToBeValidated);

E terminamos o nosso manager, ele envia para os workers produtos que precisam ser validados para exibição em algum módulo de vendas.
Em seguida passamos para o nosso worker.

var identity = Guid.NewGuid().ToString();

using (var receiver = new PullSocket(">tcp://localhost:10000"))
using (var validationPush = new PushSocket(">tcp://localhost:11000"))
{
    while (true)
    {
        var productTohavePriceValidated = receiver.ReceiveFrameString();
        Console.WriteLine(productTohavePriceValidated);

        var product = Newtonsoft.Json.JsonConvert.DeserializeObject<Product>(productTohavePriceValidated);

        var validationMessages = ValidateProduct(product);
        validationMessages.AddRange(ValidatePrice(product));

        var message = new ValidatedProductMessage(product, identity,validationMessages.ToArray());

        validationPush.SendFrame(Newtonsoft.Json.JsonConvert.SerializeObject(message));
    }
}

O worker recebe a requisição com receiver.ReceiveFrameString(); e realiza as validações necessárias para o produto.
Podemos simular aqui todo tipo de operação, como validação em banco, regras de negócio e etc. No nosso caso, estamos validando preço e estrutura do produto.

O código completo com as validações pode ser encontrada no github do RoadToAgility, o conteúdo dos métodos ValidateProduct e ValidatePrice apenas faz algumas validações, não precisa ficar preocupado com seu conteúdo.

Ao final o worker notifica o finisher sobre a conclusão do processamento.
O código do finisher é tão simples quanto os fontes do manager e do worker.

using (var priceValidator = new PullSocket("@tcp://localhost:11000"))
{
    while (true)
    {
        var message = priceValidator.ReceiveFrameString();
        var productMessage = Newtonsoft.Json.JsonConvert.DeserializeObject<ValidatedProductMessage>(message);

        Console.WriteLine($"Product validated: {productMessage.Product.Name}");
        Console.WriteLine($"Worker id: {productMessage.ProcessedBy}");
        Console.WriteLine("Erros:");

        foreach (var error in productMessage.Erros)
        {
            Console.WriteLine(error);
        }
    }
}

O finisher cria um socket do tipo Pull, para receber o resultado dos workers e escreve no console o resultado.

Para criarmos um novo worker, basta executar o worker multiplas vezes para termos vários workes operando.
Cada requisição enviada pelo PushSocket vai para apenas 1 worker por vez, a forma como o Manager vai gerenciar o envio dos requests entre os workers que se conectaram ao manager é parte do trabalho do zeromq.

Não temos controle sobre a sequência de workes que irá receber cada request, também fica por conta do zeromq gerenciar isso.

E pronto! agora temos um meio de escalar o processamento de requests conforme cresce a nossa demanda.

Ainda temos muito a fazer, podemos por exemplo ter multiplos sockets no finisher, ou diversas filas dentro do mesmo socket gerenciando cada um utilizando um poller.
Mas isso é assunto para um próximo post 😉

O códito utilizado pode ser encontrado no github do RoadToAgility
Até mais 🙂

#### Créditos
Abertura: Photo by La-Rel Easter on Unsplash

Tags: dotnet | netmq | push-pull | pushpull | sockets | zeromq | zmq