Node.js - Paralelismo limitado com Array.map

Créditos da Imagem

O Array.map () é uma função muito útil, mas, infelizmente, só funciona com funções síncronas. Uma solução simples para executar funções async é usar o Promose.all() ou seu irmão mais tolerante Promise.allSettled():

// Falha no momento que UMA função do .map falhar
const results = await Promise.all(array.map(asynMapFunction))

// Continua executando mesmo se uma função falhar
const results = await Promise.allSettled(array.map(asynMapFunction))

Funciona assim: o .map()irá converter cada item do array em uma Promise, então teremos um array de Promises para resolver. Há duas maneiras de fazer isso:

• Promise.all(): joga um erro se a função no .map jogar um erro ( MDN )
• Promise.allSettled(): executa as funções no .map em todo o array, mesmo que algumas delas joguem um erro ( MDN )

Portanto, a saída do .allSettled() é um array de objetos que informa se a execução falhou ou não.

Cada objeto na saída de .allSettled() é parecido com este:

// Quando a função roda com sucesso
{
  status: "fulfilled",
  value: // o valor retornado da função do .map
}

// Quando a função joga um erro
{
  status: "rejected",
  reason: // o erro jogado pela função do .map
}

Porém, há um problema: ao contrário de um .map() "normal", as funções de mapa não serão executadas em série. As funções async do mapa estarão executando ao mesmo tempo. Embora o JavaScript seja normalmente uma linguagem de thread única, isso significa que os recursos alocados (como memória e portas) para cada função serão ocupados até que as Promises sejam resolvidas ou rejeitadas. Para arrays enormes, no entanto, vamos executar um grande número de funções-mapa ao mesmo tempo. Isso pode potencialmente:

• Consumir muita memória, pois cada função-mapa mantém todas as suas variáveis ​​enquanto estiver em execução. Se você estiver executando lambda, por exemplo, ele pode facilmente travar seu tempo de execução (ou você tem que pagar o custo de atingir um tempo de execução mais robusto)
• Atingir os limites de taxa: se o mapa estiver acessando uma API para cada função, a API pode retornar um erro pela quantidade alta de requisições

Seria bom se pudéssemos de alguma forma limitar essas execuções paralelas. Uma opção é usar a função eachLimit do popular módulo async. Mas e se não quisermos importar uma dependência para um caso de uso tão simples? Vamos experimentar e aprender algo.

Limitar chamadas paralelas

Logo de cara, vamos usar Generators. Sei que é um recurso do JavaScript que muitos desenvolvedores (inclusive eu) não usam com frequência, mas neste caso, isso reduzirá o uso de memória e criará um código mais limpo.

Exemplo

Vamos definir um problema hipotético primeiro. Temos 100 URLs que queremos buscar, mas não queremos mais do que 10 chamadas paralelas ao mesmo tempo. Vamos usar o Google porque eles geralmente conseguem lidar com esse tipo de carga com facilidade!

// O array de URLs que queremos buscar
const urls = []
for (let i = 0; i < 100; i++) {
    // O parâmetro de pesquisa 'q' é o número do índice
    urls.push(`https://www.google.com/search?q=${i}`)
}

// A requisição é feita em uma função map assíncrona
async function mapFn(url, i) {
    // Estamos usando https://www.npmjs.com/package/got
    const contents = await got(url)
    return { i, url, contents }
}

Agora vamos escrever um programa que pegue essas 100 URLs, e as mapeia para imprimar os resultados:

async function main() {
    const results = await mapAllSettled(urls, mapFn, 10)
    console.dir(results)
}

// Rodando a função "async main()" usando https://www.npmjs.com/package/am
am(main)

Agora precisamos escrever a função mapAllSettled() que é bem semelhante a, Promise.allSettled(array.map(asyncMapFn)), mas com um limite. Sua assinatura se parece com isso: async function mapAllSettled(array, mapFn, limit).

Mas vamos voltar um pouco e ver como será essa execução. Para simplificar, digamos que temos 10 URLs. Se fossemos buscar todas elas de uma vez, teríamos algo assim:

Mas se tivéssemos um limite de quatro buscas ao mesmo tempo, seria assim:

Assim que uma busca for concluída, prosseguiremos com a próxima. A cada vez, temos quatro buscas em andamento. Vamos reorganizar o tempo de execução em quatro linhas que serão executadas por alguns "trabalhadores":

Todos os trabalhadores "consomem" o mesmo array, mas "inserem" o resultado na posição correta no array resultante, de forma que o valor mapeado para a URL número sete termine na posição sete do array resultante.

É aqui que os geradores são úteis. Podemos definir um gerador que recebe um array e yield o que a função de mapa espera :

function* arrayGenerator(array) {
    for (let index = 0; index < array.length; index++) {
        const currentValue = array[index]
        yield [ currentValue, index, array ]
    }
}

Para manter o formato de saída consistente com o Promise.allSettled(), podemos executar as funções do mapa em um bloco try..catch e emitir o resultado em um objeto com o formato:

async function mapItem(mapFn, currentValue, index, array) {
    try {
        return {
            status: 'fulfilled',
            value: await mapFn(currentValue, index, array)
        }
    } catch (reason) {
        return {
            status: 'rejected',
            reason
        }
    }
}

Cada trabalhador usa a função do gerador para buscar o currentItem, index e uma referência ao array, então chamamos mapItem()para executar o mapFn() assíncrono:

async function worker(id, gen, mapFn, result) {
    for (let [ currentValue, index, array ] of gen) {
        console.time(`Worker ${id} --- index ${index} item ${currentValue}`)
        result[index] = await mapItem(mapFn, currentValue, index, array)
        console.timeEnd(`Worker ${id} --- index ${index} item ${currentValue}`)
    }
}

Eu adicionei alguns console.time()e console.timeEnd()para tornar a saída mais compreensível, mas, basicamente, essa função tem duas linhas de código:

1. O loop for..of consome dados do gerador
2. o mapItem()chama a função especificada pelo usuário mapFn()e retorna seus resultados em um objeto que tem o mesmo formato quePromise.allSettled()

Agora vamos escrever o mapAllSettled() que basicamente cria esses trabalhadores e espera que eles terminem, depois retorna os resultados:

async function mapAllSettled(arr, mapFn, limit = arr.length) {
    const result = []

    if (arr.length === 0) {
        return result
    }

    const gen = arrayGenerator(arr)

    limit = Math.min(limit, arr.length)

    const workers = new Array(limit)
    for (let i = 0; i < limit; i++) {
        workers.push(worker(i, gen, mapFn, result))
    }

    await Promise.all(workers)

    return result
}

A chave aqui é compartilhar o gerador ( gen) entre os trabalhadores. Obviamente, não há sentido em processar se o array estiver vazio, então tiramos esse caso extremo da linha quatro. Além disso, não faz sentido ter mais trabalhadores do que elementos do array, portanto, na linha 10, garantimos que limit é no máximo igual ao comprimento da matriz.

Conclusão

O limit padrão é o comprimento do array, o que faz com que mapAllSettled() se comporte exatamente como Promise.allSettled() porque todas as funções do mapa serão executadas em paralelo. Mas o objetivo dessa função é dar aos usuários o controle para definir um número menor de paralelismo.

O código completo está no Github se você quiser brincar com ele (licença MIT).

Obrigado pela leitura. Se você tiver comentários ou perguntas, entre em contato no Twitter .

---

Créditos

• async map with limited parallelism in Node.js, escrito originalmente por Alex Ewerlöf.