// information about a heartbeat peer
//这个类 用来记录 peer osd 信息以及 关于heartbeat 的操作函数都在这
struct HeartbeatInfo {
 int peer;           ///< peer
    ConnectionRef con_front;   ///< peer connection (front)
    ConnectionRef con_back;    ///< peer connection (back)
    utime_t first_tx;   ///< time we sent our first ping request
    utime_t last_tx;    ///< last time we sent a ping request
    utime_t last_rx_front;  ///< last time we got a ping reply on the front side
    utime_t last_rx_back;   ///< last time we got a ping reply on the back side
    epoch_t epoch;      ///< most recent epoch we wanted this peer
    /// number of connections we send and receive heartbeat pings/replies
    static constexpr int HEARTBEAT_MAX_CONN = 2;
    /// history of inflight pings, arranging by timestamp we sent
    /// send time -> deadline -> remaining replies
    map<utime_t, pair<utime_t, int>> ping_history;

    //这个map保存了 peer osd的heartbeat_peers，key是 osd的 id
    map<int,HeartbeatInfo> heartbeat_peers;  ///< map of osd id to HeartbeatInfo
//  ....
 }

// osd 接受心跳消息的handle 函数
// 在 class OSD 里面
struct HeartbeatDispatcher : public Dispatcher {
    OSD *osd;
    explicit HeartbeatDispatcher(OSD *o) : Dispatcher(o->cct), osd(o) {}
    void ms_fast_dispatch(Message *m) override {
      osd->heartbeat_dispatch(m);
    }
    bool ms_dispatch(Message *m) override {
      return osd->heartbeat_dispatch(m);
    }
//........................
    KeyStore *ms_get_auth1_authorizer_keystore() override {
      return osd->ms_get_auth1_authorizer_keystore();
    }
  } heartbeat_dispatcher;

# ceph_osd.cc
// 创建四个关于心跳的 message
int main(int argc, const char **argv) {
    Messenger *ms_hb_back_client = Messenger::create(g_ceph_context, cluster_msg_type,
  					     entity_name_t::OSD(whoami), "hb_back_client",
  					     getpid(), Messenger::HEARTBEAT);
    Messenger *ms_hb_front_client = Messenger::create(g_ceph_context, public_msg_type,
  					     entity_name_t::OSD(whoami), "hb_front_client",
  					     getpid(), Messenger::HEARTBEAT);
    Messenger *ms_hb_back_server = Messenger::create(g_ceph_context, cluster_msg_type,
  						   entity_name_t::OSD(whoami), "hb_back_server",
  						   getpid(), Messenger::HEARTBEAT);
    Messenger *ms_hb_front_server = Messenger::create(g_ceph_context, public_msg_type,
  						    entity_name_t::OSD(whoami), "hb_front_server",
  						    getpid(), Messenger::HEARTBEAT)；
//osd 初始化的类其参数就需要 关于 心跳的 Message　实例
osd = new OSD(g_ceph_context,
                  store,
                  whoami,
                  ms_cluster,
                  ms_public,
                  ms_hb_front_client,
                  ms_hb_back_client,
                  ms_hb_front_server,
                  ms_hb_back_server,
                  ms_objecter,
                  &mc,
                  data_path,
                  journal_path);

// 启动 message
    ms_hb_front_client->start();
    ms_hb_back_client->start();
    ms_hb_front_server->start();
    ms_hb_back_server->start();
}

//解
struct T_Heartbeat : public Thread {
    OSD *osd;
    explicit T_Heartbeat(OSD *o) : osd(o) {}
    void *entry() override {
        osd->heartbeat_entry();
        return 0;
    }
} heartbeat_thread;
int OSD::init()
{
// start the heartbeat thread
//传入的参数 是线程的名字
  heartbeat_thread.create("osd_srv_heartbt");
//....
}
void OSD::heartbeat_entry()
{
  std::lock_guard l(heartbeat_lock);
  if (is_stopping())
    return;
  while (!heartbeat_stop) {
    heartbeat();

    double wait;
    //等待的时间分为两种，一种是固定的 osd_heartbeat_interval，另外一种是 基于osd_heartbeat_interval生成的随机时间
    if (cct->_conf.get_val<bool>("debug_disable_randomized_ping")) {
      wait = (float)cct->_conf->osd_heartbeat_interval;
    } else {
      wait = .5 + ((float)(rand() % 10)/10.0) * (float)cct->_conf->osd_heartbeat_interval;
    }
    utime_t w;
    //转换成 utime_t
    w.set_from_double(wait);
    dout(30) << "heartbeat_entry sleeping for " << wait << dendl;

    // 这里用条件变量来等待时间，并没有释放锁
    heartbeat_cond.WaitInterval(heartbeat_lock, w);
    if (is_stopping())
      return;
    dout(30) << "heartbeat_entry woke up" << dendl;

  }
}

