int main(int argc, const char **argv) 
{
  ......

  Messenger *ms_hbclient = Messenger::create(g_ceph_context, g_conf->ms_type, // 发送ping心跳的messenger
					     entity_name_t::OSD(whoami), "hbclient",
					     getpid());
  Messenger *ms_hb_back_server = Messenger::create(g_ceph_context, g_conf->ms_type, // 接收来自back地址的ping心跳
						   entity_name_t::OSD(whoami), "hb_back_server",
						   getpid());
  Messenger *ms_hb_front_server = Messenger::create(g_ceph_context, g_conf->ms_type, // 接收来自front地址的ping心跳
						    entity_name_t::OSD(whoami), "hb_front_server",
						    getpid());
  ......
}

void OSD::heartbeat_entry()
{
  Mutex::Locker l(heartbeat_lock);
  if (is_stopping())
    return;

  while (!heartbeat_stop) {
    heartbeat(); // 发送消息
	......
  }
}

void OSD::heartbeat()
{
  ......

  // 遍历所有peers，发送心跳，peers集合的选取需要遵循一定规则
  for (map<int,HeartbeatInfo>::iterator i = heartbeat_peers.begin();
       i != heartbeat_peers.end();
       ++i) {
    int peer = i->first;
    i->second.last_tx = now;
    if (i->second.first_tx == utime_t())
      i->second.first_tx = now;
    i->second.con_back->send_message(new MOSDPing(monc->get_fsid(), // 向back地址发送
					  service.get_osdmap()->get_epoch(),
					  MOSDPing::PING,
					  now));

    if (i->second.con_front)
      i->second.con_front->send_message(new MOSDPing(monc->get_fsid(), // 向front地址发送
					     service.get_osdmap()->get_epoch(),
						     MOSDPing::PING,
						     now));
  }
  ......
}

struct HeartbeatDispatcher : public Dispatcher {
    OSD *osd;
    HeartbeatDispatcher(OSD *o) : Dispatcher(cct), osd(o) {}
    bool ms_dispatch(Message *m) {
      return osd->heartbeat_dispatch(m); // 消息处理函数
    }
	......
} heartbeat_dispatcher;

int OSD::init()
{
  ......

  // 初始化的时候注册dispatcher，收到消息后才知道怎么处理
  hbclient_messenger->add_dispatcher_head(&heartbeat_dispatcher);
  hb_front_server_messenger->add_dispatcher_head(&heartbeat_dispatcher);
  hb_back_server_messenger->add_dispatcher_head(&heartbeat_dispatcher);

  ......
}

bool OSD::heartbeat_dispatch(Message *m)
{
  switch (m->get_type()) {
  case CEPH_MSG_PING:
    m->put();
    break;

  case MSG_OSD_PING:
    handle_osd_ping(static_cast<MOSDPing*>(m)); // 处理心跳
    break;

  case CEPH_MSG_OSD_MAP: // 这个消息在heartbeat messenger内部是不会产生的
    {
      ConnectionRef self = cluster_messenger->get_loopback_connection();
      self->send_message(m);
    }
    break;

  default:
    m->put();
  }

  return true;
}

