我使用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
如果捕获并忽略了异常,则代码正常工作。但我想摆脱其原因尚不清楚的错误。
这是我的环境和最少的代码。
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);
}
}
}
}
你这样can't directly block on a ValueTask<T>
:
_Channel.Reader.WaitToReadAsync(_Cancellation.Token)。结果
你只能用ValueTask<T>
做两件事:await
它(只有一次),或者通过调用Task<T>
将它转换为AsTask()
。如果你需要做任何复杂的事情,比如await
ing它不止一次或阻塞它,那么你需要使用AsTask()
。
或者,在这种情况下,只需在标准频道消费模式中使用await
:
class AsyncChannels : IDisposable
{
Channel<TaskCompletionSource<bool>> _Channel;
Task _Thread;
CancellationTokenSource _Cancellation;
public AsyncChannels()
{
_Channel = Channel.CreateUnbounded<TaskCompletionSource<bool>>();
_Thread = Task.Run(() => WorkerAsync());
_Cancellation = new CancellationTokenSource();
}
private async Task WorkerAsync()
{
try
{
while (await _Channel.Reader.WaitToReadAsync(_Cancellation.Token))
while (_Channel.Reader.TryRead(out var item))
item.TrySetResult(true);
}
catch (OperationCanceledException)
{
}
}
public void Dispose()
{
_Cancellation.Cancel();
_Channel.Writer.TryComplete();
}
...
}