C++20协程实例：携程化的IOCP服务端/客户端

VC支持协程已经有一段时间了，之前一直想不明白协程的意义在哪里，前几天拉屎的时候突然灵光一闪：

以下是伪代码：

task server() {
    for (;;) {
        sock_context s = co_await io.accept();
        for (;;) {
            auto buf = co_await io.recv(s);
            if (!buf.length())
                break;

            std::cout << buf.data() << std::endl;
            int n = co_await io.send(s, "收到!", strlen("收到!") + 1);
        }
        co_await io.close(s);
    }
}

如果把IO库对外的接口做成上面这样，那岂不是看起来和最简单的阻塞模型相同的代码结构，但它的内在其实是异步的，用单线程相同的代码就能支撑一堆连接通信。

所以才有了接下来的研究（闲出屁才研究的），好在研究出成品了。

最终我也明白协程的意义了：

　　协程化的库越多，C++程序员的门槛会越低，做上层开发的程序员可以不用知道协程的细节，只要知道如何正确使用库即可。

好了，真正介绍协程细节的文章有一大堆，不用我来写，我直接放代码，有兴趣的可以参考我的实现以及那些细节文章自己做：

#pragma once
#include 
#include 
#include 
#pragma comment(lib, "ws2_32.lib")
#include 
#include <string>
#include 
#include "logger.hpp"

/**
* 最近花了点时间学习了一下C++20协程，初步改造实现了IOCP协程化的网络IO库
* 此前基于回调分发的机制，由于上层协议解析所需的各种上下文，导致这个库是模板化的，
* 现在有了协程，上层协议上下文已经可以在协程函数中实现，消除了模板化，也变得易于维护了一丢丢。
* 但目前协程还有多少坑是未知的，是好是坏还得再看。
* 使用协程，就意味着，这个库几乎完全失去了多线程的能力，
* 要维护好一个内部是多线程，外皮是协程的IO库，我承认我没那个脑子。
* 我个人当前的状态是不考虑过度设计，只追求上层代码优雅简洁，10几万并发对我而言已经满足了。
* 如果这还不够用，那就意味着该放弃协程了，协程不是完全没有损耗的，根据我的测试，协程相比回调函数分发的方式，有15%左右的性能损耗。
*/
#pragma warning(push)
#pragma warning(disable:4996)
namespace aqx{

    static int init_winsock() {
        WSADATA wd;
        return WSAStartup(MAKEWORD(2, 2), &wd);
    }

    static aqx::log nlog;

#ifndef _nf
#define _nf ((size_t)-1)
#endif
#ifndef __AQX_TIME_HPP
    using clock64_t = long long;
    template
    clock64_t now() {
        const clock64_t _Freq = _Query_perf_frequency();
        const clock64_t _Ctr = _Query_perf_counter();
        const clock64_t _Whole = (_Ctr / _Freq) * period::den;
        const clock64_t _Part = (_Ctr % _Freq) * period::den / _Freq;
        return _Whole + _Part;
    }
#endif

    /** 
    * 操作码与状态码定义
    */
    struct net_status {
        static constexpr unsigned int s_accept = 0x01;
        static constexpr unsigned int s_connect = 0x02;
        static constexpr unsigned int s_read = 0x04;
        static constexpr unsigned int s_write = 0x08;
        static constexpr unsigned int s_close = 0x10;

        static constexpr unsigned int t_activated = 0x40;

        static constexpr unsigned int t_acceptor = 0x0100;
        static constexpr unsigned int t_connector = 0x0200;
        static constexpr unsigned int t_await_undo = 0x0400;

        static constexpr unsigned int t_await_accept = 0x010000;
        static constexpr unsigned int t_await_connect = 0x020000;
        static constexpr unsigned int t_await_read = 0x040000;
        static constexpr unsigned int t_await_write = 0x080000;
        static constexpr unsigned int t_await_close = 0x100000;
        static constexpr unsigned int t_await = 0xFF0000;
    };

    /** net_base 主要负责衔接操作系统
    * 不考虑过度设计，写得比较辣鸡，能用就行。
    */
    class net_base {
    public:
        net_base() {
            fd = INVALID_SOCKET;
            hIocp = NULL;
            AcceptEx = NULL;
            ConnectEx = NULL;
            DisconnectEx = NULL;
            StreamCapacity = 1440;
            Timeout = 0;
            DataBacklog = 0;
            WorkerThreadId = 0;
        }

        static bool sockaddr_from_string(sockaddr_in& _Addr, const std::string& _Dest) {
            _Addr.sin_addr.S_un.S_addr = INADDR_NONE;

            size_t pos = _Dest.find(":");
            if(pos == _nf) {
                nlog("%s->错误的目标地址:(%s)\n", __FUNCTION__, _Dest.data());
                return false;
            }

            auto strip = _Dest.substr(0, pos);
            auto strport = _Dest.substr(pos + 1);
            strport.erase(strport.find_last_not_of("\r\n\t ") + 1);
            strport.erase(0, strport.find_first_not_of("\r\n\t "));
            unsigned short port = (unsigned short)atoi(strport.c_str());
            if (!port) {
                nlog("%s->目标端口号错误:(%s)\n", __FUNCTION__, _Dest.data());
                return false;
            }
            
            strip.erase(strip.find_last_not_of("\r\n\t ") + 1);
            strip.erase(0, strip.find_first_not_of("\r\n\t "));
            auto it = std::find_if(strip.begin(), strip.end(), [](char c)->bool {
                return ((c < '0' || c > '9') && (c != '.'));
                });
            _Addr.sin_family = AF_INET;
            _Addr.sin_port = htons(port);
            if (it != strip.end()) {
                hostent* host = gethostbyname(strip.c_str());
                if (!host) {
                    nlog("%s->错误的目标域名:(%s)\n", __FUNCTION__, _Dest.data());
                    return false;
                }
                _Addr.sin_addr = *(in_addr*)(host->h_addr_list[0]);
            }
            else {
                _Addr.sin_addr.S_un.S_addr = inet_addr(strip.c_str());
            }

            if (_Addr.sin_addr.S_un.S_addr == INADDR_NONE) {
                nlog("%s->错误的目标地址:(%s)\n", __FUNCTION__, _Dest.data());
                return false;
            }
            return true;
        }

        static void sockaddr_any(sockaddr_in& _Addr, unsigned short _Port) {
            _Addr.sin_family = AF_INET;
            _Addr.sin_port = htons(_Port);
            _Addr.sin_addr.S_un.S_addr = INADDR_ANY;
        }

        static void sockaddr_local(sockaddr_in& _Addr, unsigned short _Port) {
            _Addr.sin_family = AF_INET;
            _Addr.sin_port = htons(_Port);
            _Addr.sin_addr.S_un.S_addr = INADDR_LOOPBACK;
        }

