Loop for periodic processing in a background thread

Question

Occasionally I need to implement periodic checks in a loop in a background thread, a typical example being asynchronous processing using a message queue. While it’s not terribly complicated, I wonder what simpler, more elegant solutions exist.

I post my solution, and I’m eager to hear any constructive criticism!

The requirements in more details are these: I need:

• A loop that does periodic checks and actions if needed, and blocks in-between
• It runs in a background thread (since it does an active loop)
• It can be stopped properly (e.g. the loop exits and the thread freed)
• The blocking wait can be awaken if needed (e.g. new entry in queue, or exit)
• Thread safety

My solution:

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Threading;

namespace BgLoopExample
{
    /// <summary>
    /// Perform asynchronous processing in a background thread.
    /// Handle thread creation, exit, thread safety.
    /// </summary>
    public class BgLoopExample : IDisposable
    {
        /// <summary>
        /// Queue for the incoming messages.
        /// </summary>
        private Queue<Message> myQueue;

        /// <summary>
        /// Task for background processing.
        /// </summary>
        private Task myBgLoopTask;

        /// <summary>
        /// Event for awakening the processing loop, used when new entry is added to the queue or exit requested.
        /// </summary>
        private AutoResetEvent myQueueEvent;

        /// <summary>
        /// Flag to signal stop for the parallel sender thread.
        /// </summary>
        private bool myStopTaskFlag;

        /// <summary>
        /// Flag to store if this class is disposing.
        /// </summary>
        private bool myDisposed;

        /// <summary>
        /// Class initializer, start the background thread.
        /// </summary>
        public void Init()
        {
            myQueue = new Queue<Message>();
            myQueueEvent = new AutoResetEvent(false);
            myStopTaskFlag = false;
            myDisposed = false;

            StartSenderTask();
        }

        /// <summary>
        /// Adds new event to the Queue
        /// </summary>
        /// <param name="message">New event to add to the queue.</param>
        public void AddMessage(Message message)
        {
            lock (myQueue)
            {
                if (myStopTaskFlag)
                {
                    // don't add any more if stop has been requested
                    return;
                }
                myQueue?.Enqueue(message);
                myQueueEvent?.Set();
            }
        }

        /// <summary>
        /// Starts the sender parallel task.
        /// </summary>
        private void StartSenderTask()
        {
            myStopTaskFlag = false;
            myBgLoopTask = new Task(BgLoop);
            myBgLoopTask.Start();
        }

        /// <summary>
        /// Stops the sender parallel task, but send all data before exiting.
        /// </summary>
        private void StopSenderTask()
        {
            // set the stop flag, and signal change
            myStopTaskFlag = true;
            myQueueEvent.Set();
            // wait till the loop exits
            myBgLoopTask.Wait();
            Console.WriteLine("Thread exited");
        }

        /// <summary>
        /// Infinite loop for processing the incoming events from the Queue.
        /// </summary>
        private void BgLoop()
        {
            while (true)
            {
                // copy the message queue if it has entry (copy for thread safety)
                List<Message> messageList = new List<Message>();
                lock (myQueue)
                {
                    if (myQueue.Count > 0)
                    {
                        while (myQueue.Count > 0)
                        {
                            messageList.Add(myQueue.Dequeue());
                        }
                    }
                }

                // Process the events
                if (messageList.Count > 0)
                {
                    ProcessMessages(messageList);
                }

                // Stop if need to stop and queue has been emptied
                lock (myQueue)
                {
                    if (myStopTaskFlag)
                    {
                        if (myQueue.Count == 0)
                        {
                            break;
                        }
                        // stop requested but not empty yet -- loop without waiting
                        continue;
                    }
                }

                // Block for a while.  The exact timeout value does not really matter
                myQueueEvent.WaitOne(10000);
            }
            Console.WriteLine("Thread exiting...");
        }

