关于 .NET 运行时 Sockets Datagram Socket，大量 Socket 缓冲区内存问题

本文解决的问题如上所示，Datagram（数据报）类型的 Socket，这包含 “ICMP、UDP”、“RAW IP”等，那么为什么会发生这样的问题？

我们在阻塞模式下是不存在这个问题，但本文描述的问题仅仅包含在以下几个 Socket 异步接口实现上面。

以下为发生故障的 .NET API：

1、System::Net::Sockets::Socket::BeginSendTo

2、System::Net::Sockets::Socket::BeginReceiveFrom

3、System::Net::Sockets::Socket::EndSendTo

4、System::Net::Sockets::Socket::EndReceiveFrom

5、System::Net::Sockets::Socket::SendToAsync

6、System::Net::Sockets::Socket::ReceiveFromAsync

总结：

于 Windows 平台上而言，则是 .NET 基于 CompletionRoutine（完成例程）、IoCompletionPort（完成端口）实现导致的问题。

于 Linux 平台上而言，则是 .NET 基于 libuv（epoll）实现导致的问题。

但这东西可以说它并不是 BUG，但仅仅只是造成大量不必要的内存被分配，挤占系统硬件资源（服务器上是很忌讳的），C/C++ 直接利用 IOCP、EPOLL 实现多线程的工作队列仍旧有很多人会造成这样的问题存在，比如：把所有的需要 Completion Context 投递到 ROOT IoCP or epoll，所有完成工作线程从 ROOT 上等待，操作系统触发事件到达，并不是固定某个特定的线程的，多个线程等待相同的 handle，触发可能是等待的任意一个线程。

问题：（源于）

为了实现真正高效的，事件驱动状态机（EDSM），并且提供与多个平台相近的API实现，每个事件（IoCompletionContext）被投递到 .NET ThreadPool（完成事件上下文工作线程池）进行驱动，Windows 平台基于 IOCP/IOCR 模型进行驱动，Linux 由 epoll 模型进行驱动，它类似 C/C++ boost::asio 的 boost::asio::io_context（io_service）IO完成队列驱动上下文，但与 boost::asio 框架不同的是，每个 Socket 实例的每次 “完成事务、Completion Transaction” 可以由 .NET 完成队列线程池 dispatch，而 boost::asio，仅仅只允许最终由每个 asio 确切的具体对象构造时指定的 Executor（io_context） 进行驱动，即不存在 .NET Implement，可能导致的，由A线程执行可变为A线程自身直接完成，或由B/C/D任意一个线程进行完成。

而对于一个 MT（多线程）架构的 Datagram 网络应用程序而言，我们不需要为每个 Socket 都需要分配一个有效的内存缓冲区（RAM Buffer），仅仅只需要为每个线程分配固态缓冲区 64KB（65535字节）即可，纵然是分配此缓冲区大小仍旧存在 RAM 空间浪费（C/C++ 建议 64KB 尽量按照粒度分配，.NET 程式分配器自行对齐）

准确需要按照不同的网络来区别需要多大的缓冲区，例如 UDP 协议则应最大缓冲区大小为：65535 - sizeof(udp_hdr, 8)，但借助 .NET 运行时提供的 Socket 是无法办到的，所以我们必须要解决这个问题，每个 UDP Dgram-socket 都需分配 64K 内存，那么 1G RAM 满打满算顶多只能分配 1.6W 个左右缓冲区，然而现实情况是根本不可能，能够分配几千个那都是烧了高香，因为你不可以单纯只算你具体分配了多少，.NET 一个托管对象实例需要占用很多无用户内存的。

#pragma pack(push, 1)
            struct udp_hdr {
            public:
                unsigned short                  src;
                unsigned short                  dest;  
                unsigned short                  len;
                unsigned short                  chksum;

            public:
                static struct udp_hdr*          Parse(struct ip_hdr* iphdr, const void* packet, int size);
            };
#pragma pack(pop)

但 TCP/IP 类型的 Stream Socket 是不建议为分个线程都分配固态缓冲区的，理由为每个操作系统都为 TCP/IP 进行过单独的吞吐优化，如果一个线程上多个 TCP Socket 使用相同 RAM 地址缓冲区数据，那么这会导致严重的 TCP/IP Socket 网络IO的收发（Received[RX]、Sent[TX]）吞吐性能严重降速，这不是框架问题，而是系统层实现问题，大量的SYN速度能快起来那就有问题了，强行解决办法则是抛弃系统层实现的易于应用层用户调用 TCP Socket，自主实现或引入一个 3rd TCP/IP 网络协议栈。

 那么既然已知道有这么个疑难问题，处理上就很简单的，本人基于 boost::asio 封装了一个可用的 C/C++/C# 库代码，看到本贴文内容的小伙伴，可自行通过其它方法自行实现或摘要本文提供的处理代码。

