MaNGOS-Zero源碼學習之mangosd游戲邏輯服務器（二）：socket的處理方式

    本篇將介紹客戶端與游戲邏輯服務器連接建立以后，mangosd如何接收、解析和處理客戶端發過來的協議。本篇不再討論mangosd與客戶端的認證及建立最終RC4流加密的過程，想了解這部分內容請看該系列的第一篇。

 

 

一、acceptor socket的監聽啟動及注冊

    mangosd的main ()函數調用單例對象sMaster的Run ()函數，啟動監聽socket的代碼如下：

 

int Master::Run()
{
    ........

    ///- Launch the world listener socket
    uint16 wsport = sWorld.getConfig (CONFIG_UINT32_PORT_WORLD);
    std::string bind_ip = sConfig.GetStringDefault ("BindIP", "0.0.0.0");

    if (sWorldSocketMgr->StartNetwork (wsport, bind_ip) == -1)
    {
        sLog.outError ("Failed to start network");
        Log::WaitBeforeContinueIfNeed();
        World::StopNow(ERROR_EXIT_CODE);
        // go down and shutdown the server
    }

    sWorldSocketMgr->Wait ();

    ........
}

   

    mangosd用WorldSocketMgr類來管理socket。StartNetwork ()會調用StartReactiveIO ()來啟動監聽socket，處理代碼如下：

int WorldSocketMgr::StartReactiveIO (ACE_UINT16 port, const char* address)
{
    ........

    //（1）
    m_NetThreadsCount = static_cast<size_t> (num_threads + 1);
    m_NetThreads = new ReactorRunnable[m_NetThreadsCount];

    // -1 means use default
    m_SockOutKBuff = sConfig.GetIntDefault ("Network.OutKBuff", -1);
    m_SockOutUBuff = sConfig.GetIntDefault ("Network.OutUBuff", 65536);
    if ( m_SockOutUBuff <= 0 )
    {
        sLog.outError ("Network.OutUBuff is wrong in your config file");
        return -1;
    }

    //（2）
    WorldSocket::Acceptor *acc = new WorldSocket::Acceptor;
    m_Acceptor = acc;

    ACE_INET_Addr listen_addr (port, address);
    if (acc->open (listen_addr, m_NetThreads[0].GetReactor (), ACE_NONBLOCK) == -1)
    {
        sLog.outError ("Failed to open acceptor ,check if the port is free");
        return -1;
    }

    //（3）
    for (size_t i = 0; i < m_NetThreadsCount; ++i)
        m_NetThreads[i].Start ();

    return 0;
}

（1）ReactorRunnable類繼承了ACE_Task_Base，ACE_Task_Base 是ACE 中的任務或主動對象“處理結構”的基類。在ACE 中使用了此類來實現主動對象模式。所有希望成為“主動對象”的對象都必須從此類派生。可以把ACE_TASK 看作是更高級的、更為面向對象的線程類[1]。ACE_Task_Base調用時繼承類必須重寫svc方法，並且在使用時保證調用了activate ()方法。

（2）指定監聽地址，端口並把Acceptor綁定到第一個線程的Reactor上。啟動Acceptor開始監聽網絡IO。

（3）啟動所有線程，每個線程上有一個單獨的ACE_Reactor* m_Reactor;，這里的Reactor使用的是多線程的ACE_TP_Reactor。可以各自單獨完成事件的多路復用。

 

 

二、線程體函數

    線程體函數ReactorRunnable::svc () 如下：

virtual int svc ()
{
    //（1）
    WorldDatabase.ThreadStart ();

    SocketSet::iterator i, t;
    while (!m_Reactor->reactor_event_loop_done ())
    {
        // dont be too smart to move this outside the loop
        // the run_reactor_event_loop will modify interval
        ACE_Time_Value interval (0, 10000);

        //（2）
        if (m_Reactor->run_reactor_event_loop (interval) == -1)
            break;

        //（3）
        AddNewSockets ();

        for (i = m_Sockets.begin (); i != m_Sockets.end ();)
        {
            //（4）
            if ((*i)->Update () == -1)
            {
                t = i;
                ++i;
                (*t)->CloseSocket ();
                (*t)->RemoveReference ();
                --m_Connections;
                m_Sockets.erase (t);
            }
            else
                ++i;
        }
    }

    WorldDatabase.ThreadEnd ();
    DEBUG_LOG ("Network Thread Exitting");

    return 0;
}

（1）會調用mysql_thread_init ()函數，初始化與該線程相關的變量。

（2）run_reactor_event_loop ()函數為多路復用的等待函數，當注冊的事件發生、運行超時或者出現錯誤時返回。

（3）AddNewSockets ()函數會將緩存在m_NewSockets里的新到達的socket添加到SocketSet m_Sockets;里，同時檢查並處理m_Sockets;里已經closed的socket。

（4）循環每一個WorldSocket，調用其Update ()方法，這里只處理每個socket的handle_output，即每個在此線程上的寫事件，向客戶端發送數據。下一節詳細介紹：

 

 

三、WorldSocket::Update () 方法

    Update方法用於處理每個socket的輸出：

int WorldSocket::Update (void)
{
    if (closing_)
        return -1;

    //（1）
    if (m_OutActive || m_OutBuffer->length () == 0)
        return 0;

    return handle_output (get_handle ());
}

（1）m_OutBuffer有數據時才會調用handle_output，handle_output ()用於處理輸出，如果輸出不能一次性做完，會調用schedule_wakeup_output ()再次激活write事件。當輸出處理完畢后則調用cancel_wakeup_output ()取消激活write事件，使reactor恢復到正常的loop ()循環中。詳細過程如下：

 

int WorldSocket::handle_output (ACE_HANDLE)
{
    //（1）
    ACE_GUARD_RETURN (LockType, Guard, m_OutBufferLock, -1);

    if (closing_)
        return -1;

    const size_t send_len = m_OutBuffer->length ();
    if (send_len == 0)
        return cancel_wakeup_output (Guard);

#ifdef MSG_NOSIGNAL
    ssize_t n = peer ().send (m_OutBuffer->rd_ptr (), send_len, MSG_NOSIGNAL);
#else
    ssize_t n = peer ().send (m_OutBuffer->rd_ptr (), send_len);
#endif // MSG_NOSIGNAL

    if (n == 0)
        return -1;
    else if (n == -1)
    {
        if (errno == EWOULDBLOCK || errno == EAGAIN) //----------（2）
            return schedule_wakeup_output (Guard);

        return -1;
    }
    else if (n < (ssize_t)send_len) //now n > 0      //----------（3）
    {
        m_OutBuffer->rd_ptr (static_cast<size_t> (n));

        // move the data to the base of the buffer
        m_OutBuffer->crunch ();

        return schedule_wakeup_output (Guard);
    }
    else //now n == send_len                         //----------（4）
    {
        m_OutBuffer->reset ();

        if (!iFlushPacketQueue ())
            return cancel_wakeup_output (Guard);
        else
            return schedule_wakeup_output (Guard);
    }

    ACE_NOTREACHED (return 0);
}

（1）對m_OutBuffer加鎖。

（2）考慮信號打斷的情況等，暫時不能寫。

（3）只發送了部分數據則繼續wakeup該線程對應的Reactor。

（4）檢查m_OutBuffer數據發送完畢同時等待buffer（PacketQueueT m_PacketQueue;）里已經沒有數據時，cancel wakeup讓Reactor恢復正常。

 

 

四、socket到對應線程的指派

    上一節內容可以看到線程內如何處理socket，及新到達的socket。但從第一節中可知只有第一個線程注冊為acceptor線程，那么新連接到達時，是如何被指派到對應的“接待”線程的呢？

    可以先看一下ACE_Acceptor的處理時序圖：

圖3.1 連接到達處理時序 [2]

   

    上圖可以看出，當連接到達時，acceptor會調用對應的SVC_HANDLER的open ()函數，在mangosd里就是acceptor對應的int WorldSocket::open (void *a)，如下：

int WorldSocket::open (void *a)
{
    ........

    // Hook for the manager.
    if (sWorldSocketMgr->OnSocketOpen (this) == -1)
        return -1;

    ........
}

 

OnSocketOpen方法：

int WorldSocketMgr::OnSocketOpen (WorldSocket* sock)
{
    ........

    // we skip the Acceptor Thread
    size_t min = 1;

    MANGOS_ASSERT (m_NetThreadsCount >= 1);

    //（1）
    for (size_t i = 1; i < m_NetThreadsCount; ++i)
        if (m_NetThreads[i].Connections () < m_NetThreads[min].Connections ())
            min = i;

    return m_NetThreads[min].AddSocket (sock);
}

（1）將WorldSocket均衡的分配給每個線程。AddSocket ()將socket添加到m_NewSockets中做緩存，待該線程自行調用AddNewSockets ()添加到處理隊列里。

 

  

總結：

    mangosd對socket的處理因為使用了ACE，邏輯處理代碼相對比較簡單，寫事件的異常處理主要涉及

（1）一次不能寫完則不斷的wakeup Reactor。

（2）信號中斷等錯誤的判斷。似乎這里並沒有考慮全面（見附錄）

（3）使用另一個buffer緩存，因寫緩存m_OutBuffer滿而帶來的多出的數據。

 

References：

[1] http://blog.csdn.net/yecao_kinux/article/details/1546914

[2] http://postfiles12.naver.net/data41/2009/4/11/187/33_kbkim007.jpg?type=w3

 

 

附錄：

不同平台不同版本的read/write、send/recv有些不同，挺郁悶的一件事……………

EWOULDBLOCK：基於 berkeley 實現，當客戶端異常終止連接時

ECONNABORTED：基於 posix 實現，當客戶端異常終止連接時

EPROTO：基於 SVR4 實現，當客戶端異常終止連接時）以及 EINTR