How to send messages between threads using CommunityToolkit.Mvvm.Messaging and WinUI 3?

Question

The following simple multi-threaded program was meant to try out the CommunityToolkit Messenger package for which the documentation says (see: Messenger)

Both WeakReferenceMessenger and StrongReferenceMessenger also expose a Default property that offers a thread-safe implementation built-in into the package.

I had hoped this would mean I could send messages on one thread and receive them on other threads but a problem arose with what seems to be the IMessenger Interface. Details follow below.

This project starts with a vanilla TemplateStudio WinUI 3 (v1.1.5) desktop template that uses the CommunityToolkit Mvvm package (with Messenger) and a single page, MainPage. When the App launches, it starts a RandomMessageGenerator thread that periodically issues a TraceMessage using the WeakReferenceMessenger.Default channel from the Toolkit. The UI thread receives these messages and stores them in a List.

App.xaml.cs

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using CommunityToolkit.Mvvm.Messaging;
using Microsoft.UI.Xaml;

using Multi_Window.Activation;
using Multi_Window.Contracts.Services;
using Multi_Window.Core.Contracts.Services;
using Multi_Window.Core.Services;
using Multi_Window.Services;
using Multi_Window.ViewModels;
using Multi_Window.Views;
using Microsoft.UI;
using Microsoft.UI.Windowing;
using CommunityToolkit.Mvvm.Messaging.Messages;
using System.Diagnostics;

namespace Multi_Window;

// To learn more about WinUI 3, see https://docs.microsoft.com/windows/apps/winui/winui3/.
public partial class App : Application
{
    // The .NET Generic Host provides dependency injection, configuration, logging, and other services.
    // https://docs.microsoft.com/dotnet/core/extensions/generic-host
    // https://docs.microsoft.com/dotnet/core/extensions/dependency-injection
    // https://docs.microsoft.com/dotnet/core/extensions/configuration
    // https://docs.microsoft.com/dotnet/core/extensions/logging
    public IHost Host { get; }

    public static T GetService<T>()
        where T : class
    {
        if ((App.Current as App)!.Host.Services.GetService(typeof(T)) is not T service)
        {
            throw new ArgumentException($"{typeof(T)} needs to be registered in ConfigureServices within App.xaml.cs.");
        }

        return service;
    }

    public static WindowEx MainWindow { get; } = new MainWindow();
    public static ShellPage? ShellPage  { get; set; }
    private static readonly List<string> _traceMessages = new();

    private Task? messageGenerator;

    public App()
    {
        InitializeComponent();

        Host = Microsoft.Extensions.Hosting.Host.
        CreateDefaultBuilder().
        UseContentRoot(AppContext.BaseDirectory).
        ConfigureServices((context, services) =>
        {
            // Default Activation Handler
            services.AddTransient<ActivationHandler<LaunchActivatedEventArgs>, DefaultActivationHandler>();

            // Other Activation Handlers

            // Services
            services.AddTransient<INavigationViewService, NavigationViewService>();

            services.AddSingleton<IActivationService, ActivationService>();
            services.AddSingleton<IPageService, PageService>();
            services.AddSingleton<INavigationService, NavigationService>();

            // Core Services
            services.AddSingleton<IFileService, FileService>();

            // Views and ViewModels
            services.AddTransient<MainViewModel>();
            services.AddTransient<MainPage>();

            // ** NOTE ** changed to Singleton so we can refer to THE ShellPage/ShellViewModel
            services.AddSingleton<ShellPage>();
            services.AddSingleton<ShellViewModel>();

            // Configuration
        }).
        Build();

        UnhandledException += App_UnhandledException;
        System.AppDomain.CurrentDomain.UnhandledException += CurrentDomain_UnhandledException;
        Microsoft.UI.Xaml.Application.Current.UnhandledException += Current_UnhandledException;
    }

