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 }