Forem: Eduardo Klosowski

Branch para deploy de validação

Eduardo Klosowski — Fri, 26 Sep 2025 01:32:08 +0000

Em um projeto de software é comum precisar testar uma funcionalidade que está sendo ou foi desenvolvida. Nos projetos em que o código roda em um servidor, isso pode envolver o deploy em algum ambiente específico, permitindo assim testar uma versão específica do código interagindo com as aplicações cliente ou demais serviços. Porém quando mais de uma funcionalidade é desenvolvida ao mesmo tempo em branches diferentes do git, muitas vezes por pessoas diferentes, testar duas ou mais funcionalidades ao mesmo tempo pode não ser trivial.

Exemplificando o problema

Um fluxo bastante comum usado no git é criar uma branch para desenvolver uma funcionalidade. Quando essa funcionalidade estiver pronta, e preferencialmente testada, é feito o merge para a branch principal do projeto. Isso permite que diferentes funcionalidades sejam desenvolvidas em paralelo, seguindo cada uma as etapas do seu fluxo de trabalho, sem ou com a menor interferência possível, assim como também permite desistir do desenvolvimento de uma funcionalidade ou recomeçá-lo.

Um exemplo de diferentes branches do git pode ser visto a baixo:

Se considerar que o projeto é o código de uma API Rest que é executado em um servidor, e que existem alguns ambientes diferentes como:

Produção: ambiente em que o projeto está de fato sendo utilizado. Deploy realizado a partir da branch main.
Homologação: ambiente para validação antes de ir para produção. Deploy realizado a partir da branch main.
Desenvolvimento: ambiente para testar as coisas pelos desenvolvedores, sem necessariamente se comprometer em levar para produção o que é feito aqui. Deploy realizado a partir de qualquer branch.

Caso as pessoas que desenvolveram as branches feature-1 e feature-2 queiram testar seus códigos no ambiente Desenvolvimento, apenas uma versão (branch) poderá ser testada por vez, já que ao realizar o deploy da versão do commit 4, ele substituirá o deploy da versão do commit 2, por exemplo, e vice-versa.

Proposta de solução

Uma forma de abordar esse problema é através da criação uma versão com todas as funcionalidades que se deseja testar. No git isso pode ser feito criando outra branch para reunir as branches desejadas (fazer merge delas). Por exemplo, executando os seguintes comandos:

git checkout -b desenvolvimento main
git merge --no-ff feature-1
git merge --no-ff feature-2

O que geraria o seguinte histórico no git:

Assim ao realizar o deploy no ambiente Desenvolvimento sempre a partir da branch desenvolvimento, isso permite testar todas as funcionalidades desejadas ao mesmo tempo, nesse caso as funcionalidades presentes nas branches feature-1 e feature-2. Porém essas funcionalidades ainda terão suas respectivas branches separadas para continuar seu fluxo de trabalho, como abertura de pull request (PR) ou merge request (MR), revisão de código e demais etapas que podem existir sem se misturar com as outras, não se importando com a branch desenvolvimento.

A branch desenvolvimento ainda poderá receber novas funcionalidades de outras branches, alterações das branches que já havia feito o merge anteriormente. Ou seja, novas versões podem ser feitas nessa branch conforme a necessidade. Um exemplo é um teste de alteração na feature-1 (commit 7), que foi feito o merge e deploy (versão no commit 8), e se não teve o resultado esperado, poderá ser revertido:

Entrega das funcionalidades

Quando uma funcionalidade for validada e estiver dada como pronta, poderá seguir seu fluxo de trabalho normal, sendo feito o merge de sua branch com a main, como se a branch desenvolvimento não existisse. Desta forma apenas essa funcionalidade integrará a branch main, podendo ser validada no ambiente Homologação e chegar a Produção, sem as demais funcionalidades que ainda estão sendo testadas no ambiente Desenvolvimento. E ela pode se juntar a outras funcionalidades, como uma pequena correção que foi aplicada diretamente na branch main(ambientes Homologação e Produção), sem ter passado pela branch desenvolvimento.

Limpeza da branch

A funcionalidade foi testada na branch desenvolvimento e entregue na branch main, porém é possível observar que com o passar do tempo pode a surgir diferenças entre o código das branches desenvolvimento e main. Um caso foi a correção fix-3, que foi aplicada diretamente na main, sem passar pela desenvolvimento, e outro caso poderia ser a feature-1, se seu desenvolvimento for pausado ou cancelado. Isso é ruim, pois poderia gerar código que funciona na desenvolvimento, mas não na main.

Desta forma é recomendável fazer limpeza da branch desenvolvimento, que seria uma ressincronização com a branch main da onde ela surgiu. Ao fazer isso, serão removidas dela todas as funcionalidades que não chegaram na branch main, porém como essas funcionalidades ainda terão suas branches individuais, seus merges podem ser refeitos com a branch desenvolvimento se ainda for relevante e seu teste estiver ocorrendo. E se não, essa funcionalidade deixará de influenciar nos testes das demais.

Essa limpeza pode ser efetuada simplesmente recriando a branch, por exemplo:

git checkout -B desenvolvimento main

E ao unir a feature-1 para testes na desenvolvimento, o histórico seria o seguinte:

Embora essa limpeza seja boa para manter as branches desenvolvimento e main semelhantes, evitando assim alguns problemas, uma frequência muito grande também pode dificultar os testes, uma vez que desativará as funcionalidades que estão em teste. Desta forma é necessário encontrar um equilíbrio, seja uma vez por dia ou semana, e isso pode mudar conforme o projeto. Outra opção é a criação de um "botão de manual", que poderia ser usado sempre que se achar necessário, e que pode até trabalhar em conjunto com algum agendamento.

Como implementar?

Os comandos sugeridos anteriormente funcionam no repositório local. Para integrar com um repositório remoto é necessário algumas adaptações.

Criando ou limpando a branch

Para criar ou limpar a branch desenvolvimento é necessário ter a branch main atualizada, criar a branch desenvolvimento e enviar para o servidor sobreescrevendo a atual. Isso pode ser feito com a seguinte sequência de comandos:

git fetch origin main
git checkout -B desenvolvimento origin/main
git push origin +desenvolvimento

Merge de funcionalides para teste

Para adicionar uma funcionalidade na branch desenvolvimento é necessário ter as branches desenvolvimento e da funcionalidade desejada atualizada, após isso é necessário fazer o merge e enviar para o servidor. Isso pode ser feito com a seguinte sequência de comandos:

git fetch origin desenvolvimento feature-1
git checkout -B desenvolvimento origin/desenvolvimento
git merge --no-ff origin/feature-1
git push origin desenvolvimento

Caso alterações sejam feitas na branch da funcionalidade depois disso, é possível fazer um novo merge.

Revertendo alterações

Caso alguma funcionalidade apresente erro, ou se deseja removê-la sem mexer nas demais, é possível reverter um merge na desenvolvimento. A forma para ter isso é ter a branch desenvolvimento atualiza e usar o próprio git para gerar um commit desfazendo as alterações do merge. Isso pode ser feito com a seguinte sequência de comandos informando o hash do commit de merge:

git fetch origin desenvolvimento
git checkout -B desenvolvimento origin/desenvolvimento
git revert -m 1 aa9ff52

Um ponto a se observar é que ao fazer um commit revertendo as alterações de um merge, não será mais possível fazer o merge dessa funcionalidade até a branch desenvolvimento ser limpa, uma vez que os commits da branch dessa funcionalidade já estão no histórico da desenvolvimento. Porém é possível reverter um commit que reverteu o merge, trazendo a funcionalidade de volta.

Onde executar?

Uma forma de executar essas sequências de comandos é através de um script, onde quem desejar poderia executá-lo e passar qual a operação desejada, branch e afins.

Outra opção é utilizar alguma ferramenta de integração contínua (CI), assim bastaria executar um job passando as mesmas informações como parâmetro. Caso não tenha alguma ferramenta de CI e o deploy da aplicação for feito no Kubernetes, é possível utilizar o Tekton para isso, ele irá executar o job usando o próprio Kubernetes para isso.

Considerações

Não existe uma forma trivial de testar diferentes branches juntas, porém é possível fazer um fluxo com uma branch a parte para isso, sendo que essa branch não vai interferir no fluxo de trabalho anterior, ficando apenas como um passo opcional. Porém é necessário implementar algumas funcionalidades com sequências de comandos para isso.

O fluxo apresentado visa fazer um único deploy com todas as funcionalidades desejadas. Outra opção seria fazer um deploy separado para cada branch, dando um endereço diferente para acessá-los. Essa outra forma permitiria testar as funcionalidades de forma isolada, porém cada deploy consumiria recursos e poderia ter dificuldades caso seja necessário integrar esse novo deploy em algum serviço, como receber eventos ou mensagens de outros serviços.

Testando código que chama serviços da AWS

Eduardo Klosowski — Wed, 05 Feb 2025 02:42:11 +0000

Eu desenvolvo sistemas que utilizam os serviços da AWS faz algum tempo, e ao longo desse tempo houve mudanças na forma como escrevo testes de código que fazem chamadas a seus serviços. Esse texto tem como objetivo apresentar algumas abordagens para escrita de testes que utilizei, e discutir o que motivou suas evoluções, destacando características de cada abordagem. Ao final, pretendo apresentar um padrão que acredito ser uma forma bastante prática de escrever testes de código que interage com serviços da AWS, usando uma biblioteca em Python que desenvolvi implementando esse padrão, mas que também poderia ser adaptado para outros contextos (serviços) e linguagens.

Código a ser testado

Antes de iniciar a discussão sobre os testes, quero apresentar um exemplo de código para ser testado. Ele é uma função que deve consumir mensagens enviadas para uma fila SQS, realizar um processamento com a informação contida nas mensagens, e enviar o resultado para outra fila SQS.

import json

def funcao1(cliente_sqs, fila1_url, fila2_url):
    # Recebe até 10 mensagens do SQS
    msgs_recebidas = cliente_sqs.receive_message(
        QueueUrl=fila1_url,
        MaxNumberOfMessages=10,
    )

    # Percorre as mensagens recebidas
    for msg in msgs_recebidas.get('Messages', []):
        # Recupera valor da mensagem
        corpo = json.loads(msg['Body'])
        n = corpo['n']

        # Processa valor
        resultado = n * 2
        resposta = {'resultado': resultado}

        # Envia mensagem com resultado
        cliente_sqs.send_message(
            QueueUrl=fila2_url,
            MessageBody=json.dumps(resposta),
        )

        # Apaga mensagem atual
        cliente_sqs.delete_message(
            QueueUrl=fila1_url,
            ReceiptHandle=msg['ReceiptHandle'],
        )

