import quincy beta 17.1.0

[ceph.git] / ceph / src / osd / PeeringState.cc

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#include "PGPeeringEvent.h"
#include "common/ceph_releases.h"
#include "common/dout.h"
#include "PeeringState.h"

#include "messages/MOSDPGRemove.h"
#include "messages/MBackfillReserve.h"
#include "messages/MRecoveryReserve.h"
#include "messages/MOSDScrubReserve.h"
#include "messages/MOSDPGInfo2.h"
#include "messages/MOSDPGTrim.h"
#include "messages/MOSDPGLog.h"
#include "messages/MOSDPGNotify2.h"
#include "messages/MOSDPGQuery2.h"
#include "messages/MOSDPGLease.h"
#include "messages/MOSDPGLeaseAck.h"

#define dout_context cct
#define dout_subsys ceph_subsys_osd

using std::dec;
using std::hex;
using std::make_pair;
using std::map;
using std::ostream;
using std::pair;
using std::set;
using std::string;
using std::stringstream;
using std::vector;

using ceph::Formatter;
using ceph::make_message;

BufferedRecoveryMessages::BufferedRecoveryMessages(PeeringCtx &ctx)
  // steal messages from ctx
  : message_map{std::move(ctx.message_map)}
{}

void BufferedRecoveryMessages::send_notify(int to, const pg_notify_t &n)
{
  spg_t pgid(n.info.pgid.pgid, n.to);
  send_osd_message(to, TOPNSPC::make_message<MOSDPGNotify2>(pgid, n));
}

void BufferedRecoveryMessages::send_query(
  int to,
  spg_t to_spgid,
  const pg_query_t &q)
{
  send_osd_message(to, TOPNSPC::make_message<MOSDPGQuery2>(to_spgid, q));
}

void BufferedRecoveryMessages::send_info(
  int to,
  spg_t to_spgid,
  epoch_t min_epoch,
  epoch_t cur_epoch,
  const pg_info_t &info,
  std::optional<pg_lease_t> lease,
  std::optional<pg_lease_ack_t> lease_ack)
{
  send_osd_message(
    to,
    TOPNSPC::make_message<MOSDPGInfo2>(
      to_spgid,
      info,
      cur_epoch,
      min_epoch,
      lease,
      lease_ack)
  );
}

void PGPool::update(OSDMapRef map)
{
  const pg_pool_t *pi = map->get_pg_pool(id);
  if (!pi) {
    return; // pool has been deleted
  }
  info = *pi;
  name = map->get_pool_name(id);

  bool updated = false;
  if ((map->get_epoch() != cached_epoch + 1) ||
      (pi->get_snap_epoch() == map->get_epoch())) {
    updated = true;
  }

  if (info.is_pool_snaps_mode() && updated) {
    snapc = pi->get_snap_context();
  }
  cached_epoch = map->get_epoch();
}

/*-------------Peering State Helpers----------------*/
#undef dout_prefix
#define dout_prefix (dpp->gen_prefix(*_dout))
#undef psdout
#define psdout(x) ldout(cct, x)

PeeringState::PeeringState(
  CephContext *cct,
  pg_shard_t pg_whoami,
  spg_t spgid,
  const PGPool &_pool,
  OSDMapRef curmap,
  DoutPrefixProvider *dpp,
  PeeringListener *pl)
  : state_history(*pl),
    cct(cct),
    spgid(spgid),
    dpp(dpp),
    pl(pl),
    orig_ctx(0),
    osdmap_ref(curmap),
    pool(_pool),
    pg_whoami(pg_whoami),
    info(spgid),
    pg_log(cct),
    last_require_osd_release(curmap->require_osd_release),
    missing_loc(spgid, this, dpp, cct),
    machine(this, cct, spgid, dpp, pl, &state_history)
{
  machine.initiate();
}

void PeeringState::start_handle(PeeringCtx *new_ctx) {
  ceph_assert(!rctx);
  ceph_assert(!orig_ctx);
  orig_ctx = new_ctx;
  if (new_ctx) {
    if (messages_pending_flush) {
      rctx.emplace(*messages_pending_flush, *new_ctx);
    } else {
      rctx.emplace(*new_ctx);
    }
    rctx->start_time = ceph_clock_now();
  }
}

void PeeringState::begin_block_outgoing() {
  ceph_assert(!messages_pending_flush);
  ceph_assert(orig_ctx);
  ceph_assert(rctx);
  messages_pending_flush.emplace();
  rctx.emplace(*messages_pending_flush, *orig_ctx);
}

void PeeringState::clear_blocked_outgoing() {
  ceph_assert(orig_ctx);
  ceph_assert(rctx);
  messages_pending_flush = std::optional<BufferedRecoveryMessages>();
}

void PeeringState::end_block_outgoing() {
  ceph_assert(messages_pending_flush);
  ceph_assert(orig_ctx);
  ceph_assert(rctx);

  orig_ctx->accept_buffered_messages(*messages_pending_flush);
  rctx.emplace(*orig_ctx);
  messages_pending_flush = std::optional<BufferedRecoveryMessages>();
}

void PeeringState::end_handle() {
  if (rctx) {
    utime_t dur = ceph_clock_now() - rctx->start_time;
    machine.event_time += dur;
  }

  machine.event_count++;
  rctx = std::nullopt;
  orig_ctx = NULL;
}

void PeeringState::check_recovery_sources(const OSDMapRef& osdmap)
{
  /*
   * check that any peers we are planning to (or currently) pulling
   * objects from are dealt with.
   */
  missing_loc.check_recovery_sources(osdmap);
  pl->check_recovery_sources(osdmap);

  for (auto i = peer_log_requested.begin(); i != peer_log_requested.end();) {
    if (!osdmap->is_up(i->osd)) {
      psdout(10) << "peer_log_requested removing " << *i << dendl;
      peer_log_requested.erase(i++);
    } else {
      ++i;
    }
  }

  for (auto i = peer_missing_requested.begin();
       i != peer_missing_requested.end();) {
    if (!osdmap->is_up(i->osd)) {
      psdout(10) << "peer_missing_requested removing " << *i << dendl;
      peer_missing_requested.erase(i++);
    } else {
      ++i;
    }
  }
}

void PeeringState::update_history(const pg_history_t& new_history)
{
  auto mnow = pl->get_mnow();
  info.history.refresh_prior_readable_until_ub(mnow, prior_readable_until_ub);
  if (info.history.merge(new_history)) {
    psdout(20) << __func__ << " advanced history from " << new_history << dendl;
    dirty_info = true;
    if (info.history.last_epoch_clean >= info.history.same_interval_since) {
      psdout(20) << __func__ << " clearing past_intervals" << dendl;
      past_intervals.clear();
      dirty_big_info = true;
    }
    prior_readable_until_ub = info.history.get_prior_readable_until_ub(mnow);
    if (prior_readable_until_ub != ceph::signedspan::zero()) {
      dout(20) << __func__
               << " prior_readable_until_ub " << prior_readable_until_ub
               << " (mnow " << mnow << " + "
               << info.history.prior_readable_until_ub << ")" << dendl;
    }
  }
  pl->on_info_history_change();
}

hobject_t PeeringState::earliest_backfill() const
{
  hobject_t e = hobject_t::get_max();
  for (const pg_shard_t& bt : get_backfill_targets()) {
    const pg_info_t &pi = get_peer_info(bt);
    e = std::min(pi.last_backfill, e);
  }
  return e;
}

void PeeringState::purge_strays()
{
  if (is_premerge()) {
    psdout(10) << "purge_strays " << stray_set << " but premerge, doing nothing"
               << dendl;
    return;
  }
  if (cct->_conf.get_val<bool>("osd_debug_no_purge_strays")) {
    return;
  }
  psdout(10) << "purge_strays " << stray_set << dendl;

  bool removed = false;
  for (auto p = stray_set.begin(); p != stray_set.end(); ++p) {
    ceph_assert(!is_acting_recovery_backfill(*p));
    if (get_osdmap()->is_up(p->osd)) {
      psdout(10) << "sending PGRemove to osd." << *p << dendl;
      vector<spg_t> to_remove;
      to_remove.push_back(spg_t(info.pgid.pgid, p->shard));
      auto m = TOPNSPC::make_message<MOSDPGRemove>(
        get_osdmap_epoch(),
        to_remove);
      pl->send_cluster_message(p->osd, std::move(m), get_osdmap_epoch());
    } else {
      psdout(10) << "not sending PGRemove to down osd." << *p << dendl;
    }
    peer_missing.erase(*p);
    peer_info.erase(*p);
    missing_loc.remove_stray_recovery_sources(*p);
    peer_purged.insert(*p);
    removed = true;
  }

  // if we removed anyone, update peers (which include peer_info)
  if (removed)
    update_heartbeat_peers();

  stray_set.clear();

  // clear _requested maps; we may have to peer() again if we discover
  // (more) stray content
  peer_log_requested.clear();
  peer_missing_requested.clear();
}

void PeeringState::query_unfound(Formatter *f, string state)
{
  psdout(20) << "Enter PeeringState common QueryUnfound" << dendl;
  {
    f->dump_string("state", state);
    f->dump_bool("available_might_have_unfound", true);
    f->open_array_section("might_have_unfound");
    for (auto p = might_have_unfound.begin();
         p != might_have_unfound.end();
         ++p) {
      if (peer_missing.count(*p)) {
        ; // Ignore already probed OSDs
      } else {
        f->open_object_section("osd");
        f->dump_stream("osd") << *p;
        if (peer_missing_requested.count(*p)) {
          f->dump_string("status", "querying");
        } else if (!get_osdmap()->is_up(p->osd)) {
          f->dump_string("status", "osd is down");
        } else {
          f->dump_string("status", "not queried");
        }
        f->close_section();
      }
    }
    f->close_section();
  }
  psdout(20) << "Exit PeeringState common QueryUnfound" << dendl;
  return;
}

bool PeeringState::proc_replica_info(
  pg_shard_t from, const pg_info_t &oinfo, epoch_t send_epoch)
{
  auto p = peer_info.find(from);
  if (p != peer_info.end() && p->second.last_update == oinfo.last_update) {
    psdout(10) << " got dup osd." << from << " info "
               << oinfo << ", identical to ours" << dendl;
    return false;
  }

  if (!get_osdmap()->has_been_up_since(from.osd, send_epoch)) {
    psdout(10) << " got info " << oinfo << " from down osd." << from
             << " discarding" << dendl;
    return false;
  }

  psdout(10) << " got osd." << from << " " << oinfo << dendl;
  ceph_assert(is_primary());
  peer_info[from] = oinfo;
  might_have_unfound.insert(from);

  update_history(oinfo.history);

  // stray?
  if (!is_up(from) && !is_acting(from)) {
    psdout(10) << " osd." << from << " has stray content: " << oinfo << dendl;
    stray_set.insert(from);
    if (is_clean()) {
      purge_strays();
    }
  }

  // was this a new info?  if so, update peers!
  if (p == peer_info.end())
    update_heartbeat_peers();

  return true;
}


void PeeringState::remove_down_peer_info(const OSDMapRef &osdmap)
{
  // Remove any downed osds from peer_info
  bool removed = false;
  auto p = peer_info.begin();
  while (p != peer_info.end()) {
    if (!osdmap->is_up(p->first.osd)) {
      psdout(10) << " dropping down osd." << p->first << " info " << p->second << dendl;
      peer_missing.erase(p->first);
      peer_log_requested.erase(p->first);
      peer_missing_requested.erase(p->first);
      peer_info.erase(p++);
      removed = true;
    } else
      ++p;
  }

  // Remove any downed osds from peer_purged so we can re-purge if necessary
  auto it = peer_purged.begin();
  while (it != peer_purged.end()) {
    if (!osdmap->is_up(it->osd)) {
      psdout(10) << " dropping down osd." << *it << " from peer_purged" << dendl;
      peer_purged.erase(it++);
    } else {
      ++it;
    }
  }

  // if we removed anyone, update peers (which include peer_info)
  if (removed)
    update_heartbeat_peers();

  check_recovery_sources(osdmap);
}

void PeeringState::update_heartbeat_peers()
{
  if (!is_primary())
    return;

  set<int> new_peers;
  for (unsigned i=0; i<acting.size(); i++) {
    if (acting[i] != CRUSH_ITEM_NONE)
      new_peers.insert(acting[i]);
  }
  for (unsigned i=0; i<up.size(); i++) {
    if (up[i] != CRUSH_ITEM_NONE)
      new_peers.insert(up[i]);
  }
  for (auto p = peer_info.begin(); p != peer_info.end(); ++p) {
    new_peers.insert(p->first.osd);
  }
  pl->update_heartbeat_peers(std::move(new_peers));
}

void PeeringState::write_if_dirty(ObjectStore::Transaction& t)
{
  pl->prepare_write(
    info,
    last_written_info,
    past_intervals,
    pg_log,
    dirty_info,
    dirty_big_info,
    last_persisted_osdmap < get_osdmap_epoch(),
    t);
  if (dirty_info || dirty_big_info) {
    last_persisted_osdmap = get_osdmap_epoch();
    last_written_info = info;
    dirty_info = false;
    dirty_big_info = false;
  }
}

void PeeringState::advance_map(
  OSDMapRef osdmap, OSDMapRef lastmap,
  vector<int>& newup, int up_primary,
  vector<int>& newacting, int acting_primary,
  PeeringCtx &rctx)
{
  ceph_assert(lastmap == osdmap_ref);
  psdout(10) << "handle_advance_map "
            << newup << "/" << newacting
            << " -- " << up_primary << "/" << acting_primary
            << dendl;

  update_osdmap_ref(osdmap);
  pool.update(osdmap);

  AdvMap evt(
    osdmap, lastmap, newup, up_primary,
    newacting, acting_primary);
  handle_event(evt, &rctx);
  if (pool.info.last_change == osdmap_ref->get_epoch()) {
    pl->on_pool_change();
  }
  readable_interval = pool.get_readable_interval(cct->_conf);
  last_require_osd_release = osdmap->require_osd_release;
}

void PeeringState::activate_map(PeeringCtx &rctx)
{
  psdout(10) << __func__ << dendl;
  ActMap evt;
  handle_event(evt, &rctx);
  if (osdmap_ref->get_epoch() - last_persisted_osdmap >
    cct->_conf->osd_pg_epoch_persisted_max_stale) {
    psdout(20) << __func__ << ": Dirtying info: last_persisted is "
              << last_persisted_osdmap
              << " while current is " << osdmap_ref->get_epoch() << dendl;
    dirty_info = true;
  } else {
    psdout(20) << __func__ << ": Not dirtying info: last_persisted is "
              << last_persisted_osdmap
              << " while current is " << osdmap_ref->get_epoch() << dendl;
  }
  write_if_dirty(rctx.transaction);

  if (get_osdmap()->check_new_blocklist_entries()) {
    pl->check_blocklisted_watchers();
  }
}

void PeeringState::set_last_peering_reset()
{
  psdout(20) << "set_last_peering_reset " << get_osdmap_epoch() << dendl;
  if (last_peering_reset != get_osdmap_epoch()) {
    last_peering_reset = get_osdmap_epoch();
    psdout(10) << "Clearing blocked outgoing recovery messages" << dendl;
    clear_blocked_outgoing();
    if (!pl->try_flush_or_schedule_async()) {
      psdout(10) << "Beginning to block outgoing recovery messages" << dendl;
      begin_block_outgoing();
    } else {
      psdout(10) << "Not blocking outgoing recovery messages" << dendl;
    }
  }
}

void PeeringState::complete_flush()
{
  flushes_in_progress--;
  if (flushes_in_progress == 0) {
    pl->on_flushed();
  }
}

void PeeringState::check_full_transition(OSDMapRef lastmap, OSDMapRef osdmap)
{
  const pg_pool_t *pi = osdmap->get_pg_pool(info.pgid.pool());
  if (!pi) {
    return; // pool deleted
  }
  bool changed = false;
  if (pi->has_flag(pg_pool_t::FLAG_FULL)) {
    const pg_pool_t *opi = lastmap->get_pg_pool(info.pgid.pool());
    if (!opi || !opi->has_flag(pg_pool_t::FLAG_FULL)) {
      psdout(10) << " pool was marked full in " << osdmap->get_epoch() << dendl;
      changed = true;
    }
  }
  if (changed) {
    info.history.last_epoch_marked_full = osdmap->get_epoch();
    dirty_info = true;
  }
}

bool PeeringState::should_restart_peering(
  int newupprimary,
  int newactingprimary,
  const vector<int>& newup,
  const vector<int>& newacting,
  OSDMapRef lastmap,
  OSDMapRef osdmap)
{
  if (PastIntervals::is_new_interval(
        primary.osd,
        newactingprimary,
        acting,
        newacting,
        up_primary.osd,
        newupprimary,
        up,
        newup,
        osdmap.get(),
        lastmap.get(),
        info.pgid.pgid)) {
    psdout(20) << "new interval newup " << newup
               << " newacting " << newacting << dendl;
    return true;
  }
  if (!lastmap->is_up(pg_whoami.osd) && osdmap->is_up(pg_whoami.osd)) {
    psdout(10) << __func__ << " osd transitioned from down -> up"
               << dendl;
    return true;
  }
  return false;
}

/* Called before initializing peering during advance_map */
void PeeringState::start_peering_interval(
  const OSDMapRef lastmap,
  const vector<int>& newup, int new_up_primary,
  const vector<int>& newacting, int new_acting_primary,
  ObjectStore::Transaction &t)
{
  const OSDMapRef osdmap = get_osdmap();

  set_last_peering_reset();

  vector<int> oldacting, oldup;
  int oldrole = get_role();

  if (is_primary()) {
    pl->clear_ready_to_merge();
  }


  pg_shard_t old_acting_primary = get_primary();
  pg_shard_t old_up_primary = up_primary;
  bool was_old_primary = is_primary();
  bool was_old_nonprimary = is_nonprimary();

  acting.swap(oldacting);
  up.swap(oldup);
  init_primary_up_acting(
    newup,
    newacting,
    new_up_primary,
    new_acting_primary);

  if (info.stats.up != up ||
      info.stats.acting != acting ||
      info.stats.up_primary != new_up_primary ||
      info.stats.acting_primary != new_acting_primary) {
    info.stats.up = up;
    info.stats.up_primary = new_up_primary;
    info.stats.acting = acting;
    info.stats.acting_primary = new_acting_primary;
    info.stats.mapping_epoch = osdmap->get_epoch();
  }

  pl->clear_publish_stats();

  // This will now be remapped during a backfill in cases
  // that it would not have been before.
  if (up != acting)
    state_set(PG_STATE_REMAPPED);
  else
    state_clear(PG_STATE_REMAPPED);

  int role = osdmap->calc_pg_role(pg_whoami, acting);
  set_role(role);

  // did acting, up, primary|acker change?
  if (!lastmap) {
    psdout(10) << " no lastmap" << dendl;
    dirty_info = true;
    dirty_big_info = true;
    info.history.same_interval_since = osdmap->get_epoch();
  } else {
    std::stringstream debug;
    ceph_assert(info.history.same_interval_since != 0);
    bool new_interval = PastIntervals::check_new_interval(
      old_acting_primary.osd,
      new_acting_primary,
      oldacting, newacting,
      old_up_primary.osd,
      new_up_primary,
      oldup, newup,
      info.history.same_interval_since,
      info.history.last_epoch_clean,
      osdmap.get(),
      lastmap.get(),
      info.pgid.pgid,
      missing_loc.get_recoverable_predicate(),
      &past_intervals,
      &debug);
    psdout(10) << __func__ << ": check_new_interval output: "
               << debug.str() << dendl;
    if (new_interval) {
      if (osdmap->get_epoch() == pl->oldest_stored_osdmap() &&
          info.history.last_epoch_clean < osdmap->get_epoch()) {
        psdout(10) << " map gap, clearing past_intervals and faking" << dendl;
        // our information is incomplete and useless; someone else was clean
        // after everything we know if osdmaps were trimmed.
        past_intervals.clear();
      } else {
        psdout(10) << " noting past " << past_intervals << dendl;
      }
      dirty_info = true;
      dirty_big_info = true;
      info.history.same_interval_since = osdmap->get_epoch();
      if (osdmap->have_pg_pool(info.pgid.pgid.pool()) &&
          info.pgid.pgid.is_split(lastmap->get_pg_num(info.pgid.pgid.pool()),
                                  osdmap->get_pg_num(info.pgid.pgid.pool()),
                                  nullptr)) {
        info.history.last_epoch_split = osdmap->get_epoch();
      }
    }
  }

  if (old_up_primary != up_primary ||
      oldup != up) {
    info.history.same_up_since = osdmap->get_epoch();
  }
  // this comparison includes primary rank via pg_shard_t
  if (old_acting_primary != get_primary()) {
    info.history.same_primary_since = osdmap->get_epoch();
  }

  on_new_interval();
  pl->on_info_history_change();

  psdout(1) << __func__ << " up " << oldup << " -> " << up
            << ", acting " << oldacting << " -> " << acting
            << ", acting_primary " << old_acting_primary << " -> "
            << new_acting_primary
            << ", up_primary " << old_up_primary << " -> " << new_up_primary
            << ", role " << oldrole << " -> " << role
            << ", features acting " << acting_features
            << " upacting " << upacting_features
            << dendl;

  // deactivate.
  state_clear(PG_STATE_ACTIVE);
  state_clear(PG_STATE_PEERED);
  state_clear(PG_STATE_PREMERGE);
  state_clear(PG_STATE_DOWN);
  state_clear(PG_STATE_RECOVERY_WAIT);
  state_clear(PG_STATE_RECOVERY_TOOFULL);
  state_clear(PG_STATE_RECOVERING);

  peer_purged.clear();
  acting_recovery_backfill.clear();

  // reset primary/replica state?
  if (was_old_primary || is_primary()) {
    pl->clear_want_pg_temp();
  } else if (was_old_nonprimary || is_nonprimary()) {
    pl->clear_want_pg_temp();
  }
  clear_primary_state();

  pl->on_change(t);

  ceph_assert(!deleting);

  // should we tell the primary we are here?
  send_notify = !is_primary();

  if (role != oldrole ||
      was_old_primary != is_primary()) {
    // did primary change?
    if (was_old_primary != is_primary()) {
      state_clear(PG_STATE_CLEAN);
      // queue/dequeue the scrubber
      pl->on_primary_status_change(was_old_primary, is_primary());
    }

    pl->on_role_change();
  } else {
    // no role change.
    // did primary change?
    if (get_primary() != old_acting_primary) {
      psdout(10) << oldacting << " -> " << acting
               << ", acting primary "
               << old_acting_primary << " -> " << get_primary()
               << dendl;
    } else {
      // primary is the same.
      if (is_primary()) {
        // i am (still) primary. but my replica set changed.
        state_clear(PG_STATE_CLEAN);

        psdout(10) << oldacting << " -> " << acting
                 << ", replicas changed" << dendl;
      }
    }
  }

  if (is_primary() && was_old_primary) {
    pl->reschedule_scrub();
  }

  if (acting.empty() && !up.empty() && up_primary == pg_whoami) {
    psdout(10) << " acting empty, but i am up[0], clearing pg_temp" << dendl;
    pl->queue_want_pg_temp(acting);
  }
}

void PeeringState::on_new_interval()
{
  dout(20) << __func__ << dendl;
  const OSDMapRef osdmap = get_osdmap();

  // initialize features
  acting_features = CEPH_FEATURES_SUPPORTED_DEFAULT;
  upacting_features = CEPH_FEATURES_SUPPORTED_DEFAULT;
  for (auto p = acting.begin(); p != acting.end(); ++p) {
    if (*p == CRUSH_ITEM_NONE)
      continue;
    uint64_t f = osdmap->get_xinfo(*p).features;
    acting_features &= f;
    upacting_features &= f;
  }
  for (auto p = up.begin(); p != up.end(); ++p) {
    if (*p == CRUSH_ITEM_NONE)
      continue;
    upacting_features &= osdmap->get_xinfo(*p).features;
  }
  psdout(20) << __func__ << " upacting_features 0x" << std::hex
             << upacting_features << std::dec
             << " from " << acting << "+" << up << dendl;

  psdout(20) << __func__ << " checking missing set deletes flag. missing = "
             << get_pg_log().get_missing() << dendl;

  if (!pg_log.get_missing().may_include_deletes &&
      !perform_deletes_during_peering()) {
    pl->rebuild_missing_set_with_deletes(pg_log);
  }
  ceph_assert(
    pg_log.get_missing().may_include_deletes ==
    !perform_deletes_during_peering());

  init_hb_stamps();

  // update lease bounds for a new interval
  auto mnow = pl->get_mnow();
  prior_readable_until_ub = std::max(prior_readable_until_ub,
                                     readable_until_ub);
  prior_readable_until_ub = info.history.refresh_prior_readable_until_ub(
    mnow, prior_readable_until_ub);
  psdout(10) << __func__ << " prior_readable_until_ub "
             << prior_readable_until_ub << " (mnow " << mnow << " + "
             << info.history.prior_readable_until_ub << ")" << dendl;
  prior_readable_down_osds.clear(); // we populate this when we build the priorset

  readable_until =
    readable_until_ub =
    readable_until_ub_sent =
    readable_until_ub_from_primary = ceph::signedspan::zero();

  acting_readable_until_ub.clear();
  if (is_primary()) {
    acting_readable_until_ub.resize(acting.size(), ceph::signedspan::zero());
  }

  pl->on_new_interval();
}

void PeeringState::init_primary_up_acting(
  const vector<int> &newup,
  const vector<int> &newacting,
  int new_up_primary,
  int new_acting_primary)
{
  actingset.clear();
  acting = newacting;
  for (uint8_t i = 0; i < acting.size(); ++i) {
    if (acting[i] != CRUSH_ITEM_NONE)
      actingset.insert(
        pg_shard_t(
          acting[i],
          pool.info.is_erasure() ? shard_id_t(i) : shard_id_t::NO_SHARD));
  }
  upset.clear();
  up = newup;
  for (uint8_t i = 0; i < up.size(); ++i) {
    if (up[i] != CRUSH_ITEM_NONE)
      upset.insert(
        pg_shard_t(
          up[i],
          pool.info.is_erasure() ? shard_id_t(i) : shard_id_t::NO_SHARD));
  }
  if (!pool.info.is_erasure()) {
    // replicated
    up_primary = pg_shard_t(new_up_primary, shard_id_t::NO_SHARD);
    primary = pg_shard_t(new_acting_primary, shard_id_t::NO_SHARD);
  } else {
    // erasure
    up_primary = pg_shard_t();
    primary = pg_shard_t();
    for (uint8_t i = 0; i < up.size(); ++i) {
      if (up[i] == new_up_primary) {
        up_primary = pg_shard_t(up[i], shard_id_t(i));
        break;
      }
    }
    for (uint8_t i = 0; i < acting.size(); ++i) {
      if (acting[i] == new_acting_primary) {
        primary = pg_shard_t(acting[i], shard_id_t(i));
        break;
      }
    }
    ceph_assert(up_primary.osd == new_up_primary);
    ceph_assert(primary.osd == new_acting_primary);
  }
}

void PeeringState::init_hb_stamps()
{
  if (is_primary()) {
    // we care about all other osds in the acting set
    hb_stamps.resize(acting.size() - 1);
    unsigned i = 0;
    for (auto p : acting) {
      if (p == CRUSH_ITEM_NONE || p == get_primary().osd) {
        continue;
      }
      hb_stamps[i++] = pl->get_hb_stamps(p);
    }
    hb_stamps.resize(i);
  } else if (is_nonprimary()) {
    // we care about just the primary
    hb_stamps.resize(1);
    hb_stamps[0] = pl->get_hb_stamps(get_primary().osd);
  } else {
    hb_stamps.clear();
  }
  dout(10) << __func__ << " now " << hb_stamps << dendl;
}


void PeeringState::clear_recovery_state()
{
  async_recovery_targets.clear();
  backfill_targets.clear();
}

void PeeringState::clear_primary_state()
{
  psdout(10) << "clear_primary_state" << dendl;

  // clear peering state
  stray_set.clear();
  peer_log_requested.clear();
  peer_missing_requested.clear();
  peer_info.clear();
  peer_bytes.clear();
  peer_missing.clear();
  peer_last_complete_ondisk.clear();
  peer_activated.clear();
  min_last_complete_ondisk = eversion_t();
  pg_trim_to = eversion_t();
  might_have_unfound.clear();
  need_up_thru = false;
  missing_loc.clear();
  pg_log.reset_recovery_pointers();

  clear_recovery_state();

  last_update_ondisk = eversion_t();
  missing_loc.clear();
  pl->clear_primary_state();
}

/// return [start,end) bounds for required past_intervals
static pair<epoch_t, epoch_t> get_required_past_interval_bounds(
  const pg_info_t &info,
  epoch_t oldest_map) {
  epoch_t start = std::max(
    info.history.last_epoch_clean ? info.history.last_epoch_clean :
    info.history.epoch_pool_created,
    oldest_map);
  epoch_t end = std::max(
    info.history.same_interval_since,
    info.history.epoch_pool_created);
  return make_pair(start, end);
}


void PeeringState::check_past_interval_bounds() const
{
  auto oldest_epoch = pl->oldest_stored_osdmap();
  auto rpib = get_required_past_interval_bounds(
    info,
    oldest_epoch);
  if (rpib.first >= rpib.second) {
    // do not warn if the start bound is dictated by oldest_map; the
    // past intervals are presumably appropriate given the pg info.
    if (!past_intervals.empty() &&
        rpib.first > oldest_epoch) {
      pl->get_clog_error() << info.pgid << " required past_interval bounds are"
                             << " empty [" << rpib << ") but past_intervals is not: "
                             << past_intervals;
      derr << info.pgid << " required past_interval bounds are"
           << " empty [" << rpib << ") but past_intervals is not: "
           << past_intervals << dendl;
    }
  } else {
    if (past_intervals.empty()) {
      pl->get_clog_error() << info.pgid << " required past_interval bounds are"
                             << " not empty [" << rpib << ") but past_intervals "
                             << past_intervals << " is empty";
      derr << info.pgid << " required past_interval bounds are"
           << " not empty [" << rpib << ") but past_intervals "
           << past_intervals << " is empty" << dendl;
      ceph_assert(!past_intervals.empty());
    }

    auto apib = past_intervals.get_bounds();
    if (apib.first > rpib.first) {
      pl->get_clog_error() << info.pgid << " past_intervals [" << apib
                             << ") start interval does not contain the required"
                             << " bound [" << rpib << ") start";
      derr << info.pgid << " past_intervals [" << apib
           << ") start interval does not contain the required"
           << " bound [" << rpib << ") start" << dendl;
      ceph_abort_msg("past_interval start interval mismatch");
    }
    if (apib.second != rpib.second) {
      pl->get_clog_error() << info.pgid << " past_interal bound [" << apib
                             << ") end does not match required [" << rpib
                             << ") end";
      derr << info.pgid << " past_interal bound [" << apib
           << ") end does not match required [" << rpib
           << ") end" << dendl;
      ceph_abort_msg("past_interval end mismatch");
    }
  }
}