        static void* getmswsfunc(SOCKET s, GUID guid) {
            DWORD dwBytes;
            void* lpResult = nullptr;
            WSAIoctl(s, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid,
                sizeof(guid), &lpResult, sizeof(lpResult), &dwBytes, NULL, NULL);
            return lpResult;
        }

        static std::string sockaddr_to_string(const sockaddr_in &_Addr) {
            char buf[256];
            sprintf(buf, "%d.%d.%d.%d:%d", _Addr.sin_addr.S_un.S_un_b.s_b1,
                _Addr.sin_addr.S_un.S_un_b.s_b2,
                _Addr.sin_addr.S_un.S_un_b.s_b3,
                _Addr.sin_addr.S_un.S_un_b.s_b4,
                htons(_Addr.sin_port));
            std::string _Result = buf;
            return _Result;
        }

    private:
        int init(int _StreamCapacity, int _DataBacklog, int _Timeout) {
            if (fd != INVALID_SOCKET) {
                return 0;
            }
            auto reterr = [this](int n) {
                if (fd != INVALID_SOCKET) {
                    closesocket(fd);
                    fd = INVALID_SOCKET;
                }
                return n;
            };
            StreamCapacity = _StreamCapacity;
            Timeout = _Timeout;
            DataBacklog = _DataBacklog;
            fd = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
            if (fd == INVALID_SOCKET) {
                nlog("%s->创建套接字失败：%d", __FUNCTION__, WSAGetLastError());
                return reterr(-1);
            }
            ConnectEx = (LPFN_CONNECTEX)getmswsfunc(fd, WSAID_CONNECTEX);
            if (!ConnectEx) {
                nlog("%s->获取 ConnectEx 地址失败，错误号：%d", __FUNCTION__, WSAGetLastError());
                return reterr(-2);
            }
            AcceptEx = (LPFN_ACCEPTEX)getmswsfunc(fd, WSAID_ACCEPTEX);
            if (!AcceptEx) {
                nlog("%s->获取 AcceptEx 函数失败，错误号：%d", __FUNCTION__, WSAGetLastError());
                return reterr(-3);
            }
            
            // 我已经不止一次做过DisconnectEx的测试，最终结论都是DisconnectEx并不能提高并发连接数。
            // DisconnectEx 在想象中会更快是因为用IOCP队列锁去换系统全局锁带来了性能提升。
            // 还有一种方法是开一个线程搞个表去阻塞调用DisconnectEx，完事之后直接AcceptEx，也就最终把全局内核锁完全转嫁成你自己的锁了。
            // DisconnectEx首先是不同的操作系统行为不一致，真正保险的做法只能在对方关闭连接时，调用DisconnectEx来复用。
            // 对于IOCP来说，也就是在WSARecv或者WSASend 从 GetQueuedCompletionStatus 返回之后，第2个参数transferred == 0时
            // 同时它受到TCP TIME_WAIT状态的影响
            // 系统存在大量TIME_WAIT套接字时，最终得到的效果是，用了更多内存，去换来了更少的并发连接数。

            /*DisconnectEx = (LPFN_DISCONNECTEX)getmswsfunc(fd, WSAID_DISCONNECTEX);
            if (!DisconnectEx) {
                nlog("%s->获取 DisconnectEx 函数失败，错误号：%d", __FUNCTION__, WSAGetLastError());
                return reterr(-4);
            }*/

            hIocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
            if (!hIocp) {
                nlog("%s->创建完成端口失败，错误号：%d", __FUNCTION__, GetLastError());
                return reterr(-5);
            }
            CreateIoCompletionPort((HANDLE)fd, hIocp, 0, 0);
            return 0;
        }

        void close() {
            if (fd != INVALID_SOCKET) {
                closesocket(fd);
                fd = INVALID_SOCKET;
            }

            if (hIocp) {
                CloseHandle(hIocp);
                hIocp = NULL;
            }
        }

        BOOL Accept(SOCKET s, char* _Data, LPOVERLAPPED _Overlapped) {
            DWORD _Received = 0;
            return AcceptEx(fd, s, _Data, 0, sizeof(SOCKADDR_IN) << 1, sizeof(SOCKADDR_IN) << 1, &_Received, _Overlapped);
        }

        BOOL Connect(SOCKET s, sockaddr* _Addr, int _AddrLen, LPOVERLAPPED _Overlapped) {
            DWORD _Sent = 0;
            return ConnectEx(s, _Addr, _AddrLen, nullptr, 0, &_Sent, _Overlapped);
        }

        BOOL Disconnect(SOCKET s, LPOVERLAPPED _Overlapped) {
            return DisconnectEx(s, _Overlapped, TF_REUSE_SOCKET, 0);
        }

    private:
        friend class sock;
        friend class netio;
        friend class coio;
        SOCKET fd;
        HANDLE hIocp;
        LPFN_ACCEPTEX AcceptEx;
        LPFN_CONNECTEX ConnectEx;
        LPFN_DISCONNECTEX DisconnectEx;
        int StreamCapacity;
        int Timeout;
        int DataBacklog;
        DWORD WorkerThreadId;
    };

    /*直接继承一个std::string来作为套接字的各种缓冲区*/
    class sock_buffer : public std::string {
    public:
        using _Basetype = std::string;
        using _Basetype::_Basetype;
        void preset_length(size_t _Length) {
            // 直接在二进制层面去搞VC的std::string结构，修改std::string::length()的返回值
            // 这么做的好处是，免去了std::string::resize()的拷贝问题。
            // 注意这段代码仅适用于VC，G++的std::string结构和VC不一样。
            struct __stlstr {
                const char str[0x10];
                size_t len;
            };
            if (this->capacity() < _Length)
                this->reserve(_Length);
            ((__stlstr*)this)->len = _Length;
        }
    };

    /**
    * 协程task
    */
    template
    struct net_task_t {
        struct promise_type;
        using _Hty = std::coroutine_handle;
        struct promise_type {
            net_task_t get_return_object() { return { _Hty::from_promise(*this) }; }
            // initial_suspend 里返回return std::suspend_always{};表示协程初始化成功之后就挂起
            // 这里就挂起，是为了给set_sock留出操作的时间，否则一个空函数协程，会在创建完之后直接就销毁。
            auto initial_suspend() { return std::suspend_always{}; }

            auto final_suspend() noexcept { 
                s->on_destroy_coroutine(); 
                return std::suspend_never{}; 
            }
            void unhandled_exception() { std::terminate(); }
            void return_void() { }
            _Ty* s = nullptr;
        };
        _Hty _Handle;
        void resume() { _Handle.resume(); }
        void destroy() { _Handle.destroy(); }
        void set_sock(_Ty* _s) { _Handle.promise().s = _s; }
    };

    /**套接字上下文*/
    class sock {
        // 这是扩展OVERLAPPED结构
        struct binding {
            OVERLAPPED ol;
            int opt;
            sock* s;
        };