void OSD::heartbeat()
{
  ceph_assert(heartbeat_lock.is_locked_by_me());
  dout(30) << "heartbeat" << dendl;

  // get CPU load avg
  double loadavgs[1];
  int hb_interval = cct->_conf->osd_heartbeat_interval;
  int n_samples = 86400;
  if (hb_interval > 1) {
    n_samples /= hb_interval;
    if (n_samples < 1)
      n_samples = 1;
  }
  cout << "send times :  " << n_samples << std::endl;
  // 估计下每次发送osd 负载值？？？？
  if (getloadavg(loadavgs, 1) == 1) {
    cout <<loadavgs[0] << std::endl;
    logger->set(l_osd_loadavg, 100 * loadavgs[0]);
    daily_loadavg = (daily_loadavg * (n_samples - 1) + loadavgs[0]) / n_samples;

    dout(1) << "heartbeat: daily_loadavg " << daily_loadavg << dendl;
  }

  dout(1) << "heartbeat checking stats" << dendl;

  // refresh peer list and osd stats
  // hb_peers保存 id（这里是 osd 的id）
  vector<int> hb_peers;

  //遍历 peer osd, peer osd 如何更新在下面会说到
  for (map<int,HeartbeatInfo>::iterator p = heartbeat_peers.begin();
       p != heartbeat_peers.end();
       ++p)
    hb_peers.push_back(p->first);

  // 更新状态 osd 状态
  auto new_stat = service.set_osd_stat(hb_peers, get_num_pgs());
  dout(5) << __func__ << " " << new_stat << dendl;
  ceph_assert(new_stat.statfs.total);

// 这比例指 osd 使用空间变更比例
  float pratio;
  float ratio = service.compute_adjusted_ratio(new_stat, &pratio);
  std::cout << "ratio :  " << ratio << std::endl;
  service.check_full_status(ratio, pratio);

//获取当前实际  精确到纳秒
  utime_t now = ceph_clock_now();
  utime_t deadline = now;

//deadline  最大等待回应的时间
  deadline += cct->_conf->osd_heartbeat_grace;
  std::cout << "now " << now << std::endl;
  std::cout << "deadline " << deadline << std::endl;

  // send heartbeats
  // 遍历 对等 peer，逐个发送
  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;
    // 保存发送 heartbeat 记录，HEARTBEAT_MAX_CONN 为发送heartbeat 数量
    //key 发送的时间和， value 为 心跳回复的期限
    i->second.ping_history[now] = make_pair(deadline,
      HeartbeatInfo::HEARTBEAT_MAX_CONN);

    if (i->second.hb_interval_start == utime_t())
      i->second.hb_interval_start = now;

    dout(1) << "heartbeat sending ping to osd." << peer << dendl;
    //  集群内发送
    i->second.con_back->send_message(new MOSDPing(monc->get_fsid(),
					  service.get_osdmap_epoch(),
					  MOSDPing::PING, now,
					  cct->_conf->osd_heartbeat_min_size));

   // 如果有 设置 public network
    if (i->second.con_front)
      i->second.con_front->send_message(new MOSDPing(monc->get_fsid(),
					     service.get_osdmap_epoch(),
					     MOSDPing::PING, now,
					  cct->_conf->osd_heartbeat_min_size));
  }
//  MOSDPing::PING 心跳信息的类型，还有
  logger->set(l_osd_hb_to, heartbeat_peers.size());
// 如果集群只有一个 osd，他会在 osd_mon_heartbeat_interval 时间内，向mon获取新的 osdmap
  // hmm.. am i all alone?
  dout(30) << "heartbeat lonely?" << dendl;
  if (heartbeat_peers.empty()) {
    if (now - last_mon_heartbeat > cct->_conf->osd_mon_heartbeat_interval && is_active()) {
      last_mon_heartbeat = now;
      dout(1) << "i have no heartbeat peers; checking mon for new map" << dendl;
      osdmap_subscribe(get_osdmap_epoch() + 1, false);
    }
  }

  dout(30) << "heartbeat done" << dendl;
}

// tick
//get_tick_interval生成随机时间 在1s上下，不想让他太频繁
double OSD::get_tick_interval() const
{
  // vary +/- 5% to avoid scrub scheduling livelocks
  constexpr auto delta = 0.05;
  return (OSD_TICK_INTERVAL *
	  ceph::util::generate_random_number(1.0 - delta, 1.0 + delta));
}
// 在 int OSD::init()中有一个  tick_timer.add_event_after
// 从函数名 可看出 添加一个时间，在 get_tick_interval() 后执行事件
  tick_timer.add_event_after(get_tick_interval(),
			     new C_Tick(this));
// 从 OSD::C_Tick 可以看出， 执行的是 osd->tick() 函数
class OSD::C_Tick : public Context {
  OSD *osd;
  public:
  explicit C_Tick(OSD *o) : osd(o) {}
  void finish(int r) override {
    osd->tick();
  }
};

