PyChannel 1.5

A Python concurrency library inspired by Go's goroutines and channels

PyChannel

PyChannel é uma biblioteca de concorrência inspirada nas goroutines e nos channels da linguagem Go. Sua API é composta por três módulos principais: Task, Channel e Buffer.

A PyChannel utiliza processos gerenciados pelo loky. Ela é inspirada nos conceitos do Go, mas não implementa goroutines reais.

Fluxo de execução

flowchart LR
    A["Função decorada"] --> B["Task"]
    B --> C["Worker do loky"]
    C --> D["Future"]
    D --> E{"Buffer configurado?"}
    E -- "Não" --> F["Channel recebe o resultado"]
    E -- "Sim, acumulando" --> G["Channel recebe False"]
    E -- "Sim, lote pronto" --> H["Channel recebe a lista"]

A Task envia a função para um worker. O resultado retorna por um Future e, quando configurado, também é encaminhado ao Channel. Se a Task possuir um Buffer, os resultados são acumulados antes da liberação do lote.

Task

A classe Task transforma uma função comum em uma tarefa executada por processos.

Task(
    max_workers,
    name=None,
    channel=None,
    buffer=None,
)

Parâmetro	Função
max_workers	Define a quantidade máxima de processos disponíveis no pool.
name	Define o nome da tarefa exibido nos logs.
channel	Recebe os resultados e as exceções das execuções.
buffer	Ativa o agrupamento automático dos resultados.

Decorando uma função

O método run() retorna o decorator responsável por enviar cada chamada ao executor:

from PyChannel import Channel, Task

channel = Channel(name="Resultados")
task = Task(max_workers=4, name="Cálculos", channel=channel)


@task.run()
def quadrado(numero):
    return numero * numero


future = quadrado(10)

A chamada não devolve o resultado diretamente. Ela devolve um Future:

resultado = future.result()  # 100

Como existe um Channel, o mesmo resultado também pode ser consumido por ele:

resultado = channel.get()  # 100

Comportamento da Task

Configuração	Comportamento
Sem Channel e sem Buffer	O resultado fica disponível somente no Future.
Com Channel	Cada resultado é enviado ao Channel.
Com Channel e Buffer	O Channel recebe False durante o acúmulo e uma lista quando o lote fica pronto.
Com Buffer, sem Channel	O Buffer é atualizado e o resultado individual continua disponível no Future.

Se a função gerar uma exceção, ela permanece armazenada no Future. Quando existe um Channel, o objeto da exceção também é enviado para ele.

Os executores registrados pela Task são finalizados automaticamente quando o programa termina. Quando for necessário encerrar o pool antes do fim do programa, use task.shutdown().

Channel

O Channel transporta valores entre produtores e consumidores por meio de uma multiprocessing.Queue.

Channel(name=None, limit=None)

Parâmetro	Função
name	Define o nome utilizado nos logs.
limit	Define a quantidade máxima de valores pendentes. None cria uma fila sem limite explícito.

Enviando e recebendo valores

from PyChannel import Channel

channel = Channel(name="Eventos")

channel.send("processado")
resultado = channel.get()

print(resultado)  # processado

send() bloqueia se um Channel limitado estiver cheio. get() bloqueia enquanto não houver nenhum valor disponível.

Ignorando resultados falsos

O método get(loop=True) continua consumindo a fila até encontrar um valor verdadeiro:

resultado = channel.get(loop=True)

Esse modo é útil com o Buffer automático, pois ignora os valores False enviados durante o acúmulo e retorna diretamente o lote:

somar()
somar()

lote = channel.get(loop=True)
print(lote)  # [2, 2]

loop=True ignora qualquer valor considerado falso pelo Python, incluindo False, None, 0, "" e coleções vazias.

Buffer

O Buffer acumula valores temporariamente até atingir sua capacidade ou seu limite de tempo.

Buffer(capacity, timeout, name=None)

Parâmetro	Função
capacity	Quantidade máxima de valores acumulados.
timeout	Tempo máximo, em segundos, contado a partir do primeiro valor.
name	Define o nome utilizado nos logs.

O acesso interno é protegido por um RLock, permitindo chamadas concorrentes aos seus métodos.

Uso manual

from PyChannel import Buffer

buffer = Buffer(capacity=3, timeout=10, name="Lote")

buffer.sender("A")
buffer.sender("B")

print(buffer.status())  # False

buffer.sender("C")

if buffer.status():
    lote = buffer.get().copy()
    buffer.reset()
    print(lote)  # ["A", "B", "C"]

Métodos

Método	Comportamento
sender(data)	Adiciona um valor se ainda houver capacidade. O contador de tempo começa no primeiro valor.
get()	Retorna a lista atualmente armazenada.
status()	Retorna True quando a capacidade ou o timeout é atingido.
reset()	Limpa a lista e reinicia o controle de tempo.

O timeout não cria uma tarefa em segundo plano. Ele é avaliado quando status() é chamado ou quando a Task processa um novo resultado.

Integração dos módulos

from PyChannel import Buffer, Channel, Task

buffer = Buffer(capacity=2, timeout=10, name="Resultados")
channel = Channel(name="Saída")
task = Task(
    max_workers=2,
    name="Soma",
    channel=channel,
    buffer=buffer,
)


@task.run()
def somar(a, b):
    return a + b


somar(1, 1)
somar(2, 2)

lote = channel.get(loop=True)
print(lote)  # A ordem pode ser [2, 4] ou [4, 2]

Os workers executam de forma independente. Por isso, os resultados chegam ao Buffer e ao Channel na ordem de conclusão, não necessariamente na ordem de envio.