        /**
        * 返回给协程recv的对象类型
        */
        class sock_data {
            sock_data(sock* _s) : s(_s) {}
        public:
            char* data() { return s->ibuf.data(); }
            void erase(size_t _Count) { s->ibuf.erase(0, _Count); }
            size_t length() { return s->ibuf.length(); }
            void clear() { s->ibuf.clear(); }

        private:
            friend class sock;
            sock* s;
        };

        /**返回给协程connect和accept的对象类型
        * 用于异步send与close，
        * 其他线程也可以利用这个对象通信，已经处理了线程安全问题, safe_send
        */
        class asyncsock {
        public:
            /**
            * send 是未加锁的发送数据
            * 没有多线程需求时，send是安全的
            */
            int send(void* data, int len) {
                return s->send(data, len);
            }

            int send(const void* data, int len) {
                return s->send(data, len);
            }

            /**
            * 存在多线程异步写需求时，就应该所有的写操作全部用safe_send
            * 
            */
            int safe_send(void* data, int len) {
                return s->safe_send(data, len);
            }

            int safe_send(const void* data, int len) {
                return s->safe_send(data, len);
            }

            void close() {
                s->close();
            }

            bool isactivated() { return s->isactivated(); }

            operator bool() {
                return (s != nullptr);
            }

            sockaddr_in& getsockaddr() {
                return s->getsockaddr();
            }

        private:
            friend class netio;
            friend class coio;
            friend class sock;
            sock* s = nullptr;
        };

        struct recv_awaitable {
            recv_awaitable(sock* s) : data(s) { }

            __declspec(noinline) bool await_ready() {
                // 我当前的vs版本是: vs 2022 17.0.1
                // 这里发现一个编译bug，只要await_ready与await_suspend同时被inline优化
                // 会编译成如下这样的汇编：
                /*
                * 这段反汇编处于协程函数内部，在删除__declspec(noinline)之后大致就是如下的样子。
00007FF69D8A60F8  call        aqx::coio::recv (07FF69D8A2C70h)  

// 这里直接用rbx保存了recv_awaitable对象，它是被优化到寄存器的
00007FF69D8A60FD  mov         rbx,rax  
// 但是在从流程态resume()的地方跳回来协程态之后，发现它并没有维护好rbx寄存器
// 最后导致rbx = __coro_frame_ptr.__resume_address

00007FF69D8A6100  mov         rax,qword ptr [rax]  
00007FF69D8A6103  test        dword ptr [rax+20h],400h  
00007FF69D8A610A  je          `main'::`2'::$_ResumeCoro$1::operator()+448h (07FF69D8A6318h)  
00007FF69D8A6110  lea         rcx,[rax+0B8h]  
00007FF69D8A6117  cmp         qword ptr [rax+0D0h],10h  
00007FF69D8A611F  jb          `main'::`2'::$_ResumeCoro$1::operator()+258h (07FF69D8A6128h)  
00007FF69D8A6121  mov         rcx,qword ptr [rax+0B8h]  
00007FF69D8A6128  mov         qword ptr [rax+0C8h],r14  
00007FF69D8A612F  mov         byte ptr [rcx],0  
00007FF69D8A6132  mov         rax,qword ptr [rbx]  
00007FF69D8A6135  and         dword ptr [rax+20h],0FFFFFBFFh  
00007FF69D8A613C  mov         r8,qword ptr [rbx]  
00007FF69D8A613F  mov         rdx,rbx  
                */
                // 我目前尚未完全搞清楚导致这个问题的机制(还没有真正闲出屁去帮微软找bug)。
                // 所以我没上报这个问题。
                // 有兴趣的同学可以仔细跟踪一下。
                // 我目前的缓解措施是，让await_ready强制noinline，也总比在await_suspend里面resume()强。

                if (data.s->st & net_status::t_await_undo) {
                    data.s->ibuf.clear();
                    data.s->st &= (~net_status::t_await_undo);
                    return true;
                }
                return false;
            }

            
            void await_suspend(std::coroutine_handle<> handle) { }
            sock_data await_resume() const { 
                return data; 
            }
            sock_data data;
        };

        struct sock_awaitable {
            sock_awaitable(sock* _s) { s.s = _s; }
            __declspec(noinline) bool await_ready() {
                if (s.s->st & net_status::t_await_undo) {
                    s.s->st &= (~net_status::t_await_undo);
                    return true;
                }
                return false;
            }
            void await_suspend(std::coroutine_handle<> handle) { }
            sock::asyncsock await_resume() { return s; }
            sock::asyncsock s;
        };

        struct close_awaitable {
            close_awaitable(bool _IsSuspend) : IsSuspend(_IsSuspend) { }
            __declspec(noinline) bool await_ready() { return (IsSuspend == false); }
            void await_suspend(std::coroutine_handle<> handle) { }
            void await_resume() { }
            bool IsSuspend;
        };

        struct send_awaitable {
            send_awaitable(sock* _s) : s(_s) {}
            __declspec(noinline) bool await_ready() {
                if (s->st & net_status::t_await_undo) {
                    s->st &= (~net_status::t_await_undo);
                    return true;
                }
                return false;
            }
            void await_suspend(std::coroutine_handle<> handle) { }
            int await_resume() { return s->syncsendlen; }
            sock* s;
        };

        struct safe_buffer {
            sock* s = nullptr;
            sock_buffer buf;
        };

    public:
        using opcode = net_status;
        sock(net_base* _v) {
            fd = INVALID_SOCKET;
            v = _v;
            st = 0;
            memset(&input.ol, 0, sizeof(input.ol));
            memset(&output.ol, 0, sizeof(output.ol));
            
            if (v->Timeout)
                output.ol.hEvent = input.ol.hEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
            else
                output.ol.hEvent = input.ol.hEvent = NULL;
            output.s = input.s = this;
            output.opt = opcode::s_write;
            ibuf.reserve(v->StreamCapacity);
            obuf.reserve(v->StreamCapacity);
        }

        ~sock() {
            close();
            if (!output.ol.hEvent)
                return;
            CloseHandle(output.ol.hEvent);
            output.ol.hEvent = output.ol.hEvent = NULL;
            if (st & opcode::t_await) 
                co.destroy();
        }

        void on_destroy_coroutine() {
            st &= (~opcode::t_connector);
        }

        bool isactivated() {
            return ((st & opcode::t_activated) != 0);
        }

        int send(void* data, int len) {
            if (!len)
                return len;
            int n = (int)(obuf.capacity() - obuf.length());
            if (n >= len) {
                obuf.append((char*)data, len);
            }
            else {
                if (v->DataBacklog != 0 && obacklog.length() + len > v->DataBacklog) {
                    //积压值超过限制
                    close();
                    return -1;
                }
                obacklog.append((char*)data, len);
            }
            return (write() == 0) ? len : -1;
        }

