22740981351

Committed 05 Mar 2026 11:04PM UTC coverage: 33.103% (-0.2%) from 33.333%

Build # 22740981351

Build Type

push

coveralls.net

Committed by

DemoBytom

Commit Message

More work on StaThreadService and the Sta sync context

`StaSingleThreadedSynchronizationContext` - in the end, instead having a `task Completion` exposed to be awaited by the consumer to signal any errors, an event is used. The reason is that awaiting a task, while ensuring it's awaited in a different context, than the main task being executed within this context, led to quite an unintuitive design.
The code and tests were refactored to now rely solely on said event. In case there is a subscriber to the event - it is used to signal error, and a 10 second timeout is started - if the context is not stopped within those 10 seconds the application forcibly exits by exiting the process. If the context is stopped in time, the timeout is cancelled. If no subscriber is subscribed to the event the process exits immediately.

Run Details

1248 of 3770 relevant lines covered (33.1%)

0.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.59

/src/Demo.Engine.Core/Features/StaThread/StaThreadService.cs

// Copyright © Michał Dembski and contributors.
// Distributed under MIT license. See LICENSE file in the root for more information.

using System.Collections.Concurrent;
using System.Diagnostics.CodeAnalysis;
using System.Runtime.InteropServices;
using System.Threading.Channels;
using Demo.Engine.Core.Interfaces;
using Demo.Engine.Core.Interfaces.Platform;
using Demo.Engine.Core.Interfaces.Rendering;
using Demo.Engine.Core.Services;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace Demo.Engine.Core.Features.StaThread;

