void OSD::handle_pg_query(OpRequestRef op)
{
  assert(osd_lock.is_locked());

  const MOSDPGQuery *m = static_cast<const MOSDPGQuery*>(op->get_req());
  assert(m->get_type() == MSG_OSD_PG_QUERY);

  if (!require_osd_peer(op->get_req()))
    return;

  dout(7) << "handle_pg_query from " << m->get_source() << " epoch " << m->get_epoch() << dendl;
  int from = m->get_source().num();

  if (!require_same_or_newer_map(op, m->get_epoch(), false))
    return;

  op->mark_started();

  map< int, vector<pair<pg_notify_t, PastIntervals> > > notify_list;

  for (auto it = m->pg_list.begin();
       it != m->pg_list.end();
       ++it) {
    spg_t pgid = it->first;

    if (pgid.preferred() >= 0) {
      dout(10) << "ignoring localized pg " << pgid << dendl;
      continue;
    }

    if (service.splitting(pgid)) {
      peering_wait_for_split[pgid].push_back(
	PG::CephPeeringEvtRef(
	  new PG::CephPeeringEvt(
	    it->second.epoch_sent, it->second.epoch_sent,
	    PG::MQuery(pg_shard_t(from, it->second.from),
		       it->second, it->second.epoch_sent))));
      continue;
    }

    {
      RWLock::RLocker l(pg_map_lock);
      if (pg_map.count(pgid)) {
        PG *pg = 0;
        pg = _lookup_lock_pg_with_map_lock_held(pgid);
        pg->queue_query(
            it->second.epoch_sent, it->second.epoch_sent,
            pg_shard_t(from, it->second.from), it->second);
        pg->unlock();
        continue;
      }
    }

    if (!osdmap->have_pg_pool(pgid.pool()))
      continue;

    // get active crush mapping
    int up_primary, acting_primary;
    vector<int> up, acting;
    osdmap->pg_to_up_acting_osds(
      pgid.pgid, &up, &up_primary, &acting, &acting_primary);

    // same primary?
    pg_history_t history = it->second.history;
    bool valid_history = project_pg_history(
      pgid, history, it->second.epoch_sent,
      up, up_primary, acting, acting_primary);

    if (!valid_history ||
        it->second.epoch_sent < history.same_interval_since) {
      dout(10) << " pg " << pgid << " dne, and pg has changed in "
	       << history.same_interval_since
	       << " (msg from " << it->second.epoch_sent << ")" << dendl;
      continue;
    }

    dout(10) << " pg " << pgid << " dne" << dendl;
    pg_info_t empty(spg_t(pgid.pgid, it->second.to));
    /* This is racy, but that should be ok: if we complete the deletion
     * before the pg is recreated, we'll just start it off backfilling
     * instead of just empty */
    if (service.deleting_pgs.lookup(pgid))
      empty.set_last_backfill(hobject_t());
    if (it->second.type == pg_query_t::LOG ||
	it->second.type == pg_query_t::FULLLOG) {
      ConnectionRef con = service.get_con_osd_cluster(from, osdmap->get_epoch());
      if (con) {
	MOSDPGLog *mlog = new MOSDPGLog(
	  it->second.from, it->second.to,
	  osdmap->get_epoch(), empty,
	  it->second.epoch_sent);
	service.share_map_peer(from, con.get(), osdmap);
	con->send_message(mlog);
      }
    } else {
      notify_list[from].push_back(
	make_pair(
	  pg_notify_t(
	    it->second.from, it->second.to,
	    it->second.epoch_sent,
	    osdmap->get_epoch(),
	    empty),
	  PastIntervals(
	    osdmap->get_pools().at(pgid.pool()).ec_pool(),
	    *osdmap)));
    }
  }
  do_notifies(notify_list, osdmap);
}