        int send(const void* data, int len) {
            return send((void*)data, len);
        }

        int safe_send(void* data, int len) {
            std::lock_guard lg(mtx);
            return send(data, len);
        }

        int safe_send(const void* data, int len) {
            std::lock_guard lg(mtx);
            return send(data, len);
        }

        void close() {
            if (INVALID_SOCKET == fd)
                return;
            closesocket(fd);
            fd = INVALID_SOCKET;
            st &= ~opcode::t_activated;
            st |= opcode::s_close;
            set_timer(false);
            ibuf.clear();
            if (obacklog.capacity() <= 0x0F)
                return;
            sock_buffer tmp;
            obacklog.swap(tmp);
        }

        sockaddr_in& getsockaddr() { return sa; }

    private:
        int initfd() {
            if (INVALID_SOCKET != fd) {
                
                return 0;
            }
                
            fd = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
            if (INVALID_SOCKET == fd) {
                nlog("%s->创建套接字失败，错误号：%d", __FUNCTION__, WSAGetLastError());
                return -1;
            }
            LINGER linger = { 1, 0 };
            setsockopt(fd, SOL_SOCKET, SO_LINGER, (char*)&linger, sizeof(linger));
            int b = 1;
            setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char*)&b, sizeof(b));
            CreateIoCompletionPort((HANDLE)fd, v->hIocp, 0, 0);
            return 0;
        }

        int bindlocal() {
            sockaddr_in local;
            local.sin_family = AF_INET;
            local.sin_addr.S_un.S_addr = INADDR_ANY;
            local.sin_port = 0;
            if (SOCKET_ERROR == bind(fd, (LPSOCKADDR)&local, sizeof(local))) {
                nlog("%s->绑定本地端口失败，错误号：%d", __FUNCTION__, WSAGetLastError());
                return -1;
            }
            return 0;
        }

        bool set_dest(const std::string& _Dest) {
            return net_base::sockaddr_from_string(sa, _Dest);
        }

        void set_timer(bool _Enable) {
            if (_Enable) {
                if (hTimer)
                    return;
                RegisterWaitForSingleObject(&hTimer, output.ol.hEvent, [](void* Param, BOOLEAN TimerOrWaitFired) {
                    if (!TimerOrWaitFired)
                        return;
                    sock* p = (sock*)Param;
                    PostQueuedCompletionStatus(p->v->hIocp, 0, (ULONG_PTR)p, nullptr);
                }, this, (ULONG)v->Timeout, WT_EXECUTEDEFAULT);
            }
            else {
                if (!hTimer)
                    return;
                std::ignore = UnregisterWaitEx(hTimer, NULL);
                hTimer = NULL;
            }
        }

        int nat() {
            sockaddr_in _Addr;
            int _AddrLen = sizeof(_Addr);
            if (-1 == getsockname(fd, (sockaddr*)&_Addr, &_AddrLen))
                return -1;
            SOCKET fdNat = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
            LINGER linger = { 1, 0 };
            setsockopt(fdNat, SOL_SOCKET, SO_LINGER, (char*)&linger, sizeof(linger));
            CreateIoCompletionPort((HANDLE)fdNat, v->hIocp, 0, 0);
            if (-1 == bind(fdNat, (sockaddr*)&_Addr, sizeof(_Addr))) {
                closesocket(fdNat);
                return -1;
            }
            close();
            fd = fdNat;
            return connect();
        }

        int accept() {
            if (((st & 0xFF) | opcode::s_close) != opcode::s_close) {
                nlog("%s->当前套接字未断开连接!", __FUNCTION__);
                return -1;
            }

            if (initfd())
                return -1;
            DWORD _Received = 0;
            input.opt = opcode::s_accept;
            st &= (~opcode::s_close);
            st |= opcode::s_accept;
            if (!v->Accept(fd, ibuf.data(), &input.ol)) {
                int _Error = WSAGetLastError();
                if (_Error != ERROR_IO_PENDING) {
                    st &= (~opcode::s_accept);
                    nlog("%s->AcceptEx失败, 错误号:", __FUNCTION__, WSAGetLastError());
                    return -1;
                }
            }
            return 0;
        }

        int connect() {
            if (((st & 0xFF) | opcode::s_close) != opcode::s_close) {
                nlog("%s->当前套接字未断开连接!", __FUNCTION__);
                return -1;
            }
            if (INVALID_SOCKET == fd) {
                if (initfd())
                    return -1;
                if (bindlocal())
                    return -1;
            }
            input.opt = opcode::s_connect;
            st &= (~opcode::s_close);
            st |= opcode::s_connect;

            if (!v->Connect(fd, (sockaddr*)&sa, sizeof(sa), &input.ol)) {
                int _Error = WSAGetLastError();
                if (_Error != ERROR_IO_PENDING) {
                    nlog("%s->ConnectEx失败, 错误号:", __FUNCTION__, WSAGetLastError());
                    return -1;
                }
            }
            return 0;
        }

        int write() {
            if (!(st & opcode::t_activated)) {
                return -1;
            }
            if (st & (opcode::s_write | opcode::s_close | opcode::s_accept | opcode::s_connect))
                return 0;
            if (obacklog.size()) {
                size_t rl = obuf.capacity() - obuf.length();
                if (rl > obacklog.length())
                    rl = obacklog.length();
                if (rl) {
                    obuf.append(obacklog.data(), rl);
                    obacklog.erase(0, rl);
                }
            }
            WSABUF buf = { (ULONG)(obuf.length()), obuf.data() };
            if (!buf.len)
                return 0;
            st |= opcode::s_write;
            DWORD _Sent = 0;
            if (SOCKET_ERROR == WSASend(fd, &buf, 1, &_Sent, 0, &(output.ol), NULL)) {
                int _Error = WSAGetLastError();
                if (WSA_IO_PENDING != _Error) {
                    st &= (~opcode::s_write);
                    return -1;
                }
            }
            return 0;
        }

        int read() {
            if (!(st & opcode::t_activated)) {
                return -1;
            }
            if (st & (opcode::s_read | opcode::s_close | opcode::s_accept | opcode::s_connect))
                return 0;
            WSABUF buf = {
                (ULONG)(ibuf.capacity() - ibuf.length()),
                ibuf.data() + ibuf.length()
            };
            if ((int)buf.len <= 0) {
                return -1;
            }
            DWORD _Received = 0;
            DWORD _Flags = 0;
            st |= opcode::s_read;
            input.opt = opcode::s_read;
            if (SOCKET_ERROR == WSARecv(fd, &buf, 1, &_Received, &_Flags, &(input.ol), NULL)) {
                int _Error = WSAGetLastError();
                if (WSA_IO_PENDING != _Error) {
                    st &= ~(opcode::s_read);
                    return -1;
                }
            }
            return 0;
        }

    private:
        friend class coio;
        friend class netio;
        SOCKET fd;
        sockaddr_in sa;
        net_base* v;
        int st;
        binding input, output, reuse_sock;
        sock_buffer ibuf, obuf, obacklog;
        HANDLE hTimer;
        aqx::clock64_t rtime;
        net_task_t co;
        int syncsendlen;
        std::mutex mtx;
    };


    // coio是传参给协程函数的操作对象
    class coio {
        coio(sock* _s) : s(_s) {}

    public:
        using asyncsock = sock::asyncsock;
        using sock_awaitable = sock::sock_awaitable;
        using close_awaitable = sock::close_awaitable;
        using send_awaitable = sock::send_awaitable;
        using recv_awaitable = sock::recv_awaitable;

        struct nat_awaitable {
            nat_awaitable(bool _ret) : ret(_ret) {  }
            __declspec(noinline) bool await_ready() { return (ret == false); }
            void await_suspend(std::coroutine_handle<> handle) { }
            bool await_resume() { return ret; }
            bool ret;
        };

        coio() : s(nullptr) {}

        sock_awaitable connect(const std::string& _Dest) {
            if (!s->set_dest(_Dest)) {
                // 设置目标地址失败时，撤销等待。
                s->st |= net_status::t_await_undo;
                return sock_awaitable(s);
            }

            // 一个套接字的首次connect操作基本都是由其他线程引发的
            // 而且很可能在await_suspend之前，IOCP队列就已经完成
            if (GetCurrentThreadId() == s->v->WorkerThreadId) {
                if (s->connect()) {
                    // 连接失败时，撤销等待。
                    s->st |= net_status::t_await_undo;
                    return sock_awaitable(s);
                }
            }
            else {
                // 因此，不是IOCP队列线程引发的connect就发送到IOCP队列去处理
                PostQueuedCompletionStatus(s->v->hIocp, net_status::s_connect, (ULONG_PTR)s, 0);
            }

            s->st |= net_status::t_await_connect;
            return sock_awaitable(s);
        }

        sock_awaitable accept() {
            // 首次accept虽然也是其他线程调用的(一般是main线程)
            // 但首次accept时，IOCP工作线程尚未启动，因此可以无视掉connect的那个问题。
            s->st |= ((!s->accept()) ? net_status::t_await_accept : net_status::t_await_undo);
            return sock_awaitable(s);
        }

        /**
        * 以下几个成员函数中的参数asyncsock _s应该等同于私有成员s，除非强行在外部使用syncio对象
        * 使用参数而不是私有成员的原因是防止在尚未连接前调用IO操作。
        * 私有成员s将专用于accept与connect
        */
        close_awaitable close(asyncsock _s) {
            _s.s->close();
            if ((_s.s->st & 0xFF) == net_status::s_close) {
                // 如果套接字上已经没有任何IO事件，就让awaitable直接唤醒协程
                // 通常这才是正常状态，但如果有其他线程异步send时，可能就会有未决IO存在了。
                return close_awaitable(false);
            }
            _s.s->st |= net_status::t_await_close;
            return close_awaitable(true);
        }

        send_awaitable send(asyncsock _s, void *buf, int len) {
            _s.s->syncsendlen = _s.send(buf, len);
            _s.s->st |= ((_s.s->syncsendlen >= 0) ? net_status::t_await_write : net_status::t_await_undo);
            return sock::send_awaitable(_s.s);
        }

        send_awaitable send(asyncsock _s, const void* buf, int len) {
            _s.s->syncsendlen = _s.send(buf, len);
            _s.s->st |= ((_s.s->syncsendlen >= 0) ? net_status::t_await_write : net_status::t_await_undo);
            return sock::send_awaitable(_s.s);
        }
        
        send_awaitable safe_send(asyncsock _s, void* buf, int len) {
            _s.s->syncsendlen = _s.safe_send(buf, len);
            _s.s->st |= ((_s.s->syncsendlen >= 0) ? net_status::t_await_write : net_status::t_await_undo);
            return sock::send_awaitable(_s.s);
        }

        send_awaitable safe_send(asyncsock _s, const void* buf, int len) {
            _s.s->syncsendlen = _s.safe_send(buf, len);
            _s.s->st |= ((_s.s->syncsendlen >= 0) ? net_status::t_await_write : net_status::t_await_undo);
            return sock::send_awaitable(_s.s);
        }

        recv_awaitable recv(asyncsock _s) {
            int n = _s.s->read();
            if (n < 0) {
                _s.s->st |= net_status::t_await_undo;
            }
            else {
                _s.s->st |= net_status::t_await_read;
            }
            return recv_awaitable(_s.s);
        }

        nat_awaitable nat(asyncsock _s, const std::string& _Dest) {
            if ((_s.s->st & 0xFF) != net_status::t_activated) {
                // nat之前必须保证所有未决IO都已经返回，与打洞服务器保持正常连接状态，否则就是失败。
                // 到这里失败时，依旧与打洞服务器保持着正常连接。
                return nat_awaitable(false);
            }

            sockaddr_in sa = _s.s->sa;
            if (!_s.s->set_dest(_Dest)) {
                // 设置目标地址失败
                // 到这里失败时，依旧与打洞服务器保持着正常连接。
                _s.s->sa = sa;
                return nat_awaitable(false);
            }

            if (_s.s->nat()) {
                // 到这一步失败时，与打洞服务器的连接就有可能会断掉
                // nat失败时，本就应该直接close(); 
                // 都失败了，我想不出还要跟打洞服务器继续苟合的理由。
                // 如果所有状态全都对，还失败，可能就是双方正好属于无法穿透的NAT类型环境下。
                // 我对此研究不多，业界内真正懂行的也不多，资料更是少得可怜，我只知道TCP NAT在代码上的表现为:
                //     1、与打洞服务器保持连接的这个套接字设置了SO_REUSEADDR，确保这个套接字绑定的本地端口可复用。
                //          在这个库里我全都设置了可复用，但主要目的是为了缓解TIME_WAIT，并不是为了穿透。
                //     2、双方通过打洞服务器沟通好各自的远端地址
                //     3、双方都创建一个新的套接字，并将该套接字绑定到本地与打洞服务器进行连接的那个地址(getsockname可以获得)
                //          到第 3 步处理好之后，与打洞服务器连接的那个套接字，已经废了，无法再进行通信，此时应该把它close掉。
                //     4、最后双方都connect对方的地址。
                _s.s->sa = sa;
                return nat_awaitable(false);
            }

            s->st |= net_status::t_await_connect;
            return nat_awaitable(true);
        }

        bool valid() {
            return (s != nullptr);
        }

        operator bool () {
            return valid();
        }

    private:
        friend class netio;
        sock* s;
    };

    /**
    * 可以简单把netio看成是一个容器的作用
    * 它主要用于对接net_base，创建线程，处理IO事件。
    */
    class netio {
        struct IOCP_STATUS {
            DWORD transferred;
            SIZE_T key;
            typename sock::binding* pb;
            BOOL ok;
        };

    public:
        /**listener 只是一种简单的参数包装，只是为了方便构造而已
        * 构造参数：
        * _Dest 要监听的地址和端口，格式为："a.b.c.d:port"
        * _ListenBacklog 系统函数listen的第2个参数
        * _MaxClients 最多同时接受的客户端数量
        */
        class listener {
        public:
            listener() {
                max_clients = 0;
                listen_backlog = 0;
                addr.sin_addr.S_un.S_addr = INADDR_NONE;
            }

            listener(const std::string& _Dest, int _ListenBacklog, size_t _MaxClients) {
                max_clients = _MaxClients;
                listen_backlog = _ListenBacklog;
                net_base::sockaddr_from_string(addr, _Dest);
            }

        private:
            friend class netio;
            sockaddr_in addr;
            int listen_backlog;
            size_t max_clients;
        };

        using asyncsock = sock::asyncsock;
        using opcode = net_status;
        using task = net_task_t;

        int init(int _StreamCapacity = 1440, int _DataBacklog = 0, int _Timeout = 0) {
            std::lock_guard lg(mtx);
            return nwb.init(_StreamCapacity, _DataBacklog, _Timeout);
        }

        int server(const std::function &_func, const listener &param) {
            std::lock_guard lg(mtx);
            if (thd.joinable()) {
                nlog("%s->netio已启动, 请勿重复调用!", __FUNCTION__);
                return 0;
            }

            if (nwb.fd == INVALID_SOCKET)
                return -1;

            cofunc = _func;
            if (param.addr.sin_addr.S_un.S_addr != INADDR_NONE) {
                if (SOCKET_ERROR == bind(nwb.fd, (SOCKADDR*)¶m.addr, sizeof(SOCKADDR))) {
                    nlog("%s->绑定端口失败，错误号：%d", __FUNCTION__, WSAGetLastError());
                    nwb.close();
                    return -1;
                }

                if (SOCKET_ERROR == ::listen(nwb.fd, param.listen_backlog)) {
                    nlog("%s->监听失败，错误号：%d", __FUNCTION__, WSAGetLastError());
                    nwb.close();
                    return -1;
                }

                for (int i = 0; i < param.max_clients; i++) {
                    sock* psock = new sock(&nwb);
                    a_list.push_back(psock);
                    psock->st |= opcode::t_acceptor;
                    psock->co = cofunc(coio(psock));
                    psock->co.set_sock(psock);
                    psock->co.resume();
                }
            }
            __start();
            return 0;
        }

        // client是一次性的，专用于客户端
        // 让它返回asyncsock对象的理由是为了给脚本语言预留的
        // 例如可以使用lua去实现类似node.js的那种connect之后不管连没连上就先得到对象去绑定事件的机制。
        asyncsock client(const std::function& _func) {
            std::lock_guard lg(mtx);
            coio io;
            asyncsock ret;
            if (!thd.joinable()) {
                // 如果线程未启动，尝试启动线程，这之后如果要回收资源，是需要stop和release的
                if (nwb.fd == INVALID_SOCKET)
                    return ret;
                __start();
            }
            io.s = get_connector();
            ret.s = io.s;
            io.s->co = _func(io);
            io.s->co.set_sock(io.s);
            io.s->co.resume();
            return ret;
        }

        void stop() {
            std::lock_guard lg(mtx);
            if (thd.joinable()) {
                PostQueuedCompletionStatus(nwb.hIocp, -1, 0, 0);
                thd.join();
            }
        }

        void release() {
            std::lock_guard lg(mtx);
            if (thd.joinable()) {
                nlog("%s->nio正在运行，请先stop", __FUNCTION__);
                return;
            }
            for (auto p : a_list) {
                if (p->st & opcode::t_await)
                    p->co.destroy();
                delete p;
            }
            a_list.clear();

            for (auto p : c_list) {
                if (p->st & opcode::t_await)
                    p->co.destroy();
                delete p;
            }
            c_list.clear();
            nwb.close();
        }

    private:
        sock* get_connector() {
            sock* psock = nullptr;
            
            for (auto v : c_list) {
                if ((v->st & opcode::t_connector) == 0 && ((v->st & 0xFF)| opcode::s_close) == opcode::s_close) {
                    psock = v;
                    break;
                }
            }

            if (!psock) {
                psock = new sock(&nwb);
                c_list.push_back(psock);
            }

            psock->st |= opcode::t_connector;
            return psock;
        }

        void on_connect(sock& s) {
            s.ibuf.clear();
            s.obuf.clear();
            s.obacklog.clear();
            s.rtime = aqx::now() + nwb.Timeout;
            if (nwb.Timeout != 0)
                s.set_timer(true);
            s.st |= opcode::t_activated;
        }
        
        void on_accept(sock &s) {
            // 懒得去调用GetAcceptExSockAddrs，有硬编码可用
#ifndef _WIN64
            s.sa = *(sockaddr_in*)(s.ibuf.data() + 0x26);
#else
            s.sa = *(sockaddr_in*)(s.ibuf.data() + 0x20);
#endif
            on_connect(s);
        }
        
        bool on_resume(sock& s) {
            if (s.st & opcode::t_await) {
                // 清除所有协程等待标志
                s.st &= (~opcode::t_await);

                // 唤醒协程
                s.co.resume();
                return true;
            }
            return false;
        }

        void on_close(sock& s) {
            if ((s.st & 0xFF) == opcode::s_close) {
                s.st &= ~opcode::s_close;
                on_resume(s);
            }
        }

        bool error_resume(sock &s) {
            int st = s.st & opcode::t_await;
            switch (st) {
            case opcode::t_await_accept:
            case opcode::t_await_connect:
            case opcode::t_await_close:
                s.st &= (~opcode::t_await);
                s.co.resume();
                return true;
            case opcode::t_await_read:
                s.ibuf.clear();
                s.st &= (~opcode::t_await);
                s.co.resume();
                return true;
            case opcode::t_await_write:
                s.syncsendlen = -1;
                s.st &= (~opcode::t_await);
                s.co.resume();
                return true;
            default:
                break;
            }
            return false;
        }

        void on_reset(sock &s) {
            if ((s.st & 0xFF) == opcode::s_close) {
                s.st &= ~opcode::s_close;
                if (s.st & opcode::t_acceptor) {
                    // 如果服务端协程不在一个循环里，协程返回自动销毁后就会这样
                    // 此时的挽救措施就是创建一个新的协程
                    s.co = cofunc(coio(&s));
                }
            }
        }

        void on_completion(IOCP_STATUS& st) {
            sock& s = *(st.pb->s);
            int op = st.pb->opt;
            s.st &= (~op);
            if (s.st & opcode::s_close) 
                op = 0;
            //nlog("on_completion:%I64X, %d", &s, op);
            switch (op) {
            case 0:
                break;
            case opcode::s_accept:
                on_accept(s);
                break;
            case opcode::s_connect:
                if (!st.ok && WSAGetLastError() == 1225) {
                    // 出现这种错误，一般是由于服务端没有在监听指定端口，直接被操作系统拒绝了。
                    op = 0;
                    break;
                }
                on_connect(s);
                break;
            case opcode::s_read:
                if (!st.transferred) {
                    op = 0;
                    break;
                }
                s.ibuf.preset_length(s.ibuf.length() + st.transferred);
                break;
            case opcode::s_write:
                if (!st.transferred) {
                    op = 0;
                    break;
                }

                s.obuf.erase(0, st.transferred);
                if (s.obuf.length() || s.obacklog.length()) {
                    if (s.write()) {
                        op = 0;
                        break;
                    }
                }
                // write操作可能是非协程发起的，协程很可能挂起在recv，因此需要判断一下。
                if (!(s.st & opcode::t_await_write))
                    return;
                break;
            }
            
            //nlog("on_completion2:%I64X, %d", &s, op);
            if (!op) {
                if (error_resume(s))
                    return;
                // 只有当协程被销毁时，error_resume才会返回false
                s.close();
                on_reset(s);
                return;
            }
            
            on_resume(s);
            if (s.st & opcode::s_close)
                return on_close(s);
        }

        void __start() {
            thd = std::thread([this]() {
                nwb.WorkerThreadId = GetCurrentThreadId();
                srand((unsigned int)aqx::now() + nwb.WorkerThreadId);
                IOCP_STATUS st = { 0,0,0,0 };
                //nlog("netio::worker->I/O工作线程 %d 开始!", nwb.WorkerThreadId);
                for (;;) {
                    st.ok = GetQueuedCompletionStatus(nwb.hIocp,
                        &(st.transferred),
                        &(st.key),
                        (OVERLAPPED**)&(st.pb),
                        INFINITE);

                    if (!st.pb) {
                        if (!st.transferred) {
                            sock* psock = (sock*)st.key;
                            if (aqx::now() > psock->rtime && (psock->st & opcode::t_activated)) {
                                psock->close();
                                if (error_resume(*psock))
                                    return;
                                on_reset(*psock);
                            }
                        }
                        else if (st.transferred == opcode::s_connect) {
                            sock* psock = (sock*)st.key;
                            if (psock->connect()) {
                                psock->close();
                                if (error_resume(*psock))
                                    return;
                                on_reset(*psock);
                            }
                        }
                        else if (st.transferred == -1)
                            break;
                        continue;
                    }
                    on_completion(st);
                }
                
                //nlog("netio::worker->I/O工作线程 %d 已停止!", nwb.WorkerThreadId);
            });
        }

    private:
        net_base nwb;
        std::list a_list;
        std::list c_list;
        std::function cofunc;
        std::thread thd;
        std::mutex mtx;
    };
}