Esse código é uma função que recebe um objeto para interagir com o serviço SQS (cliente_sqs) e a URL de duas filas (fila1_url e fila2_url). Ele busca até 10 mensagens da fila1_url (o máximo permitido por chamada), cada mensagem é um JSON que possui um número no campo n. Após recuperar esse valor, um processamento é feito (nesse caso multiplicar o valor por 2), um novo JSON é gerado e enviado para a fila2_url. Se tudo isso ocorrer conforme esperado, a mensagem processada é removida da fila1_url, evitando que ela volte para a fila e eventualmente ser feito uma nova tentativa de processá-la. Esse processo é repetido para cada mensagem recebida.

Teste unitário com mock

Uma forma bastante simples de testar é fazer testes unitários (também chamados de testes de unidade) e usar mocks. Para o código de exemplo é possível criar um objeto que simula (mock) o cliente_sqs passado para a função, e registre as funções chamadas dele. Depois de executar a função a ser testada, basta verificar as chamadas feitas no objeto simulado, e assim validar se a função está se comportando como o esperado.

Na biblioteca padrão do Python existe o MagicMock, que facilita a criação de mocks. Bastando definir o que deve ser retornando em cada função (se ela tiver retorno), e depois verificar como suas funções foram chamadas (assert). Caso espera-se que uma função tenha sido chamada de determinada forma e isso não ocorreu, o teste falhará.

Segue a baixo um exemplo de teste unitário com mock para o código apresentado anteriormente:

import json
from exemplo.codigo1 import funcao1
from unittest.mock import MagicMock
from uuid import uuid4

def test_funcao1():
    # Mensagens com valores de entrada e resultados esperados
    msgs = [
        {'n': 0},
        {'n': 1},
        {'n': 2},
    ]
    respostas = [
        {'resultado': 0},
        {'resultado': 2},
        {'resultado': 4},
    ]

    # Define outros valores auxiliares para o teste
    fila1_url = 'http://sqs.aws/fila1'
    fila2_url = 'http://sqs.aws/fila2'
    msgs_mockadas = [
        {
            'ReceiptHandle': uuid4().hex,
            'Body': json.dumps(msg),
        } for msg in msgs
    ]

    # Cria e configura mock do SQS
    mock_cliente_sqs = MagicMock()
    mock_cliente_sqs.receive_message.return_value = {
        'Messages': msgs_mockadas,
    }

    # Chama função a ser testada
    funcao1(mock_cliente_sqs, fila1_url, fila2_url)

    # Verifica se a função buscou as mensagens na fila 1
    mock_cliente_sqs.receive_message.assert_called_once_with(
        QueueUrl=fila1_url,
        MaxNumberOfMessages=10,
    )

    # Verifica se as mensagens da fila 1 foram apagadas
    assert mock_cliente_sqs.delete_message.call_count == len(msgs)
    for msg in msgs_mockadas:
        mock_cliente_sqs.delete_message.assert_any_call(
            QueueUrl=fila1_url,
            ReceiptHandle=msg['ReceiptHandle'],
        )

    # Verifica se as respostas estão corretas na fila 2
    assert mock_cliente_sqs.send_message.call_count == len(respostas)
    for esperado in respostas:
        mock_cliente_sqs.send_message.assert_any_call(
            QueueUrl=fila2_url,
            MessageBody=json.dumps(esperado),
        )

Essa abordagem tem algumas vantagens, como: Não precisar configurar e acessar o serviço SQS para rodar os testes, executando tudo localmente; E testa um trecho de código de forma isolada (uma função nesse caso), independente de outras partes do código, como o cliente_sqs. Porém esse método também apresenta desvantagens, como: Ao isolar uma parte do código, perde-se a certeza se ele funcionará junto com as demais partes do sistema, uma vez que podem existir diferenças entre uma resposta real e a resposta do mock (nesse teste o msgs_mockadas possui apenas duas chaves, quando numa resposta verdadeira existiriam várias outras também); Além de ser necessário definir toda vez qual o retorno das funções e validar se elas foram chamadas conforme o esperado; Também é possível fazer um mock que aceite uma chamada inválida para o serviço real; E como a mensagem enviada é uma string com JSON, a ordem dos campos pode mudar, assim como sua formatação, o que pode causar un falso negativo no teste. Então qualquer alteração de parâmetros da função ou resposta, pode gerar uma incompatibilidade do mock com o comportamento real do sistema.

Teste de integração com uma reimplementação

Partindo mais para uma abordagem de testes de integração, existe uma reimplementação dos serviços da AWS em Python para ser usada em testes chamada Moto (que também pode ser utilizada em outras linguagens no seu modo servidor, semelhante ao LocalStack). Assim também é possível rodar os testes localmente, uma vez que ele simula o comportamento dos serviços da AWS. Um exemplo de teste utilizando essa biblioteca pode ser visto a baixo:

import boto3
import json
from exemplo.codigo1 import funcao1
from moto import mock_aws

@mock_aws
def test_funcao1():
    # Mensagens com valores de entrada e resultados esperados
    msgs = [
        {'n': 0},
        {'n': 1},
        {'n': 2},
    ]
    respostas = [
        {'resultado': 0},
        {'resultado': 2},
        {'resultado': 4},
    ]

    # Conecta no SQS
    cliente_sqs = boto3.client('sqs', region_name='us-east-1')

    # Cria filas de teste
    fila1_url = cliente_sqs.create_queue(QueueName='fila1')['QueueUrl']
    fila2_url = cliente_sqs.create_queue(QueueName='fila2')['QueueUrl']

    # Envia mensagens de teste
    for msg in msgs:
        cliente_sqs.send_message(
            QueueUrl=fila1_url,
            MessageBody=json.dumps(msg),
        )

    # Chama função a ser testada
    funcao1(cliente_sqs, fila1_url, fila2_url)

    # Verifica se a função buscou e apagou as mensagens na fila 1
    assert sum(int(attr) for attr in cliente_sqs.get_queue_attributes(
        QueueUrl=fila1_url,
        AttributeNames=[
            'ApproximateNumberOfMessages',
            'ApproximateNumberOfMessagesNotVisible',
        ],
    )['Attributes'].values()) == 0

    # Verifica se as respostas estão corretas na fila 2
    msgs_de_resposta = cliente_sqs.receive_message(
        QueueUrl=fila2_url,
        MaxNumberOfMessages=10,
    )['Messages']
    for msg in msgs_de_resposta:
        cliente_sqs.delete_message(
            QueueUrl=fila2_url,
            ReceiptHandle=msg['ReceiptHandle'],
        )
    assert [
        json.loads(msg['Body']) for msg in msgs_de_resposta
    ] == respostas

    # Remove filas de teste
    cliente_sqs.delete_queue(QueueUrl=fila1_url)
    cliente_sqs.delete_queue(QueueUrl=fila2_url)

Diferente do teste anterior, neste não é necessário se preocupar em montar os retornos dos serviços acessados, porém é necessário conhecer melhor o serviço para criar os recursos utilizados pela função a ser testada. Também não existe uma forma de fazer um assert para verificar se uma mensagem foi enviada, é necessário buscá-las da fila, e nesse caso, também chamar a função para removê-las, de forma que elas não voltem para a fila, evitando comportamentos não esperados (embora não seja obrigatório ao usar o Moto como um decorador, mas é bom para evitar problemas). E mesmo que possa existir alguma diferença entre o serviço real e sua reimplementação, espera-se que ele seja confiável, e ao atualizar o Moto, todos os testes já serão validados com os novos comportamentos, terceirizando essa responsabilidade do teste.

Interagindo com outro serviço (SNS)

Pensando em testar outro serviço da AWS, pode-se trocar a fila SQS para qual a resposta do código é enviada por um tópico SNS. O código a ser testado então fica:

import json

def funcao2(cliente_sqs, cliente_sns, fila_url, topico_arn):
    # Recebe até 10 mensagens do SQS
    msgs_recebidas = cliente_sqs.receive_message(
        QueueUrl=fila_url,
        MaxNumberOfMessages=10,
    )

    # Percorre as mensagens recebidas
    for msg in msgs_recebidas.get('Messages', []):
        # Recupera valor da mensagem
        corpo = json.loads(msg['Body'])
        n = corpo['n']

        # Processa valor
        resultado = n * 2
        resposta = {'resultado': resultado}

        # Publica mensagem com resultado
        cliente_sns.publish(
            TopicArn=topico_arn,
            Message=json.dumps(resposta),
        )

        # Apaga mensagem atual
        cliente_sqs.delete_message(
            QueueUrl=fila_url,
            ReceiptHandle=msg['ReceiptHandle'],
        )

Seu teste pode seguir uma estrutura bastante similar:

import boto3
import json
from exemplo.codigo2 import funcao2
from moto import mock_aws

@mock_aws
def test_funcao2():
    # Mensagens com valores de entrada e resultados esperados
    msgs = [
        {'n': 0},
        {'n': 1},
        {'n': 2},
    ]
    respostas = [
        {'result': 0},
        {'result': 2},
        {'result': 4},
    ]

    # Conecta no SQS e SNS
    cliente_sqs = boto3.client('sqs', region_name='us-east-1')
    cliente_sns = boto3.client('sns', region_name='us-east-1')

    # Cria fila e tópico de teste
    fila_url = cliente_sqs.create_queue(QueueName='fila')['QueueUrl']
    topico_arn = cliente_sns.create_topic(Name='topico')['TopicArn']

    # Envia mensagens de teste
    for msg in msgs:
        cliente_sqs.send_message(
            QueueUrl=fila_url,
            MessageBody=json.dumps(msg),
        )

    # Chama função a ser testada
    funcao2(cliente_sqs, cliente_sns, fila_url, topico_arn)

    # Como fazer um assert das mensagens publicadas no SNS?
    assert ...

    # Remove fila e tópico de teste
    cliente_sqs.delete_queue(QueueUrl=fila_url)
    cliente_sns.delete_topic(TopicArn=topico_arn)

Porém como verificar o que foi enviado para o tópico SNS? Como mocks não estão sendo utilizados, não tem como fazer um assert para verificar se a função foi chamada e com quais parâmetros. O SNS também não tem uma forma direta de recuperar as mensagens publicadas nele. Uma solução possível é criar uma fila SQS, assinar o tópico SNS, de forma que tudo que for enviado para o tópico seja encaminhado para a fila SQS, e validar o tópico SNS a partir da fila SQS (e após a execução do teste, remover a assinatura, fila e tópico). Embora possível, essa solução é bastante trabalhosa e repetitiva, ainda mais se considerar que isso precisará ser refeito em cada teste do projeto que chamar o SNS.

Simplificando e padronizando (repensando os testes)

Embora a solução apresentada tenha o problema da repetição de código, ela funciona, e para os testes não importa saber criar uma fila ou tópico e removê-los depois (que é uma parte considerável do que é repetido), só importa usá-los. É totalmente plausível (pelo menos na ideia), de na hora de rodar o teste, só perguntar por alguma fila disponível que possa ser utilizada, e depois devolvê-la. Então seria bom se tivesse uma forma de pegar filas e tópicos emprestados para os testes, ou até mesmo criá-los e removê-los, o teste em si não precisa saber o que acontecerá com esses recursos depois de utilizá-los. Na verdade, o teste nem precisaria saber do detalhe de que é necessário primeiro buscar uma mensagem da fila do SQS, e depois deletar essa mensagem, nem de que a mensagem publicada em um tópico SNS precisou passar por uma fila SQS para ser validada.

