35 · Paralelismo

python3 exercicios/run_all.py 35

adopt Arcane.Concurrent as P
adopt Arcane.Time as Time

// ── O problema ──────────────────────────────────────────────
//
// 'async', 'thread' e 'parallel' usam threads do Python, e duas
// threads do Python nunca executam bytecode ao mesmo tempo — e o GIL.
//
//     trabalho que ESPERA (rede, disco, banco)  ->  thread
//     trabalho que CALCULA (numeros, imagem)    ->  processo
//
// Para o primeiro caso a thread e perfeita: ela solta o GIL enquanto
// espera, e dez downloads acontecem juntos. Para o segundo, oito
// threads levam o mesmo tempo que uma.

out $"nucleos disponiveis: {P.nucleos()}"

action cpu(n):
    soma := 0
    cycle i from 1 to n:
        soma += i * i
    yield soma

steady BLOCOS := [150000, 150000, 150000, 150000]

action cronometrar(f):
    inicio := Time.monotonic()
    valor := f()
    yield {"ms": (Time.monotonic() - inicio) * 1000, "valor": valor}

serie := cronometrar(lambda => [cpu(b) cycle b in BLOCOS])
threads := cronometrar(lambda => P.map(cpu, BLOCOS))
processos := cronometrar(lambda => P.map_processos(cpu, BLOCOS))

// ── O resultado tem de ser o mesmo pelos tres caminhos ──────

assert serie["valor"] is threads["valor"], "as threads mudaram o resultado"
assert serie["valor"] is processos["valor"], "os processos mudaram o resultado"

out $"serie:     {round(serie['ms'])} ms"
out $"threads:   {round(threads['ms'])} ms"
out $"processos: {round(processos['ms'])} ms"

ganho_threads := serie["ms"] / threads["ms"]
ganho_processos := serie["ms"] / processos["ms"]

out $"threads   x{round(ganho_threads, 2)}"
out $"processos x{round(ganho_processos, 2)}"

// ── O que os numeros provam ─────────────────────────────────
//
// A comparacao e sempre com a SERIE medida na mesma maquina. Um
// limite absoluto ("menos de 200 ms") mediria a maquina, e nao o
// paralelismo — e reprovaria num runner ocupado que esta certissimo.

given P.nucleos() >= 4:
    assert ganho_processos bigger 1.2,
    "os processos nao ganharam da serie: a travessia voltou a rodar num nucleo so"
    assert ganho_threads < 1.5,
    "as threads ganharam em trabalho de CPU — se o GIL sumiu, este exercicio e que mudou"

out "ok"

serie := cronometrar(lambda => [cpu(b) cycle b in BLOCOS])
threads := cronometrar(lambda => P.map(cpu, BLOCOS))
processos := cronometrar(lambda => P.map_processos(cpu, BLOCOS))

assert ganho_processos bigger 1.2      // certo
assert processos["ms"] < 200           // errado

given P.nucleos() >= 4:
    assert ganho_processos bigger 1.2, …

adopt Arcane.Concurrent as P
adopt Arcane.Math as M
adopt Arcane.Database as DB

// ── O que viaja ─────────────────────────────────────────────
//
// Um processo recebe o trabalho por COPIA. O que atravessa nao e a
// acao — e a DECLARACAO dela, mais os nomes que ela le e nao cria, e
// tudo o que esses nomes alcancam. Nada disso precisa ser dito.

steady TAXA := 0.08

record Pedido:
    cliente: String
    valor: Float

enum Faixa:
    Comum
    Alta

record Nota:
    cliente: String
    total: Float
    faixa: Faixa

action imposto(v):
    yield v * TAXA

action faixa_de(v):
    yield Faixa.Alta given v bigger 200 otherwise Faixa.Comum

action emitir(p: Pedido) -> Nota:
    total := M.round(p.valor + imposto(p.valor), 2)
    yield Nota(p.cliente, total, faixa_de(total))

pedidos := [Pedido("ana", 100.0), Pedido("bia", 250.0), Pedido("caio", 30.0)]
notas := P.map_processos(emitir, pedidos)

cycle n in notas:
    out n

assert len(notas) is 3, "um item se perdeu na travessia"
assert notas[0].total is 108.0, "a steady TAXA nao atravessou"
assert notas[1].faixa is Faixa.Alta, "o enum nao atravessou"
assert notas[2].faixa is Faixa.Comum, "o enum nao atravessou"

// O tipo que VOLTA e o mesmo declarado aqui — e por isso 'with',
// que confere os campos contra o record, aceita o resultado.
copia := notas[0] with {"cliente": "outro"}
assert copia.cliente is "outro", "o record voltou como outro tipo"
assert copia.total is notas[0].total, "'with' nao preservou o resto"

// ── O que NAO viaja ─────────────────────────────────────────
//
// Uma conexao de banco, um arquivo aberto, um socket, um mutex, um
// canal e uma tarefa existem no processo que os abriu. Copia-los nao
// faria sentido: o outro lado ganharia um descritor que la nao aponta
// para nada.

banco := DB.connect(":memory:")

action usa_o_banco_de_fora(n):
    yield len(DB.query(banco, "select 1")) + n

monitor:
    P.map_processos(usa_o_banco_de_fora, [1, 2])
    assert no, "devia ter recusado"
handle Error as e:
    out e.message
    assert "banco" in e.message, "a mensagem tem de citar a VARIAVEL pelo nome"

// ── E o que fazer a respeito ────────────────────────────────
//
// Abrir o recurso DENTRO da acao: cada processo abre o seu.

action abre_o_proprio(n):
    meu := DB.connect(":memory:")
    DB.execute(meu, "create table t (n integer)")
    DB.execute(meu, "insert into t values (?)", [n])
    yield DB.count(meu, "t") * n

assert P.map_processos(abre_o_proprio, [2, 3, 4]) is [2, 3, 4]

// Um recurso que a acao NAO usa nao atrapalha: a analise so cobra o
// nome quando ele faz falta de verdade.
_trava := P.mutex()
_canal := P.canal()

action nao_usa_nada_de_fora(n):
    yield n * 10

assert P.map_processos(nao_usa_nada_de_fora, [1, 2, 3]) is [10, 20, 30]

out "ok"

steady TAXA := 0.08

record Pedido:
    cliente: String
    valor: Float

action imposto(v):
    yield v * TAXA

action emitir(p):
    yield Nota(p.cliente, p.valor + imposto(p.valor))

P.map_processos(emitir, pedidos)

copia := notas[0] with {"cliente": "outro"}

erro[DF1001]: 'banco' cannot cross into another process
  = nota: it holds a connection to a database, which exists only in
          the process that opened it
  = dica: open it INSIDE the action — each process opens its own — or
          use 'map', which uses threads and shares memory

action abre_o_proprio(n):
    meu := DB.connect(":memory:")
    …

_trava := P.mutex()
_canal := P.canal()

action nao_usa_nada_de_fora(n):
    yield n * 10           // atravessa sem problema

adopt Arcane.Concurrent as P
adopt Arcane.Time as Time

// ── O desenho ───────────────────────────────────────────────
//
//     dividir  ->  map_processos  ->  juntar
//
// Cada bloco tem de ser autossuficiente: ele atravessa para outro
// processo, faz a conta la, e volta com um resumo pequeno. O que
// atravessa custa — mandar o bloco e receber UM numero e barato;
// mandar o bloco e receber outro bloco nao e.

steady TOTAL := 60000
steady BLOCOS := 6

action dividir(quantos, em):
    tamanho := quantos ~/ em
    yield [{"de":i * tamanho + 1, "ate":(i + 1) * tamanho}
        cycle i in range(0, em)]

// O bloco vira um RESUMO, e nao uma lista: e o resumo que volta.
action resumir(bloco):
    soma := 0
    primos := 0
    cycle n from bloco["de"] to bloco["ate"]:
        soma += n
        given e_primo(n):
            primos += 1
    yield {"soma": soma, "primos": primos}

action e_primo(n):
    given n < 2:
        yield no
    given n % 2 is 0:
        yield n is 2
    d := 3
    persist d * d <= n:
        given n % d is 0:
            yield no
        d += 2
    yield yes

blocos := dividir(TOTAL, BLOCOS)
assert len(blocos) is BLOCOS, "a divisao errou a contagem"
assert blocos[0]["de"] is 1
assert blocos[BLOCOS - 1]["ate"] is TOTAL, "o ultimo bloco nao fecha o total"

inicio := Time.monotonic()
resumos := P.map_processos(resumir, blocos)
ms := (Time.monotonic() - inicio) * 1000

soma := sum([r["soma"] cycle r in resumos])
primos := sum([r["primos"] cycle r in resumos])

out $"{BLOCOS} blocos de {TOTAL ~/ BLOCOS} em {round(ms)} ms"
out $"soma:   {soma}"
out $"primos: {primos}"

// ── A conferencia ───────────────────────────────────────────
//
// Um pipeline paralelo que nao e conferido contra a resposta fechada
// e um gerador de numeros plausiveis. A soma de 1 a n tem formula:

esperado := TOTAL * (TOTAL + 1) ~/ 2
assert soma is esperado, $"a soma deu {soma}, e devia ser {esperado}"
assert primos is 6057, "a contagem de primos ate 60000 e conhecida"

// A ordem da entrada e a da saida — sem isso, quem chama teria de
// reassociar bloco e resumo, e e ai que se erra.
assert resumos[0]["soma"] < resumos[BLOCOS - 1]["soma"],
"os resumos voltaram fora de ordem"

out "ok"

dividir  ->  map_processos  ->  juntar

action resumir(bloco):
    soma := 0
    primos := 0
    cycle n from bloco["de"] to bloco["ate"]:
        …
    yield {"soma": soma, "primos": primos}

esperado := TOTAL * (TOTAL + 1) ~/ 2
assert soma is esperado
assert primos is 6057        // π(60000), que é uma constante conhecida

assert resumos[0]["soma"] < resumos[BLOCOS - 1]["soma"]