#pragma warning(pop)

这个库我已经去除了各种耦合，除了日志库，aqx::log我自己写的一个简单的格式化日志库：

logger.hpp
#pragma once
#include 
#include <string>
#include 
#include 
#include 
#include 

//aqx::log不与aqx其他库耦合
#if defined(_WIN32) || defined(_WIN64)
#ifndef _WINDOWS_
#include 
#endif
#define __aqxlog_getpid GetCurrentProcessId
#define __aqxlog_gettid GetCurrentThreadId
#include 
#else
#if defined(__linux__)
#include 
#include 
#define __aqxlog_getpid getpid
#define __aqxlog_gettid() syscall(__NR_gettid)
#endif
#endif

#pragma warning(push)
#pragma warning(disable:4996)

namespace aqx {

    class log {
    private:
        struct _format_texts {
            std::string time;
            std::string type;
            std::string pid;
            std::string tid;
        };

    public:
        static constexpr auto hs_time{ static_cast<int>(1) };
        static constexpr auto hs_type{ static_cast<int>(2) };
        static constexpr auto hs_pid{ static_cast<int>(4) };
        static constexpr auto hs_tid{ static_cast<int>(8) };

        log() {
            _stdout_fp = stdout;
            fp = stdout;
            _fmtts = { "%Y/%m/%d %H:%M:%S ", "{%s} ",  "[%d] ",  "(%d) " };
            head_style = log::hs_time;
            head_presize = _gethps();
            _EnableInfo = true;
            _EnableError = false;
            _EnableDebug = false;
            _EnableWarn = false;
            _DefType = "info";
            s.reserve(0x1000);
        }