    private void RandomMessageGenerator()
    {
        var shutdown = false;
        WeakReferenceMessenger.Default.Register<ShutDownMessage>(this, (r, m) => shutdown = true);
        Debug.WriteLine($"RandomMessageGenerator started on thread {Environment.CurrentManagedThreadId}");

        Random rnd = new();
        // not a good way to control thread shutdown in general but will do for a quick test
        while (shutdown == false)
        {
            Thread.Sleep(rnd.Next(5000));
            var tm = new TraceMessage($"{DateTime.Now:hh:mm:ss.ffff} Timer event. (Th: {Environment.CurrentManagedThreadId})");
            try
            {
                WeakReferenceMessenger.Default.Send(tm);
            }
            catch (Exception e) 
            {
                Debug.WriteLine(e.Message);
                break;
            }
        }
        Debug.WriteLine($"RandomMessageGenerator closed at {DateTime.Now:hh:mm:ss.ffff} (Th: {Environment.CurrentManagedThreadId})");
    }

    private void Current_UnhandledException(object sender, Microsoft.UI.Xaml.UnhandledExceptionEventArgs e) => throw new NotImplementedException();
    private void CurrentDomain_UnhandledException(object sender, System.UnhandledExceptionEventArgs e) => throw new NotImplementedException();

    private void App_UnhandledException(object sender, Microsoft.UI.Xaml.UnhandledExceptionEventArgs e)
    {
        // TODO: Log and handle exceptions as appropriate.
        // https://docs.microsoft.com/windows/windows-app-sdk/api/winrt/microsoft.ui.xaml.application.unhandledexception.
        throw new NotImplementedException();
    }

    protected async override void OnLaunched(LaunchActivatedEventArgs args)
    {
        base.OnLaunched(args);

        await App.GetService<IActivationService>().ActivateAsync(args);

        MainWindow.AppWindow.Closing += OnAppWindowClosing;

        WeakReferenceMessenger.Default.Register<TraceMessage>(this, (r, m) =>
        {
            _traceMessages.Add(m.Value);
            Debug.WriteLine(m.Value);
        });
        WeakReferenceMessenger.Default.Register<WindowClosedMessage>(this, (r, m) => OnStatusWindowClosed());           // StatusWindow closed events
        WeakReferenceMessenger.Default.Register<App, TraceMessagesRequest>(this, (r, m) => m.Reply(_traceMessages));    // StatusWindow requests previous messages

        messageGenerator = Task.Run(RandomMessageGenerator);
    }

    private void OnStatusWindowClosed()
    {
        if (ShellPage is not null && ShellPage.SettingsStatusWindow)
        {
            ShellPage.SettingsStatusWindow = false;                                                     // turn off toggle
            if (ShellPage.NavigationFrame.Content is MainPage settingsPage) settingsPage.StatusWindowToggle.IsOn = false;
        }
    }

    private async void OnAppWindowClosing(object sender, AppWindowClosingEventArgs e)
    {
        WeakReferenceMessenger.Default.UnregisterAll(this);                                             // stop messages and avoid memory leaks
        WeakReferenceMessenger.Default.Send(new ShutDownMessage(true));                                 // close all windows

        MainWindow.AppWindow.Closing -= OnAppWindowClosing;

        if (messageGenerator is not null) await messageGenerator;
    }
}

The user may create a StatusWindow (a secondary Window on the UI thread) by toggling a switch on MainPage. The StatusWindow should open, request and load previous messages from the App, then register for new TraceMessages. All TraceMessages (including new ones) are displayed in a ListView on the StatusWindow.

MainPage.xaml.cs

using CommunityToolkit.Mvvm.Messaging;
using Microsoft.UI.Xaml;
using Microsoft.UI.Xaml.Controls;

using Multi_Window.ViewModels;

namespace Multi_Window.Views;

public sealed partial class MainPage : Page
{
    public MainViewModel ViewModel { get; } = App.GetService<MainViewModel>();
    public ShellPage ShellPage { get; } = App.GetService<ShellPage>();
    public ShellViewModel ShellViewModel { get; } = App.GetService<ShellViewModel>();

