Custom cross-platform thread dispatcher for console / asp.net application

Question

I’ve got a .NET application that needs to call a COM object (it always has to be called from the same thread). As I have multiple threads in the application, I need to invoke an action on another thread.
The application does not have a (standard) message loop and I don’t really like the idea to add WPF / WinForms just to have a Dispatcher.
What would be a safe and effective way to implement a custom "message loop" / queue that allows invoking an Action / Func (with return type) on another thread?
It would also be nice to have a cross-platform solution for this problem.

Are you talking about a standard [message loop](https://en.wikipedia.org/wiki/Message_loop_in_Microsoft_Windows), typically installed on the UI thread by windows-based applications? — Theodor Zoulias, Feb 18 '22 at 11:50
I don’t know too much about this topic, but I think so. I need some kind of queue to store the actions in that need to be called (by the message loop) on the other thread. Problem is that the loop may not block code execution of the 2nd thread! — ShortDevelopment, Feb 18 '22 at 12:31
I don't think that you want a standard message loop, that processes messages coming from the operating system. You probably need a `BlockingCollection`-based solution, like the one in [this](https://stackoverflow.com/questions/58379898/c-sharp-moving-database-to-a-separate-thread-without-busy-wait/58397942#58397942) answer, that only invokes actions coming from your code. — Theodor Zoulias, Feb 18 '22 at 12:58
So I could use something like this? ``` Console.WriteLine($"Main Thread: {Thread.CurrentThread.ManagedThreadId}"); BlockingCollection _queue = new(); Thread thread = new(() => { DoSomeWork(); Console.WriteLine($"Worker Thread: {Thread.CurrentThread.ManagedThreadId}"); foreach (var task in _queue.GetConsumingEnumerable()) task?.Invoke(); }); thread.Start(); _queue.Add(() => { Console.WriteLine($"Invoked on Thread: {Thread.CurrentThread.ManagedThreadId}"); }); ``` — ShortDevelopment, Feb 18 '22 at 15:21
ShortDevelopment if this solves your problem, you could post it as a [self-answer](https://stackoverflow.com/help/self-answer). Honestly your reference to COM objects, message loop and `Dispatcher` made your problem look initially much more complex than just consuming a `BlockingCollection` in a thread. — Theodor Zoulias, Feb 18 '22 at 15:40
Sadly, that doesn’t seem to fix all my problems. My custom “message loop” now blocks the thread. A WPF dispatcher would not do that but as far as I know it also tightly integrates with the os. Could I modify the above code to still not block the thread while being able to execute all new action / tasks in the queue? Sorry, if that sounds weird; it might just be the fact that I’m missing quite some knowledge in this area. That’s why I’m asking… — ShortDevelopment, Feb 18 '22 at 16:13
I am not very familiar with the WPF, but in WinForms the method that starts the message loop, the [`Application.Run`](https://learn.microsoft.com/en-us/dotnet/api/system.windows.forms.application.run) method, also blocks the UI thread. Any code that follows this method will be blocked until the loop completes. You can't have a thread that serves a message loop, and can also do other things in parallel. A message loop owns the thread on which it runs. — Theodor Zoulias, Feb 18 '22 at 16:19
Yes, your right… I’ve now created a small class (based on your information) that should solve my problems very well. Fell free to add / edit if something seems odd. Thanks for your help! — ShortDevelopment, Feb 18 '22 at 20:22

ShortDevelopment · Accepted Answer · 2022-02-20T08:26:07.443

Based on the information of @theodor-zoulias, I came up with this solution.
Disclaimer: Might be that this is actually a very bad design!

public sealed class DispatcherLoop : IDisposable
{
    #region Instance
    private DispatcherLoop() { }

    static Dictionary<int, DispatcherLoop> dispatcherLoops = new();
    public static DispatcherLoop Current
    {
        get
        {
            int threadId = Thread.CurrentThread.ManagedThreadId;
            if (dispatcherLoops.ContainsKey(threadId))
                return dispatcherLoops[threadId];

            DispatcherLoop dispatcherLoop = new()
            {
                ThreadId = Thread.CurrentThread.ManagedThreadId
            };
            dispatcherLoops.Add(threadId, dispatcherLoop);
            return dispatcherLoop;
        }
    }
    #endregion

    bool isDisposed = false;
    public void Dispose()
    {
        if (isDisposed)
            throw new ObjectDisposedException(null);

        _queue.CompleteAdding();
        _queue.Dispose();
        dispatcherLoops.Remove(ThreadId);
        isDisposed = true;
    }

    public int ThreadId { get; private set; } = -1;
    public bool IsRunning { get; private set; } = false;

    BlockingCollection<Task> _queue = new();
    public void Run()
    {
        if (isDisposed)
            throw new ObjectDisposedException(null);

        if (ThreadId != Thread.CurrentThread.ManagedThreadId)
            throw new InvalidOperationException($"The {nameof(DispatcherLoop)} has been created for a different thread!");

        if (IsRunning)
            throw new InvalidOperationException("Already running!");

        IsRunning = true;

        try
        {
            // ToDo: `RunSynchronously` is not guaranteed to be executed on this thread (see comments below)!
            foreach (var task in _queue.GetConsumingEnumerable())
                task?.RunSynchronously();
        }
        catch (ObjectDisposedException) { }

        IsRunning = false;
    }

    public void BeginInvoke(Task task)
    {
        if (isDisposed)
            throw new ObjectDisposedException(null);

        if (!IsRunning)
            throw new InvalidOperationException("Not running!");

        if (ThreadId == Thread.CurrentThread.ManagedThreadId)
            task?.RunSynchronously();
        else
            _queue.Add(task);
    }

    public void Invoke(Action action)
    {
        if (isDisposed)
            throw new ObjectDisposedException(null);

        Task task = new(action);
        BeginInvoke(task);
        task.GetAwaiter().GetResult();
    }

    public T Invoke<T>(Func<T> action)
    {
        if (isDisposed)
            throw new ObjectDisposedException(null);

        Task<T> task = new(action);
        BeginInvoke(task);
        return task.GetAwaiter().GetResult();
    }
}

Nice! Just be aware that the `Task.RunSynchronously` is not actually guaranteed to execute the task synchronously. In case you do acrobatics like calling `Invoke` recursively (call `Invoke` inside the action of `Invoke`), then after ~1,400 recursions the next `RunSynchronously` will run on the `ThreadPool`. You can read [here](https://devblogs.microsoft.com/pfxteam/when-executesynchronously-doesnt-execute-synchronously/) why this happens. — Theodor Zoulias, Feb 18 '22 at 22:12
Thanks, I didn’t know about this behavior! Could I overcome this issue by using a custom `TaskScheduler` like your linked answer above did? — ShortDevelopment, Feb 19 '22 at 09:07
ShortDevelopment according to my experiments, no. The `StackOverflowException`-evading logic is baked in the task-execution mechanism, and cannot be circumvented. Your best bet is probably to ditch the delegate-based tasks approach, and use instead pairs of a `TaskCompletionSource` and an `Action`. Run the action, and then complete the `TaskCompletionSource` accordingly (successfully or faulted or canceled). If you want to get fancy, you could create a `Task`-like class that inherits from the `TaskCompletionSource` class, and add an `Action` field and a `RunSynchronously` method to it. — Theodor Zoulias, Feb 19 '22 at 09:15
Actually my previous comment is incorrect. If you create a `SingleThreadTaskScheduler` instance from [this answer](https://stackoverflow.com/a/58397942/11178549) and use it to schedule tasks, it is guaranteed that the action of the tasks will run on the dedicated thread, or not at all. If you stretch this class to its limits, by attempting to `RunSynchronously` tasks recursively, after ~1400 recursions the `TryExecuteTask` will deny to execute the task and will divert to the `QueueTask`, resulting in a deadlock. — Theodor Zoulias, Feb 19 '22 at 11:34
I just noticed that the `void Invoke(Action action)` method has different behavior than the `T Invoke(Func action)`. The later waits synchronously the action to complete, but the former doesn't. The former behaves like the `BeginInvoke` in various APIs (it launches a fire-and-forget operation). The later behaves like a true `Invoke`, with all that this entails (one `Invoke` inside another will deadlock). — Theodor Zoulias, Feb 19 '22 at 12:01
Wow, thanks for this incredible research! I’ve updated the code to reflect the `Invoke` vs `BeginInvoke` behavior. Now it's clear what can / will deadlock and what not. — ShortDevelopment, Feb 19 '22 at 21:13

score -1 · Answer 2 · answered Feb 20 '22 at 21:12

You should use Microsoft's Reactive Framework (aka Rx) - NuGet System.Reactive and add using System.Reactive.Linq; - then you can do this:

var els = new EventLoopScheduler();

Then you can do things like this:

els.Schedule(() => Console.WriteLine("Hello, World!"));

The EventLoopScheduler spins up its own thread and you can ask it to schedule any work on it you like - it'll always be that thread.

When you're finished with the scheduler, just call els.Dispose() to shut down it down cleanly.

There are also plenty of overloads for scheduling code in the future. It's a very powerful class.

Custom cross-platform thread dispatcher for console / asp.net application

2 Answers2