微服务开源框架TARS的RPC源码解析 之 初识TARS C++服务端

g_app.main(argc, argv);
g_app.waitForShutdown();

void Application::main(int argc, char *argv[])
{
	......
        //初始化Server部分
        initializeServer();
	......
}

TC_EpollServer::TC_EpollServer(unsigned int iNetThreadNum)
{
    if(_netThreadNum < 1)
    {
        _netThreadNum = 1;
    }
    //网络线程的配置数目不能15个
    if(_netThreadNum > 15)
    {
        _netThreadNum = 15;
    }
 
    for (size_t i = 0; i < _netThreadNum; ++i)
    {
        TC_EpollServer::NetThread* netThreads = new TC_EpollServer::NetThread(this);
        _netThreads.push_back(netThreads);
    }
}

void Application::main(int argc, char *argv[])
{
	......
        vector<TC_EpollServer::BindAdapterPtr> adapters;
        //绑定对象和端口
        bindAdapter(adapters);
	......
        _epollServer->createEpoll();
}
 
void Application::waitForShutdown()
{
    waitForQuit();
    ......
}

vector<TC_EpollServer::BindAdapterPtr> adapters;
//绑定对象和端口
bindAdapter(adapters);

BindAdapterPtr bindAdapter = new BindAdapter(_epollServer.get());
_epollServer->bind(bindAdapter);

int  TC_EpollServer::bind(TC_EpollServer::BindAdapterPtr &lsPtr)
{
    int iRet = 0;
    for(size_t i = 0; i < _netThreads.size(); ++i)
    {
        if(i == 0)
        {
            iRet = _netThreads[i]->bind(lsPtr);
        }
        else
        {
            //当网络线程中listeners没有监听socket时，list使用adapter中设置的最大连接数作为初始化
            _netThreads[i]->setListSize(lsPtr->getMaxConns());
        }
    }
    return iRet;
}

void TC_EpollServer::NetThread::bind(const TC_Endpoint &ep, TC_Socket &s)
{
    int type = ep.isUnixLocal()?AF_LOCAL:AF_INET;
 
    if(ep.isTcp())
    {
        s.createSocket(SOCK_STREAM, type);
    }
    else
    {
        s.createSocket(SOCK_DGRAM, type);
    }
 
    if(ep.isUnixLocal())
    {
        s.bind(ep.getHost().c_str());
    }
    else
    {
        s.bind(ep.getHost(), ep.getPort());
    }
 
    if(ep.isTcp() && !ep.isUnixLocal())
    {
        s.listen(1024);
        s.setKeepAlive();
        s.setTcpNoDelay();
        //不要设置close wait否则http服务回包主动关闭连接会有问题
        s.setNoCloseWait();
    }
    s.setblock(false);
}

_listeners[s.getfd()] = lsPtr;

_epollServer->createEpoll();

void TC_EpollServer::createEpoll()
{
    for(size_t i = 0; i < _netThreads.size(); ++i)
    {
        _netThreads[i]->createEpoll(i+1);
    }
    //必须先等所有网络线程调用createEpoll()，初始化list后，才能调用initUdp()
    for(size_t i = 0; i < _netThreads.size(); ++i)
    {
        _netThreads[i]->initUdp();
    }
}

void Application::main(int argc, char *argv[])
{
	......
 
        vector<TC_EpollServer::BindAdapterPtr> adapters;
        bindAdapter(adapters);
        //业务应用的初始化
        initialize();
 
 
        //设置HandleGroup分组，启动线程
        for (size_t i = 0; i < adapters.size(); ++i)
        {
            string name = adapters[i]->getName();
 
            string groupName = adapters[i]->getHandleGroupName();
 
            if(name != groupName)
            {
                TC_EpollServer::BindAdapterPtr ptr = _epollServer->getBindAdapter(groupName);
 
                if (!ptr)
                {
                    throw runtime_error("[TARS][adater `" + name + "` setHandle to group `" + groupName + "` fail!");
                }
 
            }
            setHandle(adapters[i]);
        }
 
        //启动业务处理线程
        _epollServer->startHandle();
	......
 
}

for (size_t i = 0; i < adapterName.size(); i++)
{
	……
     string servant = _conf.get("/tars/application/server/" + adapterName[i] + "<servant>");
     checkServantNameValid(servant, sPrefix);
 
     ServantHelperManager::getInstance()->setAdapterServant(adapterName[i], servant);
	……
}

void ServantHelperManager::setAdapterServant(const string &sAdapter, const string &sServant)
{
    _adapter_servant[sAdapter] = sServant;
    _servant_adapter[sServant] = sAdapter;
}

    /**
     * Adapter包含的Servant(Adapter名称:servant名称)
     */
    map<string, string>                     _adapter_servant;
 
    /**
     * Adapter包含的Servant(Servant名称:Adapter名称)
     */
map<string, string>                     _servant_adapter;

void
StringServer::initialize()
{
    //initialize application here:
    //...
    addServant<StringServantImp>(ServerConfig::Application + "." + ServerConfig::ServerName + ".StringServantObj");
}

    template<typename T>
    void addServant(const string &id,bool check = false)
    {
        if(check && _servant_adapter.end() == _servant_adapter.find(id))
        {
            cerr<<"[TARS]ServantHelperManager::addServant "<< id <<" not find adapter.(maybe not conf in the web)"<<endl;
            throw runtime_error("[TARS]ServantHelperManager::addServant " + id + " not find adapter.(maybe not conf in the web)");
        }
        _servant_creator[id] = new ServantCreation<T>();
}

/**
 * Servant
 */
class ServantHelperCreation : public TC_HandleBase
{
public:
    virtual ServantPtr create(const string &s) = 0;
};
typedef TC_AutoPtr<ServantHelperCreation> ServantHelperCreationPtr;
 
//////////////////////////////////////////////////////////////////////////////
/**
 * Servant
 */
template<class T>
struct ServantCreation : public ServantHelperCreation
{
    ServantPtr create(const string &s) { T *p = new T; p->setName(s); return p; }
};

//设置HandleGroup分组，启动线程
for (size_t i = 0; i < adapters.size(); ++i)
{
    string name = adapters[i]->getName();
    string groupName = adapters[i]->getHandleGroupName();
 
    if(name != groupName)
    {
        TC_EpollServer::BindAdapterPtr ptr = _epollServer->getBindAdapter(groupName);
 
        if (!ptr)
        {
            throw runtime_error("[TARS][adater `" + name + "` setHandle to group `" + groupName + "` fail!");
        }
 
    }
    setHandle(adapters[i]);
}

void Application::setHandle(TC_EpollServer::BindAdapterPtr& adapter)
{
    adapter->setHandle<ServantHandle>();
}

template<typename T> void setHandle()
{
    _pEpollServer->setHandleGroup<T>(_handleGroupName, _iHandleNum, this);
}

/**
 * 创建一个handle对象组，如果已经存在则直接返回
 * @param name
 * @return HandlePtr
 */
template<class T> void setHandleGroup(const string& groupName, int32_t handleNum, BindAdapterPtr adapter)
{
    map<string, HandleGroupPtr>::iterator it = _handleGroups.find(groupName);
 
    if (it == _handleGroups.end())
    {
        HandleGroupPtr hg = new HandleGroup();
        hg->name = groupName;
        adapter->_handleGroup = hg;
 
        for (int32_t i = 0; i < handleNum; ++i)
        {
            HandlePtr handle = new T();
            handle->setEpollServer(this);
            handle->setHandleGroup(hg);
            hg->handles.push_back(handle);
        }
 
        _handleGroups[groupName] = hg;
        it = _handleGroups.find(groupName);
    }
    it->second->adapters[adapter->getName()] = adapter;
    adapter->_handleGroup = it->second;
}

//启动业务处理线程
_epollServer->startHandle();

void TC_EpollServer::startHandle()
{
    if (!_handleStarted)
    {
        _handleStarted = true;
 
        for (auto& kv : _handleGroups)
        {
            auto& hds = kv.second->handles;
            for (auto& handle : hds)
            {
                if (!handle->isAlive())
                    handle->start();
            }
        }
    }
}

void TC_EpollServer::Handle::run()
{
    initialize();
 
    handleImp();
}

void ServantHandle::initialize()
{
    map<string, TC_EpollServer::BindAdapterPtr>::iterator adpit;
    // 获取本Handle所关联的BindAdapter
    map<string, TC_EpollServer::BindAdapterPtr>& adapters = _handleGroup->adapters;
    // 遍历所有BindAdapter
    for (adpit = adapters.begin(); adpit != adapters.end(); ++adpit)
    {
        // 借助ServantHelperManager来获取服务指针——XXXServantImp类的指针
        ServantPtr servant = ServantHelperManager::getInstance()->create(adpit->first);
        // 将指针放进map<string, ServantPtr> ServantHandle:: _servants中
        if (servant)
        {
            _servants[servant->getName()] = servant;
        }
        else
        {
            TLOGERROR("[TARS]ServantHandle initialize createServant ret null, for adapter `" + adpit->first + "`" << endl);
        }
    }
 
    ......
}

