I feel like I'm re-inventing the wheel. Dispatch work to a specific thread. I want to process events from multiple threads sequentially on one thread

Question

I know that similar things exist in WPF and forms applications with the Control.Invoke method, I also know of the existence of BackgroundWorker, ThreadPool etc.

However, I don't want to depend on Forms/WPF, and I want to make sure work is executed sequentially and on one thread.

Edit: Rationale: I want to drive a state machine from one thread. The events come from other threads tough. There is no UI.

So far I couldn't really figure out how to do this with existing framework classes but I might have misunderstood the documentation.

Edit: I forgot to mention I'm bound to .NET Framework 3.5

What I wrote so far:

public class Dispatcher
{
    string Name;
    Thread WorkerThread;
    Queue<Action> WorkQueue;
    List<Exception> Exceptions;
    ManualResetEvent Gate;
    volatile bool KeepRunning;
    readonly object WorkLocker;

    public override string ToString()
    {
        return String.Format("{0}({1})", this.GetType().Name, Name);
    }

    public Dispatcher(string name)
    {
        Name = name;
        WorkLocker = new Object();
        Gate = new ManualResetEvent(false);
        WorkQueue = new Queue<Action>();
        Exceptions = new List<Exception>();
    }

    public void Start()
    {
        if (WorkerThread == null)
        {
            WorkerThread = new Thread(doDispatch)
            {
                IsBackground = true,
                Name = this.Name
            };
            WorkerThread.Start();
        }
    }

    public void Stop()
    {
        if (WorkerThread != null && WorkerThread.IsAlive)
        {
            Dispatch(() => { KeepRunning = false; });
            WorkerThread.Join();
        }
        WorkerThread = null;
    }

    public void Reset()
    {
        Stop();
        lock (WorkLocker)
        {
            WorkQueue = new Queue<Action>();
            Exceptions = new List<Exception>();
        }
    }

    public void Dispatch(Action a)
    {
        lock (WorkLocker)
        {
            WorkQueue.Enqueue(a);
        }
        Gate.Set();
    }

    public List<Exception> CollectExceptions()
    {
        List<Exception> result = new List<Exception>();
        lock(WorkLocker)
        {
            foreach(Exception e in Exceptions)
            {
                result.Add(e);
            }
            Exceptions.Clear();
        }
        return result;
    }

    private void doDispatch()
    {
        KeepRunning = true;
        while (KeepRunning)
        {
            Gate.WaitOne();
            lock (WorkLocker)
            {
                while (WorkQueue.Count > 0)
                {
                    try
                    {
                        WorkQueue.Dequeue()?.Invoke();
                    }
                    catch (Exception e)
                    {
                        Exceptions.Add(e);
                    }
                }
            }
        }
    }
}

Is there a way to do something like this in a simpler way? Another nice feature would be being able to dispatch calls that have multiple arguments.

You could use a [ConcurrentQueue](https://learn.microsoft.com/en-us/dotnet/api/system.collections.concurrent.concurrentqueue-1?view=netframework-4.7.2) and ditch the locking. Have you considered that? — Fildor, Nov 05 '18 at 07:55
Good point. I completely forget to mention that I'm currently bound to Framework 3.5. As far as I know the concurrent queue is not available in 3.5. — katzenversteher, Nov 05 '18 at 07:57
That would have been a very useful piece of information to put into the question ;) — Fildor, Nov 05 '18 at 07:58
@katzenversteher nothing is available in 3.5, not even 3.5 - it's not supported anymore. Unless you target the (unsupported as well) Windows XP, there's no reason to use 3.5 and *every* reason to avoid it. No TLS 1.2, no async/await. — Panagiotis Kanavos, Nov 05 '18 at 07:58
So, I guess the 3.5 - thing is not in your hands - you cannot upgrade? Because that would be the first thing I'd recommend. — Fildor, Nov 05 '18 at 08:01
@katzenversteher btw what you ask is already implemented through the Dataflow classes, esp ActionBlock. Queeing and processing in a single thread [is the simplest case](https://learn.microsoft.com/en-us/dotnet/standard/parallel-programming/how-to-perform-action-when-a-dataflow-block-receives-data). — Panagiotis Kanavos, Nov 05 '18 at 08:02
@katzenversteher You can create a pipeline of processing blocks and even have the final block run on the UI thread in order to update the UI, as shown in [Walkthrough: Using Dataflow in a Windows Forms Application](https://learn.microsoft.com/en-us/dotnet/standard/parallel-programming/walkthrough-using-dataflow-in-a-windows-forms-application) — Panagiotis Kanavos, Nov 05 '18 at 08:02
I would guess that the op has some sort of embedded device that is not upgradable, rather than being stuck on XP on a PC. Still, it's worth pointing out that the software will take longer (and be more expensive) that possibly buying new hardware. — Neil, Nov 05 '18 at 08:03
I would use a BlockingCollection which is an implementation of the Producer-Consumer pattern. It's basically doing all the work of your collection, collection lock, collection notify event (Gate) and KeepRunning in one. — ckuri, Nov 05 '18 at 08:09
@ckuri BlockingCollection is available from 4.0 , OP is locked in on 3.5 — Fildor, Nov 05 '18 at 08:23
I'll read a bit through the Dataflow things. I have no UI though. There is a state machine I want to drive with only one thread. Events however are generated by other threads. — katzenversteher, Nov 05 '18 at 08:26
Your Start() Stop() and Reset() are not multithreading safe ! "WorkQueue.Dequeue()?.Invoke()" will not compile under 3.5. in doDispatch() you should not lock (WorkLocker) while executing the Invoke. This defeats the purpose of your class, because nobody can Dispatch anything while doDispatch() holds the lock. — Peter Huber, Nov 05 '18 at 18:25
@PeterHuber "WorkQueue.Dequeue()?.Invoke()" does compile on my system. AFAIK it's a language (compiler) feature, not a framework feature. The intention of the lock in doDispatch was to avoid reading from the queue while someone else is writing into it. However, I can probably release the lock immediately after dequeueing and before invoking. Thank you! — katzenversteher, Nov 06 '18 at 06:39
The purpose of the lock is to protect access to the WorkQueue. In general, a lock should be applied for a short time as possible. Please note that you need to make also Start() Stop() and Reset() multithreading safe. You can do that by using locking or using the Interlocked Class. — Peter Huber, Nov 06 '18 at 13:39
Multithreading code is extremely difficult to write. Do you know how to test it ? Use some threads writing as quickly as possible to the queue, each thread writing an identifier and a running number. The comsuming thread will then check if every message it dequeues is in the proper sequence. For more see: https://stackoverflow.com/questions/499634/how-to-detect-and-debug-multi-threading-problems/24549418#24549418 — Peter Huber, Nov 06 '18 at 13:43

score 1 · Accepted Answer · answered Nov 05 '18 at 11:53

Since you are bound to 3.5 you can't use BlockingCollection or the DataFlow library...you'll have to roll your own implementation.

The sample code you provided is a good start, but you should apply the Single Responsibility Principle to make it cleaner and easier to refactor when(if?) you upgrade the .NET Framework.

I would do it like this:

Create a thread safe wrapper class around Queue that somewhat mimics BlockingCollection, this answer provides a nice example
Structure your code around a consumer/producer flow and inject the wrapper

I feel like I'm re-inventing the wheel. Dispatch work to a specific thread. I want to process events from multiple threads sequentially on one thread

1 Answers1