22785668159

Committed 06 Mar 2026 11:05PM UTC coverage: 33.06% (-0.04%) from 33.103%

Build # 22785668159

Build Type

push

coveralls.net

Committed by

DemoBytom

Commit Message

Experimenting with object pooling for context work items

So far about 41% less memory allocated when using object pooling.

Run Details

1252 of 3787 relevant lines covered (33.06%)

0.36 hits per line

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

83.2

/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;
using Microsoft.Extensions.ObjectPool;

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 readonly ObjectPool<WorkItem>? _workItemPool;
        private bool _isDisposed;

        public event EventHandler<Exception>? OnError;

        /// <inheritdoc/>
        public override void Post(SendOrPostCallback d, object? state)
            => _workQueue.Add(_workItemPool?
                    .Get()
                    .Set(d, state, isSynchronous: false)
                ?? 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 = _workItemPool?
                    .Get()
                    .Set(d, state, isSynchronous: true)
                ?? WorkItem.Synchronous(d, state);

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

                if (workItem.Exception is { } exception)
                {
                    throw exception;
                }
            }
            finally
            {
                if (_workItemPool is not null)
                {
                    _workItemPool.Return(workItem);
                }
                else
                {
                    workItem.Dispose();
                }
            }
        }

        public StaSingleThreadedSynchronizationContext(
            ObjectPool<WorkItem>? workItemPool = null)
        {
            _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();
            _workItemPool = workItemPool;
        }

        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 if (_workItemPool is not null)
                        {
                            _workItemPool.Return(workItem);
                        }
                        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();
        }

        internal sealed class WorkItem
            : IDisposable,
              IResettable
        {
            private bool _isDisposed;

            public SendOrPostCallback Callback { get; private set; }

            public object? State { get; private set; }

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

            public ManualResetEventSlim? SyncEvent { get; private set; }

            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 WorkItem()
            {
                Callback = null!;
            }

            public WorkItem Set(
                SendOrPostCallback callback,
                object? state,
                bool isSynchronous)
            {
                Callback = callback;
                State = state;
                IsSynchronous = isSynchronous;

                if (IsSynchronous)
                {
                    (SyncEvent ??= new ManualResetEventSlim())
                        .Reset();
                }

                return this;
            }

            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();
            }

            public bool TryReset()
            {
                SyncEvent?.Reset();
                Exception = null;

                return true;
            }
        }
    }
}

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

DemoBytom / DemoEngine / 22785668159

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