ceph中Dispatcher模块的示例分析

80次阅读

共计 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 行业资讯频道，更多相关知识等着你来学习!

正文完