System.Threading.Channels を使用して非同期キューを作成しました。しかし、テストのためにプログラムを実行すると、ランダムなタイミングで次の例外がスローされ、ワーカー スレッドが停止しました。
System.InvalidOperationException: The asynchronous operation has not completed.
at System.Threading.Channels.AsyncOperation.ThrowIncompleteOperationException()
at System.Threading.Channels.AsyncOperation`1.GetResult(Int16 token)
at AsyncChannels.Worker() in g:\src\gitrepos\dotnet-sandbox\channelstest\AsyncChannelsTest.cs:line 26
例外がキャッチされて無視された場合、コードは機能しています。でも、原因不明のエラーをなくしたい。
ここに私の環境と最小のコードがあります。
- TargetFramework = netcoreapp2.1
- System.Threading.Channels バージョン = 4.5.0
using System.Threading.Channels;
using System.Threading;
using System.Threading.Tasks;
using System;
using System.Linq;
class AsyncChannels : IDisposable
{
Channel<TaskCompletionSource<bool>> _Channel;
Thread _Thread;
CancellationTokenSource _Cancellation;
public AsyncChannels()
{
_Channel = Channel.CreateUnbounded<TaskCompletionSource<bool>>();
_Thread = new Thread(Worker);
_Thread.Start();
_Cancellation = new CancellationTokenSource();
}
private void Worker()
{
while (!_Cancellation.IsCancellationRequested)
{
// System.InvalidOperationException is thrown
if (!_Channel.Reader.WaitToReadAsync(_Cancellation.Token).Result)
{
break;
}
while (_Channel.Reader.TryRead(out var item))
{
item.TrySetResult(true);
}
}
}
public void Dispose()
{
_Cancellation.Cancel();
_Channel.Writer.TryComplete();
_Thread.Join();
}
public Task<bool> Enqueue()
{
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
_Channel.Writer.TryWrite(tcs);
return tcs.Task;
}
public static async Task Test()
{
using (var queue = new AsyncChannels())
{
for (int i = 0; i < 100000; i++)
{
await queue.Enqueue().ConfigureAwait(false);
}
}
}
}