MongoDB网络传输层模块源码实现二

//网络会话链接，消息处理管理相关的类，在createWithConfig构造该类存入_tls  
 class TransportLayerManager final : public TransportLayer {  
     //以下四个接口真正实现在TransportLayerASIO类中具体实现  
     Ticket sourceMessage(...) override;  
     Ticket sinkMessage(...) override;      
     Status wait(Ticket&& ticket) override;  
     void asyncWait(...) override;  
     //配置初始化实现  
     std::unique_ptr<TransportLayer> createWithConfig(...);  
     
       //createWithConfig中赋值,对应TransportLayerASIO,  
       //实际上容器中就一个成员,就是TransportLayerASIO  
       std::vector<std::unique_ptr<TransportLayer>> _tls;  
   };  

//根据配置构造相应类信息  _initAndListen中调用  
std::unique_ptr<TransportLayer> TransportLayerManager::createWithConfig(...) {  
    std::unique_ptr<TransportLayer> transportLayer;  
    //服务类型，也就是本实例是mongos还是mongod  
    //mongos对应ServiceEntryPointMongod,mongod对应ServiceEntryPointMongos  
    auto sep = ctx->getServiceEntryPoint();  
    //net.transportLayer配置模式，默认asio, legacy模式已淘汰  
    if (config->transportLayer == "asio") {  
         //同步方式还是异步方式，默认synchronous  
        if (config->serviceExecutor == "adaptive") {  
            //动态线程池模型,也就是异步模式  
            opts.transportMode = transport::Mode::kAsynchronous;  
        } else if (config->serviceExecutor == "synchronous") {  
            //一个链接一个线程模型，也就是同步模式  
            opts.transportMode = transport::Mode::kSynchronous;  
        }   
        //如果配置是asio,构造TransportLayerASIO类  
        auto transportLayerASIO = stdx::make_unique<transport::TransportLayerASIO>(opts, sep);  
        if (config->serviceExecutor == "adaptive") { //异步方式  
             //构造动态线程模型对应的执行器ServiceExecutorAdaptive  
            ctx->setServiceExecutor(stdx::make_unique<ServiceExecutorAdaptive>(  
                ctx, transportLayerASIO->getIOContext()));  
         } else if (config->serviceExecutor == "synchronous") { //同步方式  
            //构造一个链接一个线程模型对应的执行器ServiceExecutorSynchronous  
            ctx->setServiceExecutor(stdx::make_unique<ServiceExecutorSynchronous>(ctx));  
         }  
         //transportLayerASIO转换为transportLayer类  
         transportLayer = std::move(transportLayerASIO);  
    }   
   //transportLayer转存到对应retVector数组中并返回  
    std::vector<std::unique_ptr<TransportLayer>> retVector;  
    retVector.emplace_back(std::move(transportLayer));  
    return stdx::make_unique<TransportLayerManager>(std::move(retVector));  
}

class TransportLayerASIO final : public TransportLayer {  
    //以下四个接口主要和套接字数据读写相关  
    Ticket sourceMessage(...);  
    Ticket sinkMessage(...);  
    Status wait(Ticket&& ticket);  
    void asyncWait(Ticket&& ticket, TicketCallback callback);  
    void end(const SessionHandle& session);  
    //新链接处理  
    void _acceptConnection(GenericAcceptor& acceptor);  
      
    //adaptive线程模型网络IO上下文处理  
    std::shared_ptr<asio::io_context> _workerIOContext;   
    //accept接收客户端链接对应的IO上下文  
    std::unique_ptr<asio::io_context> _acceptorIOContext;    
    //bindIp配置中的ip地址列表，用于bind监听，accept客户端请求  
    std::vector<std::pair<SockAddr, GenericAcceptor>> _acceptors;  
    //listener线程负责接收客户端新链接  
    stdx::thread _listenerThread;  
    //服务类型，也就是本实例是mongos还是mongod  
    //mongos对应ServiceEntryPointMongod,mongod对应ServiceEntryPointMongos  
    ServiceEntryPoint* const _sep = nullptr;  
    //当前运行状态  
    AtomicWord<bool> _running{false};  
    //listener处理相关的配置信息  
    Options _listenerOptions;  
}  