Pensando dessa forma, só é necessário executar algum código antes do teste para criar os recursos (filas, tópicos...) que serão utilizados, e executar algum código depois dos testes para remover esses recursos criados anteriormente. Além disso, alguma abstração poderia ser criada para simplificar a interação com esses recursos na parte dos testes para facilitar a verificação de sua utilização, que é justamente a parte que importa dos testes.

Para executar algum código antes e depois de algo, no Python pode-se ser usado um contexto gerenciado (aquele criado com a estrutura with ...), que poderia criar os recursos, passá-lo para o teste, e assim que ele acabar, fazer sua remoção. Porém, o que ele deveria retornar?

Para testar um código que envia mensagens para uma fila SQS, só é necessário saber qual a fila e alguma forma de receber as mensagens enviadas. A identificação de uma fila no SQS se dá através de uma URL, e um gerador poderia ser criado para abstrair toda a lógica de consumir a fila. Já para testar um código que recebe mensagens de uma fila SQS é necessário saber a URL da fila, e uma função para enviar as mensagens que serão consumidas pelo código a ser testado.

Agora considerando o caso do tópico SNS, toda vez que um tópico for criado, também poderia se criar uma fila SQS que assina esse tópico. A função geradora também poderia buscar nessa fila as mensagens. Isso deixaria o SQS totalmente transparente para os testes (o teste nem saberia que tem uma fila SQS sendo utilizada). Porém para fazer a assinatura, a URL da fila não serve, é necessário seu ARN, em outros contextos pode ser necessário o nome e não URL ou ARN.

Para o contexto retornar mais de uma coisa, pode-se utilizar as tuplas do Python. Porém a questão continua, o que estará nessa tupla? URL da fila? ARN da fila? Nome? Função para enviar mensagens? Gerador para consumir mensagens? Para não ser necessário criar diferentes tuplas para cada caso, pode-se criar uma tupla retornando tudo. Mas pode ser complicado lembrar em que posição está cada coisa, e um código que recebe uma tupla com vários valores e não utiliza a maioria pode gerar confusão. Então em vez de usar tuplas, pode-se criar um objeto seguindo o modelo de dados do Python, assim informações (como nome, ARN e URL) podem ir como atributos, a função para enviar mensagens poderia ser um método, e esse objeto também poderia se comportar como um iterador, que é um comportamento do gerador. E por seguir o modelo de dados do Python, isso tudo ainda pareceria algo nativo do Python, onde um objeto abstrairia a fila SQS ou tópico SNS (algo semelhante ao page object apresentado no curso de Selenium do Dunossauro).

E para facilitar ainda mais, esses contextos que criam os recursos podem se tornar fixtures do pytest, assim bastaria informar o nome da fixture como parâmetro da função de teste que o recurso já seria criado, e assim que o teste terminar, removido.

Implementando esses padrões nos testes

Os padrões apresentados foram implementados numa biblioteca chamada pytest-moto-fixtures (também publicada no PyPI), assim é possível reutilizá-lo em diferentes testes. Para códigos que utilizam SQS, segue a implementação do objeto da fila SQS, contexto para criar fila SQS e fixture da fila SQS no repositório. Da mesma forma, para códigos que utilizam SNS, segue a implementação do objeto do tópico SNS, contexto para criar tópico SNS e fixture do tópico SNS.

Após instalar a biblioteca, o teste da função que recebe mensagens de uma fila SQS e publica a resposta em um tópico SNS fica da seguinte forma:

import json
from exemplo.codigo2 import funcao2

def test_funcao2(sqs_queue, sns_topic):
    # Mensagens com valores de entrada e resultados esperados
    msgs = [
        {'n': 0},
        {'n': 1},
        {'n': 2},
    ]
    respostas = [
        {'resultado': 0},
        {'resultado': 2},
        {'resultado': 4},
    ]

    # Envia mensagens de teste
    for msg in msgs:
        sqs_queue.send_message(body=msg)

    # Chama função a ser testada
    funcao2(sqs_queue.client, sns_topic.client, sqs_queue.url, sns_topic.arn)

    # Verifica se a função buscou e apagou as mensagens na fila do SQS
    assert len(sqs_queue) == 0

    # Verifica se as respostas estão corretas no tópico SNS
    assert len(sns_topic) == len(respostas)
    assert [
        json.loads(msg['Message']) for msg in sns_topic
    ] == respostas

Esse teste ficou muito mais simples, focando no que realmente importa para validar se o código está se comportando como deveria, e se aproveita de recursos do Python, como a função len e compreensões de lista. Toda a criação da fila SQS e tópico SNS se resumiu em adicionar as fixtures sqs_queue e sns_topic como argumentos da função de teste, assim como sua remoção posteriormente.

Quando precisa de múltiplos recursos do mesmo tipo

O teste anterior foi simples porque precisou de apenas um recurso de cada tipo. Porém se voltar a primeira função, aquela que recebe mensagens de uma fila SQS e envia o resultado para outra fila SQS, é necessário duas filas, mas a biblioteca entrega apenas uma fixture que cria uma única fila SQS. Uma opção seria criar outras fixtures para criar outras filas, porém quantas? Se criar poucas fixtures continuaria faltando filas, se criar muitas elas seriam carregadas sem necessidade.

Outra opção é dar um passo a traz, usando diretamente os contextos gerenciados em vez usá-los através de fixtures. O código de teste fica um pouco mais poluído, com comandos para criar as filas, porém possibilita uma maior flexibilidade também, como: Criar mais de uma fila para o teste; Criar filas com nomes específicos (ou que o nome siga um determinado padrão); E controlar quando a fila será criada e removida (ao sair do contexto). Segue um exemplo do teste:

import json
from exemplo.codigo1 import funcao1
from pytest_moto_fixtures.services.sqs import sqs_create_queue

def test_funcao1(sqs_client):
    # Mensagens com valores de entrada e resultados esperados
    msgs = [
        {'n': 0},
        {'n': 1},
        {'n': 2},
    ]
    respostas = [
        {'resultado': 0},
        {'resultado': 2},
        {'resultado': 4},
    ]

    # Cria filas de teste
    with (
        sqs_create_queue(sqs_client=sqs_client) as fila1,
        sqs_create_queue(sqs_client=sqs_client) as fila2,
    ):
        # Envia mensagens de teste
        for msg in msgs:
            fila1.send_message(body=msg)

        # Chama função a ser testada
        funcao1(sqs_client, fila1.url, fila2.url)

        # Verifica se a função buscou e apagou as mensagens na fila1
        assert len(fila1) == 0

        # Verifica se as respostas estão corretas na fila2
        assert len(fila2) == len(respostas)
        assert [
            json.loads(msg['Body']) for msg in fila2
        ] == respostas

Considerações finais

Esse texto apresentou uma sequência de passos e pensamentos que levaram até a criação da biblioteca pytest-moto-fixtures para auxiliar na escrita de testes de código que utilizam serviços da AWS. Também apresentou como sua arquitetura foi surgindo das necessidades, sempre com o foco em simplificação, e como resolveu essas necessidades.

A biblioteca também possui algumas facilidades que deixaram o código de teste bastante simples, como o uso das fixtures para criação de recursos. Junto com o uso das funcionalidades da própria linguagem, isso permitiu um código de teste bastante simples e direto, sem precisar expor detalhes dos serviços utilizados. E para coisas mais complexas ou específicas, a biblioteca também permite utilizar manualmente algumas de suas camadas mais profundas, porém ainda assim de forma simplificada, sem sujar o código tanto quando sua versão sem o uso da biblioteca.

Outra lição que fica é a possibilidade de criarmos bibliotecas para facilitar as coisas repetitivas do dia a dia, e não só usar as feitas por terceiros. Porém isso deve ser feito com estudo para realmente simplificar, caso contrário poderia trazer mais ou outras dificuldades para as atividades. Nessa forma, essa biblioteca seguiu a linha voltada para os testes de integração, outra biblioteca poderia surgir seguindo a ideia de uso de mocks, possuindo características diferentes.

Obs: O projeto com os códigos desse artigo pode ser acessado aqui.

Discussão sobre o Advent of Code 2023 - Dia 6: Modelando problema com matemática

Eduardo Klosowski — Wed, 13 Dec 2023 00:41:20 +0000

O problema do dia 6 do Advent of Code desse ano pode ser resolvido usando matemática. Então resolvi testando o SymPy e ver o quanto ele facilita nas análises e contas. O resultado encontra-se no seguinte notebook.

Discussão sobre o Advent of Code 2022 - Dia 11: Complexidade de operações com números grandes

Eduardo Klosowski — Sun, 19 Mar 2023 20:31:19 +0000

No décimo primeiro dia do Advent of Code de 2022 tem um problema interessante para discutir sobre a complexidade de operações com números grandes.

O problema do dia 11

O problema do dia 11, "macaco no meio", consiste em seguir uma sequência de passos, que pode ser implementada em código, para simular o ocorrido e calcular a resposta. Recomendo tentar resolvê-lo primeiro.

Resolução da parte 1

Não existe nenhuma pegadinha na parte 1 do problema. Talvez a parte mais difícil possa ser ler a entrada, porém como o programa não precisa ser dinâmico, os dados dessa entrada podem estar fixos dentro do código, não sendo necessário ler o arquivo da entrada, simplificando o código. No demais é escrever um código que itere sobre os valores e tome as ações conforme descrito no problema.

Resolução da parte 2

A parte 2 é igual à parte 1, mas com duas diferenças: o valor de preocupação com o item não é divido por 3 e devem ser calculado mais rodadas (passando de 20 para 10.000). Esses ajustes são simples de fazer no código, porém na hora de executar o programa ele fica calculando e não termina. O gráfico a baixo mostra o tempo que a solução que implementei leva para calcular cada rodada até a 135 no meu computador:

No gráfico pode ser visto que as primeiras rodadas tiveram um tempo relativamente baixo, porém com o passar do tempo, aparecem picos cada vez mais altos, e os vales também estão subindo, ou seja, o tempo para se calcular uma nova rodada está crescendo conforme as rodadas foram passando. A pergunta que fica é porque isso ocorre se as operações são as mesmas e a quantidade de itens também não varia? O que está deixando o código mais lento?

Bibliotecas para números grandes

Como eu fiz o meu programa em Python esse foi o comportamento do meu código, porém em outras linguagens poderia dar erro, ou até rodar relativamente rápido, mas apresentar resultados incorretos. Isso ocorre porque, como um valor manipulado pelo programa cresce, ele deixa de caber no espaço de uma variável de alguns bits (normalmente 16, 32, ou até 64 bits), ocorrendo um overflow de inteiro. Uma forma de lidar com isso é utilizando bibliotecas que consigam alocar mais espaço para as variáveis, e no Python, por exemplo, isso já é implementado nativamente. Mas por que fica cada vez mais lento?

