9
\$\begingroup\$

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;
        }
    }
}
\$\endgroup\$
6
  • 3
    \$\begingroup\$ Instead of Queue use ConcurrentQueue and you can avoid your locks. See: msdn.microsoft.com/en-us/library/dd267265(v=vs.110).aspx Also this probably belongs better on codereview. \$\endgroup\$
    – Bauss
    Jan 3, 2017 at 14:43
  • \$\begingroup\$ ConcurrentQueue seems like a good way to make the code more compact, however, the lock here does not only ensures the thread-safety of the queue object, but also synchronization with the event triggering. I'm not sure that is crucial, though. \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 14:51
  • \$\begingroup\$ I was not aware of codereview site. Thanks for migrating. \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 15:03
  • \$\begingroup\$ Update: I received several improvement suggestions that are relatively minor -- I have incorporated several of them 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 have added a few sample usage scenarios to the description \$\endgroup\$
    – Adam B
    Jan 4, 2017 at 14:00
  • 1
    \$\begingroup\$ I have rolled back the last edit. Please see What to do when someone answers. \$\endgroup\$ Jan 4, 2017 at 14:04

6 Answers 6

1
\$\begingroup\$

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.

\$\endgroup\$
2
  • \$\begingroup\$ I like using BlockingCollecion, it seems to achieve the same thing in a more compact code. I rewrote my code (but I cant' post it rright here in a comment...). \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 15:27
  • \$\begingroup\$ Glad it was useful to you. As you probably read in my answer, even I didn't get to it on the first shot! But unless you have INSANE performance requirements (high, frequent concurrence) then it "just works" very well for the typical cases. \$\endgroup\$ Jan 3, 2017 at 16:00
7
\$\begingroup\$

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.

\$\endgroup\$
5
  • \$\begingroup\$ The point about making the queue general w.r.t the message type is accepted \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 15:36
  • \$\begingroup\$ You are right about using up a Threadpool thread; in my solution the background thread continuously blocks a thread. A solution for this would be to use TPL async pattern. However, as I remember, async waiting on an event is not yet implemented :( \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 15:39
  • \$\begingroup\$ I will need to digest your solution, it looks promising :) \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 15:39
  • \$\begingroup\$ @AdamRotaru Your question doesn't have any code that uses any event at all, but even if you did want to do something after an event fires, you can trivially create a Task that represents the next time an event fires, if that's what you need. It's only a few lines of code. If you just mean the AutoResetEvent, then there's no need for an ARE in this solution, it will just automatically perform the next operation in the queue as soon as its able to. \$\endgroup\$
    – Servy
    Jan 3, 2017 at 15:53
  • \$\begingroup\$ I referred to my original solution -- I used an ARE to make sure to wake up the processing thread as soon as a message is available (without it if the polling loop makes too frequent checks, we waste CPU, if not too frequent, there is a long delay till an message is noticed). In your solution there is no need for an ARE, nor for a long-running task, I agree. \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 16:05
4
\$\begingroup\$

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;
        }
    }
} 
\$\endgroup\$
4
  • \$\begingroup\$ Bruno Costa: about your first comment on the principle: I am considering it... \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 16:14
  • \$\begingroup\$ Bruno Costa: thanks for the minor improvement suggestions. \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 16:15
  • \$\begingroup\$ Not having a class at all and calling Task.Run would result in operations being performed in parallel if multiple users added work at the same time, rather than the operations being performed sequentially. (Unless you have a class to synchronize them and prevent that from happening, of course.) \$\endgroup\$
    – Servy
    Jan 3, 2017 at 16:19
  • \$\begingroup\$ @Servy You're right. I didn't find a scenario where that might be a issue so far though. Should the Op have one he should prefer your solution \$\endgroup\$ Jan 3, 2017 at 16:27
0
\$\begingroup\$

/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);
        }
    }
}
\$\endgroup\$
-1
\$\begingroup\$

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 :)

\$\endgroup\$
5
  • \$\begingroup\$ I have some problems with using timers: it's hard to 'wake' them up, to process immediately; and also I have many cases when timers where used improperly and led to cascading endless loops; I found timers quite error-prone and hard to debug and read, and my advise is to avoid them. \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 15:31
  • \$\begingroup\$ To the point about using a long-running task for the background task: you are right. A solution for this would be to use TPL async pattern. However, as I remember, async waiting on an event is not yet implemented :( \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 15:43
  • \$\begingroup\$ About Init(): I think this is neutral, and not relevant to the main issue. I probably did this way from the effect of dependency injection patterns, where constructors are used indirectly, and an Init() is a convention for further initialization. But I agree that the member variables could and should be created by the constructor, and Init() just starts the processing loop. If it is renamed to Start(), we arrive at your suggested solution :) \$\endgroup\$
    – Adam B
    Jan 3, 2017 at 15:47
  • 1
    \$\begingroup\$ You can just start a long running task like that:Task.Factory.StartNew(BgLoop, TaskCreationOptions.LongRunning). That call creates a new thread for the task instead of using one of the ThreadPool. \$\endgroup\$
    – JanDotNet
    Jan 3, 2017 at 15:47
  • 1
    \$\begingroup\$ There are multiple problems with your current Init implementation: If you call it multiple times, the instance variables are reinitialized (the overwritten queue object may contain items). If you don't call it, the method AddMessage doesn't work (without error). \$\endgroup\$
    – JanDotNet
    Jan 3, 2017 at 15:50
-1
\$\begingroup\$

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;
            }
        }
    }
    
\$\endgroup\$
6
  • 1
    \$\begingroup\$ You've removed the thread safety from the OP's code. The lock isn't just there to prevent mutations from the queue from multiple threads, it's also to ensure that the operations on the ARE and other state of the type is logically atomic. If you're going to remove the lock you need to use a tool that allows you to avoid keeping track of any other state so that you have nothing to synchronize. \$\endgroup\$
    – Servy
    Jan 3, 2017 at 18:34
  • \$\begingroup\$ @Servy I do not understand your comment. Could you please detail your concerns? In the author's case the lock is used only to synchronize access to queue. \$\endgroup\$ Jan 3, 2017 at 18:42
  • 1
    \$\begingroup\$ But it's not. it's also used to ensure that the producer and consumer are properly notified of what should be processed, not just what items are in the queue. Without proper synchronization that state can be incorrect, stating that there's work to do when there isn't, for example, or not being notified that the queue has stopped. \$\endgroup\$
    – Servy
    Jan 3, 2017 at 18:52
  • \$\begingroup\$ @Servy, if you mean AutoResetEvent synchronization, then there is no need to sync, because it only tells thread to wake up if there is an element appeared in the queue. Stopping is regulated by flag that is also no need to synchronize, because it is changed only once and in the worst case its state will be re-read almost immediately. We can discuss it in the chat if you like. \$\endgroup\$ Jan 3, 2017 at 19:32
  • 1
    \$\begingroup\$ But the logic is specifically designed such that it only actually notifies the consumer that there's an item when there is an item. When you remove the synchronization you can notify the consumer of an item when there is no item. For the flag the worst case isn't that it'll be read immediately. If there's no synchronization in place the flag value could be cached as the runtime could prove that it's never being mutated from that thread, and preventing the consumers from ever being updated. \$\endgroup\$
    – Servy
    Jan 3, 2017 at 19:42

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.