//对几种情况
void OSD::tick()
{
  ceph_assert(osd_lock.is_locked());
  dout(10) << "tick" << dendl;
//等待 osd心跳健康，
  if (is_active() || is_waiting_for_healthy()) {
 //更新 heartbeat_peers
    maybe_update_heartbeat_peers();
  }

  if (is_waiting_for_healthy()) {
    start_boot();
  }
//更新 osdmap epoch
  if (is_waiting_for_healthy() || is_booting()) {
    std::lock_guard l(heartbeat_lock);
    utime_t now = ceph_clock_now();
    if (now - last_mon_heartbeat > cct->_conf->osd_mon_heartbeat_interval) {
      last_mon_heartbeat = now;
      dout(1) << __func__ << " checking mon for new map" << dendl;
      osdmap_subscribe(get_osdmap_epoch() + 1, false);
    }
  }

  do_waiters();
//继续添加 事件，循环调用  OSD::tick()
  tick_timer.add_event_after(get_tick_interval(), new C_Tick(this));
}

void OSD::maybe_update_heartbeat_peers()
{
  ceph_assert(osd_lock.is_locked());
  if (is_waiting_for_healthy() || is_active()) {
    utime_t now = ceph_clock_now();
    //第一次启动时候
    if (last_heartbeat_resample == utime_t()) {
      last_heartbeat_resample = now;
      heartbeat_set_peers_need_update();
    } else if (!heartbeat_peers_need_update()) {
      utime_t dur = now - last_heartbeat_resample;
      //仅仅在超出grace时间后才更新
      if (dur > cct->_conf->osd_heartbeat_grace) {
	dout(10) << "maybe_update_heartbeat_peers forcing update after " << dur << " seconds" << dendl;
        // 更新 peer的标志位
	heartbeat_set_peers_need_update();
        ////
	last_heartbeat_resample = now;
	// automatically clean up any stale heartbeat peers
	// if we are unhealthy, then clean all
        //清空 peers
	reset_heartbeat_peers(is_waiting_for_healthy());
      }
    }
  }

// ............
  // build heartbeat from set
  // 从同一个 pg下的osd中 获取 peer
  if (is_active()) {
    vector<PGRef> pgs;
    _get_pgs(&pgs);
    for (auto& pg : pgs) {
      pg->with_heartbeat_peers([&](int peer) {
	  if (get_osdmap()->is_up(peer)) {
	    _add_heartbeat_peer(peer);
	  }
	});
    }
  }

  // include next and previous up osds to ensure we have a fully-connected set
  //want: 保存 osd的相邻osd id，以防止 peer数量不够
  set<int> want, extras;
  const int next = get_osdmap()->get_next_up_osd_after(whoami);
  if (next >= 0)
    want.insert(next);
  int prev = get_osdmap()->get_previous_up_osd_before(whoami);
  if (prev >= 0 && prev != next)
    want.insert(prev);

  // make sure we have at least **min_down** osds coming from different
  // subtree level (e.g., hosts) for fast failure detection.
  // 确保至少有 min_down 个osd peer（理论上）， 从 level 故障域 和  min_down 从整个集群 随机抽取两个osd
  auto min_down = cct->_conf.get_val<uint64_t>("mon_osd_min_down_reporters");
  auto subtree = cct->_conf.get_val<string>("mon_osd_reporter_subtree_level");

  get_osdmap()->get_random_up_osds_by_subtree(
    whoami, subtree, min_down, want, &want);

  for (set<int>::iterator p = want.begin(); p != want.end(); ++p) {
    dout(10) << " adding neighbor peer osd." << *p << dendl;
    extras.insert(*p);
    _add_heartbeat_peer(*p);
  }
  //删除已经down的osd
  // remove down peers; enumerate extras
  map<int,HeartbeatInfo>::iterator p = heartbeat_peers.begin();
  while (p != heartbeat_peers.end()) {
    if (!get_osdmap()->is_up(p->first)) {
      int o = p->first;
      ++p;
      _remove_heartbeat_peer(o);
      continue;
    }
    if (p->second.epoch < get_osdmap_epoch()) {
      extras.insert(p->first);
    }
    ++p;
  }

  // too few?
  //peer数量过少，默认 10个，从 want 里面添加
  for (int n = next; n >= 0; ) {
    if ((int)heartbeat_peers.size() >= cct->_conf->osd_heartbeat_min_peers)
      break;
    if (!extras.count(n) && !want.count(n) && n != whoami) {
      dout(10) << " adding random peer osd." << n << dendl;
      extras.insert(n);
      _add_heartbeat_peer(n);
    }
    n = get_osdmap()->get_next_up_osd_after(n);
    if (n == next)
      break;  // came full circle; stop
  }

  // too many?
  for (set<int>::iterator p = extras.begin();
       (int)heartbeat_peers.size() > cct->_conf->osd_heartbeat_min_peers && p != extras.end();
       ++p) {
    if (want.count(*p))
      continue;
    _remove_heartbeat_peer(*p);
  }

  dout(10) << "maybe_update_heartbeat_peers " << heartbeat_peers.size() << " peers, extras " << extras << dendl;
}

void OSD::handle_pg_create(OpRequestRef op)
{
  const MOSDPGCreate *m = static_cast<const MOSDPGCreate*>(op->get_req());
  ceph_assert(m->get_type() == MSG_OSD_PG_CREATE);

  dout(10) << "handle_pg_create " << *m << dendl;
 //........................

  maybe_update_heartbeat_peers();
}