int PeeringState::clamp_recovery_priority(int priority, int pool_recovery_priority, int max)
{
  static_assert(OSD_RECOVERY_PRIORITY_MIN < OSD_RECOVERY_PRIORITY_MAX, "Invalid priority range");
  static_assert(OSD_RECOVERY_PRIORITY_MIN >= 0, "Priority range must match unsigned type");

  ceph_assert(max <= OSD_RECOVERY_PRIORITY_MAX);

  // User can't set this too high anymore, but might be a legacy value
  if (pool_recovery_priority > OSD_POOL_PRIORITY_MAX)
    pool_recovery_priority = OSD_POOL_PRIORITY_MAX;
  if (pool_recovery_priority < OSD_POOL_PRIORITY_MIN)
    pool_recovery_priority = OSD_POOL_PRIORITY_MIN;
  // Shift range from min to max to 0 to max - min
  pool_recovery_priority += (0 - OSD_POOL_PRIORITY_MIN);
  ceph_assert(pool_recovery_priority >= 0 && pool_recovery_priority <= (OSD_POOL_PRIORITY_MAX - OSD_POOL_PRIORITY_MIN));

  priority += pool_recovery_priority;

  // Clamp to valid range
  if (priority > max) {
    return max;
  } else if (priority < OSD_RECOVERY_PRIORITY_MIN) {
    return OSD_RECOVERY_PRIORITY_MIN;
  } else {
    return priority;
  }
}

unsigned PeeringState::get_recovery_priority()
{
  // a higher value -> a higher priority
  int ret = OSD_RECOVERY_PRIORITY_BASE;
  int base = ret;

  if (state & PG_STATE_FORCED_RECOVERY) {
    ret = OSD_RECOVERY_PRIORITY_FORCED;
  } else {
    // XXX: This priority boost isn't so much about inactive, but about data-at-risk
    if (is_degraded() && info.stats.avail_no_missing.size() < pool.info.min_size) {
      base = OSD_RECOVERY_INACTIVE_PRIORITY_BASE;
      // inactive: no. of replicas < min_size, highest priority since it blocks IO
      ret = base + (pool.info.min_size - info.stats.avail_no_missing.size());
    }

    int64_t pool_recovery_priority = 0;
    pool.info.opts.get(pool_opts_t::RECOVERY_PRIORITY, &pool_recovery_priority);

    ret = clamp_recovery_priority(ret, pool_recovery_priority, max_prio_map[base]);
  }
  psdout(20) << __func__ << " recovery priority is " << ret << dendl;
  return static_cast<unsigned>(ret);
}

unsigned PeeringState::get_backfill_priority()
{
  // a higher value -> a higher priority
  int ret = OSD_BACKFILL_PRIORITY_BASE;
  int base = ret;

  if (state & PG_STATE_FORCED_BACKFILL) {
    ret = OSD_BACKFILL_PRIORITY_FORCED;
  } else {
    if (actingset.size() < pool.info.min_size) {
      base = OSD_BACKFILL_INACTIVE_PRIORITY_BASE;
      // inactive: no. of replicas < min_size, highest priority since it blocks IO
      ret = base + (pool.info.min_size - actingset.size());

    } else if (is_undersized()) {
      // undersized: OSD_BACKFILL_DEGRADED_PRIORITY_BASE + num missing replicas
      ceph_assert(pool.info.size > actingset.size());
      base = OSD_BACKFILL_DEGRADED_PRIORITY_BASE;
      ret = base + (pool.info.size - actingset.size());

    } else if (is_degraded()) {
      // degraded: baseline degraded
      base = ret = OSD_BACKFILL_DEGRADED_PRIORITY_BASE;
    }

    // Adjust with pool's recovery priority
    int64_t pool_recovery_priority = 0;
    pool.info.opts.get(pool_opts_t::RECOVERY_PRIORITY, &pool_recovery_priority);

    ret = clamp_recovery_priority(ret, pool_recovery_priority, max_prio_map[base]);
  }

  psdout(20) << __func__ << " backfill priority is " << ret << dendl;
  return static_cast<unsigned>(ret);
}

unsigned PeeringState::get_delete_priority()
{
  auto state = get_osdmap()->get_state(pg_whoami.osd);
  if (state & (CEPH_OSD_BACKFILLFULL |
               CEPH_OSD_FULL)) {
    return OSD_DELETE_PRIORITY_FULL;
  } else if (state & CEPH_OSD_NEARFULL) {
    return OSD_DELETE_PRIORITY_FULLISH;
  } else {
    return OSD_DELETE_PRIORITY_NORMAL;
  }
}

bool PeeringState::set_force_recovery(bool b)
{
  bool did = false;
  if (b) {
    if (!(state & PG_STATE_FORCED_RECOVERY) &&
        (state & (PG_STATE_DEGRADED |
                  PG_STATE_RECOVERY_WAIT |
                  PG_STATE_RECOVERING))) {
      psdout(20) << __func__ << " set" << dendl;
      state_set(PG_STATE_FORCED_RECOVERY);
      pl->publish_stats_to_osd();
      did = true;
    }
  } else if (state & PG_STATE_FORCED_RECOVERY) {
    psdout(20) << __func__ << " clear" << dendl;
    state_clear(PG_STATE_FORCED_RECOVERY);
    pl->publish_stats_to_osd();
    did = true;
  }
  if (did) {
    psdout(20) << __func__ << " state " << get_current_state()
             << dendl;
    pl->update_local_background_io_priority(get_recovery_priority());
  }
  return did;
}

bool PeeringState::set_force_backfill(bool b)
{
  bool did = false;
  if (b) {
    if (!(state & PG_STATE_FORCED_BACKFILL) &&
        (state & (PG_STATE_DEGRADED |
                  PG_STATE_BACKFILL_WAIT |
                  PG_STATE_BACKFILLING))) {
      psdout(10) << __func__ << " set" << dendl;
      state_set(PG_STATE_FORCED_BACKFILL);
      pl->publish_stats_to_osd();
      did = true;
    }
  } else if (state & PG_STATE_FORCED_BACKFILL) {
    psdout(10) << __func__ << " clear" << dendl;
    state_clear(PG_STATE_FORCED_BACKFILL);
    pl->publish_stats_to_osd();
    did = true;
  }
  if (did) {
    psdout(20) << __func__ << " state " << get_current_state()
             << dendl;
    pl->update_local_background_io_priority(get_backfill_priority());
  }
  return did;
}

void PeeringState::schedule_renew_lease()
{
  pl->schedule_renew_lease(
    last_peering_reset,
    readable_interval / 2);
}

void PeeringState::send_lease()
{
  epoch_t epoch = pl->get_osdmap_epoch();
  for (auto peer : actingset) {
    if (peer == pg_whoami) {
      continue;
    }
    pl->send_cluster_message(
      peer.osd,
      TOPNSPC::make_message<MOSDPGLease>(epoch,
                      spg_t(spgid.pgid, peer.shard),
                      get_lease()),
      epoch);
  }
}

void PeeringState::proc_lease(const pg_lease_t& l)
{
  assert(HAVE_FEATURE(upacting_features, SERVER_OCTOPUS));
  if (!is_nonprimary()) {
    psdout(20) << __func__ << " no-op, !nonprimary" << dendl;
    return;
  }
  psdout(10) << __func__ << " " << l << dendl;
  if (l.readable_until_ub > readable_until_ub_from_primary) {
    readable_until_ub_from_primary = l.readable_until_ub;
  }

  ceph::signedspan ru = ceph::signedspan::zero();
  if (l.readable_until != ceph::signedspan::zero() &&
      hb_stamps[0]->peer_clock_delta_ub) {
    ru = l.readable_until - *hb_stamps[0]->peer_clock_delta_ub;
    psdout(20) << " peer_clock_delta_ub " << *hb_stamps[0]->peer_clock_delta_ub
               << " -> ru " << ru << dendl;
  }
  if (ru > readable_until) {
    readable_until = ru;
    psdout(20) << __func__ << " readable_until now " << readable_until << dendl;
    // NOTE: if we ever decide to block/queue ops on the replica,
    // we'll need to wake them up here.
  }

  ceph::signedspan ruub;
  if (hb_stamps[0]->peer_clock_delta_lb) {
    ruub = l.readable_until_ub - *hb_stamps[0]->peer_clock_delta_lb;
    psdout(20) << " peer_clock_delta_lb " << *hb_stamps[0]->peer_clock_delta_lb
               << " -> ruub " << ruub << dendl;
  } else {
    ruub = pl->get_mnow() + l.interval;
    psdout(20) << " no peer_clock_delta_lb -> ruub " << ruub << dendl;
  }
  if (ruub > readable_until_ub) {
    readable_until_ub = ruub;
    psdout(20) << __func__ << " readable_until_ub now " << readable_until_ub
               << dendl;
  }
}

void PeeringState::proc_lease_ack(int from, const pg_lease_ack_t& a)
{
  assert(HAVE_FEATURE(upacting_features, SERVER_OCTOPUS));
  auto now = pl->get_mnow();
  bool was_min = false;
  for (unsigned i = 0; i < acting.size(); ++i) {
    if (from == acting[i]) {
      // the lease_ack value is based on the primary's clock
      if (a.readable_until_ub > acting_readable_until_ub[i]) {
        if (acting_readable_until_ub[i] == readable_until) {
          was_min = true;
        }
        acting_readable_until_ub[i] = a.readable_until_ub;
        break;
      }
    }
  }
  if (was_min) {
    auto old_ru = readable_until;
    recalc_readable_until();
    if (now < old_ru) {
      pl->recheck_readable();
    }
  }
}

void PeeringState::proc_renew_lease()
{
  assert(HAVE_FEATURE(upacting_features, SERVER_OCTOPUS));
  renew_lease(pl->get_mnow());
  send_lease();
  schedule_renew_lease();
}

void PeeringState::recalc_readable_until()
{
  assert(is_primary());
  ceph::signedspan min = readable_until_ub_sent;
  for (unsigned i = 0; i < acting.size(); ++i) {
    if (acting[i] == pg_whoami.osd || acting[i] == CRUSH_ITEM_NONE) {
      continue;
    }
    dout(20) << __func__ << " peer osd." << acting[i]
             << " ruub " << acting_readable_until_ub[i] << dendl;
    if (acting_readable_until_ub[i] < min) {
      min = acting_readable_until_ub[i];
    }
  }
  readable_until = min;
  readable_until_ub = min;
  dout(20) << __func__ << " readable_until[_ub] " << readable_until
           << " (sent " << readable_until_ub_sent << ")" << dendl;
}

bool PeeringState::check_prior_readable_down_osds(const OSDMapRef& map)
{
  assert(HAVE_FEATURE(upacting_features, SERVER_OCTOPUS));
  bool changed = false;
  auto p = prior_readable_down_osds.begin();
  while (p != prior_readable_down_osds.end()) {
    if (map->is_dead(*p)) {
      dout(10) << __func__ << " prior_readable_down_osds osd." << *p
               << " is dead as of epoch " << map->get_epoch()
               << dendl;
      p = prior_readable_down_osds.erase(p);
      changed = true;
    } else {
      ++p;
    }
  }
  if (changed && prior_readable_down_osds.empty()) {
    psdout(10) << " empty prior_readable_down_osds, clearing ub" << dendl;
    clear_prior_readable_until_ub();
    return true;
  }
  return false;
}

bool PeeringState::adjust_need_up_thru(const OSDMapRef osdmap)
{
  epoch_t up_thru = osdmap->get_up_thru(pg_whoami.osd);
  if (need_up_thru &&
      up_thru >= info.history.same_interval_since) {
    psdout(10) << "adjust_need_up_thru now "
               << up_thru << ", need_up_thru now false" << dendl;
    need_up_thru = false;
    return true;
  }
  return false;
}

PastIntervals::PriorSet PeeringState::build_prior()
{
  if (1) {
    // sanity check
    for (auto it = peer_info.begin(); it != peer_info.end(); ++it) {
      ceph_assert(info.history.last_epoch_started >=
                  it->second.history.last_epoch_started);
    }
  }

  const OSDMap &osdmap = *get_osdmap();
  PastIntervals::PriorSet prior = past_intervals.get_prior_set(
    pool.info.is_erasure(),
    info.history.last_epoch_started,
    &missing_loc.get_recoverable_predicate(),
    [&](epoch_t start, int osd, epoch_t *lost_at) {
      const osd_info_t *pinfo = 0;
      if (osdmap.exists(osd)) {
        pinfo = &osdmap.get_info(osd);
        if (lost_at)
          *lost_at = pinfo->lost_at;
      }

      if (osdmap.is_up(osd)) {
        return PastIntervals::UP;
      } else if (!pinfo) {
        return PastIntervals::DNE;
      } else if (pinfo->lost_at > start) {
        return PastIntervals::LOST;
      } else {
        return PastIntervals::DOWN;
      }
    },
    up,
    acting,
    dpp);

  if (prior.pg_down) {
    state_set(PG_STATE_DOWN);
  }

  if (get_osdmap()->get_up_thru(pg_whoami.osd) <
      info.history.same_interval_since) {
    psdout(10) << "up_thru " << get_osdmap()->get_up_thru(pg_whoami.osd)
               << " < same_since " << info.history.same_interval_since
               << ", must notify monitor" << dendl;
    need_up_thru = true;
  } else {
    psdout(10) << "up_thru " << get_osdmap()->get_up_thru(pg_whoami.osd)
               << " >= same_since " << info.history.same_interval_since
               << ", all is well" << dendl;
    need_up_thru = false;
  }
  pl->set_probe_targets(prior.probe);
  return prior;
}

bool PeeringState::needs_recovery() const
{
  ceph_assert(is_primary());

  auto &missing = pg_log.get_missing();

  if (missing.num_missing()) {
    psdout(10) << __func__ << " primary has " << missing.num_missing()
               << " missing" << dendl;
    return true;
  }

  ceph_assert(!acting_recovery_backfill.empty());
  for (const pg_shard_t& peer : acting_recovery_backfill) {
    if (peer == get_primary()) {
      continue;
    }
    auto pm = peer_missing.find(peer);
    if (pm == peer_missing.end()) {
      psdout(10) << __func__ << " osd." << peer << " doesn't have missing set"
                 << dendl;
      continue;
    }
    if (pm->second.num_missing()) {
      psdout(10) << __func__ << " osd." << peer << " has "
                 << pm->second.num_missing() << " missing" << dendl;
      return true;
    }
  }

  psdout(10) << __func__ << " is recovered" << dendl;
  return false;
}

bool PeeringState::needs_backfill() const
{
  ceph_assert(is_primary());

  // We can assume that only possible osds that need backfill
  // are on the backfill_targets vector nodes.
  for (const pg_shard_t& peer : backfill_targets) {
    auto pi = peer_info.find(peer);
    ceph_assert(pi != peer_info.end());
    if (!pi->second.last_backfill.is_max()) {
      psdout(10) << __func__ << " osd." << peer
                 << " has last_backfill " << pi->second.last_backfill << dendl;
      return true;
    }
  }

  psdout(10) << __func__ << " does not need backfill" << dendl;
  return false;
}

/*
 * Returns true unless there is a non-lost OSD in might_have_unfound.
 */
bool PeeringState::all_unfound_are_queried_or_lost(
  const OSDMapRef osdmap) const
{
  ceph_assert(is_primary());

  auto peer = might_have_unfound.begin();
  auto mend = might_have_unfound.end();
  for (; peer != mend; ++peer) {
    if (peer_missing.count(*peer))
      continue;
    auto iter = peer_info.find(*peer);
    if (iter != peer_info.end() &&
        (iter->second.is_empty() || iter->second.dne()))
      continue;
    if (!osdmap->exists(peer->osd))
      continue;
    const osd_info_t &osd_info(osdmap->get_info(peer->osd));
    if (osd_info.lost_at <= osd_info.up_from) {
      // If there is even one OSD in might_have_unfound that isn't lost, we
      // still might retrieve our unfound.
      return false;
    }
  }
  psdout(10) << "all_unfound_are_queried_or_lost all of might_have_unfound "
             << might_have_unfound
             << " have been queried or are marked lost" << dendl;
  return true;
}


void PeeringState::reject_reservation()
{
  pl->unreserve_recovery_space();
  pl->send_cluster_message(
    primary.osd,
    TOPNSPC::make_message<MBackfillReserve>(
      MBackfillReserve::REJECT_TOOFULL,
      spg_t(info.pgid.pgid, primary.shard),
      get_osdmap_epoch()),
    get_osdmap_epoch());
}

/**
 * find_best_info
 *
 * Returns an iterator to the best info in infos sorted by:
 *  1) Prefer newer last_update
 *  2) Prefer longer tail if it brings another info into contiguity
 *  3) Prefer current primary
 */
map<pg_shard_t, pg_info_t>::const_iterator PeeringState::find_best_info(
  const map<pg_shard_t, pg_info_t> &infos,
  bool restrict_to_up_acting,
  bool *history_les_bound) const
{
  ceph_assert(history_les_bound);
  /* See doc/dev/osd_internals/last_epoch_started.rst before attempting
   * to make changes to this process.  Also, make sure to update it
   * when you find bugs! */
  epoch_t max_last_epoch_started_found = 0;
  for (auto i = infos.begin(); i != infos.end(); ++i) {
    if (!cct->_conf->osd_find_best_info_ignore_history_les &&
        max_last_epoch_started_found < i->second.history.last_epoch_started) {
      *history_les_bound = true;
      max_last_epoch_started_found = i->second.history.last_epoch_started;
    }
    if (!i->second.is_incomplete() &&
        max_last_epoch_started_found < i->second.last_epoch_started) {
      *history_les_bound = false;
      max_last_epoch_started_found = i->second.last_epoch_started;
    }
  }
  eversion_t min_last_update_acceptable = eversion_t::max();
  for (auto i = infos.begin(); i != infos.end(); ++i) {
    if (max_last_epoch_started_found <= i->second.last_epoch_started) {
      if (min_last_update_acceptable > i->second.last_update)
        min_last_update_acceptable = i->second.last_update;
    }
  }
  if (min_last_update_acceptable == eversion_t::max())
    return infos.end();

  auto best = infos.end();
  // find osd with newest last_update (oldest for ec_pool).
  // if there are multiples, prefer
  //  - a longer tail, if it brings another peer into log contiguity
  //  - the current primary
  for (auto p = infos.begin(); p != infos.end(); ++p) {
    if (restrict_to_up_acting && !is_up(p->first) &&
        !is_acting(p->first))
      continue;
    // Only consider peers with last_update >= min_last_update_acceptable
    if (p->second.last_update < min_last_update_acceptable)
      continue;
    // Disqualify anyone with a too old last_epoch_started
    if (p->second.last_epoch_started < max_last_epoch_started_found)
      continue;
    // Disqualify anyone who is incomplete (not fully backfilled)
    if (p->second.is_incomplete())
      continue;
    if (best == infos.end()) {
      best = p;
      continue;
    }
    // Prefer newer last_update
    if (pool.info.require_rollback()) {
      if (p->second.last_update > best->second.last_update)
        continue;
      if (p->second.last_update < best->second.last_update) {
        best = p;
        continue;
      }
    } else {
      if (p->second.last_update < best->second.last_update)
        continue;
      if (p->second.last_update > best->second.last_update) {
        best = p;
        continue;
      }
    }

    // Prefer longer tail
    if (p->second.log_tail > best->second.log_tail) {
      continue;
    } else if (p->second.log_tail < best->second.log_tail) {
      best = p;
      continue;
    }

    if (!p->second.has_missing() && best->second.has_missing()) {
      psdout(10) << __func__ << " prefer osd." << p->first
               << " because it is complete while best has missing"
               << dendl;
      best = p;
      continue;
    } else if (p->second.has_missing() && !best->second.has_missing()) {
      psdout(10) << __func__ << " skipping osd." << p->first
               << " because it has missing while best is complete"
               << dendl;
      continue;
    } else {
      // both are complete or have missing
      // fall through
    }

    // prefer current primary (usually the caller), all things being equal
    if (p->first == pg_whoami) {
      psdout(10) << "calc_acting prefer osd." << p->first
                 << " because it is current primary" << dendl;
      best = p;
      continue;
    }
  }
  return best;
}

void PeeringState::calc_ec_acting(
  map<pg_shard_t, pg_info_t>::const_iterator auth_log_shard,
  unsigned size,
  const vector<int> &acting,
  const vector<int> &up,
  const map<pg_shard_t, pg_info_t> &all_info,
  bool restrict_to_up_acting,
  vector<int> *_want,
  set<pg_shard_t> *backfill,
  set<pg_shard_t> *acting_backfill,
  ostream &ss)
{
  vector<int> want(size, CRUSH_ITEM_NONE);
  map<shard_id_t, set<pg_shard_t> > all_info_by_shard;
  for (auto i = all_info.begin();
       i != all_info.end();
       ++i) {
    all_info_by_shard[i->first.shard].insert(i->first);
  }
  for (uint8_t i = 0; i < want.size(); ++i) {
    ss << "For position " << (unsigned)i << ": ";
    if (up.size() > (unsigned)i && up[i] != CRUSH_ITEM_NONE &&
        !all_info.find(pg_shard_t(up[i], shard_id_t(i)))->second.is_incomplete() &&
        all_info.find(pg_shard_t(up[i], shard_id_t(i)))->second.last_update >=
        auth_log_shard->second.log_tail) {
      ss << " selecting up[i]: " << pg_shard_t(up[i], shard_id_t(i)) << std::endl;
      want[i] = up[i];
      continue;
    }
    if (up.size() > (unsigned)i && up[i] != CRUSH_ITEM_NONE) {
      ss << " backfilling up[i]: " << pg_shard_t(up[i], shard_id_t(i))
         << " and ";
      backfill->insert(pg_shard_t(up[i], shard_id_t(i)));
    }

    if (acting.size() > (unsigned)i && acting[i] != CRUSH_ITEM_NONE &&
        !all_info.find(pg_shard_t(acting[i], shard_id_t(i)))->second.is_incomplete() &&
        all_info.find(pg_shard_t(acting[i], shard_id_t(i)))->second.last_update >=
        auth_log_shard->second.log_tail) {
      ss << " selecting acting[i]: " << pg_shard_t(acting[i], shard_id_t(i)) << std::endl;
      want[i] = acting[i];
    } else if (!restrict_to_up_acting) {
      for (auto j = all_info_by_shard[shard_id_t(i)].begin();
           j != all_info_by_shard[shard_id_t(i)].end();
           ++j) {
        ceph_assert(j->shard == i);
        if (!all_info.find(*j)->second.is_incomplete() &&
            all_info.find(*j)->second.last_update >=
            auth_log_shard->second.log_tail) {
          ss << " selecting stray: " << *j << std::endl;
          want[i] = j->osd;
          break;
        }
      }
      if (want[i] == CRUSH_ITEM_NONE)
        ss << " failed to fill position " << (int)i << std::endl;
    }
  }

  for (uint8_t i = 0; i < want.size(); ++i) {
    if (want[i] != CRUSH_ITEM_NONE) {
      acting_backfill->insert(pg_shard_t(want[i], shard_id_t(i)));
    }
  }
  acting_backfill->insert(backfill->begin(), backfill->end());
  _want->swap(want);
}

std::pair<map<pg_shard_t, pg_info_t>::const_iterator, eversion_t>
PeeringState::select_replicated_primary(
  map<pg_shard_t, pg_info_t>::const_iterator auth_log_shard,
  uint64_t force_auth_primary_missing_objects,
  const std::vector<int> &up,
  pg_shard_t up_primary,
  const map<pg_shard_t, pg_info_t> &all_info,
  const OSDMapRef osdmap,
  ostream &ss)
{
  pg_shard_t auth_log_shard_id = auth_log_shard->first;

  ss << __func__ << " newest update on osd." << auth_log_shard_id
     << " with " << auth_log_shard->second << std::endl;

  // select primary
  auto primary = all_info.find(up_primary);
  if (up.size() &&
      !primary->second.is_incomplete() &&
      primary->second.last_update >=
        auth_log_shard->second.log_tail) {
    assert(HAVE_FEATURE(osdmap->get_up_osd_features(), SERVER_NAUTILUS));
    auto approx_missing_objects =
      primary->second.stats.stats.sum.num_objects_missing;
    auto auth_version = auth_log_shard->second.last_update.version;
    auto primary_version = primary->second.last_update.version;
    if (auth_version > primary_version) {
      approx_missing_objects += auth_version - primary_version;
    } else {
      approx_missing_objects += primary_version - auth_version;
    }
    if ((uint64_t)approx_missing_objects >
        force_auth_primary_missing_objects) {
      primary = auth_log_shard;
      ss << "up_primary: " << up_primary << ") has approximate "
         << approx_missing_objects
         << "(>" << force_auth_primary_missing_objects <<") "
         << "missing objects, osd." << auth_log_shard_id
         << " selected as primary instead"
         << std::endl;
    } else {
      ss << "up_primary: " << up_primary << ") selected as primary"
         << std::endl;
    }
  } else {
    ceph_assert(!auth_log_shard->second.is_incomplete());
    ss << "up[0] needs backfill, osd." << auth_log_shard_id
       << " selected as primary instead" << std::endl;
    primary = auth_log_shard;
  }

  ss << __func__ << " primary is osd." << primary->first
     << " with " << primary->second << std::endl;

  /* We include auth_log_shard->second.log_tail because in GetLog,
   * we will request logs back to the min last_update over our
   * acting_backfill set, which will result in our log being extended
   * as far backwards as necessary to pick up any peers which can
   * be log recovered by auth_log_shard's log */
  eversion_t oldest_auth_log_entry =
    std::min(primary->second.log_tail, auth_log_shard->second.log_tail);

  return std::make_pair(primary, oldest_auth_log_entry);
}


/**
 * calculate the desired acting set.
 *
 * Choose an appropriate acting set.  Prefer up[0], unless it is
 * incomplete, or another osd has a longer tail that allows us to
 * bring other up nodes up to date.
 */
void PeeringState::calc_replicated_acting(
  map<pg_shard_t, pg_info_t>::const_iterator primary,
  eversion_t oldest_auth_log_entry,
  unsigned size,
  const vector<int> &acting,
  const vector<int> &up,
  pg_shard_t up_primary,
  const map<pg_shard_t, pg_info_t> &all_info,
  bool restrict_to_up_acting,
  vector<int> *want,
  set<pg_shard_t> *backfill,
  set<pg_shard_t> *acting_backfill,
  const OSDMapRef osdmap,
  const PGPool& pool,
  ostream &ss)
{
  ss << __func__ << (restrict_to_up_acting ? " restrict_to_up_acting" : "")
     << std::endl;

  want->push_back(primary->first.osd);
  acting_backfill->insert(primary->first);

  // select replicas that have log contiguity with primary.
  // prefer up, then acting, then any peer_info osds
  for (auto i : up) {
    pg_shard_t up_cand = pg_shard_t(i, shard_id_t::NO_SHARD);
    if (up_cand == primary->first)
      continue;
    const pg_info_t &cur_info = all_info.find(up_cand)->second;
    if (cur_info.is_incomplete() ||
        cur_info.last_update < oldest_auth_log_entry) {
      ss << " shard " << up_cand << " (up) backfill " << cur_info << std::endl;
      backfill->insert(up_cand);
      acting_backfill->insert(up_cand);
    } else {
      want->push_back(i);
      acting_backfill->insert(up_cand);
      ss << " osd." << i << " (up) accepted " << cur_info << std::endl;
    }
  }

  if (want->size() >= size) {
    return;
  }

  std::vector<std::pair<eversion_t, int>> candidate_by_last_update;
  candidate_by_last_update.reserve(acting.size());
  // This no longer has backfill OSDs, but they are covered above.
  for (auto i : acting) {
    pg_shard_t acting_cand(i, shard_id_t::NO_SHARD);
    // skip up osds we already considered above
    if (acting_cand == primary->first)
      continue;
    auto up_it = find(up.begin(), up.end(), i);
    if (up_it != up.end())
      continue;

    const pg_info_t &cur_info = all_info.find(acting_cand)->second;
    if (cur_info.is_incomplete() ||
        cur_info.last_update < oldest_auth_log_entry) {
      ss << " shard " << acting_cand << " (acting) REJECTED "
         << cur_info << std::endl;
    } else {
      candidate_by_last_update.emplace_back(cur_info.last_update, i);
    }
  }

  auto sort_by_eversion =[](const std::pair<eversion_t, int> &lhs,
                            const std::pair<eversion_t, int> &rhs) {
    return lhs.first > rhs.first;
  };
  // sort by last_update, in descending order.
  std::sort(candidate_by_last_update.begin(),
            candidate_by_last_update.end(), sort_by_eversion);
  for (auto &p: candidate_by_last_update) {
    ceph_assert(want->size() < size);
    want->push_back(p.second);
    pg_shard_t s = pg_shard_t(p.second, shard_id_t::NO_SHARD);
    acting_backfill->insert(s);
    ss << " shard " << s << " (acting) accepted "
       << all_info.find(s)->second << std::endl;
    if (want->size() >= size) {
      return;
    }
  }

  if (restrict_to_up_acting) {
    return;
  }
  candidate_by_last_update.clear();
  candidate_by_last_update.reserve(all_info.size()); // overestimate but fine
  // continue to search stray to find more suitable peers
  for (auto &i : all_info) {
    // skip up osds we already considered above
    if (i.first == primary->first)
      continue;
    auto up_it = find(up.begin(), up.end(), i.first.osd);
    if (up_it != up.end())
      continue;
    auto acting_it = find(
      acting.begin(), acting.end(), i.first.osd);
    if (acting_it != acting.end())
      continue;

    if (i.second.is_incomplete() ||
        i.second.last_update < oldest_auth_log_entry) {
      ss << " shard " << i.first << " (stray) REJECTED " << i.second
         << std::endl;
    } else {
      candidate_by_last_update.emplace_back(
        i.second.last_update, i.first.osd);
    }
  }

  if (candidate_by_last_update.empty()) {
    // save us some effort
    return;
  }

  // sort by last_update, in descending order.
  std::sort(candidate_by_last_update.begin(),
            candidate_by_last_update.end(), sort_by_eversion);

  for (auto &p: candidate_by_last_update) {
    ceph_assert(want->size() < size);
    want->push_back(p.second);
    pg_shard_t s = pg_shard_t(p.second, shard_id_t::NO_SHARD);
    acting_backfill->insert(s);
    ss << " shard " << s << " (stray) accepted "
       << all_info.find(s)->second << std::endl;
    if (want->size() >= size) {
      return;
    }
  }
}

// Defines osd preference order: acting set, then larger last_update
using osd_ord_t = std::tuple<bool, eversion_t>; // <acting, last_update>
using osd_id_t = int;

class bucket_candidates_t {
  std::deque<std::pair<osd_ord_t, osd_id_t>> osds;
  int selected = 0;

public:
  void add_osd(osd_ord_t ord, osd_id_t osd) {
    // osds will be added in smallest to largest order
    assert(osds.empty() || osds.back().first <= ord);
    osds.push_back(std::make_pair(ord, osd));
  }
  osd_id_t pop_osd() {
    ceph_assert(!is_empty());
    auto ret = osds.front();
    osds.pop_front();
    return ret.second;
  }

  void inc_selected() { selected++; }
  unsigned get_num_selected() const { return selected; }

  osd_ord_t get_ord() const {
    return osds.empty() ? std::make_tuple(false, eversion_t())
      : osds.front().first;
  }

  bool is_empty() const { return osds.empty(); }

  bool operator<(const bucket_candidates_t &rhs) const {
    return std::make_tuple(-selected, get_ord()) <
      std::make_tuple(-rhs.selected, rhs.get_ord());
  }

  friend std::ostream &operator<<(std::ostream &, const bucket_candidates_t &);
};

std::ostream &operator<<(std::ostream &lhs, const bucket_candidates_t &cand)
{
  return lhs << "candidates[" << cand.osds << "]";
}

class bucket_heap_t {
  using elem_t = std::reference_wrapper<bucket_candidates_t>;
  std::vector<elem_t> heap;

  // Max heap -- should emit buckets in order of preference
  struct comp {
    bool operator()(const elem_t &lhs, const elem_t &rhs) {
      return lhs.get() < rhs.get();
    }
  };
public:
  void push_if_nonempty(elem_t e) {
    if (!e.get().is_empty()) {
      heap.push_back(e);
      std::push_heap(heap.begin(), heap.end(), comp());
    }
  }
  elem_t pop() {
    std::pop_heap(heap.begin(), heap.end(), comp());
    auto ret = heap.back();
    heap.pop_back();
    return ret;
  }

  bool is_empty() const { return heap.empty(); }
};

/**
 * calc_replicated_acting_stretch
 *
 * Choose an acting set using as much of the up set as possible; filling 
 * in the remaining slots so as to maximize the number of crush buckets at
 * level pool.info.peering_crush_bucket_barrier represented.
 *
 * Stretch clusters are a bit special: while they have a "size" the
 * same way as normal pools, if we happen to lose a data center
 * (we call it a "stretch bucket", but really it'll be a data center or
 * a cloud availability zone), we don't actually want to shove
 * 2 DC's worth of replication into a single site -- it won't fit!
 * So we locally calculate a bucket_max, based
 * on the targeted number of stretch buckets for the pool and
 * its size. Then we won't pull more than bucket_max from any
 * given ancestor even if it leaves us undersized.

 * There are two distinct phases: (commented below)
 */
void PeeringState::calc_replicated_acting_stretch(
  map<pg_shard_t, pg_info_t>::const_iterator primary,
  eversion_t oldest_auth_log_entry,
  unsigned size,
  const vector<int> &acting,
  const vector<int> &up,
  pg_shard_t up_primary,
  const map<pg_shard_t, pg_info_t> &all_info,
  bool restrict_to_up_acting,
  vector<int> *want,
  set<pg_shard_t> *backfill,
  set<pg_shard_t> *acting_backfill,
  const OSDMapRef osdmap,
  const PGPool& pool,
  ostream &ss)
{
  ceph_assert(want);
  ceph_assert(acting_backfill);
  ceph_assert(backfill);
  ss << __func__ << (restrict_to_up_acting ? " restrict_to_up_acting" : "")
     << std::endl;

  auto used = [want](int osd) {
    return std::find(want->begin(), want->end(), osd) != want->end();
  };

  auto usable_info = [&](const auto &cur_info) mutable {
    return !(cur_info.is_incomplete() ||
             cur_info.last_update < oldest_auth_log_entry);
  };

  auto osd_info = [&](int osd) mutable -> const pg_info_t & {
    pg_shard_t cand = pg_shard_t(osd, shard_id_t::NO_SHARD);
    const pg_info_t &cur_info = all_info.find(cand)->second;
    return cur_info;
  };

  auto usable_osd = [&](int osd) mutable {
    return usable_info(osd_info(osd));
  };

  std::map<int, bucket_candidates_t> ancestors;
  auto get_ancestor = [&](int osd) mutable {
    int ancestor = osdmap->crush->get_parent_of_type(
      osd,
      pool.info.peering_crush_bucket_barrier,
      pool.info.crush_rule);
    return &ancestors[ancestor];
  };

  unsigned bucket_max = pool.info.size / pool.info.peering_crush_bucket_target;
  if (bucket_max * pool.info.peering_crush_bucket_target < pool.info.size) {
    ++bucket_max; 
  }

  /* 1) Select all usable osds from the up set as well as the primary
   *
   * We also stash any unusable osds from up into backfill.
   */
  auto add_required = [&](int osd) {
    if (!used(osd)) {
      want->push_back(osd);
      acting_backfill->insert(
        pg_shard_t(osd, shard_id_t::NO_SHARD));
      get_ancestor(osd)->inc_selected();
    }
  };
  add_required(primary->first.osd);
  ss << " osd " << primary->first.osd << " primary accepted "
     << osd_info(primary->first.osd) << std::endl;
  for (auto upcand: up) {
    auto upshard = pg_shard_t(upcand, shard_id_t::NO_SHARD);
    auto &curinfo = osd_info(upcand);
    if (usable_osd(upcand)) {
      ss << " osd " << upcand << " (up) accepted " << curinfo << std::endl;
      add_required(upcand);
    } else {
      ss << " osd " << upcand << " (up) backfill " << curinfo << std::endl;
      backfill->insert(upshard);
      acting_backfill->insert(upshard);
    }
  }

  if (want->size() >= pool.info.size) { // non-failed CRUSH mappings are valid
    ss << " up set sufficient" << std::endl;
    return;
  }
  ss << " up set insufficient, considering remaining osds" << std::endl;

  /* 2) Fill out remaining slots from usable osds in all_info
   *    while maximizing the number of ancestor nodes at the
   *    barrier_id crush level.
   */
  {
    std::vector<std::pair<osd_ord_t, osd_id_t>> candidates;
    /* To do this, we first filter the set of usable osd into an ordered
     * list of usable osds
     */
    auto get_osd_ord = [&](bool is_acting, const pg_info_t &info) -> osd_ord_t {
      return std::make_tuple(
        !is_acting /* acting should sort first */,
        info.last_update);
    };
    for (auto &cand : acting) {
      auto &cand_info = osd_info(cand);
      if (!used(cand) && usable_info(cand_info)) {
        ss << " acting candidate " << cand << " " << cand_info << std::endl;
        candidates.push_back(std::make_pair(get_osd_ord(true, cand_info), cand));
      }
    }
    if (!restrict_to_up_acting) {
      for (auto &[cand, info] : all_info) {
        if (!used(cand.osd) && usable_info(info) &&
            (std::find(acting.begin(), acting.end(), cand.osd)
             == acting.end())) {
          ss << " other candidate " << cand << " " << info << std::endl;
          candidates.push_back(
            std::make_pair(get_osd_ord(false, info), cand.osd));
        }
      }
    }
    std::sort(candidates.begin(), candidates.end());

    // We then filter these candidates by ancestor
    std::for_each(candidates.begin(), candidates.end(), [&](auto cand) {
      get_ancestor(cand.second)->add_osd(cand.first, cand.second);
    });
  }

  auto pop_ancestor = [&](auto &ancestor) {
    ceph_assert(!ancestor.is_empty());
    auto osd = ancestor.pop_osd();

    ss << " accepting candidate " << osd << std::endl;

    ceph_assert(!used(osd));
    ceph_assert(usable_osd(osd));

    want->push_back(osd);
    acting_backfill->insert(
      pg_shard_t(osd, shard_id_t::NO_SHARD));
    ancestor.inc_selected();
  };

  /* Next, we use the ancestors map to grab a descendant of the
   * peering_crush_mandatory_member if not already represented.
   *
   * TODO: using 0 here to match other users.  Prior to merge, I
   * expect that this and other users should instead check against
   * CRUSH_ITEM_NONE.
   */
  if (pool.info.peering_crush_mandatory_member != CRUSH_ITEM_NONE) {
    auto aiter = ancestors.find(pool.info.peering_crush_mandatory_member);
    if (aiter != ancestors.end() &&
        !aiter->second.get_num_selected()) {
      ss << " adding required ancestor " << aiter->first << std::endl;
      ceph_assert(!aiter->second.is_empty()); // wouldn't exist otherwise
      pop_ancestor(aiter->second);
    }
  }

  /* We then place the ancestors in a heap ordered by fewest selected
   * and then by the ordering token of the next osd */
  bucket_heap_t aheap;
  std::for_each(ancestors.begin(), ancestors.end(), [&](auto &anc) {
    aheap.push_if_nonempty(anc.second);
  });

  /* and pull from this heap until it's empty or we have enough.
   * "We have enough" is a sufficient check here for
   * stretch_set_can_peer() because our heap sorting always
   * pulls from ancestors with the least number of included OSDs,
   * so if it is possible to satisfy the bucket_count constraints we
   * will do so.
   */
  while (!aheap.is_empty() && want->size() < pool.info.size) {
    auto next = aheap.pop();
    pop_ancestor(next.get());
    if (next.get().get_num_selected() < bucket_max) {
      aheap.push_if_nonempty(next);
    }
  }

  /* The end result is that we should have as many buckets covered as
   * possible while respecting up, the primary selection,
   * the pool size (given bucket count constraints),
   * and the mandatory member.
   */
}


bool PeeringState::recoverable(const vector<int> &want) const
{
  unsigned num_want_acting = 0;
  set<pg_shard_t> have;
  for (int i = 0; i < (int)want.size(); ++i) {
    if (want[i] != CRUSH_ITEM_NONE) {
      ++num_want_acting;
      have.insert(
        pg_shard_t(
          want[i],
          pool.info.is_erasure() ? shard_id_t(i) : shard_id_t::NO_SHARD));
    }
  }

  if (num_want_acting < pool.info.min_size) {
    if (!cct->_conf.get_val<bool>("osd_allow_recovery_below_min_size")) {
      psdout(10) << __func__ << " failed, recovery below min size not enabled" << dendl;
      return false;
    }
  }
  if (missing_loc.get_recoverable_predicate()(have)) {
    return true;
  } else {
    psdout(10) << __func__ << " failed, not recoverable " << dendl;
    return false;
  }
}

void PeeringState::choose_async_recovery_ec(
  const map<pg_shard_t, pg_info_t> &all_info,
  const pg_info_t &auth_info,
  vector<int> *want,
  set<pg_shard_t> *async_recovery,
  const OSDMapRef osdmap) const
{
  set<pair<int, pg_shard_t> > candidates_by_cost;
  for (uint8_t i = 0; i < want->size(); ++i) {
    if ((*want)[i] == CRUSH_ITEM_NONE)
      continue;

    // Considering log entries to recover is accurate enough for
    // now. We could use minimum_to_decode_with_cost() later if
    // necessary.
    pg_shard_t shard_i((*want)[i], shard_id_t(i));
    // do not include strays
    if (stray_set.find(shard_i) != stray_set.end())
      continue;
    // Do not include an osd that is not up, since choosing it as
    // an async_recovery_target will move it out of the acting set.
    // This results in it being identified as a stray during peering,
    // because it is no longer in the up or acting set.
    if (!is_up(shard_i))
      continue;
    auto shard_info = all_info.find(shard_i)->second;
    // for ec pools we rollback all entries past the authoritative
    // last_update *before* activation. This is relatively inexpensive
    // compared to recovery, since it is purely local, so treat shards
    // past the authoritative last_update the same as those equal to it.
    version_t auth_version = auth_info.last_update.version;
    version_t candidate_version = shard_info.last_update.version;
    assert(HAVE_FEATURE(osdmap->get_up_osd_features(), SERVER_NAUTILUS));
    auto approx_missing_objects =
      shard_info.stats.stats.sum.num_objects_missing;
    if (auth_version > candidate_version) {
      approx_missing_objects += auth_version - candidate_version;
    }
    if (static_cast<uint64_t>(approx_missing_objects) >
       cct->_conf.get_val<uint64_t>("osd_async_recovery_min_cost")) {
      candidates_by_cost.emplace(approx_missing_objects, shard_i);
    }
  }

  psdout(20) << __func__ << " candidates by cost are: " << candidates_by_cost
             << dendl;

  // take out as many osds as we can for async recovery, in order of cost
  for (auto rit = candidates_by_cost.rbegin();
       rit != candidates_by_cost.rend(); ++rit) {
    pg_shard_t cur_shard = rit->second;
    vector<int> candidate_want(*want);
    candidate_want[cur_shard.shard.id] = CRUSH_ITEM_NONE;
    if (recoverable(candidate_want)) {
      want->swap(candidate_want);
      async_recovery->insert(cur_shard);
    }
  }
  psdout(20) << __func__ << " result want=" << *want
             << " async_recovery=" << *async_recovery << dendl;
}

void PeeringState::choose_async_recovery_replicated(
  const map<pg_shard_t, pg_info_t> &all_info,
  const pg_info_t &auth_info,
  vector<int> *want,
  set<pg_shard_t> *async_recovery,
  const OSDMapRef osdmap) const
{
  set<pair<int, pg_shard_t> > candidates_by_cost;
  for (auto osd_num : *want) {
    pg_shard_t shard_i(osd_num, shard_id_t::NO_SHARD);
    // do not include strays
    if (stray_set.find(shard_i) != stray_set.end())
      continue;
    // Do not include an osd that is not up, since choosing it as
    // an async_recovery_target will move it out of the acting set.
    // This results in it being identified as a stray during peering,
    // because it is no longer in the up or acting set.
    if (!is_up(shard_i))
      continue;
    auto shard_info = all_info.find(shard_i)->second;
    // use the approximate magnitude of the difference in length of
    // logs plus historical missing objects as the cost of recovery
    version_t auth_version = auth_info.last_update.version;
    version_t candidate_version = shard_info.last_update.version;
    assert(HAVE_FEATURE(osdmap->get_up_osd_features(), SERVER_NAUTILUS));
    auto approx_missing_objects =
      shard_info.stats.stats.sum.num_objects_missing;
    if (auth_version > candidate_version) {
      approx_missing_objects += auth_version - candidate_version;
    } else {
      approx_missing_objects += candidate_version - auth_version;
    }
    if (static_cast<uint64_t>(approx_missing_objects)  >
       cct->_conf.get_val<uint64_t>("osd_async_recovery_min_cost")) {
      candidates_by_cost.emplace(approx_missing_objects, shard_i);
    }
  }

  psdout(20) << __func__ << " candidates by cost are: " << candidates_by_cost
             << dendl;
  // take out as many osds as we can for async recovery, in order of cost
  for (auto rit = candidates_by_cost.rbegin();
       rit != candidates_by_cost.rend(); ++rit) {
    if (want->size() <= pool.info.min_size) {
      break;
    }
    pg_shard_t cur_shard = rit->second;
    vector<int> candidate_want(*want);
    for (auto it = candidate_want.begin(); it != candidate_want.end(); ++it) {
      if (*it == cur_shard.osd) {
        candidate_want.erase(it);
        if (pool.info.stretch_set_can_peer(candidate_want, *osdmap, NULL)) {
          // if we're in stretch mode, we can only remove the osd if it doesn't
          // break peering limits.
          want->swap(candidate_want);
          async_recovery->insert(cur_shard);
        }
        break;
      }
    }
  }

  psdout(20) << __func__ << " result want=" << *want
             << " async_recovery=" << *async_recovery << dendl;
}

/**
 * choose acting
 *
 * calculate the desired acting, and request a change with the monitor
 * if it differs from the current acting.
 *
 * if restrict_to_up_acting=true, we filter out anything that's not in
 * up/acting.  in order to lift this restriction, we need to
 *  1) check whether it's worth switching the acting set any time we get
 *     a new pg info (not just here, when recovery finishes)
 *  2) check whether anything in want_acting went down on each new map
 *     (and, if so, calculate a new want_acting)
 *  3) remove the assertion in PG::PeeringState::Active::react(const AdvMap)
 * TODO!
 */
bool PeeringState::choose_acting(pg_shard_t &auth_log_shard_id,
                                 bool restrict_to_up_acting,
                                 bool *history_les_bound,
                                 bool request_pg_temp_change_only)
{
  map<pg_shard_t, pg_info_t> all_info(peer_info.begin(), peer_info.end());
  all_info[pg_whoami] = info;

  if (cct->_conf->subsys.should_gather<dout_subsys, 10>()) {
    for (auto p = all_info.begin(); p != all_info.end(); ++p) {
      psdout(10) << __func__ << " all_info osd." << p->first << " "
                 << p->second << dendl;
    }
  }

  auto auth_log_shard = find_best_info(all_info, restrict_to_up_acting,
                                       history_les_bound);

  if (auth_log_shard == all_info.end()) {
    if (up != acting) {
      psdout(10) << __func__ << " no suitable info found (incomplete backfills?),"
                 << " reverting to up" << dendl;
      want_acting = up;
      vector<int> empty;
      pl->queue_want_pg_temp(empty);
    } else {
      psdout(10) << __func__ << " failed" << dendl;
      ceph_assert(want_acting.empty());
    }
    return false;
  }

  ceph_assert(!auth_log_shard->second.is_incomplete());
  auth_log_shard_id = auth_log_shard->first;

  set<pg_shard_t> want_backfill, want_acting_backfill;
  vector<int> want;
  stringstream ss;
  if (pool.info.is_replicated()) {
    auto [primary_shard, oldest_log] = select_replicated_primary(
      auth_log_shard,
      cct->_conf.get_val<uint64_t>(
        "osd_force_auth_primary_missing_objects"),
      up,
      up_primary,
      all_info,
      get_osdmap(),
      ss);
    if (pool.info.is_stretch_pool()) {
      calc_replicated_acting_stretch(
        primary_shard,
        oldest_log,
        get_osdmap()->get_pg_size(info.pgid.pgid),
        acting,
        up,
        up_primary,
        all_info,
        restrict_to_up_acting,
        &want,
        &want_backfill,
        &want_acting_backfill,
        get_osdmap(),
        pool,
        ss);
    } else {
      calc_replicated_acting(
        primary_shard,
        oldest_log,
        get_osdmap()->get_pg_size(info.pgid.pgid),
        acting,
        up,
        up_primary,
        all_info,
        restrict_to_up_acting,
        &want,
        &want_backfill,
        &want_acting_backfill,
        get_osdmap(),
        pool,
        ss);
    }
  } else {
    calc_ec_acting(
      auth_log_shard,
      get_osdmap()->get_pg_size(info.pgid.pgid),
      acting,
      up,
      all_info,
      restrict_to_up_acting,
      &want,
      &want_backfill,
      &want_acting_backfill,
      ss);
  }
  psdout(10) << ss.str() << dendl;

  if (!recoverable(want)) {
    want_acting.clear();
    return false;
  }

  set<pg_shard_t> want_async_recovery;
  if (HAVE_FEATURE(get_osdmap()->get_up_osd_features(), SERVER_MIMIC)) {
    if (pool.info.is_erasure()) {
      choose_async_recovery_ec(
        all_info, auth_log_shard->second, &want, &want_async_recovery,
        get_osdmap());
    } else {
      choose_async_recovery_replicated(
        all_info, auth_log_shard->second, &want, &want_async_recovery,
        get_osdmap());
    }
  }
  while (want.size() > pool.info.size) {
    // async recovery should have taken out as many osds as it can.
    // if not, then always evict the last peer
    // (will get synchronously recovered later)
    psdout(10) << __func__ << " evicting osd." << want.back()
               << " from oversized want " << want << dendl;
    want.pop_back();
  }
  if (want != acting) {
    psdout(10) << __func__ << " want " << want << " != acting " << acting
               << ", requesting pg_temp change" << dendl;
    want_acting = want;

    if (!cct->_conf->osd_debug_no_acting_change) {
      if (want_acting == up) {
        // There can't be any pending backfill if
        // want is the same as crush map up OSDs.
        ceph_assert(want_backfill.empty());
        vector<int> empty;
        pl->queue_want_pg_temp(empty);
      } else
        pl->queue_want_pg_temp(want);
    }
    return false;
  }

  if (request_pg_temp_change_only)
    return true;
  want_acting.clear();
  acting_recovery_backfill = want_acting_backfill;
  psdout(10) << "acting_recovery_backfill is "
             << acting_recovery_backfill << dendl;
  ceph_assert(
    backfill_targets.empty() ||
    backfill_targets == want_backfill);
  if (backfill_targets.empty()) {
    // Caller is GetInfo
    backfill_targets = want_backfill;
  }
  // Adding !needs_recovery() to let the async_recovery_targets reset after recovery is complete
  ceph_assert(
    async_recovery_targets.empty() ||
    async_recovery_targets == want_async_recovery ||
    !needs_recovery());
  if (async_recovery_targets.empty() || !needs_recovery()) {
    async_recovery_targets = want_async_recovery;
  }
  // Will not change if already set because up would have had to change
  // Verify that nothing in backfill is in stray_set
  for (auto i = want_backfill.begin(); i != want_backfill.end(); ++i) {
    ceph_assert(stray_set.find(*i) == stray_set.end());
  }
  psdout(10) << "choose_acting want=" << want << " backfill_targets="
           << want_backfill << " async_recovery_targets="
           << async_recovery_targets << dendl;
  return true;
}

void PeeringState::log_weirdness()
{
  if (pg_log.get_tail() != info.log_tail)
    pl->get_clog_error() << info.pgid
                           << " info mismatch, log.tail " << pg_log.get_tail()
                           << " != info.log_tail " << info.log_tail;
  if (pg_log.get_head() != info.last_update)
    pl->get_clog_error() << info.pgid
                           << " info mismatch, log.head " << pg_log.get_head()
                           << " != info.last_update " << info.last_update;

  if (!pg_log.get_log().empty()) {
    // sloppy check
    if ((pg_log.get_log().log.begin()->version <= pg_log.get_tail()))
      pl->get_clog_error() << info.pgid
                             << " log bound mismatch, info (tail,head] ("
                             << pg_log.get_tail() << ","
                             << pg_log.get_head() << "]"
                             << " actual ["
                             << pg_log.get_log().log.begin()->version << ","
                             << pg_log.get_log().log.rbegin()->version << "]";
  }

  if (pg_log.get_log().caller_ops.size() > pg_log.get_log().log.size()) {
    pl->get_clog_error() << info.pgid
                           << " caller_ops.size "
                           << pg_log.get_log().caller_ops.size()
                           << " > log size " << pg_log.get_log().log.size();
  }
}

/*
 * Process information from a replica to determine if it could have any
 * objects that i need.
 *
 * TODO: if the missing set becomes very large, this could get expensive.
 * Instead, we probably want to just iterate over our unfound set.
 */
bool PeeringState::search_for_missing(
  const pg_info_t &oinfo, const pg_missing_t &omissing,
  pg_shard_t from,
  PeeringCtxWrapper &ctx)
{
  uint64_t num_unfound_before = missing_loc.num_unfound();
  bool found_missing = missing_loc.add_source_info(
    from, oinfo, omissing, ctx.handle);
  if (found_missing && num_unfound_before != missing_loc.num_unfound())
    pl->publish_stats_to_osd();
  // avoid doing this if the peer is empty.  This is abit of paranoia
  // to avoid doing something rash if add_source_info() above
  // incorrectly decided we found something new. (if the peer has
  // last_update=0'0 that's impossible.)
  if (found_missing &&
      oinfo.last_update != eversion_t()) {
    pg_info_t tinfo(oinfo);
    tinfo.pgid.shard = pg_whoami.shard;
    ctx.send_info(
      from.osd,
      spg_t(info.pgid.pgid, from.shard),
      get_osdmap_epoch(),  // fixme: use lower epoch?
      get_osdmap_epoch(),
      tinfo);
  }
  return found_missing;
}

bool PeeringState::discover_all_missing(
  BufferedRecoveryMessages &rctx)
{
  auto &missing = pg_log.get_missing();
  uint64_t unfound = get_num_unfound();
  bool any = false;  // did we start any queries

  psdout(10) << __func__ << " "
             << missing.num_missing() << " missing, "
             << unfound << " unfound"
             << dendl;

  auto m = might_have_unfound.begin();
  auto mend = might_have_unfound.end();
  for (; m != mend; ++m) {
    pg_shard_t peer(*m);

    if (!get_osdmap()->is_up(peer.osd)) {
      psdout(20) << __func__ << " skipping down osd." << peer << dendl;
      continue;
    }

    if (peer_purged.count(peer)) {
      psdout(20) << __func__ << " skipping purged osd." << peer << dendl;
      continue;
    }

    auto iter = peer_info.find(peer);
    if (iter != peer_info.end() &&
        (iter->second.is_empty() || iter->second.dne())) {
      // ignore empty peers
      continue;
    }

    // If we've requested any of this stuff, the pg_missing_t information
    // should be on its way.
    // TODO: coalsce requested_* into a single data structure
    if (peer_missing.find(peer) != peer_missing.end()) {
      psdout(20) << __func__ << ": osd." << peer
                 << ": we already have pg_missing_t" << dendl;
      continue;
    }
    if (peer_log_requested.find(peer) != peer_log_requested.end()) {
      psdout(20) << __func__ << ": osd." << peer
                 << ": in peer_log_requested" << dendl;
      continue;
    }
    if (peer_missing_requested.find(peer) != peer_missing_requested.end()) {
      psdout(20) << __func__ << ": osd." << peer
                 << ": in peer_missing_requested" << dendl;
      continue;
    }

    // Request missing
    psdout(10) << __func__ << ": osd." << peer << ": requesting pg_missing_t"
               << dendl;
    peer_missing_requested.insert(peer);
    rctx.send_query(
      peer.osd,
      spg_t(info.pgid.pgid, peer.shard),
      pg_query_t(
        pg_query_t::FULLLOG,
        peer.shard, pg_whoami.shard,
        info.history, get_osdmap_epoch()));
    any = true;
  }
  return any;
}

/* Build the might_have_unfound set.
 *
 * This is used by the primary OSD during recovery.
 *
 * This set tracks the OSDs which might have unfound objects that the primary
 * OSD needs. As we receive pg_missing_t from each OSD in might_have_unfound, we
 * will remove the OSD from the set.
 */
void PeeringState::build_might_have_unfound()
{
  ceph_assert(might_have_unfound.empty());
  ceph_assert(is_primary());

  psdout(10) << __func__ << dendl;

  check_past_interval_bounds();

  might_have_unfound = past_intervals.get_might_have_unfound(
    pg_whoami,
    pool.info.is_erasure());

  // include any (stray) peers
  for (auto p = peer_info.begin(); p != peer_info.end(); ++p)
    might_have_unfound.insert(p->first);

  psdout(15) << __func__ << ": built " << might_have_unfound << dendl;
}

