Best way in .NET to manage queue of tasks on a separate (single) thread

Question

I know that asynchronous programming has seen a lot of changes over the years. I'm somewhat embarrassed that I let myself get this rusty at just 34 years old, but I'm counting on StackOverflow to bring me up to speed.

What I am trying to do is manage a queue of "work" on a separate thread, but in such a way that only one item is processed at a time. I want to post work on this thread and it doesn't need to pass anything back to the caller. Of course I could simply spin up a new Thread object and have it loop over a shared Queue object, using sleeps, interrupts, wait handles, etc. But I know things have gotten better since then. We have BlockingCollection, Task, async/await, not to mention NuGet packages that probably abstract a lot of that.

I know that "What's the best..." questions are generally frowned upon so I'll rephrase it by saying "What is the currently recommended..." way to accomplish something like this using built-in .NET mechanisms preferably. But if a third party NuGet package simplifies things a bunch, it's just as well.

I considered a TaskScheduler instance with a fixed maximum concurrency of 1, but seems there is probably a much less clunky way to do that by now.

Background

Specifically, what I am trying to do in this case is queue an IP geolocation task during a web request. The same IP might wind up getting queued for geolocation multiple times, but the task will know how to detect that and skip out early if it's already been resolved. But the request handler is just going to throw these () => LocateAddress(context.Request.UserHostAddress) calls into a queue and let the LocateAddress method handle duplicate work detection. The geolocation API I am using doesn't like to be bombarded with requests which is why I want to limit it to a single concurrent task at a time. However, it would be nice if the approach was allowed to easily scale to more concurrent tasks with a simple parameter change.

Answer 1

Actually you don't need to run tasks in one thread, you need them to run serially (one after another), and FIFO. Here is my very lightweight, non-blocking Task-based implementation with tests. https://github.com/Gentlee/SerialQueue

Comparison:

• Unlike TPL Dataflow ActionBlock from @i3arnon answer, my enqueue method completes when task completes executing, and it throws exceptions if one happened in the task. Also, it is a bit faster.

• Benchmark shows that @Servy implementation with SemaphoreSlim is much slower than mine.

Example:

private readonly SerialQueue queue = new SerialQueue();

async Task SomeAsyncMethod()
{
    var result = await queue.Enqueue(DoSomething);
}

Benchmark (lower - better):

Answer 2

I know this thread is old, but it seems all the present solutions are extremely onerous. The simplest way I could find uses the Linq Aggregate function to create a daisy-chained list of tasks.

var arr = new int[] { 1, 2, 3, 4, 5};
var queue = arr.Aggregate(Task.CompletedTask, 
    (prev, item) => prev.ContinueWith(antecedent => PerformWorkHere(item)));

