void OSD::handle_osd_ping(MOSDPing *m)
{
  if (superblock.cluster_fsid != m->fsid) {
    dout(20) << "handle_osd_ping from " << m->get_source_inst()
	     << " bad fsid " << m->fsid << " != " << superblock.cluster_fsid << dendl;
    m->put();
    return;
  }

  int from = m->get_source().num();

  heartbeat_lock.Lock();
  if (is_stopping()) {
    heartbeat_lock.Unlock();
    m->put();
    return;
  }

  OSDMapRef curmap = service.get_osdmap();
  if (!curmap) {
    heartbeat_lock.Unlock();
    m->put();
    return;
  }

  switch (m->op) {

  case MOSDPing::PING:
    {
      if (cct->_conf->osd_debug_drop_ping_probability > 0) {
	auto heartbeat_drop = debug_heartbeat_drops_remaining.find(from);
	if (heartbeat_drop != debug_heartbeat_drops_remaining.end()) {
	  if (heartbeat_drop->second == 0) {
	    debug_heartbeat_drops_remaining.erase(heartbeat_drop);
	  } else {
	    --heartbeat_drop->second;
	    dout(5) << "Dropping heartbeat from " << from
		    << ", " << heartbeat_drop->second
		    << " remaining to drop" << dendl;
	    break;
	  }
	} else if (cct->_conf->osd_debug_drop_ping_probability >
	           ((((double)(rand()%100))/100.0))) {
	  heartbeat_drop =
	    debug_heartbeat_drops_remaining.insert(std::make_pair(from,
	                     cct->_conf->osd_debug_drop_ping_duration)).first;
	  dout(5) << "Dropping heartbeat from " << from
		  << ", " << heartbeat_drop->second
		  << " remaining to drop" << dendl;
	  break;
	}
      }

      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);

      if (curmap->is_up(from)) {
	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());
	  }
	}
      } 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);
      }
    }
    break;

  case MOSDPing::PING_REPLY:
    {
      map<int,HeartbeatInfo>::iterator i = heartbeat_peers.find(from);
      if (i != heartbeat_peers.end()) {
	if (m->get_connection() == i->second.con_back) {
	  dout(25) << "handle_osd_ping got reply from osd." << from
		   << " first_tx " << i->second.first_tx
		   << " last_tx " << i->second.last_tx
		   << " last_rx_back " << i->second.last_rx_back << " -> " << m->stamp
		   << " last_rx_front " << i->second.last_rx_front
		   << dendl;
	  i->second.last_rx_back = m->stamp;
	  // if there is no front con, set both stamps.
	  if (i->second.con_front == NULL)
	    i->second.last_rx_front = m->stamp;
	} else if (m->get_connection() == i->second.con_front) {
	  dout(25) << "handle_osd_ping got reply from osd." << from
		   << " first_tx " << i->second.first_tx
		   << " last_tx " << i->second.last_tx
		   << " last_rx_back " << i->second.last_rx_back
		   << " last_rx_front " << i->second.last_rx_front << " -> " << m->stamp
		   << dendl;
	  i->second.last_rx_front = m->stamp;
	}

        utime_t cutoff = ceph_clock_now();
        cutoff -= cct->_conf->osd_heartbeat_grace;
        if (i->second.is_healthy(cutoff)) {
          // Cancel false reports
	  auto failure_queue_entry = failure_queue.find(from);
	  if (failure_queue_entry != failure_queue.end()) {
            dout(10) << "handle_osd_ping canceling queued "
                     << "failure report for osd." << from << dendl;
            failure_queue.erase(failure_queue_entry);
          }

	  auto failure_pending_entry = failure_pending.find(from);
	  if (failure_pending_entry != failure_pending.end()) {
            dout(10) << "handle_osd_ping canceling in-flight "
                     << "failure report for osd." << from << dendl;
            send_still_alive(curmap->get_epoch(),
			     failure_pending_entry->second.second);
            failure_pending.erase(failure_pending_entry);
          }
        }
      }

      if (m->map_epoch &&
	  curmap->is_up(from)) {
	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());
	  }
	}
      }
    }
    break;

  case MOSDPing::YOU_DIED:
    dout(10) << "handle_osd_ping " << m->get_source_inst()
	     << " says i am down in " << m->map_epoch << dendl;
    osdmap_subscribe(curmap->get_epoch()+1, false);
    break;
  }

  heartbeat_lock.Unlock();
  m->put();
}