作为分布式存储系统，Msg(src/msg)模块可谓是Ceph的基石之一。Ceph发展到Luminous ，已经支持的3大通信机制：simple，async和xio，其中simple历史最为悠久，是Ceph最早的通信模块，原理简单但性能较差（多线程模型，对于较大的集群，会产生很多线程，对主机的要求也比较高。）。async作为后起之秀，优良的性能使其自Luminous开始已经作为了缺省msg方案。xio拥有众多实验特性，目前距离生产环境还有很大距离。

在使用OOP设计一个通信模块时，往往少不了以下几个抽象：

• Messenger 用于在最高层次管理所有通信，通常包括通信策略，工作线程等等
• Connection表示对一个连接的抽象，经常会设计一个状态机来供上层管理该Connection
• Message是对消息的封装，通常包含’Header + Data + Checking’ 几部分，Message和报文流的常常需要转换方法，Message->buf:encode()，Message<-buf:decode()
• Stack 负责实现真实的通信，比如TCP/IP协议栈、RDMA协议栈

以此为基础，Ceph Msg框架就显得十分清晰，下面就是Async机制核心的封装：AsyncMessenger，Processor，AsyncConnection，NetworkStack，Worker，EventCenter，Message，本文将逐个讨论这些角色。

上图是Ceph Msg模块框图，可以看出，msg模块可以分为3个层次：

层次	内容
1	async/simple/xio
2	Generic NetworkStack（通用）
3	Specific NetworkStack（特定）

上层将”通信机制”抽象出来，下层聚焦于协议栈，包括硬件无关的部分以及硬件相关的部分，比如RDMAStack针对配置了IB卡的存储节点，DPDKStack用于使用X86 DPDK技术的存储节点，而PosixStack则是Linux原生的Socket通信接口。

AsyncMessenger

AsynMessenger类继承自Messenger:

class  AsyncMessenger:public SimplePolicyMessenger

class  SimplePolicyMessenger:public Messenger

SimplePolicyMessenger是和”连接策略”相关的类。AsyncMessenger作为整个通信模块的核心和中转，不论是作为底层通信的NetworkStack，还是作为连接的抽象的Connection，抑或处理对端链接请求的Processor，都要围着它转。以OSD进程的main函数为例(src/ceph_)，整个OSD进程也不过7个Messenger实例：

ms_public， ms_cluster, ms_hb_back_client, ms_hb_front_client, ms_hb_back_server, ms_hb_front_server, ms_objecter，

这7个Messenger分别用来处理不同类型的消息。

//src/msg/async/AsyncMessenger.h
 75 class AsyncMessenger : public SimplePolicyMessenger {
139   int send_message(Message *m, const entity_inst_t& dest) override {}
176   void ready() override; //启动Processor和dispatch_queue线程
197   AsyncConnectionRef create_connect(const entity_addr_t& addr, int type);
221   NetworkStack *stack;
222   std::vector<Processor*> processors;
223   friend class Processor;
224   DispatchQueue dispatch_queue;
273   ceph::unordered_map<entity_addr_t, AsyncConnectionRef> conns;
280   set<AsyncConnectionRef> accepting_conns;
343   AsyncConnectionRef lookup_conn(const entity_addr_t& k) {
346   }
348   int accept_conn(AsyncConnectionRef conn) {
368   }
371   void add_accept(Worker *w, ConnectedSocket cli_socket, entity_addr_t &addr);
448 } ;