Usage：

        [MTAThread]
        private static void Main(string[] args)
        {
            IPAddress interfaceIP = IPAddress.Parse("192.168.0.24");
            AsyncScheduler scheduler = new AsyncScheduler(1);
            AsyncContext context = scheduler.GetContext();

            AsyncSocket socket = context.CreateSocket(new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp));
            socket.Socket.Bind(new IPEndPoint(interfaceIP, 0));

            byte[] buffer = context.Buffer;
            socket.SendTo(new byte[] { (byte)'1', (byte)'2', (byte)'3' }, 0, 3, new IPEndPoint(interfaceIP, 7000), null);
            socket.ReceiveFrom(buffer, 0, buffer.Length, (bytesTransferred, sourceEP) =>
            {
                string RX = Encoding.UTF8.GetString(buffer, 0, Math.Max(bytesTransferred, 0));
                Console.WriteLine($"bytesTransferred[{bytesTransferred}] RX[{RX}]");
            });
            Console.ReadKey(false);
        }

C/C++ Dlllibrary Implement： 

#include 
#include 
#include 
#ifdef _WIN32
#include 
#include 

#pragma comment(lib, "Ws2_32.lib")
#else
#include 
#include 
#include 
#include 
#endif

#include 
#include 
#include 
#include 

#ifndef LIBASIO_API
#ifdef __cplusplus 
#ifdef _WIN32
#define LIBASIO_API extern "C" __declspec(dllexport)
#else
#define LIBASIO_API extern "C" __attribute__((visibility("default")))
#endif
#else
#define LIBASIO_API
#endif
#endif

typedef struct {
    uint32_t            v4_or_v6_;
    union {
        struct {
            uint32_t    address_;
            uint32_t    port_;
        } in4_;
        struct {
            char        address_[16];
            uint32_t    port_;
        } in6_;
        char            data_[20];
    };
} libasio_endpoint;
typedef void(*libasio_post_callback)(void* context_, uint64_t key_);
typedef void(*libasio_sendto_callback)(void* socket_, uint64_t key_, int length_);
typedef void(*libasio_recvfrom_callback)(void* socket_, uint64_t key_, int length_, libasio_endpoint* remoteEP_);

typedef std::shared_ptr                      libasio_socket;
typedef std::shared_ptr                           libasio_context;
typedef std::unordered_map      libasio_context_hashtable;
typedef std::unordered_map  libasio_socket_hashtable;
typedef std::unordered_map libasio_so2ctx;

static libasio_so2ctx            _so2ctxs_;
static libasio_socket_hashtable  _sockets_;
static libasio_context_hashtable _contexts_;
static std::mutex                _syncobj_;

#define __lock__(obj) 
do {
    std::lock_guard scoped_(obj);
#define __unlock__ 
} while(0);

static libasio_context libasio_getcontext(boost::asio::io_context* context_) {
    libasio_context_hashtable::iterator tail = _contexts_.find(context_);
    libasio_context_hashtable::iterator endl = _contexts_.end();
    if (tail == endl) {
        return NULL;
    }
    return tail->second;
}

static libasio_context libasio_getcontext(boost::asio::ip::udp::socket* socket_) {
    libasio_so2ctx::iterator tail = _so2ctxs_.find(socket_);
    libasio_so2ctx::iterator endl = _so2ctxs_.end();
    if (tail == endl) {
        return NULL;
    }
    return tail->second;
}

static libasio_socket libasio_getsocket(boost::asio::ip::udp::socket* socket_) {
    libasio_socket_hashtable::iterator tail = _sockets_.find(socket_);
    libasio_socket_hashtable::iterator endl = _sockets_.end();
    if (tail == endl) {
        return NULL;
    }
    return tail->second;
}

LIBASIO_API
boost::asio::io_context* libasio_newcontext() {
    std::shared_ptr context_ = std::make_shared();
    std::thread([context_] {
        #ifdef _WIN32
        SetThreadPriority(GetCurrentProcess(), THREAD_PRIORITY_HIGHEST);
        #else
        
        struct sched_param param_;
        param_.sched_priority = sched_get_priority_max(SCHED_FIFO); // SCHED_RR
        pthread_setschedparam(pthread_self(), SCHED_FIFO, ¶m_);
        #endif

        boost::asio::io_context::work work_(*context_);
        boost::system::error_code ec_;
        context_->run(ec_);

        __lock__(_syncobj_){
            std::shared_ptr p = libasio_getcontext(context_.get());
            if (p) {
                _contexts_.erase(context_.get());
            }
        } __unlock__;
    }).detach();
    boost::asio::io_context* p = context_.get();
    __lock__(_syncobj_) {
        _contexts_[p] = std::move(context_);
    } __unlock__;
    return p;
}

