Hey~~ I think most gardeners should be more interested in the words "high performance". In order to attract attention, the title must be prominent. In fact, the title I prefer is "Monkey Sai Lei, a TCP service written in C#" Various poses
Hey~~ I think most gardeners should be more interested in the word "high performance". In order to attract attention, the title must be prominent. In fact, the title I prefer is "Monkey Sai Lei, written in C# TCP service tricks! 》.
The main purpose of this article is different ways to implement TCP high-performance services using .NET/C#, including but not limited to the following:
Socket support in .NET/C# is based on Windows I/O Completion Ports to complete the port technology encapsulation, and uses different Non-Blocking encapsulation structures to meet different programming needs. The above methods have been fully implemented in Cowboy.Sockets, and the APM and TAP methods have been applied in actual projects. Cowboy.Sockets is still evolving and improving. If there are any problems, please correct them in time.
Although there are so many implementation methods, in the abstract, they are the same, which can be described by two Loops: Accept Loop and Read Loop, as shown below shown. (The "Loop" mentioned here refers to a loop method, not specifically to keywords such as while/for.)
If the Accept loop is blocked, the connection cannot be established quickly and the server Pending Backlog is full, which in turn causes the Client to receive a Connect Timeout exception. If the Read loop is blocked, it will obviously result in the inability to receive data from the client in time, which will cause the client's Send Buffer to be full and no longer able to send data.
From the perspective of implementation details, the location that can cause service blocking may be:
1-2 involves the Accept process and the Connection establishment process, 3-4 involves the ReceiveBuffer processing process, and 5-6 involves the implementation of the application logic side.
The famous Netty network library in Java has made a new attempt at the Buffer part since version 4.0. It adopts a design called ByteBuf to implement Buffer Zero Copy to reduce the performance loss and GC caused by Buffer copy under high concurrency conditions. pressure. Projects such as DotNetty, Orleans, and Helios are trying to implement similar ByteBuf in C#.
The implementation of TcpSocketServer is based on the further encapsulation of TcpListener and TcpClient that comes with the .NET Framework, and is implemented using the APM-based BeginXXX and EndXXX interfaces.
The Accept Loop in TcpSocketServer refers to,
Each successfully established Connection is processed by TcpSocketsession, so TcpSocketSession will contain Read Loop,
TcpSocketServer implements Connection establishment and disconnection and data reception notifications by exposing Events.
<span style="color: #0000ff;">event</span> EventHandler<TcpClientConnectedEventArgs><span style="color: #000000;"> ClientConnected; </span><span style="color: #0000ff;">event</span> EventHandler<TcpClientDisconnectedEventArgs><span style="color: #000000;"> ClientDisconnected; </span><span style="color: #0000ff;">event</span> EventHandler<TcpClientDataReceivedEventArgs> ClientDataReceived;
Using is also simple and straightforward, just subscribe to event notifications directly.
<span style="color: #0000ff;">private</span> <span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span><span style="color: #000000;"> StartServer()
{
_server </span>= <span style="color: #0000ff;">new</span> TcpSocketServer(<span style="color: #800080;">22222</span><span style="color: #000000;">);
_server.ClientConnected </span>+=<span style="color: #000000;"> server_ClientConnected;
_server.ClientDisconnected </span>+=<span style="color: #000000;"> server_ClientDisconnected;
_server.ClientDataReceived </span>+=<span style="color: #000000;"> server_ClientDataReceived;
_server.Listen();
}
</span><span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span> server_ClientConnected(<span style="color: #0000ff;">object</span><span style="color: #000000;"> sender, TcpClientConnectedEventArgs e)
{
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">TCP client {0} has connected {1}.</span><span style="color: #800000;">"</span><span style="color: #000000;">, e.Session.RemoteEndPoint, e.Session));
}
</span><span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span> server_ClientDisconnected(<span style="color: #0000ff;">object</span><span style="color: #000000;"> sender, TcpClientDisconnectedEventArgs e)
{
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">TCP client {0} has disconnected.</span><span style="color: #800000;">"</span><span style="color: #000000;">, e.Session));
}
</span><span style="color: #0000ff;">static</span> <span style="color: #0000ff;">void</span> server_ClientDataReceived(<span style="color: #0000ff;">object</span><span style="color: #000000;"> sender, TcpClientDataReceivedEventArgs e)
{
</span><span style="color: #0000ff;">var</span> text =<span style="color: #000000;"> Encoding.UTF8.GetString(e.Data, e.DataOffset, e.DataLength);
Console.Write(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">Client : {0} {1} --> </span><span style="color: #800000;">"</span><span style="color: #000000;">, e.Session.RemoteEndPoint, e.Session));
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">{0}</span><span style="color: #800000;">"</span><span style="color: #000000;">, text));
_server.Broadcast(Encoding.UTF8.GetBytes(text));
}</span>The implementation of AsyncTcpSocketServer is based on a further encapsulation of TcpListener and TcpClient that comes with the .NET Framework, and is implemented using the XXXAsync interface of async/await based on TAP.
However, in fact, XXXAsync does not create any magical effects. Its internal implementation just converts APM methods into TAP calling methods.
<span style="color: #008000;">//</span><span style="color: #008000;">************* Task-based async public methods *************************</span>
[HostProtection(ExternalThreading = <span style="color: #0000ff;">true</span><span style="color: #000000;">)]
</span><span style="color: #0000ff;">public</span> Task<Socket><span style="color: #000000;"> AcceptSocketAsync()
{
</span><span style="color: #0000ff;">return</span> Task<Socket>.Factory.FromAsync(BeginAcceptSocket, EndAcceptSocket, <span style="color: #0000ff;">null</span><span style="color: #000000;">);
}
[HostProtection(ExternalThreading </span>= <span style="color: #0000ff;">true</span><span style="color: #000000;">)]
</span><span style="color: #0000ff;">public</span> Task<TcpClient><span style="color: #000000;"> AcceptTcpClientAsync()
{
</span><span style="color: #0000ff;">return</span> Task<TcpClient>.Factory.FromAsync(BeginAcceptTcpClient, EndAcceptTcpClient, <span style="color: #0000ff;">null</span><span style="color: #000000;">);
}</span>The Accept Loop in AsyncTcpSocketServer refers to,
<span style="color: #0000ff;">while</span><span style="color: #000000;"> (IsListening)
{
</span><span style="color: #0000ff;">var</span> tcpClient = <span style="color: #0000ff;">await</span><span style="color: #000000;"> _listener.AcceptTcpClientAsync();
}</span>每一个建立成功的 Connection 由 AsyncTcpSocketSession 来处理，所以 AsyncTcpSocketSession 中会包含 Read Loop，
<span style="color: #0000ff;">while</span> (State ==<span style="color: #000000;"> TcpSocketConnectionState.Connected)
{
</span><span style="color: #0000ff;">int</span> receiveCount = <span style="color: #0000ff;">await</span> _stream.ReadAsync(_receiveBuffer, <span style="color: #800080;">0</span><span style="color: #000000;">, _receiveBuffer.Length);
}</span>为了将 async/await 异步到底，AsyncTcpSocketServer 所暴露的接口也同样是 Awaitable 的。
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">interface</span><span style="color: #000000;"> IAsyncTcpSocketServerMessageDispatcher
{
Task OnSessionStarted(AsyncTcpSocketSession session);
Task OnSessionDataReceived(AsyncTcpSocketSession session, </span><span style="color: #0000ff;">byte</span>[] data, <span style="color: #0000ff;">int</span> offset, <span style="color: #0000ff;">int</span><span style="color: #000000;"> count);
Task OnSessionClosed(AsyncTcpSocketSession session);
}</span>使用时仅需将一个实现了该接口的对象注入到 AsyncTcpSocketServer 的构造函数中即可。
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> SimpleMessageDispatcher : IAsyncTcpSocketServerMessageDispatcher
{
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span><span style="color: #000000;"> Task OnSessionStarted(AsyncTcpSocketSession session)
{
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">TCP session {0} has connected {1}.</span><span style="color: #800000;">"</span><span style="color: #000000;">, session.RemoteEndPoint, session));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> Task.CompletedTask;
}
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span> Task OnSessionDataReceived(AsyncTcpSocketSession session, <span style="color: #0000ff;">byte</span>[] data, <span style="color: #0000ff;">int</span> offset, <span style="color: #0000ff;">int</span><span style="color: #000000;"> count)
{
</span><span style="color: #0000ff;">var</span> text =<span style="color: #000000;"> Encoding.UTF8.GetString(data, offset, count);
Console.Write(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">Client : {0} --> </span><span style="color: #800000;">"</span><span style="color: #000000;">, session.RemoteEndPoint));
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">{0}</span><span style="color: #800000;">"</span><span style="color: #000000;">, text));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> session.SendAsync(Encoding.UTF8.GetBytes(text));
}
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span><span style="color: #000000;"> Task OnSessionClosed(AsyncTcpSocketSession session)
{
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">TCP session {0} has disconnected.</span><span style="color: #800000;">"</span><span style="color: #000000;">, session));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> Task.CompletedTask;
}
}</span>当然，对于接口的实现也不是强制了，也可以在构造函数中直接注入方法的实现。
<span style="color: #0000ff;">public</span><span style="color: #000000;"> AsyncTcpSocketServer(
ipEndPoint listenedEndPoint,
Func</span><AsyncTcpSocketSession, <span style="color: #0000ff;">byte</span>[], <span style="color: #0000ff;">int</span>, <span style="color: #0000ff;">int</span>, Task> onSessionDataReceived = <span style="color: #0000ff;">null</span><span style="color: #000000;">,
Func</span><AsyncTcpSocketSession, Task> onSessionStarted = <span style="color: #0000ff;">null</span><span style="color: #000000;">,
Func</span><AsyncTcpSocketSession, Task> onSessionClosed = <span style="color: #0000ff;">null</span><span style="color: #000000;">,
AsyncTcpSocketServerConfiguration configuration </span>= <span style="color: #0000ff;">null</span><span style="color: #000000;">)
{}</span>SAEA 是 SocketAsyncEventArgs 的简写。SocketAsyncEventArgs 是 .NET Framework 3.5 开始支持的一种支持高性能 Socket 通信的实现。SocketAsyncEventArgs 相比于 APM 方式的主要优点可以描述如下：
The main feature of these enhancements is the avoidance of the repeated allocation and synchronization of objects during high-volume asynchronous socket I/O. The Begin/End design pattern currently implemented by the Socket class for asynchronous socket I/O requires a System.IAsyncResult object be allocated for each asynchronous socket Operation.
也就是说，优点就是无需为每次调用都生成 IAsyncResult 等对象，向原生 Socket 更靠近一些。
使用 SocketAsyncEventArgs 的推荐步骤如下：
重点在于池化（Pooling），池化的目的就是为了重用和减少运行时分配和垃圾回收的压力。
TcpSocketSaeaServer 即是对 SocketAsyncEventArgs 的应用和封装，并实现了 Pooling 技术。TcpSocketSaeaServer 中的重点是 SaeaAwaitable 类，SaeaAwaitable 中内置了一个 SocketAsyncEventArgs，并通过 GetAwaiter 返回 SaeaAwaiter 来支持 async/await 操作。同时，通过 SaeaExtensions 扩展方法对来扩展 SocketAsyncEventArgs 的 Awaitable 实现。
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> SaeaAwaitable AcceptAsync(<span style="color: #0000ff;">this</span><span style="color: #000000;"> Socket socket, SaeaAwaitable awaitable) </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> SaeaAwaitable ConnectAsync(<span style="color: #0000ff;">this</span><span style="color: #000000;"> Socket socket, SaeaAwaitable awaitable) </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> SaeaAwaitable DisonnectAsync(<span style="color: #0000ff;">this</span><span style="color: #000000;"> Socket socket, SaeaAwaitable awaitable) </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> SaeaAwaitable ReceiveAsync(<span style="color: #0000ff;">this</span><span style="color: #000000;"> Socket socket, SaeaAwaitable awaitable) </span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">static</span> SaeaAwaitable SendAsync(<span style="color: #0000ff;">this</span> Socket socket, SaeaAwaitable awaitable)
SaeaPool 则是一个 QueuedObjectPool
TcpSocketSaeaServer 中的 Accept Loop 指的就是，
<span style="color: #0000ff;">while</span><span style="color: #000000;"> (IsListening)
{
</span><span style="color: #0000ff;">var</span> saea =<span style="color: #000000;"> _acceptSaeaPool.Take();
</span><span style="color: #0000ff;">var</span> socketError = <span style="color: #0000ff;">await</span><span style="color: #000000;"> _listener.AcceptAsync(saea);
</span><span style="color: #0000ff;">if</span> (socketError ==<span style="color: #000000;"> SocketError.Success)
{
</span><span style="color: #0000ff;">var</span> acceptedSocket =<span style="color: #000000;"> saea.Saea.AcceptSocket;
}
_acceptSaeaPool.Return(saea);
}</span>每一个建立成功的 Connection 由 TcpSocketSaeaSession 来处理，所以 TcpSocketSaeaSession 中会包含 Read Loop，
<span style="color: #0000ff;">var</span> saea =<span style="color: #000000;"> _saeaPool.Take();
saea.Saea.SetBuffer(_receiveBuffer, </span><span style="color: #800080;">0</span><span style="color: #000000;">, _receiveBuffer.Length);
</span><span style="color: #0000ff;">while</span> (State ==<span style="color: #000000;"> TcpSocketConnectionState.Connected)
{
saea.Saea.SetBuffer(</span><span style="color: #800080;">0</span><span style="color: #000000;">, _receiveBuffer.Length);
</span><span style="color: #0000ff;">var</span> socketError = <span style="color: #0000ff;">await</span><span style="color: #000000;"> _socket.ReceiveAsync(saea);
</span><span style="color: #0000ff;">if</span> (socketError !=<span style="color: #000000;"> SocketError.Success)
</span><span style="color: #0000ff;">break</span><span style="color: #000000;">;
</span><span style="color: #0000ff;">var</span> receiveCount =<span style="color: #000000;"> saea.Saea.BytesTransferred;
</span><span style="color: #0000ff;">if</span> (receiveCount == <span style="color: #800080;">0</span><span style="color: #000000;">)
</span><span style="color: #0000ff;">break</span><span style="color: #000000;">;
}</span>同样，TcpSocketSaeaServer 对外所暴露的接口也同样是 Awaitable 的。
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">interface</span><span style="color: #000000;"> ITcpSocketSaeaServerMessageDispatcher
{
Task OnSessionStarted(TcpSocketSaeaSession session);
Task OnSessionDataReceived(TcpSocketSaeaSession session, </span><span style="color: #0000ff;">byte</span>[] data, <span style="color: #0000ff;">int</span> offset, <span style="color: #0000ff;">int</span><span style="color: #000000;"> count);
Task OnSessionClosed(TcpSocketSaeaSession session);
}</span>使用起来也是简单直接：
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> SimpleMessageDispatcher : ITcpSocketSaeaServerMessageDispatcher
{
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span><span style="color: #000000;"> Task OnSessionStarted(TcpSocketSaeaSession session)
{
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">TCP session {0} has connected {1}.</span><span style="color: #800000;">"</span><span style="color: #000000;">, session.RemoteEndPoint, session));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> Task.CompletedTask;
}
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span> Task OnSessionDataReceived(TcpSocketSaeaSession session, <span style="color: #0000ff;">byte</span>[] data, <span style="color: #0000ff;">int</span> offset, <span style="color: #0000ff;">int</span><span style="color: #000000;"> count)
{
</span><span style="color: #0000ff;">var</span> text =<span style="color: #000000;"> Encoding.UTF8.GetString(data, offset, count);
Console.Write(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">Client : {0} --> </span><span style="color: #800000;">"</span><span style="color: #000000;">, session.RemoteEndPoint));
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">{0}</span><span style="color: #800000;">"</span><span style="color: #000000;">, text));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> session.SendAsync(Encoding.UTF8.GetBytes(text));
}
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span><span style="color: #000000;"> Task OnSessionClosed(TcpSocketSaeaSession session)
{
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">TCP session {0} has disconnected.</span><span style="color: #800000;">"</span><span style="color: #000000;">, session));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> Task.CompletedTask;
}
}</span>从 Windows 8.1 / Windows Server 2012 R2 开始，微软推出了 Registered I/O Networking Extensions 来支持高性能 Socket 服务的实现，简称 RIO。
The following functions are supported for Windows Store apps on Windows 8.1, Windows Server 2012 R2, and later. Microsoft Visual Studio 2013 Update 3 or later is required for Windows Store apps.
到目前为止，.NET Framework 还没有推出对 RIO 的支持，所以若想在 C# 中实现 RIO 则只能通过 P/Invoke 方式，RioSharp 是开源项目中的一个比较完整的实现。
Cowboy.Sockets 直接引用了 RioSharp 的源代码，放置在 Cowboy.Sockets.Experimental 名空间下，以供实验和测试使用。
同样，通过 TcpSocketRioServer 来实现 Accept Loop，
_listener.OnAccepted = (acceptedSocket) =><span style="color: #000000;">
{
Task.Run(</span><span style="color: #0000ff;">async</span> () =><span style="color: #000000;">
{
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> Process(acceptedSocket);
})
.Forget();
};</span>通过 TcpSocketRioSession 来处理 Read Loop，
<span style="color: #0000ff;">while</span> (State ==<span style="color: #000000;"> TcpSocketConnectionState.Connected)
{
</span><span style="color: #0000ff;">int</span> receiveCount = <span style="color: #0000ff;">await</span> _stream.ReadAsync(_receiveBuffer, <span style="color: #800080;">0</span><span style="color: #000000;">, _receiveBuffer.Length);
</span><span style="color: #0000ff;">if</span> (receiveCount == <span style="color: #800080;">0</span><span style="color: #000000;">)
</span><span style="color: #0000ff;">break</span><span style="color: #000000;">;
}</span>测试代码一如既往的类似：
<span style="color: #0000ff;">public</span> <span style="color: #0000ff;">class</span><span style="color: #000000;"> SimpleMessageDispatcher : ITcpSocketRioServerMessageDispatcher
{
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span><span style="color: #000000;"> Task OnSessionStarted(TcpSocketRioSession session)
{
</span><span style="color: #008000;">//</span><span style="color: #008000;">Console.WriteLine(string.Format("TCP session {0} has connected {1}.", session.RemoteEndPoint, session));</span>
Console.WriteLine(<span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">TCP session has connected {0}.</span><span style="color: #800000;">"</span><span style="color: #000000;">, session));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> Task.CompletedTask;
}
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span> Task OnSessionDataReceived(TcpSocketRioSession session, <span style="color: #0000ff;">byte</span>[] data, <span style="color: #0000ff;">int</span> offset, <span style="color: #0000ff;">int</span><span style="color: #000000;"> count)
{
</span><span style="color: #0000ff;">var</span> text =<span style="color: #000000;"> Encoding.UTF8.GetString(data, offset, count);
</span><span style="color: #008000;">//</span><span style="color: #008000;">Console.Write(string.Format("Client : {0} --> ", session.RemoteEndPoint));</span>
Console.Write(<span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">Client : --> </span><span style="color: #800000;">"</span><span style="color: #000000;">));
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">{0}</span><span style="color: #800000;">"</span><span style="color: #000000;">, text));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> session.SendAsync(Encoding.UTF8.GetBytes(text));
}
</span><span style="color: #0000ff;">public</span> <span style="color: #0000ff;">async</span><span style="color: #000000;"> Task OnSessionClosed(TcpSocketRioSession session)
{
Console.WriteLine(</span><span style="color: #0000ff;">string</span>.Format(<span style="color: #800000;">"</span><span style="color: #800000;">TCP session {0} has disconnected.</span><span style="color: #800000;">"</span><span style="color: #000000;">, session));
</span><span style="color: #0000ff;">await</span><span style="color: #000000;"> Task.CompletedTask;
}
}</span>本篇文章《C#高性能TCP服务的多种实现方式》由 Dennis Gao 发表自博客园个人博客，未经作者本人同意禁止以任何的形式转载，任何自动的或人为的爬虫转载行为均为耍流氓。
![[Web front-end] Node.js quick start](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