Status TransportLayerASIO::setup() {  
    std::vector<std::string> listenAddrs;  
    //如果没有配置bindIp,则默认监听"127.0.0.1：27017"
    if (_listenerOptions.ipList.empty()) {  
        listenAddrs = {"127.0.0.1"};  
    } else {  
        //配置文件中的bindIp:1.1.1.1,2.2.2.2，以逗号分隔符获取ip列表存入ipList  
        boost::split(listenAddrs, _listenerOptions.ipList, boost::is_any_of(","), boost::token_compress_on);  
    }  
    //遍历ip地址列表  
    for (auto& ip : listenAddrs) {  
        //根据IP和端口构造对应SockAddr结构  
        const auto addrs = SockAddr::createAll(  
            ip, _listenerOptions.port, _listenerOptions.enableIPv6 ? AF_UNSPEC : AF_INET);  
        ......  
        //根据addr构造endpoint  
        asio::generic::stream_protocol::endpoint endpoint(addr.raw(), addr.addressSize);  
        //_acceptorIOContext和_acceptors关联  
        GenericAcceptor acceptor(*_acceptorIOContext);  
        //epoll注册，也就是fd和epoll关联  
        //basic_socket_acceptor::open  
        acceptor.open(endpoint.protocol());   
         //SO_REUSEADDR配置 basic_socket_acceptor::set_option  
        acceptor.set_option(GenericAcceptor::reuse_address(true));  
        //非阻塞设置 basic_socket_acceptor::non_blocking  
        acceptor.non_blocking(true, ec);    
        //bind绑定    
        acceptor.bind(endpoint, ec);   
        if (ec) {  
            return errorCodeToStatus(ec);  
        }  
    }  
}  

//_initAndListen中调用执行   
Status TransportLayerASIO::start() { //listen线程处理  
    ......  
    //这里专门起一个线程做listen相关的accept事件处理  
    _listenerThread = stdx::thread([this] {  
        //修改线程名  
        setThreadName("listener");   
        //该函数中循环处理accept事件  
        while (_running.load()) {  
            asio::io_context::work work(*_acceptorIOContext);   
            try {  
                //accept事件调度处理  
                 _acceptorIOContext->run();    
            } catch (...) { //异常处理  
                severe() << "Uncaught exception in the listener: " << exceptionToStatus();  
                fassertFailed(40491);  
            }  
        }  
    });   
   遍历_acceptors，进行listen监听处理  
   for (auto& acceptor : _acceptors) {   
        acceptor.second.listen(serverGlobalParams.listenBacklog);  
        //异步accept回调注册在该函数中  
        _acceptConnection(acceptor.second);       
    }  
}

//accept新连接到来的回调注册 
void TransportLayerASIO::_acceptConnection(GenericAcceptor& acceptor) {  
      //新链接到来时候的回调函数，服务端接收到新连接都会执行该回调
    //注意这里面是递归执行，保证所有accept事件都会一次处理完毕
    auto acceptCb = [this, &acceptor](const std::error_code& ec, GenericSocket peerSocket) mutable {  
        if (!_running.load())  
            return;  
  
        ......  
        //每个新的链接都会new一个新的ASIOSession  
        std::shared_ptr<ASIOSession> session(new ASIOSession(this, std::move(peerSocket)));  
        //新的链接处理ServiceEntryPointImpl::startSession，  
        //和ServiceEntryPointImpl服务入口点模块关联起来  
        _sep->startSession(std::move(session));  
        //递归，直到处理完所有的网络accept事件  
        _acceptConnection(acceptor);   
    };  
    //accept新连接到来后服务端的回调处理在这里注册  
    acceptor.async_accept(*_workerIOContext, std::move(acceptCb));  
}  

//根据asioSession, expiration, message三个信息构造数据接收类ASIOSourceTicket  
Ticket TransportLayerASIO::sourceMessage(...) {  
    ......  
    auto asioSession = checked_pointer_cast<ASIOSession>(session);  
    //根据asioSession, expiration, message三个信息构造ASIOSourceTicket  
    auto ticket = stdx::make_unique<ASIOSourceTicket>(asioSession, expiration, message);  
    return {this, std::move(ticket)};  
}  
  
//根据asioSession, expiration, message三个信息构造数据发送类ASIOSinkTicket  
Ticket TransportLayerASIO::sinkMessage(...) {  
    auto asioSession = checked_pointer_cast<ASIOSession>(session);  
    auto ticket = stdx::make_unique<ASIOSinkTicket>(asioSession, expiration, message);  
    return {this, std::move(ticket)};  
}  
  
