对于一个分布式存储系统，需要一个稳定的底层网络通信模块，用于各个节点的之间的互联互通。
对于一个网络通信系统，要求：
在msg的子目录下， 分别对应三种不同的实现方式：Simple， Async， XIO
Simple是相对比较简单，目前可以在生产环境中使用的模式。 它最大的特点是，每一个链接，都创建两个线程，一个专门用于接收，一个专门用于发送。
这样的模式实现起来比较简单，但是对于大规模的集群部署，大量的链接会产生大量的线程，占用太多资源，影响网络的性能。
Async模式使用了基于事件的IO 多路复用模式。这是目前比较通用的方式，没有用第三方库，实现起来比较复杂。目前还处于试验阶段。
XIO模式：使用了开源的的网络通信模块accelio 来实现。这种需要依赖第三方库。实现起来可以简单一些。但是需要对accelio的使用方式熟悉，accelio出问题要有bug fix的能力。目前也处于试验阶段。
特别注意的是，前两种方式只支持 tcp/ip 协议，XIO 可以支持 Infiniband网络。
在src/msg 的目录里，定义了网络模块的接口。
在源代码src/msg 里实现了ceph 的网络通信模块。在msg目录下，定义了网络通信的抽象接口。
msg/Message.cc
msg/Message.h 定义了message
msg/Connection.h 关于connection 的接口定义
msg/Dispatcher.h 关于Dis
msg/Messenger.cc 定义了Messenger
msg/Messenger.h
msg/msg_types.cc 定义了消息的类型
msg/msg_types.h
Message 是所有消息的基类，任何要发送的消息，都要继承该类。对于消息，其发送格式如下：
header + user_data + footerheader是消息头，类似一个消息的信封(envelope)，保存消息的一些描述信息。user_data 是用于要发送的实际数据, footer是一个消息尾，保存了消息的效验信息和结束标志。
user_data
payload + middle + data用户数据在Message里 有三部分组成： payload， 一般保存ceph操作的元数据；middle 预留，目前没有使用到；data 一般为读写的数据。
其对应的数据结构如下：
ceph_msg_header
struct ceph_msg_header { __le64 seq; /* message seq# for this session */ 当前session内 消息的唯一 序号 __le64 tid; /* transaction id */ 消息的全局唯一的 id __le16 type; /* message type */ 消息类型 __le16 priority; /* priority. higher value == higher priority */ 优先级 __le16 version; /* version of message encoding */ 版本 __le32 front_len; /* bytes in main payload */ payload 的长度 __le32 middle_len; /* bytes in middle payload */ middle 的长度 __le32 data_len; /* bytes of data payload */ data 的 长度 __le16 data_off; /* sender: include full offset; 对象的数据偏移量 receiver: mask against ~PAGE_MASK */ struct ceph_entity_name src; //消息源 /* oldest code we think can decode this. unknown if zero. */ __le16 compat_version; __le16 reserved; __le32 crc; /* header crc32c */} __attribute__ ((packed));ceph_msg_footer
struct ceph_msg_footer { __le32 front_crc, middle_crc, data_crc; //分别对应crc 效验码 __le64 sig; // 消息的64位 signature __u8 flags; //结束标志} __attribute__ ((packed));Message
class Message{ ceph_msg_header header; // 消息头 ceph_msg_footer footer; // 消息尾 bufferlist payload; // "front" unaligned blob bufferlist middle; // "middle" unaligned blob bufferlist data; // data payload (page-alignment will be preserved where possible) /* recv_stamp is set when the Messenger starts reading the * Message off the wire */ utime_t recv_stamp; //开始接收数据的时间戳 /* dispatch_stamp is set when the Messenger starts calling dispatch() on * its endpoints */ utime_t dispatch_stamp; // dispatch 的时间戳 /* throttle_stamp is the point at which we got throttle */ utime_t throttle_stamp; /* time at which message was fully read */ utime_t recv_complete_stamp; //接收完成的时间戳 ConnectionRef connection; //链接 uint32_t magic; bi::list_member_hook<> dispatch_q; //boost::intrusive list 的 member}类Connection 就对应的一个链接，它是socket的port 对port 链接的封装。其最重要的接口，就是可以发送消息
struct Connection : public RefCountedObject { mutable Mutex lock; //锁包括 Connection的所有字段 Messenger *msgr; RefCountedObject *priv; //私有数据 int peer_type; //链接的类型 entity_addr_t peer_addr; //对方的地址 utime_t last_keepalive, last_keepalive_ack; //最后一次发送keeplive的时间 和最后一次接受keepalive的时间private: uint64_t features; //一些feature的标志位public: bool failed; // true if we are a lossy connection that has failed. int rx_buffers_version; //接收缓存区的版本 map<ceph_tid_t,pair<bufferlist,int> > rx_buffers; //接收缓冲区 ceph_tid --> (buffer, rx_buffers_version)}其最重要的功能，就是发送消息的接口
virtual int send_message(Message *m) = 0;
类Dispatcher 是接收消息的接口。 其接收消息的接口为：
virtual bool ms_dispatch(Message *m) = 0;virtual void ms_fast_dispatch(Message *m);无论是Server端，还是Client 端， 都需要实现一个Dispatcher 函数，对于Server 来接收请求，对应client 端，来接收ack应答。
Messenger 是整个网络模块功能类的抽象。其定义了网络模块的基本功能接口。网络模块对外提供的基本的功能，就是能在节点之间发送消息。
向一个节点发送消息
virtual int send_message(Message *m, const entity_inst_t& dest) = 0;注册一个，用来接收消息
void add_dispatcher_head(Dispatcher *d)通过下面的最基本的服务器和客户端的程序的展示，了解如何调用网络通信来完成收发请求的功能。
其源代码在 test/simple_server.cc里，这里只展示有关网络部分的核心流程。
1.调用 Messenger的函数create 创建一个Messenger的实例，g_conf->ms_type为实现的类型，目前有三种方式，simple，async，xio.
messenger = Messenger::create(g_ceph_context, g_conf->ms_type, entity_name_t::MON(-1), "simple_server", 0 /* nonce */);2.设置 messenger 的属性
messenger->set_magic(MSG_MAGIC_TRACE_CTR);messenger->set_default_policy( Messenger::Policy::stateless_server(CEPH_FEATURES_ALL, 0));3.对于 server，需要bind 服务端地址
r = messenger->bind(bind_addr);if (r < 0) goto out;common_init_finish(g_ceph_context);4.创建一个Dispatcher，并添加到Messenger
dispatcher = new SimpleDispatcher(messenger);messenger->add_dispatcher_head(dispatcher); 5.启动messenger
messenger->start();messenger->wait(); // can't be called until ready()SimpleDispatcher 函数里实现了ms_dispatch，用于处理接收到的各种请求消息。
1.调用 Messenger的函数create 创建一个Messenger的实例
messenger = Messenger::create(g_ceph_context, g_conf->ms_type, entity_name_t::MON(-1), "client", getpid());messenger->set_magic(MSG_MAGIC_TRACE_CTR);messenger->set_default_policy(Messenger::Policy::lossy_client(0, 0));3.创建Dispatcher 类并添加，用于接收消息
dispatcher = new SimpleDispatcher(messenger);messenger->add_dispatcher_head(dispatcher);dispatcher->set_active(); // this side is the pinger4.启动消息
r = messenger->start();if (r < 0) goto out;5.下面开始发送请求，先获取目标server 的链接
conn = messenger->get_connection(dest_server);6.通过Connection来发送请求消息。需要注意的是，这里的消息发送都是异步发送，请求的ack应对消息回来后在Dispatcher的 ms_dispatch或者ms_fast_dispatch里处理。
Message *m;for (msg_ix = 0; msg_ix < n_msgs; ++msg_ix) { /* add a data payload if asked */ if (! n_dsize) { m = new MPing(); } else { m = new_simple_ping_with_data("simple_client", n_dsize); } conn->send_message(m);}Simple 是ceph里比较早，目前也比较稳定，在生产环境中使用的网络通信模块。
Accepter accepter; 用于接受客户端的链接请求DispatchQueue dispatch_queue; 接收到的请求的消息分发队列 bool did_bind; 是否绑定 /// counter for the global seq our connection protocol uses __u32 global_seq; /// lock to protect the global_seq ceph_spinlock_t global_seq_lock; /** * hash map of addresses to Pipes * * NOTE: a Pipe* with state CLOSED may still be in the map but is considered * invalid and can be replaced by anyone holding the msgr lock */ ceph::unordered_map<entity_addr_t, Pipe*> rank_pipe; /** * list of pipes are in teh process of accepting * * These are not yet in the rank_pipe map. */ set<Pipe*> accepting_pipes; /// a set of all the Pipes we have which are somehow active set<Pipe*> pipes; /// a list of Pipes we want to tear down list<Pipe*> pipe_reap_queue; /// internal cluster protocol version, if any, for talking to entities of the same type. int cluster_protocol;类Accepter 用来在server端监听，接受链接。其继承了Thread类，本身是一个线程，可以不断的监听server 的端口。
DispatchQueue 类用于把接收到的请求保存在内部， 通过其内部的线程，调用SimpleMessenger 类注册的 Dispatch 类的处理函数来处理相应的消息。
class DispatchQueue { ...... mutable Mutex lock; Cond cond; class QueueItem { int type; ConnectionRef con;MessageRef m;...... }； PrioritizedQueue<QueueItem, uint64_t> mqueue; //基于优先级的 优先队列 set<pair<double, Message*> > marrival; 集合 (recv_time, message) map<Message *, set<pair<double, Message*> >::iterator> marrival_map; 消息 到 所在集合位置的 映射}；其内部的mqueue 为优先级队列，用来保存消息， marriavl 保存了接收到的消息。marrival_map 保存消息在 集合中的位置。
函数DispatchQueue::enqueue 用来把接收到的消息添加到 队列中,函数DispatchQueue::entry 为线程的处理函数，用于调用用户注册的Dispatcher类相应的处理函数来处理消息 。
类Pipe 是PipeConnection的具体实现类。其实现了两个端口之间类似管道的通信接口。
对于每一个pipe，内部有一个Reader线程 和 一个Writer 线程，分别用来处理有关这个Pipe的消息接收和发送。线程DelayedDelivery用于故障注入测试使用。
类Pipe的数据结构介绍如下：
SimpleMessenger *msgr; // msgr的指针uint64_t conn_id; //分配给Pipe 自己唯一的idchar *recv_buf; //接收缓存区int recv_max_prefetch; //接收缓冲区一次预期的最大值int recv_ofs; //接收的偏移量int recv_len; //接收的长度int sd; //pipe 对应的 socked fdstruct iovec msgvec[IOV_MAX]; //发送消息的 iovec 结构int port; //链接短裤int peer_type; //链接对方的类型entity_addr_t peer_addr; //对方地址Messenger::Policy policy; //策略Mutex pipe_lock;int state; //当前链接的状态atomic_t state_closed; // non-zero iff state = STATE_CLOSEDPipeConnectionRef connection_state; //PipeConnection 的引用utime_t backoff; // backoff timebool reader_running, reader_needs_join;bool reader_dispatching; /// reader thread is dispatching without pipe_lockbool notify_on_dispatch_done; /// something wants a signal when dispatch donebool writer_running;map<int, list<Message*> > out_q; // priority queue for outbound msgs 准备发送的消息 优先队列DispatchQueue *in_q; //接收消息的DispatchQueuelist<Message*> sent; //要发送的消息Cond cond;bool send_keepalive;bool send_keepalive_ack;utime_t keepalive_ack_stamp;bool halt_delivery; //if a pipe's queue is destroyed, stop adding to it__u32 connect_seq, peer_global_seq;uint64_t out_seq; 发送消息的序列号 uint64_t in_seq, in_seq_acked; 接收到消息序号和 应对的序号1．当发送一个消息时，首先要通过Messenger类，获取对应的 Connection
conn = messenger->get_connection(dest_server);
具体到 SimpleMessenger的实现：
2．当获得一个Connection之后，就可以调用 Connection 的 发送函数，发送消息
conn->send_message(m);
其最终调用了SimpleMessenger::submit_message 函数
3．发送线程writer把消息发送出去
通过步骤2，要发送的消息Messae已经保存在相应Pipe的out_q队列里。并触发了发送线程。每个Pipe的Writer 线程负责发送out_q 的消息，其线程入口函数为Pipe::writer, 实现功能：
接收消息,每个Pipe对应的线程 Reader 用于接收消息
其入口函数为 Pipe::reader， 其功能如下：
在这里，需要注意的是 ms_fast_dispath 和 ms_dispatch 两种处理的区别。ms_dispatch 是由DispatchQueue的线程处理的，它是一个单线程；ms_fast_dispatch的调用是由Pipe的接收线程直接处理的，因此性能比前者要好。
网络模块，最重要的是如何处理网络错误。无论是在接收消息还是发送消息的过程中，都会出现各种异常错误，包括返回异常错误码，接收数据的magic验证不一致，接收的数据的效验验证不一致等等。 错误的原因主要是网络本身的错误（物理链路等），或者字节跳变引起的。
处理的方法比较简单：
函数 Pipe::fault 用来处理错误
联系客服