A pergunta que pode esclarecer o que está ocorrendo é: qual a complexidade de tempo para operações básicas (soma, subtração, multiplicação e divisão)? Para o tipo inteiro essas operações têm tempo constante (O(1)), porém quando o valor deixa de caber num inteiro, usando o espaço de memória de múltiplos inteiros, essa operação deixa de ser contante. Imagine fazer a soma 139 + 183 manualmente, é possível começar somando as unidades (9 + 3), depois as dezenas (3 + 8 + 1, considerando o 1 a dezena da soma nas unidades), depois as centenas. Dessa forma é possível observar que quanto mais casas os números tiverem, mas operações precisam ser feitas, assim essa soma tem complexidade linear a quantidade de casas do valor (O(n)). E embora o computador use os números em binário, essa lógica se repete para as bibliotecas que lidam com números grandes, mesmo operando um ou até quatro bytes por vez, só muda a base numérica. Então pelo fato dos valores não caberem mais em um número pequeno de bits a complexidade do algoritmo aumenta junto com o aumento dos valores.

Reduzindo o valor a ser tratado

Porém ainda é necessário resolver a parte 2, e embora essas bibliotecas permitam com que o código execute corretamente, elas não fazem o mesmo rodar em tempo hábil. Então como isso pode ser feito? Aqui pode ser utilizado um pouco de matemática para evitar que os números cresçam de mais. Como o valor da preocupação utilizado no algoritmo não importa de fato, e sim o resto da divisão dele por outro número, isso pode ser utilizado para reduzir o valor da preocupação que o algoritmo precisa lidar, uma vez que os valores dessa operação ficam dentro de uma faixa menor e se repetem, mesmo que cresçam até o infinito, isso permite descartar parte da preocupação. Exemplo:

0 % 4 == 0
1 % 4 == 1
2 % 4 == 2
3 % 4 == 3
4 % 4 == 0
5 % 4 == 1
6 % 4 == 2
7 % 4 == 3
8 % 4 == 0
9 % 4 == 1
...
39 % 4 == 3
40 % 4 == 0
41 % 4 == 1
...
4000 % 4 == 0
...

Porém a lógica apresentada considera apenas um único número (o 4 nesse exemplo). Para que isso possa ser aplicada para dois ou mais números é necessário encontrar um valor que possa dividir a preocupação de forma que esse ciclo continue se repetindo normalmente, da mesma forma que ocorreu no exemplo anterior. Uma forma simples de fazer isso é multiplicando todos as bases das divisões que o algoritmo precisa fazer, e toda vez que uma nova preocupação for calculada, basta guardar na variável o resto da divisão dela por esse valor calculado.

Dessa forma os valores da preocupação não crescem até o infinito, e a parte 2 é calculada rapidamente.

Considerações

Um algoritmo pode ter sua complexidade alterada apenas por ter que lidar com inteiros que podem estar além de uma quantidade fixa de bits. E não é eficiente aumentar essa quantidade de bits para todos os números, visto que se o processador não conseguir operar todos de uma vez, terá que fazer diversas operações para chegar no resultado, isso deixaria o código mais lento em sua execução e precisaria de mais memória, mesmo que os números não usem todos os bits disponíveis para a variável.

Nesse texto foi considerado a multiplicação dos valores da base das divisões como fator para reduzir os valores da preocupação. Porém nem sempre ele será o valor mais otimizado, para ter a garantia de optimização desse fator pode ser utilizado o mínimo múltiplo comum desses valores. Porém como no caso da minha entrada todos os valores eram primos, o mínimo múltiplo comum e a multiplicação dos valores da o mesmo resultado, não alterando o desempenho do código. E ainda sobre esse assunto, é interessante como uma representação gráfica desse ciclo do resto das divisões pode ser:

Abakcus

@abakcus

Polygon laps. Beautiful geometry. ♥️ [bityl.co/GcOc]

16:10 PM - 11 Jan 2023

Discussão sobre o Advent of Code 2022 - Dia 10: Divisão de responsabilidades em geradores

Eduardo Klosowski — Mon, 06 Mar 2023 02:23:04 +0000

No décimo dia do Advent of Code de 2022 tem um problema interessante para discutir sobre divisão de responsabilidades de funções geradoras.

O problema do dia 10

O problema do dia 10, "tubo de raios catódicos", consiste em implementar um emulador simples. Na parte 1 deve ser feito a decodificação e execução de instruções de um processador, enquanto na parte 2 deve-se implementar uma saída de vídeo. Recomendo tentar resolvê-lo primeiro.

Resolução da parte 1

A parte 1 pede para ler as instruções da entrada, executá-las considerando a quantidade de ciclos que cada tipo de instrução leva, e em determinados ciclos fazer um calculo para obter a resposta, guardando o resultado em um acumulador. Nessa descrição podem ser identificado três responsabilidades diferentes: ler e tratar as instruções da entrada; controlar a execução das instruções no processador e seus ciclos necessários; e controlar o ciclo atual, fazendo o cálculo do valor desejado. Uma forma de escrever funções separando essas responsabilidades é utilizando geradores do Python, que como eles permitem executar um trecho de código, parar e voltar depois, isso permite com que cada função foque apenas na sua responsabilidade, sem precisar misturá-las.

A função para tratar a entrada pode ser feita como um gerador que lê um arquivo e retorna cada instrução com seus valores já convertidos (quando a instrução possui valores extras). No caso da instrução noop, seria apenas a informação que é uma instrução NOOP. Enquanto a instrução addx tem também um valor inteiro a ser somado no registrador x. Segue um exemplo onde cada valor retornado pelo gerador é uma lista contendo a instrução na primeira posição e o inteiro associado a ela, quando presente, na segunda posição:

def ler_instrucoes(entrada):
    for linha in entrada:
        instrucao = linha.strip().split(' ')
        if len(instrucao) >= 2:
            instrucao[1] = int(instrucao[1])
        yield instrucao

Esse gerador permite com que a entrada possa ser carregada e tradada conforme as instruções forem sendo executadas, não sendo necessário carregar tudo para a memória, nem tratar toda a entrada primeiro, antes de seguir para a próxima parte do problema. Além de simplificar o processo para quem for executar as operações, onde não será necessário se preocupar com conversão de tipos, por exemplo.

A execução das instruções também pode ser feita através de um gerador. Como um gerador permite executar um trecho de código e parar, isso pode ser utilizado para simular os ciclos do processador, onde cada valor retornado pelo gerador (yield) representar o estado no processador naquele ciclo, que nesse problema se resume ao valor do registrador x, que pode ser implementado como uma variável local da função geradora. Segue um exemplo de código:

def processador(instrucoes):
    x = 1
    for instrucao in instrucoes:
        if instrucao[0] == 'noop':
            yield x
        elif instrucao[0] == 'addx':
            yield x
            yield x
            x += instrucao[1]

Essa função itera sobre um conjunto de instruções recebida como argumento (gerador apresentado anteriormente) e executa essas instruções. No caso de instruções noop, nada é feito, porém o yield x faz com que o gerador pare nesse ponto, sendo necessário a leitura do próximo valor do gerador do processador para a continuação da execução desse código, simulando o ciclo do processador. Já as instruções addx, como possuem dois yield x, faz com que seja necessário mais leitura de valores para que finalmente o valor da instrução seja adicionado ao registrador x, simulando o processador parado dois ciclos, para só no final deles a instrução ser completada.

Com a decodificação das instruções e o processador feitos, basta instanciar um gerador da função ler_instrucoes e um da função processador, passando o primeiro como argumento para o segundo. Como o segundo gerador é o responsável por iterar sobre o primeiro (ele faz um for nesse iterador), basta iterar sobre o segundo para simular os ciclos e execução das instruções, e ao utilizar um enumerate é possível saber em que ciclo a execução se encontra e realizar as operações desejadas para se calcular o valor pedido pelo problema. Segue um exemplo de código:

p = processador(ler_instrucoes(open('entrada.txt')))
total = 0
for i, x in enumerate(p, start=1):
    if (i - 20) % 40 == 0:
        total += i * x
    if i == 220:
        break
print(total)

Resolução da parte 2

Entendido como decodificar as instruções e controlar os ciclos para a execução das mesmas, a parte 2 pede para simular uma tela de tubo de raios catódicos. Essa tela possui algumas regras, como mostrar um pixel por vez, e a cada 40 pixeis, ir para a linha a baixo, reiniciando o processo. Isso também pode ser implementado como um gerador, encapsulando essas regras, porém em vez de retornar valores, ele deve receber valores, o que também é chamado de consumidor, mas tem a mesma sintaxe, com a diferença que em vez de usar yield x para retornar um valor x, é usado x = yield para guardar numa variável o valor recebido, e ainda continua possuindo a função de parar a execução do código, nesse caso aguardando o próximo valor. Segue um exemplo de implementação:

def crt():
    while True:
        for i in range(40):
            x = yield
            if x - 1 <= i <= x + 1:
                print('\u2588', end='')
            else:
                print(' ', end='')
        print()

Essa função é um loop infinito, que percorre 40 valores imprimindo na tela um bloco preenchido ou um espaço (fiz essa mudança uma vez que facilitou a visualização do resultado), conforme o valor do registrador x recebido, e após isso faz uma quebra de linha.

Como as funções ler_instrucoes e processador já estão implementadas, e não é necessário nenhuma alteração em suas lógicas, basta acoplar o consumidor crt com elas para ter o resultado desejado. Porém como ao criar um gerador, nenhum código do mesmo é executado até que seu primeiro valor seja lido, é necessário instanciar o consumidor crt e fazer uma primeira execução para que o mesmo pare no yield e esteja pronto para receber o primeiro valor, o que pode ser feito utilizando a função next, assim como é possível fazer para recuperar um valor do gerador. Segue um exemplo dessa implementação:

tela = crt()
next(tela)
for x in processador(ler_instrucoes(open('entrada.txt'))):
    tela.send(x)

Considerações

A utilização de geradores permitiu a implementação do código separando responsabilidades em diferentes funções, além de encapsular regras de controle do processador numa função e da tela em outra, sem misturar as duas coisas, deixando-as desacopladas e que poderiam ser utilizadas separadamente. E embora as lógicas estejam separadas, a execução intercala elas, como se o código fosse um loop que lê uma instrução, executado ela e mostrado na tela o caractere resultante, para então repetir essas ações, o que permite ser eficiente em relação ao uso de memória, como já comentado anteriormente nessa série.

Geradores também são a forma como corrotinas ou funções assíncronas eram feitas em Python antes deles entrarem como um recurso da linguagem, recomendo a série "Geradores e uma Introdução histórica à corrotinas com Python" do Dunossauro sobre o assunto.

