确保调用 Socket.XXXAsync 的线程保持活动状态以完成 IO 请求（IO 完成端口，C#）

Question

我正在实现一个小型库，以便更轻松地使用 System.Net.Sockets.Socket。它应该处理任意数量的侦听和接收 TCP 套接字，并且实现速度快非常重要。

我使用 XXXAsync 方法和 CLR ThreadPool 回调库用户的委托（例如，每当成功发送消息或接收到某些数据时） 。最好该库不会自行启动任何线程。

该库的用户可以访问我的 Sockets 包装器的接口来发送消息或开始接收消息（以及许多其他方法和重载）：

public interface IClientSocket {
    // will eventually call Socket.SendAsync
    IMessageHandle SendAsync(byte[] buffer, int offset, int length);

    // will eventually call Socket.RecieveAsync
    void StartReceiveAsync();
}

XXXAsync 方法使用 IO Completion港口。因此，调用这些方法的线程必须保持活动状态，直到操作完成，否则操作将失败并显示 SocketError.OperationAborted（我认为是这种情况，或者不是？）。 对库的用户施加这样的限制是丑陋的并且容易出错。

这里最好的选择是什么？

• 使用委托调用ThreadPool.QueueUserWorkItem来调用XXXAsync方法？这样安全吗？我在某处读到，ThreadPool 不会停止具有任何 IOCP 的空闲线程。那就太好了，因为它解决了上述问题。
  但对于许多 TCP 连接也可能会很糟糕。 在这种情况下，每个 ThreadPool 线程很可能都会调用一个待处理的 ReceiveAsync 调用。因此，即使当前工作负载较低并且许多线程处于空闲状态（并且浪费内存），ThreadPool 也永远不会收缩。

• 启动一个始终处于活动状态的专用线程并调用XXXAsync方法。例如，当库用户想要发送数据时，它将委托放入同步队列中，线程将其弹出并调用 SendAsync 方法。
  我不太喜欢这个解决方案，因为它浪费了一个线程，并且在多核机器上，发送仅由一个线程执行。

此外，这两种解决方案都不是最好的，因为它们将调用异步方法的作业传递给另一个线程。这可以避免吗？

你怎么认为？ （谢谢！！）

Thomas

编辑1：

我可以使用以下测试程序重现SocketError.OperationAborted 问题（我认为它是正确的）。
编译、启动并telnet到端口127.0.0.1:10000。发送“t”或“T”并等待> 3秒。发送“T”时，ReceiveAsync 调用在 ThreadPool 中完成（有效），发送“t”时将启动一个新线程，并在 3 秒后终止（失败）。

using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Collections.Generic;

namespace Test {
    class Program {
        static List<Socket> _referenceToSockets = new List<Socket>();
        static void Main(string[] args) {
            Thread.CurrentThread.Name = "Main Thread";

            // Create a listening socket on Port 10000.
            Socket ServerSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            _referenceToSockets.Add(ServerSocket);
            var endPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 10000);
            ServerSocket.Bind(endPoint);
            ServerSocket.Listen(50);

            // Start listening.
            var saeaAccept = new SocketAsyncEventArgs();
            saeaAccept.Completed += OnCompleted;
            ServerSocket.AcceptAsync(saeaAccept);

            Console.WriteLine(String.Format("Listening on {0}.", endPoint));
            Console.ReadLine();
        }