internal sealed class StaThreadService
    : IStaThreadService
{
    private readonly ILogger<StaThreadService> _logger;
    private readonly IHostApplicationLifetime _hostApplicationLifetime;
    private readonly ChannelReader<StaThreadRequests> _channelReader;
    private readonly IMainLoopLifetime _mainLoopLifetime;

    public Task ExecutingTask { get; }
    public bool IsRunning { get; private set; }

    public StaThreadService(
        ILogger<StaThreadService> logger,
        IHostApplicationLifetime hostApplicationLifetime,
        IRenderingEngine renderingEngine,
        IOSMessageHandler osMessageHandler,
        ChannelReader<StaThreadRequests> channelReader,
        IMainLoopLifetime mainLoopLifetime)
    {
        _logger = logger;
        _hostApplicationLifetime = hostApplicationLifetime;
        _channelReader = channelReader;
        _mainLoopLifetime = mainLoopLifetime;
        IsRunning = true;
        ExecutingTask = RunSTAThread(
            renderingEngine,
            osMessageHandler);
    }

    private async Task RunSTAThread(
        IRenderingEngine renderingEngine,
        IOSMessageHandler osMessageHandler)
    {
        var originalSynContext = SynchronizationContext.Current;
        StaSingleThreadedSynchronizationContext? syncContext = null;
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(
            _hostApplicationLifetime.ApplicationStopping,
            _mainLoopLifetime.Token);

        try
        {

            syncContext = new StaSingleThreadedSynchronizationContext();
            SynchronizationContext.SetSynchronizationContext(syncContext);
            // Task.Yield is used here to force a context switch to STA thread
            // Awaiting here allows the STA thread to start and pump messages, which is necessary for the STAThread method to function correctly.
            await Task.Yield();
            System.Diagnostics.Debug.Assert(
                SynchronizationContext.Current == syncContext,
                "SynchronizationContext should be set to the STA context.");

            try
            {
                syncContext.OnError += OnStaContextError;

                await STAThread(
                        renderingEngine,
                        osMessageHandler,
                        cts.Token)
                    .ConfigureAwait(continueOnCapturedContext: true);
            }
            catch (OperationCanceledException)
            {
                _logger.LogStaThreadServiceWasCancelled();
            }
            catch (Exception)
            {
                IsRunning = false;
                _mainLoopLifetime.Cancel();
                throw;
            }
            IsRunning = false;
            _mainLoopLifetime.Cancel();

        }
        finally
        {
            SynchronizationContext.SetSynchronizationContext(originalSynContext);
            // Force context switch back to original context to ensure all STA thread work is completed before disposing the syncContext
            await Task.Yield();
            syncContext?.OnError -= OnStaContextError;
            syncContext?.Dispose();
        }

        void OnStaContextError(object? sender, Exception ex)
        {
            _logger.LogCritical(ex,
                "An error occured within the STA thread synchronization context.");
            cts.Cancel();
        }
    }

    private async Task STAThread(
        IRenderingEngine renderingEngine,
        IOSMessageHandler osMessageHandler,
        CancellationToken cancellationToken)
    {
        var doEventsOk = true;

        await foreach (var staAction in _channelReader
            .ReadAllAsync(cancellationToken)
            .ConfigureAwait(continueOnCapturedContext: true)
            .WithCancellation(cancellationToken))
        {
            switch (staAction)
            {
                case StaThreadRequests.DoEventsOkRequest doEventsOkRequest:
                    doEventsOk &= await doEventsOkRequest
                        .InvokeAsync(renderingEngine, osMessageHandler, cancellationToken)
                        .ConfigureAwait(continueOnCapturedContext: true);
                    break;

                default:
                    _ = await staAction
                        .InvokeAsync(renderingEngine, osMessageHandler, cancellationToken)
                        .ConfigureAwait(continueOnCapturedContext: true);
                    break;
            }

            if (!doEventsOk || !IsRunning || cancellationToken.IsCancellationRequested)
            {
                break;
            }
        }
    }

    internal sealed class StaSingleThreadedSynchronizationContext
        : SynchronizationContext,
          IDisposable
    {
        private readonly BlockingCollection<WorkItem> _workQueue = [];

        private readonly Thread _thread;
        private readonly CancellationTokenSource _cancellationTokenSource = new();
        private bool _isDisposed;

        public event EventHandler<Exception>? OnError;

        /// <inheritdoc/>
        public override void Post(SendOrPostCallback d, object? state)
            => _workQueue.Add(WorkItem.Asynchronous(d, state));

        /// <inheritdoc/>
        public override void Send(SendOrPostCallback d, object? state)
        {
            // If we're already on the STA thread, execute the callback directly to avoid deadlock.
            if (Environment.CurrentManagedThreadId == _thread.ManagedThreadId)
            {
                d.Invoke(state);
                return;
            }

            var workItem = WorkItem.Synchronous(d, state);

            try
            {
                _workQueue.Add(workItem);
                workItem.SyncEvent!.Wait();

                if (workItem.Exception is { } exception)
                {
                    throw exception;
                }
            }
            finally
            {
                workItem.Dispose();
            }
        }

        public StaSingleThreadedSynchronizationContext()
        {
            _thread = new Thread(ThreadInner)
            {
                IsBackground = false,
            };

            if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
            {
                //Can only by set on the Windows machine. Doesn't work on Linux/MacOS
                _thread.SetApartmentState(ApartmentState.STA);
                _thread.Name = "Main STA thread";
            }
            else
            {
                _thread.Name = "Main thread";
            }

            _thread.Start();
        }

        private void ThreadInner()
        {
            SetSynchronizationContext(this);
            try
            {
                foreach (var workItem in _workQueue.GetConsumingEnumerable(_cancellationTokenSource.Token))
                {
                    try
                    {
                        workItem.Callback.Invoke(workItem.State);
                    }
                    catch (Exception ex)
                    {
                        if (workItem.IsSynchronous)
                        {
                            workItem.Exception = ex;
                        }
                        else
                        {
                            /* An exception cannot be thrown to the caller from here
                             * since the caller has already continued execution after posting the work item. 
                             * Instead an event is rised, if there's subscribers, or the process is terminated
                             * If there are subscribers, but they don't cancel the sync context within 10 seconds
                             * the process is terminated to avoid running in an unstable state. */
                            if (OnError is not null)
                            {
                                OnError.Invoke(this, ex);
                                _ = ExitAfter10Seconds(_cancellationTokenSource.Token);
                            }
                            else
                            {
                                Environment.Exit(-1);
                            }
                        }

                        static async Task ExitAfter10Seconds(CancellationToken cancellationToken)
                        {
                            try
                            {
                                await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken)
                                    .ConfigureAwait(false);
                            }
                            catch (Exception)
                                when (cancellationToken.IsCancellationRequested)
                            {
                                return;
                            }

                            Environment.Exit(-1);
                        }
                    }
                    finally
                    {
                        if (workItem.IsSynchronous)
                        {
                            workItem.SyncEvent.Set();
                        }
                        else
                        {
                            workItem.Dispose();
                        }
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // Gracefully exit the thread when cancellation is requested.
            }
            catch (InvalidOperationException)
            {
                // The BlockingCollection has been marked as complete for adding, which means we're shutting down.
            }
        }

        public void Dispose()
        {
            if (_isDisposed)
            {
                return;
            }
            _isDisposed = true;

            // Gracefully shutdown the message pump
            _cancellationTokenSource.Cancel();
            _workQueue.CompleteAdding();

            _thread.Join();
            _workQueue.Dispose();
            _cancellationTokenSource.Dispose();
        }

        private sealed class WorkItem
            : IDisposable
        {
            private bool _isDisposed;

            public SendOrPostCallback Callback { get; }

            public object? State { get; }

            [MemberNotNullWhen(true, nameof(SyncEvent))]
            public bool IsSynchronous { get; }

            public ManualResetEventSlim? SyncEvent { get; }

            public Exception? Exception { get; set; }

            private WorkItem(
                SendOrPostCallback callback,
                object? state,
                bool isSynchronous)
            {
                Callback = callback;
                State = state;
                IsSynchronous = isSynchronous;
                if (isSynchronous)
                {
                    SyncEvent = new ManualResetEventSlim();
                }
            }

            public static WorkItem Synchronous(SendOrPostCallback callback, object? state)
                => new(
                    callback: callback,
                    state: state,
                    isSynchronous: true);

            public static WorkItem Asynchronous(SendOrPostCallback callback, object? state)
                => new(
                    callback: callback,
                    state: state,
                    isSynchronous: false);

            public void Dispose()
            {
                if (_isDisposed)
                {
                    return;
                }

                _isDisposed = true;

                SyncEvent?.Dispose();
            }
        }
    }
}

1	// Copyright © Michał Dembski and contributors.
2	// Distributed under MIT license. See LICENSE file in the root for more information.
3
4	using System.Collections.Concurrent;
5	using System.Diagnostics.CodeAnalysis;
6	using System.Runtime.InteropServices;
7	using System.Threading.Channels;
8	using Demo.Engine.Core.Interfaces;
9	using Demo.Engine.Core.Interfaces.Platform;
10	using Demo.Engine.Core.Interfaces.Rendering;
11	using Demo.Engine.Core.Services;
12	using Microsoft.Extensions.Hosting;
13	using Microsoft.Extensions.Logging;
14
15	namespace Demo.Engine.Core.Features.StaThread;
16
17	internal sealed class StaThreadService
18	: IStaThreadService
19	{
20	private readonly ILogger<StaThreadService> _logger;
21	private readonly IHostApplicationLifetime _hostApplicationLifetime;
22	private readonly ChannelReader<StaThreadRequests> _channelReader;
23	private readonly IMainLoopLifetime _mainLoopLifetime;
24
25	public Task ExecutingTask { get; }	1✔
26	public bool IsRunning { get; private set; }	1✔
27
28	public StaThreadService(	1✔
29	ILogger<StaThreadService> logger,	1✔
30	IHostApplicationLifetime hostApplicationLifetime,	1✔
31	IRenderingEngine renderingEngine,	1✔
32	IOSMessageHandler osMessageHandler,	1✔
33	ChannelReader<StaThreadRequests> channelReader,	1✔
34	IMainLoopLifetime mainLoopLifetime)	1✔
35	{
36	_logger = logger;	1✔
37	_hostApplicationLifetime = hostApplicationLifetime;	1✔
38	_channelReader = channelReader;	1✔
39	_mainLoopLifetime = mainLoopLifetime;	1✔
40	IsRunning = true;	1✔
41	ExecutingTask = RunSTAThread(	1✔
42	renderingEngine,	1✔
43	osMessageHandler);	1✔
44	}	1✔
45
46	private async Task RunSTAThread(
47	IRenderingEngine renderingEngine,
48	IOSMessageHandler osMessageHandler)
49	{
50	var originalSynContext = SynchronizationContext.Current;	1✔
51	StaSingleThreadedSynchronizationContext? syncContext = null;	1✔
52	using var cts = CancellationTokenSource.CreateLinkedTokenSource(	1✔
53	_hostApplicationLifetime.ApplicationStopping,	1✔
54	_mainLoopLifetime.Token);	1✔
55
56	try
57	{
58
59	syncContext = new StaSingleThreadedSynchronizationContext();	1✔
60	SynchronizationContext.SetSynchronizationContext(syncContext);	1✔
61	// Task.Yield is used here to force a context switch to STA thread
62	// Awaiting here allows the STA thread to start and pump messages, which is necessary for the STAThread method to function correctly.
63	await Task.Yield();	1✔
64	System.Diagnostics.Debug.Assert(
65	SynchronizationContext.Current == syncContext,
66	"SynchronizationContext should be set to the STA context.");
67
68	try
69	{
70	syncContext.OnError += OnStaContextError;	1✔
71
72	await STAThread(	1✔
73	renderingEngine,	1✔
74	osMessageHandler,	1✔
75	cts.Token)	1✔
76	.ConfigureAwait(continueOnCapturedContext: true);	1✔
77	}
78	catch (OperationCanceledException)	1✔
79	{
80	_logger.LogStaThreadServiceWasCancelled();	1✔
81	}	1✔
82	catch (Exception)
83	{
84	IsRunning = false;
85	_mainLoopLifetime.Cancel();
86	throw;
87	}
88	IsRunning = false;	1✔
89	_mainLoopLifetime.Cancel();	1✔
90
91	}
92	finally
93	{
94	SynchronizationContext.SetSynchronizationContext(originalSynContext);	1✔
95	// Force context switch back to original context to ensure all STA thread work is completed before disposing the syncContext
96	await Task.Yield();	1✔
97	syncContext?.OnError -= OnStaContextError;
98	syncContext?.Dispose();
99	}
100
101	void OnStaContextError(object? sender, Exception ex)
102	{
103	_logger.LogCritical(ex,
104	"An error occured within the STA thread synchronization context.");
105	cts.Cancel();
106	}
107	}	1✔
108
109	private async Task STAThread(
110	IRenderingEngine renderingEngine,
111	IOSMessageHandler osMessageHandler,
112	CancellationToken cancellationToken)
113	{
114	var doEventsOk = true;	1✔
115
116	await foreach (var staAction in _channelReader
117	.ReadAllAsync(cancellationToken)	1✔
118	.ConfigureAwait(continueOnCapturedContext: true)	1✔
119	.WithCancellation(cancellationToken))	1✔
120	{
121	switch (staAction)
122	{
123	case StaThreadRequests.DoEventsOkRequest doEventsOkRequest:
124	doEventsOk &= await doEventsOkRequest
125	.InvokeAsync(renderingEngine, osMessageHandler, cancellationToken)
126	.ConfigureAwait(continueOnCapturedContext: true);
127	break;
128
129	default:
130	_ = await staAction	1✔
131	.InvokeAsync(renderingEngine, osMessageHandler, cancellationToken)	1✔
132	.ConfigureAwait(continueOnCapturedContext: true);	1✔
133	break;
134	}
135
136	if (!doEventsOk \|\| !IsRunning \|\| cancellationToken.IsCancellationRequested)
137	{
138	break;
139	}
140	}
141	}
142
143	internal sealed class StaSingleThreadedSynchronizationContext
144	: SynchronizationContext,
145	IDisposable
146	{
147	private readonly BlockingCollection<WorkItem> _workQueue = [];	2✔
148
149	private readonly Thread _thread;
150	private readonly CancellationTokenSource _cancellationTokenSource = new();	2✔
151	private bool _isDisposed;
152
153	public event EventHandler<Exception>? OnError;
154
155	/// <inheritdoc/>
156	public override void Post(SendOrPostCallback d, object? state)
157	=> _workQueue.Add(WorkItem.Asynchronous(d, state));	2✔
158
159	/// <inheritdoc/>
160	public override void Send(SendOrPostCallback d, object? state)
161	{
162	// If we're already on the STA thread, execute the callback directly to avoid deadlock.
163	if (Environment.CurrentManagedThreadId == _thread.ManagedThreadId)
164	{
165	d.Invoke(state);
166	return;
167	}
168
169	var workItem = WorkItem.Synchronous(d, state);
170
171	try
172	{
173	_workQueue.Add(workItem);
174	workItem.SyncEvent!.Wait();
175
176	if (workItem.Exception is { } exception)
177	{
178	throw exception;
179	}
180	}
181	finally
182	{
183	workItem.Dispose();
184	}
185	}
186
187	public StaSingleThreadedSynchronizationContext()	2✔
188	{
189	_thread = new Thread(ThreadInner)	2✔
190	{	2✔
191	IsBackground = false,	2✔
192	};	2✔
193
194	if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
195	{
196	//Can only by set on the Windows machine. Doesn't work on Linux/MacOS
197	_thread.SetApartmentState(ApartmentState.STA);	2✔
198	_thread.Name = "Main STA thread";	2✔
199	}
200	else
201	{
202	_thread.Name = "Main thread";
203	}
204
205	_thread.Start();	2✔
206	}	2✔
207
208	private void ThreadInner()
209	{
210	SetSynchronizationContext(this);	2✔
211	try
212	{
213	foreach (var workItem in _workQueue.GetConsumingEnumerable(_cancellationTokenSource.Token))
214	{
215	try
216	{
217	workItem.Callback.Invoke(workItem.State);	2✔
218	}	2✔
219	catch (Exception ex)	2✔
220	{
221	if (workItem.IsSynchronous)
222	{
223	workItem.Exception = ex;
224	}
225	else
226	{
227	/* An exception cannot be thrown to the caller from here
228	* since the caller has already continued execution after posting the work item.
229	* Instead an event is rised, if there's subscribers, or the process is terminated
230	* If there are subscribers, but they don't cancel the sync context within 10 seconds
231	* the process is terminated to avoid running in an unstable state. */
232	if (OnError is not null)
233	{
234	OnError.Invoke(this, ex);	2✔
235	_ = ExitAfter10Seconds(_cancellationTokenSource.Token);	2✔
236	}
237	else
238	{
239	Environment.Exit(-1);
240	}
241	}
242
243	static async Task ExitAfter10Seconds(CancellationToken cancellationToken)
244	{
245	try
246	{
247	await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken)	2✔
248	.ConfigureAwait(false);	2✔
249	}
250	catch (Exception)
251	when (cancellationToken.IsCancellationRequested)	2✔
252	{
253	return;	2✔
254	}
255
256	Environment.Exit(-1);
257	}	2✔
258	}	2✔
259	finally
260	{
261	if (workItem.IsSynchronous)
262	{
263	workItem.SyncEvent.Set();
264	}
265	else
266	{
267	workItem.Dispose();	2✔
268	}
269	}	2✔
270	}
271	}
272	catch (OperationCanceledException)	2✔
273	{
274	// Gracefully exit the thread when cancellation is requested.
275	}	2✔
276	catch (InvalidOperationException)
277	{
278	// The BlockingCollection has been marked as complete for adding, which means we're shutting down.
279	}
280	}	2✔
281
282	public void Dispose()
283	{
284	if (_isDisposed)
285	{
286	return;
287	}
288	_isDisposed = true;	2✔
289
290	// Gracefully shutdown the message pump
291	_cancellationTokenSource.Cancel();	2✔
292	_workQueue.CompleteAdding();	2✔
293
294	_thread.Join();	2✔
295	_workQueue.Dispose();	2✔
296	_cancellationTokenSource.Dispose();	2✔
297	}	2✔
298
299	private sealed class WorkItem
300	: IDisposable
301	{
302	private bool _isDisposed;
303
304	public SendOrPostCallback Callback { get; }	3✔
305
306	public object? State { get; }	3✔
307
308	[MemberNotNullWhen(true, nameof(SyncEvent))]
309	public bool IsSynchronous { get; }	3✔
310
311	public ManualResetEventSlim? SyncEvent { get; }	3✔
312
313	public Exception? Exception { get; set; }	×
314
315	private WorkItem(	3✔
316	SendOrPostCallback callback,	3✔
317	object? state,	3✔
318	bool isSynchronous)	3✔
319	{
320	Callback = callback;	3✔
321	State = state;	3✔
322	IsSynchronous = isSynchronous;	3✔
323	if (isSynchronous)
324	{
325	SyncEvent = new ManualResetEventSlim();	×
326	}
327	}	3✔
328
329	public static WorkItem Synchronous(SendOrPostCallback callback, object? state)
330	=> new(	×
331	callback: callback,	×
332	state: state,	×
333	isSynchronous: true);	×
334
335	public static WorkItem Asynchronous(SendOrPostCallback callback, object? state)
336	=> new(	3✔
337	callback: callback,	3✔
338	state: state,	3✔
339	isSynchronous: false);	3✔
340
341	public void Dispose()
342	{
343	if (_isDisposed)
344	{
345	return;	×
346	}
347
348	_isDisposed = true;	3✔
349
350	SyncEvent?.Dispose();
351	}	×
352	}
353	}
354	}

DemoBytom / DemoEngine / 22740981351

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous