int OSD::handle_pg_peering_evt(
  spg_t pgid,
  const pg_history_t& orig_history,
  const PastIntervals& pi,
  epoch_t epoch,
  PG::CephPeeringEvtRef evt)
{
  if (service.splitting(pgid)) {
    peering_wait_for_split[pgid].push_back(evt);
    return -EEXIST;
  }

  PG *pg = _lookup_lock_pg(pgid);
  if (!pg) {
    // same primary?
    if (!osdmap->have_pg_pool(pgid.pool()))
      return -EINVAL;
    int up_primary, acting_primary;
    vector<int> up, acting;
    osdmap->pg_to_up_acting_osds(
      pgid.pgid, &up, &up_primary, &acting, &acting_primary);

    pg_history_t history = orig_history;
    bool valid_history = project_pg_history(
      pgid, history, epoch, up, up_primary, acting, acting_primary);

    if (!valid_history || epoch < history.same_interval_since) {
      dout(10) << __func__ << pgid << " acting changed in "
	       << history.same_interval_since << " (msg from " << epoch << ")"
	       << dendl;
      return -EINVAL;
    }

    if (service.splitting(pgid)) {
      ceph_abort();
    }

    const bool is_mon_create =
      evt->get_event().dynamic_type() == PG::NullEvt::static_type();
    if (maybe_wait_for_max_pg(pgid, is_mon_create)) {
      return -EAGAIN;
    }
    // do we need to resurrect a deleting pg?
    spg_t resurrected;
    PGRef old_pg_state;
    res_result result = _try_resurrect_pg(
      service.get_osdmap(),
      pgid,
      &resurrected,
      &old_pg_state);

    PG::RecoveryCtx rctx = create_context();
    switch (result) {
    case RES_NONE: {
      const pg_pool_t* pp = osdmap->get_pg_pool(pgid.pool());
      if (pp->has_flag(pg_pool_t::FLAG_EC_OVERWRITES) &&
	  store->get_type() != "bluestore") {
	clog->warn() << "pg " << pgid
		     << " is at risk of silent data corruption: "
		     << "the pool allows ec overwrites but is not stored in "
		     << "bluestore, so deep scrubbing will not detect bitrot";
      }
      PG::_create(*rctx.transaction, pgid, pgid.get_split_bits(pp->get_pg_num()));
      PG::_init(*rctx.transaction, pgid, pp);

      int role = osdmap->calc_pg_role(whoami, acting, acting.size());
      if (!pp->is_replicated() && role != pgid.shard)
	role = -1;

      pg = _create_lock_pg(
	get_map(epoch),
	pgid, false, false,
	role,
	up, up_primary,
	acting, acting_primary,
	history, pi,
	*rctx.transaction);
      pg->handle_create(&rctx);
      pg->write_if_dirty(*rctx.transaction);
      dispatch_context(rctx, pg, osdmap);

      dout(10) << *pg << " is new" << dendl;

      pg->queue_peering_event(evt);
      wake_pg_waiters(pg);
      pg->unlock();
      return 0;
    }
    case RES_SELF: {
      old_pg_state->lock();
      OSDMapRef old_osd_map = old_pg_state->get_osdmap();
      int old_role = old_pg_state->role;
      vector<int> old_up = old_pg_state->up;
      int old_up_primary = old_pg_state->up_primary.osd;
      vector<int> old_acting = old_pg_state->acting;
      int old_primary = old_pg_state->primary.osd;
      pg_history_t old_history = old_pg_state->info.history;
      PastIntervals old_past_intervals = old_pg_state->past_intervals;
      old_pg_state->unlock();
      pg = _create_lock_pg(
	old_osd_map,
	resurrected,
	false,
	true,
	old_role,
	old_up,
	old_up_primary,
	old_acting,
	old_primary,
	old_history,
	old_past_intervals,
	*rctx.transaction);
      pg->handle_create(&rctx);
      pg->write_if_dirty(*rctx.transaction);
      dispatch_context(rctx, pg, osdmap);

      dout(10) << *pg << " is new (resurrected)" << dendl;

      pg->queue_peering_event(evt);
      wake_pg_waiters(pg);
      pg->unlock();
      return 0;
    }
    case RES_PARENT: {
      assert(old_pg_state);
      old_pg_state->lock();
      OSDMapRef old_osd_map = old_pg_state->get_osdmap();
      int old_role = old_pg_state->role;
      vector<int> old_up = old_pg_state->up;
      int old_up_primary = old_pg_state->up_primary.osd;
      vector<int> old_acting = old_pg_state->acting;
      int old_primary = old_pg_state->primary.osd;
      pg_history_t old_history = old_pg_state->info.history;
      PastIntervals old_past_intervals = old_pg_state->past_intervals;
      old_pg_state->unlock();
      PG *parent = _create_lock_pg(
	old_osd_map,
	resurrected,
	false,
	true,
	old_role,
	old_up,
	old_up_primary,
	old_acting,
	old_primary,
	old_history,
	old_past_intervals,
	*rctx.transaction
	);
      parent->handle_create(&rctx);
      parent->write_if_dirty(*rctx.transaction);
      dispatch_context(rctx, parent, osdmap);

      dout(10) << *parent << " is new" << dendl;

      assert(service.splitting(pgid));
      peering_wait_for_split[pgid].push_back(evt);

      //parent->queue_peering_event(evt);
      parent->queue_null(osdmap->get_epoch(), osdmap->get_epoch());
      wake_pg_waiters(parent);
      parent->unlock();
      return 0;
    }
    default:
      assert(0);
      return 0;
    }
  } else {
    // already had it.  did the mapping change?
    if (epoch < pg->info.history.same_interval_since) {
      dout(10) << *pg << __func__ << " acting changed in "
	       << pg->info.history.same_interval_since
	       << " (msg from " << epoch << ")" << dendl;
    } else {
      pg->queue_peering_event(evt);
    }
    pg->unlock();
    return -EEXIST;
  }
}