gpt4 book ai didi

c# - 如何避免在 ChannelReader.WaitToReadAsync 时抛出 InvalidOperationException?

转载 作者:行者123 更新时间:2023-12-03 13:18:52 32 4
gpt4 key购买 nike

我使用 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);
}
}
}
}

最佳答案

can't directly block on a ValueTask<T> 像这样:

_Channel.Reader.WaitToReadAsync(_Cancellation.Token).Result

你只能用 ValueTask<T> 做两件事: await它(仅一次),或将其转换为 Task<T>调用AsTask() .如果你需要做任何复杂的事情,比如 await多次访问它或阻塞它,那么你需要使用 AsTask() .

或者,在这种情况下,只需使用 await在标准 channel 消费模式中:

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

关于c# - 如何避免在 ChannelReader.WaitToReadAsync 时抛出 InvalidOperationException?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/55840824/

32 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com