//src/msg/async/AsyncMessenger.h
class AsyncMessenger : public SimplePolicyMessenger
{
    //提供给上层提交发送请求的接口，AsyncMessenger::send_message()
    int send_message(Message *m, const entity_inst_t &dest) override
    {
    }
    //启动Processor和dispatch_queue线程
    void ready() override;
    //创建Connection实例的接口
    AsyncConnectionRef create_connect(const entity_addr_t &addr, int type);//
    //Messenger关联的Network实例，收发流程都要经过该实例的传递
    NetworkStack *stack;
    std::vector<Processor *> processors;
    friend class Processor;
    //向上层提交收到的Message的队列
    DispatchQueue dispatch_queue;
    //管理Connection实例的数据结构，所有已经成功建立起来的Connection都可以在这里找到
    ceph::unordered_map<entity_addr_t, AsyncConnectionRef> conns;
    //对于Connection建立过程中的Server端，accept返回但还在协商阶段的Connection被放置于此，连接建立成功后将从这里移到conns中
    set<AsyncConnectionRef> accepting_conns;
    //根据地址在conns中查找对应的Connection实例
    AsyncConnectionRef lookup_conn(const entity_addr_t &k)
    {
    }
    //将链接从accepting_conns中删除，并添加到conns中
    int accept_conn(AsyncConnectionRef conn)
    {
    }
    //被Processor::accept()调用，Server端一旦侦听到Client端的链接，就会在Server端建立一个Connection实例，并将其加入到accepting_conns中以待后续处理，这个建立和加入的过程就是在该函数中完成的。
    void add_accept(Worker *w, ConnectedSocket cli_socket, entity_addr_t &addr);
} ;

–139–>提供给上层提交发送请求的接口，AsyncMessenger::send_message()–>AsyncMessenger::_send_message()–>AsyncMessenger::submit_message()–>AsyncConnection::send_message()

–176–>启动Processor和dispatch_queue线程
–197–>创建Connection实例的接口
–221–>Messenger关联的Network实例，收发流程都要经过该实例的传递
–224–>向上层提交收到的Message的队列
–273–>管理Connection实例的数据结构，所有已经成功建立起来的Connection都可以在这里找到
–280–>对于Connection建立过程中的Server端，accept返回但还在协商阶段的Connection被放置于此，连接建立成功后将从这里移到conns中
–343–>根据地址在conns中查找对应的Connection实例
–348–>将链接从accepting_conns中删除，并添加到conns中
–371–>被Processor::accept()调用，Server端一旦侦听到Client端的链接，就会在Server端建立一个Connection实例，并将其加入到accepting_conns中以待后续处理，这个建立和加入的过程就是在该函数中完成的。

AsyncMessenger::AsyncMessenger()