        ~log() {
            if (fp != _stdout_fp)
                fclose(fp);
        }

        void enable(const std::string_view& _Type, bool _Enable) {
            std::lock_guard lg(_Mtx);
            if (_Type == "info")
                _EnableInfo = _Enable;
            else if (_Type == "error")
                _EnableError = _Enable;
            else if (_Type == "debug")
                _EnableDebug = _Enable;
            else if (_Type == "warn")
                _EnableWarn = _Enable;
        }

        void seths(int hs) {
            std::lock_guard lg(_Mtx);
            head_style = hs;
            head_presize = _gethps();
        }

        void sethfmt(int _Style, const char* _Fmt) {
            std::lock_guard lg(_Mtx);
            switch (_Style) {
            case hs_time:
                _fmtts.time = _Fmt;
                break;
            case hs_type:
                _fmtts.type = _Fmt;
                break;
            case hs_pid:
                _fmtts.pid = _Fmt;
                break;
            case hs_tid:
                _fmtts.tid = _Fmt;
                break;
            }
            head_presize = _gethps();
        }

        bool setvfs(const char* _FileName, bool _PutStdout = false) {
            std::lock_guard lg(_Mtx);
            FILE* _tmp = fopen(_FileName, "ab");
            if (!_tmp)
                return false;
            if (fp != _stdout_fp)
                fclose(fp);
            fp = _tmp;
            PutStdout = _PutStdout;
            return true;
        }