void OSD::handle_osd_map(MOSDMap *m)
{
  //..................
  if (is_active() || is_waiting_for_healthy())
    maybe_update_heartbeat_peers();
  //..................
}

//heartbeat_dispatcher 是一个结构体，里面有处理心跳信息的函数

int OSD::init()
{
...
//注册dispatcher， heartbeat_dispatcher 处理心跳信息的函数
hb_front_client_messenger->add_dispatcher_head(&heartbeat_dispatcher);
hb_back_client_messenger->add_dispatcher_head(&heartbeat_dispatcher);
hb_front_server_messenger->add_dispatcher_head(&heartbeat_dispatcher);
hb_back_server_messenger->add_dispatcher_head(&heartbeat_dispatcher);
...
}

// heartbeat_dispatch
// osd 收到心跳消息 的类型是 MSG_OSD_PING ，执行 handle_osd_ping
bool OSD::heartbeat_dispatch(Message *m)
{
  dout(30) << "heartbeat_dispatch " << m << dendl;
  switch (m->get_type()) {

  case CEPH_MSG_PING:
    dout(10) << "ping from " << m->get_source_inst() << dendl;
    m->put();
    break;

  case MSG_OSD_PING:
    handle_osd_ping(static_cast<MOSDPing*>(m));
    break;

  default:
    dout(0) << "dropping unexpected message " << *m << " from " << m->get_source_inst() << dendl;
    m->put();
  }

  return true;
}

//收到的是心跳通知
  case MOSDPing::PING:
    {
      //debug 模式
     //这里判断是否超时
      if (!cct->get_heartbeat_map()->is_healthy()) {
		dout(10) << "internal heartbeat not healthy, dropping ping request" << dendl;
		break;
      }
      //回复 心跳通知
      Message *r = new MOSDPing(monc->get_fsid(),
				curmap->get_epoch(),
				MOSDPing::PING_REPLY, m->stamp,
				cct->_conf->osd_heartbeat_min_size);
      m->get_connection()->send_message(r);
      //验证 发送方的osd是否 为 up
      if (curmap->is_up(from)) {
      //  从收到消息中 更新osd peer的epoch
	service.note_peer_epoch(from, m->map_epoch);
	if (is_active()) {
	  ConnectionRef con = service.get_con_osd_cluster(from, curmap->get_epoch());
	  if (con) {
	    service.share_map_peer(from, con.get());
	  }
	}
      //没有在 osdmap中发现你
      } else if (!curmap->exists(from) ||
		 curmap->get_down_at(from) > m->map_epoch) {
	// tell them they have died
	Message *r = new MOSDPing(monc->get_fsid(),
				  curmap->get_epoch(),
				  MOSDPing::YOU_DIED,
				  m->stamp,
				  cct->_conf->osd_heartbeat_min_size);
	m->get_connection()->send_message(r);
      }
    }

int OSD::init()
{
    //.......
  tick_timer.add_event_after(get_tick_interval(),
			     new C_Tick(this));
  {
    std::lock_guard l(tick_timer_lock);
   //每隔 get_tick_interval()时间  调C_Tick_WithoutOSDLock类里的 osd->tick_without_osd_lock() 函数
    tick_timer_without_osd_lock.add_event_after(get_tick_interval(),
						new C_Tick_WithoutOSDLock(this));
  }
    //.......
}

class OSD::C_Tick_WithoutOSDLock : public Context {
  OSD *osd;
  public:
  explicit C_Tick_WithoutOSDLock(OSD *o) : osd(o) {}
  void finish(int r) override {
    osd->tick_without_osd_lock();
  }
};

void OSD::tick_without_osd_lock()
{
 //......
  if (is_active() || is_waiting_for_healthy()) {
    heartbeat_lock.Lock();
    //关键函数 heartbeat_check
    //heartbeat_check 会将 有错误的peer都会记录到 failure_queue
    heartbeat_check();
    heartbeat_lock.Unlock();

    map_lock.get_read();
    std::lock_guard l(mon_report_lock);

    // mon report?
    utime_t now = ceph_clock_now();
    if (service.need_fullness_update() ||
	now - last_mon_report > cct->_conf->osd_mon_report_interval) {
      last_mon_report = now;
      send_full_update();
      //send_failures 会遍历 failure_queue，然后将错误的消息发送给 mon
      send_failures();
    }
    map_lock.put_read();

    epoch_t max_waiting_epoch = 0;
    for (auto s : shards) {
      max_waiting_epoch = std::max(max_waiting_epoch,
				   s->get_max_waiting_epoch());
    }
    if (max_waiting_epoch > get_osdmap()->get_epoch()) {
      dout(20) << __func__ << " max_waiting_epoch " << max_waiting_epoch
	       << ", requesting new map" << dendl;
      osdmap_subscribe(superblock.newest_map + 1, false);
    }
  }

  if (is_active()) {
 .....
    const auto now = ceph::coarse_mono_clock::now();
    {
//    周期性的向 mon报告
        std::lock_guard l{min_last_epoch_clean_lock};
      const auto elapsed = now - last_sent_beacon;
      if (chrono::duration_cast<chrono::seconds>(elapsed).count() >
        cct->_conf->osd_beacon_report_interval) {
        need_send_beacon = true;
      }
    }
    if (need_send_beacon) {
//将自己当前的 eponch 发送给mon
      send_beacon(now);
    }
  }
....
//更新 定时器任务
  tick_timer_without_osd_lock.add_event_after(get_tick_interval(),
					      new C_Tick_WithoutOSDLock(this));
}

