fix: make STA Word queue lifetime safe

This commit is contained in:
codex
2026-07-30 16:09:15 +03:00
parent f7fa0363e3
commit 25e1def353
3 changed files with 54 additions and 9 deletions
@@ -0,0 +1,3 @@
using System.Runtime.CompilerServices;
[assembly: InternalsVisibleTo("DocxWorkbench.Tests")]
@@ -6,6 +6,9 @@ internal sealed class StaWordQueue : IDisposable
{
private readonly BlockingCollection<Action> _jobs = new();
private readonly Thread _thread;
private readonly ManualResetEventSlim _stopped = new(false);
private readonly object _lifetimeGate = new();
private bool _accepting = true;
public StaWordQueue()
{
@@ -16,25 +19,42 @@ internal sealed class StaWordQueue : IDisposable
public Task<T> Enqueue<T>(Func<T> operation, CancellationToken cancellationToken)
{
if (operation is null) throw new ArgumentNullException(nameof(operation));
if (cancellationToken.IsCancellationRequested) return Task.FromCanceled<T>(cancellationToken);
var completion = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
_jobs.Add(() =>
lock (_lifetimeGate)
{
try { cancellationToken.ThrowIfCancellationRequested(); completion.SetResult(operation()); }
catch (OperationCanceledException ex) { completion.SetCanceled(ex.CancellationToken); }
catch (Exception ex) { completion.SetException(ex); }
}, cancellationToken);
if (!_accepting) throw new ObjectDisposedException(nameof(StaWordQueue));
try { _jobs.Add(() => Execute(operation, completion, cancellationToken), cancellationToken); }
catch (InvalidOperationException) { throw new ObjectDisposedException(nameof(StaWordQueue)); }
catch (OperationCanceledException) { completion.TrySetCanceled(cancellationToken); }
}
return completion.Task;
}
private static void Execute<T>(Func<T> operation, TaskCompletionSource<T> completion, CancellationToken cancellationToken)
{
try { cancellationToken.ThrowIfCancellationRequested(); completion.TrySetResult(operation()); }
catch (OperationCanceledException ex) { completion.TrySetCanceled(ex.CancellationToken); }
catch (Exception ex) { completion.TrySetException(ex); }
}
private void Run()
{
foreach (var job in _jobs.GetConsumingEnumerable()) job();
try { foreach (var job in _jobs.GetConsumingEnumerable()) job(); }
finally { _stopped.Set(); }
}
public void Dispose()
{
_jobs.CompleteAdding();
if (Thread.CurrentThread != _thread) _thread.Join(TimeSpan.FromSeconds(5));
_jobs.Dispose();
lock (_lifetimeGate)
{
if (!_accepting) return;
_accepting = false;
_jobs.CompleteAdding();
}
if (Thread.CurrentThread != _thread) _stopped.Wait(TimeSpan.FromSeconds(30));
// Do not dispose the collection or stop event while the STA worker may
// still be draining it. The worker owns the final enumeration lifetime.
}
}
@@ -0,0 +1,22 @@
using DocxWorkbench.WordInterop;
namespace DocxWorkbench.Tests;
public sealed class StaWordQueueTests
{
[Fact]
public async Task Dispose_DrainsActiveJobAndIsIdempotent()
{
using var queue = new StaWordQueue(); using var started = new ManualResetEventSlim(false); using var release = new ManualResetEventSlim(false);
var job = queue.Enqueue(() => { started.Set(); Assert.Equal(ApartmentState.STA, Thread.CurrentThread.GetApartmentState()); release.Wait(); return 42; }, CancellationToken.None); Assert.True(started.Wait(TimeSpan.FromSeconds(2)));
var dispose = Task.Run(queue.Dispose); var completed = await Task.WhenAny(dispose, Task.Delay(100)); Assert.NotSame(dispose, completed); release.Set(); await dispose; Assert.Equal(42, await job); queue.Dispose(); Assert.Throws<ObjectDisposedException>(() => EnqueueAfterDispose(queue));
}
[Fact]
public async Task Enqueue_CancellationCompletesCanceledTask()
{
using var queue = new StaWordQueue(); using var cancellation = new CancellationTokenSource(); cancellation.Cancel(); await Assert.ThrowsAnyAsync<OperationCanceledException>(() => queue.Enqueue(() => 1, cancellation.Token));
}
private static void EnqueueAfterDispose(StaWordQueue queue) => queue.Enqueue(() => 1, CancellationToken.None);
}