2019獨角獸企業重金招聘Python工程師標準>>>

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 }

?

轉載于:https://my.oschina.net/u/2326998/blog/1814357

總結

以上是生活随笔為你收集整理的ceph Dispatcher模块分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。