class Message : public RefCountedObject {
protected:
  ceph_msg_header  header;                 //消息头
  ceph_msg_footer  footer;                 //消息尾

  /* user_data*/
  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;

  /* 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需要的字段，当前只在XIO模式的实现中用到。
};

struct ceph_msg_header {
	__le64 seq;                    //当前session内消息的唯一序号
	__le64 tid;                    //事务ID
	__le16 type;                   //消息类型
	__le16 priority;               //优先级。值越大，消息的优先级越高
	__le16 version;                //消息编码的版本号

	__le32 front_len;             //bytes in main payload(主payload,也称为front playload, 主净荷数据的长度）
	__le32 middle_len;            //middle的长度
	__le32 data_len;              //data的长度
	__le16 data_off;              //对象数据的偏移量，通常需要CEPH_PAGE_SIZE对齐

	struct ceph_entity_name src;  //消息源，比如'mds0' or 'osd3'.

	/* oldest code we think can decode this.  unknown if zero. */
	__le16 compat_version;
	__le16 reserved;
	__le32 crc;                   //消息头的crc32c校验信息
} __attribute__ ((packed));

struct ceph_msg_footer {
	__le32 front_crc, middle_crc, data_crc;

	// sig holds the 64 bits of the digital signature for the message PLR
	__le64  sig;                 //消息的64位signature
	__u8 flags;                  //结束标志
} __attribute__ ((packed));

struct Connection : public RefCountedObject {
  mutable Mutex lock;                              //锁保护Connection的所有字段
  Messenger *msgr;                                 //该Connection对应的消息发送器： Messenger可以管理很多链接，并采用适当的策略来进行消息发送
  RefCountedObject *priv;                          链接的私有数据

  int peer_type;                                  //peer的类型，可以是CEPH_ENTITY_TYPE_MON、CEPH_ENTITY_TYPE_MDS等
  entity_addr_t peer_addr;                        //peer的地址

  //最后一次发送keepalive的时间和最后一次接收keepalive的ACK的时间
  utime_t last_keepalive, last_keepalive_ack;     


private:
  uint64_t features;                              //一些feature的标志位
public:
  bool failed;                                    //当值为true时，该链接为lossy链接已经失效了

  int rx_buffers_version;                         //接收缓冲区的版本

  //接收缓冲区。消息的标识ceph_tid --> (buffer, rx_buffers_version)的映射
  map<ceph_tid_t,pair<bufferlist,int> > rx_buffers;
};

  virtual int send_message(Message *m) = 0;

virtual bool ms_dispatch(Message *m) = 0;
virtual void ms_fast_dispatch(Message *m) { assert(0); };

virtual int send_message(Message *m, const entity_inst_t& dest) = 0;

void add_dispatcher_head(Dispatcher *d);

void add_dispatcher_tail(Dispatcher *d);

struct Policy {
    bool lossy;                     //如果为true，则当该连接出现错误时就删除
    
    bool server;                    //假如为true，则说明为服务端，则自己不能进行主动的重连操作
    
    bool standby;                   //假如为true，则当连接处于空闲状态时，处于standby状态
    
    bool resetcheck;                //假如为true，则在连接出错的情况下会进行重连
    
    //该connection(s)相应的流控操作
    Throttle *throttler_bytes;
    Throttle *throttler_messages;

    /// Specify features supported locally by the endpoint.
    uint64_t features_supported;

    /// Specify features any remotes must have to talk to this endpoint.
    uint64_t features_required;
};

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

messenger->set_magic(MSG_MAGIC_TRACE_CTR);
messenger->set_default_policy(
	Messenger::Policy::stateless_server(CEPH_FEATURES_ALL, 0));

r = messenger->bind(bind_addr);
if (r < 0)
	goto out;

// Set up crypto, daemonize, etc.
//global_init_daemonize(g_ceph_context, 0);
common_init_finish(g_ceph_context);

dispatcher = new SimpleDispatcher(messenger);

messenger->add_dispatcher_head(dispatcher); // should reach ready()

