blocks|key|261229|text|取消是协作的。NetworkStream.ReadAsync必须配合才能被取消。它很难做到这一点，因为这可能会使流处于未定义的状态。哪些字节已从Windows+TCP堆栈读取，哪些尚未读取？IO是不容易取消的。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|261230|Reflector显示NetworkStream不会覆盖ReadAsync。这意味着它将获得Stream.ReadAsync的默认行为，即只丢弃令牌。没有通用的方法可以取消流操作，因此BCL+Stream类甚至不尝试(它不能尝试-没有办法做到这一点)。|261231|您应该在Socket上设置超时。|261232|entityMap^0|7|N|0|B|D|S|9|1A|G|2O|6|0|4|6|0^^$0|@$1|2|3|4|5|6|7|L|8|@$9|M|A|N|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|O|8|@$9|P|A|Q|B|C]|$9|R|A|S|B|C]|$9|T|A|U|B|C]|$9|V|A|W|B|C]]|D|@]|E|$]]|$1|H|3|I|5|6|7|X|8|@$9|Y|A|Z|B|C]]|D|@]|E|$]]|$1|J|3|-4|5|6|7|10|8|@]|D|@]|E|$]]]|K|$]]

Cancellation is cooperative. <code>NetworkStream.ReadAsync</code> must cooperate to be able to be cancelled. It is kind of hard for it to do that because that would potentially leave the stream in an undefined state. What bytes have already been read from the Windows TCP stack and what haven't? IO is not easily cancellable.

Reflector shows that <code>NetworkStream</code> does not override <code>ReadAsync</code>. This means that it will get the default behavior of <code>Stream.ReadAsync</code> which just throws the token away. There is no generic way Stream operations can be cancelled so the BCL <code>Stream</code> class does not even try (it cannot try - there is no way to do this).

You should set a timeout on the <code>Socket</code>.

blocks|key|478092|text|我终于找到了一个变通办法。使用Task.WaitAny将异步呼叫与延迟任务(Task.Delay)相结合。当io任务之前经过延迟时，关闭流。这将强制任务停止。您应该正确处理io任务上的异步异常。并且您应该为延迟任务和io任务添加一个延续任务。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|478093|它也适用于tcp连接。关闭另一个线程中的连接(您可以将其视为延迟任务线程)会强制所有使用/等待此连接的异步任务停止。|478094|-编辑--|478095|@vtortola建议的另一个更干净的解决方案是:使用取消令牌注册对stream.Close的调用：|478096|async+ValueTask+Read(NetworkStream+stream,+TimeSpan+timeout+=+default)
{
++++if(timeout+==+default(TimeSpan))
++++++timeout+=+TimeSpan.FromSeconds(5);

++++using+var+cts+=+new+CancellationTokenSource(timeout);+//C#+8+syntax
++++using(cts.Token.Register(()+=>+stream.Close()))
++++{
+++++++int+receivedCount;
+++++++try
+++++++{
+++++++++++var+buffer+=+new+byte[30000];
+++++++++++receivedCount+=+await+stream.ReadAsync(buffer,+0,+30000,+tcs.Token).ConfigureAwait(false);
+++++++}
+++++++catch+(TimeoutException)
+++++++{
+++++++++++receivedCount+=+-1;
+++++++}
++++}
}|code-block|syntax|javascript|478097|entityMap^0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|O|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|P|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|Q|8|@]|9|@]|A|$]]|$1|F|3|G|5|6|7|R|8|@]|9|@]|A|$]]|$1|H|3|I|5|J|7|S|8|@]|9|@]|A|$K|L]]|$1|M|3|-4|5|6|7|T|8|@]|9|@]|A|$]]]|N|$]]

I finally found a workaround. Combine the async call with a delay task (Task.Delay) using Task.WaitAny. When the delay elapses before the io task, close the stream. This will force the task to stop. You should handle the async exception on the io task correctly. And you should add a continuation task for both the delayed task and the io task.

It also work with tcp connections. Closing the connection in another thread (you could consider it is the delay task thread) forces all async tasks using/waiting for this connection to stop.

--EDIT--

Another cleaner solution suggested by @vtortola: use the cancellation token to register a call to stream.Close:

<pre><code>async ValueTask Read(NetworkStream stream, TimeSpan timeout = default)
{
 if(timeout == default(TimeSpan))
 timeout = TimeSpan.FromSeconds(5);

 using var cts = new CancellationTokenSource(timeout); //C# 8 syntax
 using(cts.Token.Register(() =&gt; stream.Close()))
 {
 int receivedCount;
 try
 {
 var buffer = new byte[30000];
 receivedCount = await stream.ReadAsync(buffer, 0, 30000, tcs.Token).ConfigureAwait(false);
 }
 catch (TimeoutException)
 {
 receivedCount = -1;
 }
 }
}
</code></pre>