void PeeringState::activate(
  ObjectStore::Transaction& t,
  epoch_t activation_epoch,
  PeeringCtxWrapper &ctx)
{
  ceph_assert(!is_peered());

  // twiddle pg state
  state_clear(PG_STATE_DOWN);

  send_notify = false;

  if (is_primary()) {
    // only update primary last_epoch_started if we will go active
    if (acting_set_writeable()) {
      ceph_assert(cct->_conf->osd_find_best_info_ignore_history_les ||
             info.last_epoch_started <= activation_epoch);
      info.last_epoch_started = activation_epoch;
      info.last_interval_started = info.history.same_interval_since;
    }
  } else if (is_acting(pg_whoami)) {
    /* update last_epoch_started on acting replica to whatever the primary sent
     * unless it's smaller (could happen if we are going peered rather than
     * active, see doc/dev/osd_internals/last_epoch_started.rst) */
    if (info.last_epoch_started < activation_epoch) {
      info.last_epoch_started = activation_epoch;
      info.last_interval_started = info.history.same_interval_since;
    }
  }

  auto &missing = pg_log.get_missing();

  min_last_complete_ondisk = eversion_t(0,0);  // we don't know (yet)!
  if (is_primary()) {
    last_update_ondisk = info.last_update;
  }
  last_update_applied = info.last_update;
  last_rollback_info_trimmed_to_applied = pg_log.get_can_rollback_to();

  need_up_thru = false;

  // write pg info, log
  dirty_info = true;
  dirty_big_info = true; // maybe

  pl->schedule_event_on_commit(
    t,
    std::make_shared<PGPeeringEvent>(
      get_osdmap_epoch(),
      get_osdmap_epoch(),
      ActivateCommitted(
        get_osdmap_epoch(),
        activation_epoch)));

  // init complete pointer
  if (missing.num_missing() == 0) {
    psdout(10) << "activate - no missing, moving last_complete " << info.last_complete
             << " -> " << info.last_update << dendl;
    info.last_complete = info.last_update;
    info.stats.stats.sum.num_objects_missing = 0;
    pg_log.reset_recovery_pointers();
  } else {
    psdout(10) << "activate - not complete, " << missing << dendl;
    info.stats.stats.sum.num_objects_missing = missing.num_missing();
    pg_log.activate_not_complete(info);
  }

  log_weirdness();

  if (is_primary()) {
    // initialize snap_trimq
    interval_set<snapid_t> to_trim;
    auto& removed_snaps_queue = get_osdmap()->get_removed_snaps_queue();
    auto p = removed_snaps_queue.find(info.pgid.pgid.pool());
    if (p != removed_snaps_queue.end()) {
      dout(20) << "activate - purged_snaps " << info.purged_snaps
               << " removed_snaps " << p->second
               << dendl;
      for (auto q : p->second) {
        to_trim.insert(q.first, q.second);
      }
    }
    interval_set<snapid_t> purged;
    purged.intersection_of(to_trim, info.purged_snaps);
    to_trim.subtract(purged);

    assert(HAVE_FEATURE(upacting_features, SERVER_OCTOPUS));
    renew_lease(pl->get_mnow());
    // do not schedule until we are actually activated

    // adjust purged_snaps: PG may have been inactive while snaps were pruned
    // from the removed_snaps_queue in the osdmap.  update local purged_snaps
    // reflect only those snaps that we thought were pruned and were still in
    // the queue.
    info.purged_snaps.swap(purged);

    // start up replicas
    if (prior_readable_down_osds.empty()) {
      dout(10) << __func__ << " no prior_readable_down_osds to wait on, clearing ub"
               << dendl;
      clear_prior_readable_until_ub();
    }
    info.history.refresh_prior_readable_until_ub(pl->get_mnow(),
                                                 prior_readable_until_ub);

    ceph_assert(!acting_recovery_backfill.empty());
    for (auto i = acting_recovery_backfill.begin();
         i != acting_recovery_backfill.end();
         ++i) {
      if (*i == pg_whoami) continue;
      pg_shard_t peer = *i;
      ceph_assert(peer_info.count(peer));
      pg_info_t& pi = peer_info[peer];

      psdout(10) << "activate peer osd." << peer << " " << pi << dendl;

      #if defined(WITH_SEASTAR)
      MURef<MOSDPGLog> m;
      #else
      MRef<MOSDPGLog> m;
      #endif
      ceph_assert(peer_missing.count(peer));
      pg_missing_t& pm = peer_missing[peer];

      bool needs_past_intervals = pi.dne();

      // Save num_bytes for backfill reservation request, can't be negative
      peer_bytes[peer] = std::max<int64_t>(0, pi.stats.stats.sum.num_bytes);

      if (pi.last_update == info.last_update) {
        // empty log
        if (!pi.last_backfill.is_max())
          pl->get_clog_info() << info.pgid << " continuing backfill to osd."
                                << peer
                                << " from (" << pi.log_tail << "," << pi.last_update
                                << "] " << pi.last_backfill
                                << " to " << info.last_update;
        if (!pi.is_empty()) {
          psdout(10) << "activate peer osd." << peer
                     << " is up to date, queueing in pending_activators" << dendl;
          ctx.send_info(
            peer.osd,
            spg_t(info.pgid.pgid, peer.shard),
            get_osdmap_epoch(), // fixme: use lower epoch?
            get_osdmap_epoch(),
            info,
            get_lease());
        } else {
          psdout(10) << "activate peer osd." << peer
                     << " is up to date, but sending pg_log anyway" << dendl;
          m = TOPNSPC::make_message<MOSDPGLog>(
            i->shard, pg_whoami.shard,
            get_osdmap_epoch(), info,
            last_peering_reset);
        }
      } else if (
        pg_log.get_tail() > pi.last_update ||
        pi.last_backfill == hobject_t() ||
        (backfill_targets.count(*i) && pi.last_backfill.is_max())) {
        /* ^ This last case covers a situation where a replica is not contiguous
         * with the auth_log, but is contiguous with this replica.  Reshuffling
         * the active set to handle this would be tricky, so instead we just go
         * ahead and backfill it anyway.  This is probably preferrable in any
         * case since the replica in question would have to be significantly
         * behind.
         */
        // backfill
        pl->get_clog_debug() << info.pgid << " starting backfill to osd." << peer
                               << " from (" << pi.log_tail << "," << pi.last_update
                               << "] " << pi.last_backfill
                               << " to " << info.last_update;

        pi.last_update = info.last_update;
        pi.last_complete = info.last_update;
        pi.set_last_backfill(hobject_t());
        pi.last_epoch_started = info.last_epoch_started;
        pi.last_interval_started = info.last_interval_started;
        pi.history = info.history;
        pi.hit_set = info.hit_set;
        pi.stats.stats.clear();
        pi.stats.stats.sum.num_bytes = peer_bytes[peer];

        // initialize peer with our purged_snaps.
        pi.purged_snaps = info.purged_snaps;

        m = TOPNSPC::make_message<MOSDPGLog>(
          i->shard, pg_whoami.shard,
          get_osdmap_epoch(), pi,
          last_peering_reset /* epoch to create pg at */);

        // send some recent log, so that op dup detection works well.
        m->log.copy_up_to(cct, pg_log.get_log(),
                          cct->_conf->osd_max_pg_log_entries);
        m->info.log_tail = m->log.tail;
        pi.log_tail = m->log.tail;  // sigh...

        pm.clear();
      } else {
        // catch up
        ceph_assert(pg_log.get_tail() <= pi.last_update);
        m = TOPNSPC::make_message<MOSDPGLog>(
          i->shard, pg_whoami.shard,
          get_osdmap_epoch(), info,
          last_peering_reset /* epoch to create pg at */);
        // send new stuff to append to replicas log
        m->log.copy_after(cct, pg_log.get_log(), pi.last_update);
      }

      // share past_intervals if we are creating the pg on the replica
      // based on whether our info for that peer was dne() *before*
      // updating pi.history in the backfill block above.
      if (m && needs_past_intervals)
        m->past_intervals = past_intervals;

      // update local version of peer's missing list!
      if (m && pi.last_backfill != hobject_t()) {
        for (auto p = m->log.log.begin(); p != m->log.log.end(); ++p) {
          if (p->soid <= pi.last_backfill &&
              !p->is_error()) {
            if (perform_deletes_during_peering() && p->is_delete()) {
              pm.rm(p->soid, p->version);
            } else {
              pm.add_next_event(*p);
            }
          }
        }
      }

      if (m) {
        dout(10) << "activate peer osd." << peer << " sending " << m->log
                 << dendl;
        m->lease = get_lease();
        pl->send_cluster_message(peer.osd, std::move(m), get_osdmap_epoch());
      }

      // peer now has
      pi.last_update = info.last_update;

      // update our missing
      if (pm.num_missing() == 0) {
        pi.last_complete = pi.last_update;
        psdout(10) << "activate peer osd." << peer << " " << pi
                   << " uptodate" << dendl;
      } else {
        psdout(10) << "activate peer osd." << peer << " " << pi
                   << " missing " << pm << dendl;
      }
    }

    // Set up missing_loc
    set<pg_shard_t> complete_shards;
    for (auto i = acting_recovery_backfill.begin();
         i != acting_recovery_backfill.end();
         ++i) {
      psdout(20) << __func__ << " setting up missing_loc from shard " << *i
                 << " " << dendl;
      if (*i == get_primary()) {
        missing_loc.add_active_missing(missing);
        if (!missing.have_missing())
          complete_shards.insert(*i);
      } else {
        auto peer_missing_entry = peer_missing.find(*i);
        ceph_assert(peer_missing_entry != peer_missing.end());
        missing_loc.add_active_missing(peer_missing_entry->second);
        if (!peer_missing_entry->second.have_missing() &&
            peer_info[*i].last_backfill.is_max())
          complete_shards.insert(*i);
      }
    }

    // If necessary, create might_have_unfound to help us find our unfound objects.
    // NOTE: It's important that we build might_have_unfound before trimming the
    // past intervals.
    might_have_unfound.clear();
    if (needs_recovery()) {
      // If only one shard has missing, we do a trick to add all others as recovery
      // source, this is considered safe since the PGLogs have been merged locally,
      // and covers vast majority of the use cases, like one OSD/host is down for
      // a while for hardware repairing
      if (complete_shards.size() + 1 == acting_recovery_backfill.size()) {
        missing_loc.add_batch_sources_info(complete_shards, ctx.handle);
      } else {
        missing_loc.add_source_info(pg_whoami, info, pg_log.get_missing(),
                                    ctx.handle);
        for (auto i = acting_recovery_backfill.begin();
             i != acting_recovery_backfill.end();
             ++i) {
          if (*i == pg_whoami) continue;
          psdout(10) << __func__ << ": adding " << *i << " as a source" << dendl;
          ceph_assert(peer_missing.count(*i));
          ceph_assert(peer_info.count(*i));
          missing_loc.add_source_info(
            *i,
            peer_info[*i],
            peer_missing[*i],
            ctx.handle);
        }
      }
      for (auto i = peer_missing.begin(); i != peer_missing.end(); ++i) {
        if (is_acting_recovery_backfill(i->first))
          continue;
        ceph_assert(peer_info.count(i->first));
        search_for_missing(
          peer_info[i->first],
          i->second,
          i->first,
          ctx);
      }

      build_might_have_unfound();

      // Always call now so update_calc_stats() will be accurate
      discover_all_missing(ctx.msgs);

    }

    // num_objects_degraded if calculated should reflect this too, unless no
    // missing and we are about to go clean.
    if (get_osdmap()->get_pg_size(info.pgid.pgid) > actingset.size()) {
      state_set(PG_STATE_UNDERSIZED);
    }

    state_set(PG_STATE_ACTIVATING);
    pl->on_activate(std::move(to_trim));
  }
  if (acting_set_writeable()) {
    PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(t)};
    pg_log.roll_forward(rollbacker.get());
  }
}

void PeeringState::share_pg_info()
{
  psdout(10) << "share_pg_info" << dendl;

  info.history.refresh_prior_readable_until_ub(pl->get_mnow(),
                                               prior_readable_until_ub);

  // share new pg_info_t with replicas
  ceph_assert(!acting_recovery_backfill.empty());
  for (auto pg_shard : acting_recovery_backfill) {
    if (pg_shard == pg_whoami) continue;
    if (auto peer = peer_info.find(pg_shard); peer != peer_info.end()) {
      peer->second.last_epoch_started = info.last_epoch_started;
      peer->second.last_interval_started = info.last_interval_started;
      peer->second.history.merge(info.history);
    }
    auto m = TOPNSPC::make_message<MOSDPGInfo2>(spg_t{info.pgid.pgid, pg_shard.shard},
                          info,
                          get_osdmap_epoch(),
                          get_osdmap_epoch(),
                          std::optional<pg_lease_t>{get_lease()},
                          std::nullopt);
    pl->send_cluster_message(pg_shard.osd, std::move(m), get_osdmap_epoch());
  }
}

void PeeringState::merge_log(
  ObjectStore::Transaction& t, pg_info_t &oinfo, pg_log_t&& olog,
  pg_shard_t from)
{
  PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(t)};
  pg_log.merge_log(
    oinfo, std::move(olog), from, info, rollbacker.get(),
    dirty_info, dirty_big_info);
}

void PeeringState::rewind_divergent_log(
  ObjectStore::Transaction& t, eversion_t newhead)
{
  PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(t)};
  pg_log.rewind_divergent_log(
    newhead, info, rollbacker.get(), dirty_info, dirty_big_info);
}


void PeeringState::proc_primary_info(
  ObjectStore::Transaction &t, const pg_info_t &oinfo)
{
  ceph_assert(!is_primary());

  update_history(oinfo.history);
  if (!info.stats.stats_invalid && info.stats.stats.sum.num_scrub_errors) {
    info.stats.stats.sum.num_scrub_errors = 0;
    info.stats.stats.sum.num_shallow_scrub_errors = 0;
    info.stats.stats.sum.num_deep_scrub_errors = 0;
    dirty_info = true;
  }

  if (!(info.purged_snaps == oinfo.purged_snaps)) {
    psdout(10) << __func__ << " updating purged_snaps to "
               << oinfo.purged_snaps
               << dendl;
    info.purged_snaps = oinfo.purged_snaps;
    dirty_info = true;
    dirty_big_info = true;
  }
}

void PeeringState::proc_master_log(
  ObjectStore::Transaction& t, pg_info_t &oinfo,
  pg_log_t&& olog, pg_missing_t&& omissing, pg_shard_t from)
{
  psdout(10) << "proc_master_log for osd." << from << ": "
             << olog << " " << omissing << dendl;
  ceph_assert(!is_peered() && is_primary());

  // merge log into our own log to build master log.  no need to
  // make any adjustments to their missing map; we are taking their
  // log to be authoritative (i.e., their entries are by definitely
  // non-divergent).
  merge_log(t, oinfo, std::move(olog), from);
  peer_info[from] = oinfo;
  psdout(10) << " peer osd." << from << " now " << oinfo
             << " " << omissing << dendl;
  might_have_unfound.insert(from);

  // See doc/dev/osd_internals/last_epoch_started
  if (oinfo.last_epoch_started > info.last_epoch_started) {
    info.last_epoch_started = oinfo.last_epoch_started;
    dirty_info = true;
  }
  if (oinfo.last_interval_started > info.last_interval_started) {
    info.last_interval_started = oinfo.last_interval_started;
    dirty_info = true;
  }
  update_history(oinfo.history);
  ceph_assert(cct->_conf->osd_find_best_info_ignore_history_les ||
         info.last_epoch_started >= info.history.last_epoch_started);

  peer_missing[from].claim(std::move(omissing));
}

void PeeringState::proc_replica_log(
  pg_info_t &oinfo,
  const pg_log_t &olog,
  pg_missing_t&& omissing,
  pg_shard_t from)
{
  psdout(10) << "proc_replica_log for osd." << from << ": "
             << oinfo << " " << olog << " " << omissing << dendl;

  pg_log.proc_replica_log(oinfo, olog, omissing, from);

  peer_info[from] = oinfo;
  psdout(10) << " peer osd." << from << " now "
             << oinfo << " " << omissing << dendl;
  might_have_unfound.insert(from);

  for (auto i = omissing.get_items().begin();
       i != omissing.get_items().end();
       ++i) {
    psdout(20) << " after missing " << i->first
               << " need " << i->second.need
               << " have " << i->second.have << dendl;
  }
  peer_missing[from].claim(std::move(omissing));
}

void PeeringState::fulfill_info(
  pg_shard_t from, const pg_query_t &query,
  pair<pg_shard_t, pg_info_t> &notify_info)
{
  ceph_assert(from == primary);
  ceph_assert(query.type == pg_query_t::INFO);

  // info
  psdout(10) << "sending info" << dendl;
  notify_info = make_pair(from, info);
}

void PeeringState::fulfill_log(
  pg_shard_t from, const pg_query_t &query, epoch_t query_epoch)
{
  psdout(10) << "log request from " << from << dendl;
  ceph_assert(from == primary);
  ceph_assert(query.type != pg_query_t::INFO);

  auto mlog = TOPNSPC::make_message<MOSDPGLog>(
    from.shard, pg_whoami.shard,
    get_osdmap_epoch(),
    info, query_epoch);
  mlog->missing = pg_log.get_missing();

  // primary -> other, when building master log
  if (query.type == pg_query_t::LOG) {
    psdout(10) << " sending info+missing+log since " << query.since
               << dendl;
    if (query.since != eversion_t() && query.since < pg_log.get_tail()) {
      pl->get_clog_error() << info.pgid << " got broken pg_query_t::LOG since "
                             << query.since
                             << " when my log.tail is " << pg_log.get_tail()
                             << ", sending full log instead";
      mlog->log = pg_log.get_log();           // primary should not have requested this!!
    } else
      mlog->log.copy_after(cct, pg_log.get_log(), query.since);
  }
  else if (query.type == pg_query_t::FULLLOG) {
    psdout(10) << " sending info+missing+full log" << dendl;
    mlog->log = pg_log.get_log();
  }

  psdout(10) << " sending " << mlog->log << " " << mlog->missing << dendl;

  pl->send_cluster_message(from.osd, std::move(mlog), get_osdmap_epoch(), true);
}

void PeeringState::fulfill_query(const MQuery& query, PeeringCtxWrapper &rctx)
{
  if (query.query.type == pg_query_t::INFO) {
    pair<pg_shard_t, pg_info_t> notify_info;
    // note this refreshes our prior_readable_until_ub value
    update_history(query.query.history);
    fulfill_info(query.from, query.query, notify_info);
    rctx.send_notify(
      notify_info.first.osd,
      pg_notify_t(
        notify_info.first.shard, pg_whoami.shard,
        query.query_epoch,
        get_osdmap_epoch(),
        notify_info.second,
        past_intervals));
  } else {
    update_history(query.query.history);
    fulfill_log(query.from, query.query, query.query_epoch);
  }
}

void PeeringState::try_mark_clean()
{
  if (actingset.size() == get_osdmap()->get_pg_size(info.pgid.pgid)) {
    state_clear(PG_STATE_FORCED_BACKFILL | PG_STATE_FORCED_RECOVERY);
    state_set(PG_STATE_CLEAN);
    info.history.last_epoch_clean = get_osdmap_epoch();
    info.history.last_interval_clean = info.history.same_interval_since;
    past_intervals.clear();
    dirty_big_info = true;
    dirty_info = true;
  }

  if (!is_active() && is_peered()) {
    if (is_clean()) {
      bool target;
      if (pool.info.is_pending_merge(info.pgid.pgid, &target)) {
        if (target) {
          psdout(10) << "ready to merge (target)" << dendl;
          pl->set_ready_to_merge_target(
            info.last_update,
            info.history.last_epoch_started,
            info.history.last_epoch_clean);
        } else {
          psdout(10) << "ready to merge (source)" << dendl;
          pl->set_ready_to_merge_source(info.last_update);
        }
      }
    } else {
      psdout(10) << "not clean, not ready to merge" << dendl;
      // we should have notified OSD in Active state entry point
    }
  }

  state_clear(PG_STATE_FORCED_RECOVERY | PG_STATE_FORCED_BACKFILL);

  share_pg_info();
  pl->publish_stats_to_osd();
  clear_recovery_state();
}

void PeeringState::split_into(
  pg_t child_pgid, PeeringState *child, unsigned split_bits)
{
  child->update_osdmap_ref(get_osdmap());
  child->pool = pool;

  // Log
  pg_log.split_into(child_pgid, split_bits, &(child->pg_log));
  child->info.last_complete = info.last_complete;

  info.last_update = pg_log.get_head();
  child->info.last_update = child->pg_log.get_head();

  child->info.last_user_version = info.last_user_version;

  info.log_tail = pg_log.get_tail();
  child->info.log_tail = child->pg_log.get_tail();

  // reset last_complete, we might have modified pg_log & missing above
  pg_log.reset_complete_to(&info);
  child->pg_log.reset_complete_to(&child->info);

  // Info
  child->info.history = info.history;
  child->info.history.epoch_created = get_osdmap_epoch();
  child->info.purged_snaps = info.purged_snaps;

  if (info.last_backfill.is_max()) {
    child->info.set_last_backfill(hobject_t::get_max());
  } else {
    // restart backfill on parent and child to be safe.  we could
    // probably do better in the bitwise sort case, but it's more
    // fragile (there may be special work to do on backfill completion
    // in the future).
    info.set_last_backfill(hobject_t());
    child->info.set_last_backfill(hobject_t());
    // restarting backfill implies that the missing set is empty,
    // since it is only used for objects prior to last_backfill
    pg_log.reset_backfill();
    child->pg_log.reset_backfill();
  }

  child->info.stats = info.stats;
  child->info.stats.parent_split_bits = split_bits;
  info.stats.stats_invalid = true;
  child->info.stats.stats_invalid = true;
  child->info.last_epoch_started = info.last_epoch_started;
  child->info.last_interval_started = info.last_interval_started;

  // There can't be recovery/backfill going on now
  int primary, up_primary;
  vector<int> newup, newacting;
  get_osdmap()->pg_to_up_acting_osds(
    child->info.pgid.pgid, &newup, &up_primary, &newacting, &primary);
  child->init_primary_up_acting(
    newup,
    newacting,
    up_primary,
    primary);
  child->role = OSDMap::calc_pg_role(pg_whoami, child->acting);

  // this comparison includes primary rank via pg_shard_t
  if (get_primary() != child->get_primary())
    child->info.history.same_primary_since = get_osdmap_epoch();

  child->info.stats.up = newup;
  child->info.stats.up_primary = up_primary;
  child->info.stats.acting = newacting;
  child->info.stats.acting_primary = primary;
  child->info.stats.mapping_epoch = get_osdmap_epoch();

  // History
  child->past_intervals = past_intervals;

  child->on_new_interval();

  child->send_notify = !child->is_primary();

  child->dirty_info = true;
  child->dirty_big_info = true;
  dirty_info = true;
  dirty_big_info = true;
}

void PeeringState::merge_from(
  map<spg_t,PeeringState *>& sources,
  PeeringCtx &rctx,
  unsigned split_bits,
  const pg_merge_meta_t& last_pg_merge_meta)
{
  bool incomplete = false;
  if (info.last_complete != info.last_update ||
      info.is_incomplete() ||
      info.dne()) {
    psdout(10) << __func__ << " target incomplete" << dendl;
    incomplete = true;
  }
  if (last_pg_merge_meta.source_pgid != pg_t()) {
    if (info.pgid.pgid != last_pg_merge_meta.source_pgid.get_parent()) {
      psdout(10) << __func__ << " target doesn't match expected parent "
                 << last_pg_merge_meta.source_pgid.get_parent()
                 << " of source_pgid " << last_pg_merge_meta.source_pgid
                 << dendl;
      incomplete = true;
    }
    if (info.last_update != last_pg_merge_meta.target_version) {
      psdout(10) << __func__ << " target version doesn't match expected "
               << last_pg_merge_meta.target_version << dendl;
      incomplete = true;
    }
  }

  PGLog::LogEntryHandlerRef handler{pl->get_log_handler(rctx.transaction)};
  pg_log.roll_forward(handler.get());

  info.last_complete = info.last_update;  // to fake out trim()
  pg_log.reset_recovery_pointers();
  pg_log.trim(info.last_update, info);

  vector<PGLog*> log_from;
  for (auto& i : sources) {
    auto& source = i.second;
    if (!source) {
      psdout(10) << __func__ << " source " << i.first << " missing" << dendl;
      incomplete = true;
      continue;
    }
    if (source->info.last_complete != source->info.last_update ||
        source->info.is_incomplete() ||
        source->info.dne()) {
      psdout(10) << __func__ << " source " << source->pg_whoami
                 << " incomplete"
                 << dendl;
      incomplete = true;
    }
    if (last_pg_merge_meta.source_pgid != pg_t()) {
      if (source->info.pgid.pgid != last_pg_merge_meta.source_pgid) {
        dout(10) << __func__ << " source " << source->info.pgid.pgid
                 << " doesn't match expected source pgid "
                 << last_pg_merge_meta.source_pgid << dendl;
        incomplete = true;
      }
      if (source->info.last_update != last_pg_merge_meta.source_version) {
        dout(10) << __func__ << " source version doesn't match expected "
                 << last_pg_merge_meta.target_version << dendl;
        incomplete = true;
      }
    }

    // prepare log
    PGLog::LogEntryHandlerRef handler{
      source->pl->get_log_handler(rctx.transaction)};
    source->pg_log.roll_forward(handler.get());
    source->info.last_complete = source->info.last_update;  // to fake out trim()
    source->pg_log.reset_recovery_pointers();
    source->pg_log.trim(source->info.last_update, source->info);
    log_from.push_back(&source->pg_log);

    // combine stats
    info.stats.add(source->info.stats);

    // pull up last_update
    info.last_update = std::max(info.last_update, source->info.last_update);

    // adopt source's PastIntervals if target has none.  we can do this since
    // pgp_num has been reduced prior to the merge, so the OSD mappings for
    // the PGs are identical.
    if (past_intervals.empty() && !source->past_intervals.empty()) {
      psdout(10) << __func__ << " taking source's past_intervals" << dendl;
      past_intervals = source->past_intervals;
    }
  }

  info.last_complete = info.last_update;
  info.log_tail = info.last_update;
  if (incomplete) {
    info.last_backfill = hobject_t();
  }

  // merge logs
  pg_log.merge_from(log_from, info.last_update);

  // make sure we have a meaningful last_epoch_started/clean (if we were a
  // placeholder)
  if (info.history.epoch_created == 0) {
    // start with (a) source's history, since these PGs *should* have been
    // remapped in concert with each other...
    info.history = sources.begin()->second->info.history;

    // we use the last_epoch_{started,clean} we got from
    // the caller, which are the epochs that were reported by the PGs were
    // found to be ready for merge.
    info.history.last_epoch_clean = last_pg_merge_meta.last_epoch_clean;
    info.history.last_epoch_started = last_pg_merge_meta.last_epoch_started;
    info.last_epoch_started = last_pg_merge_meta.last_epoch_started;
    psdout(10) << __func__
               << " set les/c to " << last_pg_merge_meta.last_epoch_started << "/"
               << last_pg_merge_meta.last_epoch_clean
               << " from pool last_dec_*, source pg history was "
               << sources.begin()->second->info.history
               << dendl;

    // above we have pulled down source's history and we need to check
    // history.epoch_created again to confirm that source is not a placeholder
    // too. (peering requires a sane history.same_interval_since value for any
    // non-newly created pg and below here we know we are basically iterating
    // back a series of past maps to fake a merge process, hence we need to
    // fix history.same_interval_since first so that start_peering_interval()
    // will not complain)
    if (info.history.epoch_created == 0) {
      dout(10) << __func__ << " both merge target and source are placeholders,"
               << " set sis to lec " << info.history.last_epoch_clean
               << dendl;
      info.history.same_interval_since = info.history.last_epoch_clean;
    }

    // if the past_intervals start is later than last_epoch_clean, it
    // implies the source repeered again but the target didn't, or
    // that the source became clean in a later epoch than the target.
    // avoid the discrepancy but adjusting the interval start
    // backwards to match so that check_past_interval_bounds() will
    // not complain.
    auto pib = past_intervals.get_bounds();
    if (info.history.last_epoch_clean < pib.first) {
      psdout(10) << __func__ << " last_epoch_clean "
                 << info.history.last_epoch_clean << " < past_interval start "
                 << pib.first << ", adjusting start backwards" << dendl;
      past_intervals.adjust_start_backwards(info.history.last_epoch_clean);
    }

    // Similarly, if the same_interval_since value is later than
    // last_epoch_clean, the next interval change will result in a
    // past_interval start that is later than last_epoch_clean.  This
    // can happen if we use the pg_history values from the merge
    // source.  Adjust the same_interval_since value backwards if that
    // happens.  (We trust the les and lec values more because they came from
    // the real target, whereas the history value we stole from the source.)
    if (info.history.last_epoch_started < info.history.same_interval_since) {
      psdout(10) << __func__ << " last_epoch_started "
                 << info.history.last_epoch_started << " < same_interval_since "
                 << info.history.same_interval_since
                 << ", adjusting pg_history backwards" << dendl;
      info.history.same_interval_since = info.history.last_epoch_clean;
      // make sure same_{up,primary}_since are <= same_interval_since
      info.history.same_up_since = std::min(
        info.history.same_up_since, info.history.same_interval_since);
      info.history.same_primary_since = std::min(
        info.history.same_primary_since, info.history.same_interval_since);
    }
  }

  dirty_info = true;
  dirty_big_info = true;
}

void PeeringState::start_split_stats(
  const set<spg_t>& childpgs, vector<object_stat_sum_t> *out)
{
  out->resize(childpgs.size() + 1);
  info.stats.stats.sum.split(*out);
}

void PeeringState::finish_split_stats(
  const object_stat_sum_t& stats, ObjectStore::Transaction &t)
{
  info.stats.stats.sum = stats;
  write_if_dirty(t);
}

void PeeringState::update_blocked_by()
{
  // set a max on the number of blocking peers we report. if we go
  // over, report a random subset.  keep the result sorted.
  unsigned keep = std::min<unsigned>(
    blocked_by.size(), cct->_conf->osd_max_pg_blocked_by);
  unsigned skip = blocked_by.size() - keep;
  info.stats.blocked_by.clear();
  info.stats.blocked_by.resize(keep);
  unsigned pos = 0;
  for (auto p = blocked_by.begin(); p != blocked_by.end() && keep > 0; ++p) {
    if (skip > 0 && (rand() % (skip + keep) < skip)) {
      --skip;
    } else {
      info.stats.blocked_by[pos++] = *p;
      --keep;
    }
  }
}

static bool find_shard(const set<pg_shard_t> & pgs, shard_id_t shard)
{
    for (auto&p : pgs)
      if (p.shard == shard)
        return true;
    return false;
}

static pg_shard_t get_another_shard(const set<pg_shard_t> & pgs, pg_shard_t skip, shard_id_t shard)
{
    for (auto&p : pgs) {
      if (p == skip)
        continue;
      if (p.shard == shard)
        return p;
    }
    return pg_shard_t();
}