    public MainPage()
    {
        DataContext = ViewModel;
        InitializeComponent();
    }

    private void StatusWindow_Toggled(object sender, RoutedEventArgs e)
    {
        if (StatusWindowToggle.IsOn && ShellPage.SettingsStatusWindow == false)
        {
            StatusWindow window = new() { Title = "Prosper Status" };
            window.Activate();
            ShellPage.SettingsStatusWindow = true;
        }
        else if (StatusWindowToggle.IsOn == false && ShellPage.SettingsStatusWindow == true)
            WeakReferenceMessenger.Default.Send(new CloseWindowMessage(true));
    }
}

MainPage.xaml

<Page
    x:Class="Multi_Window.Views.MainPage"
    xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
    xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
    xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
    xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
    mc:Ignorable="d">

    <Grid x:Name="ContentArea">
        <ToggleSwitch x:Name="StatusWindowToggle" x:FieldModifier="public" Grid.Row="2" Grid.Column="1" Header="Show Status Window"
                          Toggled="StatusWindow_Toggled" IsOn="{x:Bind ShellPage.SettingsStatusWindow, Mode=OneTime}" />
    </Grid>
</Page>

StatusWindow.xaml.cs

using System.Collections.ObjectModel;
using CommunityToolkit.Mvvm.Messaging;
using Microsoft.UI.Xaml;

// To learn more about WinUI, the WinUI project structure,
// and more about our project templates, see: http://aka.ms/winui-project-info.

namespace Multi_Window.Views;
/// <summary>
/// An empty window that can be used on its own or navigated to within a Frame.
/// </summary>
public sealed partial class StatusWindow : Window
{
    private ObservableCollection<string> _traceMessages { get; } = new();

    public StatusWindow()
    {
        InitializeComponent();
        var sw = new WindowPrimitives(this);
        sw.AppWindow.SetIcon("Assets/wip.ico");

        WeakReferenceMessenger.Default.Register<CloseWindowMessage>(this, (r, m) => Close());
        WeakReferenceMessenger.Default.Register<ShutDownMessage>(this, (r, m) => Close());
    }

    private void StatusWindow_Closed(object sender, WindowEventArgs args)
    {
        WeakReferenceMessenger.Default.UnregisterAll(this);                                             // stop getting messages and avoid memory leaks
        WeakReferenceMessenger.Default.Send(new WindowClosedMessage(true));                             // acknowledge closure
    }

    private void StatusMessages_Loaded(object sender, RoutedEventArgs e)
    {
        // get current Trace messages
        var messages = WeakReferenceMessenger.Default.Send<TraceMessagesRequest>();
        if (messages != null && messages.Responses.Count > 0)
            foreach (var response in messages.Responses)
                foreach (var trace in response)
                    _traceMessages.Add(trace);

        // register for Trace messages and, when they arrive, add them to list
        WeakReferenceMessenger.Default.Register<TraceMessage>(this, (r, m) => _traceMessages.Add(m.Value));
    }
}

StatusPage.xaml

<Window
    x:Class="Multi_Window.Views.StatusWindow"
    xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
    xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
    xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
    xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
    Closed="StatusWindow_Closed"
    mc:Ignorable="d">

    <Grid>
        <Grid.RowDefinitions>
            <RowDefinition Height="*"/>
            <RowDefinition Height="Auto"/>
        </Grid.RowDefinitions>
        <ListView x:Name="StatusMessages" x:FieldModifier="public" VerticalAlignment="Top" Margin="20" SelectionMode="None" BorderBrush="Black" BorderThickness="1"
                    ItemsSource="{x:Bind _traceMessages, Mode=OneWay}"
                    ScrollViewer.HorizontalScrollMode="Enabled"
                    ScrollViewer.HorizontalScrollBarVisibility="Visible"
                    ScrollViewer.IsHorizontalRailEnabled="True" 
                    ScrollViewer.IsDeferredScrollingEnabled="False"
                    Loaded="StatusMessages_Loaded">
            <ListView.ItemsPanel>
                <ItemsPanelTemplate>
                    <ItemsStackPanel VerticalAlignment="Bottom" ItemsUpdatingScrollMode="KeepLastItemInView"/>
                </ItemsPanelTemplate>
            </ListView.ItemsPanel>
        </ListView>
    </Grid>
