void OSD::consume_map()
{
  assert(osd_lock.is_locked());
  dout(7) << "consume_map version " << osdmap->get_epoch() << dendl;

  /** make sure the cluster is speaking in SORTBITWISE, because we don't
   *  speak the older sorting version any more. Be careful not to force
   *  a shutdown if we are merely processing old maps, though.
   */
  if (!osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE) && is_active()) {
    derr << __func__ << " SORTBITWISE flag is not set" << dendl;
    ceph_abort();
  }

  int num_pg_primary = 0, num_pg_replica = 0, num_pg_stray = 0;
  list<PGRef> to_remove;

  // scan pg's
  {
    RWLock::RLocker l(pg_map_lock);
    for (ceph::unordered_map<spg_t,PG*>::iterator it = pg_map.begin();
        it != pg_map.end();
        ++it) {
      PG *pg = it->second;
      pg->lock();
      if (pg->is_primary())
        num_pg_primary++;
      else if (pg->is_replica())
        num_pg_replica++;
      else
        num_pg_stray++;

      if (!osdmap->have_pg_pool(pg->info.pgid.pool())) {
        //pool is deleted!
        to_remove.push_back(PGRef(pg));
      } else {
        service.init_splits_between(it->first, service.get_osdmap(), osdmap);
      }

      pg->unlock();
    }

    lock_guard<mutex> pending_creates_locker{pending_creates_lock};
    for (auto pg = pending_creates_from_osd.cbegin();
	 pg != pending_creates_from_osd.cend();) {
      if (osdmap->get_pg_acting_rank(pg->first, whoami) < 0) {
	pg = pending_creates_from_osd.erase(pg);
      } else {
	++pg;
      }
    }
  }

  for (list<PGRef>::iterator i = to_remove.begin();
       i != to_remove.end();
       to_remove.erase(i++)) {
    RWLock::WLocker locker(pg_map_lock);
    (*i)->lock();
    _remove_pg(&**i);
    (*i)->unlock();
  }

  service.expand_pg_num(service.get_osdmap(), osdmap);

  service.pre_publish_map(osdmap);
  service.await_reserved_maps();
  service.publish_map(osdmap);

  service.maybe_inject_dispatch_delay();

  dispatch_sessions_waiting_on_map();

  service.maybe_inject_dispatch_delay();

  // remove any PGs which we no longer host from the session waiting_for_pg lists
  dout(20) << __func__ << " checking waiting_for_pg" << dendl;
  op_shardedwq.prune_pg_waiters(osdmap, whoami);

  service.maybe_inject_dispatch_delay();

  // scan pg's
  {
    RWLock::RLocker l(pg_map_lock);
    for (ceph::unordered_map<spg_t,PG*>::iterator it = pg_map.begin();
        it != pg_map.end();
        ++it) {
      PG *pg = it->second;
      pg->lock();
      pg->queue_null(osdmap->get_epoch(), osdmap->get_epoch());
      pg->unlock();
    }

    logger->set(l_osd_pg, pg_map.size());
  }
  logger->set(l_osd_pg_primary, num_pg_primary);
  logger->set(l_osd_pg_replica, num_pg_replica);
  logger->set(l_osd_pg_stray, num_pg_stray);
  logger->set(l_osd_pg_removing, remove_wq.get_remove_queue_len());
}
void OSD::consume_map()
{
  assert(osd_lock.is_locked());
  dout(7) << "consume_map version " << osdmap->get_epoch() << dendl;

   
  if (!osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE) && is_active()) {
    derr << __func__ << " SORTBITWISE flag is not set" << dendl;
    ceph_abort();
  }

  int num_pg_primary = 0, num_pg_replica = 0, num_pg_stray = 0;
  list<PGRef> to_remove;

   
  {
    RWLock::RLocker l(pg_map_lock);
    for (ceph::unordered_map<spg_t,PG*>::iterator it = pg_map.begin();
        it != pg_map.end();
        ++it) {
      PG *pg = it->second;
      pg->lock();
      if (pg->is_primary())
        num_pg_primary++;
      else if (pg->is_replica())
        num_pg_replica++;
      else
        num_pg_stray++;

      if (!osdmap->have_pg_pool(pg->info.pgid.pool())) {
         
        to_remove.push_back(PGRef(pg));
      } else {
        service.init_splits_between(it->first, service.get_osdmap(), osdmap);
      }

      pg->unlock();
    }

    lock_guard<mutex> pending_creates_locker{pending_creates_lock};
    for (auto pg = pending_creates_from_osd.cbegin();
	 pg != pending_creates_from_osd.cend();) {
      if (osdmap->get_pg_acting_rank(pg->first, whoami) < 0) {
	pg = pending_creates_from_osd.erase(pg);
      } else {
	++pg;
      }
    }
  }

  for (list<PGRef>::iterator i = to_remove.begin();
       i != to_remove.end();
       to_remove.erase(i++)) {
    RWLock::WLocker locker(pg_map_lock);
    (*i)->lock();
    _remove_pg(&**i);
    (*i)->unlock();
  }

  service.expand_pg_num(service.get_osdmap(), osdmap);

  service.pre_publish_map(osdmap);
  service.await_reserved_maps();
  service.publish_map(osdmap);

  service.maybe_inject_dispatch_delay();

  dispatch_sessions_waiting_on_map();

  service.maybe_inject_dispatch_delay();

   
  dout(20) << __func__ << " checking waiting_for_pg" << dendl;
  op_shardedwq.prune_pg_waiters(osdmap, whoami);

  service.maybe_inject_dispatch_delay();

   
  {
    RWLock::RLocker l(pg_map_lock);
    for (ceph::unordered_map<spg_t,PG*>::iterator it = pg_map.begin();
        it != pg_map.end();
        ++it) {
      PG *pg = it->second;
      pg->lock();
      pg->queue_null(osdmap->get_epoch(), osdmap->get_epoch());
      pg->unlock();
    }

    logger->set(l_osd_pg, pg_map.size());
  }
  logger->set(l_osd_pg_primary, num_pg_primary);
  logger->set(l_osd_pg_replica, num_pg_replica);
  logger->set(l_osd_pg_stray, num_pg_stray);
  logger->set(l_osd_pg_removing, remove_wq.get_remove_queue_len());
}