        log& info(const char* _Fmt, ...) {
            std::lock_guard lg(_Mtx);
            if (!_EnableInfo)
                return *this;
            va_list vl;
            va_start(vl, _Fmt);
            _build("info", _Fmt, vl);
            va_end(vl);
            _putlog();
            return *this;
        }

        log& debug(const char* _Fmt, ...) {
            std::lock_guard lg(_Mtx);
            if (!_EnableDebug)
                return *this;
            va_list vl;
            va_start(vl, _Fmt);
            _build("info", _Fmt, vl);
            va_end(vl);
            _putlog();
            return *this;
        }

        log& error(const char* _Fmt, ...) {
            std::lock_guard lg(_Mtx);
            if (!_EnableError)
                return *this;
            va_list vl;
            va_start(vl, _Fmt);
            _build("info", _Fmt, vl);
            va_end(vl);
            _putlog();
            return *this;
        }

        log& warn(const char* _Fmt, ...) {
            std::lock_guard lg(_Mtx);
            if (!_EnableWarn)
                return *this;
            va_list vl;
            va_start(vl, _Fmt);
            _build("info", _Fmt, vl);
            va_end(vl);
            _putlog();
            return *this;
        }

        log& operator()(const char* _Fmt, ...) {
            std::lock_guard lg(_Mtx);
            if (!_EnableInfo)
                return *this;
            va_list vl;
            va_start(vl, _Fmt);
            _build(_DefType.c_str(), _Fmt, vl);
            va_end(vl);
            _putlog();
            return *this;
        }