</Window>

Other Sundry Classes

// Allows Win32 access to a Window through WinAPI
public class WindowPrimitives
{
    public IntPtr HWnd { get; }
    private WindowId WindowId { get; }
    public AppWindow AppWindow { get; }
    public Window Window { get; }

    public WindowPrimitives(Window window)
    {
        HWnd = WinRT.Interop.WindowNative.GetWindowHandle(window);
        WindowId = Win32Interop.GetWindowIdFromWindow(HWnd);
        AppWindow = AppWindow.GetFromWindowId(WindowId);
        Window = window;
    }
}

// Message Definitions

public class CloseWindowMessage : ValueChangedMessage<bool>
{
    public CloseWindowMessage(bool value) : base(value) { }
}

public class WindowClosedMessage : ValueChangedMessage<bool>
{
    public WindowClosedMessage(bool value) : base(value) { }
}

public class ShutDownMessage : ValueChangedMessage<bool>
{
    public ShutDownMessage(bool value) : base(value) { }
}

public class TraceMessage : ValueChangedMessage<string>
{
    public TraceMessage(string value) : base(value) { }
}

public class TraceMessagesRequest : CollectionRequestMessage<List<string>>
{
}

The problem is that, on the first new TraceMessage sent after the StatusWindow has been opened, the same WeakReferenceMessenger.Default.Send() method that has been happily sending messages between the RandomMessageGenerator thread and the UI thread throws a "The application called an interface that was marshalled for a different thread. (0x8001010E (RPC_E_WRONG_THREAD))" exception and the RandomMessageGenerator thread dies.

The exception is thrown by the WeakReferenceMessenger.Default.Send(tm); statement in the RandomMessageGenerator() method. I assume the issue is in the IMessenger interface (the only interface involved here). Briefly, as I understand it, this interface builds a table of subscribed receivers for each message type. Each receiver is then signaled on each Send().

One possibility is that references in the receiver list are all assumed to marshal to the same thread. If that were so, none of the messages between threads would work but they do before the StatusWindow is opened so that's unlikely. Changing the list of receivers is an obvious place where threading issues might occur. As WeakReferenceMessenger.Default is thread-safe, I thought adding (and deleting) registered receivers would be thread-safe but doesn't seem to be the case here. Finally, it could be the message itself (a string in this case) that is at fault. I don't know for sure but assumed that the Send method took a private copy of the message to deliver to the marshaled thread.

Could any of you please help me understand the mistake I've made here?

Answer 1

I did find a solution for this particular issue. As expected, it's because an object is accessed from a thread other than the thread on which it was created.

To fix the error, add

    public static DispatcherQueue UIDispatcherQueue = DispatcherQueue.GetForCurrentThread();

to the App class which will allow any thread access to the UI thread DispatcherQueue. Then, change the message registration in StatusWindow.xaml.cs to

    // register for Trace messages and, when they arrive, add them to list
    WeakReferenceMessenger.Default.Register<TraceMessage>(this, (r, m) => App.UIDispatcherQueue.TryEnqueue(() => _traceMessages.Add(m.Value)));

This will now marshal the _traceMessages.Add call in the message handler to the UI thread, which is where

    private ObservableCollection<string> _traceMessages { get; } = new();

was constructed.

This was easy enough to figure out once I realized that the point where the exception was thrown and the exception message were both rather deceptive. Although the sending of the message is the cause of the message being received, it's the attempt to handle the message on the wrong thread that really throws the exception.

At any rate, this appears to show that the message receiver handler executes on the same thread as the sender. I was hoping that "thread-safe" in the documentation meant messages were automatically marshalled to the receiver's thread.