//同步接收或者发送，最终调用ASIOSourceTicket::fill 或者 ASIOSinkTicket::fill  
Status TransportLayerASIO::wait(Ticket&& ticket) {  
    //获取对应Ticket，接收对应ASIOSourceTicket，发送对应ASIOSinkTicket  
    auto ownedASIOTicket = getOwnedTicketImpl(std::move(ticket));  
    auto asioTicket = checked_cast<ASIOTicket*>(ownedASIOTicket.get());  
    ......  
    //调用对应fill接口 同步接收ASIOSourceTicket::fill 或者 同步发送ASIOSinkTicket::fill  
    asioTicket->fill(true, [&waitStatus](Status result) { waitStatus = result; });  
    return waitStatus;  
}  
//异步接收或者发送，最终调用ASIOSourceTicket::fill 或者 ASIOSinkTicket::fill  
void TransportLayerASIO::asyncWait(Ticket&& ticket, TicketCallback callback) {  
    //获取对应数据收发的Ticket，接收对应ASIOSourceTicket，发送对应ASIOSinkTicket  
    auto ownedASIOTicket = std::shared_ptr<TicketImpl>(getOwnedTicketImpl(std::move(ticket)));  
    auto asioTicket = checked_cast<ASIOTicket*>(ownedASIOTicket.get());  
  
   //调用对应ASIOTicket::fill  
    asioTicket->fill(  
        false,   [ callback = std::move(callback),  
        ownedASIOTicket = std::move(ownedASIOTicket) ](Status status) { callback(status); });  
}  

class ServiceEntryPointImpl : public ServiceEntryPoint {  
    MONGO_DISALLOW_COPYING(ServiceEntryPointImpl);  
public:  
    //构造函数  
    explicit ServiceEntryPointImpl(ServiceContext* svcCtx);     
    //以下三个接口进行session会话处理控制  
    void startSession(transport::SessionHandle session) final;  
    void endAllSessions(transport::Session::TagMask tags) final;  
    bool shutdown(Milliseconds timeout) final;  
    //session会话统计  
    Stats sessionStats() const final;  
    ......  
private:  
    //该list结构管理所有的ServiceStateMachine信息  
    using SSMList = stdx::list<std::shared_ptr<ServiceStateMachine>>;  
    //SSMList对应的迭代器  
    using SSMListIterator = SSMList::iterator;  
    //赋值ServiceEntryPointImpl::ServiceEntryPointImpl  
    //对应ServiceContextMongoD(mongod)或者ServiceContextNoop(mongos)类  
    ServiceContext* const _svcCtx;   
    //该成员变量在代码中没有使用  
    AtomicWord<std::size_t> _nWorkers;  
    //锁  
    mutable stdx::mutex _sessionsMutex;  
    //一个新链接对应一个ssm保存到ServiceEntryPointImpl._sessions中  
    SSMList _sessions;  
    //最大链接数控制  
    size_t _maxNumConnections{DEFAULT_MAX_CONN};  
    //当前的总链接数，不包括关闭的链接  
    AtomicWord<size_t> _currentConnections{0};  
    //所有的链接，包括已经关闭的链接  
    AtomicWord<size_t> _createdConnections{0};  
};  

//新链接到来后的回调处理  
void ServiceEntryPointImpl::startSession(transport::SessionHandle session) {   
    //获取该新连接对应的服务端和客户端地址信息  
    const auto& remoteAddr = session->remote().sockAddr();  
    const auto& localAddr = session->local().sockAddr();  
    //服务端和客户端地址记录到session中  
    auto restrictionEnvironment =  stdx::make_unique<RestrictionEnvironment>(*remoteAddr, *localAddr);  
    RestrictionEnvironment::set(session, std::move(restrictionEnvironment));  
    ......  
  
    //获取transportMode，kAsynchronous或者kSynchronous  
    auto transportMode = _svcCtx->getServiceExecutor()->transportMode();  
    //构造ssm  
    auto ssm = ServiceStateMachine::create(_svcCtx, session, transportMode);  
    {//该{}体内实现链接计数，同时把ssm统一添加到_sessions列表管理  
        stdx::lock_guard<decltype(_sessionsMutex)> lk(_sessionsMutex);  
        connectionCount = _sessions.size() + 1; //连接数自增  
        if (connectionCount <= _maxNumConnections) {  
            //新来的链接对应的session保存到_sessions链表    
            //一个新链接对应一个ssm保存到ServiceEntryPointImpl._sessions中  
            ssmIt = _sessions.emplace(_sessions.begin(), ssm);  
            _currentConnections.store(connectionCount);  
            _createdConnections.addAndFetch(1);  
        }  
    }  
    //链接超限,直接退出  
    if (connectionCount > _maxNumConnections) {   
        ......  
        return;  
    }  
    //链接关闭的回收处理  
    ssm->setCleanupHook([ this, ssmIt, session = std::move(session) ] {  
         ......  
    });  
    //获取transport模式为同步模式还是异步模式，也就是adaptive线程模式还是synchronous线程模式  
    auto ownership = ServiceStateMachine::Ownership::kOwned;  
    if (transportMode == transport::Mode::kSynchronous) {  
        ownership = ServiceStateMachine::Ownership::kStatic;  
    }  
    //ServiceStateMachine::start，这里和服务状态机模块衔接起来  
    ssm->start(ownership);  
}  