blocks|key|478218|text|提供了关于三种不同方法的更多上下文。我的服务监视其他web应用程序的可用性。因此，它需要建立到各种网站的大量连接。它们中的一些崩溃/返回错误/变得无响应。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|478219|轴Y-挂起的测试(会话)数。由于部署/重新启动而降至0。|478220|I.+(1月25日)在修改服务后，最初的实现使用了带有取消令牌的ReadAsync。这导致了许多测试挂起(对这些网站运行请求表明服务器有时确实不返回内容)。|478221|II.+(2月17日)部署了一个更改，以保护与Task.Delay的取消。这完全解决了这个问题。|478222|private+async+Task<int>+StreamReadWithCancellationTokenAsync(Stream+stream,+byte[]+buffer,+int+count,+Task+cancellationDelayTask)
{
++++if+(cancellationDelayTask.IsCanceled)
++++{
++++++++throw+new+TaskCanceledException();
++++}

++++//+Stream.ReadAsync+doesn't+honor+cancellation+token.+It+only+checks+it+at+the+beginning.+The+actual
++++//+operation+is+not+guarded.+As+a+result+if+remote+server+never+responds+and+connection+never+closed
++++//+it+will+lead+to+this+operation+hanging+forever.
++++Task<int>+readBytesTask+=+stream.ReadAsync(
++++++++buffer,
++++++++0,
++++++++count);
++++await+Task.WhenAny(readBytesTask,+cancellationDelayTask).ConfigureAwait(false);

++++//+Check+whether+cancellation+task+is+cancelled+(or+completed).
++++if+(cancellationDelayTask.IsCanceled+%7C%7C+cancellationDelayTask.IsCompleted)
++++{
++++++++throw+new+TaskCanceledException();
++++}

++++//+Means+that+main+task+completed.+We+use+Result+directly.
++++//+If+the+main+task+failed+the+following+line+will+throw+an+exception+and
++++//+we'll+catch+it+above.
++++int+readBytes+=+readBytesTask.Result;

++++return+readBytes;
}|code-block|syntax|javascript|478223|III+(3月3日)之后，StackOverflow实现了基于超时的流关闭：|478224|using+(timeoutToken.Register(()+=>+stream.Close()))
{
++++//+Stream.ReadAsync+doesn't+honor+cancellation+token.+It+only+checks+it+at+the+beginning.+The+actual
++++//+operation+is+not+guarded.+As+a+result+if+a+remote+server+never+responds+and+connection+never+closed
++++//+it+will+lead+to+this+operation+hanging+forever.
++++//+ReSharper+disable+once+MethodSupportsCancellation
++++readBytes+=+await+targetStream.ReadAsync(
++++++++buffer,
++++++++0,
++++++++Math.Min(responseBodyLimitInBytes+-+totalReadBytes,+buffer.Length)).ConfigureAwait(false);
}|478225|这种实现带来了挂起(与最初的方法不同)：|478226|​|478227|📷|atomic|offset|length|478228|478229|已恢复到Task.Delay解决方案。|478230|entityMap|0|IMAGE|mutability|IMMUTABLE|imageUrl|https://ask.qcloudimg.com/http-save/yehe-900000/f0a7107d7d82b0880ead5e3f0f2e86d7.png|imageAlt^0|0|0|0|0|0|0|0|0|0|0|1|0|0|0|0^^$0|@$1|2|3|4|5|6|7|1B|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|1C|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|1D|8|@]|9|@]|A|$]]|$1|F|3|G|5|6|7|1E|8|@]|9|@]|A|$]]|$1|H|3|I|5|J|7|1F|8|@]|9|@]|A|$K|L]]|$1|M|3|N|5|6|7|1G|8|@]|9|@]|A|$]]|$1|O|3|P|5|J|7|1H|8|@]|9|@]|A|$K|L]]|$1|Q|3|R|5|6|7|1I|8|@]|9|@]|A|$]]|$1|S|3|T|5|6|7|1J|8|@]|9|@]|A|$]]|$1|U|3|V|5|W|7|1K|8|@]|9|@$X|1L|Y|1M|1|1N]]|A|$]]|$1|Z|3|T|5|6|7|1O|8|@]|9|@]|A|$]]|$1|10|3|11|5|6|7|1P|8|@]|9|@]|A|$]]|$1|12|3|-4|5|6|7|1Q|8|@]|9|@]|A|$]]]|13|$14|$5|15|16|17|A|$18|19|1A|-4]]]]

Providing more context on three different approaches. My service monitors other web applications availability. So, it needs to establish lots of connections to various web sites. Some of them crash/return errors/become unresponsive.

Axis Y - number of hung tests (sessions). Drops to 0 caused by deployments/restarts.

