Tarefas paralelas e prazos

Resultados paralelos nomeados, prazos de await e delegação lazy.

Resultados paralelos nomeados¶

Tarefas nomeadas em parallel produzem um record tipado após a junção estruturada. Cada tarefa captura o escopo externo; rótulos não viram variáveis para tarefas irmãs. Falhas cancelam e aguardam as irmãs cooperativamente. Resultados transitivamente lineares são rejeitados.

Pratique na By Example · Programa fonte

// Feature: typed named results from structured parallel tasks
// Each label names a result field; each RHS captures only the outer scope.
struct User { name: str }
async fn load_user() -> User = User { name: "Ada" }
async fn load_orders() -> [int] = [101, 102]

let user = "outer"
let data = parallel {
    user = await load_user()
    orders = await load_orders()
    captured = user
}
print(data.user.name)
print(data.orders.len())
print(data.captured)
print(user)
// expected:
//   Ada
//   2
//   outer
//   outer

Execute este exemplo localmente com a CLI instalada. O playground do navegador não fornece o scheduler do host.

Prazos de await¶

await tarefa within duração retorna um Result tipado de timeout. Ao expirar, o scheduler solicita cancelamento e aguarda a limpeza cooperativa da tarefa. Uma tarefa que retorna Result produz um Result aninhado; escolha qual camada propagar. Trabalho nativo sem checkpoints não pode ser interrompido à força.

Pratique na By Example · Programa fonte

async fn load() -> int {
    sleep 20ms
    42
}

fn main() {
    let result = await load() within 1s
    print(result.unwrap())
}

Execute este exemplo localmente com a CLI instalada. O playground do navegador não fornece o scheduler do host.

Delegação lazy de generators¶

yield ... filho delega um valor produzido por vez e mantém o cursor do filho entre retomadas. Não materializa um array. Fechar o generator externo fecha o filho ativo e executa sua limpeza; yield comum mantém seu comportamento.

Pratique na By Example · Programa fonte

fn* middle()->int{yield 2;yield 3}
fn* values()->int{yield 1;yield ... middle();yield 4}
for value in values(){print(value)}

Lendo uma fonte assíncrona¶

for await pede um valor por vez à fonte e aguarda next() antes de executar o corpo do laço. Nil encerra a iteração. O método conhecido next() deve retornar Future<T?> ou Task<T?>, e a fonte deve implementar core::cleanup::Cleanup. O exemplo consome uma fonte inteira e interrompe outra antes do fim; ambas são fechadas.

use core::cleanup::Cleanup
struct Events { remaining: int }
impl Events {
    async fn next(self) -> int? {
        if self.remaining <= 0 { return nil }
        let value = self.remaining
        self.remaining -= 1
        return value
    }
}
impl Cleanup for Events {
    fn close(self) { print("closed") }
}
let complete = Events { remaining: 3 }
var sum = 0
for await value in complete { sum += value }
assert_eq(sum, 6)
let early = Events { remaining: 5 }
:events for await value in early {
    if value == 4 { break :events }
    print(value)
}
print(sum)

A fonte é adquirida uma vez e fechada ao terminar, interromper, retornar, falhar ou receber cancelamento cooperativo. Padrões, inclusive for await case, examinam o valor aguardado. Arrays síncronos não se tornam fontes assíncronas. Execute com a CLI; wasm-aot rejeita operações do escalonador explicitamente.

By Example

Limitando o trabalho concorrente¶

parallel(limit: 2) [for id in ids => await load_profile(id)] processa um array com no máximo dois trabalhos ativos e devolve os resultados na ordem de entrada. O exemplo conta os trabalhos ativos para conferir o limite e depois verifica uma entrada vazia e a cópia dos itens da fonte.

var active = 0
var peak = 0
async fn load_profile(id: int) -> int {
    active += 1
    if active > peak { peak = active }
    defer { active -= 1 }
    sleep 2ms
    return id * 10
}
let profiles = parallel(limit: 2) [for id in [1, 2, 3, 4] => await load_profile(id)]
assert_eq(peak, 2)
assert_eq(active, 0)
assert_eq(profiles.len(), 4)
assert_eq(profiles[0], 10)
assert_eq(profiles[3], 40)
print(peak)
print(profiles[0])
print(profiles[3])
let no_ids: [int] = []
let empty = parallel(limit: 4) [for id in no_ids => id]
assert_eq(empty.len(), 0)

// The pool snapshots source items before evaluating any mapper.
let mut ids = [1, 2, 3]
let original = parallel(limit: 1) [for id in ids => { ids[1] = 99; id }]
assert_eq(original[1], 2)
assert_eq(ids[1], 99)

O limite inteiro positivo e a fonte são avaliados uma vez, nessa ordem. Um grupo fixo de trabalhadores pega o próximo item ao terminar o atual; não há uma tarefa criada para cada entrada. A cópia da fonte protege os itens pendentes de alterações posteriores. Uma falha cancela os outros trabalhos e aguarda a limpeza antes de propagar o erro. A concorrência é cooperativa; cálculos sem suspensão não ganham threads do sistema. Disponível na VM, nativo e LLVM; wasm-aot informa a limitação do escalonador.

By Example

Continue aprendendo¶

Documentação · Pratique na By Example