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

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了如何避免在ChannelReader.WaitToReadAsync时抛出InvalidOperationException?相关的知识,希望对你有一定的参考价值。

我使用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:srcgitreposdotnet-sandboxchannelstestAsyncChannelsTest.cs:line 26

如果捕获并忽略了异常,则代码正常工作。但我想摆脱其原因尚不清楚的错误。

这是我的环境和最少的代码。

  • TargetFramework = netcoreapp2.1
  • System.Threading.Channels version = 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)。结果

你只能用ValueTask<T>做两件事:await它(只有一次),或者通过调用Task<T>将它转换为AsTask()。如果你需要做任何复杂的事情,比如awaiting它不止一次或阻塞它,那么你需要使用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();
  }
  ...
}

以上是关于如何避免在ChannelReader.WaitToReadAsync时抛出InvalidOperationException?的主要内容,如果未能解决你的问题,请参考以下文章

VBA如何在使用许多整数时避免递归

如何避免 Touches 取消事件?

在“点击,点击”事件触发两次。如何避免?

如何避免在散点图中重叠文本?

如何避免在Reactjs中每次都调用API?

如何避免codeigniter中的sql注入