void OSD::heartbeat_check()
{
  ceph_assert(heartbeat_lock.is_locked());
  utime_t now = ceph_clock_now();

  // check for incoming heartbeats (move me elsewhere?)
  // 遍历 heartbeat_peers
  for (map<int,HeartbeatInfo>::iterator p = heartbeat_peers.begin();
       p != heartbeat_peers.end();
       ++p) {

    if (p->second.first_tx == utime_t()) {
      dout(30) << "heartbeat_check we haven't sent ping to osd." << p->first
               << " yet, skipping" << dendl;
      continue;
    }
.....
    // 收到回复后，都会根据接收时间 记录 ping_history里
    if (p->second.is_unhealthy(now)) {
      utime_t oldest_deadline = p->second.ping_history.begin()->second.first;
      //
      if (p->second.last_rx_back == utime_t() ||
	  p->second.last_rx_front == utime_t()) {
	// fail
        //有错误的peer都会记录到 failure_queue，会统一发送给mon
	failure_queue[p->first] = p->second.first_tx;
      } else {
.....
	// fail
	failure_queue[p->first] = std::min(p->second.last_rx_back, p->second.last_rx_front);
      }
    }
  }
}

bool PaxosService::dispatch(MonOpRequestRef op)
{
........
  // 预处理 操作
  // preprocess
  if (preprocess_query(op))
    return true;  // easy!
...........
.......
  // update
  if (!prepare_update(op)) {
    // no changes made.
    return true;
  }

  if (need_immediate_propose) {
    dout(10) << __func__ << " forced immediate propose" << dendl;
    need_immediate_propose = false;
    propose_pending();
    return true;
  }

......
}

bool OSDMonitor::preprocess_query(MonOpRequestRef op)
{
  op->mark_osdmon_event(__func__);
  Message *m = op->get_req();
  switch (m->get_type()) {
    // READs
.....
    // damp updates
  case MSG_OSD_MARK_ME_DOWN:
    return preprocess_mark_me_down(op);
  case MSG_OSD_FULL:
    return preprocess_full(op);
  case MSG_OSD_FAILURE:
    return preprocess_failure(op);
......
  default:
    ceph_abort();
    return true;
  }
}

bool OSDMonitor::preprocess_failure(MonOpRequestRef op)
{
  op->mark_osdmon_event(__func__);
  MOSDFailure *m = static_cast<MOSDFailure*>(op->get_req());
  // who is target_osd
  //获取osd id
  int badboy = m->get_target_osd();
  // check permissions
  if (check_source(op, m->fsid))
    goto didit;
  dout(5) << "osd test "<< badboy << dendl;
  // first, verify the reporting host is valid
  //先从 osdmap中验证有无这个osd
  if (m->get_orig_source().is_osd()) {
    int from = m->get_orig_source().num();
    if (!osdmap.exists(from) ||
	!osdmap.get_addrs(from).legacy_equals(m->get_orig_source_addrs()) ||
	(osdmap.is_down(from) && m->if_osd_failed())) {
      dout(5) << "preprocess_failure from dead osd." << from
	      << ", ignoring" << dendl;
      send_incremental(op, m->get_epoch()+1);
      goto didit;
    }
  }
  // weird?
  //看osd状态
  if (osdmap.is_down(badboy)) {
    dout(5) << "preprocess_failure dne(/dup?): osd." << m->get_target_osd()
	    << " " << m->get_target_addrs()
	    << ", from " << m->get_orig_source() << dendl;
    if (m->get_epoch() < osdmap.get_epoch())
      send_incremental(op, m->get_epoch()+1);
    goto didit;
  }
  //核查上报错误osd 的地址
  if (osdmap.get_addrs(badboy) != m->get_target_addrs()) {
    dout(5) << "preprocess_failure wrong osd: report osd." << m->get_target_osd()
	    << " " << m->get_target_addrs()
	    << " != map's " << osdmap.get_addrs(badboy)
	    << ", from " << m->get_orig_source() << dendl;
    if (m->get_epoch() < osdmap.get_epoch())
      send_incremental(op, m->get_epoch()+1);
    goto didit;
  }

  // already reported?
  //再次验证 验证epoch
  if (osdmap.is_down(badboy) ||
      osdmap.get_up_from(badboy) > m->get_epoch()) {
    dout(5) << "preprocess_failure dup/old: osd." << m->get_target_osd()
	    << " " << m->get_target_addrs()
	    << ", from " << m->get_orig_source() << dendl;
    if (m->get_epoch() < osdmap.get_epoch())
      send_incremental(op, m->get_epoch()+1);
    goto didit;
  }

  if (!can_mark_down(badboy)) {
    dout(5) << "preprocess_failure ignoring report of osd."
	    << m->get_target_osd() << " " << m->get_target_addrs()
	    << " from " << m->get_orig_source() << dendl;
    goto didit;
 }
  return false;
 didit:
  mon->no_reply(op);
  return true;
}