void TC_EpollServer::Handle::handleImp()
{
    ......
    struct timespec ts;
 
    while (!getEpollServer()->isTerminate())
    {
        {
            TC_ThreadLock::Lock lock(_handleGroup->monitor);
 
            if (allAdapterIsEmpty() && allFilterIsEmpty())
            {
                _handleGroup->monitor.timedWait(_iWaitTime);
            }
        }
}
......
}

_epoller.add(kv.first, H64(ET_LISTEN) | kv.first, EPOLLIN);

try
{
    const epoll_event &ev = _epoller.get(i);
    uint32_t h = ev.data.u64 >> 32;
 
    switch(h)
    {
    case ET_LISTEN:
        {
            //监听端口有请求
            auto it = _listeners.find(ev.data.u32);
            if( it != _listeners.end())
            {
                if(ev.events & EPOLLIN)
                {
                    bool ret;
                    do
                    {
                        ret = accept(ev.data.u32);
                    }while(ret);
                }
            }
        }
        break;
    case ET_CLOSE:
        //关闭请求
        break;
    case ET_NOTIFY:
        //发送通知
        ......
        break;
     case ET_NET:
        //网络请求
         ......
        break;
      default:
         assert(true);
      }
}

//接收连接
TC_Socket s;
s.init(fd, false, AF_INET);
int iRetCode = s.accept(cs, (struct sockaddr *) &stSockAddr, iSockAddrSize);

cs.setblock(false);
cs.setKeepAlive();
cs.setTcpNoDelay();
cs.setCloseWaitDefault();

int timeout = _listeners[fd]->getEndpoint().getTimeout()/1000;
 
Connection *cPtr = new Connection(_listeners[fd].get(), fd, (timeout < 2 ? 2 : timeout), cs.getfd(), ip, port);

// 服务连接
Connection::Connection(TC_EpollServer::BindAdapter *pBindAdapter, int lfd, int timeout, int fd, const string& ip, uint16_t port)
{
    ......
    _sock.init(fd, true, AF_INET);
}

//addTcpConnection(cPtr);
_epollServer->addConnection(cPtr, cs.getfd(), TCP_CONNECTION);

void TC_EpollServer::addConnection(TC_EpollServer::NetThread::Connection * cPtr, int fd, int iType)
{
    TC_EpollServer::NetThread* netThread = getNetThreadOfFd(fd);
 
    if(iType == 0)
    {
        netThread->addTcpConnection(cPtr);
    }
    else
    {
        netThread->addUdpConnection(cPtr);
    }
}

NetThread* getNetThreadOfFd(int fd)
{
    return _netThreads[fd % _netThreads.size()];
}

try
{
    const epoll_event &ev = _epoller.get(i);
    uint32_t h = ev.data.u64 >> 32;
 
    switch(h)
    {
    case ET_LISTEN:
        ……
        break;
    case ET_CLOSE:
        //关闭请求
        break;
    case ET_NOTIFY:
        //发送通知
        ......
        break;
     case ET_NET:
        //网络请求
        processNet(ev);
        break;
      default:
         assert(true);
      }
}

void TC_EpollServer::NetThread::processNet(const epoll_event &ev)
{
    uint32_t uid = ev.data.u32;
 
    Connection *cPtr = getConnectionPtr(uid);
    ......
}

if(ev.events & EPOLLIN)               //有数据需要读取
{
    recv_queue::queue_type vRecvData;
    int ret = recvBuffer(cPtr, vRecvData);
 
    if(ret < 0)
    {
        delConnection(cPtr,true,EM_CLIENT_CLOSE);
        return;
    }
 
    if(!vRecvData.empty())
    {
        cPtr->insertRecvQueue(vRecvData);
    }
}

int  NetThread::recvBuffer(NetThread::Connection *cPtr, recv_queue::queue_type &v)
{
    return cPtr->recv(v);
}

iBytesReceived = ::read(_sock.getfd(), (void*)buffer, sizeof(buffer))

