Proteger estado compartilhado entre corrotinas

import asyncio

CAPACIDADE = 1
reservas_confirmadas = 0

async def tentar_reservar(nome: str) -> bool:
    global reservas_confirmadas

    if reservas_confirmadas >= CAPACIDADE:
        print(f"{nome}: sem vaga")
        return False

    print(f"{nome}: encontrei vaga")
    await asyncio.sleep(0)  # cede uma oportunidade de execução
    reservas_confirmadas += 1
    print(f"{nome}: confirmei")
    return True

async def main() -> None:
    global reservas_confirmadas

    async with asyncio.TaskGroup() as grupo:
        grupo.create_task(tentar_reservar("Ana"))
        grupo.create_task(tentar_reservar("Bruno"))

    print(f"Reservas finais: {reservas_confirmadas}")
    assert reservas_confirmadas <= CAPACIDADE, "Capacidade ultrapassada!"

asyncio.run(main())

import asyncio

async def confirmar_reserva(
    cliente: str,
    vagas: dict[str, int],
    lock: asyncio.Lock,
) -> bool:
    async with lock:
        if vagas["disponiveis"] == 0:
            print(f"{cliente}: sem vaga")
            return False

        # Simula uma etapa assíncrona que faz parte da confirmação.
        await asyncio.sleep(0)
        vagas["disponiveis"] -= 1
        print(f"{cliente}: reserva confirmada")
        return True

async def main() -> None:
    vagas = {"disponiveis": 1}
    lock = asyncio.Lock()

    async with asyncio.TaskGroup() as grupo:
        grupo.create_task(confirmar_reserva("Ana", vagas, lock))
        grupo.create_task(confirmar_reserva("Bruno", vagas, lock))

    print(vagas)

asyncio.run(main())

# Não use este padrão para a regra de vagas.
if vagas["disponiveis"] > 0:
    await asyncio.sleep(0)
    async with lock:
        vagas["disponiveis"] -= 1

import asyncio

capacidade = 1
confirmadas = 0
lock = asyncio.Lock()

async def tentar_reserva(nome: str) -> bool:
    global confirmadas

    # Trabalho independente do estado compartilhado.
    mensagem = f"Pedido de {nome} pronto"
    await asyncio.sleep(0)

    # Consulta, decisão e alteração formam uma unidade.
    async with lock:
        if confirmadas >= capacidade:
            print(f"{mensagem}: sem vaga")
            return False
        confirmadas += 1
        print(f"{mensagem}: reserva confirmada")
        return True

async def tentar_reserva_incorreta() -> bool:
    global confirmadas

    async with lock:
        tem_vaga = confirmadas < capacidade

    await asyncio.sleep(0)  # outra tarefa pode confirmar a última vaga

    if tem_vaga:
        async with lock:
            confirmadas += 1
        return True
    return False

import asyncio

vagas = 1
lock = asyncio.Lock()


async def confirmar_pagamento() -> None:
    # Representa uma operação cancelável que ainda não mudou `vagas`.
    await asyncio.sleep(0)


async def reservar(nome: str) -> bool:
    global vagas

    async with lock:
        if vagas == 0:
            return False

        await confirmar_pagamento()
        vagas -= 1
        print(f"{nome}: reserva confirmada; vagas = {vagas}")
        return True


async def main() -> None:
    tarefa = asyncio.create_task(reservar("Ana"))
    await asyncio.sleep(0)
    tarefa.cancel()

    try:
        await tarefa
    except asyncio.CancelledError:
        print(f"Ana: cancelada; vagas = {vagas}")

    # O Lock foi liberado pela saída cancelada do async with.
    await reservar("Bruno")


asyncio.run(main())

import asyncio

lock = asyncio.Lock()

# INCORRETO: registrar tenta adquirir um lock já mantido
# pela própria tarefa chamadora.
async def registrar() -> None:
    async with lock:
        print("registro concluído")

async def reservar_incorreto() -> None:
    async with lock:
        await registrar()  # espera sem saída

# CORRETO: reservar é responsável por adquirir o lock;
# a função interna pressupõe que ele já está adquirido.
def _registrar_com_lock() -> None:
    print("registro concluído")

async def reservar() -> None:
    async with lock:
        _registrar_com_lock()

