21806835161

Committed 08 Feb 2026 09:00PM UTC coverage: 31.062% (-0.08%) from 31.138%

Build # 21806835161

Build Type

push

coveralls.net

Committed by

DemoBytom

Commit Message

Refactor STA Thread `SingleThreadedSynchronizationContext`

New implementation now uses `Channel<T>` instead `BlockingCollection<T>` as the queue for collecting and executing async `SendOrPostCallback`s.

Run Details

1117 of 3596 relevant lines covered (31.06%)

0.32 hits per line

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

0.0

/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.Runtime.InteropServices;
using System.Threading.Channels;
using Demo.Engine.Core.Interfaces;
using Demo.Engine.Core.Interfaces.Platform;
using Demo.Engine.Core.Interfaces.Rendering;
using Microsoft.Extensions.Hosting;

namespace Demo.Engine.Core.Features.StaThread;

internal sealed class StaThreadService
    : IStaThreadService,
      IDisposable
{
    private readonly IHostApplicationLifetime _hostApplicationLifetime;
    private readonly ChannelReader<StaThreadRequests> _channelReader;
    private readonly IMainLoopLifetime _mainLoopLifetime;
    private bool _disposedValue;

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

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

    private Task RunSTAThread(
        IRenderingEngine renderingEngine,
        IOSMessageHandler osMessageHandler)
    {
        var tcs = new TaskCompletionSource();
        var thread = new Thread(()
            =>
        {
            try
            {
                using var cts = CancellationTokenSource.CreateLinkedTokenSource(
                    _hostApplicationLifetime.ApplicationStopping,
                    _mainLoopLifetime.Token);

                SingleThreadedSynchronizationContextChannel.Await(async ()
                    => await STAThread(
                        renderingEngine: renderingEngine,
                        osMessageHandler: osMessageHandler,
                        cancellationToken: cts.Token));

                tcs.SetResult();
            }
            catch (OperationCanceledException)
            {
                tcs.SetResult();
            }
            catch (Exception ex)
            {
                tcs.SetException(ex);
            }
            IsRunning = false;
            _mainLoopLifetime.Cancel();
        });
        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();

        return tcs.Task;
    }

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

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

                default:
                    _ = staAction.Invoke(renderingEngine, osMessageHandler);
                    break;
            }

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

    private void Dispose(bool disposing)
    {
        if (!_disposedValue)
        {
            if (disposing)
            {
            }

            _disposedValue = true;
        }
    }

    public void Dispose()
    {
        // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }

    private sealed class SingleThreadedSynchronizationContextChannel
        : SynchronizationContext
    {
        private readonly Channel<(SendOrPostCallback d, object? state)> _channel =
            Channel.CreateUnbounded<(SendOrPostCallback d, object? state)>(
                new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });

        public override void Post(SendOrPostCallback d, object? state)
            => _channel.Writer.TryWrite((d, state));

        public override void Send(SendOrPostCallback d, object? state)
            => throw new InvalidOperationException("Synchronous operations are not supported!");

        public static void Await(Func<Task> taskInvoker)
        {
            var originalContext = Current;
            try
            {
                var context = new SingleThreadedSynchronizationContextChannel();
                SetSynchronizationContext(context);

                Task task;
                try
                {
                    task = taskInvoker.Invoke();
                }
                catch (Exception ex)
                {
                    // If the invoker throws synchronously, complete the channel so the pump can exit.
                    context._channel.Writer.Complete(ex);
                    throw;
                }

                _ = task.ContinueWith(t
                    => context._channel.Writer.Complete(t.Exception),
                    TaskScheduler.Default);

                // Pump loop: block synchronously until items are available or the writer completes.
                while (context._channel.Reader.WaitToReadAsync().Preserve().GetAwaiter().GetResult())
                {
                    while (context._channel.Reader.TryRead(out var work))
                    {
                        work.d.Invoke(work.state);
                    }
                }

                task.GetAwaiter().GetResult();
            }
            finally
            {
                SetSynchronizationContext(originalContext);
            }
        }
    }
}

