Limitación de tareas asíncronas

Digamos que tienes 1000 URLs, y solo quieres tener 50 solicitudes abiertas en una vez; pero tan pronto como se completa una solicitud, se abre una conexión a la siguiente URL de la lista. De esa manera, siempre hay exactamente 50 las conexiones se abren a la vez, hasta que la lista de URL se agota.

La siguiente solución simple ha surgido muchas veces aquí en SO. No usa código de bloqueo y no crea hilos explícitamente, por lo que escala muy bien:

const int MAX_DOWNLOADS = 50;

static async Task DownloadAsync(string[] urls)
{
    using (var semaphore = new SemaphoreSlim(MAX_DOWNLOADS))
    using (var httpClient = new HttpClient())
    {
        var tasks = urls.Select(async url => 
        {
            await semaphore.WaitAsync();
            try
            {
                var data = await httpClient.GetStringAsync(url);
                Console.WriteLine(data);
            }
            finally
            {
                semaphore.Release();
            }
        });

        await Task.WhenAll(tasks);
    }
}

La cosa es, el procesamiento de los datos descargados debe hacerse en una tubería diferente, con un nivel de paralelismo diferente, especialmente si se trata de un procesamiento vinculado a la CPU.

Por ejemplo, es probable que desee tener 4 subprocesos simultáneamente haciendo el procesamiento de datos (el número de núcleos de CPU), y hasta 50 solicitudes pendientes de más datos (que no utilizan subprocesos en absoluto). AFAICT, esto no es lo que su código está haciendo actualmente.

Ahí es donde TPL Dataflow o Rx pueden sea útil como una solución preferida. Sin embargo, es ciertamente posible implementar algo como esto con TPL simple. Nota, el único código de bloqueo aquí es el que hace el procesamiento de datos real dentro de Task.Run:

const int MAX_DOWNLOADS = 50;
const int MAX_PROCESSORS = 4;

// process data
class Processing
{
    SemaphoreSlim _semaphore = new SemaphoreSlim(MAX_PROCESSORS);
    HashSet<Task> _pending = new HashSet<Task>();
    object _lock = new Object();

    async Task ProcessAsync(string data)
    {
        await _semaphore.WaitAsync();
        try
        {
            await Task.Run(() =>
            {
                // simuate work
                Thread.Sleep(1000);
                Console.WriteLine(data);
            });
        }
        finally
        {
            _semaphore.Release();
        }
    }

    public async void QueueItemAsync(string data)
    {
        var task = ProcessAsync(data);
        lock (_lock)
            _pending.Add(task);
        try
        {
            await task;
        }
        catch
        {
            if (!task.IsCanceled && !task.IsFaulted)
                throw; // not the task's exception, rethrow
            // don't remove faulted/cancelled tasks from the list
            return;
        }
        // remove successfully completed tasks from the list 
        lock (_lock)
            _pending.Remove(task);
    }

    public async Task WaitForCompleteAsync()
    {
        Task[] tasks;
        lock (_lock)
            tasks = _pending.ToArray();
        await Task.WhenAll(tasks);
    }
}

// download data
static async Task DownloadAsync(string[] urls)
{
    var processing = new Processing();

    using (var semaphore = new SemaphoreSlim(MAX_DOWNLOADS))
    using (var httpClient = new HttpClient())
    {
        var tasks = urls.Select(async (url) =>
        {
            await semaphore.WaitAsync();
            try
            {
                var data = await httpClient.GetStringAsync(url);
                // put the result on the processing pipeline
                processing.QueueItemAsync(data);
            }
            finally
            {
                semaphore.Release();
            }
        });

        await Task.WhenAll(tasks.ToArray());
        await processing.WaitForCompleteAsync();
    }
}

Como se solicitó, aquí está el código con el que terminé yendo.

El trabajo se configura en una configuración master-detail, y cada master se procesa como un lote. Cada unidad de trabajo está en cola de esta manera:

var success = true;

// Start processing all the master records.
Master master;
while (null != (master = await StoredProcedures.ClaimRecordsAsync(...)))
{
    await masterBuffer.SendAsync(master);
}