LIBASIO_API
void libasio_closecontext(boost::asio::io_context* context_) {
    if (!context_) {
        return;
    }
    __lock__(_syncobj_) {
        std::shared_ptr p = libasio_getcontext(context_);
        if (!p) {
            return;
        }
        context_->stop();
        _contexts_.erase(context_);
    } __unlock__;
}

LIBASIO_API
bool libasio_postcontext(boost::asio::io_context* context_, uint64_t key_, libasio_post_callback callback_) {
    if (!context_ || !callback_) {
        return false;
    }
    __lock__(_syncobj_) {
        std::shared_ptr p = libasio_getcontext(context_);
        if (!p) {
            return false;
        }
        context_->post([context_, key_, callback_] {
            callback_(context_, key_);
        });
    } __unlock__;
    return true;
}

LIBASIO_API
boost::asio::ip::udp::socket* libasio_createsocket(boost::asio::io_context* context_, int sockfd_, bool v4_or_v6_) {
    if (!context_ || sockfd_ == -1) {
        return NULL;
    }
    __lock__(_syncobj_) {
        libasio_context context = libasio_getcontext(context_);
        if (!context) {
            return NULL;
        }
        libasio_socket socket_ = std::make_shared(*context_);
        if (v4_or_v6_) {
            socket_->assign(boost::asio::ip::udp::v4(), sockfd_);
        }
        else {
            socket_->assign(boost::asio::ip::udp::v6(), sockfd_);
        }
        boost::asio::ip::udp::socket* r_ = socket_.get();
        _sockets_[r_] = std::move(socket_);
        _so2ctxs_[r_] = std::move(context);
        return r_;
    } __unlock__;
}

LIBASIO_API
void libasio_closesocket(boost::asio::ip::udp::socket* socket_) {
    if (!socket_) {
        return;
    }
    __lock__(_syncobj_) {
        libasio_context context = libasio_getcontext(socket_);
        if (!context) {
            return;
        }
        libasio_socket socket = libasio_getsocket(socket_);
        boost::asio::post(*context, [context, socket] {
            if (socket->is_open()) {
                boost::system::error_code ec;
                try {
                    socket->cancel(ec);
                }
                catch (std::exception&) {}
                try {
                    socket->shutdown(boost::asio::ip::udp::socket::shutdown_send, ec);
                }
                catch (std::exception&) {}
                try {
                    socket->close(ec);
                }
                catch (std::exception&) {}
            }
        });
        _sockets_.erase(socket_);
        _so2ctxs_.erase(socket_);
    } __unlock__;
}

LIBASIO_API
bool libasio_recvfrom(boost::asio::ip::udp::socket* socket_, uint64_t key_, char* buf_, int size_, libasio_recvfrom_callback callback_) {
    if (!socket_ || !buf_ || size_ <= 0 || !callback_) {
        return false;
    }
    std::shared_ptr endpoint_ = std::make_shared();
    __lock__(_syncobj_) {
        libasio_context context_ = libasio_getcontext(socket_);
        if (!context_) {
            return false;
        }
        libasio_socket socket = libasio_getsocket(socket_);
        socket->async_receive_from(boost::asio::buffer(buf_, size_), *endpoint_,
            [context_, socket, key_, callback_, endpoint_](const boost::system::error_code& ec, uint32_t sz) {
                int length_ = -1;
                if (!ec) {
                    length_ = sz;
                }
                libasio_endpoint stack_;
                if (endpoint_->protocol() == boost::asio::ip::udp::v4()) {
                    stack_.v4_or_v6_ = 1;
                    stack_.in4_.address_ = htonl(endpoint_->address().to_v4().to_uint());
                    stack_.in4_.port_ = endpoint_->port();

                    callback_(socket.get(), key_, length_, &stack_);
                }
                else if (endpoint_->protocol() == boost::asio::ip::udp::v6()) {
                    stack_.v4_or_v6_ = 0;
                    stack_.in6_.port_ = endpoint_->port();

                    boost::asio::ip::address_v6::bytes_type addr_bytes_ = endpoint_->address().to_v6().to_bytes();
                    memcpy(stack_.in6_.address_, addr_bytes_.data(), addr_bytes_.size());

                    callback_(socket.get(), key_, length_, &stack_);
                }
                else {
                    callback_(socket.get(), key_, length_, NULL);
                }
            });
    } __unlock__;
    return true;
}

