1
\$\begingroup\$

I have made a multithreaded command runner that works, just wanted to share the code and see if you guys can see any pitfalls. Comments are not part of actual code

protected async Task<bool> PollQueue(CancellationToken cancellationToken)
{
    async Task<Batch> SyncQueueTask(CancellationToken token)
    {
        await WaitForQueue(token);
        return null;
    }

    var queue = new Queue<Batch>();
    var runners = new Dictionary<Guid, Task<Batch>>();

    Task<Batch> syncTask = null;
    var cancelSyncTask = new CancellationTokenSource();
    bool QueueWasReleased() => syncTask == null;

    do
    {
        //Do not spawn new work if application is closing down, but let ongoing work have a chance to complete
        if (!cancellationToken.IsCancellationRequested)
        {
            var slots = Environment.ProcessorCount - runners.Count;
            //Poll SQL queue if slots are available and we have gotten a signal that work is is available
            if (QueueWasReleased())
            {
                using (var scope = _serviceProvider.CreateScope())
                {
                    var batch = (await scope.ServiceProvider.GetService<ICommandRepository>()
                            .ListByAsync(CommandState.Pending))
                            .Where(c => !runners.ContainsKey(c.BatchId))
                            .GroupBy(c => c.BatchId)
                            .Select(grp => new Batch {Id = grp.Key, Commands = grp.ToList()});

                    queue = new Queue<Batch>(batch);
                }
            }
            //Start new work 
            var tempQueue = queue;
            Enumerable
                .Range(0, Math.Min(slots, queue.Count))
                .ForEach(i =>
                {
                    var batch = tempQueue.Dequeue();
                    var runner = Task.Run(async () =>
                    {
                        await batch
                            .Commands
                            .Select(ExecuteCommand)
                            .WaitAll(cancellationToken);

                        return batch;
                    }, CancellationToken.None);
                    runners[batch.Id] = runner;
                });
        }

        if (runners.Count == 0) break;
        //Create a sync task thats only purpose is to release await if new work is available. This is to that long running work does not starve queue
        if (syncTask == null)
            syncTask = SyncQueueTask(cancelSyncTask.Token);

        //Wait for next completed work or that new work are avialable on queue    
        var completed = await Task.WhenAny(runners.Values.Concat(syncTask));
        if (completed == syncTask)
            syncTask = null;
        else
            runners.Remove(completed.Result.Id);

    } while (true);

    //Cleanup if we are waiting
    if (syncTask != null)
        cancelSyncTask.Cancel();

    return false;
}

WaitForQueue is just a semaphoreslim. (With an extensionmethod that swallows cancelation exceptions)

private Task WaitForQueue(CancellationToken token)
{
    return _queueSync.SilentWaitAsync(Delay, token);
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.