bool OSDMonitor::prepare_update(MonOpRequestRef op)
{
  op->mark_osdmon_event(__func__);
  Message *m = op->get_req();
  dout(7) << "prepare_update " << *m << " from " << m->get_orig_source_inst() << dendl;

  switch (m->get_type()) {
    // damp updates
 .....
  case MSG_OSD_FAILURE:
    return prepare_failure(op);
 .....
  }

  return false;
}

bool OSDMonitor::prepare_failure(MonOpRequestRef op)
{
  op->mark_osdmon_event(__func__);
  //转换消息类型
  MOSDFailure *m = static_cast<MOSDFailure*>(op->get_req());
  dout(1) << "prepare_failure osd." << m->get_target_osd()
	  << " " << m->get_target_addrs()
	  << " from " << m->get_orig_source()
          << " is reporting failure:" << m->if_osd_failed() << dendl;
 //获取上报的osd
  int target_osd = m->get_target_osd();
 //
  int reporter = m->get_orig_source().num();
  ceph_assert(osdmap.is_up(target_osd));
  ceph_assert(osdmap.get_addrs(target_osd) == m->get_target_addrs());

  mon->no_reply(op);

  if (m->if_osd_failed()) {
    // calculate failure time
    utime_t now = ceph_clock_now();
    utime_t failed_since =
      m->get_recv_stamp() - utime_t(m->failed_for, 0);

    // add a report
    if (m->is_immediate()) {
      mon->clog->debug() << "osd." << m->get_target_osd()
			 << " reported immediately failed by "
			 << m->get_orig_source();
      force_failure(target_osd, reporter);
      return true;
    }
    mon->clog->debug() << "osd." << m->get_target_osd() << " reported failed by "
		      << m->get_orig_source();

    failure_info_t& fi = failure_info[target_osd];

     // 在这个添加失败的报告
    MonOpRequestRef old_op = fi.add_report(reporter, failed_since, op);
    if (old_op) {
      mon->no_reply(old_op);
    }

    return check_failure(now, target_osd, fi);
  } else {
    // remove the report
    mon->clog->debug() << "osd." << m->get_target_osd()
		       << " failure report canceled by "
		       << m->get_orig_source();
    if (failure_info.count(target_osd)) {
      failure_info_t& fi = failure_info[target_osd];
      MonOpRequestRef report_op = fi.cancel_report(reporter);
      if (report_op) {
        mon->no_reply(report_op);
      }
      if (fi.reporters.empty()) {
	dout(10) << " removing last failure_info for osd." << target_osd
		 << dendl;
	failure_info.erase(target_osd);
      } else {
	dout(10) << " failure_info for osd." << target_osd << " now "
		 << fi.reporters.size() << " reporters" << dendl;
      }
    } else {
      dout(10) << " no failure_info for osd." << target_osd << dendl;
    }
  }

  return false;
}