        /// <summary>
        /// Send the next group of events using the data sender.
        /// </summary>
        /// <param name="messages">Event list to send.</param>
        private void ProcessMessages(List<Message> messages)
        {
            foreach (Message m in messages)
            {
                try
                {
                    Console.WriteLine($"{DateTime.Now}: Processing message {m} ...");
                    // ... do processing ...
                }
                catch (Exception ex)
                {
                    // ...
                }
            }
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (myDisposed)
            {
                return;
            }
            if (disposing)
            {
                StopSenderTask();
                myQueueEvent.Dispose();
            }
            myDisposed = true;
        }
    }
}

Answer 1

I would say that my solution to another question applies here: How to queue up delegates to be executed in series in the background with C#?

You use a BlockingCollection class and use it's CompleteAdding() method when you want the whole thing to die. Hooking up workers to it is also easy, as I show in the other answer.

Answer 2

So the first thing is that you'll want to have a more generalized queue. You don't want to have to create a different type of queue for every situation. Write it once in such a way that each time you need it you can provide it with the operations to perform and it will perform them, without needing to know anything about those operations.

The thread pool is specifically designed for short lived operations. You're providing an operation to the thread pool that is going to be long running, and blocking it when there is nothing to do. Don't do that. Rather, each time you have an operation to run, ask the thread pool to run it, and simply don't have any thread doing anything whenever you don't have any work to do. This means that you don't have a thread pool thread sitting there doing nothing when you have no work, and it means the thread pool will be able to more effectively manage its work as you'll be using it in a way that's in line with its expected usage. Doing this also means that you have no need to stop it; since it's just doing nothing when you don't have any work to it, all you need to do to clean it up is stop giving it more work and there is no longer anything to clean up.

Doing this is actually rather straightforward. You keep a Task representing the "last" operation in the queue, and then each time you go to add a new operation to the queue you have it await that task and then set "itself" as "the last item in the queue".

public class TaskQueue
{
    private Task previous = Task.FromResult(false);
    private object key = new object();

    public Task<T> Enqueue<T>(Func<Task<T>> taskGenerator)
    {
        lock (key)
        {
            var next = AddContinuation(taskGenerator);
            previous = next;
            return next;
        }
    }

    public Task<T> Enqueue<T>(Func<T> function)
    {
        return Enqueue(() => Task.Run(function));
    }

    public Task Enqueue(Func<Task> taskGenerator)
    {
        lock (key)
        {
            var next = AddContinuation(taskGenerator);
            previous = next;
            return next;
        }
    }

    public Task Enqueue(Action action)
    {
        return Enqueue(() => Task.Run(action));
    }

    private async Task<T> AddContinuation<T>(Func<Task<T>> taskGenerator)
    {
        await previous
            .ContinueWith(t => { }); //ignore errors of previous task here
        return await taskGenerator();
    }

    private async Task AddContinuation(Func<Task> taskGenerator)
    {
        await previous
            .ContinueWith(t => { }); //ignore errors of previous task here
        await taskGenerator();
    }
}

So we still need a lock here, to make sure that two threads don't end up adding continuations to the same Task, and end up running in parallel. There are also a number of trivial variations that we can make for each overload to add useful features. First, there are generic and non-generic versions, so that the operation itself can either compute a value or not. Additionally, we can trivially create overloads for delegates to be executed in a thread pool thread, as a convenience, but more importantly, we can provide operations that return Task or Task<T>, which means the queue can support inherently asynchronous operations rather than just synchronous operations that need to be executed in a thread pool thread.

Another highly useful feature of this implementation is that every time you add an operation to the queue you're given a Task (or Task<T>) representing that operation, so you can tell when it finished, as well as being given information about whether it completed successfully or errored.

Answer 3

I don't see any value on making such abstraction.

ThreadPool already manages a queue of things to do, you only want to use it. In fact the only thing that you would have to write would be something like your ProcessMessages method:

Task.Run(() => {
    foreach (Message m in messages)
    {
        try
        {
            Console.WriteLine($"{DateTime.Now}: Processing message {m} ...");
            // ... do processing ...
        }
        catch (Exception ex)
        {
            // ...
        }
    }
});

The work is done in other thread, you don't have to deal with any concurrency issues (unless the threads are cooperative) and the job is done... just like that.

---

In case you are reinventing the wheel however:

my is not part of conventional field prefixs, it's common the use of _ instead.

Init should be your constructor. You might still want to start the task latter, in which case you should put StartSenderTask as public.

preferably AddMessage would throw a InvalidOperationException instead of returning silently and not adding any message to the queue when the looper is stopped.

You are using both a AutoResetEvent and a monitor, I bet the monitor is plenty enough as long as you know how to use Monitor.Wait and Monitor.Pulse variants.

List<Message> messageList = new List<Message>();
lock (myQueue)
{
    if (myQueue.Count > 0)
    {
        while (myQueue.Count > 0)
        {
            messageList.Add(myQueue.Dequeue());
        }
    }
}

Can be written as

List<Message> messageList;
lock(myQueue){
    messageList = myQueue.ToList();
    myQueue.Clear();
}

---

lock (myQueue)
{
    if (myStopTaskFlag)
    {
        if (myQueue.Count == 0)
        {
            break;
        }
        // stop requested but not empty yet -- loop without waiting
        continue;
    }
}

If you put some of this logic into a boolean property your code will become more readable:

private bool ShouldStop{
    lock (myQueue)
    {
        if (myStopTaskFlag)
        {
            if (myQueue.Count == 0)
            {
                return true;
            }
            // stop requested but not empty yet -- loop without waiting
            return false;
        }
    }
} 

Answer 4

/Self answer with updated code/

• I received several improvement suggestions that are relatively minor -- I have incorporated several of them, and reproduce my updated solution below.
• I received a suggestion based on BlockingCollection, I reproduce this solution as well.
• I have also received a suggestion based on Tasks and continuation, I think this is also valuable.
• I received some comments questioning the need for such an abstraction, as using threadpool processing can solve these scenarios ‘directly’. This is a useful conclusion for me, and shows that my original solution is based on traditional thread-based sequential paradigm, and a TPL-based approach can result in quite different (and likely more elegant) solutions. I also show a very simple solution inspired by this, which sovles the orginal problem too.

Here is my improved original solution, keeping the original basic idea:

/// <summary>
/// Perform asynchronous processing in a background thread.
/// Handle thread creation, exit, thread safety.
/// </summary>
public class BgLoop_ConcQueue : IBgLoop, IDisposable
{
    /// <summary>Message handling delegate.</summary>
    private ProcessMessage _processMessageDelegate;

    /// <summary>Queue for the incoming messages.</summary>
    private ConcurrentQueue<Message> _queue = new ConcurrentQueue<Message>();

    /// <summary>Task for background processing.</summary>
    private Task _bgLoopTask;

    /// <summary>Event for awakening the processing loop, used when new entry is added to the queue or exit requested.</summary>
    private AutoResetEvent _queueEvent = new AutoResetEvent(false);

    /// <summary>Flag to signal stop for the parallel sender thread.</summary>
    private bool _stopTaskFlag;

    /// <summary>Flag to store if this class is disposing</summary>
    private bool _disposed;

    /// <summary>
    /// Start the background thread.
    /// </summary>
    public void Start(ProcessMessage processMessageDelegate)
    {
        if (_bgLoopTask != null) return;  // prevent starting more than once
        _processMessageDelegate = processMessageDelegate;
        _bgLoopTask = Task.Factory.StartNew(BgLoop, TaskCreationOptions.LongRunning);
    }

    /// <summary>
    /// Adds new event to the Queue
    /// </summary>
    /// <param name="message">New event to add to the queue.</param>
    public void AddMessage(Message message)
    {
        if (_stopTaskFlag)
        {
            // don't add any more if stop has been requested
            return;
        }
        _queue?.Enqueue(message);
        _queueEvent?.Set();
    }

    /// <summary>
    /// Stops the sender parallel task, but send all data before exiting.
    /// </summary>
    private void StopSenderTask()
    {
        // set the stop flag, and signal change
        _stopTaskFlag = true;
        _queueEvent.Set();
        // wait till the loop exits
        _bgLoopTask.Wait();
        Console.WriteLine("Thread exited");
    }

    /// <summary>
    /// Infinite loop for processing the incoming events from the Queue.
    /// </summary>
    private void BgLoop()
    {
        while (true)
        {
            // copy the message queue if it has entry (copy for thread safety)
            List<Message> messageList = new List<Message>();
            while (_queue.Count > 0)
            {
                Message msg;
                if (_queue.TryDequeue(out msg))
                {
                    messageList.Add(msg);
                }
            }

            // Process the events
            if (messageList.Count > 0 && _processMessageDelegate != null)
            {
                foreach (Message m in messageList)
                {
                    _processMessageDelegate(m);
                }
            }

            // Stop if need to stop and queue has been emptied
            if (_stopTaskFlag)
            {
                if (_queue.Count == 0)
                {
                    break;
                }
                // stop requested but not empty yet -- loop without waiting
                continue;
            }

            // Block for a while, until an event happens.  The exact timeout value does not really matter
            _queueEvent.WaitOne(10000);
        }
        Console.WriteLine("Thread exiting...");
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    protected virtual void Dispose(bool disposing)
    {
        if (_disposed)
        {
            return;
        }
        if (disposing)
        {
            StopSenderTask();
            _queueEvent.Dispose();
        }
        _disposed = true;
    }
}

Here is the solution based on BlockingCollection -- I find it more elegant:

/// <summary>Perform asynchronous processing in a background thread.</summary>
public class BgLoop_BlockingCollection : IBgLoop, IDisposable
{
    /// <summary>Message handling delegate.</summary>
    private ProcessMessage _processMessageDelegate;

    /// <summary>Keep the messages in a BlockingCollection.</summary>
    private BlockingCollection<Action> _actionQueue = new BlockingCollection<Action>();

    /// <summary>Task for background processing.</summary>
    private Task _bgLoopTask;

    /// <summary>Flag to store if this class is disposing.</summary>
    private bool _disposed;

    public void Start(ProcessMessage processMessageDelegate)
    {
        if (_bgLoopTask != null) return;  // prevent starting more than once
        _processMessageDelegate = processMessageDelegate;
        _bgLoopTask = Task.Factory.StartNew(BgLoop, TaskCreationOptions.LongRunning);
    }

    public void AddMessage(Message message)
    {
        _actionQueue.Add(() =>
        {
            if (_processMessageDelegate != null)
            {
                _processMessageDelegate(message);
            }
        });
    }

    /// <summary>
    /// Infinite loop for processing the incoming events from the Queue.
    /// </summary>
    private void BgLoop()
    {
        foreach(var item in _actionQueue.GetConsumingEnumerable())
        {
            item(); // Run the task
        } // Exits when the BlockingCollection is marked for no more actions
    }

    /// <summary>
    /// Stops the sender parallel task, but send all data before exiting.
    /// </summary>
    private void StopSenderTask()
    {
        _actionQueue.CompleteAdding();
        _bgLoopTask.Wait();
        Console.WriteLine("Thread exited");
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    protected virtual void Dispose(bool disposing)
    {
        if (_disposed)
        {
            return;
        }
        if (disposing)
        {
            StopSenderTask();
        }
        _disposed = true;
    }
}

Here is the simple solution doing no background loop at all:

public class BgLoop_Simple : IBgLoop, IDisposable
{
    /// <summary>Message handling delegate.</summary>
    private ProcessMessage _processMessageDelegate;