The idea is to get your data into an IEnumerable (I'm using an int array), and then reduce that enumerable to a chain of tasks, starting with a default, completed, task.

Answer 3

My lib, It can:

1. Run random in queue list
2. Multi queue
3. Run prioritize first
4. Re-queue
5. Event all queue completed
6. Cancel running or cancel wait for running
7. Dispatch event to UI thread

---

public interface IQueue
  {
    bool IsPrioritize { get; }
    bool ReQueue { get; }
    /// <summary>
    /// Dont use async
    /// </summary>
    /// <returns></returns>
    Task DoWork();
    bool CheckEquals(IQueue queue);
    void Cancel();
  }

  public delegate void QueueComplete<T>(T queue) where T : IQueue;
  public delegate void RunComplete();
  public class TaskQueue<T> where T : IQueue
  {
    readonly List<T> Queues = new List<T>();
    readonly List<T> Runnings = new List<T>();

    [Browsable(false), DefaultValue((string)null)]
    public Dispatcher Dispatcher { get; set; }
    public event RunComplete OnRunComplete;
    public event QueueComplete<T> OnQueueComplete;
    int _MaxRun = 1;
    public int MaxRun
    {
      get { return _MaxRun; }
      set
      {
        bool flag = value > _MaxRun;
        _MaxRun = value;
        if (flag && Queues.Count != 0) RunNewQueue();
      }
    }
    public int RunningCount
    {
      get { return Runnings.Count; }
    }
    public int QueueCount
    {
      get { return Queues.Count; }
    }

    public bool RunRandom { get; set; } = false;

    //need lock Queues first
    void StartQueue(T queue)
    {
      if (null != queue)
      {
        Queues.Remove(queue);
        lock (Runnings) Runnings.Add(queue);
        queue.DoWork().ContinueWith(ContinueTaskResult, queue);
      }
    }

    void RunNewQueue()
    {
      lock (Queues)//Prioritize
      {
        foreach (var q in Queues.Where(x => x.IsPrioritize)) StartQueue(q);
      }

      if (Runnings.Count >= MaxRun) return;//other
      else if (Queues.Count == 0)
      {
        if (Runnings.Count == 0 && OnRunComplete != null)
        {
          if (Dispatcher != null && !Dispatcher.CheckAccess()) Dispatcher.Invoke(OnRunComplete);
          else OnRunComplete.Invoke();//on completed
        }
        else return;
      }
      else
      {
        lock (Queues)
        {
          T queue;
          if (RunRandom) queue = Queues.OrderBy(x => Guid.NewGuid()).FirstOrDefault();
          else queue = Queues.FirstOrDefault();
          StartQueue(queue);
        }
        if (Queues.Count > 0 && Runnings.Count < MaxRun) RunNewQueue();
      }
    }

    void ContinueTaskResult(Task Result, object queue_obj) => QueueCompleted((T)queue_obj);

    void QueueCompleted(T queue)
    {
      lock (Runnings) Runnings.Remove(queue);
      if (queue.ReQueue) lock (Queues) Queues.Add(queue);
      if (OnQueueComplete != null)
      {
        if (Dispatcher != null && !Dispatcher.CheckAccess()) Dispatcher.Invoke(OnQueueComplete, queue);
        else OnQueueComplete.Invoke(queue);
      }
      RunNewQueue();
    }

    public void Add(T queue)
    {
      if (null == queue) throw new ArgumentNullException(nameof(queue));
      lock (Queues) Queues.Add(queue);
      RunNewQueue();
    }

    public void Cancel(T queue)
    {
      if (null == queue) throw new ArgumentNullException(nameof(queue));
      lock (Queues) Queues.RemoveAll(o => o.CheckEquals(queue));
      lock (Runnings) Runnings.ForEach(o => { if (o.CheckEquals(queue)) o.Cancel(); });
    }

    public void Reset(T queue)
    {
      if (null == queue) throw new ArgumentNullException(nameof(queue));
      Cancel(queue);
      Add(queue);
    }

    public void ShutDown()
    {
      MaxRun = 0;
      lock (Queues) Queues.Clear();
      lock (Runnings) Runnings.ForEach(o => o.Cancel());
    }
  }

Answer 4

I'm posting a different solution here. To be honest I'm not sure whether this is a good solution.

I'm used to use BlockingCollection to implement a producer/consumer pattern, with a dedicated thread consuming those items. It's fine if there are always data coming in and consumer thread won't sit there and do nothing.

I encountered a scenario that one of the application would like to send emails on a different thread, but total number of emails is not that big. My initial solution was to have a dedicated consumer thread (created by Task.Run()), but a lot of time it just sits there and does nothing.

Old solution:

private readonly BlockingCollection<EmailData> _Emails =
    new BlockingCollection<EmailData>(new ConcurrentQueue<EmailData>());

// producer can add data here
public void Add(EmailData emailData)
{
    _Emails.Add(emailData);
}

public void Run()
{
    // create a consumer thread
    Task.Run(() => 
    {
        foreach (var emailData in _Emails.GetConsumingEnumerable())
        {
            SendEmail(emailData);
        }
    });
}

// sending email implementation
private void SendEmail(EmailData emailData)
{
    throw new NotImplementedException();
}

As you can see, if there are not enough emails to be sent (and it is my case), the consumer thread will spend most of them sitting there and do nothing at all.

I changed my implementation to:

// create an empty task
private Task _SendEmailTask = Task.Run(() => {});

// caller will dispatch the email to here
// continuewith will use a thread pool thread (different to
// _SendEmailTask thread) to send this email
private void Add(EmailData emailData)
{
    _SendEmailTask = _SendEmailTask.ContinueWith((t) =>
    {
        SendEmail(emailData);
    });
}

// actual implementation
private void SendEmail(EmailData emailData)
{
    throw new NotImplementedException();
}

It's no longer a producer/consumer pattern, but it won't have a thread sitting there and does nothing, instead, every time it is to send an email, it will use thread pool thread to do it.

Answer 5

To create an asynchronous single degree of parallelism queue of work you can simply create a SemaphoreSlim, initialized to one, and then have the enqueing method await on the acquisition of that semaphore before starting the requested work.

public class TaskQueue
{
    private SemaphoreSlim semaphore;
    public TaskQueue()
    {
        semaphore = new SemaphoreSlim(1);
    }

    public async Task<T> Enqueue<T>(Func<Task<T>> taskGenerator)
    {
        await semaphore.WaitAsync();
        try
        {
            return await taskGenerator();
        }
        finally
        {
            semaphore.Release();
        }
    }
    public async Task Enqueue(Func<Task> taskGenerator)
    {
        await semaphore.WaitAsync();
        try
        {
            await taskGenerator();
        }
        finally
        {
            semaphore.Release();
        }
    }
}

Of course, to have a fixed degree of parallelism other than one simply initialize the semaphore to some other number.

Answer 6

Your best option as I see it is using TPL Dataflow's ActionBlock:

var actionBlock = new ActionBlock<string>(address =>
{
    if (!IsDuplicate(address))
    {
        LocateAddress(address);
    }
});

actionBlock.Post(context.Request.UserHostAddress);

TPL Dataflow is robust, thread-safe, async-ready and very configurable actor-based framework (available as a nuget)

Here's a simple example for a more complicated case. Let's assume you want to:

• Enable concurrency (limited to the available cores).
• Limit the queue size (so you won't run out of memory).
• Have both LocateAddress and the queue insertion be async.
• Cancel everything after an hour.

var actionBlock = new ActionBlock<string>(async address =>
{
    if (!IsDuplicate(address))
    {
        await LocateAddressAsync(address);
    }
}, new ExecutionDataflowBlockOptions
{
    BoundedCapacity = 10000,
    MaxDegreeOfParallelism = Environment.ProcessorCount,
    CancellationToken = new CancellationTokenSource(TimeSpan.FromHours(1)).Token
});

await actionBlock.SendAsync(context.Request.UserHostAddress);

Answer 7

Use BlockingCollection<Action> to create a producer/consumer pattern with one consumer (only one thing running at a time like you want) and one or many producers.

First define a shared queue somewhere:

BlockingCollection<Action> queue = new BlockingCollection<Action>();

In your consumer Thread or Task you take from it:

//This will block until there's an item available
Action itemToRun = queue.Take()

Then from any number of producers on other threads, simply add to the queue:

queue.Add(() => LocateAddress(context.Request.UserHostAddress));