bool OSDMonitor::check_failure(utime_t now, int target_osd, failure_info_t& fi)
{
  // already pending failure?
  if (pending_inc.new_state.count(target_osd) &&
      pending_inc.new_state[target_osd] & CEPH_OSD_UP) {
    dout(10) << " already pending failure" << dendl;
    return true;
  }

  set<string> reporters_by_subtree;
  //故障域
  auto reporter_subtree_level = g_conf().get_val<string>("mon_osd_reporter_subtree_level");
  //
  utime_t orig_grace(g_conf()->osd_heartbeat_grace, 0);
  utime_t max_failed_since = fi.get_failed_since();
  utime_t failed_for = now - max_failed_since;

  utime_t grace = orig_grace;
  double my_grace = 0, peer_grace = 0;
  double decay_k = 0;

  //根据以往 osd的错误报告来 调整下  osd_heartbeat_grace 时间
  if (g_conf()->mon_osd_adjust_heartbeat_grace) {
 // decay_k 为 -0.000192541     ？？？
 //::log(.5)  -0.693147
    double halflife = (double)g_conf()->mon_osd_laggy_halflife;
    decay_k = ::log(.5) / halflife;

    // scale grace period based on historical probability of 'lagginess'
    //根据历史上的 "滞后性 "概率确定宽限期的规模
    // (false positive failures due to slowness).

    const osd_xinfo_t& xi = osdmap.get_xinfo(target_osd);
    //计算 log ？
    double decay = exp((double)failed_for * decay_k);
    dout(1) << " halflife " << halflife << " decay_k " << decay_k
	     << " failed_for " << failed_for << " decay " << decay << dendl;

    my_grace = decay * (double)xi.laggy_interval * xi.laggy_probability;
    grace += my_grace;
  }

  // consider the peers reporting a failure a proxy for a potential
  // 'subcluster' over the overall cluster that is similarly
  // laggy.  this is clearly not true in all cases, but will sometimes
  // help us localize the grace correction to a subset of the system
  // (say, a rack with a bad switch) that is unhappy.
  ceph_assert(fi.reporters.size());
  //遍历上报错误
  for (auto p = fi.reporters.begin(); p != fi.reporters.end();) {
    // get the parent bucket whose type matches with "reporter_subtree_level".
    // fall back to OSD if the level doesn't exist.
    if (osdmap.exists(p->first)) {
      //通过osdmap 获取完整的osd位置
      auto reporter_loc = osdmap.crush->get_full_location(p->first);
      //不在 指定的故障域
      if (auto iter = reporter_loc.find(reporter_subtree_level);
          iter == reporter_loc.end()) {
        reporters_by_subtree.insert("osd." + to_string(p->first));

        dout(1) <<"reporters_by_subtree, reporter_subtree_level:" <<reporter_subtree_level
                       << " vname:" << "osd." + to_string(p->first) << dendl;

      } else {
        string name = get_phy_name(iter->second);
        dout(1) <<"reporters_by_subtree, reporter_subtree_level:" <<reporter_subtree_level
                       << " vname:" << iter->second << " pname:" << name << dendl;
         reporters_by_subtree.insert(name);
      }
      if (g_conf()->mon_osd_adjust_heartbeat_grace) {
        const osd_xinfo_t& xi = osdmap.get_xinfo(p->first);
        //距离上次标记down的时间
        utime_t elapsed = now - xi.down_stamp;
        double decay = exp((double)elapsed * decay_k);
        //被标记为滞后和恢复的平均时间间隔
        peer_grace += decay * (double)xi.laggy_interval * xi.laggy_probability;
      }
      ++p;
    } else {
      fi.cancel_report(p->first);;
      p = fi.reporters.erase(p);
    }
  }

  if (g_conf()->mon_osd_adjust_heartbeat_grace) {
    peer_grace /= (double)fi.reporters.size();
    grace += peer_grace;
  }

  dout(10) << " osd." << target_osd << " has "
	   << fi.reporters.size() << " reporters, "
	   << grace << " grace (" << orig_grace << " + " << my_grace
	   << " + " << peer_grace << "), max_failed_since " << max_failed_since
	   << dendl;

// 以上都是在调整 grace的时间

// 实际 osd报错的时间 和 调整过的grace 时间做比较
// 以及报错的上报数量
  if (failed_for >= grace &&
      reporters_by_subtree.size() >= g_conf().get_val<uint64_t>("mon_osd_min_down_reporters")) {
    dout(1) << " we have enough reporters to mark osd." << target_osd
	    << " down" << dendl;
    // 将目标状态标记为 CEPH_OSD_UP，看注释是说 XORed onto previous state，异或？
    pending_inc.new_state[target_osd] = CEPH_OSD_UP;

    mon->clog->info() << "osd." << target_osd << " failed ("
		      << osdmap.crush->get_full_location_ordered_string(
			target_osd)
		      << ") ("
		      << (int)reporters_by_subtree.size()
		      << " reporters from different "
		      << reporter_subtree_level << " after "
		      << failed_for << " >= grace " << grace << ")";
    return true;
  }
  return false;
}

bool OSDMonitor::preprocess_query(MonOpRequestRef op)
{
  op->mark_osdmon_event(__func__);
  //取得 message
  Message *m = op->get_req();
  dout(10) << "preprocess_query " << *m << " from " << m->get_orig_source_inst() << dendl;
  //判断 message类性
  switch (m->get_type()) {
.....
  //osd上报错误的处理函数
  case MSG_OSD_FAILURE:
    return preprocess_failure(op);
.......
  default:
    ceph_abort();
    return true;
  }
}