    public void Start(ProcessMessage processMessageDelegate)
    {
        if (_processMessageDelegate != null) return;  // prevent starting more than once
        _processMessageDelegate = processMessageDelegate;
        // nothing else to do
    }

    public void AddMessage(Message message)
    {
        // call directly in a threadpool thread
        Task.Run(() => _processMessageDelegate?.Invoke(message));
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    protected virtual void Dispose(bool disposing) {}
}

And finally here is a simple Main and surroundings, for testing:

public delegate void ProcessMessage(Message msg);

public interface IBgLoop : IDisposable
{
    void Start(ProcessMessage processMessageDelegate);
    void AddMessage(Message message);
}

public class Message
{
    public string _msg;
    public override string ToString() { return _msg; }
}

public class Program
{
    public static void ProcessMessage(Message msg)
    {
        try
        {
            Console.WriteLine($"{DateTime.Now}: Processing message {msg} ...");
            // ... do processing ...
        }
        catch (Exception)
        {
            // ...
        }
    }

    public static void Main(string[] args)
    {
        //using (IBgLoop loop = new BgLoop_Queue())
        using (IBgLoop loop = new BgLoop_ConcQueue())
        //using (IBgLoop loop = new BgLoop_BlockingCollection())
        //using (IBgLoop loop = new BgLoop_Simple())
        {
            loop.Start(new ProcessMessage(Program.ProcessMessage));

            int cnt = 0;
            // add some messages slowly
            for (int i = 0; i < 5; ++i)
            {
                Thread.Sleep(500);
                loop.AddMessage(new Message{ _msg = $"msg{++cnt}" });
            }

            // add some messages quickly
            for (int i = 0; i < 10; ++i)
            {
                loop.AddMessage(new Message{ _msg = $"msg{++cnt}" });
            }

            Thread.Sleep(2000);
        }
    }
}

Answer 5

It is unusual to call Init before the object can be used. The constructor should create a fully functional and working object.

Therefore, I would drop the method Init and move the initialization logic to the constructor.

Additional, a method Start can be used to start the background thread.

---

You should use a long-running task for the background task. Otherwise the thread will be taken from the ThreadPool (which is used for short running tasks usually)

---

General solution for "async task processing"

I use the following simple solution for that kind of tasks:

• Using a ConcurrentQueue as queue to avoid locking
• Using a timer (concrete timer depends on the use case) to check periodically if the queue has items to process.
• Stop the timer if items are available
• Restart the timer if all items from the queue were processed

That solutions works perfectly without locking and thread signaling :)

Answer 6

I would propose slightly modified solution.

• Use ConcurrentQueue<T>. It will allow you to avoid locking your queue during reading\writing. If it is not possible due to some reasons, try use ReaderWriterLockSlim instead of Monitor (lock).
• Implementation of the method Init is not thread safe and allows you to re-initialize object after it is disposed, that violates disposing pattern principles. If you want to re-init your object, I would suggest to create Stop method and rename Init to Start.

• Indefinite Wait is generally not a good idea in dispose flow. I would use some timeout and then try to force the thread to finish.

  using System;
  using System.Collections.Concurrent;
  using System.Threading.Tasks;
  using System.Threading;
  
  namespace BgLoopExample
  {
      /// <summary>
      /// Perform asynchronous processing in a background thread.
      /// Handle thread creation, exit, thread safety.
      /// </summary>
      public class BgLoopExample : IDisposable
      {
  
          private CancellationTokenSource cancellationToken;
  
          private readonly object _syncObject = new object();
          /// <summary>
          /// Queue for the incoming messages.
          /// </summary>
          private ConcurrentQueue<Message> myQueue;
  
          /// <summary>
          /// Task for background processing.
          /// </summary>
          private Task myBgLoopTask;
  
          /// <summary>
          /// Event for awakening the processing loop, used when new entry is added to the queue or exit requested.
          /// </summary>
          private AutoResetEvent myQueueEvent;
  
