ceph中Dispatcher模塊的示例分析

185次閱讀

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

這篇文章主要介紹了 ceph 中 Dispatcher 模塊的示例分析，具有一定借鑒價值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓丸趣 TV 小編帶著大家一起了解一下。

Dipatcher 類是消息分發的接口，OSD、MON、等類都繼承該類，并實現了 Dipatcher 的消息分發接口

 1079 class OSD : public Dispatcher,
- 1080 public md_config_obs_t {| 1081 /** OSD **/

 128 class Monitor : public Dispatcher,
- 129 public md_config_obs_t {
| 130 public:
| 131 // me
| 132 string name;

在 OSD::init() 函數中把不同類型的 Dipatcher 加入到 SimpleMessenger 實例中

| 2146 // i m ready!
| 2147 client_messenger- add_dispatcher_head(this);
| 2148 cluster_messenger- add_dispatcher_head(this);
| 2149 
| 2150 hbclient_messenger- add_dispatcher_head(heartbeat_dispatcher);
| 2151 hb_front_server_messenger- add_dispatcher_head(heartbeat_dispatcher);
| 2152 hb_back_server_messenger- add_dispatcher_head(heartbeat_dispatcher);
| 2153 
| 2154 objecter_messenger- add_dispatcher_head(service.objecter);

在 Messenger::add_dispatcher_head(Dispatcher *d) 中加入 Messenger::list Dispatcher* dispatchers 中，并調用 ready（）,SimpleMessenger::ready() 重寫了基類的 ready，

|- 358 void add_dispatcher_head(Dispatcher *d) {|| 359 bool first = dispatchers.empty();
|| 360 dispatchers.push_front(d);
|| 361 if (d- ms_can_fast_dispatch_any())
|| 362 fast_dispatchers.push_front(d);
|| 363 if (first)
|| 364 ready();
|| 365 }

在 ready 函數中調用 DispatchQueue::start, start() 函數啟動 DispatchQueue::DispatchThread 和 DispatchQueue::LocalDeliveryThread 線程類，最終調用 DispatchQueue::entry() 和 DispatchQueue::run_local_delivery。

 216 void DispatchQueue::start() 
- 217 { 
| 218 assert(!stop);
| 219 assert(!dispatch_thread.is_started());
| 220 dispatch_thread.create( ms_dispatch  // 調用 Thread::create- Thread::try_create- Thread::_entry_func- Thread::entry_wrapper- DispatchThread::entry
| 221 local_delivery_thread.create( ms_local 
| 222 }

|- 98 class DispatchThread : public Thread {
|| 99 DispatchQueue *dq;
|| 100 public:
|| 101 explicit DispatchThread(DispatchQueue *dq) : dq(dq) {}
||- 102 void *entry() {||| 103 dq- entry();
||| 104 return 0;
||| 105 } 
|| 106 } dispatch_thread;

在 DispatchQueue::entry() 中調用根據不同的命令碼調用不同的 Messenger 類中的處理函數

void DispatchQueue::entry()
 .
 .
 
 switch (qitem.get_code()) {
 case D_BAD_REMOTE_RESET:
  msgr- ms_deliver_handle_remote_reset(qitem.get_connection());
  break;
 case D_CONNECT:
  msgr- ms_deliver_handle_connect(qitem.get_connection());
  break;
 case D_ACCEPT:
  msgr- ms_deliver_handle_accept(qitem.get_connection());
  break;
 case D_BAD_RESET:
  msgr- ms_deliver_handle_reset(qitem.get_connection());
  break;
 default:
  assert(0);
 } else {Message *m = qitem.get_message();
 if (stop) { ldout(cct,10)     stop flag set, discarding     m         *m   dendl;
  m- put();} else { uint64_t msize = pre_dispatch(m);
  msgr- ms_deliver_dispatch(m);
  post_dispatch(m, msize);
 }
 .
 .
}

在 Messenger::ms_deliver_dispatch 中最終調用不同的 Dipatcher 繼承類的 ms_dispatch 進行處理

|- 579 void ms_deliver_dispatch(Message *m) {|| 580 m- set_dispatch_stamp(ceph_clock_now(cct));
|| 581 for (list Dispatcher* ::iterator p = dispatchers.begin(); 
|| 582 p != dispatchers.end();
||- 583 ++p) {||| 584 if ((*p)- ms_dispatch(m)) // 在 Dispatcher 繼承類中進行處理
||| 585 return;
||| 586 }
|| 587 lsubdout(cct, ms, 0)    ms_deliver_dispatch: unhandled message     m         *m     from  
|| 588   m- get_source_inst()   dendl;
|| 589 assert(!cct- _conf- ms_die_on_unhandled_msg);
|| 590 m- put();
|| 591 }

感謝你能夠認真閱讀完這篇文章，希望丸趣 TV 小編分享的“ceph 中 Dispatcher 模塊的示例分析”這篇文章對大家有幫助，同時也希望大家多多支持丸趣 TV，關注丸趣 TV 行業資訊頻道，更多相關知識等著你來學習!

正文完