Discussão sobre o Advent of Code 2022 - Dia 8: Valor da constante de um algoritmo linear

Eduardo Klosowski — Wed, 15 Feb 2023 02:42:54 +0000

No oitavo dia do Advent of Code de 2022 tem um problema interessante para discutir sobre o termo constante na complexidade de um algoritmo.

O problema do dia 8

O problema do dia 8, "casa nas copas das árvores", consiste em analisar as alturas das árvores e contá-las. Na parte 1 isso deve ser feito olhando de fora da floresta, enquanto na parte 2 deve-se olhar do topo de cada árvore. Recomendo tentar resolvê-lo primeiro.

Resolução da parte 1

A parte 1 do problema pergunta quantas árvores são visível de fora da floresta. É possível resolvê-lo com um algoritmo de complexidade linear ao número de árvores, e uma das formas de fazer isso é analisando a partir de cada orientação (norte, sul, leste e oeste) guardando quais foram as árvores visíveis para não contar a mesma árvore mais de uma vez. Isso faz com que todas as árvores sejam analisadas quatro vezes (uma vez a partir de cada direção), mas mesmo percorrendo todas as árvores quatro vezes, não tira a característica linear do algoritmo, uma vez que a complexidade O(4*n) é considerado igual a O(n), só mudando a constante que multiplica a complexidade, e essa constante normalmente é omitida na notação.

Um exemplo de implementação desse algoritmo pode ser visto a baixo, onde é possível percorrer todas as direções mexendo nos laços das variáveis x e y:

entrada = [[int(i) for i in line.rstrip()] for line in open(0)]

largura = len(entrada[0])
altura = len(entrada)

arvores_visiveis = set()

# Análise do oeste para leste
for y in range(altura):
    maior_tamanho = -1
    for x in range(largura):
        tamanho = entrada[y][x]
        if tamanho > maior_tamanho:
            arvores_visiveis.add((x, y))
            maior_tamanho = tamanho

# Análise do leste para o oeste
for y in range(altura):
    maior_tamanho = -1
    for x in reversed(range(largura)):
        tamanho = entrada[y][x]
        if tamanho > maior_tamanho:
            arvores_visiveis.add((x, y))
            maior_tamanho = tamanho

# Análise do norte para o sul
for x in range(largura):
    maior_tamanho = -1
    for y in range(altura):
        tamanho = entrada[y][x]
        if tamanho > maior_tamanho:
            arvores_visiveis.add((x, y))
            maior_tamanho = tamanho

# Análise do sul para o norte
for x in range(largura):
    maior_tamanho = -1
    for y in reversed(range(altura)):
        tamanho = entrada[y][x]
        if tamanho > maior_tamanho:
            arvores_visiveis.add((x, y))
            maior_tamanho = tamanho

print(len(arvores_visiveis))

Ainda seria possível fazer uma otimização, como verificar se o maior tamanho de árvore já foi encontrado, o que impediria de encontrar outra árvore visível naquela linha ou coluna, e nesse caso o laço mais interno poderia ser interrompido (if maior_tamanho == 9: break). Isso pode reduz o número de análises para alguns problemas, mas não reduz a complexidade do algoritmo, que continua sendo linear, até porque essa otimização não vai conseguir melhorar o desempenho para todas as entradas, um caso é uma entrada que não tenha árvores de altura 9, por exemplo.

Também é possível utilizar outra matriz para guardar as árvores visíveis em vez do conjunto utilizado no exemplo (set()), iniciando todas as árvores como não visíveis e quando se encontra uma visível, troca-se seu estado. E para saber quantas árvores são visíveis bastaria percorrer essa segunda matriz para contá-las, o que faria ser necessário percorrer a matriz mais uma vez, porém ainda manteria a complexidade linear do algoritmo.

Resolução da parte 2

Na parte 2 do problema, a visibilidade deve ser feita a partir de cada árvore. Uma implementação simples disso é olhar as quatro direções para cada árvore. Considerando l a quantidade de árvores de leste para oeste e h de note para sul, a complexidade dessa solução simples seria O(n*l) ou O(n*h), o que for maior. Porém existe um algoritmo capaz de resolver esse problema de forma linear (O(n)).

Para resolver esse problema de forma linear são necessários ajustes na estratégia da parte 1, uma vez que a mudança do referencial pode fazer com que árvores que não eram visíveis antes se tornem visíveis, ou que eram visíveis não sejam mais. Uma estratégia é considerar todas as alturas possíveis para cada árvore que está sendo analisada, somando 1 no contador dessa possível altura caso seja visível, ou reiniciando o contador caso a visão esteja obstruída, e na hora de analisar uma direção, fazer isso de trás para frente. Assim é possível dizer para cada árvore sua visão naquela direção, e aproveitar esses dados para o cálculo de visão de cada próxima árvore. E para o cálculo da pontuação de cada árvore, como é a multiplicação da visibilidade nas quatro direções, pode-se iniciar com pontuação 1 para cada árvore e ir multiplicando conforme se calcula a visibilidade nas direções.

Um exemplo desse algoritmo pode ser visto a baixo:

entrada = [[int(i) for i in line.rstrip()] for line in open(0)]

largura = len(entrada[0])
altura = len(entrada)
tamanhos = 10

pontuacoes = [[1 for _ in range(largura)] for _ in range(altura)]

for y in range(altura):
    tamanho_anterior = -1
    visibilidade_anterior = [-1 for _ in range(tamanhos)]
    for x in range(largura):
        tamanho = entrada[y][x]
        visibilidade = []
        for i in range(tamanhos):
            if i > tamanho_anterior:
                visibilidade.append(visibilidade_anterior[i] + 1)
            else:
                visibilidade.append(1)
        pontuacoes[y][x] *= visibilidade[tamanho]
        tamanho_anterior = tamanho
        visibilidade_anterior = visibilidade

for y in range(altura):
    tamanho_anterior = -1
    visibilidade_anterior = [-1 for _ in range(tamanhos)]
    for x in reversed(range(largura)):
        tamanho = entrada[y][x]
        visibilidade = []
        for i in range(tamanhos):
            if i > tamanho_anterior:
                visibilidade.append(visibilidade_anterior[i] + 1)
            else:
                visibilidade.append(1)
        pontuacoes[y][x] *= visibilidade[tamanho]
        tamanho_anterior = tamanho
        visibilidade_anterior = visibilidade

for x in range(largura):
    tamanho_anterior = -1
    visibilidade_anterior = [-1 for _ in range(tamanhos)]
    for y in range(altura):
        tamanho = entrada[y][x]
        visibilidade = []
        for i in range(tamanhos):
            if i > tamanho_anterior:
                visibilidade.append(visibilidade_anterior[i] + 1)
            else:
                visibilidade.append(1)
        pontuacoes[y][x] *= visibilidade[tamanho]
        tamanho_anterior = tamanho
        visibilidade_anterior = visibilidade

for x in range(largura):
    tamanho_anterior = -1
    visibilidade_anterior = [-1 for _ in range(tamanhos)]
    for y in reversed(range(altura)):
        tamanho = entrada[y][x]
        visibilidade = []
        for i in range(tamanhos):
            if i > tamanho_anterior:
                visibilidade.append(visibilidade_anterior[i] + 1)
            else:
                visibilidade.append(1)
        pontuacoes[y][x] *= visibilidade[tamanho]
        tamanho_anterior = tamanho
        visibilidade_anterior = visibilidade

maior_pontuacao = 0
for y in range(altura):
    for x in range(largura):
        pontuacao = pontuacoes[y][x]
        if pontuacao > maior_pontuacao:
            maior_pontuacao = pontuacao
print(maior_pontuacao)

Esse algoritmo também percorre todas as árvores quatro vezes para calcular a visibilidade, porém toda vez que analisa uma árvore em uma determinada direção, o faz dez vezes, uma para cada tamanho possível, resultando em 40 análises por árvore (O(40*n)), além de percorrer todas as árvores uma vez a mais para encontrar a maior pontuação. Como a quantidade de tamanhos possíveis é fixa, não variando conforme a entrada do problema, isso não muda a complexidade do algoritmo, que se mantem linar (O(n)). Também seria possível calcular a maior pontuação junto do cálculo da visibilidade da última direção, isso removeria o último laço do algoritmo, porém só deslocaria as instruções da comparação da pontuação de um laço para o outro, cada instrução ainda precisaria ser executada (menos as instruções de controle do laço de repetição), não tendo grandes ganhos de desempenho e misturaria coisas diferentes no laço, o que o deixaria mais difícil de ler.

Outra coisa que pode ser observada nesse algoritmo é a quantidade de memória necessária para executá-lo. Ele cria uma cópia da matriz para guardar a pontuação calculada, e quando está calculando as direções, precisa manter um contador diferente para cada tamanho possível. E diferente da parte 1 onde é possível guardar só as árvores visíveis, o que reduz o consumo de memória ao usar um conjunto (set) para isso, na parte 2 é necessário guardar a pontuação parcial de cada árvore, uma vez que não é possível calcular a pontuação final de cada árvore, uma a uma, sem mexer na complexidade do algoritmo, mas isso permitiria guardar em memória apenas a maior pontuação encontrada até então, por exemplo.

Considerações

Foi possível encontrar algoritmos com complexidade linear para as partes desse problema, porém devido ao fato de ter uma quantidade de direções e tamanhos fixos para o problema. Se tivesse variações como para certas entradas incluir visibilidade nas diagonais também e para outras entradas não, ou mudar a diferença entre a maior e menor altura das árvores, isso teria que aparecer como uma variável multiplicando a complexidade do algoritmo, e não como uma constante, já que isso alteraria a quantidade de análises que precisariam ser feitas.

Como esses algoritmos têm complexidade linear de execução, isso os tornam uma boa opção para grandes entradas. Porém caso as entradas sejam pequenas, um algoritmo com complexidade superior poderia rodar mais rápido devido as grandes constantes que multiplicam a complexidade das soluções lineares. Isso deve ser levado em consideração na hora da escolha do algoritmo a ser utilizado para resolver o problema.

Para esses algoritmos foi necessário carregar toda a matriz em memória, uma vez que os dados eram lidos diversas vezes, e em diversas ordens. Assim não foi possível ler a entrada sob-demanda para reduzir o consumo de memória como nos problemas dos dias anteriores. Se o tamanho da entrada fosse muito grande, impossibilitando guardar tudo em memória, ainda seria possível guardar os dados em arquivos, calculando a posição do dado a ser lido ou escrito e usando funções como seek para ir direto para o dado desejado, mas considerando que o acesso ao HD ou SSD é mais lento que a memória RAM, isso deixaria a execução mais lenta. E nesse caso talvez outros algoritmos, mesmo com complexidades maiores poderiam lidar melhor com esse problema.

Discussão sobre o Advent of Code 2022 - Dia 6: Otimizando o algoritmo e utilização de estruturas de dados

Eduardo Klosowski — Mon, 12 Dec 2022 16:40:47 +0000

O sexto dia do Advent of Code deste ano tem um problema interessante para discutir sobre otimização do algoritmo e utilização de estruturas de dados.

O problema do dia 6

O problema do dia 6 "problema de sintonização" consiste em encontrar uma subsequência onde não existam elementos repetidos. Recomendo tentar resolvê-lo primeiro.

Solução simples

Uma forma simples de resolver os problemas no Advent of Code é encontrar soluções simples que não tenham grandes complicações. Para esse problema, uma solução baseada em força bruta, que consiste em testar todas as possíveis subsequências, atende esse requisito. Ela pode ser implementada percorrendo a entrada do problema, extraindo as subsequências e verificando se nelas todos os elementos são únicos, esse último pode ser feito utilizando a estrutura de conjuntos (comentado no problema do dia 3) verificando se o tamanho do mesmo é igual da subsequência, uma vez que a estrutura de conjuntos "remove" elementos duplicados. Exemplo:

from itertools import count

tamanho = 4
for i in count():
    subsequencia = entrada[i:i + tamanho]
    if len(set(entrada[i:i + tamanho])) == tamanho:
        # resposta encontrada

1ª Questão - Desempenho

A solução apresentada funciona, porém ela percorre toda a entrada, e para cada posição da entrada, precisa executar mais uma quantidade de operações igual ao tamanho da subsequência desejada (O(n * tamanho)). Isso pode deixar esse algoritmo cada vez mais lento conforme se deseja uma subsequência maior.

O ideal seria um algoritmo que conseguisse depender apenas do tamanho da entrada (O(n)), para isso é preciso desenvolver algum mecanismo para guardar quais caracteres estão na subsequência, verificar se o novo caractere lido está nela ou não e descartar o caractere que saiu da subsequência ao ler esse novo caractere da entrada, tudo isso em tempo constante (O(1)). Se pensarmos em guardar a posição que tal caractere apareceu pela última vez, é fácil responder se ele se encontra dentro ou já está fora da subsequência olhando pelo tamanho dela e posição do caractere lido. E ao guardamos a posição em que a última duplicidade foi encontrada é fácil determinarmos quantos caracteres foram lidos sem duplicidade, desta forma, ao atingir o tamanho desejado, a resposta é encontrada. Juntando esses pensamentos, temos o seguinte algoritmo:

tamanho = 4
posicao_das_letras = {}
ultima_duplicidade = -1
for i, c in enumerate(entrada):
    if posicao_das_letras.get(c, -1) > ultima_duplicidade:
        ultima_duplicidade = posicao_das_letras.get(c, -1)
    else:
        if i - ultima_duplicidade == tamanho:
            # resposta encontrada
    posicao_das_letras[c] = i

Considerando que a implementação de dicionário do Python utiliza uma tabela de espalhamento, e que as operações para guardar e recuperar um valor espera-se que ocorram em tempo constante (O(1)), temos como resultado um algoritmo em tempo linear (O(n)).

2ª Questão - Estrutura de dados

Porém tabelas de espalhamento podem ter problemas com colisão de hash, além de ter uma implementação relativamente complexa. Como nesse problema existe um conjunto relativamente restrito de chaves, 256 caracteres considerando todos os valores possíveis em 1 Byte (ou ainda 26 caracteres, considerando apenas as letras de a até z), e que esses valores podem ser facilmente interpretados como números, é possível utilizar uma simples lista no lugar do dicionário, usando a posição da lista como sua chave. Exemplo:

tamanho = 4
posicao_das_letras = [-1 for _ in range(256)]
ultima_duplicidade = -1
for i, c in enumerate(ord(c) for c in entrada):
    if posicao_das_letras[c] > ultima_duplicidade:
        ultima_duplicidade = posicao_das_letras[c]
    else:
        if i - ultima_duplicidade == tamanho:
            # resposta encontrada
    posicao_das_letras[c] = i

Desta forma é utilizado uma estrutura de dados muito mais simples, e obtendo o mesmo resultado. Essa abordagem também não vai ter problemas com colisão de hash, e nem problemas com o tempo de cálculo do mesmo, uma vez que não utiliza funções de hash, deixando o algoritmo menos suscetível a problemas.

Um outro desafio que observei onde a utilização de estruturas de dados mais simples é possível foi no 2376 da beecrowd, que se resume a montar uma chave de um campeonato, dizendo no final o vencedor. Por ter uma estrutura visual bastante semelhante a uma árvore, um primeiro impulso resolvê-lo implementando uma, porém olhando a sequência dos jogos, é possível resolver esse problema utilizando uma fila, e que pode ser até mais fácil de implementar uma vez que não é necessário se preocupar em que profundidade da árvore está se processando.

Considerações

Nem todo problema é possível resolver em tempo linear, e quando é, pode-se exigir um pouco mais de estudo sobre o problema e busca da melhor estrutura de dados.

Discussão sobre o Advent of Code 2022 - Dia 4: Algebra booliana e ordenação

Eduardo Klosowski — Thu, 08 Dec 2022 01:22:09 +0000

Quarto dia do Advent of Code deste ano, o problema desse dia é bastante interessante para discutir sobre álgebra booliana e ordenação completa ou parcial, e como ela é implementado em linguagens como Rust.

O problema do dia 4

O problema do dia 4 "limpeza do acampamento" consiste em ler uma sequência de intervalos e encontrar se eles se sobrepõem. Novamente, recomendo que tentem resolvê-lo primeiro, mas se tiver dificuldades pode seguir com a leitura desse texto.

1ª Questão - Álgebra booliana

A parte 1 do problema pede para encontrar sobreposições completas, algo como um dos dois casos da imagem a baixo, que mostra as bordas dos dois conjuntos em ordem numérica:

Visualizando a imagem é fácil chegar a conclusão de que a condição para uma sobreposição completa é (b[0] <= a[0] <= a[1] <= b[1]) || (a[0] <= b[0] <= b[1] <= a[1]) já que é a ordem em que elas aparecem na imagem.

A parte 2 do problema pode para encontrar sobreposições parciais também, novamente uma imagem pode ajudar:

Na imagem vemos as ordens (a[0] <= b[0] <= a[1] <= b[1]) || (b[0] <= a[0] <= b[1] <= a[1]). Pegando apenas o primeiro caso para analisar, nessa parte do problema condições como a[0] <= b[0] <= a[1] <= b[1] (sobreposição parcial) e a[0] <= b[0] <= b[1] <= a[1] (sobreposição completa) são as procuradas. Analisando um pouco, pode-se observar que b[1] não importa tanto para essa condição, já que ele pode estar tanto antes quanto depois de a[1], sendo assim, pode-se removê-lo da condição: a[0] <= b[0] <= a[1]. E isso se repete para o segundo caso, chegando na condição final (a[0] <= b[0] <= a[1]) || (b[0] <= a[0] <= b[1]).

Porém essa é apenas uma das formas a se chegar a condição. Outra forma que pode ser interessante para a parte 2 é pensar o oposto, o que seria uma não sobreposição:

Esse caso é muito mais simples, bastando observar que se a[1] < b[0] || b[1] <= a[0] não existe nenhum tipo de sobreposição, e para encontrar uma sobreposição basta inverter essa condição, negando-a: !(a[1] < b[0] || b[1] < a[0]).

Ainda existe a possibilidade de trabalhar com essa condição utilizando álgebra booliana para remover a negação, (eu já comentei sobre isso nesse texto, recomendo a leitura), e isso permite chegar na condição a[1] >= b[0] && b[1] >= a[0], que é mais simples do que (a[0] <= b[0] <= a[1]) || (b[0] <= a[0] <= b[1]) que havia sido encontrado anteriormente.

2ª Questão - Ordenação

Em toda essa discussão assumi que a[0] <= a[1] e b[0] <= b[1], e que o problema nos garante isso (pelo menos não existe caso onde isso não seja válido na entrada dada). Isso permite que não seja necessário verificar se o intervalo está invertido, e assumir que se a < b e b < c então a < c, uma vez que está lidando com números inteiros que também tem essa propriedade.

Voltando ao problema do "pedra papel tesoura" do dia 2, também poderia se utilizar das operações de maior e menor para dizer quem ganha ou perde de quem, exemplo: pedra < papel e papel < tesoura, porém observe que isso não nos permite assumir que pedra < tesoura, visto que o correto seria tesoura < padra. Quando isso ocorre, temos um conjunto parcialmente ordenado.

Embora isso não faça diferença para a resolução do problema do Advent of Code, permite entender a diferença entre as traits Ord e PartialOrd do Rust, onde se quiser comparar valores com > e <, deve-se implementar a PartialOrd, e se for seguro assumir que se a < b e b < c então a < c pode-se implementar também a Ord, e essa é a diferença entre elas. Isso também ocorre de forma análoga para as comparações com as traits Eq e PartialEq (se a == b e b == c então a == c, pode-se implementar também a Eq). Apesar de não fazer parte do problema, fica como curiosidade e conhecimento.

Considerações

Se estiver difícil de escrever ou entender alguma condição, às vezes vale a pena pensar o oposto e negá-la, sendo opcional o trabalho com a álgebra booliana para tentar reduzir a quantidade de operações.

Quando estiver trabalhando com comparações (maior, menor, igualdade, diferença) vale a pena observar se ela é uma comparação apenas parcial, ou se pode-se assumir outras condições a partir do que já foi verificado. Algoritmos de ordenação, por exemplo, tiram bastante proveito disso para reduzir seus tempos de execução. Quando aplicado, pode ser algo trabalhado no algoritmo para buscar melhorar sua performance.

Discussão sobre o Advent of Code 2022 - Dia 3: Complexidade de operações em estruturas de dados

Eduardo Klosowski — Tue, 06 Dec 2022 18:09:40 +0000

Terceiro dia do Advent of Code deste ano, o problema desse dia é bastante interessante para discutir sobre a complexidade de operações em estruturas de dados.

O problema do dia 3

O problema do dia 3 "reorganização da mochila" consiste em encontrar elementos repetidos em diferentes coleções de valores. Novamente, recomendo que tentem resolvê-lo primeiro, e o vídeo do Bruno Rocha:

Questão - Estrutura de dados e complexidade de operações

Uma forma bastante simples de representar os compartimentos das mochilas na primeira parte, ou as mochilas na segunda parte, é com uma lista ou array, onde cada posição é um item. E isso é bastante direto e prático, já que muitas linguagens permitem acessar o caractere de uma posição da string, fazendo a string já se comportar como uma lista de certa forma.

Para encontrar os itens em comum em duas strings bastaria comparar todos seus elementos. Exemplo:

itens_em_comum = []
for a in compartimento1:
    for b in compartimento2:
        if a == b:
            itens_em_comum.append(a)

Isso ainda pode ser simplificado para o código a baixo:

for a in compartimento1:
    if a in compartimento2:
        itens_em_comum.append(a)

Apensar da mudança, o código vai executar exatamente os mesmos passos, que são comparar cada item do compartimento1 com cada item do compartimento2. Isso faz com que sejam necessárias n * m comparações, sendo n a quantidade de itens no compartimento1 e m a quantidade de itens no compartimento2, que como na parte 1 os dois compartimentos têm a mesma quantidade de itens (n = m), é possível afirmar que são n² comparações (O(n²)).

Porém listas não são a única estrutura de dados que poderiam representar os compartimentos. Como nesse problema itens repetidos no mesmo compartimento e sua ordem não são importantes, é possível utilizar conjuntos (conhecidos como set em algumas linguagens de programação) e que podem ser implementados com uma árvore de busca binária ou tabela de espalhamento. Exemplo:

compartimento2 = set(compartimento2)
for a in compartimento1:
    if a in compartimento2:
        itens_em_comum.append(a)

Uma vantagem de se utilizar um conjunto é que itens repetidos são removidos, ou seja, uma mesma comparação não é feita com outra cópia do mesmo item. Outra vantagem é que a operação para verificar se um item existe em outro conjunto, quando implementado em uma árvore de busca binária, reduz o espaço de busca pela metade a cada comparação, sendo assim aproximadamente n * log2(n) comparações para se encontrar todos os itens repetidos, que é menor que n².

Entretanto, enquanto numa lista, para adicionar um item basta copiá-lo no final da mesma (O(1)), para adicionar um item em uma árvore de busca binária é necessário encontrar a posição para aquele item na árvore (O(log2(n))), e isso se repete para cada item, além de as vezes ser necessário rebalancear a árvore. Resumidamente existe algum processamento para se construir a árvore, e isso deve ser levado em consideração. Porém como a construção da árvore tem a mesma complexidade da busca que ocorre depois (O(n * log2(n))), faz compensar para uma grande quantidade de itens, mas pode não compensar para uma baixa quantidade, veja a baixo um gráfico comparando a quantidade de comparações para ambos os casos.

Considerações

Uma mesma operação pode ter complexidades diferentes para diferentes estruturas de dados, como no caso a função para verificar se um item existe em outra coleção (O(n²) na lista, e O(n * log2(n)) no conjunto). Porém além da operação, pode ser necessário considerar a construção da estrutura de dados, que neste caso compensa para muitos itens, mas pode não compensar para quantidades pequenas.

Ainda existem outras opções para resolver esse problema, como converter os dois compartimentos para conjuntos e utilizar funções que busquem a interseção dos mesmos (o que foi feito pelo Bruno Rocha nos seus vídeos). Porém isso obriga a construção de mais um conjunto, e se o algoritmo não tirar vantagem dessa estrutura de dados, como percorrer os itens de uma coleção buscando se existe na outra, que é o que ocorre usando uma lista para um compartimento e um conjunto para outro, isso pode não mudar a complexidade ciclomática do algoritmo e deixa o código mais lento.

Também pode ser analisado a complexidade ao utilizar uma tabela de espalhamento para implementar os conjuntos (o Python faz isso), que traria outra complexidade para essas operações, podendo ter ou não um melhor desempenho, como já discutido no dia 2.

Ou ainda trabalhando com as listas, poderia ser ordenados os itens, o que permitiria fazer uma busca binária na lista (recomendo esse vídeo sobre o assunto), isso não removeria os itens duplicados no processamento, porém visto a complexidade O(n * log2(n)) dos algoritmos de ordenação e O(log2(n)) da busca binária, teria uma eficiência do algoritmo similar as outras opções analisadas (O(n * log2(n))).

Discussão sobre o Advent of Code 2022 - Dia 2: Sequência de condições

Eduardo Klosowski — Fri, 02 Dec 2022 20:58:41 +0000

Segundo dia do Advent of Code deste ano, na questão de optimização do algoritmo, ele tem bastante semelhança com o dia 1 sobre tratar a entrada, mas tem uma questão que acredito que vale uma observação no seu processamento.

O problema do dia 2

O problema do dia 2 "pedra papel tesoura" consiste basicamente em transcrever as regras do jogo de mesmo nome para um algoritmo que processe seus resultados. Novamente recomendo que tentem resolver o desafio primeiro, e o vídeo do Bruno Rocha:

Questão - Sequência de condições

Na solução apresentada pelo Bruno Rocha foi utilizado o match do Rust para definir a pontuação ganha em cada rodada. Nem todas as linguagens têm essa estrutura de controle (ou um switch ... case que poderia substituí-lo em alguns casos), mas é possível fazer algo similar utilizando if. Exemplo:

if oponente == 'A' and voce == 'X':
    score += 0
elif oponente == 'A' and voce == 'Y':
    score += 0
elif oponente == 'A' and voce == 'Z':
    score += 0
elif oponente == 'B' and voce == 'X':
    score += 0
elif oponente == 'B' and voce == 'Y':
    score += 0
elif oponente == 'B' and voce == 'Z':
    score += 0
elif oponente == 'C' and voce == 'X':
    score += 0
elif oponente == 'C' and voce == 'Y':
    score += 0
elif oponente == 'C' and voce == 'Z':
    score += 0

Um ponto dessa abordagem é que para resultados que tem sua condição verificada mais no início, como A X, tendem a ser computados mais rápido que condições verificadas mais para o final, como C Z. Dependendo do ambiente, se confidencialidade for importante, por exemplo, medir o tempo de cálculo poderia vazar quais foram as opções escolhidas, quebrando a confidencialidade (mas não é o caso aqui). Isso pode ser contornado trocando todos os elif para if, o que faria todas as opções ficarem mais lentas iguais, já que toda vez todas as condições seriam verificadas.

Mas considerando reduzir o tempo de execução, usar if aninhados é uma outra abordagem possível. Exemplo:

if oponente == 'A':
    if voce == 'X':
        score += 0
    elif voce == 'Y':
        score += 0
    elif voce == 'Z':
        score += 0
elif oponente == 'B':
    if voce == 'X':
        score += 0
    elif voce == 'Y':
        score += 0
    elif voce == 'Z':
        score += 0
elif oponente == 'C'
    if voce == 'X':
        score += 0
    elif voce == 'Y':
        score += 0
    elif voce == 'Z':
        score += 0

Assim, em vez de ter que passar por 9 condições até chegar na opção C Z (pior caso), seriam necessário apenas 6 condições, sendo que as demais combinações também tem ganhos.

Outra abordagem em vez de usar if, como o Bruno comentou, seria utilizando estruturas como mapas ou dicionários (o nome varia de acordo com a linguagem). Exemplo:

scores = {
    ('A', 'X'): 0,
    ('A', 'Y'): 0,
    ('A', 'Z'): 0,
    ('B', 'X'): 0,
    ('B', 'Y'): 0,
    ('B', 'Z'): 0,
    ('C', 'X'): 0,
    ('C', 'Y'): 0,
    ('C', 'Z'): 0,
}

score += scores[oponente, voce]

O desempenho dessa solução depende da estrutura de dados utilizada para implementar o mapa/dicionário. Se ele for implementado em cima de uma árvore de busca binária (algo com alguma semelhante a uma BTreeMap do Rust) teria um desempenho semelhante a solução de if anilhados, se for implementado em cima de uma tabela de espalhamento (estrutura HashMap do Rust) se resumiria a calcular um hash em cima dos dados e acessar diretamente o valor desejado, o que poderia ser um desempenho ótimo para todas as condições, só dependendo da eficiência do cálculo da hash.

Olhando para a questão do Rust agora, é possível que BTreeMap tenha um desempenho melhor do que o HashMap, por ser poucos dados e não precisar chamar a função de hash. E sobre o match, não sei como ele foi implementado na linguagem, se ele seguiria uma ordem sequencial, como na primeira abordagem mostrada, ou se conseguiria fazer alguma otimização como no if aninhado. Porém para as poucas condições do problema a diferença no tempo seria mínima.

Considerações

Nos exemplos foram utilizados valores 0, por não ser o foco da discussão, e seus valores mudar na parte 1 e 2, mas uma implementação real para resolver o problema traria os pontos ganhos em cada condição.

Para um universo de 9 possibilidades diferentes, como no problema do dia 2 do Advent of Code, qualquer uma das soluções apresentadas vai conseguir atender. Porém isso não muda o fato de algumas serem mais otimizadas que outras, e que a lógica utilizada poderiam ser reaproveitada, e que conseguiriam lidar melhor com problemas onde existem muito mais possibilidades a serem analisadas.

Também existe uma discussão se seria mais eficiente tratar a string inteira ('A X'), ou processar isso e tratar como tuplas (('A', 'X')). Em outros casos poderia ser tratar direto a string, ou converter para um número inteiro. Quanto menos conversões necessárias melhor, porém as vezes isso poderia mudar a complexidade de uma operação como comparação, o que valeria pagar o custo para converter o dado.

Infelizmente no Python não existe a possibilidade de escolher a implementação do dicionário, ele sempre funcionará como uma tabela de espalhamento, e por isso valores que não podem ser convertidos para hash não podem ser utilizados como chaves (documentação oficial).

E todas essas otimizações fazem pouquíssima diferença para esse problema, mas essas mesmas ideias podem ser aplicadas a outros problemas e lá trazerem diferenças significativas. Estou usando esse problema só como desculpa para falar desses detalhes.

Discussão sobre o Advent of Code 2022 - Dia 1: Processando lista de valores

Eduardo Klosowski — Fri, 02 Dec 2022 02:54:59 +0000

Começou o Advent of Code deste ano, que embora tenha uma competição de quem resolve primeiro os problemas propostos, também é uma oportunidade para se desafiar e treinar o entendimento e a resolução de problemas. Porém aqui não quero discutir como resolver os problemas, mas sim pensar sobre os algoritmos usados para resolvê-los.

O problema do dia 1

Para quem não viu, ou se quiser relembrar, o problema do dia 1 é o "contando calorias", e que consiste basicamente de somar grupos de números e encontrar as maiores somas.

Recomendo primeiramente tentar resolver esse problema. Se tiver dificuldades, ou se quiser ter uma ideia melhor da linha de pensamento que vou usar como base da discussão, recomendo o vídeo do Bruno Rocha:

1ª Questão - Uso de memória

Muitas soluções mais simplistas e rápidas de se programar acabam criando várias cópias dos valores em memória, por exemplo: guardando todo o conteúdo do arquivo de entrada em uma variável, criando uma lista (ou array) com todo os valores, outra lista com os valores convertidos para inteiro, outra com a soma dos valores... Para exemplificar em Python, seria algo como:

entrada1 = open('entrada.txt').read()
entrada2 = entrada.strip().split('\n\n')
entrada3 = [e.split() for e in entrada2]
entrada4 = [[int(f) for f in e] for e in entrada3]
entrada5 = [sum(e) for e in entrada4]
...