void PeeringState::update_calc_stats()
{
  info.stats.version = info.last_update;
  info.stats.created = info.history.epoch_created;
  info.stats.last_scrub = info.history.last_scrub;
  info.stats.last_scrub_stamp = info.history.last_scrub_stamp;
  info.stats.last_deep_scrub = info.history.last_deep_scrub;
  info.stats.last_deep_scrub_stamp = info.history.last_deep_scrub_stamp;
  info.stats.last_clean_scrub_stamp = info.history.last_clean_scrub_stamp;
  info.stats.last_epoch_clean = info.history.last_epoch_clean;

  info.stats.log_size = pg_log.get_head().version - pg_log.get_tail().version;
  info.stats.ondisk_log_size = info.stats.log_size;
  info.stats.log_start = pg_log.get_tail();
  info.stats.ondisk_log_start = pg_log.get_tail();
  info.stats.snaptrimq_len = pl->get_snap_trimq_size();

  unsigned num_shards = get_osdmap()->get_pg_size(info.pgid.pgid);

  // In rare case that upset is too large (usually transient), use as target
  // for calculations below.
  unsigned target = std::max(num_shards, (unsigned)upset.size());
  // For undersized actingset may be larger with OSDs out
  unsigned nrep = std::max(actingset.size(), upset.size());
  // calc num_object_copies
  info.stats.stats.calc_copies(std::max(target, nrep));
  info.stats.stats.sum.num_objects_degraded = 0;
  info.stats.stats.sum.num_objects_unfound = 0;
  info.stats.stats.sum.num_objects_misplaced = 0;
  info.stats.avail_no_missing.clear();
  info.stats.object_location_counts.clear();

  // We should never hit this condition, but if end up hitting it,
  // make sure to update num_objects and set PG_STATE_INCONSISTENT.
  if (info.stats.stats.sum.num_objects < 0) {
    psdout(0) << __func__ << " negative num_objects = "
              << info.stats.stats.sum.num_objects << " setting it to 0 "
              << dendl;
    info.stats.stats.sum.num_objects = 0;
    state_set(PG_STATE_INCONSISTENT);
  }

  if ((is_remapped() || is_undersized() || !is_clean()) &&
      (is_peered()|| is_activating())) {
    psdout(20) << __func__ << " actingset " << actingset << " upset "
               << upset << " acting_recovery_backfill " << acting_recovery_backfill << dendl;

    ceph_assert(!acting_recovery_backfill.empty());

    bool estimate = false;

    // NOTE: we only generate degraded, misplaced and unfound
    // values for the summation, not individual stat categories.
    int64_t num_objects = info.stats.stats.sum.num_objects;

    // Objects missing from up nodes, sorted by # objects.
    boost::container::flat_set<pair<int64_t,pg_shard_t>> missing_target_objects;
    // Objects missing from nodes not in up, sort by # objects
    boost::container::flat_set<pair<int64_t,pg_shard_t>> acting_source_objects;

    // Fill missing_target_objects/acting_source_objects

    {
      int64_t missing;

      // Primary first
      missing = pg_log.get_missing().num_missing();
      ceph_assert(acting_recovery_backfill.count(pg_whoami));
      if (upset.count(pg_whoami)) {
        missing_target_objects.emplace(missing, pg_whoami);
      } else {
        acting_source_objects.emplace(missing, pg_whoami);
      }
      info.stats.stats.sum.num_objects_missing_on_primary = missing;
      if (missing == 0)
        info.stats.avail_no_missing.push_back(pg_whoami);
      psdout(20) << __func__ << " shard " << pg_whoami
                 << " primary objects " << num_objects
                 << " missing " << missing
                 << dendl;
    }

    // All other peers
    for (auto& peer : peer_info) {
      // Primary should not be in the peer_info, skip if it is.
      if (peer.first == pg_whoami) continue;
      int64_t missing = 0;
      int64_t peer_num_objects =
        std::max((int64_t)0, peer.second.stats.stats.sum.num_objects);
      // Backfill targets always track num_objects accurately
      // all other peers track missing accurately.
      if (is_backfill_target(peer.first)) {
        missing = std::max((int64_t)0, num_objects - peer_num_objects);
      } else {
        if (peer_missing.count(peer.first)) {
          missing = peer_missing[peer.first].num_missing();
        } else {
          psdout(20) << __func__ << " no peer_missing found for "
                     << peer.first << dendl;
          if (is_recovering()) {
            estimate = true;
          }
          missing = std::max((int64_t)0, num_objects - peer_num_objects);
        }
      }
      if (upset.count(peer.first)) {
        missing_target_objects.emplace(missing, peer.first);
      } else if (actingset.count(peer.first)) {
        acting_source_objects.emplace(missing, peer.first);
      }
      peer.second.stats.stats.sum.num_objects_missing = missing;
      if (missing == 0)
        info.stats.avail_no_missing.push_back(peer.first);
      psdout(20) << __func__ << " shard " << peer.first
                 << " objects " << peer_num_objects
                 << " missing " << missing
                 << dendl;
    }

    // Compute object_location_counts
    for (auto& ml: missing_loc.get_missing_locs()) {
      info.stats.object_location_counts[ml.second]++;
      psdout(30) << __func__ << " " << ml.first << " object_location_counts["
                 << ml.second << "]=" << info.stats.object_location_counts[ml.second]
                 << dendl;
    }
    int64_t not_missing = num_objects - missing_loc.get_missing_locs().size();
    if (not_missing) {
        // During recovery we know upset == actingset and is being populated
        // During backfill we know that all non-missing objects are in the actingset
        info.stats.object_location_counts[actingset] = not_missing;
    }
    psdout(30) << __func__ << " object_location_counts["
               << upset << "]=" << info.stats.object_location_counts[upset]
               << dendl;
    psdout(20) << __func__ << " object_location_counts "
               << info.stats.object_location_counts << dendl;

    // A misplaced object is not stored on the correct OSD
    int64_t misplaced = 0;
    // a degraded objects has fewer replicas or EC shards than the pool specifies.
    int64_t degraded = 0;

    if (is_recovering()) {
      for (auto& sml: missing_loc.get_missing_by_count()) {
        for (auto& ml: sml.second) {
          int missing_shards;
          if (sml.first == shard_id_t::NO_SHARD) {
            psdout(20) << __func__ << " ml " << ml.second
                       << " upset size " << upset.size()
                       << " up " << ml.first.up << dendl;
            missing_shards = (int)upset.size() - ml.first.up;
          } else {
            // Handle shards not even in upset below
            if (!find_shard(upset, sml.first))
              continue;
            missing_shards = std::max(0, 1 - ml.first.up);
            psdout(20) << __func__
                       << " shard " << sml.first
                       << " ml " << ml.second
                       << " missing shards " << missing_shards << dendl;
          }
          int odegraded = ml.second * missing_shards;
          // Copies on other osds but limited to the possible degraded
          int more_osds = std::min(missing_shards, ml.first.other);
          int omisplaced = ml.second * more_osds;
          ceph_assert(omisplaced <= odegraded);
          odegraded -= omisplaced;

          misplaced += omisplaced;
          degraded += odegraded;
        }
      }

      psdout(20) << __func__ << " missing based degraded "
                 << degraded << dendl;
      psdout(20) << __func__ << " missing based misplaced "
                 << misplaced << dendl;

      // Handle undersized case
      if (pool.info.is_replicated()) {
        // Add degraded for missing targets (num_objects missing)
        ceph_assert(target >= upset.size());
        unsigned needed = target - upset.size();
        degraded += num_objects * needed;
      } else {
        for (unsigned i = 0 ; i < num_shards; ++i) {
          shard_id_t shard(i);

          if (!find_shard(upset, shard)) {
            pg_shard_t pgs = get_another_shard(actingset, pg_shard_t(), shard);

            if (pgs != pg_shard_t()) {
              int64_t missing;

              if (pgs == pg_whoami)
                missing = info.stats.stats.sum.num_objects_missing_on_primary;
              else
                missing = peer_info[pgs].stats.stats.sum.num_objects_missing;

              degraded += missing;
              misplaced += std::max((int64_t)0, num_objects - missing);
            } else {
              // No shard anywhere
              degraded += num_objects;
            }
          }
        }
      }
      goto out;
    }

    // Handle undersized case
    if (pool.info.is_replicated()) {
      // Add to missing_target_objects
      ceph_assert(target >= missing_target_objects.size());
      unsigned needed = target - missing_target_objects.size();
      if (needed)
        missing_target_objects.emplace(num_objects * needed, pg_shard_t(pg_shard_t::NO_OSD));
    } else {
      for (unsigned i = 0 ; i < num_shards; ++i) {
        shard_id_t shard(i);
        bool found = false;
        for (const auto& t : missing_target_objects) {
          if (std::get<1>(t).shard == shard) {
            found = true;
            break;
          }
        }
        if (!found)
          missing_target_objects.emplace(num_objects, pg_shard_t(pg_shard_t::NO_OSD,shard));
      }
    }

    for (const auto& item : missing_target_objects)
      psdout(20) << __func__ << " missing shard " << std::get<1>(item)
                 << " missing= " << std::get<0>(item) << dendl;
    for (const auto& item : acting_source_objects)
      psdout(20) << __func__ << " acting shard " << std::get<1>(item)
                 << " missing= " << std::get<0>(item) << dendl;

    // Handle all objects not in missing for remapped
    // or backfill
    for (auto m = missing_target_objects.rbegin();
        m != missing_target_objects.rend(); ++m) {

      int64_t extra_missing = -1;

      if (pool.info.is_replicated()) {
        if (!acting_source_objects.empty()) {
          auto extra_copy = acting_source_objects.begin();
          extra_missing = std::get<0>(*extra_copy);
          acting_source_objects.erase(extra_copy);
        }
      } else {  // Erasure coded
        // Use corresponding shard
        for (const auto& a : acting_source_objects) {
          if (std::get<1>(a).shard == std::get<1>(*m).shard) {
            extra_missing = std::get<0>(a);
            acting_source_objects.erase(a);
            break;
          }
        }
      }

      if (extra_missing >= 0 && std::get<0>(*m) >= extra_missing) {
        // We don't know which of the objects on the target
        // are part of extra_missing so assume are all degraded.
        misplaced += std::get<0>(*m) - extra_missing;
        degraded += extra_missing;
      } else {
        // 1. extra_missing == -1, more targets than sources so degraded
        // 2. extra_missing > std::get<0>(m), so that we know that some extra_missing
        //    previously degraded are now present on the target.
        degraded += std::get<0>(*m);
      }
    }
    // If there are still acting that haven't been accounted for
    // then they are misplaced
    for (const auto& a : acting_source_objects) {
      int64_t extra_misplaced = std::max((int64_t)0, num_objects - std::get<0>(a));
      psdout(20) << __func__ << " extra acting misplaced " << extra_misplaced
                 << dendl;
      misplaced += extra_misplaced;
    }
out:
    // NOTE: Tests use these messages to verify this code
    psdout(20) << __func__ << " degraded " << degraded
               << (estimate ? " (est)": "") << dendl;
    psdout(20) << __func__ << " misplaced " << misplaced
               << (estimate ? " (est)": "")<< dendl;

    info.stats.stats.sum.num_objects_degraded = degraded;
    info.stats.stats.sum.num_objects_unfound = get_num_unfound();
    info.stats.stats.sum.num_objects_misplaced = misplaced;
  }
}

std::optional<pg_stat_t> PeeringState::prepare_stats_for_publish(
  const std::optional<pg_stat_t> &pg_stats_publish,
  const object_stat_collection_t &unstable_stats)
{
  if (info.stats.stats.sum.num_scrub_errors) {
    psdout(10) << __func__ << " inconsistent due to " <<
      info.stats.stats.sum.num_scrub_errors << " scrub errors" << dendl;
    state_set(PG_STATE_INCONSISTENT);
  } else {
    state_clear(PG_STATE_INCONSISTENT);
    state_clear(PG_STATE_FAILED_REPAIR);
  }

  utime_t now = ceph_clock_now();
  if (info.stats.state != state) {
    info.stats.last_change = now;
    // Optimistic estimation, if we just find out an inactive PG,
    // assume it is active till now.
    if (!(state & PG_STATE_ACTIVE) &&
        (info.stats.state & PG_STATE_ACTIVE))
      info.stats.last_active = now;

    if ((state & PG_STATE_ACTIVE) &&
        !(info.stats.state & PG_STATE_ACTIVE))
      info.stats.last_became_active = now;
    if ((state & (PG_STATE_ACTIVE|PG_STATE_PEERED)) &&
        !(info.stats.state & (PG_STATE_ACTIVE|PG_STATE_PEERED)))
      info.stats.last_became_peered = now;
    info.stats.state = state;
  }

  update_calc_stats();
  if (info.stats.stats.sum.num_objects_degraded) {
    state_set(PG_STATE_DEGRADED);
  } else {
    state_clear(PG_STATE_DEGRADED);
  }
  update_blocked_by();

  pg_stat_t pre_publish = info.stats;
  pre_publish.stats.add(unstable_stats);
  utime_t cutoff = now;
  cutoff -= cct->_conf->osd_pg_stat_report_interval_max;

  // share (some of) our purged_snaps via the pg_stats. limit # of intervals
  // because we don't want to make the pg_stat_t structures too expensive.
  unsigned max = cct->_conf->osd_max_snap_prune_intervals_per_epoch;
  unsigned num = 0;
  auto i = info.purged_snaps.begin();
  while (num < max && i != info.purged_snaps.end()) {
    pre_publish.purged_snaps.insert(i.get_start(), i.get_len());
    ++num;
    ++i;
  }
  psdout(20) << __func__ << " reporting purged_snaps "
             << pre_publish.purged_snaps << dendl;

  if (pg_stats_publish && pre_publish == *pg_stats_publish &&
      info.stats.last_fresh > cutoff) {
    psdout(15) << "publish_stats_to_osd " << pg_stats_publish->reported_epoch
               << ": no change since " << info.stats.last_fresh << dendl;
    return std::nullopt;
  } else {
    // update our stat summary and timestamps
    info.stats.reported_epoch = get_osdmap_epoch();
    ++info.stats.reported_seq;

    info.stats.last_fresh = now;

    if (info.stats.state & PG_STATE_CLEAN)
      info.stats.last_clean = now;
    if (info.stats.state & PG_STATE_ACTIVE)
      info.stats.last_active = now;
    if (info.stats.state & (PG_STATE_ACTIVE|PG_STATE_PEERED))
      info.stats.last_peered = now;
    info.stats.last_unstale = now;
    if ((info.stats.state & PG_STATE_DEGRADED) == 0)
      info.stats.last_undegraded = now;
    if ((info.stats.state & PG_STATE_UNDERSIZED) == 0)
      info.stats.last_fullsized = now;

    psdout(15) << "publish_stats_to_osd " << pre_publish.reported_epoch
               << ":" << pre_publish.reported_seq << dendl;
    return std::make_optional(std::move(pre_publish));
  }
}

void PeeringState::init(
  int role,
  const vector<int>& newup, int new_up_primary,
  const vector<int>& newacting, int new_acting_primary,
  const pg_history_t& history,
  const PastIntervals& pi,
  ObjectStore::Transaction &t)
{
  psdout(10) << "init role " << role << " up "
             << newup << " acting " << newacting
             << " history " << history
             << " past_intervals " << pi
             << dendl;

  set_role(role);
  init_primary_up_acting(
    newup,
    newacting,
    new_up_primary,
    new_acting_primary);

  info.history = history;
  past_intervals = pi;

  info.stats.up = up;
  info.stats.up_primary = new_up_primary;
  info.stats.acting = acting;
  info.stats.acting_primary = new_acting_primary;
  info.stats.mapping_epoch = info.history.same_interval_since;

  if (!perform_deletes_during_peering()) {
    pg_log.set_missing_may_contain_deletes();
  }

  on_new_interval();

  dirty_info = true;
  dirty_big_info = true;
  write_if_dirty(t);
}

void PeeringState::dump_peering_state(Formatter *f)
{
  f->dump_string("state", get_pg_state_string());
  f->dump_unsigned("epoch", get_osdmap_epoch());
  f->open_array_section("up");
  for (auto p = up.begin(); p != up.end(); ++p)
    f->dump_unsigned("osd", *p);
  f->close_section();
  f->open_array_section("acting");
  for (auto p = acting.begin(); p != acting.end(); ++p)
    f->dump_unsigned("osd", *p);
  f->close_section();
  if (!backfill_targets.empty()) {
    f->open_array_section("backfill_targets");
    for (auto p = backfill_targets.begin(); p != backfill_targets.end(); ++p)
      f->dump_stream("shard") << *p;
    f->close_section();
  }
  if (!async_recovery_targets.empty()) {
    f->open_array_section("async_recovery_targets");
    for (auto p = async_recovery_targets.begin();
         p != async_recovery_targets.end();
         ++p)
      f->dump_stream("shard") << *p;
    f->close_section();
  }
  if (!acting_recovery_backfill.empty()) {
    f->open_array_section("acting_recovery_backfill");
    for (auto p = acting_recovery_backfill.begin();
         p != acting_recovery_backfill.end();
         ++p)
      f->dump_stream("shard") << *p;
    f->close_section();
  }
  f->open_object_section("info");
  update_calc_stats();
  info.dump(f);
  f->close_section();

  f->open_array_section("peer_info");
  for (auto p = peer_info.begin(); p != peer_info.end(); ++p) {
    f->open_object_section("info");
    f->dump_stream("peer") << p->first;
    p->second.dump(f);
    f->close_section();
  }
  f->close_section();
}

void PeeringState::update_stats(
  std::function<bool(pg_history_t &, pg_stat_t &)> f,
  ObjectStore::Transaction *t) {
  if (f(info.history, info.stats)) {
    pl->publish_stats_to_osd();
  }
  pl->reschedule_scrub();

  if (t) {
    dirty_info = true;
    write_if_dirty(*t);
  }
}

void PeeringState::update_stats_wo_resched(
  std::function<void(pg_history_t &, pg_stat_t &)> f)
{
  f(info.history, info.stats);
}

bool PeeringState::append_log_entries_update_missing(
  const mempool::osd_pglog::list<pg_log_entry_t> &entries,
  ObjectStore::Transaction &t, std::optional<eversion_t> trim_to,
  std::optional<eversion_t> roll_forward_to)
{
  ceph_assert(!entries.empty());
  ceph_assert(entries.begin()->version > info.last_update);

  PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(t)};
  bool invalidate_stats =
    pg_log.append_new_log_entries(
      info.last_backfill,
      entries,
      rollbacker.get());

  if (roll_forward_to && entries.rbegin()->soid > info.last_backfill) {
    pg_log.roll_forward(rollbacker.get());
  }
  if (roll_forward_to && *roll_forward_to > pg_log.get_can_rollback_to()) {
    pg_log.roll_forward_to(*roll_forward_to, rollbacker.get());
    last_rollback_info_trimmed_to_applied = *roll_forward_to;
  }

  info.last_update = pg_log.get_head();

  if (pg_log.get_missing().num_missing() == 0) {
    // advance last_complete since nothing else is missing!
    info.last_complete = info.last_update;
  }
  info.stats.stats_invalid = info.stats.stats_invalid || invalidate_stats;

  psdout(20) << __func__ << " trim_to bool = " << bool(trim_to)
             << " trim_to = " << (trim_to ? *trim_to : eversion_t()) << dendl;
  if (trim_to)
    pg_log.trim(*trim_to, info);
  dirty_info = true;
  write_if_dirty(t);
  return invalidate_stats;
}

void PeeringState::merge_new_log_entries(
  const mempool::osd_pglog::list<pg_log_entry_t> &entries,
  ObjectStore::Transaction &t,
  std::optional<eversion_t> trim_to,
  std::optional<eversion_t> roll_forward_to)
{
  psdout(10) << __func__ << " " << entries << dendl;
  ceph_assert(is_primary());

  bool rebuild_missing = append_log_entries_update_missing(entries, t, trim_to, roll_forward_to);
  for (auto i = acting_recovery_backfill.begin();
       i != acting_recovery_backfill.end();
       ++i) {
    pg_shard_t peer(*i);
    if (peer == pg_whoami) continue;
    ceph_assert(peer_missing.count(peer));
    ceph_assert(peer_info.count(peer));
    pg_missing_t& pmissing(peer_missing[peer]);
    psdout(20) << __func__ << " peer_missing for " << peer
               << " = " << pmissing << dendl;
    pg_info_t& pinfo(peer_info[peer]);
    bool invalidate_stats = PGLog::append_log_entries_update_missing(
      pinfo.last_backfill,
      entries,
      true,
      NULL,
      pmissing,
      NULL,
      dpp);
    pinfo.last_update = info.last_update;
    pinfo.stats.stats_invalid = pinfo.stats.stats_invalid || invalidate_stats;
    rebuild_missing = rebuild_missing || invalidate_stats;
  }

  if (!rebuild_missing) {
    return;
  }

  for (auto &&i: entries) {
    missing_loc.rebuild(
      i.soid,
      pg_whoami,
      acting_recovery_backfill,
      info,
      pg_log.get_missing(),
      peer_missing,
      peer_info);
  }
}

void PeeringState::add_log_entry(const pg_log_entry_t& e, bool applied)
{
  // raise last_complete only if we were previously up to date
  if (info.last_complete == info.last_update)
    info.last_complete = e.version;

  // raise last_update.
  ceph_assert(e.version > info.last_update);
  info.last_update = e.version;

  // raise user_version, if it increased (it may have not get bumped
  // by all logged updates)
  if (e.user_version > info.last_user_version)
    info.last_user_version = e.user_version;

  // log mutation
  pg_log.add(e, applied);
  psdout(10) << "add_log_entry " << e << dendl;
}


void PeeringState::append_log(
  vector<pg_log_entry_t>&& logv,
  eversion_t trim_to,
  eversion_t roll_forward_to,
  eversion_t mlcod,
  ObjectStore::Transaction &t,
  bool transaction_applied,
  bool async)
{
  /* The primary has sent an info updating the history, but it may not
   * have arrived yet.  We want to make sure that we cannot remember this
   * write without remembering that it happened in an interval which went
   * active in epoch history.last_epoch_started.
   */
  if (info.last_epoch_started != info.history.last_epoch_started) {
    info.history.last_epoch_started = info.last_epoch_started;
  }
  if (info.last_interval_started != info.history.last_interval_started) {
    info.history.last_interval_started = info.last_interval_started;
  }
  psdout(10) << "append_log " << pg_log.get_log() << " " << logv << dendl;

  PGLog::LogEntryHandlerRef handler{pl->get_log_handler(t)};
  if (!transaction_applied) {
     /* We must be a backfill or async recovery peer, so it's ok if we apply
      * out-of-turn since we won't be considered when
      * determining a min possible last_update.
      *
      * We skip_rollforward() here, which advances the crt, without
      * doing an actual rollforward. This avoids cleaning up entries
      * from the backend and we do not end up in a situation, where the
      * object is deleted before we can _merge_object_divergent_entries().
      */
    pg_log.skip_rollforward();
  }

  for (auto p = logv.begin(); p != logv.end(); ++p) {
    add_log_entry(*p, transaction_applied);

    /* We don't want to leave the rollforward artifacts around
     * here past last_backfill.  It's ok for the same reason as
     * above */
    if (transaction_applied &&
        p->soid > info.last_backfill) {
      pg_log.roll_forward(handler.get());
    }
  }
  if (transaction_applied && roll_forward_to > pg_log.get_can_rollback_to()) {
    pg_log.roll_forward_to(
      roll_forward_to,
      handler.get());
    last_rollback_info_trimmed_to_applied = roll_forward_to;
  }

  psdout(10) << __func__ << " approx pg log length =  "
             << pg_log.get_log().approx_size() << dendl;
  psdout(10) << __func__ << " transaction_applied = "
             << transaction_applied << dendl;
  if (!transaction_applied || async)
    psdout(10) << __func__ << " " << pg_whoami
               << " is async_recovery or backfill target" << dendl;
  pg_log.trim(trim_to, info, transaction_applied, async);

  // update the local pg, pg log
  dirty_info = true;
  write_if_dirty(t);

  if (!is_primary())
    min_last_complete_ondisk = mlcod;
}

void PeeringState::recover_got(
  const hobject_t &oid, eversion_t v,
  bool is_delete,
  ObjectStore::Transaction &t)
{
  if (v > pg_log.get_can_rollback_to()) {
    /* This can only happen during a repair, and even then, it would
     * be one heck of a race.  If we are repairing the object, the
     * write in question must be fully committed, so it's not valid
     * to roll it back anyway (and we'll be rolled forward shortly
     * anyway) */
    PGLog::LogEntryHandlerRef handler{pl->get_log_handler(t)};
    pg_log.roll_forward_to(v, handler.get());
  }

  psdout(10) << "got missing " << oid << " v " << v << dendl;
  pg_log.recover_got(oid, v, info);
  if (pg_log.get_log().log.empty()) {
    psdout(10) << "last_complete now " << info.last_complete
               << " while log is empty" << dendl;
  } else if (pg_log.get_log().complete_to != pg_log.get_log().log.end()) {
    psdout(10) << "last_complete now " << info.last_complete
               << " log.complete_to " << pg_log.get_log().complete_to->version
               << dendl;
  } else {
    psdout(10) << "last_complete now " << info.last_complete
               << " log.complete_to at end" << dendl;
    //below is not true in the repair case.
    //assert(missing.num_missing() == 0);  // otherwise, complete_to was wrong.
    ceph_assert(info.last_complete == info.last_update);
  }

  if (is_primary()) {
    ceph_assert(missing_loc.needs_recovery(oid));
    if (!is_delete)
      missing_loc.add_location(oid, pg_whoami);
  }

  // update pg
  dirty_info = true;
  write_if_dirty(t);
}

void PeeringState::update_backfill_progress(
  const hobject_t &updated_backfill,
  const pg_stat_t &updated_stats,
  bool preserve_local_num_bytes,
  ObjectStore::Transaction &t) {
  info.set_last_backfill(updated_backfill);
  if (preserve_local_num_bytes) {
    psdout(25) << __func__ << " primary " << updated_stats.stats.sum.num_bytes
               << " local " << info.stats.stats.sum.num_bytes << dendl;
    int64_t bytes = info.stats.stats.sum.num_bytes;
    info.stats = updated_stats;
    info.stats.stats.sum.num_bytes = bytes;
  } else {
    psdout(20) << __func__ << " final " << updated_stats.stats.sum.num_bytes
               << " replaces local " << info.stats.stats.sum.num_bytes << dendl;
    info.stats = updated_stats;
  }

  dirty_info = true;
  write_if_dirty(t);
}

void PeeringState::adjust_purged_snaps(
  std::function<void(interval_set<snapid_t> &snaps)> f) {
  f(info.purged_snaps);
  dirty_info = true;
  dirty_big_info = true;
}

void PeeringState::on_peer_recover(
  pg_shard_t peer,
  const hobject_t &soid,
  const eversion_t &version)
{
  pl->publish_stats_to_osd();
  // done!
  peer_missing[peer].got(soid, version);
  missing_loc.add_location(soid, peer);
}

void PeeringState::begin_peer_recover(
  pg_shard_t peer,
  const hobject_t soid)
{
  peer_missing[peer].revise_have(soid, eversion_t());
}

void PeeringState::force_object_missing(
  const set<pg_shard_t> &peers,
  const hobject_t &soid,
  eversion_t version)
{
  for (auto &&peer : peers) {
    if (peer != primary) {
      peer_missing[peer].add(soid, version, eversion_t(), false);
    } else {
      pg_log.missing_add(soid, version, eversion_t());
      pg_log.reset_complete_to(&info);
      pg_log.set_last_requested(0);
    }
  }

  missing_loc.rebuild(
    soid,
    pg_whoami,
    acting_recovery_backfill,
    info,
    pg_log.get_missing(),
    peer_missing,
    peer_info);
}

void PeeringState::pre_submit_op(
  const hobject_t &hoid,
  const vector<pg_log_entry_t>& logv,
  eversion_t at_version)
{
  if (at_version > eversion_t()) {
    for (auto &&i : get_acting_recovery_backfill()) {
      if (i == primary) continue;
      pg_info_t &pinfo = peer_info[i];
      // keep peer_info up to date
      if (pinfo.last_complete == pinfo.last_update)
        pinfo.last_complete = at_version;
      pinfo.last_update = at_version;
    }
  }

  bool requires_missing_loc = false;
  for (auto &&i : get_async_recovery_targets()) {
    if (i == primary || !get_peer_missing(i).is_missing(hoid))
      continue;
    requires_missing_loc = true;
    for (auto &&entry: logv) {
      peer_missing[i].add_next_event(entry);
    }
  }

  if (requires_missing_loc) {
    for (auto &&entry: logv) {
      psdout(30) << __func__ << " missing_loc before: "
                 << missing_loc.get_locations(entry.soid) << dendl;
      missing_loc.add_missing(entry.soid, entry.version,
                              eversion_t(), entry.is_delete());
      // clear out missing_loc
      missing_loc.clear_location(entry.soid);
      for (auto &i: get_actingset()) {
        if (!get_peer_missing(i).is_missing(entry.soid))
          missing_loc.add_location(entry.soid, i);
      }
      psdout(30) << __func__ << " missing_loc after: "
                 << missing_loc.get_locations(entry.soid) << dendl;
    }
  }
}

void PeeringState::recovery_committed_to(eversion_t version)
{
  psdout(10) << __func__ << " version " << version
             << " now ondisk" << dendl;
  last_complete_ondisk = version;

  if (last_complete_ondisk == info.last_update) {
    if (!is_primary()) {
      // Either we are a replica or backfill target.
      // we are fully up to date.  tell the primary!
      pl->send_cluster_message(
        get_primary().osd,
        TOPNSPC::make_message<MOSDPGTrim>(
          get_osdmap_epoch(),
          spg_t(info.pgid.pgid, primary.shard),
          last_complete_ondisk),
        get_osdmap_epoch());
    } else {
      calc_min_last_complete_ondisk();
    }
  }
}

void PeeringState::complete_write(eversion_t v, eversion_t lc)
{
  last_update_ondisk = v;
  last_complete_ondisk = lc;
  calc_min_last_complete_ondisk();
}

void PeeringState::calc_trim_to()
{
  size_t target = pl->get_target_pg_log_entries();

  eversion_t limit = std::min(
    min_last_complete_ondisk,
    pg_log.get_can_rollback_to());
  if (limit != eversion_t() &&
      limit != pg_trim_to &&
      pg_log.get_log().approx_size() > target) {
    size_t num_to_trim = std::min(pg_log.get_log().approx_size() - target,
                             cct->_conf->osd_pg_log_trim_max);
    if (num_to_trim < cct->_conf->osd_pg_log_trim_min &&
        cct->_conf->osd_pg_log_trim_max >= cct->_conf->osd_pg_log_trim_min) {
      return;
    }
    auto it = pg_log.get_log().log.begin();
    eversion_t new_trim_to;
    for (size_t i = 0; i < num_to_trim; ++i) {
      new_trim_to = it->version;
      ++it;
      if (new_trim_to > limit) {
        new_trim_to = limit;
        psdout(10) << "calc_trim_to trimming to min_last_complete_ondisk" << dendl;
        break;
      }
    }
    psdout(10) << "calc_trim_to " << pg_trim_to << " -> " << new_trim_to << dendl;
    pg_trim_to = new_trim_to;
    assert(pg_trim_to <= pg_log.get_head());
    assert(pg_trim_to <= min_last_complete_ondisk);
  }
}

