本篇將介紹客戶端與游戲邏輯服務器連接建立以后,mangosd如何接收、解析和處理客戶端發過來的協議。本篇不再討論mangosd與客戶端的認證及建立最終RC4流加密的過程,想了解這部分內容請看該系列的
第一篇。
一、acceptor socket的監聽啟動及注冊
mangosd的main ()函數調用單例對象sMaster的Run ()函數,啟動監聽socket的代碼如下:
1: int Master::Run()
2: {
3: ........
4:
5: ///- Launch the world listener socket
6: uint16 wsport = sWorld.getConfig (CONFIG_UINT32_PORT_WORLD);
7: std::string bind_ip = sConfig.GetStringDefault ("BindIP", "0.0.0.0");
8:
9: if (sWorldSocketMgr->StartNetwork (wsport, bind_ip) == -1)
10: {
11: sLog.outError ("Failed to start network");
12: Log::WaitBeforeContinueIfNeed();
13: World::StopNow(ERROR_EXIT_CODE);
14: // go down and shutdown the server
15: }
16:
17: sWorldSocketMgr->Wait ();
18:
19: ........
20: }
mangosd用WorldSocketMgr類來管理socket。StartNetwork ()會調用StartReactiveIO ()來啟動監聽socket,處理代碼如下:
1: int WorldSocketMgr::StartReactiveIO (ACE_UINT16 port, const char* address)
2: {
3: ........
4:
5: //(1)
6: m_NetThreadsCount = static_cast<size_t> (num_threads + 1);
7: m_NetThreads = new ReactorRunnable[m_NetThreadsCount];
8:
9: // -1 means use default
10: m_SockOutKBuff = sConfig.GetIntDefault ("Network.OutKBuff", -1);
11: m_SockOutUBuff = sConfig.GetIntDefault ("Network.OutUBuff", 65536);
12: if ( m_SockOutUBuff <= 0 )
13: {
14: sLog.outError ("Network.OutUBuff is wrong in your config file");
15: return -1;
16: }
17:
18: //(2)
19: WorldSocket::Acceptor *acc = new WorldSocket::Acceptor;
20: m_Acceptor = acc;
21:
22: ACE_INET_Addr listen_addr (port, address);
23: if (acc->open (listen_addr, m_NetThreads[0].GetReactor (), ACE_NONBLOCK) == -1)
24: {
25: sLog.outError ("Failed to open acceptor ,check if the port is free");
26: return -1;
27: }
28:
29: //(3)
30: for (size_t i = 0; i < m_NetThreadsCount; ++i)
31: m_NetThreads[i].Start ();
32:
33: return 0;
34: }
(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 () 如下:
1: virtual int svc ()
2: {
3: //(1)
4: WorldDatabase.ThreadStart ();
5:
6: SocketSet::iterator i, t;
7: while (!m_Reactor->reactor_event_loop_done ())
8: {
9: // dont be too smart to move this outside the loop
10: // the run_reactor_event_loop will modify interval
11: ACE_Time_Value interval (0, 10000);
12:
13: //(2)
14: if (m_Reactor->run_reactor_event_loop (interval) == -1)
15: break;
16:
17: //(3)
18: AddNewSockets ();
19:
20: for (i = m_Sockets.begin (); i != m_Sockets.end ();)
21: {
22: //(4)
23: if ((*i)->Update () == -1)
24: {
25: t = i;
26: ++i;
27: (*t)->CloseSocket ();
28: (*t)->RemoveReference ();
29: --m_Connections;
30: m_Sockets.erase (t);
31: }
32: else
33: ++i;
34: }
35: }
36:
37: WorldDatabase.ThreadEnd ();
38: DEBUG_LOG ("Network Thread Exitting");
39:
40: return 0;
41: }
(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的輸出:
1: int WorldSocket::Update (void)
2: {
3: if (closing_)
4: return -1;
5:
6: //(1)
7: if (m_OutActive || m_OutBuffer->length () == 0)
8: return 0;
9:
10: return handle_output (get_handle ());
11: }
(1)m_OutBuffer有數據時才會調用handle_output,handle_output ()用于處理輸出,如果輸出不能一次性做完,會調用schedule_wakeup_output ()再次激活write事件。當輸出處理完畢后則調用cancel_wakeup_output ()取消激活write事件,使reactor恢復到正常的loop ()循環中。詳細過程如下:
1: int WorldSocket::handle_output (ACE_HANDLE)
2: {
3: //(1)
4: ACE_GUARD_RETURN (LockType, Guard, m_OutBufferLock, -1);
5:
6: if (closing_)
7: return -1;
8:
9: const size_t send_len = m_OutBuffer->length ();
10: if (send_len == 0)
11: return cancel_wakeup_output (Guard);
12:
13: #ifdef MSG_NOSIGNAL
14: ssize_t n = peer ().send (m_OutBuffer->rd_ptr (), send_len, MSG_NOSIGNAL);
15: #else
16: ssize_t n = peer ().send (m_OutBuffer->rd_ptr (), send_len);
17: #endif // MSG_NOSIGNAL
18:
19: if (n == 0)
20: return -1;
21: else if (n == -1)
22: {
23: if (errno == EWOULDBLOCK || errno == EAGAIN) //----------(2)
24: return schedule_wakeup_output (Guard);
25:
26: return -1;
27: }
28: else if (n < (ssize_t)send_len) //now n > 0 //----------(3)
29: {
30: m_OutBuffer->rd_ptr (static_cast<size_t> (n));
31:
32: // move the data to the base of the buffer
33: m_OutBuffer->crunch ();
34:
35: return schedule_wakeup_output (Guard);
36: }
37: else //now n == send_len //----------(4)
38: {
39: m_OutBuffer->reset ();
40:
41: if (!iFlushPacketQueue ())
42: return cancel_wakeup_output (Guard);
43: else
44: return schedule_wakeup_output (Guard);
45: }
46:
47: ACE_NOTREACHED (return 0);
48: }
(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),如下:
1: int WorldSocket::open (void *a)
2: {
3: ........
4:
5: // Hook for the manager.
6: if (sWorldSocketMgr->OnSocketOpen (this) == -1)
7: return -1;
8:
9: ........
10: }
OnSocketOpen方法:
1: int WorldSocketMgr::OnSocketOpen (WorldSocket* sock)
2: {
3: ........
4:
5: // we skip the Acceptor Thread
6: size_t min = 1;
7:
8: MANGOS_ASSERT (m_NetThreadsCount >= 1);
9:
10: //(1)
11: for (size_t i = 1; i < m_NetThreadsCount; ++i)
12: if (m_NetThreads[i].Connections () < m_NetThreads[min].Connections ())
13: min = i;
14:
15: return m_NetThreads[min].AddSocket (sock);
16: }
(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