# Se duas operações precisarem dos locks A e B, todas devem
# adquiri-los na mesma ordem: primeiro A, depois B.
async def operar_com_dois_locks(lock_a: asyncio.Lock, lock_b: asyncio.Lock) -> None:
    async with lock_a:
        async with lock_b:
            print("operação exclusiva")

# Evite também manter lock e aguardar uma tarefa que precisa dele.
# Reorganize o fluxo para que a tarefa dependente termine antes da
# aquisição, ou para que a espera aconteça depois de liberar o lock.

import asyncio


async def calcular_subtotal(valores: list[int]) -> int:
    await asyncio.sleep(0)  # simula uma operação assíncrona
    return sum(valor * 2 for valor in valores)


async def main() -> None:
    lotes = [[3, 5], [2, 7], [11]]

    async with asyncio.TaskGroup() as grupo:
        tarefas = [
            grupo.create_task(calcular_subtotal(lote))
            for lote in lotes
        ]

    total = sum(tarefa.result() for tarefa in tarefas)
    print(total)  # 56


asyncio.run(main())

import asyncio

CAPACIDADE = 1


def verificar(estado: dict[str, object]) -> None:
    vagas = estado["vagas"]
    confirmadas = estado["confirmadas"]
    assert isinstance(vagas, int)
    assert isinstance(confirmadas, list)
    assert vagas >= 0
    assert vagas + len(confirmadas) == CAPACIDADE


async def confirmar(
    nome: str,
    estado: dict[str, object],
    lock: asyncio.Lock,
    *,
    falhar: bool = False,
) -> bool:
    # VULNERÁVEL: esta função ainda não usa lock.
    vagas = estado["vagas"]
    assert isinstance(vagas, int)
    if vagas == 0:
        return False

    await asyncio.sleep(0)
    if falhar:
        raise RuntimeError(f"falha simulada de {nome}")

    estado["vagas"] = vagas - 1
    confirmadas = estado["confirmadas"]
    assert isinstance(confirmadas, list)
    confirmadas.append(nome)
    verificar(estado)
    return True


async def bloqueada_ate_cancelamento(
    lock: asyncio.Lock, entrou: asyncio.Event
) -> None:
    async with lock:
        entrou.set()
        await asyncio.Event().wait()


async def main() -> None:
    lock = asyncio.Lock()

    # Disputa por uma única vaga.
    estado: dict[str, object] = {"vagas": CAPACIDADE, "confirmadas": []}
    resultados = await asyncio.gather(
        confirmar("Ana", estado, lock),
        confirmar("Bia", estado, lock),
        return_exceptions=True,
    )
    print("disputa:", resultados, estado)

    # Falha antes da alteração final: o estado deve continuar consistente.
    estado = {"vagas": CAPACIDADE, "confirmadas": []}
    try:
        await confirmar("Caio", estado, lock, falhar=True)
    except RuntimeError as erro:
        print("exceção observada:", erro)
    verificar(estado)
    print("após exceção:", estado)

    # Cancelamento após adquirir o lock, mas antes de qualquer mutação.
    estado = {"vagas": CAPACIDADE, "confirmadas": []}
    entrou = asyncio.Event()
    tarefa = asyncio.create_task(bloqueada_ate_cancelamento(lock, entrou))
    await entrou.wait()
    tarefa.cancel()
    try:
        await tarefa
    except asyncio.CancelledError:
        print("tarefa cancelada")

    # Esta chamada só termina se o lock foi liberado na saída cancelada.
    resultado_sonda = await confirmar("Sonda", estado, lock)
    verificar(estado)
    print("sonda após cancelamento:", resultado_sonda, estado)


asyncio.run(main())

async def confirmar(
    nome: str,
    estado: dict[str, object],
    lock: asyncio.Lock,
    *,
    falhar: bool = False,
) -> bool:
    async with lock:
        vagas = estado["vagas"]
        assert isinstance(vagas, int)
        if vagas == 0:
            return False

        # Está dentro da seção crítica porque a decisão ainda depende
        # da vaga que acabamos de consultar.
        await asyncio.sleep(0)
        if falhar:
            raise RuntimeError(f"falha simulada de {nome}")

        estado["vagas"] = vagas - 1
        confirmadas = estado["confirmadas"]
        assert isinstance(confirmadas, list)
        confirmadas.append(nome)
        verificar(estado)
        return True