tagRecvData* recv = new tagRecvData();
recv->buffer           = std::move(ro);
recv->ip               = _ip;
recv->port             = _port;
recv->recvTimeStamp    = TNOWMS;
recv->uid              = getId();
recv->isOverload       = false;
recv->isClosed         = false;
recv->fd               = getfd();
//收到完整的包才算
this->_bEmptyConn = false;
 
//收到完整包
o.push_back(recv);

void NetThread::processNet(const epoll_event &ev)
{
    ......
 
    if(ev.events & EPOLLIN)               //有数据需要读取
    {
        ......
 
        if(!vRecvData.empty())
        {
            cPtr->insertRecvQueue(vRecvData);
        }
    }
    ......
}

_rbuffer.push_back(vtRecvData)

_handleGroup->monitor.notify()

bool BindAdapter::waitForRecvQueue(tagRecvData* &recv, uint32_t iWaitTime)
{
    bool bRet = false;
 
    bRet = _rbuffer.pop_front(recv, iWaitTime);
 
    if(!bRet)
    {
        return bRet;
    }
 
    return bRet;
}

void ServantHandle::handle(const TC_EpollServer::tagRecvData &stRecvData)
{
    TarsCurrentPtr current = createCurrent(stRecvData);
 
......
}

void ServantHandle::handle(const TC_EpollServer::tagRecvData &stRecvData)
{
    // 构造请求上下文
    TarsCurrentPtr current = createCurrent(stRecvData);
 
    if (!current) return;
    // 处理请求
    if (current->getBindAdapter()->isTarsProtocol())
    {
        handleTarsProtocol(current);
    }
    else
    {
        handleNoTarsProtocol(current);
    }
}

void ServantHandle::handleTarsProtocol(const TarsCurrentPtr &current)
{
    // 1-对请求上下文current进行预处理
    // 2-寻找合适的服务servant
    map<string, ServantPtr>::iterator sit = _servants.find(current->getServantName());
 
    if (sit == _servants.end())
    {
        current->sendResponse(TARSSERVERNOSERVANTERR);
 
        return;
    }
 
    int ret = TARSSERVERUNKNOWNERR;
 
    string sResultDesc = "";
 
    vector<char> buffer;
 
    try
    {
        //3-业务逻辑处理
        ret = sit->second->dispatch(current, buffer);
    }
    catch(TarsDecodeException &ex)
    {
       ……
    }
    catch(TarsEncodeException &ex)
    {
        ……
    }
    catch(exception &ex)
    {
       ……
    }
    catch(...)
    {
        ……
    }
//回送响应，第3小点再分析吧
……
}

int Servant::dispatch(TarsCurrentPtr current, vector<char> &buffer)
{
    int ret = TARSSERVERUNKNOWNERR;
 
    if (current->getFuncName() == "tars_ping")
    {
        //略
    }
    else if (!current->getBindAdapter()->isTarsProtocol())
    {
        //略
    }
    else
    {
        TC_LockT<TC_ThreadRecMutex> lock(*this);
 
        ret = onDispatch(current, buffer);
    }
    return ret;
}

void ServantHandle::handleTarsProtocol(const TarsCurrentPtr &current)
{
    // 1-对请求上下文current进行预处理
    // 2-寻找合适的服务servant
    //3-业务逻辑处理
//回送响应，本节分析
    if (current->isResponse())
    {
        current->sendResponse(ret, buffer, TarsCurrent::TARS_STATUS(), sResultDesc);
    }
}

void TC_EpollServer::NetThread::send(uint32_t uid, const string &s, const string &ip, uint16_t port)
{
    if(_bTerminate)
    {
        return;
    }
 
    tagSendData* send = new tagSendData();
 
    send->uid = uid;
    send->cmd = 's';
    send->buffer = s;
    send->ip = ip;
send->port = port;
 
    _sbuffer.push_back(send);
 
    //通知epoll响应, 有数据要发送
    _epoller.mod(_notify.getfd(), H64(ET_NOTIFY), EPOLLOUT);
}

try
{
    const epoll_event &ev = _epoller.get(i);
    uint32_t h = ev.data.u64 >> 32;
 
    switch(h)
    {
    case ET_LISTEN:
        ……
        break;
    case ET_CLOSE:
        //关闭请求
        break;
    case ET_NOTIFY:
        //发送通知
        processPipe();
        break;
     case ET_NET:
        //网络请求
        ……
        break;
      default:
         assert(true);
      }
}