Status launchServiceWorkerThread(stdx::function<void()> task) {  
        static const size_t kStackSize = 1024 * 1024;  //1M  
        struct rlimit limits;  
        //或者系统堆栈大小  
        invariant(getrlimit(RLIMIT_STACK, &limits) == 0);  
        //如果系统堆栈大小大于1M，则默认设置线程栈大小为1M  
        if (limits.rlim_cur > kStackSize) {  
            size_t stackSizeToSet = kStackSize;  
            int failed = pthread_attr_setstacksize(&attrs, stackSizeToSet);  
            if (failed) {  
                const auto ewd = errnoWithDescription(failed);  
                warning() << "pthread_attr_setstacksize failed: " << ewd;  
            }  
        } else if (limits.rlim_cur < 1024 * 1024) {  
            //如果系统栈大小小于1M，则已系统堆栈为准，同时给出告警  
            warning() << "Stack size set to " << (limits.rlim_cur / 1024) << "KB. We suggest 1MB";  
        }}  
        ......  
        //task参数传递给新建线程  
        auto ctx = stdx::make_unique<stdx::function<void()>>(std::move(task));  
        int failed = pthread_create(&thread, &attrs, runFunc, ctx.get());   
        ......  
} 

//下面的ASIOSinkTicket和ASIOSourceTicket继承该类,用于控制数据的发送和接收  
class TransportLayerASIO::ASIOTicket : public TicketImpl {  
public:  
    //初始化构造  
    explicit ASIOTicket(const ASIOSessionHandle& session, Date_t expiration);  
    //获取sessionId  
    SessionId sessionId() const final {  
        return _sessionId;  
    }  
    //asio模式没用，针对legacy模型  
    Date_t expiration() const final {  
        return _expiration;  
    }  

    //以下四个接口用于数据收发相关处理  
    void fill(bool sync, TicketCallback&& cb);  
protected:  
    void finishFill(Status status);  
    bool isSync() const;  
    virtual void fillImpl() = 0;  
private:  
    //会话信息，一个链接一个session  
    std::weak_ptr<ASIOSession> _session;  
    //每个session有一个唯一id  
    const SessionId _sessionId;  
    //asio模型没用，针对legacy生效  
    const Date_t _expiration;  
    //数据发送或者接收成功后的回调处理  
    TicketCallback _fillCallback;  
    //同步方式还是异步方式进行数据处理，默认异步  
    bool _fillSync;  
}; 

//数据接收的ticket  
class TransportLayerASIO::ASIOSourceTicket : public TransportLayerASIO::ASIOTicket {  
public:  
    //初始化构造  
    ASIOSourceTicket(const ASIOSessionHandle& session, Date_t expiration, Message* msg);  
protected:  
    //数据接收Impl  
    void fillImpl() final;  
private:  
    //接收到MongoDB头部数据后的回调处理  
    void _headerCallback(const std::error_code& ec, size_t size);  
    //接收到MongoDB包体数据后的回调处理    
    void _bodyCallback(const std::error_code& ec, size_t size);  
  
    //存储数据的buffer，网络IO读取到的原始数据内容  
    SharedBuffer _buffer;  
    //数据Message管理，数据来源为_buffer  
    Message* _target;  
};  

  
//数据发送的ticket  
class TransportLayerASIO::ASIOSinkTicket : public TransportLayerASIO::ASIOTicket {  
 public:  
    //初始化构造  
    ASIOSinkTicket(const ASIOSessionHandle& session, Date_t expiration, const Message& msg);  
protected:  
    //数据发送Impl  
    void fillImpl() final;  
private:  
    //发送数据完成的回调处理  
    void _sinkCallback(const std::error_code& ec, size_t size);  
    //需要发送的数据message信息  
    Message _msgToSend;  
}; 

//数据接收的fillImpl接口实现  
void TransportLayerASIO::ASIOSourceTicket::fillImpl() {    
    //获取对应session信息  
    auto session = getSession();  
    if (!session)  
        return;  
    //收到读取MongoDB头部数据，头部数据长度是固定的kHeaderSize字节  
    const auto initBufSize = kHeaderSize;  
    _buffer = SharedBuffer::allocate(initBufSize);  
  
    //调用TransportLayerASIO::ASIOSession::read读取底层数据存入_buffer  
    //读完头部数据后执行对应的_headerCallback回调函数  
    session->read(isSync(),  
                  asio::buffer(_buffer.get(), initBufSize), //先读取头部字段出来  
                  [this](const std::error_code& ec, size_t size) { _headerCallback(ec, size); });  
}  
  
//读取到MongoDB header头部信息后的回调处理  
void TransportLayerASIO::ASIOSourceTicket::_headerCallback(const std::error_code& ec, size_t size) {  
    ......  
    //获取session信息  
    auto session = getSession();  
    if (!session)  
        return;  
    //从_buffer中获取头部信息  
    MSGHEADER::View headerView(_buffer.get());  
    //获取message长度  
    auto msgLen = static_cast<size_t>(headerView.getMessageLength());  
    //长度太小或者太大，直接报错  
    if (msgLen < kHeaderSize || msgLen > MaxMessageSizeBytes) {  
        .......  
        return;  
    }  
    ....  
   //内容还不够一个mongo协议报文，继续读取body长度字节的数据，读取完毕后开始body处理  
   //注意这里是realloc，保证头部和body在同一个buffer中  
    _buffer.realloc(msgLen);   
    MsgData::View msgView(_buffer.get());  
  
    //调用底层TransportLayerASIO::ASIOSession::read读取数据body   
    session->read(isSync(),  
      //数据读取到该buffer                  
      asio::buffer(msgView.data(), msgView.dataLen()),  
      //读取成功后的回调处理  
      [this](const std::error_code& ec, size_t size) { _bodyCallback(ec, size); });  
}  
  
//_headerCallback对header读取后解析header头部获取到对应的msg长度，然后开始body部分处理  
void TransportLayerASIO::ASIOSourceTicket::_bodyCallback(const std::error_code& ec, size_t size) {  
    ......  
    //buffer转存到_target中  
    _target->setData(std::move(_buffer));  
    //流量统计  
    networkCounter.hitPhysicalIn(_target->size());  
    //TransportLayerASIO::ASIOTicket::finishFill    
    finishFill(Status::OK()); //包体内容读完后，开始下一阶段的处理    
    //报文读取完后的下一阶段就是报文内容处理，开始走ServiceStateMachine::_processMessage  
}  

//发送数据成功后的回调处理  
void TransportLayerASIO::ASIOSinkTicket::_sinkCallback(const std::error_code& ec, size_t size) {  
    //发送的网络字节数统计  
    networkCounter.hitPhysicalOut(_msgToSend.size());   
    //执行SSM中对应的状态流程  
    finishFill(ec ? errorCodeToStatus(ec) : Status::OK());  
}  
  
//发送数据的fillImpl  
void TransportLayerASIO::ASIOSinkTicket::fillImpl() {  
    //获取对应session  
    auto session = getSession();  
    if (!session)  
        return;  
  
    //调用底层TransportLayerASIO::ASIOSession::write发送数据，发送成功后执行_sinkCallback回调   
    session->write(isSync(),  
       asio::buffer(_msgToSend.buf(), _msgToSend.size()),  
       //发送数据成功后的callback回调  
       [this](const std::error_code& ec, size_t size) { _sinkCallback(ec, size); });  
}  

class TransportLayerASIO::ASIOSession : public Session {  
    //初始化构造  
    ASIOSession(TransportLayerASIO* tl, GenericSocket socket);  
    //获取本session使用的tl  
    TransportLayer* getTransportLayer();  
    //以下四个接口套接字相关，本端/对端地址获取，获取fd,关闭fd等  
    const HostAndPort& remote();  
    const HostAndPort& local();  
    GenericSocket& getSocket();  
    void shutdown();  
  