void PeeringState::calc_trim_to_aggressive()
{
  size_t target = pl->get_target_pg_log_entries();

  // limit pg log trimming up to the can_rollback_to value
  eversion_t limit = std::min({
    pg_log.get_head(),
    pg_log.get_can_rollback_to(),
    last_update_ondisk});
  psdout(10) << __func__ << " limit = " << limit << dendl;

  if (limit != eversion_t() &&
      limit != pg_trim_to &&
      pg_log.get_log().approx_size() > target) {
    psdout(10) << __func__ << " approx pg log length =  "
             << pg_log.get_log().approx_size() << dendl;
    uint64_t num_to_trim = std::min<uint64_t>(pg_log.get_log().approx_size() - target,
                                              cct->_conf->osd_pg_log_trim_max);
    psdout(10) << __func__ << " num_to_trim =  " << num_to_trim << dendl;
    if (num_to_trim < cct->_conf->osd_pg_log_trim_min &&
        cct->_conf->osd_pg_log_trim_max >= cct->_conf->osd_pg_log_trim_min) {
      return;
    }
    auto it = pg_log.get_log().log.begin(); // oldest log entry
    auto rit = pg_log.get_log().log.rbegin();
    eversion_t by_n_to_keep; // start from tail
    eversion_t by_n_to_trim = eversion_t::max(); // start from head
    for (size_t i = 0; it != pg_log.get_log().log.end(); ++it, ++rit) {
      i++;
      if (i > target && by_n_to_keep == eversion_t()) {
        by_n_to_keep = rit->version;
      }
      if (i >= num_to_trim && by_n_to_trim == eversion_t::max()) {
        by_n_to_trim = it->version;
      }
      if (by_n_to_keep != eversion_t() &&
          by_n_to_trim != eversion_t::max()) {
        break;
      }
    }

    if (by_n_to_keep == eversion_t()) {
      return;
    }

    pg_trim_to = std::min({by_n_to_keep, by_n_to_trim, limit});
    psdout(10) << __func__ << " pg_trim_to now " << pg_trim_to << dendl;
    ceph_assert(pg_trim_to <= pg_log.get_head());
  }
}

void PeeringState::apply_op_stats(
  const hobject_t &soid,
  const object_stat_sum_t &delta_stats)
{
  info.stats.stats.add(delta_stats);
  info.stats.stats.floor(0);

  for (auto i = get_backfill_targets().begin();
       i != get_backfill_targets().end();
       ++i) {
    pg_shard_t bt = *i;
    pg_info_t& pinfo = peer_info[bt];
    if (soid <= pinfo.last_backfill)
      pinfo.stats.stats.add(delta_stats);
  }
}

void PeeringState::update_complete_backfill_object_stats(
  const hobject_t &hoid,
  const pg_stat_t &stats)
{
  for (auto &&bt: get_backfill_targets()) {
    pg_info_t& pinfo = peer_info[bt];
    //Add stats to all peers that were missing object
    if (hoid > pinfo.last_backfill)
      pinfo.stats.add(stats);
  }
}

void PeeringState::update_peer_last_backfill(
  pg_shard_t peer,
  const hobject_t &new_last_backfill)
{
  pg_info_t &pinfo = peer_info[peer];
  pinfo.last_backfill = new_last_backfill;
  if (new_last_backfill.is_max()) {
    /* pinfo.stats might be wrong if we did log-based recovery on the
     * backfilled portion in addition to continuing backfill.
     */
    pinfo.stats = info.stats;
  }
}

void PeeringState::set_revert_with_targets(
  const hobject_t &soid,
  const set<pg_shard_t> &good_peers)
{
  for (auto &&peer: good_peers) {
    missing_loc.add_location(soid, peer);
  }
}

void PeeringState::prepare_backfill_for_missing(
  const hobject_t &soid,
  const eversion_t &version,
  const vector<pg_shard_t> &targets) {
  for (auto &&peer: targets) {
    peer_missing[peer].add(soid, version, eversion_t(), false);
  }
}

void PeeringState::update_hset(const pg_hit_set_history_t &hset_history)
{
  info.hit_set = hset_history;
}

/*------------ Peering State Machine----------------*/
#undef dout_prefix
#define dout_prefix (context< PeeringMachine >().dpp->gen_prefix(*_dout) \
                    << "state<" << get_state_name() << ">: ")
#undef psdout
#define psdout(x) ldout(context< PeeringMachine >().cct, x)

#define DECLARE_LOCALS                                  \
  PeeringState *ps = context< PeeringMachine >().state; \
  std::ignore = ps;                                     \
  PeeringListener *pl = context< PeeringMachine >().pl; \
  std::ignore = pl


/*------Crashed-------*/
PeeringState::Crashed::Crashed(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Crashed")
{
  context< PeeringMachine >().log_enter(state_name);
  ceph_abort_msg("we got a bad state machine event");
}


/*------Initial-------*/
PeeringState::Initial::Initial(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Initial")
{
  context< PeeringMachine >().log_enter(state_name);
}

boost::statechart::result PeeringState::Initial::react(const MNotifyRec& notify)
{
  DECLARE_LOCALS;
  ps->proc_replica_info(
    notify.from, notify.notify.info, notify.notify.epoch_sent);
  ps->set_last_peering_reset();
  return transit< Primary >();
}

boost::statechart::result PeeringState::Initial::react(const MInfoRec& i)
{
  DECLARE_LOCALS;
  ceph_assert(!ps->is_primary());
  post_event(i);
  return transit< Stray >();
}

boost::statechart::result PeeringState::Initial::react(const MLogRec& i)
{
  DECLARE_LOCALS;
  ceph_assert(!ps->is_primary());
  post_event(i);
  return transit< Stray >();
}

void PeeringState::Initial::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_initial_latency, dur);
}

/*------Started-------*/
PeeringState::Started::Started(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started")
{
  context< PeeringMachine >().log_enter(state_name);
}

boost::statechart::result
PeeringState::Started::react(const IntervalFlush&)
{
  psdout(10) << "Ending blocked outgoing recovery messages" << dendl;
  context< PeeringMachine >().state->end_block_outgoing();
  return discard_event();
}

boost::statechart::result PeeringState::Started::react(const AdvMap& advmap)
{
  DECLARE_LOCALS;
  psdout(10) << "Started advmap" << dendl;
  ps->check_full_transition(advmap.lastmap, advmap.osdmap);
  if (ps->should_restart_peering(
        advmap.up_primary,
        advmap.acting_primary,
        advmap.newup,
        advmap.newacting,
        advmap.lastmap,
        advmap.osdmap)) {
    psdout(10) << "should_restart_peering, transitioning to Reset"
                       << dendl;
    post_event(advmap);
    return transit< Reset >();
  }
  ps->remove_down_peer_info(advmap.osdmap);
  return discard_event();
}

boost::statechart::result PeeringState::Started::react(const QueryState& q)
{
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;
  q.f->close_section();
  return discard_event();
}

boost::statechart::result PeeringState::Started::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "Started");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::Started::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_started_latency, dur);
  ps->state_clear(PG_STATE_WAIT | PG_STATE_LAGGY);
}

/*--------Reset---------*/
PeeringState::Reset::Reset(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Reset")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;

  ps->flushes_in_progress = 0;
  ps->set_last_peering_reset();
  ps->log_weirdness();
}

boost::statechart::result
PeeringState::Reset::react(const IntervalFlush&)
{
  psdout(10) << "Ending blocked outgoing recovery messages" << dendl;
  context< PeeringMachine >().state->end_block_outgoing();
  return discard_event();
}

boost::statechart::result PeeringState::Reset::react(const AdvMap& advmap)
{
  DECLARE_LOCALS;
  psdout(10) << "Reset advmap" << dendl;

  ps->check_full_transition(advmap.lastmap, advmap.osdmap);

  if (ps->should_restart_peering(
        advmap.up_primary,
        advmap.acting_primary,
        advmap.newup,
        advmap.newacting,
        advmap.lastmap,
        advmap.osdmap)) {
    psdout(10) << "should restart peering, calling start_peering_interval again"
                       << dendl;
    ps->start_peering_interval(
      advmap.lastmap,
      advmap.newup, advmap.up_primary,
      advmap.newacting, advmap.acting_primary,
      context< PeeringMachine >().get_cur_transaction());
  }
  ps->remove_down_peer_info(advmap.osdmap);
  ps->check_past_interval_bounds();
  return discard_event();
}

boost::statechart::result PeeringState::Reset::react(const ActMap&)
{
  DECLARE_LOCALS;
  if (ps->should_send_notify() && ps->get_primary().osd >= 0) {
    ps->info.history.refresh_prior_readable_until_ub(
      pl->get_mnow(),
      ps->prior_readable_until_ub);
    context< PeeringMachine >().send_notify(
      ps->get_primary().osd,
      pg_notify_t(
        ps->get_primary().shard, ps->pg_whoami.shard,
        ps->get_osdmap_epoch(),
        ps->get_osdmap_epoch(),
        ps->info,
        ps->past_intervals));
  }

  ps->update_heartbeat_peers();

  return transit< Started >();
}

boost::statechart::result PeeringState::Reset::react(const QueryState& q)
{
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;
  q.f->close_section();
  return discard_event();
}

boost::statechart::result PeeringState::Reset::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "Reset");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::Reset::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_reset_latency, dur);
}

/*-------Start---------*/
PeeringState::Start::Start(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Start")
{
  context< PeeringMachine >().log_enter(state_name);

  DECLARE_LOCALS;
  if (ps->is_primary()) {
    psdout(1) << "transitioning to Primary" << dendl;
    post_event(MakePrimary());
  } else { //is_stray
    psdout(1) << "transitioning to Stray" << dendl;
    post_event(MakeStray());
  }
}

void PeeringState::Start::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_start_latency, dur);
}

/*---------Primary--------*/
PeeringState::Primary::Primary(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;
  ceph_assert(ps->want_acting.empty());

  // set CREATING bit until we have peered for the first time.
  if (ps->info.history.last_epoch_started == 0) {
    ps->state_set(PG_STATE_CREATING);
    // use the history timestamp, which ultimately comes from the
    // monitor in the create case.
    utime_t t = ps->info.history.last_scrub_stamp;
    ps->info.stats.last_fresh = t;
    ps->info.stats.last_active = t;
    ps->info.stats.last_change = t;
    ps->info.stats.last_peered = t;
    ps->info.stats.last_clean = t;
    ps->info.stats.last_unstale = t;
    ps->info.stats.last_undegraded = t;
    ps->info.stats.last_fullsized = t;
    ps->info.stats.last_scrub_stamp = t;
    ps->info.stats.last_deep_scrub_stamp = t;
    ps->info.stats.last_clean_scrub_stamp = t;
  }
}

boost::statechart::result PeeringState::Primary::react(const MNotifyRec& notevt)
{
  DECLARE_LOCALS;
  psdout(7) << "handle_pg_notify from osd." << notevt.from << dendl;
  ps->proc_replica_info(
    notevt.from, notevt.notify.info, notevt.notify.epoch_sent);
  return discard_event();
}

boost::statechart::result PeeringState::Primary::react(const ActMap&)
{
  DECLARE_LOCALS;
  psdout(7) << "handle ActMap primary" << dendl;
  pl->publish_stats_to_osd();
  return discard_event();
}

boost::statechart::result PeeringState::Primary::react(
  const SetForceRecovery&)
{
  DECLARE_LOCALS;
  ps->set_force_recovery(true);
  return discard_event();
}

boost::statechart::result PeeringState::Primary::react(
  const UnsetForceRecovery&)
{
  DECLARE_LOCALS;
  ps->set_force_recovery(false);
  return discard_event();
}

boost::statechart::result PeeringState::Primary::react(
  const RequestScrub& evt)
{
  DECLARE_LOCALS;
  if (ps->is_primary()) {
    pl->scrub_requested(evt.deep, evt.repair);
    psdout(10) << "marking for scrub" << dendl;
  }
  return discard_event();
}

boost::statechart::result PeeringState::Primary::react(
  const SetForceBackfill&)
{
  DECLARE_LOCALS;
  ps->set_force_backfill(true);
  return discard_event();
}

boost::statechart::result PeeringState::Primary::react(
  const UnsetForceBackfill&)
{
  DECLARE_LOCALS;
  ps->set_force_backfill(false);
  return discard_event();
}

void PeeringState::Primary::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  ps->want_acting.clear();
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_primary_latency, dur);
  pl->clear_primary_state();
  ps->state_clear(PG_STATE_CREATING);
}

/*---------Peering--------*/
PeeringState::Peering::Peering(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Peering"),
    history_les_bound(false)
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;

  ceph_assert(!ps->is_peered());
  ceph_assert(!ps->is_peering());
  ceph_assert(ps->is_primary());
  ps->state_set(PG_STATE_PEERING);
}

boost::statechart::result PeeringState::Peering::react(const AdvMap& advmap)
{
  DECLARE_LOCALS;
  psdout(10) << "Peering advmap" << dendl;
  if (prior_set.affected_by_map(*(advmap.osdmap), ps->dpp)) {
    psdout(1) << "Peering, affected_by_map, going to Reset" << dendl;
    post_event(advmap);
    return transit< Reset >();
  }

  ps->adjust_need_up_thru(advmap.osdmap);
  ps->check_prior_readable_down_osds(advmap.osdmap);

  return forward_event();
}

boost::statechart::result PeeringState::Peering::react(const QueryState& q)
{
  DECLARE_LOCALS;

  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;

  q.f->open_array_section("past_intervals");
  ps->past_intervals.dump(q.f);
  q.f->close_section();

  q.f->open_array_section("probing_osds");
  for (auto p = prior_set.probe.begin(); p != prior_set.probe.end(); ++p)
    q.f->dump_stream("osd") << *p;
  q.f->close_section();

  if (prior_set.pg_down)
    q.f->dump_string("blocked", "peering is blocked due to down osds");

  q.f->open_array_section("down_osds_we_would_probe");
  for (auto p = prior_set.down.begin(); p != prior_set.down.end(); ++p)
    q.f->dump_int("osd", *p);
  q.f->close_section();

  q.f->open_array_section("peering_blocked_by");
  for (auto p = prior_set.blocked_by.begin();
       p != prior_set.blocked_by.end();
       ++p) {
    q.f->open_object_section("osd");
    q.f->dump_int("osd", p->first);
    q.f->dump_int("current_lost_at", p->second);
    q.f->dump_string("comment", "starting or marking this osd lost may let us proceed");
    q.f->close_section();
  }
  q.f->close_section();

  if (history_les_bound) {
    q.f->open_array_section("peering_blocked_by_detail");
    q.f->open_object_section("item");
    q.f->dump_string("detail","peering_blocked_by_history_les_bound");
    q.f->close_section();
    q.f->close_section();
  }

  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::Peering::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "Peering");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::Peering::exit()
{

  DECLARE_LOCALS;
  psdout(10) << "Leaving Peering" << dendl;
  context< PeeringMachine >().log_exit(state_name, enter_time);
  ps->state_clear(PG_STATE_PEERING);
  pl->clear_probe_targets();

  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_peering_latency, dur);
}


/*------Backfilling-------*/
PeeringState::Backfilling::Backfilling(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/Backfilling")
{
  context< PeeringMachine >().log_enter(state_name);


  DECLARE_LOCALS;
  ps->backfill_reserved = true;
  pl->on_backfill_reserved();
  ps->state_clear(PG_STATE_BACKFILL_TOOFULL);
  ps->state_clear(PG_STATE_BACKFILL_WAIT);
  ps->state_set(PG_STATE_BACKFILLING);
  pl->publish_stats_to_osd();
}

void PeeringState::Backfilling::backfill_release_reservations()
{
  DECLARE_LOCALS;
  pl->cancel_local_background_io_reservation();
  for (auto it = ps->backfill_targets.begin();
       it != ps->backfill_targets.end();
       ++it) {
    ceph_assert(*it != ps->pg_whoami);
    pl->send_cluster_message(
      it->osd,
      TOPNSPC::make_message<MBackfillReserve>(
        MBackfillReserve::RELEASE,
        spg_t(ps->info.pgid.pgid, it->shard),
        ps->get_osdmap_epoch()),
      ps->get_osdmap_epoch());
  }
}

void PeeringState::Backfilling::cancel_backfill()
{
  DECLARE_LOCALS;
  backfill_release_reservations();
  pl->on_backfill_canceled();
}

boost::statechart::result
PeeringState::Backfilling::react(const Backfilled &c)
{
  backfill_release_reservations();
  return transit<Recovered>();
}

boost::statechart::result
PeeringState::Backfilling::react(const DeferBackfill &c)
{
  DECLARE_LOCALS;

  psdout(10) << "defer backfill, retry delay " << c.delay << dendl;
  ps->state_set(PG_STATE_BACKFILL_WAIT);
  ps->state_clear(PG_STATE_BACKFILLING);
  cancel_backfill();

  pl->schedule_event_after(
    std::make_shared<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      RequestBackfill()),
    c.delay);
  return transit<NotBackfilling>();
}

boost::statechart::result
PeeringState::Backfilling::react(const UnfoundBackfill &c)
{
  DECLARE_LOCALS;
  psdout(10) << "backfill has unfound, can't continue" << dendl;
  ps->state_set(PG_STATE_BACKFILL_UNFOUND);
  ps->state_clear(PG_STATE_BACKFILLING);
  cancel_backfill();
  return transit<NotBackfilling>();
}

boost::statechart::result
PeeringState::Backfilling::react(const RemoteReservationRevokedTooFull &)
{
  DECLARE_LOCALS;

  ps->state_set(PG_STATE_BACKFILL_TOOFULL);
  ps->state_clear(PG_STATE_BACKFILLING);
  cancel_backfill();

  pl->schedule_event_after(
    std::make_shared<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      RequestBackfill()),
    ps->cct->_conf->osd_backfill_retry_interval);

  return transit<NotBackfilling>();
}

boost::statechart::result
PeeringState::Backfilling::react(const RemoteReservationRevoked &)
{
  DECLARE_LOCALS;
  ps->state_set(PG_STATE_BACKFILL_WAIT);
  cancel_backfill();
  if (ps->needs_backfill()) {
    return transit<WaitLocalBackfillReserved>();
  } else {
    // raced with MOSDPGBackfill::OP_BACKFILL_FINISH, ignore
    return discard_event();
  }
}

void PeeringState::Backfilling::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  ps->backfill_reserved = false;
  ps->state_clear(PG_STATE_BACKFILLING);
  ps->state_clear(PG_STATE_FORCED_BACKFILL | PG_STATE_FORCED_RECOVERY);
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_backfilling_latency, dur);
}

/*--WaitRemoteBackfillReserved--*/

PeeringState::WaitRemoteBackfillReserved::WaitRemoteBackfillReserved(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/WaitRemoteBackfillReserved"),
    backfill_osd_it(context< Active >().remote_shards_to_reserve_backfill.begin())
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;

  ps->state_set(PG_STATE_BACKFILL_WAIT);
  pl->publish_stats_to_osd();
  post_event(RemoteBackfillReserved());
}

boost::statechart::result
PeeringState::WaitRemoteBackfillReserved::react(const RemoteBackfillReserved &evt)
{
  DECLARE_LOCALS;

  int64_t num_bytes = ps->info.stats.stats.sum.num_bytes;
  psdout(10) << __func__ << " num_bytes " << num_bytes << dendl;
  if (backfill_osd_it !=
      context< Active >().remote_shards_to_reserve_backfill.end()) {
    // The primary never backfills itself
    ceph_assert(*backfill_osd_it != ps->pg_whoami);
    pl->send_cluster_message(
      backfill_osd_it->osd,
      TOPNSPC::make_message<MBackfillReserve>(
        MBackfillReserve::REQUEST,
        spg_t(context< PeeringMachine >().spgid.pgid, backfill_osd_it->shard),
        ps->get_osdmap_epoch(),
        ps->get_backfill_priority(),
        num_bytes,
        ps->peer_bytes[*backfill_osd_it]),
      ps->get_osdmap_epoch());
    ++backfill_osd_it;
  } else {
    ps->peer_bytes.clear();
    post_event(AllBackfillsReserved());
  }
  return discard_event();
}

void PeeringState::WaitRemoteBackfillReserved::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;

  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_waitremotebackfillreserved_latency, dur);
}

void PeeringState::WaitRemoteBackfillReserved::retry()
{
  DECLARE_LOCALS;
  pl->cancel_local_background_io_reservation();

  // Send CANCEL to all previously acquired reservations
  set<pg_shard_t>::const_iterator it, begin, end;
  begin = context< Active >().remote_shards_to_reserve_backfill.begin();
  end = context< Active >().remote_shards_to_reserve_backfill.end();
  ceph_assert(begin != end);
  for (it = begin; it != backfill_osd_it; ++it) {
    // The primary never backfills itself
    ceph_assert(*it != ps->pg_whoami);
    pl->send_cluster_message(
      it->osd,
      TOPNSPC::make_message<MBackfillReserve>(
        MBackfillReserve::RELEASE,
        spg_t(context< PeeringMachine >().spgid.pgid, it->shard),
        ps->get_osdmap_epoch()),
      ps->get_osdmap_epoch());
  }

  ps->state_clear(PG_STATE_BACKFILL_WAIT);
  pl->publish_stats_to_osd();

  pl->schedule_event_after(
    std::make_shared<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      RequestBackfill()),
    ps->cct->_conf->osd_backfill_retry_interval);
}

boost::statechart::result
PeeringState::WaitRemoteBackfillReserved::react(const RemoteReservationRejectedTooFull &evt)
{
  DECLARE_LOCALS;
  ps->state_set(PG_STATE_BACKFILL_TOOFULL);
  retry();
  return transit<NotBackfilling>();
}

boost::statechart::result
PeeringState::WaitRemoteBackfillReserved::react(const RemoteReservationRevoked &evt)
{
  retry();
  return transit<NotBackfilling>();
}

/*--WaitLocalBackfillReserved--*/
PeeringState::WaitLocalBackfillReserved::WaitLocalBackfillReserved(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/WaitLocalBackfillReserved")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;

  ps->state_set(PG_STATE_BACKFILL_WAIT);
  pl->request_local_background_io_reservation(
    ps->get_backfill_priority(),
    std::make_unique<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      LocalBackfillReserved()),
    std::make_unique<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      DeferBackfill(0.0)));
  pl->publish_stats_to_osd();
}

void PeeringState::WaitLocalBackfillReserved::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_waitlocalbackfillreserved_latency, dur);
}

/*----NotBackfilling------*/
PeeringState::NotBackfilling::NotBackfilling(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/NotBackfilling")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;
  ps->state_clear(PG_STATE_REPAIR);
  pl->publish_stats_to_osd();
}

boost::statechart::result PeeringState::NotBackfilling::react(const QueryUnfound& q)
{
  DECLARE_LOCALS;

  ps->query_unfound(q.f, "NotBackfilling");
  return discard_event();
}

boost::statechart::result
PeeringState::NotBackfilling::react(const RemoteBackfillReserved &evt)
{
  return discard_event();
}

boost::statechart::result
PeeringState::NotBackfilling::react(const RemoteReservationRejectedTooFull &evt)
{
  return discard_event();
}

void PeeringState::NotBackfilling::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;
  ps->state_clear(PG_STATE_BACKFILL_UNFOUND);
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_notbackfilling_latency, dur);
}

/*----NotRecovering------*/
PeeringState::NotRecovering::NotRecovering(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/NotRecovering")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;
  ps->state_clear(PG_STATE_REPAIR);
  pl->publish_stats_to_osd();
}

boost::statechart::result PeeringState::NotRecovering::react(const QueryUnfound& q)
{
  DECLARE_LOCALS;

  ps->query_unfound(q.f, "NotRecovering");
  return discard_event();
}

void PeeringState::NotRecovering::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;
  ps->state_clear(PG_STATE_RECOVERY_UNFOUND);
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_notrecovering_latency, dur);
}

/*---RepNotRecovering----*/
PeeringState::RepNotRecovering::RepNotRecovering(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/ReplicaActive/RepNotRecovering")
{
  context< PeeringMachine >().log_enter(state_name);
}

boost::statechart::result
PeeringState::RepNotRecovering::react(const RejectTooFullRemoteReservation &evt)
{
  DECLARE_LOCALS;
  ps->reject_reservation();
  post_event(RemoteReservationRejectedTooFull());
  return discard_event();
}

void PeeringState::RepNotRecovering::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_repnotrecovering_latency, dur);
}

/*---RepWaitRecoveryReserved--*/
PeeringState::RepWaitRecoveryReserved::RepWaitRecoveryReserved(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/ReplicaActive/RepWaitRecoveryReserved")
{
  context< PeeringMachine >().log_enter(state_name);
}

boost::statechart::result
PeeringState::RepWaitRecoveryReserved::react(const RemoteRecoveryReserved &evt)
{
  DECLARE_LOCALS;
  pl->send_cluster_message(
    ps->primary.osd,
    TOPNSPC::make_message<MRecoveryReserve>(
      MRecoveryReserve::GRANT,
      spg_t(ps->info.pgid.pgid, ps->primary.shard),
      ps->get_osdmap_epoch()),
    ps->get_osdmap_epoch());
  return transit<RepRecovering>();
}

boost::statechart::result
PeeringState::RepWaitRecoveryReserved::react(
  const RemoteReservationCanceled &evt)
{
  DECLARE_LOCALS;
  pl->unreserve_recovery_space();

  pl->cancel_remote_recovery_reservation();
  return transit<RepNotRecovering>();
}

void PeeringState::RepWaitRecoveryReserved::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_repwaitrecoveryreserved_latency, dur);
}

/*-RepWaitBackfillReserved*/
PeeringState::RepWaitBackfillReserved::RepWaitBackfillReserved(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/ReplicaActive/RepWaitBackfillReserved")
{
  context< PeeringMachine >().log_enter(state_name);
}

boost::statechart::result
PeeringState::RepNotRecovering::react(const RequestBackfillPrio &evt)
{

  DECLARE_LOCALS;

  if (!pl->try_reserve_recovery_space(
        evt.primary_num_bytes, evt.local_num_bytes)) {
    post_event(RejectTooFullRemoteReservation());
  } else {
    PGPeeringEventURef preempt;
    if (HAVE_FEATURE(ps->upacting_features, RECOVERY_RESERVATION_2)) {
      // older peers will interpret preemption as TOOFULL
      preempt = std::make_unique<PGPeeringEvent>(
        pl->get_osdmap_epoch(),
        pl->get_osdmap_epoch(),
        RemoteBackfillPreempted());
    }
    pl->request_remote_recovery_reservation(
      evt.priority,
      std::make_unique<PGPeeringEvent>(
        pl->get_osdmap_epoch(),
        pl->get_osdmap_epoch(),
        RemoteBackfillReserved()),
      std::move(preempt));
  }
  return transit<RepWaitBackfillReserved>();
}

boost::statechart::result
PeeringState::RepNotRecovering::react(const RequestRecoveryPrio &evt)
{
  DECLARE_LOCALS;

  // fall back to a local reckoning of priority of primary doesn't pass one
  // (pre-mimic compat)
  int prio = evt.priority ? evt.priority : ps->get_recovery_priority();

  PGPeeringEventURef preempt;
  if (HAVE_FEATURE(ps->upacting_features, RECOVERY_RESERVATION_2)) {
    // older peers can't handle this
    preempt = std::make_unique<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      RemoteRecoveryPreempted());
  }

  pl->request_remote_recovery_reservation(
    prio,
    std::make_unique<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      RemoteRecoveryReserved()),
    std::move(preempt));
  return transit<RepWaitRecoveryReserved>();
}

void PeeringState::RepWaitBackfillReserved::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_repwaitbackfillreserved_latency, dur);
}

boost::statechart::result
PeeringState::RepWaitBackfillReserved::react(const RemoteBackfillReserved &evt)
{
  DECLARE_LOCALS;


  pl->send_cluster_message(
      ps->primary.osd,
      TOPNSPC::make_message<MBackfillReserve>(
        MBackfillReserve::GRANT,
        spg_t(ps->info.pgid.pgid, ps->primary.shard),
        ps->get_osdmap_epoch()),
      ps->get_osdmap_epoch());
  return transit<RepRecovering>();
}

boost::statechart::result
PeeringState::RepWaitBackfillReserved::react(
  const RejectTooFullRemoteReservation &evt)
{
  DECLARE_LOCALS;
  ps->reject_reservation();
  post_event(RemoteReservationRejectedTooFull());
  return discard_event();
}

boost::statechart::result
PeeringState::RepWaitBackfillReserved::react(
  const RemoteReservationRejectedTooFull &evt)
{
  DECLARE_LOCALS;
  pl->unreserve_recovery_space();

  pl->cancel_remote_recovery_reservation();
  return transit<RepNotRecovering>();
}

boost::statechart::result
PeeringState::RepWaitBackfillReserved::react(
  const RemoteReservationCanceled &evt)
{
  DECLARE_LOCALS;
  pl->unreserve_recovery_space();

  pl->cancel_remote_recovery_reservation();
  return transit<RepNotRecovering>();
}

/*---RepRecovering-------*/
PeeringState::RepRecovering::RepRecovering(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/ReplicaActive/RepRecovering")
{
  context< PeeringMachine >().log_enter(state_name);
}

boost::statechart::result
PeeringState::RepRecovering::react(const RemoteRecoveryPreempted &)
{
  DECLARE_LOCALS;


  pl->unreserve_recovery_space();
  pl->send_cluster_message(
    ps->primary.osd,
    TOPNSPC::make_message<MRecoveryReserve>(
      MRecoveryReserve::REVOKE,
      spg_t(ps->info.pgid.pgid, ps->primary.shard),
      ps->get_osdmap_epoch()),
    ps->get_osdmap_epoch());
  return discard_event();
}

boost::statechart::result
PeeringState::RepRecovering::react(const BackfillTooFull &)
{
  DECLARE_LOCALS;


  pl->unreserve_recovery_space();
  pl->send_cluster_message(
    ps->primary.osd,
    TOPNSPC::make_message<MBackfillReserve>(
      MBackfillReserve::REVOKE_TOOFULL,
      spg_t(ps->info.pgid.pgid, ps->primary.shard),
      ps->get_osdmap_epoch()),
    ps->get_osdmap_epoch());
  return discard_event();
}

boost::statechart::result
PeeringState::RepRecovering::react(const RemoteBackfillPreempted &)
{
  DECLARE_LOCALS;


  pl->unreserve_recovery_space();
  pl->send_cluster_message(
    ps->primary.osd,
    TOPNSPC::make_message<MBackfillReserve>(
      MBackfillReserve::REVOKE,
      spg_t(ps->info.pgid.pgid, ps->primary.shard),
      ps->get_osdmap_epoch()),
    ps->get_osdmap_epoch());
  return discard_event();
}

void PeeringState::RepRecovering::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  pl->unreserve_recovery_space();

  pl->cancel_remote_recovery_reservation();
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_reprecovering_latency, dur);
}

