2
\$\begingroup\$

I implemented a multi-producer single consumer class, with an important restriction that once the consumer started working, it must continue its work on the same thread (the reasoning behind this is to allocate and free COM objects from a third party library, and these actions must be on the same thread according to that library).

Here's the code, followed by an example:

interface WorkerLogic<T>
{
    void Work(T item);
}

class ThreadedWorker<T> where T : class
{
    private readonly object locker = new object();
    private readonly Queue<T> queue = new Queue<T>();
    private readonly WorkerLogic<T> logic;
    private readonly Thread actualThread;
    private bool started = false;
    private readonly ManualResetEvent resetEvent = new ManualResetEvent(false);
    private volatile bool shouldWork = true;

    public ThreadedWorker(WorkerLogic<T> logic)
    {

        this.logic = logic;
        actualThread = new Thread(Spin);
    }

    private void Spin()
    {
        while (shouldWork)
        {
            resetEvent.WaitOne();
            while (shouldWork)
            {
                T item;
                lock (locker)
                {
                    if (queue.Count == 0)
                        break; // back to main loop and WaitOne
                    item = queue.Dequeue();
                    resetEvent.Reset();
                }
                try
                {
                    logic.Work(item);
                }
                catch (Exception ex)
                {
                    // log
                }
            }
        }
    }

    public void Stop()
    {
        Stop(TimeSpan.FromMilliseconds(500));
    }

    public void Stop(TimeSpan timeout)
    {
        shouldWork = false;
        lock (locker)
        {
            resetEvent.Set();
        }
        bool joined = actualThread.Join(timeout);
        if (!joined)
        {
            try
            {
                actualThread.Abort();
            }
            catch (ThreadStateException)
            {
                // swallow
            }
        }
    }


    public void Push(T item)
    {
        lock (locker)
        {
            queue.Enqueue(item);
            if (!started)
            {
                started = true;
                actualThread.Start();
            }
            resetEvent.Set();
        }
    }
}

Usage example for the code above:

public class UsageExample
{
    public class MyLogic : WorkerLogic<string>
    {
        public void Work(string s)
        {
            Console.WriteLine("working on '{0}', thread id = {1}", s, Thread.CurrentThread.ManagedThreadId);
        }
    }

    public static void Main()
    {
        ThreadedWorker<string> threadedWorker = new ThreadedWorker<string>(new MyLogic());
        threadedWorker.Push("My");
        threadedWorker.Push("name");
        threadedWorker.Push("is");
        threadedWorker.Push("Luca");
        threadedWorker.Stop();
    }

}

Example output:

working on 'My', thread id = 3
working on 'name', thread id = 3
working on 'is', thread id = 3
working on 'Luca', thread id = 3

I created a repl for this, feel free to play with it.

I'd like to have a peer review focused on correctness: is it thread safe? Am I missing an edge case?

Thanks!

\$\endgroup\$
  • 1
    \$\begingroup\$ Not sure if this is the kind of feedback you're after, but it looks like you could simplify your ThreadedWorker<T> considerably by using a BlockingCollection<T>. \$\endgroup\$ – rob.earwaker Aug 29 at 9:17
  • 1
    \$\begingroup\$ In case of .NET Core System.Threading.Channels may also help. \$\endgroup\$ – aepot Aug 29 at 12:56
  • \$\begingroup\$ Please do not update the original code in your question. The comments and the code might be out of sync. Instead post as an answer or if you wish to receive further review suggestions then please open a new question. \$\endgroup\$ – Peter Csala Sep 3 at 14:37
  • 1
    \$\begingroup\$ @PeterCsala I made changes according to your suggestions. \$\endgroup\$ – Ron Klein Sep 3 at 15:18
0
\$\begingroup\$

I took rob.earwaker's advice (thanks!), and I implemented it using a BlockingCollection. The code is much simpler now :)

class ThreadedWorker<T> where T : class
{
    private readonly object locker = new object();
    BlockingCollection<T> queue = new BlockingCollection<T>(new ConcurrentQueue<T>());
    private readonly WorkerLogic<T> logic;
    private readonly Thread actualThread;
    private bool started = false;
    private volatile bool shouldWork = true;

    public ThreadedWorker(WorkerLogic<T> logic)
    {

        this.logic = logic;
        actualThread = new Thread(Spin);
        
    }

    private void Spin()
    {
        while (shouldWork)
        {
            T item;
            var dequeued = queue.TryTake(out item, TimeSpan.FromMilliseconds(100));
            if (!dequeued) continue;
            try
            {
                logic.Work(item);
            }
            catch (Exception ex)
            {
                // log
            }

        }            
    }

    public void Stop()
    {
        Stop(TimeSpan.FromMilliseconds(500));
    }

    public void Stop(TimeSpan timeout)
    {
        shouldWork = false;            
        bool joined = actualThread.Join(timeout);
        if (!joined)
        {
            try
            {
                actualThread.Abort();
            }
            catch (ThreadStateException)
            {
                // swallow
            }
        }
    }


    public void Push(T item)
    {
        if (!started)
        {
            lock (locker)
            {
                if (!started)
                {
                    started = true;
                    actualThread.Start();
                }
            }
        }
        queue.Add(item);
    }
}
| improve this answer | |
\$\endgroup\$

Your Answer

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

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