void OSD::handle_osd_ping(MOSDPing *m)
{
  ......

  switch (m->op) {

  case MOSDPing::PING: // 处理心跳消息
    {

	  ......

      // 当进程内部状态不正确的时候，丢弃心跳消息，此时处理心跳已经变得没有意义
	  // 很多线程池会设置timeout时间，如果超时状态就会是unhealthy
      if (!cct->get_heartbeat_map()->is_healthy()) {
		break;
      }

      Message *r = new MOSDPing(monc->get_fsid(),
				curmap->get_epoch(),
				MOSDPing::PING_REPLY, // 注意是PING_REPLY
				m->stamp);
      m->get_connection()->send_message(r); // 发送回包
	  ......

    }
    break;

  case MOSDPing::PING_REPLY: // 处理心跳回包
    {
	  ......
	
	  // 更新时间戳，避免心跳超时
	  // osd有专门的tick线程进行周期性的检查，如果发现有心跳超时的，就会上报monitor
      map<int,HeartbeatInfo>::iterator i = heartbeat_peers.find(from);
      if (i != heartbeat_peers.end()) {
		if (m->get_connection() == i->second.con_back) {
			i->second.last_rx_back = m->stamp;
		if (i->second.con_front == NULL)
			i->second.last_rx_front = m->stamp;
		} else if (m->get_connection() == i->second.con_front) {
			i->second.last_rx_front = m->stamp;
		}
      }

    }
    break;
	......
}

void OSD::tick()
{
	......

    heartbeat_lock.Lock();
    heartbeat_check(); // 检查心跳是否超时
    heartbeat_lock.Unlock();

	......
}

void OSD::heartbeat_check()
{
  ......

  for (map<int,HeartbeatInfo>::iterator p = heartbeat_peers.begin();
       p != heartbeat_peers.end();
       ++p) {
	if (p->second.is_unhealthy(cutoff)) { // 检测超时
      if (p->second.last_rx_back == utime_t() ||
			p->second.last_rx_front == utime_t()) {
		failure_queue[p->first] = p->second.last_tx; // 插入队列，等待上报给monitor
      } else {
		failure_queue[p->first] = MIN(p->second.last_rx_back, p->second.last_rx_front);
      }
    }
  }
}

void OSD::do_mon_report()
{
  ......
  send_failures();
  ......
}

void OSD::send_failures()
{
  ......
  while (!failure_queue.empty()) {
    int osd = failure_queue.begin()->first;
    int failed_for = (int)(double)(now - failure_queue.begin()->second);
    entity_inst_t i = osdmap->get_inst(osd);
    monc->send_mon_message(new MOSDFailure(monc->get_fsid(), i, failed_for, osdmap->get_epoch())); // 向monitor发送消息，报告osd心跳超时
    failure_pending[osd] = i;
    failure_queue.erase(osd);
  }
  ......
}

void OSD::maybe_update_heartbeat_peers()
{
  if (is_waiting_for_healthy()) { // 在osd启动的过程中，或者在osd收到更新osdmap的消息，osd状态可能变为waiting，此时需要更新peers集合
    utime_t now = ceph_clock_now(cct);
    if (last_heartbeat_resample == utime_t()) { // 第一次设置需要更新，这时候应该是osd刚启动
      last_heartbeat_resample = now;
      heartbeat_set_peers_need_update(); // 设置需要更新peers标志
    } else if (!heartbeat_peers_need_update()) { // 后续更新，应该是收到osdmap变更的消息
      utime_t dur = now - last_heartbeat_resample;
      if (dur > cct->_conf->osd_heartbeat_grace) { // 仅仅在超出grace时间后才更新，因为超过grace，osdmap的变更才可能导致pgmap变化
		heartbeat_set_peers_need_update(); // 设置需要更新peers标志
		last_heartbeat_resample = now;
		reset_heartbeat_peers();   // we want *new* peers!
      }
    }
  }

  Mutex::Locker l(heartbeat_lock);
  if (!heartbeat_peers_need_update())
    return; // 不需要更新直接返回
  heartbeat_need_update = false;

  heartbeat_epoch = osdmap->get_epoch();
  if (is_active()) { // 需要osd状态是active，不然更新没意义
    RWLock::RLocker l(pg_map_lock);
    for (ceph::unordered_map<spg_t, PG*>::iterator i = pg_map.begin(); // 遍历osd负责的所有pg
	 i != pg_map.end();
	 ++i) {
      PG *pg = i->second;
      pg->heartbeat_peer_lock.Lock();

      for (set<int>::iterator p = pg->heartbeat_peers.begin(); // 遍历pg对应的peers
	   p != pg->heartbeat_peers.end();
	   ++p)
		if (osdmap->is_up(*p)) // 如果为up，则加入心跳集合
			_add_heartbeat_peer(*p);

      for (set<int>::iterator p = pg->probe_targets.begin(); // 遍历probe目标集合
	   p != pg->probe_targets.end();
	   ++p)
		if (osdmap->is_up(*p)) // 如果为up，则加入心跳集合
			_add_heartbeat_peer(*p);

      pg->heartbeat_peer_lock.Unlock();
    }
  }

  // 后面流程就比较简单
  // 1) 加入'仅挨着'当前osd编号的下一个和上一个为up的节点
  // 2) 删除down的节点
  // 3) 对peers集合做调整
  ......
}

void PG::update_heartbeat_peers()
{
  ......
	  
  bool need_update = false;
  heartbeat_peer_lock.Lock();
  if (new_peers == heartbeat_peers) {
  } else {
    heartbeat_peers.swap(new_peers);
    need_update = true; // 需要update
  }

  if (need_update)
    osd->need_heartbeat_peer_update(); // 更新
}

class HeartbeatMap {
 public:
  // register/unregister
  heartbeat_handle_d *add_worker(std::string name); // 注册handler
  void remove_worker(heartbeat_handle_d *h);

 private:
  CephContext *m_cct;
  RWLock m_rwlock;
  time_t m_inject_unhealthy_until;
  std::list<heartbeat_handle_d*> m_workers; // 注册的所有timeout handler集合

  .....
  bool _check(heartbeat_handle_d *h, const char *who, time_t now); // 检查是否超时
};

struct heartbeat_handle_d {
  std::string name;
  atomic_t timeout, suicide_timeout;
  time_t grace, suicide_grace; // 超过grace时间，表示timeout；超过suicide_grace，进程会退出
  std::list<heartbeat_handle_d*>::iterator list_item;

  heartbeat_handle_d(const std::string& n)
    : name(n), grace(0), suicide_grace(0)
  { }
};