/*------Activating--------*/
PeeringState::Activating::Activating(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/Activating")
{
  context< PeeringMachine >().log_enter(state_name);
}

void PeeringState::Activating::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_activating_latency, dur);
}

PeeringState::WaitLocalRecoveryReserved::WaitLocalRecoveryReserved(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/WaitLocalRecoveryReserved")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;

  // Make sure all nodes that part of the recovery aren't full
  if (!ps->cct->_conf->osd_debug_skip_full_check_in_recovery &&
      ps->get_osdmap()->check_full(ps->acting_recovery_backfill)) {
    post_event(RecoveryTooFull());
    return;
  }

  ps->state_clear(PG_STATE_RECOVERY_TOOFULL);
  ps->state_set(PG_STATE_RECOVERY_WAIT);
  pl->request_local_background_io_reservation(
    ps->get_recovery_priority(),
    std::make_unique<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      LocalRecoveryReserved()),
    std::make_unique<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      DeferRecovery(0.0)));
  pl->publish_stats_to_osd();
}

boost::statechart::result
PeeringState::WaitLocalRecoveryReserved::react(const RecoveryTooFull &evt)
{
  DECLARE_LOCALS;
  ps->state_set(PG_STATE_RECOVERY_TOOFULL);
  pl->schedule_event_after(
    std::make_shared<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      DoRecovery()),
    ps->cct->_conf->osd_recovery_retry_interval);
  return transit<NotRecovering>();
}

void PeeringState::WaitLocalRecoveryReserved::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_waitlocalrecoveryreserved_latency, dur);
}

PeeringState::WaitRemoteRecoveryReserved::WaitRemoteRecoveryReserved(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/WaitRemoteRecoveryReserved"),
    remote_recovery_reservation_it(context< Active >().remote_shards_to_reserve_recovery.begin())
{
  context< PeeringMachine >().log_enter(state_name);
  post_event(RemoteRecoveryReserved());
}

boost::statechart::result
PeeringState::WaitRemoteRecoveryReserved::react(const RemoteRecoveryReserved &evt) {
  DECLARE_LOCALS;

  if (remote_recovery_reservation_it !=
      context< Active >().remote_shards_to_reserve_recovery.end()) {
    ceph_assert(*remote_recovery_reservation_it != ps->pg_whoami);
    pl->send_cluster_message(
      remote_recovery_reservation_it->osd,
      TOPNSPC::make_message<MRecoveryReserve>(
        MRecoveryReserve::REQUEST,
        spg_t(context< PeeringMachine >().spgid.pgid,
              remote_recovery_reservation_it->shard),
        ps->get_osdmap_epoch(),
        ps->get_recovery_priority()),
      ps->get_osdmap_epoch());
    ++remote_recovery_reservation_it;
  } else {
    post_event(AllRemotesReserved());
  }
  return discard_event();
}

void PeeringState::WaitRemoteRecoveryReserved::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_waitremoterecoveryreserved_latency, dur);
}

PeeringState::Recovering::Recovering(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/Recovering")
{
  context< PeeringMachine >().log_enter(state_name);

  DECLARE_LOCALS;
  ps->state_clear(PG_STATE_RECOVERY_WAIT);
  ps->state_clear(PG_STATE_RECOVERY_TOOFULL);
  ps->state_set(PG_STATE_RECOVERING);
  pl->on_recovery_reserved();
  ceph_assert(!ps->state_test(PG_STATE_ACTIVATING));
  pl->publish_stats_to_osd();
}

void PeeringState::Recovering::release_reservations(bool cancel)
{
  DECLARE_LOCALS;
  ceph_assert(cancel || !ps->pg_log.get_missing().have_missing());

  // release remote reservations
  for (auto i = context< Active >().remote_shards_to_reserve_recovery.begin();
       i != context< Active >().remote_shards_to_reserve_recovery.end();
       ++i) {
    if (*i == ps->pg_whoami) // skip myself
      continue;
    pl->send_cluster_message(
      i->osd,
      TOPNSPC::make_message<MRecoveryReserve>(
        MRecoveryReserve::RELEASE,
        spg_t(ps->info.pgid.pgid, i->shard),
        ps->get_osdmap_epoch()),
      ps->get_osdmap_epoch());
  }
}

boost::statechart::result
PeeringState::Recovering::react(const AllReplicasRecovered &evt)
{
  DECLARE_LOCALS;
  ps->state_clear(PG_STATE_FORCED_RECOVERY);
  release_reservations();
  pl->cancel_local_background_io_reservation();
  return transit<Recovered>();
}

boost::statechart::result
PeeringState::Recovering::react(const RequestBackfill &evt)
{
  DECLARE_LOCALS;

  release_reservations();

  ps->state_clear(PG_STATE_FORCED_RECOVERY);
  pl->cancel_local_background_io_reservation();
  pl->publish_stats_to_osd();
  // transit any async_recovery_targets back into acting
  // so pg won't have to stay undersized for long
  // as backfill might take a long time to complete..
  if (!ps->async_recovery_targets.empty()) {
    pg_shard_t auth_log_shard;
    bool history_les_bound = false;
    // FIXME: Uh-oh we have to check this return value; choose_acting can fail!
    ps->choose_acting(auth_log_shard, true, &history_les_bound);
  }
  return transit<WaitLocalBackfillReserved>();
}

boost::statechart::result
PeeringState::Recovering::react(const DeferRecovery &evt)
{
  DECLARE_LOCALS;
  if (!ps->state_test(PG_STATE_RECOVERING)) {
    // we may have finished recovery and have an AllReplicasRecovered
    // event queued to move us to the next state.
    psdout(10) << "got defer recovery but not recovering" << dendl;
    return discard_event();
  }
  psdout(10) << "defer recovery, retry delay " << evt.delay << dendl;
  ps->state_set(PG_STATE_RECOVERY_WAIT);
  pl->cancel_local_background_io_reservation();
  release_reservations(true);
  pl->schedule_event_after(
    std::make_shared<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      DoRecovery()),
    evt.delay);
  return transit<NotRecovering>();
}

boost::statechart::result
PeeringState::Recovering::react(const UnfoundRecovery &evt)
{
  DECLARE_LOCALS;
  psdout(10) << "recovery has unfound, can't continue" << dendl;
  ps->state_set(PG_STATE_RECOVERY_UNFOUND);
  pl->cancel_local_background_io_reservation();
  release_reservations(true);
  return transit<NotRecovering>();
}

void PeeringState::Recovering::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  ps->state_clear(PG_STATE_RECOVERING);
  pl->get_peering_perf().tinc(rs_recovering_latency, dur);
}

PeeringState::Recovered::Recovered(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/Recovered")
{
  pg_shard_t auth_log_shard;

  context< PeeringMachine >().log_enter(state_name);

  DECLARE_LOCALS;

  ceph_assert(!ps->needs_recovery());

  // if we finished backfill, all acting are active; recheck if
  // DEGRADED | UNDERSIZED is appropriate.
  ceph_assert(!ps->acting_recovery_backfill.empty());
  if (ps->get_osdmap()->get_pg_size(context< PeeringMachine >().spgid.pgid) <=
      ps->acting_recovery_backfill.size()) {
    ps->state_clear(PG_STATE_FORCED_BACKFILL | PG_STATE_FORCED_RECOVERY);
    pl->publish_stats_to_osd();
  }

  // adjust acting set?  (e.g. because backfill completed...)
  bool history_les_bound = false;
  if (ps->acting != ps->up && !ps->choose_acting(auth_log_shard,
                                                 true, &history_les_bound)) {
    ceph_assert(ps->want_acting.size());
  } else if (!ps->async_recovery_targets.empty()) {
    // FIXME: Uh-oh we have to check this return value; choose_acting can fail!
    ps->choose_acting(auth_log_shard, true, &history_les_bound);
  }

  if (context< Active >().all_replicas_activated  &&
      ps->async_recovery_targets.empty())
    post_event(GoClean());
}

void PeeringState::Recovered::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;

  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_recovered_latency, dur);
}

PeeringState::Clean::Clean(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active/Clean")
{
  context< PeeringMachine >().log_enter(state_name);

  DECLARE_LOCALS;

  if (ps->info.last_complete != ps->info.last_update) {
    ceph_abort();
  }


  ps->try_mark_clean();

  context< PeeringMachine >().get_cur_transaction().register_on_commit(
    pl->on_clean());
}

void PeeringState::Clean::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;
  ps->state_clear(PG_STATE_CLEAN);
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_clean_latency, dur);
}

template <typename T>
set<pg_shard_t> unique_osd_shard_set(const pg_shard_t & skip, const T &in)
{
  set<int> osds_found;
  set<pg_shard_t> out;
  for (auto i = in.begin(); i != in.end(); ++i) {
    if (*i != skip && !osds_found.count(i->osd)) {
      osds_found.insert(i->osd);
      out.insert(*i);
    }
  }
  return out;
}

/*---------Active---------*/
PeeringState::Active::Active(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Active"),
    remote_shards_to_reserve_recovery(
      unique_osd_shard_set(
        context< PeeringMachine >().state->pg_whoami,
        context< PeeringMachine >().state->acting_recovery_backfill)),
    remote_shards_to_reserve_backfill(
      unique_osd_shard_set(
        context< PeeringMachine >().state->pg_whoami,
        context< PeeringMachine >().state->backfill_targets)),
    all_replicas_activated(false)
{
  context< PeeringMachine >().log_enter(state_name);


  DECLARE_LOCALS;

  ceph_assert(!ps->backfill_reserved);
  ceph_assert(ps->is_primary());
  psdout(10) << "In Active, about to call activate" << dendl;
  ps->start_flush(context< PeeringMachine >().get_cur_transaction());
  ps->activate(context< PeeringMachine >().get_cur_transaction(),
               ps->get_osdmap_epoch(),
               context< PeeringMachine >().get_recovery_ctx());

  // everyone has to commit/ack before we are truly active
  ps->blocked_by.clear();
  for (auto p = ps->acting_recovery_backfill.begin();
       p != ps->acting_recovery_backfill.end();
       ++p) {
    if (p->shard != ps->pg_whoami.shard) {
      ps->blocked_by.insert(p->shard);
    }
  }
  pl->publish_stats_to_osd();
  psdout(10) << "Activate Finished" << dendl;
}

boost::statechart::result PeeringState::Active::react(const AdvMap& advmap)
{
  DECLARE_LOCALS;

  if (ps->should_restart_peering(
        advmap.up_primary,
        advmap.acting_primary,
        advmap.newup,
        advmap.newacting,
        advmap.lastmap,
        advmap.osdmap)) {
    psdout(10) << "Active advmap interval change, fast return" << dendl;
    return forward_event();
  }
  psdout(10) << "Active advmap" << dendl;
  bool need_publish = false;

  pl->on_active_advmap(advmap.osdmap);
  if (ps->dirty_big_info) {
    // share updated purged_snaps to mgr/mon so that we (a) stop reporting
    // purged snaps and (b) perhaps share more snaps that we have purged
    // but didn't fit in pg_stat_t.
    need_publish = true;
    ps->share_pg_info();
  }

  bool need_acting_change = false;
  for (size_t i = 0; i < ps->want_acting.size(); i++) {
    int osd = ps->want_acting[i];
    if (!advmap.osdmap->is_up(osd)) {
      pg_shard_t osd_with_shard(osd, shard_id_t(i));
      if (!ps->is_acting(osd_with_shard) && !ps->is_up(osd_with_shard)) {
        psdout(10) << "Active stray osd." << osd << " in want_acting is down"
                   << dendl;
        need_acting_change = true;
      }
    }
  }
  if (need_acting_change) {
     psdout(10) << "Active need acting change, call choose_acting again"
                << dendl;
    // possibly because we re-add some strays into the acting set and
    // some of them then go down in a subsequent map before we could see
    // the map changing the pg temp.
    // call choose_acting again to clear them out.
    // note that we leave restrict_to_up_acting to false in order to
    // not overkill any chosen stray that is still alive.
    pg_shard_t auth_log_shard;
    bool history_les_bound = false;
    ps->remove_down_peer_info(advmap.osdmap);
    ps->choose_acting(auth_log_shard, false, &history_les_bound, true);
  }

  /* Check for changes in pool size (if the acting set changed as a result,
   * this does not matter) */
  if (advmap.lastmap->get_pg_size(ps->info.pgid.pgid) !=
      ps->get_osdmap()->get_pg_size(ps->info.pgid.pgid)) {
    if (ps->get_osdmap()->get_pg_size(ps->info.pgid.pgid) <=
        ps->actingset.size()) {
      ps->state_clear(PG_STATE_UNDERSIZED);
    } else {
      ps->state_set(PG_STATE_UNDERSIZED);
    }
    // degraded changes will be detected by call from publish_stats_to_osd()
    need_publish = true;
  }

  // if we haven't reported our PG stats in a long time, do so now.
  if (ps->info.stats.reported_epoch + ps->cct->_conf->osd_pg_stat_report_interval_max < advmap.osdmap->get_epoch()) {
    psdout(20) << "reporting stats to osd after " << (advmap.osdmap->get_epoch() - ps->info.stats.reported_epoch)
                       << " epochs" << dendl;
    need_publish = true;
  }

  if (need_publish)
    pl->publish_stats_to_osd();

  if (ps->check_prior_readable_down_osds(advmap.osdmap)) {
    pl->recheck_readable();
  }

  return forward_event();
}

boost::statechart::result PeeringState::Active::react(const ActMap&)
{
  DECLARE_LOCALS;
  psdout(10) << "Active: handling ActMap" << dendl;
  ceph_assert(ps->is_primary());

  pl->on_active_actmap();

  if (ps->have_unfound()) {
    // object may have become unfound
    ps->discover_all_missing(context<PeeringMachine>().get_recovery_ctx().msgs);
  }

  uint64_t unfound = ps->missing_loc.num_unfound();
  if (unfound > 0 &&
      ps->all_unfound_are_queried_or_lost(ps->get_osdmap())) {
    if (ps->cct->_conf->osd_auto_mark_unfound_lost) {
      pl->get_clog_error() << context< PeeringMachine >().spgid.pgid << " has " << unfound
                            << " objects unfound and apparently lost, would automatically "
                            << "mark these objects lost but this feature is not yet implemented "
                            << "(osd_auto_mark_unfound_lost)";
    } else
      pl->get_clog_error() << context< PeeringMachine >().spgid.pgid << " has "
                             << unfound << " objects unfound and apparently lost";
  }

  return forward_event();
}

boost::statechart::result PeeringState::Active::react(const MNotifyRec& notevt)
{

  DECLARE_LOCALS;
  ceph_assert(ps->is_primary());
  if (ps->peer_info.count(notevt.from)) {
    psdout(10) << "Active: got notify from " << notevt.from
                       << ", already have info from that osd, ignoring"
                       << dendl;
  } else if (ps->peer_purged.count(notevt.from)) {
    psdout(10) << "Active: got notify from " << notevt.from
                       << ", already purged that peer, ignoring"
                       << dendl;
  } else {
    psdout(10) << "Active: got notify from " << notevt.from
                       << ", calling proc_replica_info and discover_all_missing"
                       << dendl;
    ps->proc_replica_info(
      notevt.from, notevt.notify.info, notevt.notify.epoch_sent);
    if (ps->have_unfound() || (ps->is_degraded() && ps->might_have_unfound.count(notevt.from))) {
      ps->discover_all_missing(
        context<PeeringMachine>().get_recovery_ctx().msgs);
    }
    // check if it is a previous down acting member that's coming back.
    // if so, request pg_temp change to trigger a new interval transition
    pg_shard_t auth_log_shard;
    bool history_les_bound = false;
    // FIXME: Uh-oh we have to check this return value; choose_acting can fail!
    ps->choose_acting(auth_log_shard, false, &history_les_bound, true);
    if (!ps->want_acting.empty() && ps->want_acting != ps->acting) {
      psdout(10) << "Active: got notify from previous acting member "
                 << notevt.from << ", requesting pg_temp change"
                 << dendl;
    }
  }
  return discard_event();
}

boost::statechart::result PeeringState::Active::react(const MTrim& trim)
{
  DECLARE_LOCALS;
  ceph_assert(ps->is_primary());

  // peer is informing us of their last_complete_ondisk
  ldout(ps->cct,10) << " replica osd." << trim.from << " lcod " << trim.trim_to << dendl;
  ps->update_peer_last_complete_ondisk(pg_shard_t{trim.from, trim.shard},
                                       trim.trim_to);
  // trim log when the pg is recovered
  ps->calc_min_last_complete_ondisk();
  return discard_event();
}

boost::statechart::result PeeringState::Active::react(const MInfoRec& infoevt)
{
  DECLARE_LOCALS;
  ceph_assert(ps->is_primary());

  ceph_assert(!ps->acting_recovery_backfill.empty());
  if (infoevt.lease_ack) {
    ps->proc_lease_ack(infoevt.from.osd, *infoevt.lease_ack);
  }
  // don't update history (yet) if we are active and primary; the replica
  // may be telling us they have activated (and committed) but we can't
  // share that until _everyone_ does the same.
  if (ps->is_acting_recovery_backfill(infoevt.from) &&
      ps->peer_activated.count(infoevt.from) == 0) {
    psdout(10) << " peer osd." << infoevt.from
               << " activated and committed" << dendl;
    ps->peer_activated.insert(infoevt.from);
    ps->blocked_by.erase(infoevt.from.shard);
    pl->publish_stats_to_osd();
    if (ps->peer_activated.size() == ps->acting_recovery_backfill.size()) {
      all_activated_and_committed();
    }
  }
  return discard_event();
}

boost::statechart::result PeeringState::Active::react(const MLogRec& logevt)
{
  DECLARE_LOCALS;
  psdout(10) << "searching osd." << logevt.from
                     << " log for unfound items" << dendl;
  ps->proc_replica_log(
    logevt.msg->info, logevt.msg->log, std::move(logevt.msg->missing), logevt.from);
  bool got_missing = ps->search_for_missing(
    ps->peer_info[logevt.from],
    ps->peer_missing[logevt.from],
    logevt.from,
    context< PeeringMachine >().get_recovery_ctx());
  // If there are missing AND we are "fully" active then start recovery now
  if (got_missing && ps->state_test(PG_STATE_ACTIVE)) {
    post_event(DoRecovery());
  }
  return discard_event();
}

boost::statechart::result PeeringState::Active::react(const QueryState& q)
{
  DECLARE_LOCALS;

  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;

  {
    q.f->open_array_section("might_have_unfound");
    for (auto p = ps->might_have_unfound.begin();
         p != ps->might_have_unfound.end();
         ++p) {
      q.f->open_object_section("osd");
      q.f->dump_stream("osd") << *p;
      if (ps->peer_missing.count(*p)) {
        q.f->dump_string("status", "already probed");
      } else if (ps->peer_missing_requested.count(*p)) {
        q.f->dump_string("status", "querying");
      } else if (!ps->get_osdmap()->is_up(p->osd)) {
        q.f->dump_string("status", "osd is down");
      } else {
        q.f->dump_string("status", "not queried");
      }
      q.f->close_section();
    }
    q.f->close_section();
  }
  {
    q.f->open_object_section("recovery_progress");
    q.f->open_array_section("backfill_targets");
    for (auto p = ps->backfill_targets.begin();
         p != ps->backfill_targets.end(); ++p)
      q.f->dump_stream("replica") << *p;
    q.f->close_section();
    pl->dump_recovery_info(q.f);
    q.f->close_section();
  }

  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::Active::react(const QueryUnfound& q)
{
  DECLARE_LOCALS;

  ps->query_unfound(q.f, "Active");
  return discard_event();
}

boost::statechart::result PeeringState::Active::react(
  const ActivateCommitted &evt)
{
  DECLARE_LOCALS;
  ceph_assert(!ps->peer_activated.count(ps->pg_whoami));
  ps->peer_activated.insert(ps->pg_whoami);
  psdout(10) << "_activate_committed " << evt.epoch
             << " peer_activated now " << ps->peer_activated
             << " last_interval_started "
             << ps->info.history.last_interval_started
             << " last_epoch_started "
             << ps->info.history.last_epoch_started
             << " same_interval_since "
             << ps->info.history.same_interval_since
             << dendl;
  ceph_assert(!ps->acting_recovery_backfill.empty());
  if (ps->peer_activated.size() == ps->acting_recovery_backfill.size())
    all_activated_and_committed();
  return discard_event();
}

boost::statechart::result PeeringState::Active::react(const AllReplicasActivated &evt)
{

  DECLARE_LOCALS;
  pg_t pgid = context< PeeringMachine >().spgid.pgid;

  all_replicas_activated = true;

  ps->state_clear(PG_STATE_ACTIVATING);
  ps->state_clear(PG_STATE_CREATING);
  ps->state_clear(PG_STATE_PREMERGE);

  bool merge_target;
  if (ps->pool.info.is_pending_merge(pgid, &merge_target)) {
    ps->state_set(PG_STATE_PEERED);
    ps->state_set(PG_STATE_PREMERGE);

    if (ps->actingset.size() != ps->get_osdmap()->get_pg_size(pgid)) {
      if (merge_target) {
        pg_t src = pgid;
        src.set_ps(ps->pool.info.get_pg_num_pending());
        assert(src.get_parent() == pgid);
        pl->set_not_ready_to_merge_target(pgid, src);
      } else {
        pl->set_not_ready_to_merge_source(pgid);
      }
    }
  } else if (!ps->acting_set_writeable()) {
    ps->state_set(PG_STATE_PEERED);
  } else {
    ps->state_set(PG_STATE_ACTIVE);
  }

  auto mnow = pl->get_mnow();
  if (ps->prior_readable_until_ub > mnow) {
    psdout(10) << " waiting for prior_readable_until_ub "
               << ps->prior_readable_until_ub << " > mnow " << mnow << dendl;
    ps->state_set(PG_STATE_WAIT);
    pl->queue_check_readable(
      ps->last_peering_reset,
      ps->prior_readable_until_ub - mnow);
  } else {
    psdout(10) << " mnow " << mnow << " >= prior_readable_until_ub "
               << ps->prior_readable_until_ub << dendl;
  }

  if (ps->pool.info.has_flag(pg_pool_t::FLAG_CREATING)) {
    pl->send_pg_created(pgid);
  }

  ps->info.history.last_epoch_started = ps->info.last_epoch_started;
  ps->info.history.last_interval_started = ps->info.last_interval_started;
  ps->dirty_info = true;

  ps->share_pg_info();
  pl->publish_stats_to_osd();

  pl->on_activate_complete();

  return discard_event();
}

boost::statechart::result PeeringState::Active::react(const RenewLease& rl)
{
  DECLARE_LOCALS;
  ps->proc_renew_lease();
  return discard_event();
}

boost::statechart::result PeeringState::Active::react(const MLeaseAck& la)
{
  DECLARE_LOCALS;
  ps->proc_lease_ack(la.from, la.lease_ack);
  return discard_event();
}


boost::statechart::result PeeringState::Active::react(const CheckReadable &evt)
{
  DECLARE_LOCALS;
  pl->recheck_readable();
  return discard_event();
}

/*
 * update info.history.last_epoch_started ONLY after we and all
 * replicas have activated AND committed the activate transaction
 * (i.e. the peering results are stable on disk).
 */
void PeeringState::Active::all_activated_and_committed()
{
  DECLARE_LOCALS;
  psdout(10) << "all_activated_and_committed" << dendl;
  ceph_assert(ps->is_primary());
  ceph_assert(ps->peer_activated.size() == ps->acting_recovery_backfill.size());
  ceph_assert(!ps->acting_recovery_backfill.empty());
  ceph_assert(ps->blocked_by.empty());

  assert(HAVE_FEATURE(ps->upacting_features, SERVER_OCTOPUS));
  // this is overkill when the activation is quick, but when it is slow it
  // is important, because the lease was renewed by the activate itself but we
  // don't know how long ago that was, and simply scheduling now may leave
  // a gap in lease coverage.  keep it simple and aggressively renew.
  ps->renew_lease(pl->get_mnow());
  ps->send_lease();
  ps->schedule_renew_lease();

  // Degraded?
  ps->update_calc_stats();
  if (ps->info.stats.stats.sum.num_objects_degraded) {
    ps->state_set(PG_STATE_DEGRADED);
  } else {
    ps->state_clear(PG_STATE_DEGRADED);
  }

  post_event(PeeringState::AllReplicasActivated());
}


void PeeringState::Active::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);


  DECLARE_LOCALS;
  pl->cancel_local_background_io_reservation();

  ps->blocked_by.clear();
  ps->backfill_reserved = false;
  ps->state_clear(PG_STATE_ACTIVATING);
  ps->state_clear(PG_STATE_DEGRADED);
  ps->state_clear(PG_STATE_UNDERSIZED);
  ps->state_clear(PG_STATE_BACKFILL_TOOFULL);
  ps->state_clear(PG_STATE_BACKFILL_WAIT);
  ps->state_clear(PG_STATE_RECOVERY_WAIT);
  ps->state_clear(PG_STATE_RECOVERY_TOOFULL);
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_active_latency, dur);
  pl->on_active_exit();
}