        private static void OnCompleted(object obj, SocketAsyncEventArgs evt) {
            var socket = (Socket)obj;
            Console.WriteLine(String.Format("Async operation completed: {0}; Error: {1}; Callback-Thread: \"{2}\" ({3} threadpool)", evt.LastOperation, evt.SocketError, Thread.CurrentThread.Name, Thread.CurrentThread.IsThreadPoolThread?"is":"no"));
            switch (evt.LastOperation) {
                case SocketAsyncOperation.Accept:
                    // Client connected. Listen for more.
                    Socket clientSocket = evt.AcceptSocket;
                    _referenceToSockets.Add(clientSocket);
                    evt.AcceptSocket = null;
                    socket.AcceptAsync(evt);

                    // Start receiving data.
                    var saeaReceive = new SocketAsyncEventArgs();
                    saeaReceive.Completed += OnCompleted;
                    saeaReceive.SetBuffer(new byte[1024], 0, 1024);
                    clientSocket.ReceiveAsync(saeaReceive);
                    break;
                case SocketAsyncOperation.Disconnect:
                    socket.Close();
                    evt.Dispose();
                    break;
                case SocketAsyncOperation.Receive:
                    if (evt.SocketError != SocketError.Success) {
                        socket.DisconnectAsync(evt);
                        return;
                    }
                    var asText = Encoding.ASCII.GetString(evt.Buffer, evt.Offset, evt.BytesTransferred);
                    Console.WriteLine(String.Format("Received: {0} bytes: \"{1}\"", evt.BytesTransferred, asText));
                    if (evt.BytesTransferred == 0) {
                        socket.Close();
                        evt.Dispose();
                    }
                    if (asText.ToUpper().StartsWith("T")) {
                        Action<object> action = (object o) => {
                            socket.ReceiveAsync(evt);
                            Console.WriteLine(String.Format("Called ReceiveAsync {0}...", o));
                            Thread.Sleep(3000);
                            Console.WriteLine("End of Action...");
                        };
                        if (asText.StartsWith("T")) {
                            ThreadPool.QueueUserWorkItem(o=>action(o), "in ThreadPool");
                        } else {
                            new Thread(o=>action(o)).Start("in a new Thread");
                        }
                    } else {
                        socket.ReceiveAsync(evt);
                    }
                    break;
            }
        }
    }
}

编辑 #3

以下是我打算使用的解决方案：

我让库用户的线程直接调用 XXXAsync（SendAsync 除外）操作。在大多数情况下，调用将会成功（因为调用线程很少终止）。
如果操作因 SocketError.OperationAborted 失败，则库仅使用异步回调中的当前线程再次调用该操作。这是一个 ThreadPool 线程，它很有可能成功（如果出现 SocketError.OperationAborted 的原因，将设置一个附加标志，最多使用一次此解决方法）是由于其他一些错误造成的）。 这应该可以工作，因为套接字本身仍然没问题，只是之前的操作失败了。

对于 SendAsync，此解决方法不起作用，因为它可能会打乱消息的顺序。在本例中，我会将消息放入 FIFO 列表中进行排队。我将使用 ThreadPool 将它们出列并通过 SendAsync 发送它们。

Answer 1

这里有一些危险信号。如果你在保持线程活动时遇到困难，那么你的线程就没有做任何有用的事情。在这种情况下，使用异步方法没有意义。如果速度是您唯一关心的问题，那么您不应该使用异步方法。它们要求 Socket 抓住一个 tp 线程来进行回调。这是很小的开销，但如果常规线程进行阻塞调用，则没有任何开销。

仅当您需要使应用程序在处理许多并发连接时能够很好地扩展时，才应考虑异步方法。这会产生非常突发的 CPU 负载，非常适合 tp 线程。

Answer 2

调用这些方法的线程必须保持活动状态，直到操作完成”

我不相信这个说法的真实性。当您启动网络操作时，它会被移交给 Windows IO 系统。当发生网络事件（即数据已收到并且您的应用程序已收到通知），这可以很高兴地由另一个线程处理（即使启动线程早已死亡）

您确定不会删除对重要内容的引用并允许它被删除吗？垃圾收集了吗？

如果如果您处于允许线程死亡的情况，那么我怀疑您的代码还没有准备好扩展到超高速/一流的客户端，如果您真的希望扩展，那么您对线程的使用应该是。非常严格地控制让一个线程死亡的成本（我假设另一个线程需要旋转来取代它）不是你应该允许发生太多的事情，线程池绝对是这里的方法。