    private:
        void _putlog() {
            fputs(s.data(), fp);
            if (fp != _stdout_fp) {
                //fflush(fp);
                if (PutStdout)
                    fputs(s.data(), _stdout_fp);
            }
        }

        size_t _build(const char* _Type, const char* _Fmt, va_list vl) {
            s.clear();
            size_t n = vsnprintf(nullptr, 0, _Fmt, vl);
            if (n <= 0)
                return _build_head(_Type);
            if (n >= s.capacity()) {
                s.clear();
                s.reserve(n + head_presize);
            }
            size_t _Pos = _build_head(_Type);
            char* p = (char*)s.data();
            _Pos += vsnprintf(p + _Pos, s.capacity(), _Fmt, vl);
            char c = p[_Pos - 1];
#ifdef _WINDOWS_
            if (c != '\r' && c != '\n') {
                p[_Pos++] = '\r';
                p[_Pos++] = '\n';
                p[_Pos] = '\0';
            }

#else
            if (c != '\r' && c != '\n') {
                p[_Pos++] = '\n';
                p[_Pos] = '\0';
            }
#endif

            return _Pos;
        }

        size_t _build_time(size_t _Pos) {
            if (!(head_style & log::hs_time))
                return _Pos;
            time_t t = time(NULL);
            auto _Tm = localtime(&t);
            _Pos += strftime((char*)s.data() + _Pos, head_presize, _fmtts.time.c_str(), _Tm);
            return _Pos;
        }

        size_t _build_type(size_t _Pos, const char* _Type) {
            if (!(head_style & log::hs_type))
                return _Pos;
            _Pos += sprintf((char*)s.data() + _Pos, _fmtts.type.c_str(), _Type);
            return _Pos;
        }

        size_t _build_pid(size_t _Pos) {
            if (!(head_style & log::hs_pid))
                return _Pos;
            auto _Pid = __aqxlog_getpid();
            _Pos += sprintf((char*)s.data() + _Pos, _fmtts.pid.c_str(), _Pid);
            return _Pos;
        }

        size_t _build_tid(size_t _Pos) {
            if (!(head_style & log::hs_tid))
                return _Pos;
            auto _Tid = __aqxlog_gettid();
            _Pos += sprintf((char*)s.data() + _Pos, _fmtts.tid.c_str(), _Tid);
            return _Pos;
        }

        size_t _build_head(const char* _Type) {
            return _build_tid(_build_pid(_build_type(_build_time(0), _Type)));
        }

        size_t _gethps() {
            size_t _Result = 3;
            if (head_style & log::hs_time)
                _Result += ((_fmtts.time.length() << 1) + 30);
            if (head_style & log::hs_type)
                _Result += ((_fmtts.pid.length() << 1) + 12);
            if (head_style & log::hs_pid)
                _Result += ((_fmtts.pid.length() << 1) + 20);
            if (head_style & log::hs_tid)
                _Result += ((_fmtts.pid.length() << 1) + 20);
            return _Result;
        }

    private:
        std::vector<char> s;
        FILE* fp;
        _format_texts _fmtts;
        int head_style;
        size_t head_presize;
        bool PutStdout;
        FILE* _stdout_fp;
        std::mutex _Mtx;
        std::string _DefType;
        bool _EnableInfo;
        bool _EnableDebug;
        bool _EnableError;
        bool _EnableWarn;
    };
}

static aqx::log logger;
#pragma warning(pop)

最后是测试代码：客户端和服务端放在一起了，要分离就从nio.init后面的几个地方分离一下。

// main.cpp
#include 
#include 

int main()
{
    aqx::init_winsock();

    aqx::netio nio;
    nio.init(1440, 0x10000);

    // 一个简单的echo服务器例子：

    nio.server([](aqx::coio io)->aqx::netio::task {
        // 服务端始终应该放在一个死循环里，否则兜底逻辑会反复创建新协程。
        for (;;) {
            // io.accept会返回一个可用于异步send和close的对象
            auto s = co_await io.accept();
            logger("客户端连入:%s", aqx::net_base::sockaddr_to_string(s.getsockaddr()));
            for (;;) {
                auto buf = co_await io.recv(s);
                if (!buf.length()) {
                    logger("断开连接!");
                    break;
                }

                puts(buf.data());
                buf.clear();
                // 异步发送，协程不会在这里挂起
                s.send("收到!", 5);
                
            }
            co_await io.close(s);
            logger("已关闭!");
        }
    }, aqx::netio::listener("0.0.0.0:55554", 100, 100));



    // 我已经懒到让客户端和服务端都放在一起了，要分自己分
    auto sock1 = nio.client([](aqx::coio io)->aqx::netio::task {
        // 客户端只有需要自动重连，才放在循环里处理
        for (;;) {
            auto s = co_await io.connect("127.0.0.1:55554");
            if (!s) {
                co_await io.close(s);
                continue;
            }

            for (;;) {
                auto buf = co_await io.recv(s);
                if (!buf.length()) {
                    break;
                }
                puts(buf.data());
                buf.clear();
            }
            
            co_await io.close(s);
        }
       
    });

    // 我已经懒到让客户端和服务端都放在一起了，要分自己分
    auto sock2 = nio.client([](aqx::coio io)->aqx::netio::task {
        // 客户端只有需要自动重连，才放在循环里处理
        for (;;) {
            auto s = co_await io.connect("127.0.0.1:55554");
            if (!s) {
                co_await io.close(s);
                continue;
            }

            for (;;) {
                auto buf = co_await io.recv(s);
                if (!buf.length()) {
                    break;
                }
                puts(buf.data());
                buf.clear();
            }

            co_await io.close(s);
        }

    });
    
    std::string str;
    for (;;) {
        std::cin >> str;
        if (str == "exit")
            break;

        std::string sd = "sock1:";
        sd += str;
        sock1.safe_send(sd.data(), (int)sd.length() + 1);

        sd = "sock2:";
        sd += str;
        sock2.safe_send(sd.data(), (int)sd.length() + 1);
    }

    nio.stop();
    nio.release();
}