I. (Jan 25th) After revamping a service, the initial implementation used ReadAsync with a cancellation token. This resulted in lots of tests hanging (running requests against those web sites showed that servers indeed sometimes didn't return content).

II. (Feb 17th) Deployed a change which guarded cancellation with Task.Delay. This completely fixed this issue.

<pre><code>private async Task&lt;int&gt; StreamReadWithCancellationTokenAsync(Stream stream, byte[] buffer, int count, Task cancellationDelayTask)
{
 if (cancellationDelayTask.IsCanceled)
 {
 throw new TaskCanceledException();
 }

 // Stream.ReadAsync doesn't honor cancellation token. It only checks it at the beginning. The actual
 // operation is not guarded. As a result if remote server never responds and connection never closed
 // it will lead to this operation hanging forever.
 Task&lt;int&gt; readBytesTask = stream.ReadAsync(
 buffer,
 0,
 count);
 await Task.WhenAny(readBytesTask, cancellationDelayTask).ConfigureAwait(false);

 // Check whether cancellation task is cancelled (or completed).
 if (cancellationDelayTask.IsCanceled || cancellationDelayTask.IsCompleted)
 {
 throw new TaskCanceledException();
 }

 // Means that main task completed. We use Result directly.
 // If the main task failed the following line will throw an exception and
 // we'll catch it above.
 int readBytes = readBytesTask.Result;

 return readBytes;
}
</code></pre>

III (March 3rd) Following this StackOverflow implemented closing a stream based on timeout:

<pre><code>using (timeoutToken.Register(() =&gt; stream.Close()))
{
 // Stream.ReadAsync doesn't honor cancellation token. It only checks it at the beginning. The actual
 // operation is not guarded. As a result if a remote server never responds and connection never closed
 // it will lead to this operation hanging forever.
 // ReSharper disable once MethodSupportsCancellation
 readBytes = await targetStream.ReadAsync(
 buffer,
 0,
 Math.Min(responseBodyLimitInBytes - totalReadBytes, buffer.Length)).ConfigureAwait(false);
}
</code></pre>

This implementation brought hangs back (not to the same extent as the initial approach):

<a href="https://i.stack.imgur.com/AqrWp.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/AqrWp.png" alt="enter image description here"></a>

Reverted back to Task.Delay solution.

blocks|key|261575|text|作为提醒，await+_stream.WriteAsync(message,cancellationToken);+(_stream是一个SslStream)在执行BeginEndWriteAsync之前会在幕后检查取消令牌是否已被取消，因此您必须在令牌开始写入之前取消令牌。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|261576|public+virtual+Task+WriteAsync(byte[]+buffer,+int+offset,+int+count,+CancellationToken+cancellationToken)
++++{
++++++++//+If+cancellation+was+requested,+bail+early+with+an+already+completed+task.
++++++++//+Otherwise,+return+a+task+that+represents+the+Begin/End+methods.
++++++++return+cancellationToken.IsCancellationRequested
++++++++++++++++++++?+Task.FromCanceled(cancellationToken)
++++++++++++++++++++:+BeginEndWriteAsync(buffer,+offset,+count);
++++}|code-block|syntax|javascript|261577|entityMap^0|5|1G|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@$9|N|A|O|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|P|8|@]|D|@]|E|$I|J]]|$1|K|3|-4|5|6|7|Q|8|@]|D|@]|E|$]]]|L|$]]

Just for the heads-up, <code>await _stream.WriteAsync(message,cancellationToken);</code> (_stream is a SslStream) checks behind the scenes if the cancellation token has been cancelled before performing the BeginEndWriteAsync, so you must cancel your token before it starts to write.
<pre><code>public virtual Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
 {
 // If cancellation was requested, bail early with an already completed task.
 // Otherwise, return a task that represents the Begin/End methods.
 return cancellationToken.IsCancellationRequested
 ? Task.FromCanceled(cancellationToken)
 : BeginEndWriteAsync(buffer, offset, count);
 }
</code></pre>

Here the proof. 
Any idea what is wrong in this code ?

<pre><code> [TestMethod]
 public void TestTest()
 {
 var tcp = new TcpClient() { ReceiveTimeout = 5000, SendTimeout = 20000 };
 tcp.Connect(IPAddress.Parse("176.31.100.115"), 25);
 bool ok = Read(tcp.GetStream()).Wait(30000);
 Assert.IsTrue(ok);
 }

 async Task Read(NetworkStream stream)
 {
 using (var cancellationTokenSource = new CancellationTokenSource(5000))
 {
 int receivedCount;
 try
 {
 var buffer = new byte[1000];
 receivedCount = await stream.ReadAsync(buffer, 0, 1000, cancellationTokenSource.Token);
 }
 catch (TimeoutException e)
 {
 receivedCount = -1;
 }
 }
 }
</code></pre>

NetworkStream.ReadAsync with a cancellation token never cancels

服务器

Windows

这就是证据。你知道这段代码出了什么问题吗？    [TestMethod]    public void TestTest()    {        var tcp = new TcpClient() { ReceiveTimeout = 5000, SendTimeout = 20000 };        tcp...

问带有取消令牌的NetworkStream.ReadAsync从不取消
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问带有取消令牌的NetworkStream.ReadAsync从不取消EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问带有取消令牌的NetworkStream.ReadAsync从不取消
EN