Nesse exemplo estou exagerando em criar várias variáveis, muitas dessas instruções poderiam ser feitas juntas de outras, economizando algumas linhas (e variáveis também). Porém a questão que quero apontar aqui é que todos os dados são carregados para a memória do computador de uma vez só, e depois são criadas várias cópias ligeiramente diferentes do mesmo dado (convertido de string para inteiro, lista da soma dos valores no lugar da lista de lista...).

Isso pode ser um problema quando existem muitos dados para tratar, como no caso de arquivos com gigabytes de dados, ou quando se tem uma quantidade limitada de memória RAM, como em um dispositivo embarcado ou um AWS Lambda. E isso continua sendo um problema mesmo reduzindo a quantidade de cópias dos valores para duas, ou até mesmo uma única cópia em memória, só será necessário mais dados para o problema ocorrer.

A solução apresentada pelo Bruno Rocha é bastante interessante nesse ponto, que pelo menos para a parte 1 do problema, vai lendo, processando e descartando dados conforme eles são e deixam de ser necessários, com a exceção da entrada que é lida inteira, mas poderia ser adaptada para uma solução com um buffer que leia poucos bytes por vez, ou conforme for necessário. O código dele se aproveita bastante das funcionalidades do Rust para isso, mas também é possível fazer algo semelhante sem essas funcionalidades, ou em outras linguagens, exemplo:

max_calorias = 0
calorias = 0
for linha in open('entrada.txt'):
    linha = linha.strip()
    if linha == '':
        max_calorias = max(calorias, max_calorias)
        calorias = 0
    else:
        calorias += int(linha)
max_calorias = max(calorias, max_calorias)
print(max_calorias)

Nessa solução apresentada ainda existe alguma duplicidade de valores na memória, como ter ao mesmo tempo o valor como string e inteiro, mas se limita apenas ao que está sendo tratando no momento e não tudo, não usa nenhuma lista, por exemplo. Ela também permite tratar arquivos de qualquer tamanho com pouca memória RAM, até mesmo com uma quantidade de memória RAM menor do que o tamanho do arquivo de entrada.

Entretanto a parte 2 do problema exige um pouco mais de memória, já que não busca só o maior valor, vou discutir sobre isso junto com a próxima questão.

2ª Questão - Ordenação

A parte 2 do problema pede a soma dos 3 maiores valores, uma forma de resolver isso é gerar uma lista com as somas dos valores, ordenar eles e pegar os 3 maiores. Além dessa solução precisar de mais memória conforme a quantidade de valores, ela também irá ordenar todos os valores, sendo que precisamos apenas dos 3 primeiros (ou últimos, dependendo da lógica e ponto de vista). Isso é um problema porque algoritmos de ordenação tendem a ter complexidade ciclomática O(n * log2(n)), que basicamente diz que o tempo para ordenar os valores cresce (demora mais para rodar) conforme tem mais valores para ordenar (n é igual a quantidade de valores nesse caso), devido a maior quantidade de comparações de valores necessárias para isso.

Considerando a entrada presente no problema, são 241 somas que devem ser ordenadas para se buscar as 3 maiores, isso da algo em torno da grandeza de 1900 comparações (241 * log2(241)) para se ordenar essa lista com um algoritmo como o quick sort, que é largamente utilizado. Porém olhando o bubble sort, que embora seja conhecidamente mais lento que o quick sort na maioria dos casos, e por isso normalmente não utilizado quando se precisa de desempenho, ele permite interromper sua execução logo após a ordenação dos 3 valores desejados, sem precisar ordenar desnecessariamente os demais. Exemplo:

valores = [...]
for i in range(3):
    for j in range(i + 1, len(valores)):
        if valores[j] > valores[i]:
            aux = valores[i]
            valores[i] = valores[j]
            valores[j] = aux
print(valores[:3])

Isso faria a ordenação apenas dos 3 valores desejados com 717 comparações, algo em torno da grandeza de O(3 * n), sendo 3 o número de valores desejados (3 * 241 = 723 apenas para comparação). Desta forma são menos comparações que o quick sort, e bem menos do que as 28920 comparações que a execução completa do bubble sort levaria.

Entretanto o problema aqui está no tamanho da lista a ser ordenada. Imagine que se em vez de ordenar a lista só no final, toda vez que um valor fosse inserido na lista isso já fosse feito de forma ordenada, e descartando os valores desnecessários (menores que os 3 maiores já encontrados até aquele momento). Isso reduziria muito a lista a ser ordenada (4 valores, os 3 maiores e o valor sendo processado), e sabendo que os valores que estão na lista já estão ordenados, não é necessário ordená-los novamente, só inserir o novo valor no local certo. Desta forma seria necessário menos comparações, o código rodaria mais rápido, e com o descarte dos valores menores, voltaria a ser possível executá-lo com uma quantidade reduzida de memória RAM.

Considerações

Nesse texto discuti sobre variações de algoritmos que resolvem o problema visando tempo de execução e consumo de memória, que é um foco diferente da competição do Advent of Code em si, que é de quem dá a resposta certa mais próximo da liberação do problema. Também é diferente do foco das competições de programação que visam apenas tempo de execução (e não consideram uso de memória).

Para a competição do Advent of Code, uma solução mais rápida de programar pode ser mais interessante, mesmo que em alguns casos ela demore mais para executar, visto que a velocidade do processador pode compensar o tempo que uma pessoa levaria para pensar e implementar um algoritmo mais otimizado. Isso mostra que dependendo da onde for utilizado, nem sempre uma solução mais otimizada para o computador é a melhor.

Outro ponto positivo de participar do Advent of Code são os exercícios de tentar entender o problema e de tentar pensar em como representar os dados para que um algoritmo possa processá-los, mesmo sem conseguir encontrar a resposta, só de pensar nas estruturas de dados e organização deles pode ser um ótimo exercício.

Eu não pretendo fazer uma análise dessa de cada dia, até porque são bastantes problemas, e os mais avançados tendem a ser mais complexos e misturar várias coisas. Esses pontos que levantei provavelmente vão se repetir nos próximos problemas, o que ficaria redundante também. Mas volto a escrever outro texto se eu observar alguma questão que seja interessante trazer para a discussão.

Questões para estudo de algoritmos

Eduardo Klosowski — Mon, 17 Oct 2022 17:56:34 +0000

Recentemente li o texto do Maycon Alves, "3 algoritmos para você sua lógica", onde são apresentados 3 problemas para treinar a lógica e escrita de algoritmos: cálculo de fatorial, identificar se um número é primo, e calcular os valores da sequência de Fibonacci. São problemas interessantes, e após resolvê-los, pode-se fazer outras perguntas que levam a um entendimento mais profundo desses algoritmos. Eu recomendo que leiam o texto do Maycon primeiro e tentem implementar uma solução para esses problemas propostos, e com isso feito, vamos discutir um pouco sobre eles.

Analisando as soluções

No texto do Maycon, tem uma dica sobre o problema da sequência de Fibonacci, onde é dito que ele pode ser resolvido usando recursividade ou loops. Vamos analisar essas opções.

Uma solução recursiva pode ser implementada como a baixo. O código dessa solução é simples e se aproxima bastante da descrição matemática do problema.

def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n - 2) + fibonacci(n - 1)

Enquanto uma solução iterativa (com loop) pode ser um pouco mais complicada de se ler:

def fibonacci(n):
    a = 0
    b = 1
    for i in range(n - 1):
        a, b = b, a + b
    return b

Essas mesmas técnicas podem ser utilizadas para resolver o cálculo do fatorial. Onde uma implementação recursiva e outra iterativa podem ser vistas a baixo:

def fatorial(n):
    if n <= 1:
        return 1
    return n * fatorial(n - 1)

def fatorial(n):
    valor = 0
    for i in range(1, n + 1):
        valor *= i
    return valor

Com essas soluções implementadas, vem uma pergunta: Existe alguma diferença entre elas, além da diferença de como isso está expresso no código? Um primeiro teste que pode ser feito é de desempenho visando observar quanto tempo cada implementação leva para calcular a resposta. Os testes foram executados em um notebook com processador Intel Core i7-6500U CPU @ 2.50GHz, memória RAM DDR4 2133MHz, no Python 3.9.2 do Debian 11, desativamente o garbage collector do Python durante a execução das funções para ter um resultado com menos variação, apresentados como uma média de 10 execuções (código utilizado).

O gráfico a baixo mostra o tempo de execução das implementações que calculam os valores da sequência de Fibonacci, onde é possível observar que a implementação iterativa mantém uma linearidade do tempo conforme vai se pedindo números maiores da sequência, diferente da implementação recursiva, que a cada valor quase dobra o tempo de execução.

E a baixo pode-se observar o gráfico para as implementações que calculam o fatorial, que devido aos tempos serem mais baixos, possui uma variação um pouco maior, e é possível observar uma tendência de reta para as duas implementações, com a implementação recursiva tendo um ângulo um pouco mais íngreme, implicando em um algoritmo mais lento.

A partir desses dois gráficos algumas perguntas podem ser feitas: Por que a implementação recursiva do Fibonacci apresentou uma curva que se aproxima de uma exponencial e não de uma reta como as demais? Qual a diferença para a implementação recursiva do fatorial que explicar isso? Implementações recursivas são sempre piores que as implementações iterativas, ou existem casos em elas superem ou se equivalem as iterativas?

Saindo um pouco desses gráficos, outras questões podem ser levantadas, como: Existe mais algum aspecto ou característica além do tempo de execução (e facilidade de leitura do código) para a escolha entre uma implementação ou outra? Considerando que essas funções vão rodar em um servidor, existe alguma coisa que possa ser feita para melhorar a performance dessas funções, como reaproveitar resultados já calculados para se calcular novos resultados? Como isso poderia ser feito? Quais são as desvantagens dessa abordagem?

Olhando para o problema de verificar se um número é primo, existe o crivo de Eratóstenes, ele é uma implementação eficiente? Existem casos em que ele pode ser uma boa solução ou não? O exemplo a baixo (retirado da Wikipédia) mostra o processo para encontrar todos os números primos até 120, existe alguma forma de adaptá-lo para executar diversas vezes reaproveitando o que já foi calculado, como sempre retornar o próximo número primo?

Considerações

Se você nunca se deparou com perguntas desse tipo, seja bem-vindo a área de análise de algoritmos, onde após se ter uma solução, busca-se analisar e descrever o comportamento do algoritmo, e até a busca de algoritmos mais eficientes. E trazendo para o dia a dia de um desenvolvedor, essas questões podem ser a resposta do motivo do código funcionar muito bem no computador de quem desenvolveu, mas demorar muito ou apresentar problemas para rodar no servidor de produção, ou com o tempo (e crescimento do volume de dados) começar a dar problemas.

Nesse artigo eu apenas levantei as perguntas, deixo que cada um busque as respostas, que existem. Sintam-se livres para me procurar para discutir alguma questão ou orientações para encontrar as respostas, seja nos comentários do texto no dev.to ou no Twitter.