MultiProducerConcurrentConsumer – Start it

In the previous instalment, we discussed the technique of preallocating and reuse to save allocations during runtime. In this post, I will introduce the heart of the MPCC which is the start method and the timer loop outline.

The MPCC can get started at any time after construction. If the component is started, it will start pushing items to the pump delegate whenever the batch size is reached or the push interval. This is how the Start method looks like

public void Start(Func<List<TItem>, int, object, CancellationToken, Task> pump)
{
    Start(pump, null);
}

public void Start(Func<List<TItem>, int, object, CancellationToken, Task> pump, object state)
{
    if (started)
    {
        throw new InvalidOperationException("Already started");
    }

    tokenSource = new CancellationTokenSource();
    timer = Task.Run(TimerLoop);
    this.pump = pump;
    this.state = state;
    started = true;
}

The MPCC can be started once only. There are two variations for starting, a state based overload and a func which doesn’t provide a state. The state based overload is here to enable scenarios where you want to avoid closure capturing. So you can pass in the state that is required to execute the functionality inside the pump delegate implementation.

Furthermore, the component requires a cancellation token source and a timer loop that executes the interval based pushing of items to the pump. Let’s have a quick look at the timer loop implementation (a few details are left out since we will be exploring those in future posts):

async Task TimerLoop()
{
    var token = tokenSource.Token;
    while (!tokenSource.IsCancellationRequested)
    {
        try
        {
            await Task.Delay(pushInterval, token).ConfigureAwait(false);
            await PushInBatches().ConfigureAwait(false);
        }
        catch (Exception)
        {
            // intentionally ignored
        }
    }
}

The timer loop is scheduled to be executed on the worker thread pool. In its heart, we have a while loop that loops until the token indicates a cancellation was requested. Inside the loop, the component uses a Task.Delay to asynchronously wait until the push interval was reached or the token was cancelled. After the push interval is over the method PushInBatches is called. In this implementation, I choose to ignore any exceptions that might be happening either in the Task.Delay (i.ex. OperationCanceledException was thrown) or exceptions bubbling up from PushInBatches.

Now we have a simple approach to push items in batches based on the push interval defined in the constructor. But this doesn’t address the batch size based pushing of items. We’ll cover this in the next posts.