// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
	// vim: ts=8 sw=2 smarttab
	/*
	 * Ceph - scalable distributed file system
	 *
	 * Copyright (C) 2011 New Dream Network
	 * Copyright (C) 2013,2014 Cloudwatt <libre.licensing@cloudwatt.com>
	 *
	 * Author: Loic Dachary <loic@dachary.org>
	 *
	 * This is free software; you can redistribute it and/or
	 * modify it under the terms of the GNU Lesser General Public
	 * License version 2.1, as published by the Free Software
	 * Foundation.  See file COPYING.
	 *
	 */
	
	#include <list>
	#include <map>
	#include <ostream>
	#include <sstream>
	#include <set>
	#include <string>
	#include <utility>
	#include <vector>
	
	
	#include <boost/assign/list_of.hpp>
	
	#include "include/ceph_features.h"
	#include "include/encoding.h"
	#include "include/stringify.h"
	extern "C" {
	#include "crush/hash.h"
	}
	
	#include "common/Formatter.h"
	#include "OSDMap.h"
	#include "osd_types.h"
	#include "os/Transaction.h"
	
	using std::list;
	using std::make_pair;
	using std::map;
	using std::ostream;
	using std::ostringstream;
	using std::pair;
	using std::set;
	using std::string;
	using std::stringstream;
	using std::unique_ptr;
	using std::vector;
	
	using ceph::decode;
	using ceph::decode_nohead;
	using ceph::encode;
	using ceph::encode_nohead;
	using ceph::Formatter;
	
	using namespace std::literals;
	
	const char *ceph_osd_flag_name(unsigned flag)
	{
	  switch (flag) {
	  case CEPH_OSD_FLAG_ACK: return "ack";
	  case CEPH_OSD_FLAG_ONNVRAM: return "onnvram";
	  case CEPH_OSD_FLAG_ONDISK: return "ondisk";
	  case CEPH_OSD_FLAG_RETRY: return "retry";
	  case CEPH_OSD_FLAG_READ: return "read";
	  case CEPH_OSD_FLAG_WRITE: return "write";
	  case CEPH_OSD_FLAG_ORDERSNAP: return "ordersnap";
	  case CEPH_OSD_FLAG_PEERSTAT_OLD: return "peerstat_old";
	  case CEPH_OSD_FLAG_BALANCE_READS: return "balance_reads";
	  case CEPH_OSD_FLAG_PARALLELEXEC: return "parallelexec";
	  case CEPH_OSD_FLAG_PGOP: return "pgop";
	  case CEPH_OSD_FLAG_EXEC: return "exec";
	  case CEPH_OSD_FLAG_EXEC_PUBLIC: return "exec_public";
	  case CEPH_OSD_FLAG_LOCALIZE_READS: return "localize_reads";
	  case CEPH_OSD_FLAG_RWORDERED: return "rwordered";
	  case CEPH_OSD_FLAG_IGNORE_CACHE: return "ignore_cache";
	  case CEPH_OSD_FLAG_SKIPRWLOCKS: return "skiprwlocks";
	  case CEPH_OSD_FLAG_IGNORE_OVERLAY: return "ignore_overlay";
	  case CEPH_OSD_FLAG_FLUSH: return "flush";
	  case CEPH_OSD_FLAG_MAP_SNAP_CLONE: return "map_snap_clone";
	  case CEPH_OSD_FLAG_ENFORCE_SNAPC: return "enforce_snapc";
	  case CEPH_OSD_FLAG_REDIRECTED: return "redirected";
	  case CEPH_OSD_FLAG_KNOWN_REDIR: return "known_if_redirected";
	  case CEPH_OSD_FLAG_FULL_TRY: return "full_try";
	  case CEPH_OSD_FLAG_FULL_FORCE: return "full_force";
	  case CEPH_OSD_FLAG_IGNORE_REDIRECT: return "ignore_redirect";
	  case CEPH_OSD_FLAG_RETURNVEC: return "returnvec";
	  default: return "???";
	  }
	}
	
	string ceph_osd_flag_string(unsigned flags)
	{
	  string s;
	  for (unsigned i=0; i<32; ++i) {
	    if (flags & (1u<<i)) {
	      if (s.length())
		s += "+";
	      s += ceph_osd_flag_name(1u << i);
	    }
	  }
	  if (s.length())
	    return s;
	  return string("-");
	}
	
	const char * ceph_osd_op_flag_name(unsigned flag)
	{
	  const char *name;
	
	  switch(flag) {
	    case CEPH_OSD_OP_FLAG_EXCL:
	      name = "excl";
	      break;
	    case CEPH_OSD_OP_FLAG_FAILOK:
	      name = "failok";
	      break;
	    case CEPH_OSD_OP_FLAG_FADVISE_RANDOM:
	      name = "fadvise_random";
	      break;
	    case CEPH_OSD_OP_FLAG_FADVISE_SEQUENTIAL:
	      name = "fadvise_sequential";
	      break;
	    case CEPH_OSD_OP_FLAG_FADVISE_WILLNEED:
	      name = "favise_willneed";
	      break;
	    case CEPH_OSD_OP_FLAG_FADVISE_DONTNEED:
	      name = "fadvise_dontneed";
	      break;
	    case CEPH_OSD_OP_FLAG_FADVISE_NOCACHE:
	      name = "fadvise_nocache";
	      break;
	    case CEPH_OSD_OP_FLAG_WITH_REFERENCE:
	      name = "with_reference";
	      break;
	    case CEPH_OSD_OP_FLAG_BYPASS_CLEAN_CACHE:
	      name = "bypass_clean_cache";
	      break;
	    default:
	      name = "???";
	  };
	
	  return name;
	}
	
	string ceph_osd_op_flag_string(unsigned flags)
	{
	  string s;
	  for (unsigned i=0; i<32; ++i) {
	    if (flags & (1u<<i)) {
	      if (s.length())
		s += "+";
	      s += ceph_osd_op_flag_name(1u << i);
	    }
	  }
	  if (s.length())
	    return s;
	  return string("-");
	}
	
	string ceph_osd_alloc_hint_flag_string(unsigned flags)
	{
	  string s;
	  for (unsigned i=0; i<32; ++i) {
	    if (flags & (1u<<i)) {
	      if (s.length())
		s += "+";
	      s += ceph_osd_alloc_hint_flag_name(1u << i);
	    }
	  }
	  if (s.length())
	    return s;
	  return string("-");
	}
	
	void pg_shard_t::encode(ceph::buffer::list &bl) const
	{
	  ENCODE_START(1, 1, bl);
	  encode(osd, bl);
	  encode(shard, bl);
	  ENCODE_FINISH(bl);
	}
	void pg_shard_t::decode(ceph::buffer::list::const_iterator &bl)
	{
	  DECODE_START(1, bl);
	  decode(osd, bl);
	  decode(shard, bl);
	  DECODE_FINISH(bl);
	}
	
	ostream &operator<<(ostream &lhs, const pg_shard_t &rhs)
	{
	  if (rhs.is_undefined())
	    return lhs << "?";
	  if (rhs.shard == shard_id_t::NO_SHARD)
	    return lhs << rhs.get_osd();
	  return lhs << rhs.get_osd() << '(' << (unsigned)(rhs.shard) << ')';
	}
	
	void dump(Formatter* f, const osd_alerts_t& alerts)
	{
	  for (auto& a : alerts) {
	    string s0 = " osd: ";
	    s0 += stringify(a.first);
	    string s;
	    for (auto& aa : a.second) {
	      s = s0;
	      s += " ";
	      s += aa.first;
	      s += ":";
	      s += aa.second;
	      f->dump_string("alert", s);
	    }
	  }
	}
	
	// -- osd_reqid_t --
	void osd_reqid_t::dump(Formatter *f) const
	{
	  f->dump_stream("name") << name;
	  f->dump_int("inc", inc);
	  f->dump_unsigned("tid", tid);
	}
	
	void osd_reqid_t::generate_test_instances(list<osd_reqid_t*>& o)
	{
	  o.push_back(new osd_reqid_t);
	  o.push_back(new osd_reqid_t(entity_name_t::CLIENT(123), 1, 45678));
	}
	
	// -- object_locator_t --
	
	void object_locator_t::encode(ceph::buffer::list& bl) const
	{
	  // verify that nobody's corrupted the locator
	  ceph_assert(hash == -1 || key.empty());
	  __u8 encode_compat = 3;
	  ENCODE_START(6, encode_compat, bl);
	  encode(pool, bl);
	  int32_t preferred = -1;  // tell old code there is no preferred osd (-1).
	  encode(preferred, bl);
	  encode(key, bl);
	  encode(nspace, bl);
	  encode(hash, bl);
	  if (hash != -1)
	    encode_compat = std::max<std::uint8_t>(encode_compat, 6); // need to interpret the hash
	  ENCODE_FINISH_NEW_COMPAT(bl, encode_compat);
	}
	
	void object_locator_t::decode(ceph::buffer::list::const_iterator& p)
	{
	  DECODE_START_LEGACY_COMPAT_LEN(6, 3, 3, p);
	  if (struct_v < 2) {
	    int32_t op;
	    decode(op, p);
	    pool = op;
	    int16_t pref;
	    decode(pref, p);
	  } else {
	    decode(pool, p);
	    int32_t preferred;
	    decode(preferred, p);
	  }
	  decode(key, p);
	  if (struct_v >= 5)
	    decode(nspace, p);
	  if (struct_v >= 6)
	    decode(hash, p);
	  else
	    hash = -1;
	  DECODE_FINISH(p);
	  // verify that nobody's corrupted the locator
	  ceph_assert(hash == -1 || key.empty());
	}
	
	void object_locator_t::dump(Formatter *f) const
	{
	  f->dump_int("pool", pool);
	  f->dump_string("key", key);
	  f->dump_string("namespace", nspace);
	  f->dump_int("hash", hash);
	}
	
	void object_locator_t::generate_test_instances(list<object_locator_t*>& o)
	{
	  o.push_back(new object_locator_t);
	  o.push_back(new object_locator_t(123));
	  o.push_back(new object_locator_t(123, 876));
	  o.push_back(new object_locator_t(1, "n2"));
	  o.push_back(new object_locator_t(1234, "", "key"));
	  o.push_back(new object_locator_t(12, "n1", "key2"));
	}
	
	// -- request_redirect_t --
	void request_redirect_t::encode(ceph::buffer::list& bl) const
	{
	  ENCODE_START(1, 1, bl);
	  encode(redirect_locator, bl);
	  encode(redirect_object, bl);
	  // legacy of the removed osd_instructions member
	  encode((uint32_t)0, bl);
	  ENCODE_FINISH(bl);
	}
	
	void request_redirect_t::decode(ceph::buffer::list::const_iterator& bl)
	{
	  DECODE_START(1, bl);
	  uint32_t legacy_osd_instructions_len;
	  decode(redirect_locator, bl);
	  decode(redirect_object, bl);
	  decode(legacy_osd_instructions_len, bl);
	  if (legacy_osd_instructions_len) {
	    bl.advance(legacy_osd_instructions_len);
	  }
	  DECODE_FINISH(bl);
	}
	
	void request_redirect_t::dump(Formatter *f) const
	{
	  f->dump_string("object", redirect_object);
	  f->open_object_section("locator");
	  redirect_locator.dump(f);
	  f->close_section(); // locator
	}
	
	void request_redirect_t::generate_test_instances(list<request_redirect_t*>& o)
	{
	  object_locator_t loc(1, "redir_obj");
	  o.push_back(new request_redirect_t());
	  o.push_back(new request_redirect_t(loc, 0));
	  o.push_back(new request_redirect_t(loc, "redir_obj"));
	  o.push_back(new request_redirect_t(loc));
	}
	
	void objectstore_perf_stat_t::dump(Formatter *f) const
	{
	  // *_ms values just for compatibility.
	  f->dump_float("commit_latency_ms", os_commit_latency_ns / 1000000.0);
	  f->dump_float("apply_latency_ms", os_apply_latency_ns / 1000000.0);
	  f->dump_unsigned("commit_latency_ns", os_commit_latency_ns);
	  f->dump_unsigned("apply_latency_ns", os_apply_latency_ns);
	}
	
	void objectstore_perf_stat_t::encode(ceph::buffer::list &bl, uint64_t features) const
	{
	  uint8_t target_v = 2;
	  if (!HAVE_FEATURE(features, OS_PERF_STAT_NS)) {
	    target_v = 1;
	  }
	  ENCODE_START(target_v, target_v, bl);
	  if (target_v >= 2) {
	    encode(os_commit_latency_ns, bl);
	    encode(os_apply_latency_ns, bl);
	  } else {
	    constexpr auto NS_PER_MS = std::chrono::nanoseconds(1ms).count();
	    uint32_t commit_latency_ms = os_commit_latency_ns / NS_PER_MS;
	    uint32_t apply_latency_ms = os_apply_latency_ns / NS_PER_MS;
	    encode(commit_latency_ms, bl); // for compatibility with older monitor.
	    encode(apply_latency_ms, bl); // for compatibility with older monitor.
	  }
	  ENCODE_FINISH(bl);
	}
	
	void objectstore_perf_stat_t::decode(ceph::buffer::list::const_iterator &bl)
	{
	  DECODE_START(2, bl);
	  if (struct_v >= 2) {
	    decode(os_commit_latency_ns, bl);
	    decode(os_apply_latency_ns, bl);
	  } else {
	    uint32_t commit_latency_ms;
	    uint32_t apply_latency_ms;
	    decode(commit_latency_ms, bl);
	    decode(apply_latency_ms, bl);
	    constexpr auto NS_PER_MS = std::chrono::nanoseconds(1ms).count();
	    os_commit_latency_ns = commit_latency_ms * NS_PER_MS;
	    os_apply_latency_ns = apply_latency_ms * NS_PER_MS;
	  }
	  DECODE_FINISH(bl);
	}
	
	void objectstore_perf_stat_t::generate_test_instances(std::list<objectstore_perf_stat_t*>& o)
	{
	  o.push_back(new objectstore_perf_stat_t());
	  o.push_back(new objectstore_perf_stat_t());
	  o.back()->os_commit_latency_ns = 20000000;
	  o.back()->os_apply_latency_ns = 30000000;
	}
	
	// -- osd_stat_t --
	void osd_stat_t::dump(Formatter *f) const
	{
	  f->dump_unsigned("up_from", up_from);
	  f->dump_unsigned("seq", seq);
	  f->dump_unsigned("num_pgs", num_pgs);
	  f->dump_unsigned("num_osds", num_osds);
	  f->dump_unsigned("num_per_pool_osds", num_per_pool_osds);
	  f->dump_unsigned("num_per_pool_omap_osds", num_per_pool_omap_osds);
	
	  /// dump legacy stats fields to ensure backward compatibility.
	  f->dump_unsigned("kb", statfs.kb());
	  f->dump_unsigned("kb_used", statfs.kb_used_raw());
	  f->dump_unsigned("kb_used_data", statfs.kb_used_data());
	  f->dump_unsigned("kb_used_omap", statfs.kb_used_omap());
	  f->dump_unsigned("kb_used_meta", statfs.kb_used_internal_metadata());
	  f->dump_unsigned("kb_avail", statfs.kb_avail());
	  ////////////////////
	
	  f->open_object_section("statfs");
	  statfs.dump(f);
	  f->close_section();
	  f->open_array_section("hb_peers");
	  for (auto p : hb_peers)
	    f->dump_int("osd", p);
	  f->close_section();
	  f->dump_int("snap_trim_queue_len", snap_trim_queue_len);
	  f->dump_int("num_snap_trimming", num_snap_trimming);
	  f->dump_int("num_shards_repaired", num_shards_repaired);
	  f->open_object_section("op_queue_age_hist");
	  op_queue_age_hist.dump(f);
	  f->close_section();
	  f->open_object_section("perf_stat");
	  os_perf_stat.dump(f);
	  f->close_section();
	  f->open_array_section("alerts");
	  ::dump(f, os_alerts);
	  f->close_section();
	  f->open_array_section("network_ping_times");
	  for (auto &i : hb_pingtime) {
	    f->open_object_section("entry");
	    f->dump_int("osd", i.first);
	    const time_t lu(i.second.last_update);
	    char buffer[26];
	    string lustr(ctime_r(&lu, buffer));
	    lustr.pop_back();   // Remove trailing \n
	    f->dump_string("last update", lustr);
	    f->open_array_section("interfaces");
	    f->open_object_section("interface");
	    f->dump_string("interface", "back");
	    f->open_object_section("average");
	    f->dump_format_unquoted("1min", "%s", fixed_u_to_string(i.second.back_pingtime[0],3).c_str());
	    f->dump_format_unquoted("5min", "%s", fixed_u_to_string(i.second.back_pingtime[1],3).c_str());
	    f->dump_format_unquoted("15min", "%s", fixed_u_to_string(i.second.back_pingtime[2],3).c_str());
	    f->close_section(); // average
	    f->open_object_section("min");
	    f->dump_format_unquoted("1min", "%s", fixed_u_to_string(i.second.back_min[0],3).c_str());
	    f->dump_format_unquoted("5min", "%s", fixed_u_to_string(i.second.back_min[1],3).c_str());
	    f->dump_format_unquoted("15min", "%s", fixed_u_to_string(i.second.back_min[2],3).c_str());
	    f->close_section(); // min
	    f->open_object_section("max");
	    f->dump_format_unquoted("1min", "%s", fixed_u_to_string(i.second.back_max[0],3).c_str());
	    f->dump_format_unquoted("5min", "%s", fixed_u_to_string(i.second.back_max[1],3).c_str());
	    f->dump_format_unquoted("15min", "%s", fixed_u_to_string(i.second.back_max[2],3).c_str());
	    f->close_section(); // max
	    f->dump_format_unquoted("last", "%s", fixed_u_to_string(i.second.back_last,3).c_str());
	    f->close_section(); // interface
	
	    if (i.second.front_pingtime[0] != 0) {
	      f->open_object_section("interface");
	      f->dump_string("interface", "front");
	      f->open_object_section("average");
	      f->dump_format_unquoted("1min", "%s", fixed_u_to_string(i.second.front_pingtime[0],3).c_str());
	      f->dump_format_unquoted("5min", "%s", fixed_u_to_string(i.second.front_pingtime[1],3).c_str());
	      f->dump_format_unquoted("15min", "%s", fixed_u_to_string(i.second.front_pingtime[2],3).c_str());
	      f->close_section(); // average
	      f->open_object_section("min");
	      f->dump_format_unquoted("1min", "%s", fixed_u_to_string(i.second.front_min[0],3).c_str());
	      f->dump_format_unquoted("5min", "%s", fixed_u_to_string(i.second.front_min[1],3).c_str());
	      f->dump_format_unquoted("15min", "%s", fixed_u_to_string(i.second.front_min[2],3).c_str());
	      f->close_section(); // min
	      f->open_object_section("max");
	      f->dump_format_unquoted("1min", "%s", fixed_u_to_string(i.second.front_max[0],3).c_str());
	      f->dump_format_unquoted("5min", "%s", fixed_u_to_string(i.second.front_max[1],3).c_str());
	      f->dump_format_unquoted("15min", "%s", fixed_u_to_string(i.second.front_max[2],3).c_str());
	      f->close_section(); // max
	      f->dump_format_unquoted("last", "%s", fixed_u_to_string(i.second.front_last,3).c_str());
	      f->close_section(); // interface
	    }
	    f->close_section(); // interfaces
	    f->close_section(); // entry
	  }
	  f->close_section(); // network_ping_time
	}
	
	void osd_stat_t::encode(ceph::buffer::list &bl, uint64_t features) const
	{
	  ENCODE_START(14, 2, bl);
	
	  //////// for compatibility ////////
	  int64_t kb = statfs.kb();
	  int64_t kb_used = statfs.kb_used_raw();
	  int64_t kb_avail = statfs.kb_avail();
	  encode(kb, bl);
	  encode(kb_used, bl);
	  encode(kb_avail, bl);
	  ///////////////////////////////////
	
	  encode(snap_trim_queue_len, bl);
	  encode(num_snap_trimming, bl);
	  encode(hb_peers, bl);
	  encode((uint32_t)0, bl);
	  encode(op_queue_age_hist, bl);
	  encode(os_perf_stat, bl, features);
	  encode(up_from, bl);
	  encode(seq, bl);
	  encode(num_pgs, bl);
	
	  //////// for compatibility ////////
	  int64_t kb_used_data = statfs.kb_used_data();
	  int64_t kb_used_omap = statfs.kb_used_omap();
	  int64_t kb_used_meta = statfs.kb_used_internal_metadata();
	  encode(kb_used_data, bl);
	  encode(kb_used_omap, bl);
	  encode(kb_used_meta, bl);
	  encode(statfs, bl);
	  ///////////////////////////////////
	  encode(os_alerts, bl);
	  encode(num_shards_repaired, bl);
	  encode(num_osds, bl);
	  encode(num_per_pool_osds, bl);
	  encode(num_per_pool_omap_osds, bl);
	
	  // hb_pingtime map
	  encode((int)hb_pingtime.size(), bl);
	  for (auto i : hb_pingtime) {
	    encode(i.first, bl); // osd
	    encode(i.second.last_update, bl);
	    encode(i.second.back_pingtime[0], bl);
	    encode(i.second.back_pingtime[1], bl);
	    encode(i.second.back_pingtime[2], bl);
	    encode(i.second.back_min[0], bl);
	    encode(i.second.back_min[1], bl);
	    encode(i.second.back_min[2], bl);
	    encode(i.second.back_max[0], bl);
	    encode(i.second.back_max[1], bl);
	    encode(i.second.back_max[2], bl);
	    encode(i.second.back_last, bl);
	    encode(i.second.front_pingtime[0], bl);
	    encode(i.second.front_pingtime[1], bl);
	    encode(i.second.front_pingtime[2], bl);
	    encode(i.second.front_min[0], bl);
	    encode(i.second.front_min[1], bl);
	    encode(i.second.front_min[2], bl);
	    encode(i.second.front_max[0], bl);
	    encode(i.second.front_max[1], bl);
	    encode(i.second.front_max[2], bl);
	    encode(i.second.front_last, bl);
	  }
	  ENCODE_FINISH(bl);
	}
	
	void osd_stat_t::decode(ceph::buffer::list::const_iterator &bl)
	{
	  int64_t kb, kb_used,kb_avail;
	  int64_t kb_used_data, kb_used_omap, kb_used_meta;
	  DECODE_START_LEGACY_COMPAT_LEN(14, 2, 2, bl);
	  decode(kb, bl);
	  decode(kb_used, bl);
	  decode(kb_avail, bl);
	  decode(snap_trim_queue_len, bl);
	  decode(num_snap_trimming, bl);
	  decode(hb_peers, bl);
	  vector<int> num_hb_out;
	  decode(num_hb_out, bl);
	  if (struct_v >= 3)
	    decode(op_queue_age_hist, bl);
	  if (struct_v >= 4)
	    decode(os_perf_stat, bl);
	  if (struct_v >= 6) {
	    decode(up_from, bl);
	    decode(seq, bl);
	  }
	  if (struct_v >= 7) {
	    decode(num_pgs, bl);
	  }
	  if (struct_v >= 8) {
	    decode(kb_used_data, bl);
	    decode(kb_used_omap, bl);
	    decode(kb_used_meta, bl);
	  } else {
	    kb_used_data = kb_used;
	    kb_used_omap = 0;
	    kb_used_meta = 0;
	  }
	  if (struct_v >= 9) {
	    decode(statfs, bl);
	  } else {
	    statfs.reset();
	    statfs.total = kb << 10;
	    statfs.available = kb_avail << 10;
	    // actually it's totally unexpected to have ststfs.total < statfs.available
	    // here but unfortunately legacy generate_test_instances produced such a
	    // case hence inserting some handling rather than assert
	    statfs.internally_reserved =
	      statfs.total > statfs.available ? statfs.total - statfs.available : 0;
	    kb_used <<= 10;
	    if ((int64_t)statfs.internally_reserved > kb_used) {
	      statfs.internally_reserved -= kb_used;
	    } else {
	      statfs.internally_reserved = 0;
	    }
	    statfs.allocated = kb_used_data << 10;
	    statfs.omap_allocated = kb_used_omap << 10;
	    statfs.internal_metadata = kb_used_meta << 10;
	  }
	  if (struct_v >= 10) {
	    decode(os_alerts, bl);
	  } else {
	    os_alerts.clear();
	  }
	  if (struct_v >= 11) {
	    decode(num_shards_repaired, bl);
	  } else {
	    num_shards_repaired = 0;
	  }
	  if (struct_v >= 12) {
	    decode(num_osds, bl);
	    decode(num_per_pool_osds, bl);
	  } else {
	    num_osds = 0;
	    num_per_pool_osds = 0;
	  }
	  if (struct_v >= 13) {
	    decode(num_per_pool_omap_osds, bl);
	  } else {
	    num_per_pool_omap_osds = 0;
	  }
	  hb_pingtime.clear();
	  if (struct_v >= 14) {
	    int count;
	    decode(count, bl);
	    for (int i = 0 ; i < count ; i++) {
	      int osd;
	      decode(osd, bl);
	      struct Interfaces ifs;
	      decode(ifs.last_update, bl);
	      decode(ifs.back_pingtime[0],bl);
	      decode(ifs.back_pingtime[1], bl);
	      decode(ifs.back_pingtime[2], bl);
	      decode(ifs.back_min[0],bl);
	      decode(ifs.back_min[1], bl);
	      decode(ifs.back_min[2], bl);
	      decode(ifs.back_max[0],bl);
	      decode(ifs.back_max[1], bl);
	      decode(ifs.back_max[2], bl);
	      decode(ifs.back_last, bl);
	      decode(ifs.front_pingtime[0], bl);
	      decode(ifs.front_pingtime[1], bl);
	      decode(ifs.front_pingtime[2], bl);
	      decode(ifs.front_min[0], bl);
	      decode(ifs.front_min[1], bl);
	      decode(ifs.front_min[2], bl);
	      decode(ifs.front_max[0], bl);
	      decode(ifs.front_max[1], bl);
	      decode(ifs.front_max[2], bl);
	      decode(ifs.front_last, bl);
	      hb_pingtime[osd] = ifs;
	    }
	  }
	  DECODE_FINISH(bl);
	}
	
	void osd_stat_t::generate_test_instances(std::list<osd_stat_t*>& o)
	{
	  o.push_back(new osd_stat_t);
	
	  o.push_back(new osd_stat_t);
	  list<store_statfs_t*> ll;
	  store_statfs_t::generate_test_instances(ll);
	  o.back()->statfs = *ll.back();
	  o.back()->hb_peers.push_back(7);
	  o.back()->snap_trim_queue_len = 8;
	  o.back()->num_snap_trimming = 99;
	  o.back()->num_shards_repaired = 101;
	  o.back()->os_alerts[0].emplace(
	    "some alert", "some alert details");
	  o.back()->os_alerts[1].emplace(
	    "some alert2", "some alert2 details");
	  struct Interfaces gen_interfaces = {
		123456789, { 1000, 900, 800 }, { 990, 890, 790 }, { 1010, 910, 810 }, 1001,
		 { 1100, 1000, 900 }, { 1090, 990, 890 }, { 1110, 1010, 910 }, 1101 };
	  o.back()->hb_pingtime[20] = gen_interfaces;
	  gen_interfaces = {
		987654321, { 100, 200, 300 }, { 90, 190, 290 }, { 110, 210, 310 }, 101 };
	  o.back()->hb_pingtime[30] = gen_interfaces;
	}
	
	// -- pg_t --
	
	int pg_t::print(char *o, int maxlen) const
	{
	  return snprintf(o, maxlen, "%llu.%x", (unsigned long long)pool(), ps());
	}
	
	bool pg_t::parse(const char *s)
	{
	  uint64_t ppool;
	  uint32_t pseed;
	  int r = sscanf(s, "%llu.%x", (long long unsigned *)&ppool, &pseed);
	  if (r < 2)
	    return false;
	  m_pool = ppool;
	  m_seed = pseed;
	  return true;
	}
	
	bool spg_t::parse(const char *s)
	{
	  shard = shard_id_t::NO_SHARD;
	  uint64_t ppool;
	  uint32_t pseed;
	  uint32_t pshard;
	  int r = sscanf(s, "%llu.%x", (long long unsigned *)&ppool, &pseed);
	  if (r < 2)
	    return false;
	  pgid.set_pool(ppool);
	  pgid.set_ps(pseed);
	
	  const char *p = strchr(s, 's');
	  if (p) {
	    r = sscanf(p, "s%u", &pshard);
	    if (r == 1) {
	      shard = shard_id_t(pshard);
	    } else {
	      return false;
	    }
	  }
	  return true;
	}
	
	char *spg_t::calc_name(char *buf, const char *suffix_backwords) const
	{
	  while (*suffix_backwords)
	    *--buf = *suffix_backwords++;
	
	  if (!is_no_shard()) {
	    buf = ritoa<uint8_t, 10>((uint8_t)shard.id, buf);
	    *--buf = 's';
	  }
	
	  return pgid.calc_name(buf, "");
	}
	
	ostream& operator<<(ostream& out, const spg_t &pg)
	{
	  char buf[spg_t::calc_name_buf_size];
	  buf[spg_t::calc_name_buf_size - 1] = '\0';
	  out << pg.calc_name(buf + spg_t::calc_name_buf_size - 1, "");
	  return out;
	}
	
	pg_t pg_t::get_ancestor(unsigned old_pg_num) const
	{
	  int old_bits = cbits(old_pg_num);
	  int old_mask = (1 << old_bits) - 1;
	  pg_t ret = *this;
	  ret.m_seed = ceph_stable_mod(m_seed, old_pg_num, old_mask);
	  return ret;
	}
	
	bool pg_t::is_split(unsigned old_pg_num, unsigned new_pg_num, set<pg_t> *children) const
	{
	  //ceph_assert(m_seed < old_pg_num);
	  if (m_seed >= old_pg_num) {
	    // degenerate case
	    return false;
	  }
	  if (new_pg_num <= old_pg_num)
	    return false;
	
	  bool split = false;
	  if (true) {
	    unsigned old_bits = cbits(old_pg_num);
	    unsigned old_mask = (1 << old_bits) - 1;
	    for (unsigned n = 1; ; n++) {
	      unsigned next_bit = (n << (old_bits-1));
	      unsigned s = next_bit | m_seed;
	
	      if (s < old_pg_num || s == m_seed)
		continue;
	      if (s >= new_pg_num)
		break;
	      if ((unsigned)ceph_stable_mod(s, old_pg_num, old_mask) == m_seed) {
		split = true;
		if (children)
		  children->insert(pg_t(s, m_pool));
	      }
	    }
	  }
	  if (false) {
	    // brute force
	    int old_bits = cbits(old_pg_num);
	    int old_mask = (1 << old_bits) - 1;
	    for (unsigned x = old_pg_num; x < new_pg_num; ++x) {
	      unsigned o = ceph_stable_mod(x, old_pg_num, old_mask);
	      if (o == m_seed) {
		split = true;
		children->insert(pg_t(x, m_pool));
	      }
	    }
	  }
	  return split;
	}
	
	unsigned pg_t::get_split_bits(unsigned pg_num) const {
	  if (pg_num == 1)
	    return 0;
	  ceph_assert(pg_num > 1);
	
	  // Find unique p such that pg_num \in [2^(p-1), 2^p)
	  unsigned p = cbits(pg_num);
	  ceph_assert(p); // silence coverity #751330 
	
	  if ((m_seed % (1<<(p-1))) < (pg_num % (1<<(p-1))))
	    return p;
	  else
	    return p - 1;
	}
	
	bool pg_t::is_merge_source(
	  unsigned old_pg_num,
	  unsigned new_pg_num,
	  pg_t *parent) const
	{
	  if (m_seed < old_pg_num &&
	      m_seed >= new_pg_num) {
	    if (parent) {
	      pg_t t = *this;
	      while (t.m_seed >= new_pg_num) {
		t = t.get_parent();
	      }
	      *parent = t;
	    }
	    return true;
	  }
	  return false;
	}
	
	pg_t pg_t::get_parent() const
	{
	  unsigned bits = cbits(m_seed);
	  ceph_assert(bits);
	  pg_t retval = *this;
	  retval.m_seed &= ~((~0)<<(bits - 1));
	  return retval;
	}
	
	hobject_t pg_t::get_hobj_start() const
	{
	  return hobject_t(object_t(), string(), 0, m_seed, m_pool,
			   string());
	}
	
	hobject_t pg_t::get_hobj_end(unsigned pg_num) const
	{
	  // note: this assumes a bitwise sort; with the legacy nibblewise
	  // sort a PG did not always cover a single contiguous range of the
	  // (bit-reversed) hash range.
	  unsigned bits = get_split_bits(pg_num);
	  uint64_t rev_start = hobject_t::_reverse_bits(m_seed);
	  uint64_t rev_end = (rev_start | (0xffffffff >> bits)) + 1;
	  if (rev_end >= 0x100000000) {
	    ceph_assert(rev_end == 0x100000000);
	    return hobject_t::get_max();
	  } else {
	    return hobject_t(object_t(), string(), CEPH_NOSNAP,
			   hobject_t::_reverse_bits(rev_end), m_pool,
			   string());
	  }
	}
	
	void pg_t::dump(Formatter *f) const
	{
	  f->dump_unsigned("pool", m_pool);
	  f->dump_unsigned("seed", m_seed);
	}
	
	void pg_t::generate_test_instances(list<pg_t*>& o)
	{
	  o.push_back(new pg_t);
	  o.push_back(new pg_t(1, 2));
	  o.push_back(new pg_t(13123, 3));
	  o.push_back(new pg_t(131223, 4));
	}
	
	char *pg_t::calc_name(char *buf, const char *suffix_backwords) const
	{
	  while (*suffix_backwords)
	    *--buf = *suffix_backwords++;
	
	  buf = ritoa<uint32_t, 16>(m_seed, buf);
	
	  *--buf = '.';
	
	  return  ritoa<uint64_t, 10>(m_pool, buf);
	}
	
	ostream& operator<<(ostream& out, const pg_t &pg)
	{
	  char buf[pg_t::calc_name_buf_size];
	  buf[pg_t::calc_name_buf_size - 1] = '\0';
	  out << pg.calc_name(buf + pg_t::calc_name_buf_size - 1, "");
	  return out;
	}
	
	
	// -- coll_t --
	
	void coll_t::calc_str()
	{
	  switch (type) {
	  case TYPE_META:
	    strcpy(_str_buff, "meta");
	    _str = _str_buff;
	    break;
	  case TYPE_PG:
	    _str_buff[spg_t::calc_name_buf_size - 1] = '\0';
	    _str = pgid.calc_name(_str_buff + spg_t::calc_name_buf_size - 1, "daeh_");
	    break;
	  case TYPE_PG_TEMP:
	    _str_buff[spg_t::calc_name_buf_size - 1] = '\0';
	    _str = pgid.calc_name(_str_buff + spg_t::calc_name_buf_size - 1, "PMET_");
	    break;
	  default:
	    ceph_abort_msg("unknown collection type");
	  }
	}
	
	bool coll_t::parse(const std::string& s)
	{
	  if (s == "meta") {
	    type = TYPE_META;
	    pgid = spg_t();
	    removal_seq = 0;
	    calc_str();
	    ceph_assert(s == _str);
	    return true;
	  }
	  if (s.find("_head") == s.length() - 5 &&
	      pgid.parse(s.substr(0, s.length() - 5))) {
	    type = TYPE_PG;
	    removal_seq = 0;
	    calc_str();
	    ceph_assert(s == _str);
	    return true;
	  }
	  if (s.find("_TEMP") == s.length() - 5 &&
	      pgid.parse(s.substr(0, s.length() - 5))) {
	    type = TYPE_PG_TEMP;
	    removal_seq = 0;
	    calc_str();
	    ceph_assert(s == _str);
	    return true;
	  }
	  return false;
	}
	
	void coll_t::encode(ceph::buffer::list& bl) const
	{
	  using ceph::encode;
	  // when changing this, remember to update encoded_size() too.
	  if (is_temp()) {
	    // can't express this as v2...
	    __u8 struct_v = 3;
	    encode(struct_v, bl);
	    encode(to_str(), bl);
	  } else {
	    __u8 struct_v = 2;
	    encode(struct_v, bl);
	    encode((__u8)type, bl);
	    encode(pgid, bl);
	    snapid_t snap = CEPH_NOSNAP;
	    encode(snap, bl);
	  }
	}
	
	size_t coll_t::encoded_size() const
	{
	  size_t r = sizeof(__u8);
	  if (is_temp()) {
	    // v3
	    r += sizeof(__u32);
	    if (_str) {
	      r += strlen(_str);
	    }
	  } else {
	      // v2
	      // 1. type
	      r += sizeof(__u8);
	      // 2. pgid
	      //  - encoding header
	      r += sizeof(ceph_le32) + 2 * sizeof(__u8);
	      // - pg_t
	      r += sizeof(__u8) + sizeof(uint64_t) + 2 * sizeof(uint32_t);
	      // - shard_id_t
	      r += sizeof(int8_t);
	      // 3. snapid_t
	      r += sizeof(uint64_t);
	  }
	
	  return r;
	}
	
	void coll_t::decode(ceph::buffer::list::const_iterator& bl)
	{
	  using ceph::decode;
	  __u8 struct_v;
	  decode(struct_v, bl);
	  switch (struct_v) {
	  case 1:
	    {
	      snapid_t snap;
	      decode(pgid, bl);
	      decode(snap, bl);
	
	      // infer the type
	      if (pgid == spg_t() && snap == 0) {
		type = TYPE_META;
	      } else {
		type = TYPE_PG;
	      }
	      removal_seq = 0;
	    }
	    break;
	
	  case 2:
	    {
	      __u8 _type;
	      snapid_t snap;
	      decode(_type, bl);
	      decode(pgid, bl);
	      decode(snap, bl);
	      type = (type_t)_type;
	      removal_seq = 0;
	    }
	    break;
	
	  case 3:
	    {
	      string str;
	      decode(str, bl);
	      bool ok = parse(str);
	      if (!ok)
		throw std::domain_error(std::string("unable to parse pg ") + str);
	    }
	    break;
	
	  default:
	    {
	      ostringstream oss;
	      oss << "coll_t::decode(): don't know how to decode version "
		  << struct_v;
	      throw std::domain_error(oss.str());
	    }
	  }
	}
	
	void coll_t::dump(Formatter *f) const
	{
	  f->dump_unsigned("type_id", (unsigned)type);
	  if (type != TYPE_META)
	    f->dump_stream("pgid") << pgid;
	  f->dump_string("name", to_str());
	}
	
	void coll_t::generate_test_instances(list<coll_t*>& o)
	{
	  o.push_back(new coll_t());
	  o.push_back(new coll_t(spg_t(pg_t(1, 0), shard_id_t::NO_SHARD)));
	  o.push_back(new coll_t(o.back()->get_temp()));
	  o.push_back(new coll_t(spg_t(pg_t(3, 2), shard_id_t(12))));
	  o.push_back(new coll_t(o.back()->get_temp()));
	  o.push_back(new coll_t());
	}
	
	// ---
	
	std::string pg_vector_string(const vector<int32_t> &a)
	{
	  ostringstream oss;
	  oss << "[";
	  for (auto i = a.cbegin(); i != a.cend(); ++i) {
	    if (i != a.begin())
	      oss << ",";
	    if (*i != CRUSH_ITEM_NONE)
	      oss << *i;
	    else
	      oss << "NONE";
	  }
	  oss << "]";
	  return oss.str();
	}
	
	std::string pg_state_string(uint64_t state)
	{
	  ostringstream oss;
	  if (state & PG_STATE_STALE)
	    oss << "stale+";
	  if (state & PG_STATE_CREATING)
	    oss << "creating+";
	  if (state & PG_STATE_ACTIVE)
	    oss << "active+";
	  if (state & PG_STATE_ACTIVATING)
	    oss << "activating+";
	  if (state & PG_STATE_CLEAN)
	    oss << "clean+";
	  if (state & PG_STATE_RECOVERY_WAIT)
	    oss << "recovery_wait+";
	  if (state & PG_STATE_RECOVERY_TOOFULL)
	    oss << "recovery_toofull+";
	  if (state & PG_STATE_RECOVERING)
	    oss << "recovering+";
	  if (state & PG_STATE_FORCED_RECOVERY)
	    oss << "forced_recovery+";
	  if (state & PG_STATE_DOWN)
	    oss << "down+";
	  if (state & PG_STATE_RECOVERY_UNFOUND)
	    oss << "recovery_unfound+";
	  if (state & PG_STATE_BACKFILL_UNFOUND)
	    oss << "backfill_unfound+";
	  if (state & PG_STATE_UNDERSIZED)
	    oss << "undersized+";
	  if (state & PG_STATE_DEGRADED)
	    oss << "degraded+";
	  if (state & PG_STATE_REMAPPED)
	    oss << "remapped+";
	  if (state & PG_STATE_PREMERGE)
	    oss << "premerge+";
	  if (state & PG_STATE_SCRUBBING)
	    oss << "scrubbing+";
	  if (state & PG_STATE_DEEP_SCRUB)
	    oss << "deep+";
	  if (state & PG_STATE_INCONSISTENT)
	    oss << "inconsistent+";
	  if (state & PG_STATE_PEERING)
	    oss << "peering+";
	  if (state & PG_STATE_REPAIR)
	    oss << "repair+";
	  if (state & PG_STATE_BACKFILL_WAIT)
	    oss << "backfill_wait+";
	  if (state & PG_STATE_BACKFILLING)
	    oss << "backfilling+";
	  if (state & PG_STATE_FORCED_BACKFILL)
	    oss << "forced_backfill+";
	  if (state & PG_STATE_BACKFILL_TOOFULL)
	    oss << "backfill_toofull+";
	  if (state & PG_STATE_INCOMPLETE)
	    oss << "incomplete+";
	  if (state & PG_STATE_PEERED)
	    oss << "peered+";
	  if (state & PG_STATE_SNAPTRIM)
	    oss << "snaptrim+";
	  if (state & PG_STATE_SNAPTRIM_WAIT)
	    oss << "snaptrim_wait+";
	  if (state & PG_STATE_SNAPTRIM_ERROR)
	    oss << "snaptrim_error+";
	  if (state & PG_STATE_FAILED_REPAIR)
	    oss << "failed_repair+";
	  if (state & PG_STATE_LAGGY)
	    oss << "laggy+";
	  if (state & PG_STATE_WAIT)
	    oss << "wait+";
	  string ret(oss.str());
	  if (ret.length() > 0)
	    ret.resize(ret.length() - 1);
	  else
	    ret = "unknown";
	  return ret;
	}
	
	std::optional<uint64_t> pg_string_state(const std::string& state)
	{
	  std::optional<uint64_t> type;
	  if (state == "active")
	    type = PG_STATE_ACTIVE;
	  else if (state == "clean")
	    type = PG_STATE_CLEAN;
	  else if (state == "down")
	    type = PG_STATE_DOWN;
	  else if (state == "recovery_unfound")
	    type = PG_STATE_RECOVERY_UNFOUND;
	  else if (state == "backfill_unfound")
	    type = PG_STATE_BACKFILL_UNFOUND;
	  else if (state == "premerge")
	    type = PG_STATE_PREMERGE;
	  else if (state == "scrubbing")
	    type = PG_STATE_SCRUBBING;
	  else if (state == "degraded")
	    type = PG_STATE_DEGRADED;
	  else if (state == "inconsistent")
	    type = PG_STATE_INCONSISTENT;
	  else if (state == "peering")
	    type = PG_STATE_PEERING;
	  else if (state == "repair")
	    type = PG_STATE_REPAIR;
	  else if (state == "recovering")
	    type = PG_STATE_RECOVERING;
	  else if (state == "forced_recovery")
	    type = PG_STATE_FORCED_RECOVERY;
	  else if (state == "backfill_wait")
	    type = PG_STATE_BACKFILL_WAIT;
	  else if (state == "incomplete")
	    type = PG_STATE_INCOMPLETE;
	  else if (state == "stale")
	    type = PG_STATE_STALE;
	  else if (state == "remapped")
	    type = PG_STATE_REMAPPED;
	  else if (state == "deep")
	    type = PG_STATE_DEEP_SCRUB;
	  else if (state == "backfilling")
	    type = PG_STATE_BACKFILLING;
	  else if (state == "forced_backfill")
	    type = PG_STATE_FORCED_BACKFILL;
	  else if (state == "backfill_toofull")
	    type = PG_STATE_BACKFILL_TOOFULL;
	  else if (state == "recovery_wait")
	    type = PG_STATE_RECOVERY_WAIT;
	  else if (state == "recovery_toofull")
	    type = PG_STATE_RECOVERY_TOOFULL;
	  else if (state == "undersized")
	    type = PG_STATE_UNDERSIZED;
	  else if (state == "activating")
	    type = PG_STATE_ACTIVATING;
	  else if (state == "peered")
	    type = PG_STATE_PEERED;
	  else if (state == "snaptrim")
	    type = PG_STATE_SNAPTRIM;
	  else if (state == "snaptrim_wait")
	    type = PG_STATE_SNAPTRIM_WAIT;
	  else if (state == "snaptrim_error")
	    type = PG_STATE_SNAPTRIM_ERROR;
	  else if (state == "creating")
	    type = PG_STATE_CREATING;
	  else if (state == "failed_repair")
	    type = PG_STATE_FAILED_REPAIR;
	  else if (state == "laggy")
	    type = PG_STATE_LAGGY;
	  else if (state == "wait")
	    type = PG_STATE_WAIT;
	  else if (state == "unknown")
	    type = 0;
	  else
	    type = std::nullopt;
	  return type;
	}
	
	// -- eversion_t --
	string eversion_t::get_key_name() const
	{
	  std::string key(32, ' ');
	  get_key_name(&key[0]);
	  key.resize(31); // remove the null terminator
	  return key;
	}
	
	// -- pool_snap_info_t --
	void pool_snap_info_t::dump(Formatter *f) const
	{
	  f->dump_unsigned("snapid", snapid);
	  f->dump_stream("stamp") << stamp;
	  f->dump_string("name", name);
	}
	
	void pool_snap_info_t::encode(ceph::buffer::list& bl, uint64_t features) const
	{
	  using ceph::encode;
	  if ((features & CEPH_FEATURE_PGPOOL3) == 0) {
	    __u8 struct_v = 1;
	    encode(struct_v, bl);
	    encode(snapid, bl);
	    encode(stamp, bl);
	    encode(name, bl);
	    return;
	  }
	  ENCODE_START(2, 2, bl);
	  encode(snapid, bl);
	  encode(stamp, bl);
	  encode(name, bl);
	  ENCODE_FINISH(bl);
	}
	
	void pool_snap_info_t::decode(ceph::buffer::list::const_iterator& bl)
	{
	  DECODE_START_LEGACY_COMPAT_LEN(2, 2, 2, bl);
	  decode(snapid, bl);
	  decode(stamp, bl);
	  decode(name, bl);
	  DECODE_FINISH(bl);
	}
	
	void pool_snap_info_t::generate_test_instances(list<pool_snap_info_t*>& o)
	{
	  o.push_back(new pool_snap_info_t);
	  o.push_back(new pool_snap_info_t);
	  o.back()->snapid = 1;
	  o.back()->stamp = utime_t(1, 2);
	  o.back()->name = "foo";
	}
	
	// -- pool_opts_t --
	
	typedef std::map<std::string, pool_opts_t::opt_desc_t> opt_mapping_t;
	static opt_mapping_t opt_mapping = boost::assign::map_list_of
		   ("scrub_min_interval", pool_opts_t::opt_desc_t(
		     pool_opts_t::SCRUB_MIN_INTERVAL, pool_opts_t::DOUBLE))
		   ("scrub_max_interval", pool_opts_t::opt_desc_t(
		     pool_opts_t::SCRUB_MAX_INTERVAL, pool_opts_t::DOUBLE))
		   ("deep_scrub_interval", pool_opts_t::opt_desc_t(
		     pool_opts_t::DEEP_SCRUB_INTERVAL, pool_opts_t::DOUBLE))
	           ("recovery_priority", pool_opts_t::opt_desc_t(
	             pool_opts_t::RECOVERY_PRIORITY, pool_opts_t::INT))
	           ("recovery_op_priority", pool_opts_t::opt_desc_t(
	             pool_opts_t::RECOVERY_OP_PRIORITY, pool_opts_t::INT))
	           ("scrub_priority", pool_opts_t::opt_desc_t(
	             pool_opts_t::SCRUB_PRIORITY, pool_opts_t::INT))
	           ("compression_mode", pool_opts_t::opt_desc_t(
		     pool_opts_t::COMPRESSION_MODE, pool_opts_t::STR))
	           ("compression_algorithm", pool_opts_t::opt_desc_t(
		     pool_opts_t::COMPRESSION_ALGORITHM, pool_opts_t::STR))
	           ("compression_required_ratio", pool_opts_t::opt_desc_t(
		     pool_opts_t::COMPRESSION_REQUIRED_RATIO, pool_opts_t::DOUBLE))
	           ("compression_max_blob_size", pool_opts_t::opt_desc_t(
		     pool_opts_t::COMPRESSION_MAX_BLOB_SIZE, pool_opts_t::INT))
	           ("compression_min_blob_size", pool_opts_t::opt_desc_t(
		     pool_opts_t::COMPRESSION_MIN_BLOB_SIZE, pool_opts_t::INT))
	           ("csum_type", pool_opts_t::opt_desc_t(
		     pool_opts_t::CSUM_TYPE, pool_opts_t::INT))
	           ("csum_max_block", pool_opts_t::opt_desc_t(
		     pool_opts_t::CSUM_MAX_BLOCK, pool_opts_t::INT))
	           ("csum_min_block", pool_opts_t::opt_desc_t(
		     pool_opts_t::CSUM_MIN_BLOCK, pool_opts_t::INT))
	           ("fingerprint_algorithm", pool_opts_t::opt_desc_t(
		     pool_opts_t::FINGERPRINT_ALGORITHM, pool_opts_t::STR))
	           ("pg_num_min", pool_opts_t::opt_desc_t(
		     pool_opts_t::PG_NUM_MIN, pool_opts_t::INT))
	           ("target_size_bytes", pool_opts_t::opt_desc_t(
		     pool_opts_t::TARGET_SIZE_BYTES, pool_opts_t::INT))
	           ("target_size_ratio", pool_opts_t::opt_desc_t(
		     pool_opts_t::TARGET_SIZE_RATIO, pool_opts_t::DOUBLE))
	           ("pg_autoscale_bias", pool_opts_t::opt_desc_t(
		     pool_opts_t::PG_AUTOSCALE_BIAS, pool_opts_t::DOUBLE))
	           ("read_lease_interval", pool_opts_t::opt_desc_t(
		     pool_opts_t::READ_LEASE_INTERVAL, pool_opts_t::DOUBLE));
	
	bool pool_opts_t::is_opt_name(const std::string& name)
	{
	  return opt_mapping.count(name);
	}
	
	pool_opts_t::opt_desc_t pool_opts_t::get_opt_desc(const std::string& name)
	{
	  auto i = opt_mapping.find(name);
	  ceph_assert(i != opt_mapping.end());
	  return i->second;
	}
	
	bool pool_opts_t::is_set(pool_opts_t::key_t key) const
	{
	  return opts.count(key);
	}
	
	const pool_opts_t::value_t& pool_opts_t::get(pool_opts_t::key_t key) const
	{
	  auto i = opts.find(key);
	  ceph_assert(i != opts.end());
	  return i->second;
	}
	
	bool pool_opts_t::unset(pool_opts_t::key_t key) {
	  return opts.erase(key) > 0;
	}
	
	class pool_opts_dumper_t : public boost::static_visitor<> {
	public:
	  pool_opts_dumper_t(const std::string& name_, Formatter* f_) :
	    name(name_.c_str()), f(f_) {}
	
	  void operator()(std::string s) const {
	    f->dump_string(name, s);
	  }
	  void operator()(int64_t i) const {
	    f->dump_int(name, i);
	  }
	  void operator()(double d) const {
	    f->dump_float(name, d);
	  }
	
	private:
	  const char* name;
	  Formatter* f;
	};
	
	void pool_opts_t::dump(const std::string& name, Formatter* f) const
	{
	  const opt_desc_t& desc = get_opt_desc(name);
	  auto i = opts.find(desc.key);
	  if (i == opts.end()) {
	      return;
	  }
	  boost::apply_visitor(pool_opts_dumper_t(name, f), i->second);
	}
	
	void pool_opts_t::dump(Formatter* f) const
	{
	  for (auto i = opt_mapping.cbegin(); i != opt_mapping.cend(); ++i) {
	    const std::string& name = i->first;
	    const opt_desc_t& desc = i->second;
	    auto j = opts.find(desc.key);
	    if (j == opts.end()) {
	      continue;
	    }
	    boost::apply_visitor(pool_opts_dumper_t(name, f), j->second);
	  }
	}
	
	class pool_opts_encoder_t : public boost::static_visitor<> {
	public:
	  explicit pool_opts_encoder_t(ceph::buffer::list& bl_, uint64_t features)
	    : bl(bl_),
	      features(features) {}
	
	  void operator()(const std::string &s) const {
	    encode(static_cast<int32_t>(pool_opts_t::STR), bl);
	    encode(s, bl);
	  }
	  void operator()(int64_t i) const {
	    encode(static_cast<int32_t>(pool_opts_t::INT), bl);
	    if (HAVE_FEATURE(features, SERVER_NAUTILUS)) {
	      encode(i, bl);
	    } else {
	      encode(static_cast<int32_t>(i), bl);
	    }
	  }
	  void operator()(double d) const {
	    encode(static_cast<int32_t>(pool_opts_t::DOUBLE), bl);
	    encode(d, bl);
	  }
	
	private:
	  ceph::buffer::list& bl;
	  uint64_t features;
	};
	
	void pool_opts_t::encode(ceph::buffer::list& bl, uint64_t features) const
	{
	  unsigned v = 2;
	  if (!HAVE_FEATURE(features, SERVER_NAUTILUS)) {
	    v = 1;
	  }
	  ENCODE_START(v, 1, bl);
	  uint32_t n = static_cast<uint32_t>(opts.size());
	  encode(n, bl);
	  for (auto i = opts.cbegin(); i != opts.cend(); ++i) {
	    encode(static_cast<int32_t>(i->first), bl);
	    boost::apply_visitor(pool_opts_encoder_t(bl, features), i->second);
	  }
	  ENCODE_FINISH(bl);
	}
	
	void pool_opts_t::decode(ceph::buffer::list::const_iterator& bl)
	{
	  DECODE_START(1, bl);
	  __u32 n;
	  decode(n, bl);
	  opts.clear();
	  while (n--) {
	    int32_t k, t;
	    decode(k, bl);
	    decode(t, bl);
	    if (t == STR) {
	      std::string s;
	      decode(s, bl);
	      opts[static_cast<key_t>(k)] = s;
	    } else if (t == INT) {
	      int64_t i;
	      if (struct_v >= 2) {
		decode(i, bl);
	      } else {
		int ii;
		decode(ii, bl);
		i = ii;
	      }
	      opts[static_cast<key_t>(k)] = i;
	    } else if (t == DOUBLE) {
	      double d;
	      decode(d, bl);
	      opts[static_cast<key_t>(k)] = d;
	    } else {
	      ceph_assert(!"invalid type");
	    }
	  }
	  DECODE_FINISH(bl);
	}
	
	ostream& operator<<(ostream& out, const pool_opts_t& opts)
	{
	  for (auto i = opt_mapping.begin(); i != opt_mapping.end(); ++i) {
	    const std::string& name = i->first;
	    const pool_opts_t::opt_desc_t& desc = i->second;
	    auto j = opts.opts.find(desc.key);
	    if (j == opts.opts.end()) {
	      continue;
	    }
	    out << " " << name << " " << j->second;
	  }
	  return out;
	}
	
	// -- pg_pool_t --
	
	const char *pg_pool_t::APPLICATION_NAME_CEPHFS("cephfs");
	const char *pg_pool_t::APPLICATION_NAME_RBD("rbd");
	const char *pg_pool_t::APPLICATION_NAME_RGW("rgw");
	
	void pg_pool_t::dump(Formatter *f) const
	{
	  f->dump_stream("create_time") << get_create_time();
	  f->dump_unsigned("flags", get_flags());
	  f->dump_string("flags_names", get_flags_string());
	  f->dump_int("type", get_type());
	  f->dump_int("size", get_size());
	  f->dump_int("min_size", get_min_size());
	  f->dump_int("crush_rule", get_crush_rule());
	  f->dump_int("object_hash", get_object_hash());
	  f->dump_string("pg_autoscale_mode",
			 get_pg_autoscale_mode_name(pg_autoscale_mode));
	  f->dump_unsigned("pg_num", get_pg_num());
	  f->dump_unsigned("pg_placement_num", get_pgp_num());
	  f->dump_unsigned("pg_placement_num_target", get_pgp_num_target());
	  f->dump_unsigned("pg_num_target", get_pg_num_target());
	  f->dump_unsigned("pg_num_pending", get_pg_num_pending());
	  f->dump_object("last_pg_merge_meta", last_pg_merge_meta);
	  f->dump_stream("last_change") << get_last_change();
	  f->dump_stream("last_force_op_resend") << get_last_force_op_resend();
	  f->dump_stream("last_force_op_resend_prenautilus")
	    << get_last_force_op_resend_prenautilus();
	  f->dump_stream("last_force_op_resend_preluminous")
	    << get_last_force_op_resend_preluminous();
	  f->dump_unsigned("auid", get_auid());
	  f->dump_string("snap_mode", is_pool_snaps_mode() ? "pool" : "selfmanaged");
	  f->dump_unsigned("snap_seq", get_snap_seq());
	  f->dump_unsigned("snap_epoch", get_snap_epoch());
	  f->open_array_section("pool_snaps");
	  for (auto p = snaps.cbegin(); p != snaps.cend(); ++p) {
	    f->open_object_section("pool_snap_info");
	    p->second.dump(f);
	    f->close_section();
	  }
	  f->close_section();
	  f->dump_stream("removed_snaps") << removed_snaps;
	  f->dump_unsigned("quota_max_bytes", quota_max_bytes);
	  f->dump_unsigned("quota_max_objects", quota_max_objects);
	  f->open_array_section("tiers");
	  for (auto p = tiers.cbegin(); p != tiers.cend(); ++p)
	    f->dump_unsigned("pool_id", *p);
	  f->close_section();
	  f->dump_int("tier_of", tier_of);
	  f->dump_int("read_tier", read_tier);
	  f->dump_int("write_tier", write_tier);
	  f->dump_string("cache_mode", get_cache_mode_name());
	  f->dump_unsigned("target_max_bytes", target_max_bytes);
	  f->dump_unsigned("target_max_objects", target_max_objects);
	  f->dump_unsigned("cache_target_dirty_ratio_micro",
			   cache_target_dirty_ratio_micro);
	  f->dump_unsigned("cache_target_dirty_high_ratio_micro",
			   cache_target_dirty_high_ratio_micro);
	  f->dump_unsigned("cache_target_full_ratio_micro",
			   cache_target_full_ratio_micro);
	  f->dump_unsigned("cache_min_flush_age", cache_min_flush_age);
	  f->dump_unsigned("cache_min_evict_age", cache_min_evict_age);
	  f->dump_string("erasure_code_profile", erasure_code_profile);
	  f->open_object_section("hit_set_params");
	  hit_set_params.dump(f);
	  f->close_section(); // hit_set_params
	  f->dump_unsigned("hit_set_period", hit_set_period);
	  f->dump_unsigned("hit_set_count", hit_set_count);
	  f->dump_bool("use_gmt_hitset", use_gmt_hitset);
	  f->dump_unsigned("min_read_recency_for_promote", min_read_recency_for_promote);
	  f->dump_unsigned("min_write_recency_for_promote", min_write_recency_for_promote);
	  f->dump_unsigned("hit_set_grade_decay_rate", hit_set_grade_decay_rate);
	  f->dump_unsigned("hit_set_search_last_n", hit_set_search_last_n);
	  f->open_array_section("grade_table");
	  for (unsigned i = 0; i < hit_set_count; ++i)
	    f->dump_unsigned("value", get_grade(i));
	  f->close_section();
	  f->dump_unsigned("stripe_width", get_stripe_width());
	  f->dump_unsigned("expected_num_objects", expected_num_objects);
	  f->dump_bool("fast_read", fast_read);
	  f->open_object_section("options");
	  opts.dump(f);
	  f->close_section(); // options
	  f->open_object_section("application_metadata");
	  for (auto &app_pair : application_metadata) {
	    f->open_object_section(app_pair.first.c_str());
	    for (auto &kv_pair : app_pair.second) {
	      f->dump_string(kv_pair.first.c_str(), kv_pair.second);
	    }
	    f->close_section(); // application
	  }
	  f->close_section(); // application_metadata
	}
	
	void pg_pool_t::convert_to_pg_shards(const vector<int> &from, set<pg_shard_t>* to) const {
	  for (size_t i = 0; i < from.size(); ++i) {
	    if (from[i] != CRUSH_ITEM_NONE) {
	      to->insert(
	        pg_shard_t(
	          from[i],
	          is_erasure() ? shard_id_t(i) : shard_id_t::NO_SHARD));
	    }
	  }
	}
	
	void pg_pool_t::calc_pg_masks()
	{
	  pg_num_mask = (1 << cbits(pg_num-1)) - 1;
	  pgp_num_mask = (1 << cbits(pgp_num-1)) - 1;
	}
	
	unsigned pg_pool_t::get_pg_num_divisor(pg_t pgid) const
	{
	  if (pg_num == pg_num_mask + 1)
	    return pg_num;                    // power-of-2 split
	  unsigned mask = pg_num_mask >> 1;
	  if ((pgid.ps() & mask) < (pg_num & mask))
	    return pg_num_mask + 1;           // smaller bin size (already split)
	  else
	    return (pg_num_mask + 1) >> 1;    // bigger bin (not yet split)
	}
	
	bool pg_pool_t::is_pending_merge(pg_t pgid, bool *target) const
	{
	  if (pg_num_pending >= pg_num) {
	    return false;
	  }
	  if (pgid.ps() >= pg_num_pending && pgid.ps() < pg_num) {
	    if (target) {
	      *target = false;
	    }
	    return true;
	  }
	  for (unsigned ps = pg_num_pending; ps < pg_num; ++ps) {
	    if (pg_t(ps, pgid.pool()).get_parent() == pgid) {
	      if (target) {
		*target = true;
	      }
	      return true;
	    }
	  }
	  return false;
	}
	
	/*
	 * we have two snap modes:
	 *  - pool snaps
	 *    - snap existence/non-existence defined by snaps[] and snap_seq
	 *  - user managed snaps
	 *    - existence tracked by librados user
	 */
	bool pg_pool_t::is_pool_snaps_mode() const
	{
	  return has_flag(FLAG_POOL_SNAPS);
	}
	
	bool pg_pool_t::is_unmanaged_snaps_mode() const
	{
	  return has_flag(FLAG_SELFMANAGED_SNAPS);
	}
	
	bool pg_pool_t::is_removed_snap(snapid_t s) const
	{
	  if (is_pool_snaps_mode())
	    return s <= get_snap_seq() && snaps.count(s) == 0;
	  else
	    return removed_snaps.contains(s);
	}
	
	snapid_t pg_pool_t::snap_exists(const char *s) const
	{
	  for (auto p = snaps.cbegin(); p != snaps.cend(); ++p)
	    if (p->second.name == s)
	      return p->second.snapid;
	  return 0;
	}
	
	void pg_pool_t::add_snap(const char *n, utime_t stamp)
	{
	  ceph_assert(!is_unmanaged_snaps_mode());
	  flags |= FLAG_POOL_SNAPS;
	  snapid_t s = get_snap_seq() + 1;
	  snap_seq = s;
	  snaps[s].snapid = s;
	  snaps[s].name = n;
	  snaps[s].stamp = stamp;
	}
	
	uint64_t pg_pool_t::add_unmanaged_snap(bool preoctopus_compat)
	{
	  ceph_assert(!is_pool_snaps_mode());
	  if (snap_seq == 0) {
	    if (preoctopus_compat) {
	      // kludge for pre-mimic tracking of pool vs selfmanaged snaps.  after
	      // mimic this field is not decoded but our flag is set; pre-mimic, we
	      // have a non-empty removed_snaps to signifiy a non-pool-snaps pool.
	      removed_snaps.insert(snapid_t(1));
	    }
	    snap_seq = 1;
	  }
	  flags |= FLAG_SELFMANAGED_SNAPS;
	  snap_seq = snap_seq + 1;
	  return snap_seq;
	}
	
	void pg_pool_t::remove_snap(snapid_t s)
	{
	  ceph_assert(snaps.count(s));
	  snaps.erase(s);
	  snap_seq = snap_seq + 1;
	}
	
	void pg_pool_t::remove_unmanaged_snap(snapid_t s, bool preoctopus_compat)
	{
	  ceph_assert(is_unmanaged_snaps_mode());
	  ++snap_seq;
	  if (preoctopus_compat) {
	    removed_snaps.insert(s);
	    // try to add in the new seq, just to try to keep the interval_set contiguous
	    if (!removed_snaps.contains(get_snap_seq())) {
	      removed_snaps.insert(get_snap_seq());
	    }
	  }
	}
	
	SnapContext pg_pool_t::get_snap_context() const
	{
	  vector<snapid_t> s(snaps.size());
	  unsigned i = 0;
	  for (auto p = snaps.crbegin(); p != snaps.crend(); ++p)
	    s[i++] = p->first;
	  return SnapContext(get_snap_seq(), s);
	}
	
	uint32_t pg_pool_t::hash_key(const string& key, const string& ns) const
	{
	 if (ns.empty()) 
	    return ceph_str_hash(object_hash, key.data(), key.length());
	  int nsl = ns.length();
	  int len = key.length() + nsl + 1;
	  char buf[len];
	  memcpy(&buf[0], ns.data(), nsl);
	  buf[nsl] = '\037';
	  memcpy(&buf[nsl+1], key.data(), key.length());
	  return ceph_str_hash(object_hash, &buf[0], len);
	}
	
	uint32_t pg_pool_t::raw_hash_to_pg(uint32_t v) const
	{
	  return ceph_stable_mod(v, pg_num, pg_num_mask);
	}
	
	/*
	 * map a raw pg (with full precision ps) into an actual pg, for storage
	 */
	pg_t pg_pool_t::raw_pg_to_pg(pg_t pg) const
	{
	  pg.set_ps(ceph_stable_mod(pg.ps(), pg_num, pg_num_mask));
	  return pg;
	}
	  
	/*
	 * map raw pg (full precision ps) into a placement seed.  include
	 * pool id in that value so that different pools don't use the same
	 * seeds.
	 */
	ps_t pg_pool_t::raw_pg_to_pps(pg_t pg) const
	{
	  if (flags & FLAG_HASHPSPOOL) {
	    // Hash the pool id so that pool PGs do not overlap.
	    return
	      crush_hash32_2(CRUSH_HASH_RJENKINS1,
			     ceph_stable_mod(pg.ps(), pgp_num, pgp_num_mask),
			     pg.pool());
	  } else {
	    // Legacy behavior; add ps and pool together.  This is not a great
	    // idea because the PGs from each pool will essentially overlap on
	    // top of each other: 0.5 == 1.4 == 2.3 == ...
	    return
	      ceph_stable_mod(pg.ps(), pgp_num, pgp_num_mask) +
	      pg.pool();
	  }
	}
	
	uint32_t pg_pool_t::get_random_pg_position(pg_t pg, uint32_t seed) const
	{
	  uint32_t r = crush_hash32_2(CRUSH_HASH_RJENKINS1, seed, 123);
	  if (pg_num == pg_num_mask + 1) {
	    r &= ~pg_num_mask;
	  } else {
	    unsigned smaller_mask = pg_num_mask >> 1;
	    if ((pg.ps() & smaller_mask) < (pg_num & smaller_mask)) {
	      r &= ~pg_num_mask;
	    } else {
	      r &= ~smaller_mask;
	    }
	  }
	  r |= pg.ps();
	  return r;
	}
	
	void pg_pool_t::encode(ceph::buffer::list& bl, uint64_t features) const
	{
	  using ceph::encode;
	  if ((features & CEPH_FEATURE_PGPOOL3) == 0) {
	    // this encoding matches the old struct ceph_pg_pool
	    __u8 struct_v = 2;
	    encode(struct_v, bl);
	    encode(type, bl);
	    encode(size, bl);
	    encode(crush_rule, bl);
	    encode(object_hash, bl);
	    encode(pg_num, bl);
	    encode(pgp_num, bl);
	    __u32 lpg_num = 0, lpgp_num = 0;  // tell old code that there are no localized pgs.
	    encode(lpg_num, bl);
	    encode(lpgp_num, bl);
	    encode(last_change, bl);
	    encode(snap_seq, bl);
	    encode(snap_epoch, bl);
	
	    __u32 n = snaps.size();
	    encode(n, bl);
	    n = removed_snaps.num_intervals();
	    encode(n, bl);
	
	    encode(auid, bl);
	
	    encode_nohead(snaps, bl, features);
	    encode_nohead(removed_snaps, bl);
	    return;
	  }
	
	  if ((features & CEPH_FEATURE_OSDENC) == 0) {
	    __u8 struct_v = 4;
	    encode(struct_v, bl);
	    encode(type, bl);
	    encode(size, bl);
	    encode(crush_rule, bl);
	    encode(object_hash, bl);
	    encode(pg_num, bl);
	    encode(pgp_num, bl);
	    __u32 lpg_num = 0, lpgp_num = 0;  // tell old code that there are no localized pgs.
	    encode(lpg_num, bl);
	    encode(lpgp_num, bl);
	    encode(last_change, bl);
	    encode(snap_seq, bl);
	    encode(snap_epoch, bl);
	    encode(snaps, bl, features);
	    encode(removed_snaps, bl);
	    encode(auid, bl);
	    encode(flags, bl);
	    encode((uint32_t)0, bl); // crash_replay_interval
	    return;
	  }
	
	  if ((features & CEPH_FEATURE_OSD_POOLRESEND) == 0) {
	    // we simply added last_force_op_resend here, which is a fully
	    // backward compatible change.  however, encoding the same map
	    // differently between monitors triggers scrub noise (even though
	    // they are decodable without the feature), so let's be pendantic
	    // about it.
	    ENCODE_START(14, 5, bl);
	    encode(type, bl);
	    encode(size, bl);
	    encode(crush_rule, bl);
	    encode(object_hash, bl);
	    encode(pg_num, bl);
	    encode(pgp_num, bl);
	    __u32 lpg_num = 0, lpgp_num = 0;  // tell old code that there are no localized pgs.
	    encode(lpg_num, bl);
	    encode(lpgp_num, bl);
	    encode(last_change, bl);
	    encode(snap_seq, bl);
	    encode(snap_epoch, bl);
	    encode(snaps, bl, features);
	    encode(removed_snaps, bl);
	    encode(auid, bl);
	    encode(flags, bl);
	    encode((uint32_t)0, bl); // crash_replay_interval
	    encode(min_size, bl);
	    encode(quota_max_bytes, bl);
	    encode(quota_max_objects, bl);
	    encode(tiers, bl);
	    encode(tier_of, bl);
	    __u8 c = cache_mode;
	    encode(c, bl);
	    encode(read_tier, bl);
	    encode(write_tier, bl);
	    encode(properties, bl);
	    encode(hit_set_params, bl);
	    encode(hit_set_period, bl);
	    encode(hit_set_count, bl);
	    encode(stripe_width, bl);
	    encode(target_max_bytes, bl);
	    encode(target_max_objects, bl);
	    encode(cache_target_dirty_ratio_micro, bl);
	    encode(cache_target_full_ratio_micro, bl);
	    encode(cache_min_flush_age, bl);
	    encode(cache_min_evict_age, bl);
	    encode(erasure_code_profile, bl);
	    ENCODE_FINISH(bl);
	    return;
	  }
	
	  uint8_t v = 29;
	  // NOTE: any new encoding dependencies must be reflected by
	  // SIGNIFICANT_FEATURES
	  if (!(features & CEPH_FEATURE_NEW_OSDOP_ENCODING)) {
	    // this was the first post-hammer thing we added; if it's missing, encode
	    // like hammer.
	    v = 21;
	  } else if (!HAVE_FEATURE(features, SERVER_LUMINOUS)) {
	    v = 24;
	  } else if (!HAVE_FEATURE(features, SERVER_MIMIC)) {
	    v = 26;
	  } else if (!HAVE_FEATURE(features, SERVER_NAUTILUS)) {
	    v = 27;
	  }
	
	  ENCODE_START(v, 5, bl);
	  encode(type, bl);
	  encode(size, bl);
	  encode(crush_rule, bl);
	  encode(object_hash, bl);
	  encode(pg_num, bl);
	  encode(pgp_num, bl);
	  __u32 lpg_num = 0, lpgp_num = 0;  // tell old code that there are no localized pgs.
	  encode(lpg_num, bl);
	  encode(lpgp_num, bl);
	  encode(last_change, bl);
	  encode(snap_seq, bl);
	  encode(snap_epoch, bl);
	  encode(snaps, bl, features);
	  encode(removed_snaps, bl);
	  encode(auid, bl);
	  if (v >= 27) {
	    encode(flags, bl);
	  } else {
	    auto tmp = flags;
	    tmp &= ~(FLAG_SELFMANAGED_SNAPS | FLAG_POOL_SNAPS | FLAG_CREATING);
	    encode(tmp, bl);
	  }
	  encode((uint32_t)0, bl); // crash_replay_interval
	  encode(min_size, bl);
	  encode(quota_max_bytes, bl);
	  encode(quota_max_objects, bl);
	  encode(tiers, bl);
	  encode(tier_of, bl);
	  __u8 c = cache_mode;
	  encode(c, bl);
	  encode(read_tier, bl);
	  encode(write_tier, bl);
	  encode(properties, bl);
	  encode(hit_set_params, bl);
	  encode(hit_set_period, bl);
	  encode(hit_set_count, bl);
	  encode(stripe_width, bl);
	  encode(target_max_bytes, bl);
	  encode(target_max_objects, bl);
	  encode(cache_target_dirty_ratio_micro, bl);
	  encode(cache_target_full_ratio_micro, bl);
	  encode(cache_min_flush_age, bl);
	  encode(cache_min_evict_age, bl);
	  encode(erasure_code_profile, bl);
	  encode(last_force_op_resend_preluminous, bl);
	  encode(min_read_recency_for_promote, bl);
	  encode(expected_num_objects, bl);
	  if (v >= 19) {
	    encode(cache_target_dirty_high_ratio_micro, bl);
	  }
	  if (v >= 20) {
	    encode(min_write_recency_for_promote, bl);
	  }
	  if (v >= 21) {
	    encode(use_gmt_hitset, bl);
	  }
	  if (v >= 22) {
	    encode(fast_read, bl);
	  }
	  if (v >= 23) {
	    encode(hit_set_grade_decay_rate, bl);
	    encode(hit_set_search_last_n, bl);
	  }
	  if (v >= 24) {
	    encode(opts, bl, features);
	  }
	  if (v >= 25) {
	    encode(last_force_op_resend_prenautilus, bl);
	  }
	  if (v >= 26) {
	    encode(application_metadata, bl);
	  }
	  if (v >= 27) {
	    encode(create_time, bl);
	  }
	  if (v >= 28) {
	    encode(pg_num_target, bl);
	    encode(pgp_num_target, bl);
	    encode(pg_num_pending, bl);
	    encode((epoch_t)0, bl);  // pg_num_dec_last_epoch_started from 14.1.[01]
	    encode((epoch_t)0, bl);  // pg_num_dec_last_epoch_clean from 14.1.[01]
	    encode(last_force_op_resend, bl);
	    encode(pg_autoscale_mode, bl);
	  }
	  if (v >= 29) {
	    encode(last_pg_merge_meta, bl);
	  }
	  ENCODE_FINISH(bl);
	}
	
	void pg_pool_t::decode(ceph::buffer::list::const_iterator& bl)
	{
	  DECODE_START_LEGACY_COMPAT_LEN(29, 5, 5, bl);
	  decode(type, bl);
	  decode(size, bl);
	  decode(crush_rule, bl);
	  decode(object_hash, bl);
	  decode(pg_num, bl);
	  decode(pgp_num, bl);
	  {
	    __u32 lpg_num, lpgp_num;
	    decode(lpg_num, bl);
	    decode(lpgp_num, bl);
	  }
	  decode(last_change, bl);
	  decode(snap_seq, bl);
	  decode(snap_epoch, bl);
	
	  if (struct_v >= 3) {
	    decode(snaps, bl);
	    decode(removed_snaps, bl);
	    decode(auid, bl);
	  } else {
	    __u32 n, m;
	    decode(n, bl);
	    decode(m, bl);
	    decode(auid, bl);
	    decode_nohead(n, snaps, bl);
	    decode_nohead(m, removed_snaps, bl);
	  }
	
	  if (struct_v >= 4) {
	    decode(flags, bl);
	    uint32_t crash_replay_interval;
	    decode(crash_replay_interval, bl);
	  } else {
	    flags = 0;
	  }
	  // upgrade path for selfmanaged vs pool snaps
	  if (snap_seq > 0 && (flags & (FLAG_SELFMANAGED_SNAPS|FLAG_POOL_SNAPS)) == 0) {
	    if (!removed_snaps.empty()) {
	      flags |= FLAG_SELFMANAGED_SNAPS;
	    } else {
	      flags |= FLAG_POOL_SNAPS;
	    }
	  }
	  if (struct_v >= 7) {
	    decode(min_size, bl);
	  } else {
	    min_size = size - size/2;
	  }
	  if (struct_v >= 8) {
	    decode(quota_max_bytes, bl);
	    decode(quota_max_objects, bl);
	  }
	  if (struct_v >= 9) {
	    decode(tiers, bl);
	    decode(tier_of, bl);
	    __u8 v;
	    decode(v, bl);
	    cache_mode = (cache_mode_t)v;
	    decode(read_tier, bl);
	    decode(write_tier, bl);
	  }
	  if (struct_v >= 10) {
	    decode(properties, bl);
	  }
	  if (struct_v >= 11) {
	    decode(hit_set_params, bl);
	    decode(hit_set_period, bl);
	    decode(hit_set_count, bl);
	  } else {
	    pg_pool_t def;
	    hit_set_period = def.hit_set_period;
	    hit_set_count = def.hit_set_count;
	  }
	  if (struct_v >= 12) {
	    decode(stripe_width, bl);
	  } else {
	    set_stripe_width(0);
	  }
	  if (struct_v >= 13) {
	    decode(target_max_bytes, bl);
	    decode(target_max_objects, bl);
	    decode(cache_target_dirty_ratio_micro, bl);
	    decode(cache_target_full_ratio_micro, bl);
	    decode(cache_min_flush_age, bl);
	    decode(cache_min_evict_age, bl);
	  } else {
	    target_max_bytes = 0;
	    target_max_objects = 0;
	    cache_target_dirty_ratio_micro = 0;
	    cache_target_full_ratio_micro = 0;
	    cache_min_flush_age = 0;
	    cache_min_evict_age = 0;
	  }
	  if (struct_v >= 14) {
	    decode(erasure_code_profile, bl);
	  }
	  if (struct_v >= 15) {
	    decode(last_force_op_resend_preluminous, bl);
	  } else {
	    last_force_op_resend_preluminous = 0;
	  }
	  if (struct_v >= 16) {
	    decode(min_read_recency_for_promote, bl);
	  } else {
	    min_read_recency_for_promote = 1;
	  }
	  if (struct_v >= 17) {
	    decode(expected_num_objects, bl);
	  } else {
	    expected_num_objects = 0;
	  }
	  if (struct_v >= 19) {
	    decode(cache_target_dirty_high_ratio_micro, bl);
	  } else {
	    cache_target_dirty_high_ratio_micro = cache_target_dirty_ratio_micro;
	  }
	  if (struct_v >= 20) {
	    decode(min_write_recency_for_promote, bl);
	  } else {
	    min_write_recency_for_promote = 1;
	  }
	  if (struct_v >= 21) {
	    decode(use_gmt_hitset, bl);
	  } else {
	    use_gmt_hitset = false;
	  }
	  if (struct_v >= 22) {
	    decode(fast_read, bl);
	  } else {
	    fast_read = false;
	  }
	  if (struct_v >= 23) {
	    decode(hit_set_grade_decay_rate, bl);
	    decode(hit_set_search_last_n, bl);
	  } else {
	    hit_set_grade_decay_rate = 0;
	    hit_set_search_last_n = 1;
	  }
	  if (struct_v >= 24) {
	    decode(opts, bl);
	  }
	  if (struct_v >= 25) {
	    decode(last_force_op_resend_prenautilus, bl);
	  } else {
	    last_force_op_resend_prenautilus = last_force_op_resend_preluminous;
	  }
	  if (struct_v >= 26) {
	    decode(application_metadata, bl);
	  }
	  if (struct_v >= 27) {
	    decode(create_time, bl);
	  }
	  if (struct_v >= 28) {
	    decode(pg_num_target, bl);
	    decode(pgp_num_target, bl);
	    decode(pg_num_pending, bl);
	    epoch_t old_merge_last_epoch_clean, old_merge_last_epoch_started;
	    decode(old_merge_last_epoch_started, bl);
	    decode(old_merge_last_epoch_clean, bl);
	    decode(last_force_op_resend, bl);
	    decode(pg_autoscale_mode, bl);
	    if (struct_v >= 29) {
	      decode(last_pg_merge_meta, bl);
	    } else {
	      last_pg_merge_meta.last_epoch_clean = old_merge_last_epoch_clean;
	      last_pg_merge_meta.last_epoch_started = old_merge_last_epoch_started;
	    }
	  } else {
	    pg_num_target = pg_num;
	    pgp_num_target = pgp_num;
	    pg_num_pending = pg_num;
	    last_force_op_resend = last_force_op_resend_prenautilus;
	    pg_autoscale_mode = pg_autoscale_mode_t::WARN;    // default to warn on upgrade
	  }
	  DECODE_FINISH(bl);
	  calc_pg_masks();
	  calc_grade_table();
	}
	
	void pg_pool_t::generate_test_instances(list<pg_pool_t*>& o)
	{
	  pg_pool_t a;
	  o.push_back(new pg_pool_t(a));
	
	  a.create_time = utime_t(4,5);
	  a.type = TYPE_REPLICATED;
	  a.size = 2;
	  a.crush_rule = 3;
	  a.object_hash = 4;
	  a.pg_num = 6;
	  a.pgp_num = 4;
	  a.pgp_num_target = 4;
	  a.pg_num_target = 5;
	  a.pg_num_pending = 5;
	  a.last_pg_merge_meta.last_epoch_started = 2;
	  a.last_pg_merge_meta.last_epoch_clean = 2;
	  a.last_change = 9;
	  a.last_force_op_resend = 123823;
	  a.last_force_op_resend_preluminous = 123824;
	  a.snap_seq = 10;
	  a.snap_epoch = 11;
	  a.flags = FLAG_POOL_SNAPS;
	  a.auid = 12;
	  a.quota_max_bytes = 473;
	  a.quota_max_objects = 474;
	  o.push_back(new pg_pool_t(a));
	
	  a.snaps[3].name = "asdf";
	  a.snaps[3].snapid = 3;
	  a.snaps[3].stamp = utime_t(123, 4);
	  a.snaps[6].name = "qwer";
	  a.snaps[6].snapid = 6;
	  a.snaps[6].stamp = utime_t(23423, 4);
	  o.push_back(new pg_pool_t(a));
	
	  a.flags = FLAG_SELFMANAGED_SNAPS;
	  a.snaps.clear();
	  a.removed_snaps.insert(2);
	  a.quota_max_bytes = 2473;
	  a.quota_max_objects = 4374;
	  a.tiers.insert(0);
	  a.tiers.insert(1);
	  a.tier_of = 2;
	  a.cache_mode = CACHEMODE_WRITEBACK;
	  a.read_tier = 1;
	  a.write_tier = 1;
	  a.hit_set_params = HitSet::Params(new BloomHitSet::Params);
	  a.hit_set_period = 3600;
	  a.hit_set_count = 8;
	  a.min_read_recency_for_promote = 1;
	  a.min_write_recency_for_promote = 1;
	  a.hit_set_grade_decay_rate = 50;
	  a.hit_set_search_last_n = 1;
	  a.calc_grade_table();
	  a.set_stripe_width(12345);
	  a.target_max_bytes = 1238132132;
	  a.target_max_objects = 1232132;
	  a.cache_target_dirty_ratio_micro = 187232;
	  a.cache_target_dirty_high_ratio_micro = 309856;
	  a.cache_target_full_ratio_micro = 987222;
	  a.cache_min_flush_age = 231;
	  a.cache_min_evict_age = 2321;
	  a.erasure_code_profile = "profile in osdmap";
	  a.expected_num_objects = 123456;
	  a.fast_read = false;
	  a.application_metadata = {{"rbd", {{"key", "value"}}}};
	  o.push_back(new pg_pool_t(a));
	}
	
	ostream& operator<<(ostream& out, const pg_pool_t& p)
	{
	  out << p.get_type_name();
	  if (p.get_type_name() == "erasure") {
	    out << " profile " << p.erasure_code_profile;
	  }
	  out << " size " << p.get_size()
	      << " min_size " << p.get_min_size()
	      << " crush_rule " << p.get_crush_rule()
	      << " object_hash " << p.get_object_hash_name()
	      << " pg_num " << p.get_pg_num()
	      << " pgp_num " << p.get_pgp_num();
	  if (p.get_pg_num_target() != p.get_pg_num()) {
	    out << " pg_num_target " << p.get_pg_num_target();
	  }
	  if (p.get_pgp_num_target() != p.get_pgp_num()) {
	    out << " pgp_num_target " << p.get_pgp_num_target();
	  }
	  if (p.get_pg_num_pending() != p.get_pg_num()) {
	    out << " pg_num_pending " << p.get_pg_num_pending();
	  }
	  if (p.pg_autoscale_mode != pg_pool_t::pg_autoscale_mode_t::UNKNOWN) {
	    out << " autoscale_mode " << p.get_pg_autoscale_mode_name(p.pg_autoscale_mode);
	  }
	  out << " last_change " << p.get_last_change();
	  if (p.get_last_force_op_resend() ||
	      p.get_last_force_op_resend_prenautilus() ||
	      p.get_last_force_op_resend_preluminous())
	    out << " lfor " << p.get_last_force_op_resend() << "/"
		<< p.get_last_force_op_resend_prenautilus() << "/"
		<< p.get_last_force_op_resend_preluminous();
	  if (p.get_auid())
	    out << " owner " << p.get_auid();
	  if (p.flags)
	    out << " flags " << p.get_flags_string();
	  if (p.quota_max_bytes)
	    out << " max_bytes " << p.quota_max_bytes;
	  if (p.quota_max_objects)
	    out << " max_objects " << p.quota_max_objects;
	  if (!p.tiers.empty())
	    out << " tiers " << p.tiers;
	  if (p.is_tier())
	    out << " tier_of " << p.tier_of;
	  if (p.has_read_tier())
	    out << " read_tier " << p.read_tier;
	  if (p.has_write_tier())
	    out << " write_tier " << p.write_tier;
	  if (p.cache_mode)
	    out << " cache_mode " << p.get_cache_mode_name();
	  if (p.target_max_bytes)
	    out << " target_bytes " << p.target_max_bytes;
	  if (p.target_max_objects)
	    out << " target_objects " << p.target_max_objects;
	  if (p.hit_set_params.get_type() != HitSet::TYPE_NONE) {
	    out << " hit_set " << p.hit_set_params
		<< " " << p.hit_set_period << "s"
		<< " x" << p.hit_set_count << " decay_rate "
		<< p.hit_set_grade_decay_rate
		<< " search_last_n " << p.hit_set_search_last_n;
	  }
	  if (p.min_read_recency_for_promote)
	    out << " min_read_recency_for_promote " << p.min_read_recency_for_promote;
	  if (p.min_write_recency_for_promote)
	    out << " min_write_recency_for_promote " << p.min_write_recency_for_promote;
	  out << " stripe_width " << p.get_stripe_width();
	  if (p.expected_num_objects)
	    out << " expected_num_objects " << p.expected_num_objects;
	  if (p.fast_read)
	    out << " fast_read " << p.fast_read;
	  out << p.opts;
	  if (!p.application_metadata.empty()) {
	    out << " application ";
	    for (auto it = p.application_metadata.begin();
	         it != p.application_metadata.end(); ++it) {
	      if (it != p.application_metadata.begin())
	        out << ",";
	      out << it->first;
	    }
	  }
	  return out;
	}
	
	
	// -- object_stat_sum_t --
	
	void object_stat_sum_t::dump(Formatter *f) const
	{
	  f->dump_int("num_bytes", num_bytes);
	  f->dump_int("num_objects", num_objects);
	  f->dump_int("num_object_clones", num_object_clones);
	  f->dump_int("num_object_copies", num_object_copies);
	  f->dump_int("num_objects_missing_on_primary", num_objects_missing_on_primary);
	  f->dump_int("num_objects_missing", num_objects_missing);
	  f->dump_int("num_objects_degraded", num_objects_degraded);
	  f->dump_int("num_objects_misplaced", num_objects_misplaced);
	  f->dump_int("num_objects_unfound", num_objects_unfound);
	  f->dump_int("num_objects_dirty", num_objects_dirty);
	  f->dump_int("num_whiteouts", num_whiteouts);
	  f->dump_int("num_read", num_rd);
	  f->dump_int("num_read_kb", num_rd_kb);
	  f->dump_int("num_write", num_wr);
	  f->dump_int("num_write_kb", num_wr_kb);
	  f->dump_int("num_scrub_errors", num_scrub_errors);
	  f->dump_int("num_shallow_scrub_errors", num_shallow_scrub_errors);
	  f->dump_int("num_deep_scrub_errors", num_deep_scrub_errors);
	  f->dump_int("num_objects_recovered", num_objects_recovered);
	  f->dump_int("num_bytes_recovered", num_bytes_recovered);
	  f->dump_int("num_keys_recovered", num_keys_recovered);
	  f->dump_int("num_objects_omap", num_objects_omap);
	  f->dump_int("num_objects_hit_set_archive", num_objects_hit_set_archive);
	  f->dump_int("num_bytes_hit_set_archive", num_bytes_hit_set_archive);
	  f->dump_int("num_flush", num_flush);
	  f->dump_int("num_flush_kb", num_flush_kb);
	  f->dump_int("num_evict", num_evict);
	  f->dump_int("num_evict_kb", num_evict_kb);
	  f->dump_int("num_promote", num_promote);
	  f->dump_int("num_flush_mode_high", num_flush_mode_high);
	  f->dump_int("num_flush_mode_low", num_flush_mode_low);
	  f->dump_int("num_evict_mode_some", num_evict_mode_some);
	  f->dump_int("num_evict_mode_full", num_evict_mode_full);
	  f->dump_int("num_objects_pinned", num_objects_pinned);
	  f->dump_int("num_legacy_snapsets", num_legacy_snapsets);
	  f->dump_int("num_large_omap_objects", num_large_omap_objects);
	  f->dump_int("num_objects_manifest", num_objects_manifest);
	  f->dump_int("num_omap_bytes", num_omap_bytes);
	  f->dump_int("num_omap_keys", num_omap_keys);
	  f->dump_int("num_objects_repaired", num_objects_repaired);
	}
	
	void object_stat_sum_t::encode(ceph::buffer::list& bl) const
	{
	  ENCODE_START(20, 14, bl);
	#if defined(CEPH_LITTLE_ENDIAN)
	  bl.append((char *)(&num_bytes), sizeof(object_stat_sum_t));
	#else
	  encode(num_bytes, bl);
	  encode(num_objects, bl);
	  encode(num_object_clones, bl);
	  encode(num_object_copies, bl);
	  encode(num_objects_missing_on_primary, bl);
	  encode(num_objects_degraded, bl);
	  encode(num_objects_unfound, bl);
	  encode(num_rd, bl);
	  encode(num_rd_kb, bl);
	  encode(num_wr, bl);
	  encode(num_wr_kb, bl);
	  encode(num_scrub_errors, bl);
	  encode(num_objects_recovered, bl);
	  encode(num_bytes_recovered, bl);
	  encode(num_keys_recovered, bl);
	  encode(num_shallow_scrub_errors, bl);
	  encode(num_deep_scrub_errors, bl);
	  encode(num_objects_dirty, bl);
	  encode(num_whiteouts, bl);
	  encode(num_objects_omap, bl);
	  encode(num_objects_hit_set_archive, bl);
	  encode(num_objects_misplaced, bl);
	  encode(num_bytes_hit_set_archive, bl);
	  encode(num_flush, bl);
	  encode(num_flush_kb, bl);
	  encode(num_evict, bl);
	  encode(num_evict_kb, bl);
	  encode(num_promote, bl);
	  encode(num_flush_mode_high, bl);
	  encode(num_flush_mode_low, bl);
	  encode(num_evict_mode_some, bl);
	  encode(num_evict_mode_full, bl);
	  encode(num_objects_pinned, bl);
	  encode(num_objects_missing, bl);
	  encode(num_legacy_snapsets, bl);
	  encode(num_large_omap_objects, bl);
	  encode(num_objects_manifest, bl);
	  encode(num_omap_bytes, bl);
	  encode(num_omap_keys, bl);
	  encode(num_objects_repaired, bl);
	#endif
	  ENCODE_FINISH(bl);
	}
	
	void object_stat_sum_t::decode(ceph::buffer::list::const_iterator& bl)
	{
	  bool decode_finish = false;
	  static const int STAT_SUM_DECODE_VERSION = 20;
	  DECODE_START(STAT_SUM_DECODE_VERSION, bl);
	#if defined(CEPH_LITTLE_ENDIAN)
	  if (struct_v == STAT_SUM_DECODE_VERSION) {
	    bl.copy(sizeof(object_stat_sum_t), (char*)(&num_bytes));
	    decode_finish = true;
	  }
	#endif
	  if (!decode_finish) {
	    decode(num_bytes, bl);
	    decode(num_objects, bl);
	    decode(num_object_clones, bl);
	    decode(num_object_copies, bl);
	    decode(num_objects_missing_on_primary, bl);
	    decode(num_objects_degraded, bl);
	    decode(num_objects_unfound, bl);
	    decode(num_rd, bl);
	    decode(num_rd_kb, bl);
	    decode(num_wr, bl);
	    decode(num_wr_kb, bl);
	    decode(num_scrub_errors, bl);
	    decode(num_objects_recovered, bl);
	    decode(num_bytes_recovered, bl);
	    decode(num_keys_recovered, bl);
	    decode(num_shallow_scrub_errors, bl);
	    decode(num_deep_scrub_errors, bl);
	    decode(num_objects_dirty, bl);
	    decode(num_whiteouts, bl);
	    decode(num_objects_omap, bl);
	    decode(num_objects_hit_set_archive, bl);
	    decode(num_objects_misplaced, bl);
	    decode(num_bytes_hit_set_archive, bl);
	    decode(num_flush, bl);
	    decode(num_flush_kb, bl);
	    decode(num_evict, bl);
	    decode(num_evict_kb, bl);
	    decode(num_promote, bl);
	    decode(num_flush_mode_high, bl);
	    decode(num_flush_mode_low, bl);
	    decode(num_evict_mode_some, bl);
	    decode(num_evict_mode_full, bl);
	    decode(num_objects_pinned, bl);
	    decode(num_objects_missing, bl);
	    if (struct_v >= 16) {
	      decode(num_legacy_snapsets, bl);
	    } else {
	      num_legacy_snapsets = num_object_clones;  // upper bound
	    }
	    if (struct_v >= 17) {
	      decode(num_large_omap_objects, bl);
	    }
	    if (struct_v >= 18) {
	      decode(num_objects_manifest, bl);
	    }
	    if (struct_v >= 19) {
	      decode(num_omap_bytes, bl);
	      decode(num_omap_keys, bl);
	    }
	    if (struct_v >= 20) {
	      decode(num_objects_repaired, bl);
	    }
	  }
	  DECODE_FINISH(bl);
	}
	
	void object_stat_sum_t::generate_test_instances(list<object_stat_sum_t*>& o)
	{
	  object_stat_sum_t a;
	
	  a.num_bytes = 1;
	  a.num_objects = 3;
	  a.num_object_clones = 4;
	  a.num_object_copies = 5;
	  a.num_objects_missing_on_primary = 6;
	  a.num_objects_missing = 123;
	  a.num_objects_degraded = 7;
	  a.num_objects_unfound = 8;
	  a.num_rd = 9; a.num_rd_kb = 10;
	  a.num_wr = 11; a.num_wr_kb = 12;
	  a.num_objects_recovered = 14;
	  a.num_bytes_recovered = 15;
	  a.num_keys_recovered = 16;
	  a.num_deep_scrub_errors = 17;
	  a.num_shallow_scrub_errors = 18;
	  a.num_scrub_errors = a.num_deep_scrub_errors + a.num_shallow_scrub_errors;
	  a.num_objects_dirty = 21;
	  a.num_whiteouts = 22;
	  a.num_objects_misplaced = 1232;
	  a.num_objects_hit_set_archive = 2;
	  a.num_bytes_hit_set_archive = 27;
	  a.num_flush = 5;
	  a.num_flush_kb = 6;
	  a.num_evict = 7;
	  a.num_evict_kb = 8;
	  a.num_promote = 9;
	  a.num_flush_mode_high = 0;
	  a.num_flush_mode_low = 1;
	  a.num_evict_mode_some = 1;
	  a.num_evict_mode_full = 0;
	  a.num_objects_pinned = 20;
	  a.num_large_omap_objects = 5;
	  a.num_objects_manifest = 2;
	  a.num_omap_bytes = 20000;
	  a.num_omap_keys = 200;
	  a.num_objects_repaired = 300;
	  o.push_back(new object_stat_sum_t(a));
	}
	
	void object_stat_sum_t::add(const object_stat_sum_t& o)
	{
	  num_bytes += o.num_bytes;
	  num_objects += o.num_objects;
	  num_object_clones += o.num_object_clones;
	  num_object_copies += o.num_object_copies;
	  num_objects_missing_on_primary += o.num_objects_missing_on_primary;
	  num_objects_missing += o.num_objects_missing;
	  num_objects_degraded += o.num_objects_degraded;
	  num_objects_misplaced += o.num_objects_misplaced;
	  num_rd += o.num_rd;
	  num_rd_kb += o.num_rd_kb;
	  num_wr += o.num_wr;
	  num_wr_kb += o.num_wr_kb;
	  num_objects_unfound += o.num_objects_unfound;
	  num_scrub_errors += o.num_scrub_errors;
	  num_shallow_scrub_errors += o.num_shallow_scrub_errors;
	  num_deep_scrub_errors += o.num_deep_scrub_errors;
	  num_objects_recovered += o.num_objects_recovered;
	  num_bytes_recovered += o.num_bytes_recovered;
	  num_keys_recovered += o.num_keys_recovered;
	  num_objects_dirty += o.num_objects_dirty;
	  num_whiteouts += o.num_whiteouts;
	  num_objects_omap += o.num_objects_omap;
	  num_objects_hit_set_archive += o.num_objects_hit_set_archive;
	  num_bytes_hit_set_archive += o.num_bytes_hit_set_archive;
	  num_flush += o.num_flush;
	  num_flush_kb += o.num_flush_kb;
	  num_evict += o.num_evict;
	  num_evict_kb += o.num_evict_kb;
	  num_promote += o.num_promote;
	  num_flush_mode_high += o.num_flush_mode_high;
	  num_flush_mode_low += o.num_flush_mode_low;
	  num_evict_mode_some += o.num_evict_mode_some;
	  num_evict_mode_full += o.num_evict_mode_full;
	  num_objects_pinned += o.num_objects_pinned;
	  num_legacy_snapsets += o.num_legacy_snapsets;
	  num_large_omap_objects += o.num_large_omap_objects;
	  num_objects_manifest += o.num_objects_manifest;
	  num_omap_bytes += o.num_omap_bytes;
	  num_omap_keys += o.num_omap_keys;
	  num_objects_repaired += o.num_objects_repaired;
	}
	
	void object_stat_sum_t::sub(const object_stat_sum_t& o)
	{
	  num_bytes -= o.num_bytes;
	  num_objects -= o.num_objects;
	  num_object_clones -= o.num_object_clones;
	  num_object_copies -= o.num_object_copies;
	  num_objects_missing_on_primary -= o.num_objects_missing_on_primary;
	  num_objects_missing -= o.num_objects_missing;
	  num_objects_degraded -= o.num_objects_degraded;
	  num_objects_misplaced -= o.num_objects_misplaced;
	  num_rd -= o.num_rd;
	  num_rd_kb -= o.num_rd_kb;
	  num_wr -= o.num_wr;
	  num_wr_kb -= o.num_wr_kb;
	  num_objects_unfound -= o.num_objects_unfound;
	  num_scrub_errors -= o.num_scrub_errors;
	  num_shallow_scrub_errors -= o.num_shallow_scrub_errors;
	  num_deep_scrub_errors -= o.num_deep_scrub_errors;
	  num_objects_recovered -= o.num_objects_recovered;
	  num_bytes_recovered -= o.num_bytes_recovered;
	  num_keys_recovered -= o.num_keys_recovered;
	  num_objects_dirty -= o.num_objects_dirty;
	  num_whiteouts -= o.num_whiteouts;
	  num_objects_omap -= o.num_objects_omap;
	  num_objects_hit_set_archive -= o.num_objects_hit_set_archive;
	  num_bytes_hit_set_archive -= o.num_bytes_hit_set_archive;
	  num_flush -= o.num_flush;
	  num_flush_kb -= o.num_flush_kb;
	  num_evict -= o.num_evict;
	  num_evict_kb -= o.num_evict_kb;
	  num_promote -= o.num_promote;
	  num_flush_mode_high -= o.num_flush_mode_high;
	  num_flush_mode_low -= o.num_flush_mode_low;
	  num_evict_mode_some -= o.num_evict_mode_some;
	  num_evict_mode_full -= o.num_evict_mode_full;
	  num_objects_pinned -= o.num_objects_pinned;
	  num_legacy_snapsets -= o.num_legacy_snapsets;
	  num_large_omap_objects -= o.num_large_omap_objects;
	  num_objects_manifest -= o.num_objects_manifest;
	  num_omap_bytes -= o.num_omap_bytes;
	  num_omap_keys -= o.num_omap_keys;
	  num_objects_repaired -= o.num_objects_repaired;
	}
	
	bool operator==(const object_stat_sum_t& l, const object_stat_sum_t& r)
	{
	  return
	    l.num_bytes == r.num_bytes &&
	    l.num_objects == r.num_objects &&
	    l.num_object_clones == r.num_object_clones &&
	    l.num_object_copies == r.num_object_copies &&
	    l.num_objects_missing_on_primary == r.num_objects_missing_on_primary &&
	    l.num_objects_missing == r.num_objects_missing &&
	    l.num_objects_degraded == r.num_objects_degraded &&
	    l.num_objects_misplaced == r.num_objects_misplaced &&
	    l.num_objects_unfound == r.num_objects_unfound &&
	    l.num_rd == r.num_rd &&
	    l.num_rd_kb == r.num_rd_kb &&
	    l.num_wr == r.num_wr &&
	    l.num_wr_kb == r.num_wr_kb &&
	    l.num_scrub_errors == r.num_scrub_errors &&
	    l.num_shallow_scrub_errors == r.num_shallow_scrub_errors &&
	    l.num_deep_scrub_errors == r.num_deep_scrub_errors &&
	    l.num_objects_recovered == r.num_objects_recovered &&
	    l.num_bytes_recovered == r.num_bytes_recovered &&
	    l.num_keys_recovered == r.num_keys_recovered &&
	    l.num_objects_dirty == r.num_objects_dirty &&
	    l.num_whiteouts == r.num_whiteouts &&
	    l.num_objects_omap == r.num_objects_omap &&
	    l.num_objects_hit_set_archive == r.num_objects_hit_set_archive &&
	    l.num_bytes_hit_set_archive == r.num_bytes_hit_set_archive &&
	    l.num_flush == r.num_flush &&
	    l.num_flush_kb == r.num_flush_kb &&
	    l.num_evict == r.num_evict &&
	    l.num_evict_kb == r.num_evict_kb &&
	    l.num_promote == r.num_promote &&
	    l.num_flush_mode_high == r.num_flush_mode_high &&
	    l.num_flush_mode_low == r.num_flush_mode_low &&
	    l.num_evict_mode_some == r.num_evict_mode_some &&
	    l.num_evict_mode_full == r.num_evict_mode_full &&
	    l.num_objects_pinned == r.num_objects_pinned &&
	    l.num_legacy_snapsets == r.num_legacy_snapsets &&
	    l.num_large_omap_objects == r.num_large_omap_objects &&
	    l.num_objects_manifest == r.num_objects_manifest &&
	    l.num_omap_bytes == r.num_omap_bytes &&
	    l.num_omap_keys == r.num_omap_keys &&
	    l.num_objects_repaired == r.num_objects_repaired;
	}
	
	// -- object_stat_collection_t --
	
	void object_stat_collection_t::dump(Formatter *f) const
	{
	  f->open_object_section("stat_sum");
	  sum.dump(f);
	  f->close_section();
	}
	
	void object_stat_collection_t::encode(ceph::buffer::list& bl) const
	{
	  ENCODE_START(2, 2, bl);
	  encode(sum, bl);
	  encode((__u32)0, bl);
	  ENCODE_FINISH(bl);
	}
	
	void object_stat_collection_t::decode(ceph::buffer::list::const_iterator& bl)
	{
	  DECODE_START_LEGACY_COMPAT_LEN(2, 2, 2, bl);
	  decode(sum, bl);
	  {
	    map<string,object_stat_sum_t> cat_sum;
	    decode(cat_sum, bl);
	  }
	  DECODE_FINISH(bl);
	}
	
	void object_stat_collection_t::generate_test_instances(list<object_stat_collection_t*>& o)
	{
	  object_stat_collection_t a;
	  o.push_back(new object_stat_collection_t(a));
	  list<object_stat_sum_t*> l;
	  object_stat_sum_t::generate_test_instances(l);
	  for (auto p = l.begin(); p != l.end(); ++p) {
	    a.add(**p);
	    o.push_back(new object_stat_collection_t(a));
	  }
	}
	
	
	// -- pg_stat_t --
	
	bool pg_stat_t::is_acting_osd(int32_t osd, bool primary) const
	{
	  if (primary && osd == acting_primary) {
	    return true;
	  } else if (!primary) {
	    for(auto it = acting.cbegin(); it != acting.cend(); ++it)
	    {
	      if (*it == osd)
	        return true;
	    }
	  }
	  return false;
	}
	
	void pg_stat_t::dump(Formatter *f) const
	{
	  f->dump_stream("version") << version;
	  f->dump_stream("reported_seq") << reported_seq;
	  f->dump_stream("reported_epoch") << reported_epoch;
	  f->dump_string("state", pg_state_string(state));
	  f->dump_stream("last_fresh") << last_fresh;
	  f->dump_stream("last_change") << last_change;
	  f->dump_stream("last_active") << last_active;
	  f->dump_stream("last_peered") << last_peered;
	  f->dump_stream("last_clean") << last_clean;
	  f->dump_stream("last_became_active") << last_became_active;
	  f->dump_stream("last_became_peered") << last_became_peered;
	  f->dump_stream("last_unstale") << last_unstale;
	  f->dump_stream("last_undegraded") << last_undegraded;
	  f->dump_stream("last_fullsized") << last_fullsized;
	  f->dump_unsigned("mapping_epoch", mapping_epoch);
	  f->dump_stream("log_start") << log_start;
	  f->dump_stream("ondisk_log_start") << ondisk_log_start;
	  f->dump_unsigned("created", created);
	  f->dump_unsigned("last_epoch_clean", last_epoch_clean);
	  f->dump_stream("parent") << parent;
	  f->dump_unsigned("parent_split_bits", parent_split_bits);
	  f->dump_stream("last_scrub") << last_scrub;
	  f->dump_stream("last_scrub_stamp") << last_scrub_stamp;
	  f->dump_stream("last_deep_scrub") << last_deep_scrub;
	  f->dump_stream("last_deep_scrub_stamp") << last_deep_scrub_stamp;
	  f->dump_stream("last_clean_scrub_stamp") << last_clean_scrub_stamp;
	  f->dump_int("log_size", log_size);
	  f->dump_int("ondisk_log_size", ondisk_log_size);
	  f->dump_bool("stats_invalid", stats_invalid);
	  f->dump_bool("dirty_stats_invalid", dirty_stats_invalid);
	  f->dump_bool("omap_stats_invalid", omap_stats_invalid);
	  f->dump_bool("hitset_stats_invalid", hitset_stats_invalid);
	  f->dump_bool("hitset_bytes_stats_invalid", hitset_bytes_stats_invalid);
	  f->dump_bool("pin_stats_invalid", pin_stats_invalid);
	  f->dump_bool("manifest_stats_invalid", manifest_stats_invalid);
	  f->dump_unsigned("snaptrimq_len", snaptrimq_len);
	  stats.dump(f);
	  f->open_array_section("up");
	  for (auto p = up.cbegin(); p != up.cend(); ++p)
	    f->dump_int("osd", *p);
	  f->close_section();
	  f->open_array_section("acting");
	  for (auto p = acting.cbegin(); p != acting.cend(); ++p)
	    f->dump_int("osd", *p);
	  f->close_section();
	  f->open_array_section("avail_no_missing");
	  for (auto p = avail_no_missing.cbegin(); p != avail_no_missing.cend(); ++p)
	    f->dump_stream("shard") << *p;
	  f->close_section();
	  f->open_array_section("object_location_counts");
	  for (auto p = object_location_counts.cbegin(); p != object_location_counts.cend(); ++p) {
	    f->open_object_section("entry");
	    f->dump_stream("shards") << p->first;
	    f->dump_int("objects", p->second);
	    f->close_section();
	  }
	  f->close_section();
	  f->open_array_section("blocked_by");
	  for (auto p = blocked_by.cbegin(); p != blocked_by.cend(); ++p)
	    f->dump_int("osd", *p);
	  f->close_section();
	  f->dump_int("up_primary", up_primary);
	  f->dump_int("acting_primary", acting_primary);
	  f->open_array_section("purged_snaps");
	  for (auto i = purged_snaps.begin(); i != purged_snaps.end(); ++i) {
	    f->open_object_section("interval");
	    f->dump_stream("start") << i.get_start();
	    f->dump_stream("length") << i.get_len();
	    f->close_section();
	  }
	  f->close_section();
	}
	
	void pg_stat_t::dump_brief(Formatter *f) const
	{
	  f->dump_string("state", pg_state_string(state));
	  f->open_array_section("up");
	  for (auto p = up.cbegin(); p != up.cend(); ++p)
	    f->dump_int("osd", *p);
	  f->close_section();
	  f->open_array_section("acting");
	  for (auto p = acting.cbegin(); p != acting.cend(); ++p)
	    f->dump_int("osd", *p);
	  f->close_section();
	  f->dump_int("up_primary", up_primary);
	  f->dump_int("acting_primary", acting_primary);
	}
	
	void pg_stat_t::encode(ceph::buffer::list &bl) const
	{
	  ENCODE_START(26, 22, bl);
	  encode(version, bl);
	  encode(reported_seq, bl);
	  encode(reported_epoch, bl);
	  encode((__u32)state, bl);   // for older peers
	  encode(log_start, bl);
	  encode(ondisk_log_start, bl);
	  encode(created, bl);
	  encode(last_epoch_clean, bl);
	  encode(parent, bl);
	  encode(parent_split_bits, bl);
	  encode(last_scrub, bl);
	  encode(last_scrub_stamp, bl);
	  encode(stats, bl);
	  encode(log_size, bl);
	  encode(ondisk_log_size, bl);
	  encode(up, bl);
	  encode(acting, bl);
	  encode(last_fresh, bl);
	  encode(last_change, bl);
	  encode(last_active, bl);
	  encode(last_clean, bl);
	  encode(last_unstale, bl);
	  encode(mapping_epoch, bl);
	  encode(last_deep_scrub, bl);
	  encode(last_deep_scrub_stamp, bl);
	  encode(stats_invalid, bl);
	  encode(last_clean_scrub_stamp, bl);
	  encode(last_became_active, bl);
	  encode(dirty_stats_invalid, bl);
	  encode(up_primary, bl);
	  encode(acting_primary, bl);
	  encode(omap_stats_invalid, bl);
	  encode(hitset_stats_invalid, bl);
	  encode(blocked_by, bl);
	  encode(last_undegraded, bl);
	  encode(last_fullsized, bl);
	  encode(hitset_bytes_stats_invalid, bl);
	  encode(last_peered, bl);
	  encode(last_became_peered, bl);
	  encode(pin_stats_invalid, bl);
	  encode(snaptrimq_len, bl);
	  __u32 top_state = (state >> 32);
	  encode(top_state, bl);
	  encode(purged_snaps, bl);
	  encode(manifest_stats_invalid, bl);
	  encode(avail_no_missing, bl);
	  encode(object_location_counts, bl);
	  ENCODE_FINISH(bl);
	}
	
	void pg_stat_t::decode(ceph::buffer::list::const_iterator &bl)
	{
	  bool tmp;
	  uint32_t old_state;
	  DECODE_START(26, bl);
	  decode(version, bl);
	  decode(reported_seq, bl);
	  decode(reported_epoch, bl);
	  decode(old_state, bl);
	  decode(log_start, bl);
	  decode(ondisk_log_start, bl);
	  decode(created, bl);
	  decode(last_epoch_clean, bl);
	  decode(parent, bl);
	  decode(parent_split_bits, bl);
	  decode(last_scrub, bl);
	  decode(last_scrub_stamp, bl);
	  decode(stats, bl);
	  decode(log_size, bl);
	  decode(ondisk_log_size, bl);
	  decode(up, bl);
	  decode(acting, bl);
	  decode(last_fresh, bl);
	  decode(last_change, bl);
	  decode(last_active, bl);
	  decode(last_clean, bl);
	  decode(last_unstale, bl);
	  decode(mapping_epoch, bl);
	  decode(last_deep_scrub, bl);
	  decode(last_deep_scrub_stamp, bl);
	  decode(tmp, bl);
	  stats_invalid = tmp;
	  decode(last_clean_scrub_stamp, bl);
	  decode(last_became_active, bl);
	  decode(tmp, bl);
	  dirty_stats_invalid = tmp;
	  decode(up_primary, bl);
	  decode(acting_primary, bl);
	  decode(tmp, bl);
	  omap_stats_invalid = tmp;
	  decode(tmp, bl);
	  hitset_stats_invalid = tmp;
	  decode(blocked_by, bl);
	  decode(last_undegraded, bl);
	  decode(last_fullsized, bl);
	  decode(tmp, bl);
	  hitset_bytes_stats_invalid = tmp;
	  decode(last_peered, bl);
	  decode(last_became_peered, bl);
	  decode(tmp, bl);
	  pin_stats_invalid = tmp;
	  if (struct_v >= 23) {
	    decode(snaptrimq_len, bl);
	    if (struct_v >= 24) {
	      __u32 top_state;
	      decode(top_state, bl);
	      state = (uint64_t)old_state | ((uint64_t)top_state << 32);
	      decode(purged_snaps, bl);
	    } else {
	      state = old_state;
	    }
	    if (struct_v >= 25) {
	      decode(tmp, bl);
	      manifest_stats_invalid = tmp;
	    } else {
	      manifest_stats_invalid = true;
	    }
	    if (struct_v >= 26) {
	      decode(avail_no_missing, bl);
	      decode(object_location_counts, bl);
	    }
	  }
	  DECODE_FINISH(bl);
	}
	
	void pg_stat_t::generate_test_instances(list<pg_stat_t*>& o)
	{
	  pg_stat_t a;
	  o.push_back(new pg_stat_t(a));
	
	  a.version = eversion_t(1, 3);
	  a.reported_epoch = 1;
	  a.reported_seq = 2;
	  a.state = 123;
	  a.mapping_epoch = 998;
	  a.last_fresh = utime_t(1002, 1);
	  a.last_change = utime_t(1002, 2);
	  a.last_active = utime_t(1002, 3);
	  a.last_clean = utime_t(1002, 4);
	  a.last_unstale = utime_t(1002, 5);
	  a.last_undegraded = utime_t(1002, 7);
	  a.last_fullsized = utime_t(1002, 8);
	  a.log_start = eversion_t(1, 4);
	  a.ondisk_log_start = eversion_t(1, 5);
	  a.created = 6;
	  a.last_epoch_clean = 7;
	  a.parent = pg_t(1, 2);
	  a.parent_split_bits = 12;
	  a.last_scrub = eversion_t(9, 10);
	  a.last_scrub_stamp = utime_t(11, 12);
	  a.last_deep_scrub = eversion_t(13, 14);
	  a.last_deep_scrub_stamp = utime_t(15, 16);
	  a.last_clean_scrub_stamp = utime_t(17, 18);
	  a.snaptrimq_len = 1048576;
	  list<object_stat_collection_t*> l;
	  object_stat_collection_t::generate_test_instances(l);
	  a.stats = *l.back();
	  a.log_size = 99;
	  a.ondisk_log_size = 88;
	  a.up.push_back(123);
	  a.up_primary = 123;
	  a.acting.push_back(456);
	  a.avail_no_missing.push_back(pg_shard_t(456, shard_id_t::NO_SHARD));
	  set<pg_shard_t> sset = { pg_shard_t(0), pg_shard_t(1) };
	  a.object_location_counts.insert(make_pair(sset, 10));
	  sset.insert(pg_shard_t(2));
	  a.object_location_counts.insert(make_pair(sset, 5));
	  a.acting_primary = 456;
	  o.push_back(new pg_stat_t(a));
	
	  a.up.push_back(124);
	  a.up_primary = 124;
	  a.acting.push_back(124);
	  a.acting_primary = 124;
	  a.blocked_by.push_back(155);
	  a.blocked_by.push_back(156);
	  o.push_back(new pg_stat_t(a));
	}
	
	bool operator==(const pg_stat_t& l, const pg_stat_t& r)
	{
	  return
	    l.version == r.version &&
	    l.reported_seq == r.reported_seq &&
	    l.reported_epoch == r.reported_epoch &&
	    l.state == r.state &&
	    l.last_fresh == r.last_fresh &&
	    l.last_change == r.last_change &&
	    l.last_active == r.last_active &&
	    l.last_peered == r.last_peered &&
	    l.last_clean == r.last_clean &&
	    l.last_unstale == r.last_unstale &&
	    l.last_undegraded == r.last_undegraded &&
	    l.last_fullsized == r.last_fullsized &&
	    l.log_start == r.log_start &&
	    l.ondisk_log_start == r.ondisk_log_start &&
	    l.created == r.created &&
	    l.last_epoch_clean == r.last_epoch_clean &&
	    l.parent == r.parent &&
	    l.parent_split_bits == r.parent_split_bits &&
	    l.last_scrub == r.last_scrub &&
	    l.last_deep_scrub == r.last_deep_scrub &&
	    l.last_scrub_stamp == r.last_scrub_stamp &&
	    l.last_deep_scrub_stamp == r.last_deep_scrub_stamp &&
	    l.last_clean_scrub_stamp == r.last_clean_scrub_stamp &&
	    l.stats == r.stats &&
	    l.stats_invalid == r.stats_invalid &&
	    l.log_size == r.log_size &&
	    l.ondisk_log_size == r.ondisk_log_size &&
	    l.up == r.up &&
	    l.acting == r.acting &&
	    l.avail_no_missing == r.avail_no_missing &&
	    l.object_location_counts == r.object_location_counts &&
	    l.mapping_epoch == r.mapping_epoch &&
	    l.blocked_by == r.blocked_by &&
	    l.last_became_active == r.last_became_active &&
	    l.last_became_peered == r.last_became_peered &&
	    l.dirty_stats_invalid == r.dirty_stats_invalid &&
	    l.omap_stats_invalid == r.omap_stats_invalid &&
	    l.hitset_stats_invalid == r.hitset_stats_invalid &&
	    l.hitset_bytes_stats_invalid == r.hitset_bytes_stats_invalid &&
	    l.up_primary == r.up_primary &&
	    l.acting_primary == r.acting_primary &&
	    l.pin_stats_invalid == r.pin_stats_invalid &&
	    l.manifest_stats_invalid == r.manifest_stats_invalid &&
	    l.purged_snaps == r.purged_snaps &&
	    l.snaptrimq_len == r.snaptrimq_len;
	}
	
	// -- store_statfs_t --
	
	bool store_statfs_t::operator==(const store_statfs_t& other) const
	{
	  return total == other.total
	    && available == other.available
	    && allocated == other.allocated
	    && internally_reserved == other.internally_reserved
	    && data_stored == other.data_stored
	    && data_compressed == other.data_compressed
	    && data_compressed_allocated == other.data_compressed_allocated
	    && data_compressed_original == other.data_compressed_original
	    && omap_allocated == other.omap_allocated
	    && internal_metadata == other.internal_metadata;
	}
	
	void store_statfs_t::dump(Formatter *f) const
	{
	  f->dump_int("total", total);
	  f->dump_int("available", available);
	  f->dump_int("internally_reserved", internally_reserved);
	  f->dump_int("allocated", allocated);
	  f->dump_int("data_stored", data_stored);
	  f->dump_int("data_compressed", data_compressed);
	  f->dump_int("data_compressed_allocated", data_compressed_allocated);
	  f->dump_int("data_compressed_original", data_compressed_original);
	  f->dump_int("omap_allocated", omap_allocated);
	  f->dump_int("internal_metadata", internal_metadata);
	}
	
	ostream& operator<<(ostream& out, const store_statfs_t &s)
	{
	  out << std::hex
	      << "store_statfs(0x" << s.available
	      << "/0x"  << s.internally_reserved
	      << "/0x"  << s.total
	      << ", data 0x" << s.data_stored
	      << "/0x"  << s.allocated
	      << ", compress 0x" << s.data_compressed
	      << "/0x"  << s.data_compressed_allocated
	      << "/0x"  << s.data_compressed_original
	      << ", omap 0x" << s.omap_allocated
	      << ", meta 0x" << s.internal_metadata
	      << std::dec
	      << ")";
	  return out;
	}
	
	void store_statfs_t::generate_test_instances(list<store_statfs_t*>& o)
	{
	  store_statfs_t a;
	  o.push_back(new store_statfs_t(a));
	  a.total = 234;
	  a.available = 123;
	  a.internally_reserved = 33;
	  a.allocated = 32;
	  a.data_stored = 44;
	  a.data_compressed = 21;
	  a.data_compressed_allocated = 12;
	  a.data_compressed_original = 13;
	  a.omap_allocated = 14;
	  a.internal_metadata = 15;
	  o.push_back(new store_statfs_t(a));
	}
	
	// -- pool_stat_t --
	
	void pool_stat_t::dump(Formatter *f) const
	{
	  stats.dump(f);
	  f->open_object_section("store_stats");
	  store_stats.dump(f);
	  f->close_section();
	  f->dump_int("log_size", log_size);
	  f->dump_int("ondisk_log_size", ondisk_log_size);
	  f->dump_int("up", up);
	  f->dump_int("acting", acting);
	  f->dump_int("num_store_stats", num_store_stats);
	}
	
	void pool_stat_t::encode(ceph::buffer::list &bl, uint64_t features) const
	{
	  using ceph::encode;
	  if ((features & CEPH_FEATURE_OSDENC) == 0) {
	    __u8 v = 4;
	    encode(v, bl);
	    encode(stats, bl);
	    encode(log_size, bl);
	    encode(ondisk_log_size, bl);
	    return;
	  }
	
	  ENCODE_START(7, 5, bl);
	  encode(stats, bl);
	  encode(log_size, bl);
	  encode(ondisk_log_size, bl);
	  encode(up, bl);
	  encode(acting, bl);
	  encode(store_stats, bl);
	  encode(num_store_stats, bl);
	  ENCODE_FINISH(bl);
	}
	
	void pool_stat_t::decode(ceph::buffer::list::const_iterator &bl)
	{
	  DECODE_START_LEGACY_COMPAT_LEN(7, 5, 5, bl);
	  if (struct_v >= 4) {
	    decode(stats, bl);
	    decode(log_size, bl);
	    decode(ondisk_log_size, bl);
	    if (struct_v >= 6) {
	      decode(up, bl);
	      decode(acting, bl);
	    } else {
	      up = 0;
	      acting = 0;
	    }
	    if (struct_v >= 7) {
	      decode(store_stats, bl);
	      decode(num_store_stats, bl);
	    } else {
	      store_stats.reset();
	      num_store_stats = 0;
	    }
	
	  } else {
	    decode(stats.sum.num_bytes, bl);
	    uint64_t num_kb;
	    decode(num_kb, bl);
	    decode(stats.sum.num_objects, bl);
	    decode(stats.sum.num_object_clones, bl);
	    decode(stats.sum.num_object_copies, bl);
	    decode(stats.sum.num_objects_missing_on_primary, bl);
	    decode(stats.sum.num_objects_degraded, bl);
	    decode(log_size, bl);
	    decode(ondisk_log_size, bl);
	    if (struct_v >= 2) {
	      decode(stats.sum.num_rd, bl);
	      decode(stats.sum.num_rd_kb, bl);
	      decode(stats.sum.num_wr, bl);
	      decode(stats.sum.num_wr_kb, bl);
	    }
	    if (struct_v >= 3) {
	      decode(stats.sum.num_objects_unfound, bl);
	    }
	  }
	  DECODE_FINISH(bl);
	}
	
	void pool_stat_t::generate_test_instances(list<pool_stat_t*>& o)
	{
	  pool_stat_t a;
	  o.push_back(new pool_stat_t(a));
	
	  list<object_stat_collection_t*> l;
	  object_stat_collection_t::generate_test_instances(l);
	  list<store_statfs_t*> ll;
	  store_statfs_t::generate_test_instances(ll);
	  a.stats = *l.back();
	  a.store_stats = *ll.back();
	  a.log_size = 123;
	  a.ondisk_log_size = 456;
	  a.acting = 3;
	  a.up = 4;
	  a.num_store_stats = 1;
	  o.push_back(new pool_stat_t(a));
	}
	
	
	// -- pg_history_t --
	
	void pg_history_t::encode(ceph::buffer::list &bl) const
	{
	  ENCODE_START(10, 4, bl);
	  encode(epoch_created, bl);
	  encode(last_epoch_started, bl);
	  encode(last_epoch_clean, bl);
	  encode(last_epoch_split, bl);
	  encode(same_interval_since, bl);
	  encode(same_up_since, bl);
	  encode(same_primary_since, bl);
	  encode(last_scrub, bl);
	  encode(last_scrub_stamp, bl);
	  encode(last_deep_scrub, bl);
	  encode(last_deep_scrub_stamp, bl);
	  encode(last_clean_scrub_stamp, bl);
	  encode(last_epoch_marked_full, bl);
	  encode(last_interval_started, bl);
	  encode(last_interval_clean, bl);
	  encode(epoch_pool_created, bl);
	  encode(prior_readable_until_ub, bl);
	  ENCODE_FINISH(bl);
	}
	
	void pg_history_t::decode(ceph::buffer::list::const_iterator &bl)
	{
	  DECODE_START_LEGACY_COMPAT_LEN(10, 4, 4, bl);
	  decode(epoch_created, bl);
	  decode(last_epoch_started, bl);
	  if (struct_v >= 3)
	    decode(last_epoch_clean, bl);
	  else
	    last_epoch_clean = last_epoch_started;  // careful, it's a lie!
	  decode(last_epoch_split, bl);
	  decode(same_interval_since, bl);
	  decode(same_up_since, bl);
	  decode(same_primary_since, bl);
	  if (struct_v >= 2) {
	    decode(last_scrub, bl);
	    decode(last_scrub_stamp, bl);
	  }
	  if (struct_v >= 5) {
	    decode(last_deep_scrub, bl);
	    decode(last_deep_scrub_stamp, bl);
	  }
	  if (struct_v >= 6) {
	    decode(last_clean_scrub_stamp, bl);
	  }
	  if (struct_v >= 7) {
	    decode(last_epoch_marked_full, bl);
	  }
	  if (struct_v >= 8) {
	    decode(last_interval_started, bl);
	    decode(last_interval_clean, bl);
	  } else {
	    if (last_epoch_started >= same_interval_since) {
	      last_interval_started = same_interval_since;
	    } else {
	      last_interval_started = last_epoch_started; // best guess
	    }
	    if (last_epoch_clean >= same_interval_since) {
	      last_interval_clean = same_interval_since;
	    } else {
	      last_interval_clean = last_epoch_clean; // best guess
	    }
	  }
	  if (struct_v >= 9) {
	    decode(epoch_pool_created, bl);
	  } else {
	    epoch_pool_created = epoch_created;
	  }
	  if (struct_v >= 10) {
	    decode(prior_readable_until_ub, bl);
	  }
	  DECODE_FINISH(bl);
	}
	
	void pg_history_t::dump(Formatter *f) const
	{
	  f->dump_int("epoch_created", epoch_created);
	  f->dump_int("epoch_pool_created", epoch_pool_created);
	  f->dump_int("last_epoch_started", last_epoch_started);
	  f->dump_int("last_interval_started", last_interval_started);
	  f->dump_int("last_epoch_clean", last_epoch_clean);
	  f->dump_int("last_interval_clean", last_interval_clean);
	  f->dump_int("last_epoch_split", last_epoch_split);
	  f->dump_int("last_epoch_marked_full", last_epoch_marked_full);
	  f->dump_int("same_up_since", same_up_since);
	  f->dump_int("same_interval_since", same_interval_since);
	  f->dump_int("same_primary_since", same_primary_since);
	  f->dump_stream("last_scrub") << last_scrub;
	  f->dump_stream("last_scrub_stamp") << last_scrub_stamp;
	  f->dump_stream("last_deep_scrub") << last_deep_scrub;
	  f->dump_stream("last_deep_scrub_stamp") << last_deep_scrub_stamp;
	  f->dump_stream("last_clean_scrub_stamp") << last_clean_scrub_stamp;
	  f->dump_float(
	    "prior_readable_until_ub",
	    std::chrono::duration<double>(prior_readable_until_ub).count());
	}
	
	void pg_history_t::generate_test_instances(list<pg_history_t*>& o)
	{
	  o.push_back(new pg_history_t);
	  o.push_back(new pg_history_t);
	  o.back()->epoch_created = 1;
	  o.back()->epoch_pool_created = 1;
	  o.back()->last_epoch_started = 2;
	  o.back()->last_interval_started = 2;
	  o.back()->last_epoch_clean = 3;
	  o.back()->last_interval_clean = 2;
	  o.back()->last_epoch_split = 4;
	  o.back()->prior_readable_until_ub = make_timespan(3.1415);
	  o.back()->same_up_since = 5;
	  o.back()->same_interval_since = 6;
	  o.back()->same_primary_since = 7;
	  o.back()->last_scrub = eversion_t(8, 9);
	  o.back()->last_scrub_stamp = utime_t(10, 11);
	  o.back()->last_deep_scrub = eversion_t(12, 13);
	  o.back()->last_deep_scrub_stamp = utime_t(14, 15);
	  o.back()->last_clean_scrub_stamp = utime_t(16, 17);
	  o.back()->last_epoch_marked_full = 18;
	}
	
	
	// -- pg_info_t --
	
	void pg_info_t::encode(ceph::buffer::list &bl) const
	{
	  ENCODE_START(32, 26, bl);
	  encode(pgid.pgid, bl);
	  encode(last_update, bl);
	  encode(last_complete, bl);
	  encode(log_tail, bl);
	  encode(hobject_t(), bl);  // old (nibblewise) last_backfill
	  encode(stats, bl);
	  history.encode(bl);
	  encode(purged_snaps, bl);
	  encode(last_epoch_started, bl);
	  encode(last_user_version, bl);
	  encode(hit_set, bl);
	  encode(pgid.shard, bl);
	  encode(last_backfill, bl);
	  encode(true, bl); // was last_backfill_bitwise
	  encode(last_interval_started, bl);
	  ENCODE_FINISH(bl);
	}
	
	void pg_info_t::decode(ceph::buffer::list::const_iterator &bl)
	{
	  DECODE_START(32, bl);
	  decode(pgid.pgid, bl);
	  decode(last_update, bl);
	  decode(last_complete, bl);
	  decode(log_tail, bl);
	  {
	    hobject_t old_last_backfill;
	    decode(old_last_backfill, bl);
	  }
	  decode(stats, bl);
	  history.decode(bl);
	  decode(purged_snaps, bl);
	  decode(last_epoch_started, bl);
	  decode(last_user_version, bl);
	  decode(hit_set, bl);
	  decode(pgid.shard, bl);
	  decode(last_backfill, bl);
	  {
	    bool last_backfill_bitwise;
	    decode(last_backfill_bitwise, bl);
	    // note: we may see a false value here since the default value for
	    // the member was false, so it often didn't get set to true until
	    // peering progressed.
	  }
	  if (struct_v >= 32) {
	    decode(last_interval_started, bl);
	  } else {
	    last_interval_started = last_epoch_started;
	  }
	  DECODE_FINISH(bl);
	}
	
	// -- pg_info_t --
	
	void pg_info_t::dump(Formatter *f) const
	{
	  f->dump_stream("pgid") << pgid;
	  f->dump_stream("last_update") << last_update;
	  f->dump_stream("last_complete") << last_complete;
	  f->dump_stream("log_tail") << log_tail;
	  f->dump_int("last_user_version", last_user_version);
	  f->dump_stream("last_backfill") << last_backfill;
	  f->open_array_section("purged_snaps");
	  for (interval_set<snapid_t>::const_iterator i=purged_snaps.begin();
	       i != purged_snaps.end();
	       ++i) {
	    f->open_object_section("purged_snap_interval");
	    f->dump_stream("start") << i.get_start();
	    f->dump_stream("length") << i.get_len();
	    f->close_section();
	  }
	  f->close_section();
	  f->open_object_section("history");
	  history.dump(f);
	  f->close_section();
	  f->open_object_section("stats");
	  stats.dump(f);
	  f->close_section();
	
	  f->dump_int("empty", is_empty());
	  f->dump_int("dne", dne());
	  f->dump_int("incomplete", is_incomplete());
	  f->dump_int("last_epoch_started", last_epoch_started);
	
	  f->open_object_section("hit_set_history");
	  hit_set.dump(f);
	  f->close_section();
	}
	
	void pg_info_t::generate_test_instances(list<pg_info_t*>& o)
	{
	  o.push_back(new pg_info_t);
	  o.push_back(new pg_info_t);
	  list<pg_history_t*> h;
	  pg_history_t::generate_test_instances(h);
	  o.back()->history = *h.back();
	  o.back()->pgid = spg_t(pg_t(1, 2), shard_id_t::NO_SHARD);
	  o.back()->last_update = eversion_t(3, 4);
	  o.back()->last_complete = eversion_t(5, 6);
	  o.back()->last_user_version = 2;
	  o.back()->log_tail = eversion_t(7, 8);
	  o.back()->last_backfill = hobject_t(object_t("objname"), "key", 123, 456, -1, "");
	  {
	    list<pg_stat_t*> s;
	    pg_stat_t::generate_test_instances(s);
	    o.back()->stats = *s.back();
	  }
	  {
	    list<pg_hit_set_history_t*> s;
	    pg_hit_set_history_t::generate_test_instances(s);
	    o.back()->hit_set = *s.back();
	  }
	}
	
	// -- pg_notify_t --
	void pg_notify_t::encode(ceph::buffer::list &bl) const
	{
	  ENCODE_START(3, 2, bl);
	  encode(query_epoch, bl);
	  encode(epoch_sent, bl);
	  encode(info, bl);
	  encode(to, bl);
	  encode(from, bl);
	  encode(past_intervals, bl);
	  ENCODE_FINISH(bl);
	}
	
	void pg_notify_t::decode(ceph::buffer::list::const_iterator &bl)
	{
	  DECODE_START(3, bl);
	  decode(query_epoch, bl);
	  decode(epoch_sent, bl);
	  decode(info, bl);
	  decode(to, bl);
	  decode(from, bl);
	  if (struct_v >= 3) {
	    decode(past_intervals, bl);
	  }
	  DECODE_FINISH(bl);
	}
	
	void pg_notify_t::dump(Formatter *f) const
	{
	  f->dump_int("from", from);
	  f->dump_int("to", to);
	  f->dump_unsigned("query_epoch", query_epoch);
	  f->dump_unsigned("epoch_sent", epoch_sent);
	  {
	    f->open_object_section("info");
	    info.dump(f);
	    f->close_section();
	  }
	  f->dump_object("past_intervals", past_intervals);
	}
	
	void pg_notify_t::generate_test_instances(list<pg_notify_t*>& o)
	{
	  o.push_back(new pg_notify_t(shard_id_t(3), shard_id_t::NO_SHARD, 1, 1,
				      pg_info_t(), PastIntervals()));
	  o.push_back(new pg_notify_t(shard_id_t(0), shard_id_t(0), 3, 10,
				      pg_info_t(), PastIntervals()));
	}
	
	ostream &operator<<(ostream &lhs, const pg_notify_t &notify)
	{
	  lhs << "(query:" << notify.query_epoch
	      << " sent:" << notify.epoch_sent
	      << " " << notify.info;
	  if (notify.from != shard_id_t::NO_SHARD ||
	      notify.to != shard_id_t::NO_SHARD)
	    lhs << " " << (unsigned)notify.from
		<< "->" << (unsigned)notify.to;
	  lhs << " " << notify.past_intervals;
	  return lhs << ")";
	}
	
	// -- pg_interval_t --
	
	void PastIntervals::pg_interval_t::encode(ceph::buffer::list& bl) const
	{
	  ENCODE_START(4, 2, bl);
	  encode(first, bl);
	  encode(last, bl);
	  encode(up, bl);
	  encode(acting, bl);
	  encode(maybe_went_rw, bl);
	  encode(primary, bl);
	  encode(up_primary, bl);
	  ENCODE_FINISH(bl);
	}
	
	void PastIntervals::pg_interval_t::decode(ceph::buffer::list::const_iterator& bl)
	{
	  DECODE_START_LEGACY_COMPAT_LEN(4, 2, 2, bl);
	  decode(first, bl);
	  decode(last, bl);
	  decode(up, bl);
	  decode(acting, bl);
	  decode(maybe_went_rw, bl);
	  if (struct_v >= 3) {
	    decode(primary, bl);
	  } else {
	    if (acting.size())
	      primary = acting[0];
	  }
	  if (struct_v >= 4) {
	    decode(up_primary, bl);
	  } else {
	    if (up.size())
	      up_primary = up[0];
	  }
	  DECODE_FINISH(bl);
	}
	
	void PastIntervals::pg_interval_t::dump(Formatter *f) const
	{
	  f->dump_unsigned("first", first);
	  f->dump_unsigned("last", last);
	  f->dump_int("maybe_went_rw", maybe_went_rw ? 1 : 0);
	  f->open_array_section("up");
	  for (auto p = up.cbegin(); p != up.cend(); ++p)
	    f->dump_int("osd", *p);
	  f->close_section();
	  f->open_array_section("acting");
	  for (auto p = acting.cbegin(); p != acting.cend(); ++p)
	    f->dump_int("osd", *p);
	  f->close_section();
	  f->dump_int("primary", primary);
	  f->dump_int("up_primary", up_primary);
	}
	
	void PastIntervals::pg_interval_t::generate_test_instances(list<pg_interval_t*>& o)
	{
	  o.push_back(new pg_interval_t);
	  o.push_back(new pg_interval_t);
	  o.back()->up.push_back(1);
	  o.back()->acting.push_back(2);
	  o.back()->acting.push_back(3);
	  o.back()->first = 4;
	  o.back()->last = 5;
	  o.back()->maybe_went_rw = true;
	}
	
	WRITE_CLASS_ENCODER(PastIntervals::pg_interval_t)
	
	
	/**
	 * pi_compact_rep
	 *
	 * PastIntervals only needs to be able to answer two questions:
	 * 1) Where should the primary look for unfound objects?
	 * 2) List a set of subsets of the OSDs such that contacting at least
	 *    one from each subset guarantees we speak to at least one witness
	 *    of any completed write.
	 *
	 * Crucially, 2) does not require keeping *all* past intervals.  Certainly,
	 * we don't need to keep any where maybe_went_rw would be false.  We also
	 * needn't keep two intervals where the actingset in one is a subset
	 * of the other (only need to keep the smaller of the two sets).  In order
	 * to accurately trim the set of intervals as last_epoch_started changes
	 * without rebuilding the set from scratch, we'll retain the larger set
	 * if it in an older interval.
	 */
	struct compact_interval_t {
	  epoch_t first;
	  epoch_t last;
	  set<pg_shard_t> acting;
	  bool supersedes(const compact_interval_t &other) {
	    for (auto &&i: acting) {
	      if (!other.acting.count(i))
		return false;
	    }
	    return true;
	  }
	  void dump(Formatter *f) const {
	    f->open_object_section("compact_interval_t");
	    f->dump_stream("first") << first;
	    f->dump_stream("last") << last;
	    f->dump_stream("acting") << acting;
	    f->close_section();
	  }
	  void encode(ceph::buffer::list &bl) const {
	    ENCODE_START(1, 1, bl);
	    encode(first, bl);
	    encode(last, bl);
	    encode(acting, bl);
	    ENCODE_FINISH(bl);
	  }
	  void decode(ceph::buffer::list::const_iterator &bl) {
	    DECODE_START(1, bl);
	    decode(first, bl);
	    decode(last, bl);
	    decode(acting, bl);
	    DECODE_FINISH(bl);
	  }
	  static void generate_test_instances(list<compact_interval_t*> & o) {
	    /* Not going to be used, we'll generate pi_compact_rep directly */
	  }
	};
	ostream &operator<<(ostream &o, const compact_interval_t &rhs)
	{
	  return o << "([" << rhs.first << "," << rhs.last
		   << "] acting " << rhs.acting << ")";
	}
	WRITE_CLASS_ENCODER(compact_interval_t)
	
	class pi_compact_rep : public PastIntervals::interval_rep {
	  epoch_t first = 0;
	  epoch_t last = 0; // inclusive
	  set<pg_shard_t> all_participants;
	  list<compact_interval_t> intervals;
	  pi_compact_rep(
	    bool ec_pool,
	    std::list<PastIntervals::pg_interval_t> &&intervals) {
	    for (auto &&i: intervals)
	      add_interval(ec_pool, i);
	  }
	public:
	  pi_compact_rep() = default;
	  pi_compact_rep(const pi_compact_rep &) = default;
	  pi_compact_rep(pi_compact_rep &&) = default;
	  pi_compact_rep &operator=(const pi_compact_rep &) = default;
	  pi_compact_rep &operator=(pi_compact_rep &&) = default;
	
	  size_t size() const override { return intervals.size(); }
	  bool empty() const override {
	    return first > last || (first == 0 && last == 0);
	  }
	  void clear() override {
	    *this = pi_compact_rep();
	  }
	  pair<epoch_t, epoch_t> get_bounds() const override {
	    return make_pair(first, last + 1);
	  }
	  void adjust_start_backwards(epoch_t last_epoch_clean) {
	    first = last_epoch_clean;
	  }
	
	  set<pg_shard_t> get_all_participants(
	    bool ec_pool) const override {
	    return all_participants;
	  }
	  void add_interval(
	    bool ec_pool, const PastIntervals::pg_interval_t &interval) override {
	    if (first == 0)
	      first = interval.first;
	    ceph_assert(interval.last > last);
	    last = interval.last;
	    set<pg_shard_t> acting;
	    for (unsigned i = 0; i < interval.acting.size(); ++i) {
	      if (interval.acting[i] == CRUSH_ITEM_NONE)
		continue;
	      acting.insert(
		pg_shard_t(
		  interval.acting[i],
		  ec_pool ? shard_id_t(i) : shard_id_t::NO_SHARD));
	    }
	    all_participants.insert(acting.begin(), acting.end());
	    if (!interval.maybe_went_rw)
	      return;
	    intervals.push_back(
	      compact_interval_t{interval.first, interval.last, acting});
	    auto plast = intervals.end();
	    --plast;
	    for (auto cur = intervals.begin(); cur != plast; ) {
	      if (plast->supersedes(*cur)) {
		intervals.erase(cur++);
	      } else {
		++cur;
	      }
	    }
	  }
	  unique_ptr<PastIntervals::interval_rep> clone() const override {
	    return unique_ptr<PastIntervals::interval_rep>(new pi_compact_rep(*this));
	  }
	  ostream &print(ostream &out) const override {
	    return out << "([" << first << "," << last
		       << "] all_participants=" << all_participants
		       << " intervals=" << intervals << ")";
	  }
	  void encode(ceph::buffer::list &bl) const override {
	    ENCODE_START(1, 1, bl);
	    encode(first, bl);
	    encode(last, bl);
	    encode(all_participants, bl);
	    encode(intervals, bl);
	    ENCODE_FINISH(bl);
	  }
	  void decode(ceph::buffer::list::const_iterator &bl) override {
	    DECODE_START(1, bl);
	    decode(first, bl);
	    decode(last, bl);
	    decode(all_participants, bl);
	    decode(intervals, bl);
	    DECODE_FINISH(bl);
	  }
	  void dump(Formatter *f) const override {
	    f->open_object_section("PastIntervals::compact_rep");
	    f->dump_stream("first") << first;
	    f->dump_stream("last") << last;
	    f->open_array_section("all_participants");
	    for (auto& i : all_participants) {
	      f->dump_object("pg_shard", i);
	    }
	    f->close_section();
	    f->open_array_section("intervals");
	    for (auto &&i: intervals) {
	      i.dump(f);
	    }
	    f->close_section();
	    f->close_section();
	  }
	  static void generate_test_instances(list<pi_compact_rep*> &o) {
	    using ival = PastIntervals::pg_interval_t;
	    using ivallst = std::list<ival>;
	    o.push_back(
	      new pi_compact_rep(
		true, ivallst
		{ ival{{0, 1, 2}, {0, 1, 2}, 10, 20,  true, 0, 0}
		, ival{{   1, 2}, {   1, 2}, 21, 30,  true, 1, 1}
		, ival{{      2}, {      2}, 31, 35, false, 2, 2}
		, ival{{0,    2}, {0,    2}, 36, 50,  true, 0, 0}
		}));
	    o.push_back(
	      new pi_compact_rep(
		false, ivallst
		{ ival{{0, 1, 2}, {0, 1, 2}, 10, 20,  true, 0, 0}
		, ival{{   1, 2}, {   1, 2}, 21, 30,  true, 1, 1}
		, ival{{      2}, {      2}, 31, 35, false, 2, 2}
		, ival{{0,    2}, {0,    2}, 36, 50,  true, 0, 0}
		}));
	    o.push_back(
	      new pi_compact_rep(
		true, ivallst
		{ ival{{2, 1, 0}, {2, 1, 0}, 10, 20,  true, 1, 1}
		, ival{{   0, 2}, {   0, 2}, 21, 30,  true, 0, 0}
		, ival{{   0, 2}, {2,    0}, 31, 35,  true, 2, 2}
		, ival{{   0, 2}, {   0, 2}, 36, 50,  true, 0, 0}
		}));
	  }
	  void iterate_mayberw_back_to(
	    epoch_t les,
	    std::function<void(epoch_t, const set<pg_shard_t> &)> &&f) const override {
	    for (auto i = intervals.rbegin(); i != intervals.rend(); ++i) {
	      if (i->last < les)
		break;
	      f(i->first, i->acting);
	    }
	  }
	  virtual ~pi_compact_rep() override {}
	};
	WRITE_CLASS_ENCODER(pi_compact_rep)
	
	PastIntervals::PastIntervals()
	{
	  past_intervals.reset(new pi_compact_rep);
	}
	
	PastIntervals::PastIntervals(const PastIntervals &rhs)
	  : past_intervals(rhs.past_intervals ?
			   rhs.past_intervals->clone() :
			   nullptr) {}
	
	PastIntervals &PastIntervals::operator=(const PastIntervals &rhs)
	{
	  PastIntervals other(rhs);
	  swap(other);
	  return *this;
	}
	
	ostream& operator<<(ostream& out, const PastIntervals &i)
	{
	  if (i.past_intervals) {
	    return i.past_intervals->print(out);
	  } else {
	    return out << "(empty)";
	  }
	}
	
	ostream& operator<<(ostream& out, const PastIntervals::PriorSet &i)
	{
	  return out << "PriorSet("
		     << "ec_pool: " << i.ec_pool
		     << ", probe: " << i.probe
		     << ", down: " << i.down
		     << ", blocked_by: " << i.blocked_by
		     << ", pg_down: " << i.pg_down
		     << ")";
	}
	
	void PastIntervals::decode(ceph::buffer::list::const_iterator &bl)
	{
	  DECODE_START(1, bl);
	  __u8 type = 0;
	  decode(type, bl);
	  switch (type) {
	  case 0:
	    break;
	  case 1:
	    ceph_abort_msg("pi_simple_rep support removed post-luminous");
	    break;
	  case 2:
	    past_intervals.reset(new pi_compact_rep);
	    past_intervals->decode(bl);
	    break;
	  }
	  DECODE_FINISH(bl);
	}
	
	void PastIntervals::generate_test_instances(list<PastIntervals*> &o)
	{
	  {
	    list<pi_compact_rep *> compact;
	    pi_compact_rep::generate_test_instances(compact);
	    for (auto &&i: compact) {
	      // takes ownership of contents
	      o.push_back(new PastIntervals(i));
	    }
	  }
	  return;
	}
	
	bool PastIntervals::is_new_interval(
	  int old_acting_primary,
	  int new_acting_primary,
	  const vector<int> &old_acting,
	  const vector<int> &new_acting,
	  int old_up_primary,
	  int new_up_primary,
	  const vector<int> &old_up,
	  const vector<int> &new_up,
	  int old_size,
	  int new_size,
	  int old_min_size,
	  int new_min_size,
	  unsigned old_pg_num,
	  unsigned new_pg_num,
	  unsigned old_pg_num_pending,
	  unsigned new_pg_num_pending,
	  bool old_sort_bitwise,
	  bool new_sort_bitwise,
	  bool old_recovery_deletes,
	  bool new_recovery_deletes,
	  pg_t pgid) {
	  return old_acting_primary != new_acting_primary ||
	    new_acting != old_acting ||
	    old_up_primary != new_up_primary ||
	    new_up != old_up ||
	    old_min_size != new_min_size ||
	    old_size != new_size ||
	    pgid.is_split(old_pg_num, new_pg_num, 0) ||
	    // (is or was) pre-merge source
	    pgid.is_merge_source(old_pg_num_pending, new_pg_num_pending, 0) ||
	    pgid.is_merge_source(new_pg_num_pending, old_pg_num_pending, 0) ||
	    // merge source
	    pgid.is_merge_source(old_pg_num, new_pg_num, 0) ||
	    // (is or was) pre-merge target
	    pgid.is_merge_target(old_pg_num_pending, new_pg_num_pending) ||
	    pgid.is_merge_target(new_pg_num_pending, old_pg_num_pending) ||
	    // merge target
	    pgid.is_merge_target(old_pg_num, new_pg_num) ||
	    old_sort_bitwise != new_sort_bitwise ||
	    old_recovery_deletes != new_recovery_deletes;
	}
	
	bool PastIntervals::is_new_interval(
	  int old_acting_primary,
	  int new_acting_primary,
	  const vector<int> &old_acting,
	  const vector<int> &new_acting,
	  int old_up_primary,
	  int new_up_primary,
	  const vector<int> &old_up,
	  const vector<int> &new_up,
	  const OSDMap *osdmap,
	  const OSDMap *lastmap,
	  pg_t pgid)
	{
	  const pg_pool_t *plast = lastmap->get_pg_pool(pgid.pool());
	  if (!plast) {
	    return false; // after pool is deleted there are no more interval changes
	  }
	  const pg_pool_t *pi = osdmap->get_pg_pool(pgid.pool());
	  if (!pi) {
	    return true;  // pool was deleted this epoch -> (final!) interval change
	  }
	  return
	    is_new_interval(old_acting_primary,
			    new_acting_primary,
			    old_acting,
			    new_acting,
			    old_up_primary,
			    new_up_primary,
			    old_up,
			    new_up,
			    plast->size,
			    pi->size,
			    plast->min_size,
			    pi->min_size,
			    plast->get_pg_num(),
			    pi->get_pg_num(),
			    plast->get_pg_num_pending(),
			    pi->get_pg_num_pending(),
			    lastmap->test_flag(CEPH_OSDMAP_SORTBITWISE),
			    osdmap->test_flag(CEPH_OSDMAP_SORTBITWISE),
			    lastmap->test_flag(CEPH_OSDMAP_RECOVERY_DELETES),
			    osdmap->test_flag(CEPH_OSDMAP_RECOVERY_DELETES),
			    pgid);
	}
	
	bool PastIntervals::check_new_interval(
	  int old_acting_primary,
	  int new_acting_primary,
	  const vector<int> &old_acting,
	  const vector<int> &new_acting,
	  int old_up_primary,
	  int new_up_primary,
	  const vector<int> &old_up,
	  const vector<int> &new_up,
	  epoch_t same_interval_since,
	  epoch_t last_epoch_clean,
	  const OSDMap *osdmap,
	  const OSDMap *lastmap,
	  pg_t pgid,
	  const IsPGRecoverablePredicate &could_have_gone_active,
	  PastIntervals *past_intervals,
	  std::ostream *out)
	{
	  /*
	   * We have to be careful to gracefully deal with situations like
	   * so. Say we have a power outage or something that takes out both
	   * OSDs, but the monitor doesn't mark them down in the same epoch.
	   * The history may look like
	   *
	   *  1: A B
	   *  2:   B
	   *  3:       let's say B dies for good, too (say, from the power spike) 
	   *  4: A
	   *
	   * which makes it look like B may have applied updates to the PG
	   * that we need in order to proceed.  This sucks...
	   *
	   * To minimize the risk of this happening, we CANNOT go active if
	   * _any_ OSDs in the prior set are down until we send an MOSDAlive
	   * to the monitor such that the OSDMap sets osd_up_thru to an epoch.
	   * Then, we have something like
	   *
	   *  1: A B
	   *  2:   B   up_thru[B]=0
	   *  3:
	   *  4: A
	   *
	   * -> we can ignore B, bc it couldn't have gone active (up_thru still 0).
	   *
	   * or,
	   *
	   *  1: A B
	   *  2:   B   up_thru[B]=0
	   *  3:   B   up_thru[B]=2
	   *  4:
	   *  5: A    
	   *
	   * -> we must wait for B, bc it was alive through 2, and could have
	   *    written to the pg.
	   *
	   * If B is really dead, then an administrator will need to manually
	   * intervene by marking the OSD as "lost."
	   */
	
	  // remember past interval
	  //  NOTE: a change in the up set primary triggers an interval
	  //  change, even though the interval members in the pg_interval_t
	  //  do not change.
	  ceph_assert(past_intervals);
	  ceph_assert(past_intervals->past_intervals);
	  if (is_new_interval(
		old_acting_primary,
		new_acting_primary,
		old_acting,
		new_acting,
		old_up_primary,
		new_up_primary,
		old_up,
		new_up,
		osdmap,
		lastmap,
		pgid)) {
	    pg_interval_t i;
	    i.first = same_interval_since;
	    i.last = osdmap->get_epoch() - 1;
	    ceph_assert(i.first <= i.last);
	    i.acting = old_acting;
	    i.up = old_up;
	    i.primary = old_acting_primary;
	    i.up_primary = old_up_primary;
	
	    unsigned num_acting = 0;
	    for (auto p = i.acting.cbegin(); p != i.acting.cend(); ++p)
	      if (*p != CRUSH_ITEM_NONE)
		++num_acting;
	
	    ceph_assert(lastmap->get_pools().count(pgid.pool()));
	    const pg_pool_t& old_pg_pool = lastmap->get_pools().find(pgid.pool())->second;
	    set<pg_shard_t> old_acting_shards;
	    old_pg_pool.convert_to_pg_shards(old_acting, &old_acting_shards);
	
	    if (num_acting &&
		i.primary != -1 &&
		num_acting >= old_pg_pool.min_size &&
	        could_have_gone_active(old_acting_shards)) {
	      if (out)
		*out << __func__ << " " << i
		     << " up_thru " << lastmap->get_up_thru(i.primary)
		     << " up_from " << lastmap->get_up_from(i.primary)
		     << " last_epoch_clean " << last_epoch_clean;
	      if (lastmap->get_up_thru(i.primary) >= i.first &&
		  lastmap->get_up_from(i.primary) <= i.first) {
		i.maybe_went_rw = true;
		if (out)
		  *out << " " << i
		       << " : primary up " << lastmap->get_up_from(i.primary)
		       << "-" << lastmap->get_up_thru(i.primary)
		       << " includes interval"
	               << std::endl;
	      } else if (last_epoch_clean >= i.first &&
			 last_epoch_clean <= i.last) {
		// If the last_epoch_clean is included in this interval, then
		// the pg must have been rw (for recovery to have completed).
		// This is important because we won't know the _real_
		// first_epoch because we stop at last_epoch_clean, and we
		// don't want the oldest interval to randomly have
		// maybe_went_rw false depending on the relative up_thru vs
		// last_epoch_clean timing.
		i.maybe_went_rw = true;
		if (out)
		  *out << " " << i
		       << " : includes last_epoch_clean " << last_epoch_clean
		       << " and presumed to have been rw"
		       << std::endl;
	      } else {
		i.maybe_went_rw = false;
		if (out)
		  *out << " " << i
		       << " : primary up " << lastmap->get_up_from(i.primary)
		       << "-" << lastmap->get_up_thru(i.primary)
		       << " does not include interval"
	               << std::endl;
	      }
	    } else {
	      i.maybe_went_rw = false;
	      if (out)
		*out << __func__ << " " << i << " : acting set is too small" << std::endl;
	    }
	    past_intervals->past_intervals->add_interval(old_pg_pool.is_erasure(), i);
	    return true;
	  } else {
	    return false;
	  }
	}
	
	
	// true if the given map affects the prior set
	bool PastIntervals::PriorSet::affected_by_map(
	  const OSDMap &osdmap,
	  const DoutPrefixProvider *dpp) const
	{
	  for (auto p = probe.begin(); p != probe.end(); ++p) {
	    int o = p->osd;
	
	    // did someone in the prior set go down?
	    if (osdmap.is_down(o) && down.count(o) == 0) {