          /// <summary>
          /// Flag to signal stop for the parallel sender thread.
          /// </summary>
          private volatile bool myStopTaskFlag = true;
  
          /// <summary>
          /// Flag to store if this class is disposing.
          /// </summary>
          private bool myDisposed;
  
          /// <summary>
          /// Class initializer, start the background thread.
          /// </summary>
          public void Init()
          {
              if (myDisposed)
              {
                  throw new ObjectDisposedException("","Object disposed");
              }
              //Avoiding concurrent initialization
              if (myBgLoopTask != null) return; //Or throw exception
              lock (_syncObject)
              {
                  if (myBgLoopTask != null) return; //Or throw exception
                  myQueue = new ConcurrentQueue<Message>();
                  myQueueEvent = new AutoResetEvent(false);
                  myStopTaskFlag = false;
                  myDisposed = false;
                  StartSenderTask();
              }
          }
  
          /// <summary>
          /// Adds new event to the Queue
          /// </summary>
          /// <param name="message">New event to add to the queue.</param>
          public void AddMessage(Message message)
          {
              if (myDisposed)
              {
                  throw new ObjectDisposedException("", "Object disposed");
              }
              myQueue?.Enqueue(message);
              myQueueEvent?.Set();
          }
  
          /// <summary>
          /// Starts the sender parallel task.
          /// </summary>
          private void StartSenderTask()
          {
              myStopTaskFlag = false;
              cancellationToken = new CancellationTokenSource();
              //NOTE: LongRunning task does not occupy a thread from ThreadPool.
              myBgLoopTask = new Task(BgLoop, cancellationToken.Token, TaskCreationOptions.LongRunning);
              myBgLoopTask.Start();
          }
  
          /// <summary>
          /// Stops the sender parallel task, but send all data before exiting.
          /// </summary>
          private void StopSenderTask()
          {
              // set the stop flag, and signal change
              myStopTaskFlag = true;
              myQueueEvent.Set();
              // wait till the loop exits
              var stopped = myBgLoopTask.Wait(TimeSpan.FromSeconds(5)); //To avoid hanging
              if (!stopped)
              {
                  //Log timeout
                  //Force to stop by throwing cancellation exception
                  cancellationToken.Cancel(true);
                  myBgLoopTask.Wait();
              }
              myBgLoopTask.Dispose();
              cancellationToken.Dispose();
              myBgLoopTask = null;
              Console.WriteLine("Thread exited");
          }
  
          /// <summary>
          /// Infinite loop for processing the incoming events from the Queue.
          /// </summary>
          private void BgLoop()
          {
              while (true)
              {
                  Message message;
                  while (!myQueue.TryDequeue(out message))
                  {
                      ProcessMessage(message);
                  }
  
                  if (myStopTaskFlag)
                  {
                      if (myQueue.IsEmpty)
                      {
                          break;
                      }
                      continue;
                  }
                  // Block for a while.  The exact timeout value does not really matter
                  myQueueEvent.WaitOne(10000);
              }
              Console.WriteLine("Thread exiting...");
          }
  
          /// <summary>
          /// Send the next group of events using the data sender.
          /// </summary>
          /// <param name="message">Event to send.</param>
          private void ProcessMessage(Message message)
          {
              try
              {
                  Console.WriteLine($"{DateTime.Now}: Processing message {message} ...");
                  // ... do processing ...
              }
              catch (Exception ex)
              {
                  // ...
              }
          }
  
          public void Dispose()
          {
              Dispose(true);
              GC.SuppressFinalize(this);
          }
  
          protected virtual void Dispose(bool disposing)
          {
              if (myDisposed)
              {
                  return;
              }
              if (disposing)
              {
                  StopSenderTask(true);
                  myQueueEvent.Dispose();
              }
              myDisposed = true;
          }
      }
  }