    //以下四个接口调用asio网络库实现数据的同步收发和异步收发  
    void read(...)  
    void write(...)  
    void opportunisticRead(...)  
    void opportunisticWrite(...)  
  
    //远端地址信息  
    HostAndPort _remote;  
    //本段地址信息  
    HostAndPort _local;  
    //赋值见TransportLayerASIO::_acceptConnection  
    //也就是fd，一个新连接对应一个_socket  
    GenericSocket _socket;  
    //SSL相关不做分析，  
#ifdef MONGO_CONFIG_SSL  
    boost::optional<asio::ssl::stream<decltype(_socket)>> _sslSocket;  
    bool _ranHandshake = false;  
#endif  
  
    //本套接字对应的tl，赋值建TransportLayerASIO::_acceptConnection(...)  
    TransportLayerASIO* const _tl;  
} 

//初始化构造 TransportLayerASIO::_acceptConnection调用  
ASIOSession(TransportLayerASIO* tl, GenericSocket socket)  
    //fd描述符及TL初始化赋值  
    : _socket(std::move(socket)), _tl(tl) {  
    std::error_code ec;  
  
    //异步方式设置为非阻塞读  
    _socket.non_blocking(_tl->_listenerOptions.transportMode == Mode::kAsynchronous, ec);  
    fassert(40490, ec.value() == 0);  
  
    //获取套接字的family  
    auto family = endpointToSockAddr(_socket.local_endpoint()).getType();  
    //满足AF_INET
    if (family == AF_INET || family == AF_INET6) {  
        //no_delay keep_alive套接字系统参数设置  
        _socket.set_option(asio::ip::tcp::no_delay(true));  
        _socket.set_option(asio::socket_base::keep_alive(true));  
        //KeepAlive系统参数设置  
        setSocketKeepAliveParams(_socket.native_handle());  
    }  
  
    //获取本端和对端地址  
    _local = endpointToHostAndPort(_socket.local_endpoint());  
    _remote = endpointToHostAndPort(_socket.remote_endpoint(ec));  
    if (ec) {  
        LOG(3) << "Unable to get remote endpoint address: " << ec.message();  
    }  
}  

//读取指定长度数据，然后执行handler回调  
void opportunisticRead(...) {  
    std::error_code ec;  
    //如果是异步线程模型，在ASIOSession构造初始化的时候会设置non_blocking非阻塞模式  
    //异步线程模型这里实际上是非阻塞读取，如果是同步线程模型，则没有non_blocking设置，也就是阻塞读取  
    auto size = asio::read(stream, buffers, ec);    
  
    //如果是异步读，并且read返回would_block或者try_again说明指定长度的数据还没有读取完毕  
    if ((ec == asio::error::would_block || ec == asio::error::try_again) && !sync) {  
        //buffers有大小size，实际读最多读size字节  
        MutableBufferSequence asyncBuffers(buffers);  
        if (size > 0) {  
            asyncBuffers += size; //buffer offset向后移动  
        }  
  
        //继续异步方式读取数据，读取到指定长度数据后执行handler回调处理  
        asio::async_read(stream, asyncBuffers, std::forward<CompleteHandler>(handler));  
    } else {   
        //阻塞方式读取read返回后可以保证读取到了size字节长度的数据  
        //直接read获取到size字节数据，则直接执行handler   
        handler(ec, size);  
    }  
}  

//发送数据  
void opportunisticWrite(...) {  
    std::error_code ec;  
    //如果是同步模式，则阻塞写，直到全部写成功。异步模式则非阻塞写  
    auto size = asio::write(stream, buffers, ec);   
  
    //异步写如果返回try_again说明数据还没有发送完，则继续异步写发送  
    if ((ec == asio::error::would_block || ec == asio::error::try_again) && !sync) {  
        ConstBufferSequence asyncBuffers(buffers);  
        if (size > 0) {  //buffer中数据指针偏移计数
            asyncBuffers += size;  
        }  
        //异步写发送完成，执行handler回调  
        asio::async_write(stream, asyncBuffers, std::forward<CompleteHandler>(handler));  
    } else {  
        //同步写成功，则直接执行handler回调处理  
        handler(ec, size);  
    }  
}