【发布时间】：2020-07-14 07:42:39
【问题描述】：
我最近在一个项目中使用了 SocketAsyncEventArgs，我遇到了一些问题，即 ReceiveAsync 有时会以与通过 SendAsync 发送的顺序不同的顺序获取数据。在 SendAsync 方法中发送的每个数据块都会被维护，但这些块的顺序不一定正确。也许我对 SendAsync 方法的理解不正确，但我认为特别是使用 SocketType.Stream 和 ProtocolType.Tcp 可以确保保持顺序。我知道底层过程将不可避免地破坏消息，并且 ReceiveAsync 通常会读取少于缓冲区分配的内容。但我假设发送和接收流会保持顺序。
我开发了一个显示问题的测试控制台程序。它每次尝试使用一组不同的套接字和端口运行大约 20 次。在我的笔记本电脑上，它通常会通过一次然后第二次失败；通常在期待第二个块时会收到一个稍后的块。从其他测试中，我知道预期的块最终确实会出现，只是不按顺序。
需要注意的是，我能够在 Windows 2008 远程服务器上对其进行测试并且没有任何问题。但是，它从未接近在我的笔记本电脑上完成。事实上，如果我让调试执行在异常中断中挂起一段时间，我已经让它不止一次地完全冻结了我的笔记本电脑，并且不得不进行硬重启。这是我在 Windows 7 上使用 VS2017 运行的工作笔记本电脑。我不确定这是否是一个因素，但它正在运行 Symantec Endpoint Protection，尽管我在日志中没有发现任何内容。
所以我的问题是，我对 SocketAsyncEventArgs 的运作方式有错误的看法吗？还是我的代码是一场灾难（也许两者兼而有之）？它是我的笔记本电脑独有的吗？ （最后一个让我觉得我正在为编程新手做准备，并且您认为编译器一定有问题。）
using System;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading.Tasks;
static class DumTest
{
static void Main(string[] args)
{
for (int i = 9177; i < 9199; i++)
{
RunDum(i);
//Thread.Sleep(350);
}
Console.WriteLine("all done.");
Console.ReadLine();
}
static void RunDum(int port)
{
var dr = new DumReceiver(port);
var ds = new DumSender(port);
dr.Acception.Wait();
ds.Connection.Wait();
dr.Completion.Wait();
ds.Completion.Wait();
Console.WriteLine($"Completed {port}. " +
$"sent: {ds.SegmentsSent} segments, received: {dr.SegmentsRead} segments");
}
}
class DumReceiver
{
private readonly SocketAsyncEventArgs eva = new SocketAsyncEventArgs();
private readonly TaskCompletionSource<object> tcsAcc = new TaskCompletionSource<object>();
private TaskCompletionSource<object> tcsRcv;
private Socket socket;
internal DumReceiver(int port)
{
this.eva.Completed += this.Received;
var lstSock = new Socket(
AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
var localIP = Dns.GetHostEntry(Dns.GetHostName()).AddressList
.First(i => i.AddressFamily == AddressFamily.InterNetwork);
lstSock.Bind(new IPEndPoint(localIP, port));
lstSock.Listen(1);
var saea = new SocketAsyncEventArgs();
saea.Completed += this.AcceptCompleted;
lstSock.AcceptAsync(saea);
}
internal Task Acception => this.tcsAcc.Task;
internal Task Completion { get; private set; }
internal int SegmentsRead { get; private set; }
private void AcceptCompleted(object sender, SocketAsyncEventArgs e)
{
if (e.SocketError == SocketError.Success)
{
this.socket = e.AcceptSocket;
e.Dispose();
try
{
this.Completion = this.ReceiveLupeAsync();
}
finally
{
this.tcsAcc.SetResult(null);
}
}
else
{
this.tcsAcc.SetException(new SocketException((int)e.SocketError));
}
}
private async Task ReceiveLupeAsync()
{
var buf = new byte[8196];
byte bufSeg = 1;
int pos = 0;
while (true)
{
this.tcsRcv = new TaskCompletionSource<object>();
this.eva.SetBuffer(buf, pos, 8196 - pos);
if (this.socket.ReceiveAsync(this.eva))
{
await this.tcsRcv.Task.ConfigureAwait(false);
}
if (this.eva.SocketError != SocketError.Success)
{
throw new SocketException((int)eva.SocketError);
}
if (this.eva.BytesTransferred == 0)
{
if (pos != 0)
{
throw new EndOfStreamException();
}
break;
}
pos += this.eva.BytesTransferred;
if (pos == 8196)
{
pos = 0;
for (int i = 0; i < 8196; i++)
{
if (buf[i] != bufSeg)
{
var msg = $"Expected {bufSeg} but read {buf[i]} ({i} of 8196). " +
$"Last read: {this.eva.BytesTransferred}.";
Console.WriteLine(msg);
throw new Exception(msg);
}
}
this.SegmentsRead++;
bufSeg = (byte)(this.SegmentsRead + 1);
}
}
}
private void Received(object s, SocketAsyncEventArgs e) => this.tcsRcv.SetResult(null);
}
class DumSender
{
private readonly SocketAsyncEventArgs eva = new SocketAsyncEventArgs();
private readonly Socket socket = new Socket(
AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
private readonly TaskCompletionSource<object> tcsCon = new TaskCompletionSource<object>();
private TaskCompletionSource<object> tcsSnd;
internal DumSender(int port)
{
this.eva.Completed += this.Sent;
var saea = new SocketAsyncEventArgs();
var localIP = Dns.GetHostEntry(Dns.GetHostName()).AddressList
.First(i => i.AddressFamily == AddressFamily.InterNetwork);
saea.RemoteEndPoint = new IPEndPoint(localIP, port);
saea.Completed += this.ConnectionCompleted;
this.socket.ConnectAsync(saea);
}
internal Task Connection => this.tcsCon.Task;
internal Task Completion { get; private set; }
internal int SegmentsSent { get; private set; }
private void ConnectionCompleted(object sender, SocketAsyncEventArgs e)
{
if (e.SocketError == SocketError.Success)
{
e.Dispose();
try
{
this.Completion = this.SendLupeAsync();
}
finally
{
this.tcsCon.SetResult(null);
}
}
else
{
this.tcsCon.SetException(new SocketException((int)e.SocketError));
}
}
private async Task SendLupeAsync()
{
var buf = new byte[8196];
byte bufSeg = 1;
while (true)
{
for (int i = 0; i < 8196; i++)
{
buf[i] = bufSeg;
}
this.tcsSnd = new TaskCompletionSource<object>();
this.eva.SetBuffer(buf, 0, 8196);
if (this.socket.SendAsync(this.eva))
{
await this.tcsSnd.Task.ConfigureAwait(false);
}
if (this.eva.SocketError != SocketError.Success)
{
throw new SocketException((int)this.eva.SocketError);
}
if (this.eva.BytesTransferred != 8196)
{
throw new SocketException();
}
if (++this.SegmentsSent == 299)
{
break;
}
bufSeg = (byte)(this.SegmentsSent + 1);
}
this.socket.Shutdown(SocketShutdown.Both);
}
private void Sent(object s, SocketAsyncEventArgs e) => this.tcsSnd.SetResult(null);
}
【问题讨论】：
-
我无法使用您发布的代码重现您的问题（Windows 10 Pro）。代码对我来说看起来不错。除非我需要最大的可扩展性，否则我不会为 Socket 中的
XXXAsync()方法而烦恼；使用NetworkStream和传统的await友好的XXXAsync()方法，代码更容易读写。但是我在代码中没有看到任何实际的 bug，虽然任何可用于套接字的异步机制都存在潜在的缓冲区重新排序问题，但只有当你有两个或更多读取操作同时排队，而您没有。 -
FWIW，我增加了您的测试代码以运行 22 端口测试的 20 次迭代，使用 2990 段而不是 299 段，但仍然无法使其失败。坦率地说，如果您有任何类型的 AV 软件，那么当您开始看到 I/O 代码出现意外行为时，您应该首先禁用它。这类软件中的错误太常见了，在向更大的社区寻求帮助之前，测试这种可能性太容易了，没有任何借口不这样做。
-
感谢您抽出宝贵时间对其进行测试，如果只是浪费时间，我们深表歉意。我天真地认为，因为连接是与本地主机建立的，所以它会超出 AV 的范围。我曾尝试将其关闭并以相同的结果对其进行测试。这种公司设置并没有使它非常直观或容易做到，因此它可能仍在运行。明天我会有更多的选择来测试它。感谢您的健全性检查。
-
当然，没问题。需要明确的是：只要您一次只执行一个读取操作，您将始终以正确的顺序获取数据。它是 TCP 的一个基本保证（实际上，我提到的重新排序问题，并不是因为数据最终乱序......缓冲区仍然按照它们呈现给网络层的顺序填充，它是只是处理 I/O 完成的线程有时会出现乱序）。因此，如果您确实看到数据乱序，那么某些东西肯定会妨碍并破坏事物。
标签： c# sockets socketasynceventargs