//src/msg/async/
245 AsyncMessenger::AsyncMessenger(CephContext *cct, entity_name_t name, const std::string &type, string mname, uint64_t _nonce)
253 {
254   std::string transport_type = "posix";
255   if (type.find("rdma") != std::string::npos)
256     transport_type = "rdma";
257   else if (type.find("dpdk") != std::string::npos)
258     transport_type = "dpdk";
259 
260   ceph_spin_init(&global_seq_lock);
261   StackSingleton *single;
262   cct->lookup_or_create_singleton_object<StackSingleton>(single, "AsyncMessenger::NetworkStack::"+tr
263   single->ready(transport_type);
264   stack = single->stack.get();   //返回单例的this
265   stack->start();
266   local_worker = stack->get_worker();
267   local_connection = new AsyncConnection(cct, this, &dispatch_queue, local_worker);
268   init_local_connection();
269   reap_handler = new C_handle_reap(this);
270   unsigned processor_num = 1;
271   if (stack->support_local_listen_table())
272     processor_num = stack->get_num_worker();
273   for (unsigned i = 0; i < processor_num; ++i)
274     processors.push_back(new Processor(this, stack->get_worker(i), cct));
275 }

–254-258  –>协议栈默认是Posix，根据ceph.conf中指定的，可以为rdma或dpdk
–261-262  –>使用单例模式single(用于构造NetworkStack对象），所以在一个Ceph进程(MON/MDS/OSD/Client)中，一个类型的协议栈有且只有一个实例
–263         –>利用单例single构造NetworkStack对象，实质是调用"NetworkStack::create(cct,type)"，后者会根据type的不同来初始化不同的子类，比如RDMAStack，PosixStack或DPDKStack。在NetworkStack实例被构造的同时，还会构造一组Worker供关联于它的AsyncMessenger使用，类似的，这组Worker也会根据ceph.conf而被实例化出不同的子类实例，比如RDMAWorker。这组worker使用"AsyncMessenger->NetworkStack::vector<Worker*>workers"管理。
–264        –>将构造好的NetworkStack与AsyncMessenger关联，一个进程内可以有多个Messenger实例(参见ceph_)，但如果只有一种类型的网络，eg，public/cluster均为Posix，则该进程内只有一个NetworkStack实例
–265        –>”启动”协议栈，其主要工作就是针对每一个workers[i]启动一个线程，这个线程的核心任务就是循环执行"Worker.EventCenter.process_events()"，相关的资源都被封装在Worker及其子类中。
–266        –>get_worker()会根据Work Load获取当前Worker中的负载最轻的线程
–271-274 –>构造一组(个)Processor实例，和Worker类似，每个Processor实例都是对一个线程运行资源的封装，在"Processor::start()"中(not here)启动这些线程。使用"AsyncMessenger::vector<Processor*>processors"管理。

Processor

Processor的主要工作是”监视”，类似于Socket编程中的listen，Async机制会启动一组Processor线程，每个线程”监视”一个端口，一旦有链接建立，即构造一个AsynConnection实例并交由该Processor线程对应的Worker线程处理。这里插句题外话，Processor线程和Worker线程是”多对一”的关系，一个Worker线程不但处理Processor发给它的任务，还可能会有一堆AsyncConnection发给它的任务。在Ceph中，进程之间(eg, OSD-OSD)没有C/S之分，但两个进程之间有很多的Connection，对于每个Connection确是有C/S之分的。在没有链接的情况下，主动建立链接的为Client，被动接受链接的为Server，而Processor就是那个使一个进程可以像Server一样被动建立链接的前提。

//ceph-12.0.0\src\msg\async\AsyncMessenger.h
47 class Processor {
 48   AsyncMessenger *msgr;
 49   NetHandler net;
 50   Worker *worker;
 51   ServerSocket listen_socket;
 52   EventCallbackRef listen_handler;
 54   class C_processor_accept;
 57   Processor(AsyncMessenger *r, Worker *w, CephContext *c);
 60   void stop();
 61   int bind(const entity_addr_t &bind_addr,
 62            const set<int>& avoid_ports,
 63            entity_addr_t* bound_addr);
 64   void start();
 65   void accept();
 66 };

–48–>隶属的AsyncMessenger实例（父AsyncMessenger）
–49–>Processor建立连接的底层是使用Socket的listen-accept机制，这些机制被封装在了NetHandler中
–50–>Processor关联的Worker线程，Processor和Connection一样，所有Event最后都提交给关联的Worker处理
–51–>监听socket，本质是对Specific Stack层中的Server端Socket进行封装，比如RDMAServerScoketImpl::server_setup_socket(listen_socket.fd())
–52–>实质是C_processor_accept实例，作为listen_socket.fd()的handler被注册到Msg中的file events中，当Client端有连接过来时，该fd()会变得可读，此时就会回调该listen_handler，本质是调用Processor.accept()–>AsyncMessenger.add_accept()–>AsyncConnection.accept()来建立一个可用连接。
–61–>绑定一个地址作为listen_socket。

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

=================================================================================================

 

 

当前的版本（mimic，v13）默认的messenger框架是AsyncMessenger，下面以OSD的ms_public来介绍AsyncMessenger网络模块。

 messenger = Messenger::create(g_ceph_context, g_conf->ms_type,
                  entity_name_t::MON(-1),
                  "simple_server",
                  0 /* nonce */);

 工厂模式，在create函数中会根据g_conf->ms_type来选择创建不同类型的Messenger，Mimic默认为"async+posix"，因此会创建AsyncMessenger，返回如下：
return new AsyncMessenger(cct, name, type, std::move(lname), nonce);

 

ceph-12.0.0\src\test\msgr\perf_msgr_中的通信初始化流程

int main(int argc, char **argv) { ……   auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT,              CODE_ENVIRONMENT_UTILITY, 0);   common_init_finish(g_ceph_context);   g_ceph_context->_conf->apply_changes(NULL); ……   MessengerServer server(g_ceph_context->_conf->ms_type, args[0], worker_threads, think_time);   server.start();   return 0; }

 

在这个例子中，相比于前面讲的 src\test\messenger\simple_server.c里的实例，本例子中Messenger被封装在MessengerServer类中，是MessengerServer的成员

class MessengerServer {
  Messenger *msgr;
  string type;
  string bindaddr;
  ServerDispatcher dispatcher;

 public:
  MessengerServer(string t, string addr, int threads, int delay):
      msgr(NULL), type(t), bindaddr(addr), dispatcher(threads, delay) {
    msgr = Messenger::create(g_ceph_context, type, entity_name_t::OSD(0), "server", 0, 0);
    msgr->set_default_policy(Messenger::Policy::stateless_server(0, 0));
  }

 MessengerServer参数实例化时，类内部创建一个  msgr = Messenger::create(g_ceph_context, type, entity_name_t::OSD(0), "server", 0, 0);

class MessengerServer {
  Messenger *msgr;
  string type;
  string bindaddr;
  ServerDispatcher dispatcher;

 public:
  MessengerServer(string t, string addr, int threads, int delay):
      msgr(NULL), type(t), bindaddr(addr), dispatcher(threads, delay) {
    msgr = Messenger::create(g_ceph_context, type, entity_name_t::OSD(0), "server", 0, 0);
    msgr->set_default_policy(Messenger::Policy::stateless_server(0, 0));
  }
  ~MessengerServer() {
    msgr->shutdown();
    msgr->wait();
  }
  void start() {
    entity_addr_t addr;
    addr.parse(bindaddr.c_str());
    msgr->bind(addr);
    msgr->add_dispatcher_head(&dispatcher);
    msgr->start();
    msgr->wait();
  }
};

然后server.start(); 完成了和src\test\messenger\simple_server.c里的实例一样的messenger->start();和messenger->wait();

上面的代码是典型的服务端的启动流程：

1. 绑定服务端地址 msgr->bind(addr)
2. 添加消息分发类型 dispatcher
3. 启动 msgr->start()

和src\test\messenger\simple_server.c里的实例同样的流程

messenger = Messenger::create（）；
 
r = messenger->bind(bind_addr);
 
dispatcher = new SimpleDispatcher(messenger);
 
messenger->add_dispatcher_head(dispatcher);  //messager被观察对象 将 dispatcher观察者注册到最近的队列中
 
messenger->start();
 
messenger->wait(); // can't be called until ready()

 下面从内部具体如何实现。

• 调用processor的bind 函数，对于PosixStack， 只需要一个porcessor就可以了。

int AsyncMessenger::bind(const entity_addr_t &bind_addr) {  ......      for (auto &&p : processors) {          int r = p->bind(bind_addr, avoid_ports, &bound_addr);        }  ...... }

int Processor::bind(const entity_addr_t &bind_addr,             const set<int>& avoid_ports,             entity_addr_t* bound_addr) { ...   //向Processor对应的worker线程 投递外部事件，其回调函数为 worker的 listen函数    worker->center.submit_to(worker->center.get_id(), [this, &listen_addr, &opts, &r]() {         r = worker->listen(listen_addr, opts, &listen_socket);       }, false); ... }

 

原文：ceph Async 网络通信源代码分析（二）_changtao381的专栏-CSDN博客

 

1. Async模块

1. Async工作原理

1. Async主要组件

AsyncMessenger	管理网络连接
AsyncConnection	网路通信连接，定义网络通信应用层协议
NetworkStack	管理Worker对象及其对应地线程
Worker	网络I/O流处理单元，每个Worker对应一个工作线程
ServerSocket/ServerSocketImpl	C/S模式监听套接字，向上屏蔽了各种不同的网络编程接口
ConnectedSocket/ConnectedSocketImpl	C/S模式连接套接字，向上屏蔽了各种不同的网络编程接口
EventCenter	事件分发器，负责事件注册、事件分发
EventCallback	当对应的事件发生时，由EventCenter负责回调
EventEpoll	对epoll进行封装，轮询网络I/O事件