// Finished sending master records
masterBuffer.Complete();

// Now, wait for all the batches to complete.
await batchAction.Completion;

return success;

Los maestros se almacenan uno a la vez para guardar el trabajo para otros procesos externos. Los detalles de cada maestro se envían para el trabajo a través de la masterTransform TransformManyBlock. También se crea un BatchedJoinBlock para recopilar los detalles en un lote.

El trabajo real se realiza en el detailTransform TransformBlock, asincrónicamente, 150 a la vez. BoundedCapacity se establece en 300 para asegurar que demasiados Maestros no se almacenen en búfer al principio de la cadena, mientras que también deja espacio para que se pongan en cola suficientes registros de detalle para permitir que se procesen 150 registros a la vez. El bloque envía un object a sus destinos, porque se filtra a través de los enlaces dependiendo de si se trata de un Detail o Exception.

El batchAction ActionBlock recopila la salida de todos los lotes y realiza actualizaciones masivas de la base de datos, error registro, etc. para cada lote.

Habrá varios BatchedJoinBlock s, uno para cada maestro. Dado que cada ISourceBlock se genera secuencialmente y cada lote solo acepta el número de registros de detalles asociados con un maestro, los lotes se procesarán en orden. Cada bloque solo genera un grupo y se desvincula al completarse. Solo el último bloque de lotes propaga su finalización al ActionBlock final.

La red de flujo de datos:

// The dataflow network
BufferBlock<Master> masterBuffer = null;
TransformManyBlock<Master, Detail> masterTransform = null;
TransformBlock<Detail, object> detailTransform = null;
ActionBlock<Tuple<IList<object>, IList<object>>> batchAction = null;

// Buffer master records to enable efficient throttling.
masterBuffer = new BufferBlock<Master>(new DataflowBlockOptions { BoundedCapacity = 1 });

// Sequentially transform master records into a stream of detail records.
masterTransform = new TransformManyBlock<Master, Detail>(async masterRecord =>
{
    var records = await StoredProcedures.GetObjectsAsync(masterRecord);

    // Filter the master records based on some criteria here
    var filteredRecords = records;

    // Only propagate completion to the last batch
    var propagateCompletion = masterBuffer.Completion.IsCompleted && masterTransform.InputCount == 0;

    // Create a batch join block to encapsulate the results of the master record.
    var batchjoinblock = new BatchedJoinBlock<object, object>(records.Count(), new GroupingDataflowBlockOptions { MaxNumberOfGroups = 1 });

    // Add the batch block to the detail transform pipeline's link queue, and link the batch block to the the batch action block.
    var detailLink1 = detailTransform.LinkTo(batchjoinblock.Target1, detailResult => detailResult is Detail);
    var detailLink2 = detailTransform.LinkTo(batchjoinblock.Target2, detailResult => detailResult is Exception);
    var batchLink = batchjoinblock.LinkTo(batchAction, new DataflowLinkOptions { PropagateCompletion = propagateCompletion });

    // Unlink batchjoinblock upon completion.
    // (the returned task does not need to be awaited, despite the warning.)
    batchjoinblock.Completion.ContinueWith(task =>
    {
        detailLink1.Dispose();
        detailLink2.Dispose();
        batchLink.Dispose();
    });

    return filteredRecords;
}, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });

// Process each detail record asynchronously, 150 at a time.
detailTransform = new TransformBlock<Detail, object>(async detail => {
    try
    {
        // Perform the action for each detail here asynchronously
        await DoSomethingAsync();

        return detail;
    }
    catch (Exception e)
    {
        success = false;
        return e;
    }

}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 150, BoundedCapacity = 300 });

// Perform the proper action for each batch
batchAction = new ActionBlock<Tuple<IList<object>, IList<object>>>(async batch =>
{
    var details = batch.Item1.Cast<Detail>();
    var errors = batch.Item2.Cast<Exception>();

    // Do something with the batch here
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });

masterBuffer.LinkTo(masterTransform, new DataflowLinkOptions { PropagateCompletion = true });
masterTransform.LinkTo(detailTransform, new DataflowLinkOptions { PropagateCompletion = true });