1	// Copyright © Michał Dembski and contributors.
2	// Distributed under MIT license. See LICENSE file in the root for more information.
3
4	using System.Runtime.InteropServices;
5	using System.Threading.Channels;
6	using Demo.Engine.Core.Interfaces;
7	using Demo.Engine.Core.Interfaces.Platform;
8	using Demo.Engine.Core.Interfaces.Rendering;
9	using Microsoft.Extensions.Hosting;
10
11	namespace Demo.Engine.Core.Features.StaThread;
12
13	internal sealed class StaThreadService
14	: IStaThreadService,
15	IDisposable
16	{
17	private readonly IHostApplicationLifetime _hostApplicationLifetime;
18	private readonly ChannelReader<StaThreadRequests> _channelReader;
19	private readonly IMainLoopLifetime _mainLoopLifetime;
20	private bool _disposedValue;
21
22	public Task ExecutingTask { get; }
23	public bool IsRunning { get; private set; }
24
25	public StaThreadService(
26	IHostApplicationLifetime hostApplicationLifetime,
27	IRenderingEngine renderingEngine,
28	IOSMessageHandler osMessageHandler,
29	ChannelReader<StaThreadRequests> channelReader,
30	IMainLoopLifetime mainLoopLifetime)
31	{
32	_hostApplicationLifetime = hostApplicationLifetime;
33	_channelReader = channelReader;
34	_mainLoopLifetime = mainLoopLifetime;
35	IsRunning = true;
36	ExecutingTask = RunSTAThread(
37	renderingEngine,
38	osMessageHandler);
39	}
40
41	private Task RunSTAThread(
42	IRenderingEngine renderingEngine,
43	IOSMessageHandler osMessageHandler)
44	{
45	var tcs = new TaskCompletionSource();
46	var thread = new Thread(()
47	=>
48	{
49	try
50	{
51	using var cts = CancellationTokenSource.CreateLinkedTokenSource(
52	_hostApplicationLifetime.ApplicationStopping,
53	_mainLoopLifetime.Token);
54
55	SingleThreadedSynchronizationContextChannel.Await(async ()
56	=> await STAThread(
57	renderingEngine: renderingEngine,
58	osMessageHandler: osMessageHandler,
59	cancellationToken: cts.Token));
60
61	tcs.SetResult();
62	}
63	catch (OperationCanceledException)
64	{
65	tcs.SetResult();
66	}
67	catch (Exception ex)
68	{
69	tcs.SetException(ex);
70	}
71	IsRunning = false;
72	_mainLoopLifetime.Cancel();
73	});
74	if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
75	{
76	//Can only by set on the Windows machine. Doesn't work on Linux/MacOS
77	thread.SetApartmentState(ApartmentState.STA);
78	thread.Name = "Main STA thread";
79	}
80	else
81	{
82	thread.Name = "Main thread";
83	}
84
85	thread.Start();
86
87	return tcs.Task;
88	}
89
90	private async Task STAThread(
91	IRenderingEngine renderingEngine,
92	IOSMessageHandler osMessageHandler,
93	CancellationToken cancellationToken)
94	{
95	var doEventsOk = true;
96
97	await foreach (var staAction in _channelReader
98	.ReadAllAsync(cancellationToken)
99	.WithCancellation(cancellationToken))
100	{
101	switch (staAction)
102	{
103	case StaThreadRequests.DoEventsOkRequest doEventsOkRequest:
104	doEventsOk &= doEventsOkRequest.Invoke(renderingEngine, osMessageHandler);
105	break;
106
107	default:
108	_ = staAction.Invoke(renderingEngine, osMessageHandler);
109	break;
110	}
111
112	if (!doEventsOk \|\| !IsRunning \|\| cancellationToken.IsCancellationRequested)
113	{
114	break;
115	}
116	}
117	}
118
119	private void Dispose(bool disposing)
120	{
121	if (!_disposedValue)
122	{
123	if (disposing)
124	{
125	}
126
127	_disposedValue = true;
128	}
129	}
130
131	public void Dispose()
132	{
133	// Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method
134	Dispose(disposing: true);
135	GC.SuppressFinalize(this);
136	}
137
138	private sealed class SingleThreadedSynchronizationContextChannel
139	: SynchronizationContext
140	{
141	private readonly Channel<(SendOrPostCallback d, object? state)> _channel =	×
142	Channel.CreateUnbounded<(SendOrPostCallback d, object? state)>(	×
143	new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });	×
144
145	public override void Post(SendOrPostCallback d, object? state)
146	=> _channel.Writer.TryWrite((d, state));	×
147
148	public override void Send(SendOrPostCallback d, object? state)
149	=> throw new InvalidOperationException("Synchronous operations are not supported!");	×
150
151	public static void Await(Func<Task> taskInvoker)
152	{
153	var originalContext = Current;	×
154	try
155	{
156	var context = new SingleThreadedSynchronizationContextChannel();	×
157	SetSynchronizationContext(context);	×
158
159	Task task;
160	try
161	{
162	task = taskInvoker.Invoke();	×
163	}	×
164	catch (Exception ex)	×
165	{
166	// If the invoker throws synchronously, complete the channel so the pump can exit.
167	context._channel.Writer.Complete(ex);	×
168	throw;	×
169	}
170
171	_ = task.ContinueWith(t	×
172	=> context._channel.Writer.Complete(t.Exception),	×
173	TaskScheduler.Default);	×
174
175	// Pump loop: block synchronously until items are available or the writer completes.
176	while (context._channel.Reader.WaitToReadAsync().Preserve().GetAwaiter().GetResult())	×
177	{
178	while (context._channel.Reader.TryRead(out var work))	×
179	{
180	work.d.Invoke(work.state);	×
181	}	×
182	}
183
184	task.GetAwaiter().GetResult();	×
185	}	×
186	finally
187	{
188	SetSynchronizationContext(originalContext);	×
189	}	×
190	}	×
191	}
192	}

DemoBytom / DemoEngine / 21806835161

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