Implementation of multi-threaded consumer by message shard ID

Question

I have a command processor that i am rewriting to be sharded across a number of threads. I am going from a single threaded sequential sender. Where we have the delegate func passed into the method which then decides how to call the delegate:

internal class SingleThreadedSendStrategy
{
    public Task Send(Func, CancellationToken, Task> func, IRequest request, CancellationToken cancellationToken)
    {
        return func(request, cancellationToken);
    }
}

To a multi-threaded (n threads) setup where each thread has a queue, and commands are appended to a specific queue depending on the message hashcode as below:

internal class ShardedSendStrategy
{
    const int QueueCount = 8;
    readonly BlockingCollection[] Queues = new BlockingCollection[QueueCount];
    readonly CancellationToken CancellationToken;

    public ShardedSendStrategy(CancellationToken cancellationToken)
    {
        CancellationToken = cancellationToken;

        for (int i = 0; i < QueueCount; i++)
        {
            Queues[i] = new BlockingCollection();
            var thread = new Thread( () => OnHandlerStart(Queues[i])) { IsBackground = true };
            thread.Start();
        }
    }

    public Task Send(Func, CancellationToken, Task> func, IRequest request, CancellationToken cancellationToken = default)
    {
        var shard = request.GetHashCode() % QueueCount;

        var task = new Task(() => func(request, cancellationToken));
        Queues[shard].Add(task);

        return task; // we have a problem here, Task is expected
    }

    private void OnHandlerStart(BlockingCollection queue)
    {
        foreach (var job in queue.GetConsumingEnumerable(CancellationToken))
        {
            job.Start();
        }
    }
}

However, if in the Send function i sent the task to var task = new Task(() => func(request, cancellationToken)); I cannot return the task as the compiler throws Cannot implicitly convert type 'System.Threading.Tasks.Task' to 'System.Threading.Tasks.Task'..

If in the Send function i set var task = new Task( () => { return func(request, cancellationToken); }); it throws Cannot convert lambda expression to intended delegate type because some of the return types in the block are not implicitly convertible to the delegate return type.

How can i resolve the cast errors please so that I can have the job get dequeued and ran in the consumer (i am not fussed about any specific object types, since we set the function to call in the Task constructor)?

How also would I comminicate back to the calling thread that called Send, that the function completed in the OnHandlerStart method?

Any other best practices that I should be implementing for this multi-queue consumer setup, and if i am reinventing the wheel some code on this use case using existing TPL structure would be much appreciated?

Luaan · Accepted Answer

The Task constructor doesn't unwrap the task in the delegate. It isn't aware of it at all, and doesn't care about it. This is one of the things you get for using the "advanced scenario" - don't expect that new Task works the same as Task.Run, just without starting the delegate. It doesn't.

A more typical way of implementing what you're trying to do would be making your own task scheduler. But if you want to use your approach, you need to do the unwrapping manually. This can be a bit more complicated than it sounds, but fortunately, there's a method that does exactly what you need:

var task = new Task>(() => func(request, cancellationToken)).Unwrap();

Less fortunately, this is still a promise-style task, and cannot be started. So you need to go even further:

internal class ShardedSendStrategy
{
  const int QueueCount = 8;
  readonly BlockingCollection[] Queues 
    = new BlockingCollection[QueueCount];
  readonly CancellationToken CancellationToken;

  public ShardedSendStrategy(CancellationToken cancellationToken)
  {
    CancellationToken = cancellationToken;

    for (int i = 0; i < QueueCount; i++)
    {
      var id = i;
      Queues[id] = new BlockingCollection();
      var thread = new Thread( () => OnHandlerStart(Queues[id])) 
                             { IsBackground = true };
      thread.Start();
    }
  }

  public Task Send(
    Func, CancellationToken, Task> func, 
    IRequest request, CancellationToken cancellationToken = default)
  {
    var shard = request.GetHashCode() % QueueCount;
    var tcs = new TaskCompletionSource();

    Queues[shard].Add(() => {
      var result = func(request, cancellationToken).Result;

      tcs.SetResult(result);
    });

    return tcs.Task;
  }

  private void OnHandlerStart(BlockingCollection queue)
  {
    foreach (var job in queue.GetConsumingEnumerable(CancellationToken))
    {
      job();
    }
  }
}

The main problem is that this waits for the result of the task synchronously on your background thread. This can be a problem if there's dependencies between your tasks - you could get a deadlock where one task is waiting for another task to do its job, but that one can't get its work done because it's waiting for a spot in your "queue".

If you do decide to use a similar approach, make sure to add proper error handling. You need to handle exceptions, cancellation etc.

Again, implementing your own task scheduler is probably a better idea.

Implementation of multi-threaded consumer by message shard ID

Answers (2)

Related Questions