void OSD::_committed_osd_maps(epoch_t first, epoch_t last, MOSDMap *m)
{
  dout(10) << __func__ << " " << first << ".." << last << dendl;
  if (is_stopping()) {
    dout(10) << __func__ << " bailing, we are shutting down" << dendl;
    return;
  }
  Mutex::Locker l(osd_lock);
  if (is_stopping()) {
    dout(10) << __func__ << " bailing, we are shutting down" << dendl;
    return;
  }
  map_lock.get_write();

  bool do_shutdown = false;
  bool do_restart = false;
  bool network_error = false;

  // advance through the new maps
  for (epoch_t cur = first; cur <= last; cur++) {
    dout(10) << " advance to epoch " << cur
	     << " (<= last " << last
	     << " <= newest_map " << superblock.newest_map
	     << ")" << dendl;

    OSDMapRef newmap = get_map(cur);
    assert(newmap);  // we just cached it above!

    // start blacklisting messages sent to peers that go down.
    service.pre_publish_map(newmap);

    // kill connections to newly down osds
    bool waited_for_reservations = false;
    set<int> old;
    osdmap->get_all_osds(old);
    for (set<int>::iterator p = old.begin(); p != old.end(); ++p) {
      if (*p != whoami &&
	  osdmap->is_up(*p) && // in old map
	  newmap->is_down(*p)) {    // but not the new one
        if (!waited_for_reservations) {
          service.await_reserved_maps();
          waited_for_reservations = true;
        }
	note_down_osd(*p);
      } else if (*p != whoami &&
                osdmap->is_down(*p) &&
                newmap->is_up(*p)) {
        note_up_osd(*p);
      }
    }

    if ((osdmap->test_flag(CEPH_OSDMAP_NOUP) !=
        newmap->test_flag(CEPH_OSDMAP_NOUP)) ||
        (osdmap->is_noup(whoami) != newmap->is_noup(whoami))) {
      dout(10) << __func__ << " NOUP flag changed in " << newmap->get_epoch()
	       << dendl;
      if (is_booting()) {
	// this captures the case where we sent the boot message while
	// NOUP was being set on the mon and our boot request was
	// dropped, and then later it is cleared.  it imperfectly
	// handles the case where our original boot message was not
	// dropped and we restart even though we might have booted, but
	// that is harmless (boot will just take slightly longer).
	do_restart = true;
      }
    }
    if (osdmap->require_osd_release < CEPH_RELEASE_LUMINOUS &&
	newmap->require_osd_release >= CEPH_RELEASE_LUMINOUS) {
      dout(10) << __func__ << " require_osd_release reached luminous in "
	       << newmap->get_epoch() << dendl;
      clear_pg_stat_queue();
      clear_outstanding_pg_stats();
    }

    osdmap = newmap;
    epoch_t up_epoch;
    epoch_t boot_epoch;
    service.retrieve_epochs(&boot_epoch, &up_epoch, NULL);
    if (!up_epoch &&
	osdmap->is_up(whoami) &&
	osdmap->get_inst(whoami) == client_messenger->get_myinst()) {
      up_epoch = osdmap->get_epoch();
      dout(10) << "up_epoch is " << up_epoch << dendl;
      if (!boot_epoch) {
	boot_epoch = osdmap->get_epoch();
	dout(10) << "boot_epoch is " << boot_epoch << dendl;
      }
      service.set_epochs(&boot_epoch, &up_epoch, NULL);
    }
  }

  had_map_since = ceph_clock_now();

  epoch_t _bind_epoch = service.get_bind_epoch();
  if (osdmap->is_up(whoami) &&
      osdmap->get_addr(whoami) == client_messenger->get_myaddr() &&
      _bind_epoch < osdmap->get_up_from(whoami)) {

    if (is_booting()) {
      dout(1) << "state: booting -> active" << dendl;
      set_state(STATE_ACTIVE);

      // set incarnation so that osd_reqid_t's we generate for our
      // objecter requests are unique across restarts.
      service.objecter->set_client_incarnation(osdmap->get_epoch());
    }
  }

  if (osdmap->get_epoch() > 0 &&
      is_active()) {
    if (!osdmap->exists(whoami)) {
      dout(0) << "map says i do not exist.  shutting down." << dendl;
      do_shutdown = true;   // don't call shutdown() while we have
			    // everything paused
    } else if (!osdmap->is_up(whoami) ||
	       !osdmap->get_addr(whoami).probably_equals(
		 client_messenger->get_myaddr()) ||
	       !osdmap->get_cluster_addr(whoami).probably_equals(
		 cluster_messenger->get_myaddr()) ||
	       !osdmap->get_hb_back_addr(whoami).probably_equals(
		 hb_back_server_messenger->get_myaddr()) ||
	       (osdmap->get_hb_front_addr(whoami) != entity_addr_t() &&
                !osdmap->get_hb_front_addr(whoami).probably_equals(
		  hb_front_server_messenger->get_myaddr()))) {
      if (!osdmap->is_up(whoami)) {
	if (service.is_preparing_to_stop() || service.is_stopping()) {
	  service.got_stop_ack();
	} else {
          clog->warn() << "Monitor daemon marked osd." << whoami << " down, "
                          "but it is still running";
          clog->debug() << "map e" << osdmap->get_epoch()
                        << " wrongly marked me down at e"
                        << osdmap->get_down_at(whoami);
	}
      } else if (!osdmap->get_addr(whoami).probably_equals(
		   client_messenger->get_myaddr())) {
	clog->error() << "map e" << osdmap->get_epoch()
		      << " had wrong client addr (" << osdmap->get_addr(whoami)
		      << " != my " << client_messenger->get_myaddr() << ")";
      } else if (!osdmap->get_cluster_addr(whoami).probably_equals(
		   cluster_messenger->get_myaddr())) {
	clog->error() << "map e" << osdmap->get_epoch()
		      << " had wrong cluster addr ("
		      << osdmap->get_cluster_addr(whoami)
		      << " != my " << cluster_messenger->get_myaddr() << ")";
      } else if (!osdmap->get_hb_back_addr(whoami).probably_equals(
		   hb_back_server_messenger->get_myaddr())) {
	clog->error() << "map e" << osdmap->get_epoch()
		      << " had wrong heartbeat back addr ("
		      << osdmap->get_hb_back_addr(whoami)
		      << " != my " << hb_back_server_messenger->get_myaddr()
		      << ")";
      } else if (osdmap->get_hb_front_addr(whoami) != entity_addr_t() &&
		 !osdmap->get_hb_front_addr(whoami).probably_equals(
		   hb_front_server_messenger->get_myaddr())) {
	clog->error() << "map e" << osdmap->get_epoch()
		      << " had wrong heartbeat front addr ("
		      << osdmap->get_hb_front_addr(whoami)
		      << " != my " << hb_front_server_messenger->get_myaddr()
		      << ")";
      }

      if (!service.is_stopping()) {
        epoch_t up_epoch = 0;
        epoch_t bind_epoch = osdmap->get_epoch();
        service.set_epochs(NULL,&up_epoch, &bind_epoch);
	do_restart = true;

	//add markdown log
	utime_t now = ceph_clock_now();
	utime_t grace = utime_t(cct->_conf->osd_max_markdown_period, 0);
	osd_markdown_log.push_back(now);
	//clear all out-of-date log
	while (!osd_markdown_log.empty() &&
	       osd_markdown_log.front() + grace < now)
	  osd_markdown_log.pop_front();
	if ((int)osd_markdown_log.size() > cct->_conf->osd_max_markdown_count) {
	  dout(0) << __func__ << " marked down "
		  << osd_markdown_log.size()
		  << " > osd_max_markdown_count "
		  << cct->_conf->osd_max_markdown_count
		  << " in last " << grace << " seconds, shutting down"
		  << dendl;
	  do_restart = false;
	  do_shutdown = true;
	}

	start_waiting_for_healthy();

	set<int> avoid_ports;
#if defined(__FreeBSD__)
        // prevent FreeBSD from grabbing the client_messenger port during
        // rebinding. In which case a cluster_meesneger will connect also 
	// to the same port
	avoid_ports.insert(client_messenger->get_myaddr().get_port());
#endif
	avoid_ports.insert(cluster_messenger->get_myaddr().get_port());
	avoid_ports.insert(hb_back_server_messenger->get_myaddr().get_port());
	avoid_ports.insert(hb_front_server_messenger->get_myaddr().get_port());

	int r = cluster_messenger->rebind(avoid_ports);
	if (r != 0) {
	  do_shutdown = true;  // FIXME: do_restart?
          network_error = true;
          dout(0) << __func__ << " marked down:"
                  << " rebind cluster_messenger failed" << dendl;
        }

	r = hb_back_server_messenger->rebind(avoid_ports);
	if (r != 0) {
	  do_shutdown = true;  // FIXME: do_restart?
          network_error = true;
          dout(0) << __func__ << " marked down:"
                  << " rebind hb_back_server_messenger failed" << dendl;
        }

	r = hb_front_server_messenger->rebind(avoid_ports);
	if (r != 0) {
	  do_shutdown = true;  // FIXME: do_restart?
          network_error = true;
          dout(0) << __func__ << " marked down:" 
                  << " rebind hb_front_server_messenger failed" << dendl;
        }

	hb_front_client_messenger->mark_down_all();
	hb_back_client_messenger->mark_down_all();

	reset_heartbeat_peers();
      }
    }
  }

  map_lock.put_write();

  check_osdmap_features(store);

  // yay!
  consume_map();

  if (is_active() || is_waiting_for_healthy())
    maybe_update_heartbeat_peers();

  if (!is_active()) {
    dout(10) << " not yet active; waiting for peering wq to drain" << dendl;
    peering_wq.drain();
  } else {
    activate_map();
  }

  if (do_shutdown) {
    if (network_error) {
      Mutex::Locker l(heartbeat_lock);
      map<int,pair<utime_t,entity_inst_t>>::iterator it =
	failure_pending.begin();
      while (it != failure_pending.end()) {
        dout(10) << "handle_osd_ping canceling in-flight failure report for osd."
		 << it->first << dendl;
        send_still_alive(osdmap->get_epoch(), it->second.second);
        failure_pending.erase(it++);
      }
    }
    // trigger shutdown in a different thread
    dout(0) << __func__ << " shutdown OSD via async signal" << dendl;
    queue_async_signal(SIGINT);
  }
  else if (m->newest_map && m->newest_map > last) {
    dout(10) << " msg say newest map is " << m->newest_map
	     << ", requesting more" << dendl;
    osdmap_subscribe(osdmap->get_epoch()+1, false);
  }
  else if (is_preboot()) {
    if (m->get_source().is_mon())
      _preboot(m->oldest_map, m->newest_map);
    else
      start_boot();
  }
  else if (do_restart)
    start_boot();

}