TaskGroup: paralelismo dinámico con resultados y errores

Cuando el número de trabajos deja de estar escrito en el código y pasa a depender de una colección, de una respuesta del servidor o de una decisión del usuario, la aridad estática de async let se queda corta y hace falta una estructura que lleve la cuenta en ejecución sin renunciar a ninguna garantía. Eso es un grupo de tareas: un ámbito con nombre, creado con withTaskGroup o su variante que lanza, al que se le añaden hijas dinámicamente y que no puede cerrarse mientras alguna siga viva. La diferencia con una lista de trabajos pendientes escrita a mano es que el grupo no es un contenedor pasivo: es a la vez el dueño de las hijas, la secuencia asíncrona por la que llegan sus resultados y el mecanismo que las cancela a todas cuando algo va mal.

El ámbito con nombre

El grupo se abre con una función que recibe una clausura; dentro de ella, la variable del grupo permite añadir hijas. Al cerrarse la clausura, ninguna hija sigue viva.

func miniaturas(de urls: [URL]) async -> [Miniatura] {
    await withTaskGroup(of: Miniatura?.self) { grupo in
        for url in urls {
            grupo.addTask { await generarMiniatura(url) }   // hija por elemento
        }

        var salida: [Miniatura] = []
        for await resultado in grupo {                      // llegan segun terminan
            if let m = resultado { salida.append(m) }
        }
        return salida
    }
}

Tres detalles cargan con casi toda la semántica. El primero es of:, que fija el tipo común de resultado de todas las hijas; si los trabajos devuelven cosas distintas hay que unificarlos en un enum o en una tupla, y esa fricción es una señal útil de que quizá lo que querías era async let. El segundo es que las hijas empiezan en cuanto se añaden, no al iterar. El tercero es que el grupo se comporta como una AsyncSequence: iterar sobre él consume resultados en orden de finalización, no de creación.

Si el orden importa, la solución idiomática es devolver el índice junto al valor y reordenar al final, no intentar forzar la llegada.

await withTaskGroup(of: (Int, Miniatura?).self) { grupo in
    for (i, url) in urls.enumerated() {
        grupo.addTask { (i, await generarMiniatura(url)) }
    }
    var buffer = [Miniatura?](repeating: nil, count: urls.count)
    for await (i, m) in grupo { buffer[i] = m }
    return buffer.compactMap { $0 }
}

Errores: propagar o contener

Con withThrowingTaskGroup, una hija que lanza no derriba el grupo por sí sola: el error se materializa cuando alguien lo consume, es decir, en el for try await o en un next explícito. En ese momento se cancela el resto de hijas, se espera a que terminen y el error sale del grupo.

try await withThrowingTaskGroup(of: Documento.self) { grupo in
    for id in ids { grupo.addTask { try await descargar(id) } }
    var docs: [Documento] = []
    for try await d in grupo { docs.append(d) }   // el primer fallo aborta todo
    return docs
}

Esa es la política de «todo o nada», adecuada cuando un resultado parcial no tiene sentido. La política opuesta —seguir adelante y reportar los fallos— no se consigue con try sino haciendo que el fallo forme parte del tipo de resultado.

await withTaskGroup(of: Result<Documento, Error>.self) { grupo in
    for id in ids {
        grupo.addTask {
            do    { return .success(try await descargar(id)) }
            catch { return .failure(error) }
        }
    }
    var ok: [Documento] = []
    var fallos: [Error] = []
    for await r in grupo {
        switch r {
        case .success(let d): ok.append(d)
        case .failure(let e): fallos.append(e)
        }
    }
    return Informe(exitos: ok, errores: fallos)
}

La elección entre ambas no es estilística. «Todo o nada» convierte cualquier fallo en el fallo de la operación completa; la contención por Result mantiene vivo el trabajo restante y traslada la decisión a quien llama. Sincronizar veinte ficheros pide lo segundo; construir un objeto que necesita sus cinco partes pide lo primero.

flowchart TB
G[withThrowingTaskGroup abre el ambito] --> A1[addTask 1]
G --> A2[addTask 2]
G --> A3[addTask 3]
A1 --> C[for try await consume resultados]
A2 --> C
A3 --> C
C -->|una hija lanza| K[Se cancela el resto del grupo]
K --> W[Se espera a todas las hijas]
W --> E[El error sale del ambito]
C -->|todas terminan bien| R[Resultado agregado]
style C fill:#89b4fa,color:#11111b
style E fill:#f38ba8,color:#11111b

Limitar la concurrencia

Añadir diez mil hijas de golpe compila y funciona, pero suele ser un error de ingeniería: cada una reserva memoria, y diez mil conexiones simultáneas saturan la red, el disco o el servidor remoto mucho antes de mejorar el rendimiento. El patrón correcto es una ventana deslizante: llenar el grupo hasta un máximo y, por cada resultado consumido, añadir el siguiente trabajo.

func procesar(_ items: [Item], maximo: Int = 6) async -> [Salida] {
    await withTaskGroup(of: Salida.self) { grupo in
        var indice = 0
        var salidas: [Salida] = []

        while indice < min(maximo, items.count) {        // llenar la ventana
            let item = items[indice]
            grupo.addTask { await trabajar(item) }
            indice += 1
        }

        for await s in grupo {                            // uno sale, uno entra
            salidas.append(s)
            if indice < items.count {
                let item = items[indice]
                grupo.addTask { await trabajar(item) }
                indice += 1
            }
        }
        return salidas
    }
}

El máximo adecuado no se adivina: depende del recurso que satura primero. Para trabajo ligado a CPU, el número de núcleos activos es un punto de partida razonable; para red, lo marca el servidor y la latencia; para disco, el dispositivo. Medir dos o tres valores es más barato que discutirlo.