LIBASIO_API
bool libasio_sendto(boost::asio::ip::udp::socket* socket_, uint64_t key_, char* buf_, int size_, libasio_endpoint* endpoint_, libasio_sendto_callback callback_) {
    if (!socket_ || !buf_ || size_ <= 0 || !endpoint_) {
        return false;
    }
    boost::asio::ip::udp::endpoint sendtoEP_;
    if (endpoint_->v4_or_v6_) {
        sendtoEP_ = boost::asio::ip::udp::endpoint(boost::asio::ip::address_v4(ntohl(endpoint_->in4_.address_)), endpoint_->in4_.port_);
    }
    else {
        boost::asio::ip::address_v6::bytes_type addr_bytes_;
        memcpy(addr_bytes_.data(), endpoint_->in6_.address_, addr_bytes_.size());
        sendtoEP_ = boost::asio::ip::udp::endpoint(boost::asio::ip::address_v6(addr_bytes_), endpoint_->in6_.port_);
    }
    __lock__(_syncobj_) {
        libasio_context context_ = libasio_getcontext(socket_);
        if (!context_) {
            return false;
        }
        libasio_socket socket = libasio_getsocket(socket_);
        socket->async_send_to(boost::asio::buffer(buf_, size_), sendtoEP_,
            [context_, socket, key_, callback_](const boost::system::error_code& ec, uint32_t sz) {
                if (callback_) {
                    int length_ = -1;
                    if (!ec) {
                        length_ = sz;
                    }
                    callback_(socket.get(), key_, length_);
                }
            });
    } __unlock__;
    return true;
}

C# AsyncContext.cs

namespace Ppp.Net.Auxiliary
{
    using System;
    using System.Security;
    using System.Collections.Concurrent;
    using System.Diagnostics;
    using System.Net.Sockets;
#if NETCOREAPP
    using System.Runtime.CompilerServices;
#endif
    using System.Runtime.InteropServices;
    using System.Threading;

    public unsafe sealed class AsyncContext : IDisposable
    {
        [Dllimport("libasio", EntryPoint = "libasio_newcontext", CallingConvention = CallingConvention.Cdecl)]
        [return: MarshalAs(UnmanagedType.SysInt)]
        private static extern IntPtr libasio_new_context();

        [Dllimport("libasio", EntryPoint = "libasio_closecontext", CallingConvention = CallingConvention.Cdecl)]
        private static extern void libasio_closecontext([MarshalAs(UnmanagedType.SysInt)] IntPtr context_);

        [Dllimport("libasio", EntryPoint = "libasio_postcontext", CallingConvention = CallingConvention.Cdecl)]
        [return: MarshalAs(UnmanagedType.Bool)]
        private static extern bool libasio_postcontext([MarshalAs(UnmanagedType.SysInt)] IntPtr context_, long key_, [MarshalAs(UnmanagedType.FunctionPtr)] libasio_post_callback callback_);

        [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
        private delegate void libasio_post_callback([MarshalAs(UnmanagedType.SysInt)] IntPtr context_, long key_);

        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private IntPtr _handle = IntPtr.Zero;
        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private bool _disposed = false;
        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private long _mapkey = 0;
        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private readonly object _synobj = new object();
        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private readonly byte[] _buffer = new byte[ushort.MaxValue];
        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private GCHandle _buffer_gc = default(GCHandle);

        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private static readonly AsyncScheduler _scheduler = new AsyncScheduler(Environment.ProcessorCount);
        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private static readonly libasio_post_callback _callback = (context_, key_) =>
        {
            _callbacks.TryGetValue(context_, out ConcurrentDictionary callbacks);
            if (callbacks == null)
            {
                return;
            }
            callbacks.TryRemove(key_, out IOCompletionCallback callback_);
            if (callback_ == null)
            {
                return;
            }
            callback_.callback_(callback_.state_);
        };
        [DebuggerBrowsable(DebuggerBrowsableState.Never)]
        private static readonly ConcurrentDictionary> _callbacks =
            new ConcurrentDictionary>();

#if NETCOREAPP
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
#endif
        [SecurityCritical]
        [SecuritySafeCritical]
        static AsyncContext() => GCHandle.Alloc(_callback);

        private sealed class IOCompletionCallback
        {
            public object state_;
            public Action