/*------ReplicaActive-----*/
PeeringState::ReplicaActive::ReplicaActive(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/ReplicaActive")
{
  context< PeeringMachine >().log_enter(state_name);

  DECLARE_LOCALS;
  ps->start_flush(context< PeeringMachine >().get_cur_transaction());
}


boost::statechart::result PeeringState::ReplicaActive::react(
  const Activate& actevt) {
  DECLARE_LOCALS;
  psdout(10) << "In ReplicaActive, about to call activate" << dendl;
  ps->activate(
    context< PeeringMachine >().get_cur_transaction(),
    actevt.activation_epoch,
    context< PeeringMachine >().get_recovery_ctx());
  psdout(10) << "Activate Finished" << dendl;
  return discard_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(
  const ActivateCommitted &evt)
{
  DECLARE_LOCALS;
  psdout(10) << __func__ << " " << evt.epoch << " telling primary" << dendl;

  auto &rctx = context<PeeringMachine>().get_recovery_ctx();
  auto epoch = ps->get_osdmap_epoch();
  pg_info_t i = ps->info;
  i.history.last_epoch_started = evt.activation_epoch;
  i.history.last_interval_started = i.history.same_interval_since;
  rctx.send_info(
    ps->get_primary().osd,
    spg_t(ps->info.pgid.pgid, ps->get_primary().shard),
    epoch,
    epoch,
    i,
    {}, /* lease */
    ps->get_lease_ack());

  if (ps->acting_set_writeable()) {
    ps->state_set(PG_STATE_ACTIVE);
  } else {
    ps->state_set(PG_STATE_PEERED);
  }
  pl->on_activate_committed();

  return discard_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(const MLease& l)
{
  DECLARE_LOCALS;
  spg_t spgid = context< PeeringMachine >().spgid;
  epoch_t epoch = pl->get_osdmap_epoch();

  ps->proc_lease(l.lease);
  pl->send_cluster_message(
    ps->get_primary().osd,
    TOPNSPC::make_message<MOSDPGLeaseAck>(epoch,
                       spg_t(spgid.pgid, ps->get_primary().shard),
                       ps->get_lease_ack()),
    epoch);
  return discard_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(const MInfoRec& infoevt)
{
  DECLARE_LOCALS;
  ps->proc_primary_info(context<PeeringMachine>().get_cur_transaction(),
                        infoevt.info);
  return discard_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(const MLogRec& logevt)
{
  DECLARE_LOCALS;
  psdout(10) << "received log from " << logevt.from << dendl;
  ObjectStore::Transaction &t = context<PeeringMachine>().get_cur_transaction();
  ps->merge_log(t, logevt.msg->info, std::move(logevt.msg->log), logevt.from);
  ceph_assert(ps->pg_log.get_head() == ps->info.last_update);
  if (logevt.msg->lease) {
    ps->proc_lease(*logevt.msg->lease);
  }

  return discard_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(const MTrim& trim)
{
  DECLARE_LOCALS;
  // primary is instructing us to trim
  ps->pg_log.trim(trim.trim_to, ps->info);
  ps->dirty_info = true;
  return discard_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(const ActMap&)
{
  DECLARE_LOCALS;
  if (ps->should_send_notify() && ps->get_primary().osd >= 0) {
    ps->info.history.refresh_prior_readable_until_ub(
      pl->get_mnow(), ps->prior_readable_until_ub);
    context< PeeringMachine >().send_notify(
      ps->get_primary().osd,
      pg_notify_t(
        ps->get_primary().shard, ps->pg_whoami.shard,
        ps->get_osdmap_epoch(),
        ps->get_osdmap_epoch(),
        ps->info,
        ps->past_intervals));
  }
  return discard_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(
  const MQuery& query)
{
  DECLARE_LOCALS;
  ps->fulfill_query(query, context<PeeringMachine>().get_recovery_ctx());
  return discard_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(const QueryState& q)
{
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;
  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::ReplicaActive::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "ReplicaActive");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::ReplicaActive::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  pl->unreserve_recovery_space();

  pl->cancel_remote_recovery_reservation();
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_replicaactive_latency, dur);

  ps->min_last_complete_ondisk = eversion_t();
}

/*-------Stray---*/
PeeringState::Stray::Stray(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Stray")
{
  context< PeeringMachine >().log_enter(state_name);


  DECLARE_LOCALS;
  ceph_assert(!ps->is_peered());
  ceph_assert(!ps->is_peering());
  ceph_assert(!ps->is_primary());

  if (!ps->get_osdmap()->have_pg_pool(ps->info.pgid.pgid.pool())) {
    ldout(ps->cct,10) << __func__ << " pool is deleted" << dendl;
    post_event(DeleteStart());
  } else {
    ps->start_flush(context< PeeringMachine >().get_cur_transaction());
  }
}

boost::statechart::result PeeringState::Stray::react(const MLogRec& logevt)
{
  DECLARE_LOCALS;
  MOSDPGLog *msg = logevt.msg.get();
  psdout(10) << "got info+log from osd." << logevt.from << " " << msg->info << " " << msg->log << dendl;

  ObjectStore::Transaction &t = context<PeeringMachine>().get_cur_transaction();
  if (msg->info.last_backfill == hobject_t()) {
    // restart backfill
    ps->info = msg->info;
    pl->on_info_history_change();
    ps->dirty_info = true;
    ps->dirty_big_info = true;  // maybe.

    PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(t)};
    ps->pg_log.reset_backfill_claim_log(msg->log, rollbacker.get());

    ps->pg_log.reset_backfill();
  } else {
    ps->merge_log(t, msg->info, std::move(msg->log), logevt.from);
  }
  if (logevt.msg->lease) {
    ps->proc_lease(*logevt.msg->lease);
  }

  ceph_assert(ps->pg_log.get_head() == ps->info.last_update);

  post_event(Activate(logevt.msg->info.last_epoch_started));
  return transit<ReplicaActive>();
}

boost::statechart::result PeeringState::Stray::react(const MInfoRec& infoevt)
{
  DECLARE_LOCALS;
  psdout(10) << "got info from osd." << infoevt.from << " " << infoevt.info << dendl;

  if (ps->info.last_update > infoevt.info.last_update) {
    // rewind divergent log entries
    ObjectStore::Transaction &t = context<PeeringMachine>().get_cur_transaction();
    ps->rewind_divergent_log(t, infoevt.info.last_update);
    ps->info.stats = infoevt.info.stats;
    ps->info.hit_set = infoevt.info.hit_set;
  }

  if (infoevt.lease) {
    ps->proc_lease(*infoevt.lease);
  }

  ceph_assert(infoevt.info.last_update == ps->info.last_update);
  ceph_assert(ps->pg_log.get_head() == ps->info.last_update);

  post_event(Activate(infoevt.info.last_epoch_started));
  return transit<ReplicaActive>();
}

boost::statechart::result PeeringState::Stray::react(const MQuery& query)
{
  DECLARE_LOCALS;
  ps->fulfill_query(query, context<PeeringMachine>().get_recovery_ctx());
  return discard_event();
}

boost::statechart::result PeeringState::Stray::react(const ActMap&)
{
  DECLARE_LOCALS;
  if (ps->should_send_notify() && ps->get_primary().osd >= 0) {
    ps->info.history.refresh_prior_readable_until_ub(
      pl->get_mnow(), ps->prior_readable_until_ub);
    context< PeeringMachine >().send_notify(
      ps->get_primary().osd,
      pg_notify_t(
        ps->get_primary().shard, ps->pg_whoami.shard,
        ps->get_osdmap_epoch(),
        ps->get_osdmap_epoch(),
        ps->info,
        ps->past_intervals));
  }
  return discard_event();
}

void PeeringState::Stray::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_stray_latency, dur);
}


/*--------ToDelete----------*/
PeeringState::ToDelete::ToDelete(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/ToDelete")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;
  pl->get_perf_logger().inc(l_osd_pg_removing);
}

void PeeringState::ToDelete::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  // note: on a successful removal, this path doesn't execute. see
  // do_delete_work().
  pl->get_perf_logger().dec(l_osd_pg_removing);

  pl->cancel_local_background_io_reservation();
}

/*----WaitDeleteReserved----*/
PeeringState::WaitDeleteReserved::WaitDeleteReserved(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history,
               "Started/ToDelete/WaitDeleteReseved")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;
  context< ToDelete >().priority = ps->get_delete_priority();

  pl->cancel_local_background_io_reservation();
  pl->request_local_background_io_reservation(
    context<ToDelete>().priority,
    std::make_unique<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      DeleteReserved()),
    std::make_unique<PGPeeringEvent>(
      ps->get_osdmap_epoch(),
      ps->get_osdmap_epoch(),
      DeleteInterrupted()));
}

boost::statechart::result PeeringState::ToDelete::react(
  const ActMap& evt)
{
  DECLARE_LOCALS;
  if (ps->get_delete_priority() != priority) {
    psdout(10) << __func__ << " delete priority changed, resetting"
                   << dendl;
    return transit<ToDelete>();
  }
  return discard_event();
}

void PeeringState::WaitDeleteReserved::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
}

/*----Deleting-----*/
PeeringState::Deleting::Deleting(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/ToDelete/Deleting")
{
  context< PeeringMachine >().log_enter(state_name);

  DECLARE_LOCALS;
  ps->deleting = true;
  ObjectStore::Transaction &t = context<PeeringMachine>().get_cur_transaction();

  // clear log
  PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(t)};
  ps->pg_log.roll_forward(rollbacker.get());

  // adjust info to backfill
  ps->info.set_last_backfill(hobject_t());
  ps->pg_log.reset_backfill();
  ps->dirty_info = true;

  pl->on_removal(t);
}

boost::statechart::result PeeringState::Deleting::react(
  const DeleteSome& evt)
{
  DECLARE_LOCALS;
  std::pair<ghobject_t, bool> p;
  p = pl->do_delete_work(context<PeeringMachine>().get_cur_transaction(),
    next);
  next = p.first;
  return p.second ? discard_event() : terminate();
}

void PeeringState::Deleting::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  ps->deleting = false;
  pl->cancel_local_background_io_reservation();
}

/*--------GetInfo---------*/
PeeringState::GetInfo::GetInfo(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Peering/GetInfo")
{
  context< PeeringMachine >().log_enter(state_name);


  DECLARE_LOCALS;
  ps->check_past_interval_bounds();
  ps->log_weirdness();
  PastIntervals::PriorSet &prior_set = context< Peering >().prior_set;

  ceph_assert(ps->blocked_by.empty());

  prior_set = ps->build_prior();
  ps->prior_readable_down_osds = prior_set.down;

  if (ps->prior_readable_down_osds.empty()) {
    psdout(10) << " no prior_set down osds, will clear prior_readable_until_ub before activating"
               << dendl;
  }

  ps->reset_min_peer_features();
  get_infos();
  if (prior_set.pg_down) {
    post_event(IsDown());
  } else if (peer_info_requested.empty()) {
    post_event(GotInfo());
  }
}

void PeeringState::GetInfo::get_infos()
{
  DECLARE_LOCALS;
  PastIntervals::PriorSet &prior_set = context< Peering >().prior_set;

  ps->blocked_by.clear();
  for (auto it = prior_set.probe.begin(); it != prior_set.probe.end(); ++it) {
    pg_shard_t peer = *it;
    if (peer == ps->pg_whoami) {
      continue;
    }
    if (ps->peer_info.count(peer)) {
      psdout(10) << " have osd." << peer << " info " << ps->peer_info[peer] << dendl;
      continue;
    }
    if (peer_info_requested.count(peer)) {
      psdout(10) << " already requested info from osd." << peer << dendl;
      ps->blocked_by.insert(peer.osd);
    } else if (!ps->get_osdmap()->is_up(peer.osd)) {
      psdout(10) << " not querying info from down osd." << peer << dendl;
    } else {
      psdout(10) << " querying info from osd." << peer << dendl;
      context< PeeringMachine >().send_query(
        peer.osd,
        pg_query_t(pg_query_t::INFO,
                   it->shard, ps->pg_whoami.shard,
                   ps->info.history,
                   ps->get_osdmap_epoch()));
      peer_info_requested.insert(peer);
      ps->blocked_by.insert(peer.osd);
    }
  }

  ps->check_prior_readable_down_osds(ps->get_osdmap());

  pl->publish_stats_to_osd();
}

boost::statechart::result PeeringState::GetInfo::react(const MNotifyRec& infoevt)
{

  DECLARE_LOCALS;

  auto p = peer_info_requested.find(infoevt.from);
  if (p != peer_info_requested.end()) {
    peer_info_requested.erase(p);
    ps->blocked_by.erase(infoevt.from.osd);
  }

  epoch_t old_start = ps->info.history.last_epoch_started;
  if (ps->proc_replica_info(
        infoevt.from, infoevt.notify.info, infoevt.notify.epoch_sent)) {
    // we got something new ...
    PastIntervals::PriorSet &prior_set = context< Peering >().prior_set;
    if (old_start < ps->info.history.last_epoch_started) {
      psdout(10) << " last_epoch_started moved forward, rebuilding prior" << dendl;
      prior_set = ps->build_prior();
      ps->prior_readable_down_osds = prior_set.down;

      // filter out any osds that got dropped from the probe set from
      // peer_info_requested.  this is less expensive than restarting
      // peering (which would re-probe everyone).
      auto p = peer_info_requested.begin();
      while (p != peer_info_requested.end()) {
        if (prior_set.probe.count(*p) == 0) {
          psdout(20) << " dropping osd." << *p << " from info_requested, no longer in probe set" << dendl;
          peer_info_requested.erase(p++);
        } else {
          ++p;
        }
      }
      get_infos();
    }
    psdout(20) << "Adding osd: " << infoevt.from.osd << " peer features: "
                       << hex << infoevt.features << dec << dendl;
    ps->apply_peer_features(infoevt.features);

    // are we done getting everything?
    if (peer_info_requested.empty() && !prior_set.pg_down) {
      psdout(20) << "Common peer features: " << hex << ps->get_min_peer_features() << dec << dendl;
      psdout(20) << "Common acting features: " << hex << ps->get_min_acting_features() << dec << dendl;
      psdout(20) << "Common upacting features: " << hex << ps->get_min_upacting_features() << dec << dendl;
      post_event(GotInfo());
    }
  }
  return discard_event();
}

boost::statechart::result PeeringState::GetInfo::react(const QueryState& q)
{
  DECLARE_LOCALS;
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;

  q.f->open_array_section("requested_info_from");
  for (auto p = peer_info_requested.begin();
       p != peer_info_requested.end();
       ++p) {
    q.f->open_object_section("osd");
    q.f->dump_stream("osd") << *p;
    if (ps->peer_info.count(*p)) {
      q.f->open_object_section("got_info");
      ps->peer_info[*p].dump(q.f);
      q.f->close_section();
    }
    q.f->close_section();
  }
  q.f->close_section();

  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::GetInfo::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "GetInfo");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::GetInfo::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_getinfo_latency, dur);
  ps->blocked_by.clear();
}

/*------GetLog------------*/
PeeringState::GetLog::GetLog(my_context ctx)
  : my_base(ctx),
    NamedState(
      context< PeeringMachine >().state_history,
      "Started/Primary/Peering/GetLog"),
    msg(0)
{
  context< PeeringMachine >().log_enter(state_name);

  DECLARE_LOCALS;

  ps->log_weirdness();

  // adjust acting?
  if (!ps->choose_acting(auth_log_shard, false,
                         &context< Peering >().history_les_bound)) {
    if (!ps->want_acting.empty()) {
      post_event(NeedActingChange());
    } else {
      post_event(IsIncomplete());
    }
    return;
  }

  // am i the best?
  if (auth_log_shard == ps->pg_whoami) {
    post_event(GotLog());
    return;
  }

  const pg_info_t& best = ps->peer_info[auth_log_shard];

  // am i broken?
  if (ps->info.last_update < best.log_tail) {
    psdout(10) << " not contiguous with osd." << auth_log_shard << ", down" << dendl;
    post_event(IsIncomplete());
    return;
  }

  // how much log to request?
  eversion_t request_log_from = ps->info.last_update;
  ceph_assert(!ps->acting_recovery_backfill.empty());
  for (auto p = ps->acting_recovery_backfill.begin();
       p != ps->acting_recovery_backfill.end();
       ++p) {
    if (*p == ps->pg_whoami) continue;
    pg_info_t& ri = ps->peer_info[*p];
    if (ri.last_update < ps->info.log_tail && ri.last_update >= best.log_tail &&
        ri.last_update < request_log_from)
      request_log_from = ri.last_update;
  }

  // how much?
  psdout(10) << " requesting log from osd." << auth_log_shard << dendl;
  context<PeeringMachine>().send_query(
    auth_log_shard.osd,
    pg_query_t(
      pg_query_t::LOG,
      auth_log_shard.shard, ps->pg_whoami.shard,
      request_log_from, ps->info.history,
      ps->get_osdmap_epoch()));

  ceph_assert(ps->blocked_by.empty());
  ps->blocked_by.insert(auth_log_shard.osd);
  pl->publish_stats_to_osd();
}

boost::statechart::result PeeringState::GetLog::react(const AdvMap& advmap)
{
  // make sure our log source didn't go down.  we need to check
  // explicitly because it may not be part of the prior set, which
  // means the Peering state check won't catch it going down.
  if (!advmap.osdmap->is_up(auth_log_shard.osd)) {
    psdout(10) << "GetLog: auth_log_shard osd."
                       << auth_log_shard.osd << " went down" << dendl;
    post_event(advmap);
    return transit< Reset >();
  }

  // let the Peering state do its checks.
  return forward_event();
}

boost::statechart::result PeeringState::GetLog::react(const MLogRec& logevt)
{
  ceph_assert(!msg);
  if (logevt.from != auth_log_shard) {
    psdout(10) << "GetLog: discarding log from "
                       << "non-auth_log_shard osd." << logevt.from << dendl;
    return discard_event();
  }
  psdout(10) << "GetLog: received master log from osd."
                     << logevt.from << dendl;
  msg = logevt.msg;
  post_event(GotLog());
  return discard_event();
}

boost::statechart::result PeeringState::GetLog::react(const GotLog&)
{

  DECLARE_LOCALS;
  psdout(10) << "leaving GetLog" << dendl;
  if (msg) {
    psdout(10) << "processing master log" << dendl;
    ps->proc_master_log(context<PeeringMachine>().get_cur_transaction(),
                        msg->info, std::move(msg->log), std::move(msg->missing),
                        auth_log_shard);
  }
  ps->start_flush(context< PeeringMachine >().get_cur_transaction());
  return transit< GetMissing >();
}

boost::statechart::result PeeringState::GetLog::react(const QueryState& q)
{
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;
  q.f->dump_stream("auth_log_shard") << auth_log_shard;
  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::GetLog::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "GetLog");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::GetLog::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_getlog_latency, dur);
  ps->blocked_by.clear();
}

/*------WaitActingChange--------*/
PeeringState::WaitActingChange::WaitActingChange(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/WaitActingChange")
{
  context< PeeringMachine >().log_enter(state_name);
}

boost::statechart::result PeeringState::WaitActingChange::react(const AdvMap& advmap)
{
  DECLARE_LOCALS;
  OSDMapRef osdmap = advmap.osdmap;

  psdout(10) << "verifying no want_acting " << ps->want_acting << " targets didn't go down" << dendl;
  for (auto p = ps->want_acting.begin(); p != ps->want_acting.end(); ++p) {
    if (!osdmap->is_up(*p)) {
      psdout(10) << " want_acting target osd." << *p << " went down, resetting" << dendl;
      post_event(advmap);
      return transit< Reset >();
    }
  }
  return forward_event();
}

boost::statechart::result PeeringState::WaitActingChange::react(const MLogRec& logevt)
{
  psdout(10) << "In WaitActingChange, ignoring MLocRec" << dendl;
  return discard_event();
}

boost::statechart::result PeeringState::WaitActingChange::react(const MInfoRec& evt)
{
  psdout(10) << "In WaitActingChange, ignoring MInfoRec" << dendl;
  return discard_event();
}

boost::statechart::result PeeringState::WaitActingChange::react(const MNotifyRec& evt)
{
  psdout(10) << "In WaitActingChange, ignoring MNotifyRec" << dendl;
  return discard_event();
}

boost::statechart::result PeeringState::WaitActingChange::react(const QueryState& q)
{
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;
  q.f->dump_string("comment", "waiting for pg acting set to change");
  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::WaitActingChange::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "WaitActingChange");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::WaitActingChange::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_waitactingchange_latency, dur);
}

/*------Down--------*/
PeeringState::Down::Down(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Peering/Down")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;

  ps->state_clear(PG_STATE_PEERING);
  ps->state_set(PG_STATE_DOWN);

  auto &prior_set = context< Peering >().prior_set;
  ceph_assert(ps->blocked_by.empty());
  ps->blocked_by.insert(prior_set.down.begin(), prior_set.down.end());
  pl->publish_stats_to_osd();
}

void PeeringState::Down::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;

  ps->state_clear(PG_STATE_DOWN);
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_down_latency, dur);

  ps->blocked_by.clear();
}

boost::statechart::result PeeringState::Down::react(const QueryState& q)
{
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;
  q.f->dump_string("comment",
                   "not enough up instances of this PG to go active");
  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::Down::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "Down");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

boost::statechart::result PeeringState::Down::react(const MNotifyRec& infoevt)
{
  DECLARE_LOCALS;

  ceph_assert(ps->is_primary());
  epoch_t old_start = ps->info.history.last_epoch_started;
  if (!ps->peer_info.count(infoevt.from) &&
      ps->get_osdmap()->has_been_up_since(infoevt.from.osd, infoevt.notify.epoch_sent)) {
    ps->update_history(infoevt.notify.info.history);
  }
  // if we got something new to make pg escape down state
  if (ps->info.history.last_epoch_started > old_start) {
      psdout(10) << " last_epoch_started moved forward, re-enter getinfo" << dendl;
    ps->state_clear(PG_STATE_DOWN);
    ps->state_set(PG_STATE_PEERING);
    return transit< GetInfo >();
  }

  return discard_event();
}


/*------Incomplete--------*/
PeeringState::Incomplete::Incomplete(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Peering/Incomplete")
{
  context< PeeringMachine >().log_enter(state_name);
  DECLARE_LOCALS;

  ps->state_clear(PG_STATE_PEERING);
  ps->state_set(PG_STATE_INCOMPLETE);

  PastIntervals::PriorSet &prior_set = context< Peering >().prior_set;
  ceph_assert(ps->blocked_by.empty());
  ps->blocked_by.insert(prior_set.down.begin(), prior_set.down.end());
  pl->publish_stats_to_osd();
}

boost::statechart::result PeeringState::Incomplete::react(const AdvMap &advmap) {
  DECLARE_LOCALS;
  int64_t poolnum = ps->info.pgid.pool();

  // Reset if min_size turn smaller than previous value, pg might now be able to go active
  if (!advmap.osdmap->have_pg_pool(poolnum) ||
      advmap.lastmap->get_pools().find(poolnum)->second.min_size >
      advmap.osdmap->get_pools().find(poolnum)->second.min_size) {
    post_event(advmap);
    return transit< Reset >();
  }

  return forward_event();
}

boost::statechart::result PeeringState::Incomplete::react(const MNotifyRec& notevt) {
  DECLARE_LOCALS;
  psdout(7) << "handle_pg_notify from osd." << notevt.from << dendl;
  if (ps->proc_replica_info(
    notevt.from, notevt.notify.info, notevt.notify.epoch_sent)) {
    // We got something new, try again!
    return transit< GetLog >();
  } else {
    return discard_event();
  }
}

boost::statechart::result PeeringState::Incomplete::react(
  const QueryState& q)
{
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;
  q.f->dump_string("comment", "not enough complete instances of this PG");
  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::Incomplete::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "Incomplete");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::Incomplete::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;

  ps->state_clear(PG_STATE_INCOMPLETE);
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_incomplete_latency, dur);

  ps->blocked_by.clear();
}

/*------GetMissing--------*/
PeeringState::GetMissing::GetMissing(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Peering/GetMissing")
{
  context< PeeringMachine >().log_enter(state_name);

  DECLARE_LOCALS;
  ps->log_weirdness();
  ceph_assert(!ps->acting_recovery_backfill.empty());
  eversion_t since;
  for (auto i = ps->acting_recovery_backfill.begin();
       i != ps->acting_recovery_backfill.end();
       ++i) {
    if (*i == ps->get_primary()) continue;
    const pg_info_t& pi = ps->peer_info[*i];
    // reset this so to make sure the pg_missing_t is initialized and
    // has the correct semantics even if we don't need to get a
    // missing set from a shard. This way later additions due to
    // lost+unfound delete work properly.
    ps->peer_missing[*i].may_include_deletes = !ps->perform_deletes_during_peering();

    if (pi.is_empty())
      continue;                                // no pg data, nothing divergent

    if (pi.last_update < ps->pg_log.get_tail()) {
      psdout(10) << " osd." << *i << " is not contiguous, will restart backfill" << dendl;
      ps->peer_missing[*i].clear();
      continue;
    }
    if (pi.last_backfill == hobject_t()) {
      psdout(10) << " osd." << *i << " will fully backfill; can infer empty missing set" << dendl;
      ps->peer_missing[*i].clear();
      continue;
    }

    if (pi.last_update == pi.last_complete &&  // peer has no missing
        pi.last_update == ps->info.last_update) {  // peer is up to date
      // replica has no missing and identical log as us.  no need to
      // pull anything.
      // FIXME: we can do better here.  if last_update==last_complete we
      //        can infer the rest!
      psdout(10) << " osd." << *i << " has no missing, identical log" << dendl;
      ps->peer_missing[*i].clear();
      continue;
    }

    // We pull the log from the peer's last_epoch_started to ensure we
    // get enough log to detect divergent updates.
    since.epoch = pi.last_epoch_started;
    ceph_assert(pi.last_update >= ps->info.log_tail);  // or else choose_acting() did a bad thing
    if (pi.log_tail <= since) {
      psdout(10) << " requesting log+missing since " << since << " from osd." << *i << dendl;
      context< PeeringMachine >().send_query(
        i->osd,
        pg_query_t(
          pg_query_t::LOG,
          i->shard, ps->pg_whoami.shard,
          since, ps->info.history,
          ps->get_osdmap_epoch()));
    } else {
      psdout(10) << " requesting fulllog+missing from osd." << *i
                         << " (want since " << since << " < log.tail "
                         << pi.log_tail << ")" << dendl;
      context< PeeringMachine >().send_query(
        i->osd, pg_query_t(
          pg_query_t::FULLLOG,
          i->shard, ps->pg_whoami.shard,
          ps->info.history, ps->get_osdmap_epoch()));
    }
    peer_missing_requested.insert(*i);
    ps->blocked_by.insert(i->osd);
  }

  if (peer_missing_requested.empty()) {
    if (ps->need_up_thru) {
      psdout(10) << " still need up_thru update before going active"
                         << dendl;
      post_event(NeedUpThru());
      return;
    }

    // all good!
    post_event(Activate(ps->get_osdmap_epoch()));
  } else {
    pl->publish_stats_to_osd();
  }
}

boost::statechart::result PeeringState::GetMissing::react(const MLogRec& logevt)
{
  DECLARE_LOCALS;

  peer_missing_requested.erase(logevt.from);
  ps->proc_replica_log(logevt.msg->info,
                       logevt.msg->log,
                       std::move(logevt.msg->missing),
                       logevt.from);

  if (peer_missing_requested.empty()) {
    if (ps->need_up_thru) {
      psdout(10) << " still need up_thru update before going active"
                         << dendl;
      post_event(NeedUpThru());
    } else {
      psdout(10) << "Got last missing, don't need missing "
                         << "posting Activate" << dendl;
      post_event(Activate(ps->get_osdmap_epoch()));
    }
  }
  return discard_event();
}

boost::statechart::result PeeringState::GetMissing::react(const QueryState& q)
{
  DECLARE_LOCALS;
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;

  q.f->open_array_section("peer_missing_requested");
  for (auto p = peer_missing_requested.begin();
       p != peer_missing_requested.end();
       ++p) {
    q.f->open_object_section("osd");
    q.f->dump_stream("osd") << *p;
    if (ps->peer_missing.count(*p)) {
      q.f->open_object_section("got_missing");
      ps->peer_missing[*p].dump(q.f);
      q.f->close_section();
    }
    q.f->close_section();
  }
  q.f->close_section();

  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::GetMissing::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "GetMising");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::GetMissing::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);

  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_getmissing_latency, dur);
  ps->blocked_by.clear();
}

/*------WaitUpThru--------*/
PeeringState::WaitUpThru::WaitUpThru(my_context ctx)
  : my_base(ctx),
    NamedState(context< PeeringMachine >().state_history, "Started/Primary/Peering/WaitUpThru")
{
  context< PeeringMachine >().log_enter(state_name);
}

boost::statechart::result PeeringState::WaitUpThru::react(const ActMap& am)
{
  DECLARE_LOCALS;
  if (!ps->need_up_thru) {
    post_event(Activate(ps->get_osdmap_epoch()));
  }
  return forward_event();
}

boost::statechart::result PeeringState::WaitUpThru::react(const MLogRec& logevt)
{
  DECLARE_LOCALS;
  psdout(10) << "Noting missing from osd." << logevt.from << dendl;
  ps->peer_missing[logevt.from].claim(std::move(logevt.msg->missing));
  ps->peer_info[logevt.from] = logevt.msg->info;
  return discard_event();
}

boost::statechart::result PeeringState::WaitUpThru::react(const QueryState& q)
{
  q.f->open_object_section("state");
  q.f->dump_string("name", state_name);
  q.f->dump_stream("enter_time") << enter_time;
  q.f->dump_string("comment", "waiting for osdmap to reflect a new up_thru for this osd");
  q.f->close_section();
  return forward_event();
}

boost::statechart::result PeeringState::WaitUpThru::react(const QueryUnfound& q)
{
  q.f->dump_string("state", "WaitUpThru");
  q.f->dump_bool("available_might_have_unfound", false);
  return discard_event();
}

void PeeringState::WaitUpThru::exit()
{
  context< PeeringMachine >().log_exit(state_name, enter_time);
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  pl->get_peering_perf().tinc(rs_waitupthru_latency, dur);
}

/*----PeeringState::PeeringMachine Methods-----*/
#undef dout_prefix
#define dout_prefix dpp->gen_prefix(*_dout)

void PeeringState::PeeringMachine::log_enter(const char *state_name)
{
  DECLARE_LOCALS;
  psdout(5) << "enter " << state_name << dendl;
  pl->log_state_enter(state_name);
}

void PeeringState::PeeringMachine::log_exit(const char *state_name, utime_t enter_time)
{
  DECLARE_LOCALS;
  utime_t dur = ceph_clock_now() - enter_time;
  psdout(5) << "exit " << state_name << " " << dur << " " << event_count << " " << event_time << dendl;
  pl->log_state_exit(state_name, enter_time, event_count, event_time);
  event_count = 0;
  event_time = utime_t();
}

ostream &operator<<(ostream &out, const PeeringState &ps) {
  out << "pg[" << ps.info
      << " " << pg_vector_string(ps.up);
  if (ps.acting != ps.up)
    out << "/" << pg_vector_string(ps.acting);
  if (ps.is_ec_pg())
    out << "p" << ps.get_primary();
  if (!ps.async_recovery_targets.empty())
    out << " async=[" << ps.async_recovery_targets << "]";
  if (!ps.backfill_targets.empty())
    out << " backfill=[" << ps.backfill_targets << "]";
  out << " r=" << ps.get_role();
  out << " lpr=" << ps.get_last_peering_reset();

  if (ps.deleting)
    out << " DELETING";

  if (!ps.past_intervals.empty()) {
    out << " pi=[" << ps.past_intervals.get_bounds()
        << ")/" << ps.past_intervals.size();
  }

  if (ps.is_peered()) {
    if (ps.last_update_ondisk != ps.info.last_update)
      out << " luod=" << ps.last_update_ondisk;
    if (ps.last_update_applied != ps.info.last_update)
      out << " lua=" << ps.last_update_applied;
  }

  if (ps.pg_log.get_tail() != ps.info.log_tail ||
      ps.pg_log.get_head() != ps.info.last_update)
    out << " (info mismatch, " << ps.pg_log.get_log() << ")";

  if (!ps.pg_log.get_log().empty()) {
    if ((ps.pg_log.get_log().log.begin()->version <= ps.pg_log.get_tail())) {
      out << " (log bound mismatch, actual=["
          << ps.pg_log.get_log().log.begin()->version << ","
          << ps.pg_log.get_log().log.rbegin()->version << "]";
      out << ")";
    }
  }

  out << " crt=" << ps.pg_log.get_can_rollback_to();

  if (ps.last_complete_ondisk != ps.info.last_complete)
    out << " lcod " << ps.last_complete_ondisk;

  out << " mlcod " << ps.min_last_complete_ondisk;

  out << " " << pg_state_string(ps.get_state());
  if (ps.should_send_notify())
    out << " NOTIFY";

  if (ps.prior_readable_until_ub != ceph::signedspan::zero()) {
    out << " pruub " << ps.prior_readable_until_ub
        << "@" << ps.get_prior_readable_down_osds();
  }
  return out;
}

std::vector<pg_shard_t> PeeringState::get_replica_recovery_order() const
{
  std::vector<std::pair<unsigned int, pg_shard_t>> replicas_by_num_missing,
    async_by_num_missing;
  replicas_by_num_missing.reserve(get_acting_recovery_backfill().size() - 1);
  for (auto &p : get_acting_recovery_backfill()) {
    if (p == get_primary()) {
      continue;
    }
    auto pm = get_peer_missing().find(p);
    assert(pm != get_peer_missing().end());
    auto nm = pm->second.num_missing();
    if (nm != 0) {
      if (is_async_recovery_target(p)) {
        async_by_num_missing.push_back(make_pair(nm, p));
      } else {
        replicas_by_num_missing.push_back(make_pair(nm, p));
      }
    }
  }
  // sort by number of missing objects, in ascending order.
  auto func = [](const std::pair<unsigned int, pg_shard_t> &lhs,
                 const std::pair<unsigned int, pg_shard_t> &rhs) {
    return lhs.first < rhs.first;
  };
  // acting goes first
  std::sort(replicas_by_num_missing.begin(), replicas_by_num_missing.end(), func);
  // then async_recovery_targets
  std::sort(async_by_num_missing.begin(), async_by_num_missing.end(), func);
  replicas_by_num_missing.insert(replicas_by_num_missing.end(),
    async_by_num_missing.begin(), async_by_num_missing.end());

  std::vector<pg_shard_t> ret;
  ret.reserve(replicas_by_num_missing.size());
  for (auto p : replicas_by_num_missing) {
    ret.push_back(p.second);
  }
  return ret;
}