bool OSDMonitor::preprocess_failure(MonOpRequestRef op)
{
  op->mark_osdmon_event(__func__);
  MOSDFailure *m = static_cast<MOSDFailure*>(op->get_req());
  // who is target_osd
  int badboy = m->get_target_osd();

  // check permissions
  if (check_source(op, m->fsid))
    goto didit;

  // first, verify the reporting host is valid
  if (m->get_orig_source().is_osd()) {
    int from = m->get_orig_source().num();
    if (!osdmap.exists(from) ||
	!osdmap.get_addrs(from).legacy_equals(m->get_orig_source_addrs()) ||
	(osdmap.is_down(from) && m->if_osd_failed())) {
      dout(5) << "preprocess_failure from dead osd." << from
	      << ", ignoring" << dendl;
      send_incremental(op, m->get_epoch()+1);
      goto didit;
    }
  }

  // weird?
  if (osdmap.is_down(badboy)) {
    dout(5) << "preprocess_failure dne(/dup?): osd." << m->get_target_osd()
	    << " " << m->get_target_addrs()
	    << ", from " << m->get_orig_source() << dendl;
    if (m->get_epoch() < osdmap.get_epoch())
      send_incremental(op, m->get_epoch()+1);
    goto didit;
  }
  if (osdmap.get_addrs(badboy) != m->get_target_addrs()) {
    dout(5) << "preprocess_failure wrong osd: report osd." << m->get_target_osd()
	    << " " << m->get_target_addrs()
	    << " != map's " << osdmap.get_addrs(badboy)
	    << ", from " << m->get_orig_source() << dendl;
    if (m->get_epoch() < osdmap.get_epoch())
      send_incremental(op, m->get_epoch()+1);
    goto didit;
  }

  // already reported?
  if (osdmap.is_down(badboy) ||
      osdmap.get_up_from(badboy) > m->get_epoch()) {
    dout(5) << "preprocess_failure dup/old: osd." << m->get_target_osd()
	    << " " << m->get_target_addrs()
	    << ", from " << m->get_orig_source() << dendl;
    if (m->get_epoch() < osdmap.get_epoch())
      send_incremental(op, m->get_epoch()+1);
    goto didit;
  }

  if (!can_mark_down(badboy)) {
    dout(5) << "preprocess_failure ignoring report of osd."
	    << m->get_target_osd() << " " << m->get_target_addrs()
	    << " from " << m->get_orig_source() << dendl;
    goto didit;
  }

  dout(10) << "preprocess_failure new: osd." << m->get_target_osd()
	   << " " << m->get_target_addrs()
	   << ", from " << m->get_orig_source() << dendl;
  return false;

 didit:
  mon->no_reply(op);
  return true;
}

bool OSDMonitor::check_failure(utime_t now, int target_osd, failure_info_t& fi)
{
  // already pending failure?
  if (pending_inc.new_state.count(target_osd) &&
      pending_inc.new_state[target_osd] & CEPH_OSD_UP) {
    dout(10) << " already pending failure" << dendl;
    return true;
  }

  set<string> reporters_by_subtree;
  //故障域
  auto reporter_subtree_level = g_conf().get_val<string>("mon_osd_reporter_subtree_level");
  //
  utime_t orig_grace(g_conf()->osd_heartbeat_grace, 0);
  utime_t max_failed_since = fi.get_failed_since();
  utime_t failed_for = now - max_failed_since;

  utime_t grace = orig_grace;
  double my_grace = 0, peer_grace = 0;
  double decay_k = 0;

  //根据以往 osd的错误报告来 调整下  osd_heartbeat_grace 时间
  if (g_conf()->mon_osd_adjust_heartbeat_grace) {
 // decay_k 为 -0.000192541     ？？？
 //::log(.5)  -0.693147
    double halflife = (double)g_conf()->mon_osd_laggy_halflife;
    decay_k = ::log(.5) / halflife;

    // scale grace period based on historical probability of 'lagginess'
    //根据历史上的 "滞后性 "概率确定宽限期的规模
    // (false positive failures due to slowness).

    const osd_xinfo_t& xi = osdmap.get_xinfo(target_osd);
    //计算 log ？
    double decay = exp((double)failed_for * decay_k);
    dout(1) << " halflife " << halflife << " decay_k " << decay_k
	     << " failed_for " << failed_for << " decay " << decay << dendl;

    my_grace = decay * (double)xi.laggy_interval * xi.laggy_probability;
    grace += my_grace;
  }
  //遍历上报错误
  for (auto p = fi.reporters.begin(); p != fi.reporters.end();) {
    // get the parent bucket whose type matches with "reporter_subtree_level".
    // fall back to OSD if the level doesn't exist.
    if (osdmap.exists(p->first)) {
      //通过osdmap 获取完整的osd位置
      auto reporter_loc = osdmap.crush->get_full_location(p->first);
      //不在 指定的故障域
      if (auto iter = reporter_loc.find(reporter_subtree_level);
          iter == reporter_loc.end()) {
        reporters_by_subtree.insert("osd." + to_string(p->first));
      } else {
        string name = get_phy_name(iter->second);
           reporters_by_subtree.insert(name);
      }
      if (g_conf()->mon_osd_adjust_heartbeat_grace) {
        const osd_xinfo_t& xi = osdmap.get_xinfo(p->first);
        //距离上次标记down的时间
        utime_t elapsed = now - xi.down_stamp;
        double decay = exp((double)elapsed * decay_k);
        //被标记为滞后和恢复的平均时间间隔
        peer_grace += decay * (double)xi.laggy_interval * xi.laggy_probability;
      }
      ++p;
    } else {
      fi.cancel_report(p->first);;
      p = fi.reporters.erase(p);
    }
  }

  if (g_conf()->mon_osd_adjust_heartbeat_grace) {
    peer_grace /= (double)fi.reporters.size();
    grace += peer_grace;
  }
// 以上都是在调整 grace的时间

// 实际 osd报错的时间 和 调整过的grace 时间做比较
// 以及报错的上报数量
  if (failed_for >= grace &&
      reporters_by_subtree.size() >= g_conf().get_val<uint64_t>("mon_osd_min_down_reporters")) {

    // 将目标状态标记为 CEPH_OSD_UP，看注释是说 XORed onto previous state，异或？
    pending_inc.new_state[target_osd] = CEPH_OSD_UP;
    return true;
  }
  return false;
}