messenger->start();
messenger->wait();      // 本函数必须等start完成才能调用

messenger = Messenger::create(g_ceph_context, g_conf->ms_type,
				      entity_name_t::MON(-1),
				      "client",
				      getpid());

// enable timing prints
messenger->set_magic(MSG_MAGIC_TRACE_CTR);
messenger->set_default_policy(Messenger::Policy::lossy_client(0, 0));

dispatcher = new SimpleDispatcher(messenger);
messenger->add_dispatcher_head(dispatcher);

dispatcher->set_active(); // this side is the pinger

r = messenger->start();
if (r < 0)
	goto out;

conn = messenger->get_connection(dest_server);

int msg_ix;
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);
}

class SimpleMessenger : public SimplePolicyMessenger {
public:
  Accepter accepter;                //用于接受客户端的链接请求
  DispatchQueue dispatch_queue;     //接收到的请求的消息分发队列

  bool did_bind;                    //是否绑定
  

  __u32 global_seq;                //生成全局的消息seq
  ceph_spinlock_t global_seq_lock; //用于保护global_seq

  //addr与pipe的映射
  ceph::unordered_map<entity_addr_t, Pipe*> rank_pipe;

  //正在处理的pipes
  set<Pipe *> accepting_pipes;
  set<Pipe*>      pipes;           //所有的Pipes
  list<Pipe*>     pipe_reap_queue; //准备释放的Pipe列表
  int cluster_protocol;            //内部集群的协议版本
};

class Accepter : public Thread {
  SimpleMessenger *msgr;
  bool done;
  int listen_sd;            //所监听的socket句柄
  uint64_t nonce;

  ....
};

class DispatchQueue {
  class QueueItem {
    int type;
    ConnectionRef con;
    MessageRef m;
    ....
  };


  SimpleMessenger *msgr;
  mutable Mutex lock;
  Cond cond;

  PrioritizedQueue<QueueItem, uint64_t> mqueue;    //接收消息的优先队列

  set<pair<double, Message*> > marrival;           //接收到的消息集合。pair为(recv_time,message)

  //消息->所在集合位置的映射
  map<Message *, set<pair<double, Message*> >::iterator> marrival_map;
};

class Pipe : public RefCountedObject {
	class Reader : public Thread {
		...
	}reader_thread;                //接收线程，用于接收数据

	class Writer : public Thread {
		...
	}writer_thread;                //发送线程，用于发送数据

	
	SimpleMessenger *msgr;         //msgr的指针
	uint64_t conn_id;              //分配给Pipe自己唯一的id

	char *recv_buf;                //接收缓冲区
	size_t recv_max_prefetch;      //接收缓冲区一次预取的最大值
	size_t recv_ofs;               //接收的偏移量
	size_t recv_len;               //接收的长度

	int sd;                        //pipe对应的socket fd
	struct iovec msgvec[SM_IOV_MAX];   //发送消息的iovec结构

	int port;                      //链接端口
	int peer_type;                 //链接对方的类型： OSD、MON、MDS等
	entity_addr_t peer_addr;       //对方地址
	Messenger::Policy policy;      //策略

	Mutex pipe_lock;
	int state;                     //当前连接的状态
	atomic_t state_closed;        // non-zero iff state = STATE_CLOSED


	PipeConnectionRef connection_state;    //PipeConnection的引用
	utime_t backoff;              //backoff的时间

	map<int, list<Message*> > out_q;   //准备发送消息的优先级队列
	DispatchQueue *in_q;               //接收消息的DispatchQueue
	list<Message*> sent;               //要发送的消息
	Cond cond;
	bool send_keepalive;
	bool send_keepalive_ack;
	utime_t keepalive_ack_stamp;
	bool halt_delivery;                //如果Pipe队列销毁，停止增加

	__u32 connect_seq, peer_global_seq;
	uint64_t out_seq;                  //发送消息的序列号
	uint64_t in_seq, in_seq_acked;     //接收到消息序号和ACK的序号
};

conn = messenger->get_connection(dest_server);

conn->send_message(m);