MultiProducerConcurrentConsumer – Push in batches

In the last post, I introduced the push method and the timer loop of the MultiProducerConcurrentConsumer. In this post, I’ll focus on the actual PushInBatches method that gets called by the timer loop.

Task PushInBatches() {
    if (Interlocked.Read(ref numberOfPushedItems) == 0) {
        return TaskEx.Completed;
    }

    for (var i = 0; i < numberOfSlots; i++) { 
       var queue = queues[i]; 

       PushInBatchesUpToConcurrencyPerQueueForAGivenSlot(queue, i, pushTasks); 
    } 

    return Task.WhenAll(pushTasks).ContinueWith((t, s) => {
        var tasks = (List<Task>) s;
        tasks.Clear();
    }, pushTasks, TaskContinuationOptions.ExecuteSynchronously);
}

When the method is entered it first checks whether there is anything to push. Since numberOfPushedItems is accessed by multiple threads it uses Interlocked to read it and compare the returned value. When there is nothing to push a completed task is return. By not using the async keyword here we avoid the state machine generated by the compiler. Bear in mind this trick should not be blindly applied.

Then it loops through all the slots. The queue per slot is taken from the queues array, and PushInBatchesUpToConcurrencyPerQueueForAGivenSlot is called with the queue of the current slot and the pushTasks list. Remember the push task list and the queues array was previously allocated as members of the class for efficient reuse.

At the end of the method, a task that completes is created when all of the tasks contained in pushTasks are completed with Task.WhenAll. A synchronous continuation is scheduled that will make sure the pushTasks list gets cleared at the end of the execution. That continuation task is returned to the caller which is the timer loop. This means the timer loop will only continue when the pushTask list is cleared. So instead of creating a list of tasks for every loop cycle we reuse the task list and clear it after each push cycle which reduces the garbage collection pressure.

Here I used another not so well known trick. Instead of accessing the pushTasks list directly and therefore creating a closure over this, I’m passing the pushTasks list as a state parameter to the ContinueWith method and extract the state inside the body of the continuation. Which save an Action delegate allocation per push cycle.

In the next installment, we’ll look into PushInBatchesUpToConcurrencyPerQueueForAGivenSlot since it is complex enough to warrant a dedicated post.