Ceph網絡模塊代碼的示例分析

197次閱讀

共計 2739 個字符，預計需要花費 7 分鐘才能閱讀完成。

這篇文章主要為大家展示了“Ceph 網絡模塊代碼的示例分析”，內容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓丸趣 TV 小編帶領大家一起研究并學習一下“Ceph 網絡模塊代碼的示例分析”這篇文章吧。

1. 創建消息管理類 Messenger.

Messenger *ms_public = Messenger::create(g_ceph_context,
 entity_name_t::OSD(whoami),  client ,
 getpid());

在這里函數這里創建了若干個 Messenger 實例,ms_public 實例只是其中的一個. 從這個函數參數 client 也能看出一些端倪, 說明這個 Messenger 是專門用于
處理客戶端的消息的, 這里的客戶端指的是諸如 RGW, RBD 這樣的客戶端. 而后面的 ms_cluster 則處理與其它 OSD 的消息。

2. 綁定端口和地址

然后在后面, 就像 Linux Socket 編程一樣, 在這里也需要調用 bind() 函數完成端口以及地址的綁定, 只不過這里 bind() 函數是經過特殊封裝后的.

ms_public- bind(g_conf- public_addr);

Messenger 是一個抽象類, ms_public 是一個基類指針.
而類 SimpleMessenger 繼承自 Messenger.

最終調用的 bind() 函數是子類 SimpleMessenger::bind(const entity_addr_t bind_addr)

int SimpleMessenger::bind(const entity_addr_t  bind_addr)
 lock.Lock();
 if (started) { ldout(cct,10)    rank.bind already started    dendl;
 lock.Unlock();
 return -1; 
 }
 ldout(cct,10)    rank.bind     bind_addr   dendl;
 lock.Unlock();
 // bind to a socket
 set int  avoid_ports;
 int r = accepter.bind(bind_addr, avoid_ports);
 if (r  = 0)
 did_bind = true;
 return r;
}

這里的 entity_addr_t 實際是 socket 通信中需要用到的 sockaddr_in,AF_INET/AF_INET6 的集合.
繼續調用成員 accepter.bind(bind_addr, avoid_ports);
在 msg/Accepter.cc 中可以看到:

int Accepter::bind(const entity_addr_t  bind_addr, const set int  avoid_ports)

這個函數執行了:

1) 創建一個 socket fd.

2) 執行 bind()

3) 執行 listen()
以上都是 socket 編程中基本操作步驟.

3. 啟動服務端

ms_public- start();

實際上調用的是:

int SimpleMessenger::start()
 lock.Lock();
 ldout(cct,1)    messenger.start    dendl;
 // register at least one entity, first!
 assert(my_type  = 0);
 assert(!started);
 started = true;
 if (!did_bind) {
 my_inst.addr.nonce = nonce;
 init_local_connection();
 }
 lock.Unlock();
 reaper_started = true;
 reaper_thread.create();
 return 0;
}

從這里可以看出,reaper_thread.create() 創建了一個 reaper 線程. 找到它的 void *entry() 函數.//SimpleMessenger.h

reaper 線程實際上執行的是 void SimpleMessenger::reaper_entry(), 它最終執行 / 調用的是:

void SimpleMessenger::reaper()
 ldout(cct,10)    reaper    dendl;
 assert(lock.is_locked());
 while (!pipe_reap_queue.empty()) { Pipe *p = pipe_reap_queue.front();
 pipe_reap_queue.pop_front();
 ldout(cct,10)    reaper reaping pipe     p         p- get_peer_addr()   dendl;
 p- pipe_lock.Lock();
 p- discard_out_queue();
 if (p- connection_state) { // mark_down, mark_down_all, or fault() should have done this,
 // or accept() may have switch the Connection to a different
 // Pipe... but make sure!
 bool cleared = p- connection_state- clear_pipe(p);
 assert(!cleared);
 }
 p- pipe_lock.Unlock();
 p- unregister_pipe();
 assert(pipes.count(p));
 pipes.erase(p);
 p- join();
 if (p- sd  = 0)
 ::close(p- 
 ldout(cct,10)    reaper reaped pipe     p         p- get_peer_addr()   dendl;
 p- put();
 ldout(cct,10)    reaper deleted pipe     p   dendl;
 }
 ldout(cct,10)    reaper done    dendl;
}

從這里可以看出做出了清理 Pipe 的操作. (?)

4. 最后一步: