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

// -*- 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) 2004-2006 Sage Weil <sage@newdream.net>
 *
 * 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 "SnapMapper.h"

#define dout_context cct
#define dout_subsys ceph_subsys_osd
#undef dout_prefix
#define dout_prefix *_dout << "snap_mapper."

using std::make_pair;
using std::map;
using std::pair;
using std::set;
using std::string;
using std::vector;

using ceph::bufferlist;
using ceph::decode;
using ceph::encode;
using ceph::timespan_str;

const string SnapMapper::LEGACY_MAPPING_PREFIX = "MAP_";
const string SnapMapper::MAPPING_PREFIX = "SNA_";
const string SnapMapper::OBJECT_PREFIX = "OBJ_";

const char *SnapMapper::PURGED_SNAP_PREFIX = "PSN_";

/*

  We have a bidirectional mapping, (1) from each snap+obj to object,
  sorted by snapshot, such that we can enumerate to identify all clones
  mapped to a particular snapshot, and (2) from object to snaps, so we
  can identify which reverse mappings exist for any given object (and,
  e.g., clean up on deletion).

  "MAP_"
  + ("%016x" % snapid)
  + "_"
  + (".%x" % shard_id)
  + "_"
  + hobject_t::to_str() ("%llx.%8x.%lx.name...." % pool, hash, snap)
  -> SnapMapping::Mapping { snap, hoid }

  "SNA_"
  + ("%lld" % poolid)
  + "_"
  + ("%016x" % snapid)
  + "_"
  + (".%x" % shard_id)
  + "_"
  + hobject_t::to_str() ("%llx.%8x.%lx.name...." % pool, hash, snap)
  -> SnapMapping::Mapping { snap, hoid }

  "OBJ_" +
  + (".%x" % shard_id)
  + hobject_t::to_str()
   -> SnapMapper::object_snaps { oid, set<snapid_t> }

  */

int OSDriver::get_keys(
  const std::set<std::string> &keys,
  std::map<std::string, bufferlist> *out)
{
  return os->omap_get_values(ch, hoid, keys, out);
}

int OSDriver::get_next(
  const std::string &key,
  pair<std::string, bufferlist> *next)
{
  ObjectMap::ObjectMapIterator iter =
    os->get_omap_iterator(ch, hoid);
  if (!iter) {
    ceph_abort();
    return -EINVAL;
  }
  iter->upper_bound(key);
  if (iter->valid()) {
    if (next)
      *next = make_pair(iter->key(), iter->value());
    return 0;
  } else {
    return -ENOENT;
  }
}

string SnapMapper::get_prefix(int64_t pool, snapid_t snap)
{
  char buf[100];
  int len = snprintf(
    buf, sizeof(buf),
    "%lld_%.*X_",
    (long long)pool,
    (int)(sizeof(snap)*2), static_cast<unsigned>(snap));
  return MAPPING_PREFIX + string(buf, len);
}

string SnapMapper::to_raw_key(
  const pair<snapid_t, hobject_t> &in)
{
  return get_prefix(in.second.pool, in.first) + shard_prefix + in.second.to_str();
}

pair<string, bufferlist> SnapMapper::to_raw(
  const pair<snapid_t, hobject_t> &in)
{
  bufferlist bl;
  encode(Mapping(in), bl);
  return make_pair(
    to_raw_key(in),
    bl);
}

pair<snapid_t, hobject_t> SnapMapper::from_raw(
  const pair<std::string, bufferlist> &image)
{
  using ceph::decode;
  Mapping map;
  bufferlist bl(image.second);
  auto bp = bl.cbegin();
  decode(map, bp);
  return make_pair(map.snap, map.hoid);
}

bool SnapMapper::is_mapping(const string &to_test)
{
  return to_test.substr(0, MAPPING_PREFIX.size()) == MAPPING_PREFIX;
}

string SnapMapper::to_object_key(const hobject_t &hoid)
{
  return OBJECT_PREFIX + shard_prefix + hoid.to_str();
}

void SnapMapper::object_snaps::encode(bufferlist &bl) const
{
  ENCODE_START(1, 1, bl);
  encode(oid, bl);
  encode(snaps, bl);
  ENCODE_FINISH(bl);
}

void SnapMapper::object_snaps::decode(bufferlist::const_iterator &bl)
{
  DECODE_START(1, bl);
  decode(oid, bl);
  decode(snaps, bl);
  DECODE_FINISH(bl);
}

bool SnapMapper::check(const hobject_t &hoid) const
{
  if (hoid.match(mask_bits, match)) {
    return true;
  }
  derr << __func__ << " " << hoid << " mask_bits " << mask_bits
       << " match 0x" << std::hex << match << std::dec << " is false"
       << dendl;
  return false;
}

int SnapMapper::get_snaps(
  const hobject_t &oid,
  object_snaps *out)
{
  ceph_assert(check(oid));
  set<string> keys;
  map<string, bufferlist> got;
  keys.insert(to_object_key(oid));
  int r = backend.get_keys(keys, &got);
  if (r < 0) {
    dout(20) << __func__ << " " << oid << " got err " << r << dendl;
    return r;
  }
  if (got.empty()) {
    dout(20) << __func__ << " " << oid << " got.empty()" << dendl;
    return -ENOENT;
  }
  if (out) {
    auto bp = got.begin()->second.cbegin();
    decode(*out, bp);
    dout(20) << __func__ << " " << oid << " " << out->snaps << dendl;
    if (out->snaps.empty()) {
      dout(1) << __func__ << " " << oid << " empty snapset" << dendl;
      ceph_assert(!cct->_conf->osd_debug_verify_snaps);
    }
  } else {
    dout(20) << __func__ << " " << oid << " (out == NULL)" << dendl;
  }
  return 0;
}

void SnapMapper::clear_snaps(
  const hobject_t &oid,
  MapCacher::Transaction<std::string, bufferlist> *t)
{
  dout(20) << __func__ << " " << oid << dendl;
  ceph_assert(check(oid));
  set<string> to_remove;
  to_remove.insert(to_object_key(oid));
  if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
    for (auto& i : to_remove) {
      dout(20) << __func__ << " rm " << i << dendl;
    }
  }
  backend.remove_keys(to_remove, t);
}

void SnapMapper::set_snaps(
  const hobject_t &oid,
  const object_snaps &in,
  MapCacher::Transaction<std::string, bufferlist> *t)
{
  ceph_assert(check(oid));
  map<string, bufferlist> to_set;
  bufferlist bl;
  encode(in, bl);
  to_set[to_object_key(oid)] = bl;
  dout(20) << __func__ << " " << oid << " " << in.snaps << dendl;
  if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
    for (auto& i : to_set) {
      dout(20) << __func__ << " set " << i.first << dendl;
    }
  }
  backend.set_keys(to_set, t);
}

int SnapMapper::update_snaps(
  const hobject_t &oid,
  const set<snapid_t> &new_snaps,
  const set<snapid_t> *old_snaps_check,
  MapCacher::Transaction<std::string, bufferlist> *t)
{
  dout(20) << __func__ << " " << oid << " " << new_snaps
	   << " was " << (old_snaps_check ? *old_snaps_check : set<snapid_t>())
	   << dendl;
  ceph_assert(check(oid));
  if (new_snaps.empty())
    return remove_oid(oid, t);

  object_snaps out;
  int r = get_snaps(oid, &out);
  // Tolerate missing keys but not disk errors
  if (r < 0 && r != -ENOENT)
    return r;
  if (old_snaps_check)
    ceph_assert(out.snaps == *old_snaps_check);

  object_snaps in(oid, new_snaps);
  set_snaps(oid, in, t);

  set<string> to_remove;
  for (set<snapid_t>::iterator i = out.snaps.begin();
       i != out.snaps.end();
       ++i) {
    if (!new_snaps.count(*i)) {
      to_remove.insert(to_raw_key(make_pair(*i, oid)));
    }
  }
  if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
    for (auto& i : to_remove) {
      dout(20) << __func__ << " rm " << i << dendl;
    }
  }
  backend.remove_keys(to_remove, t);
  return 0;
}

void SnapMapper::add_oid(
  const hobject_t &oid,
  const set<snapid_t>& snaps,
  MapCacher::Transaction<std::string, bufferlist> *t)
{
  dout(20) << __func__ << " " << oid << " " << snaps << dendl;
  ceph_assert(!snaps.empty());
  ceph_assert(check(oid));
  {
    object_snaps out;
    int r = get_snaps(oid, &out);
    if (r != -ENOENT) {
      derr << __func__ << " found existing snaps mapped on " << oid
	   << ", removing" << dendl;
      ceph_assert(!cct->_conf->osd_debug_verify_snaps);
      remove_oid(oid, t);
    }
  }

  object_snaps _snaps(oid, snaps);
  set_snaps(oid, _snaps, t);

  map<string, bufferlist> to_add;
  for (set<snapid_t>::iterator i = snaps.begin();
       i != snaps.end();
       ++i) {
    to_add.insert(to_raw(make_pair(*i, oid)));
  }
  if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
    for (auto& i : to_add) {
      dout(20) << __func__ << " set " << i.first << dendl;
    }
  }
  backend.set_keys(to_add, t);
}

int SnapMapper::get_next_objects_to_trim(
  snapid_t snap,
  unsigned max,
  vector<hobject_t> *out)
{
  ceph_assert(out);
  ceph_assert(out->empty());

  // if max would be 0, we return ENOENT and the caller would mistakenly
  // trim the snaptrim queue
  ceph_assert(max > 0);
  int r = 0;
  for (set<string>::iterator i = prefixes.begin();
       i != prefixes.end() && out->size() < max && r == 0;
       ++i) {
    string prefix(get_prefix(pool, snap) + *i);
    string pos = prefix;
    while (out->size() < max) {
      pair<string, bufferlist> next;
      r = backend.get_next(pos, &next);
      dout(20) << __func__ << " get_next(" << pos << ") returns " << r
	       << " " << next << dendl;
      if (r != 0) {
	break; // Done
      }

      if (next.first.substr(0, prefix.size()) !=
	  prefix) {
	break; // Done with this prefix
      }

      ceph_assert(is_mapping(next.first));

      dout(20) << __func__ << " " << next.first << dendl;
      pair<snapid_t, hobject_t> next_decoded(from_raw(next));
      ceph_assert(next_decoded.first == snap);
      ceph_assert(check(next_decoded.second));

      out->push_back(next_decoded.second);
      pos = next.first;
    }
  }
  if (out->size() == 0) {
    return -ENOENT;
  } else {
    return 0;
  }
}


int SnapMapper::remove_oid(
  const hobject_t &oid,
  MapCacher::Transaction<std::string, bufferlist> *t)
{
  dout(20) << __func__ << " " << oid << dendl;
  ceph_assert(check(oid));
  return _remove_oid(oid, t);
}

int SnapMapper::_remove_oid(
  const hobject_t &oid,
  MapCacher::Transaction<std::string, bufferlist> *t)
{
  dout(20) << __func__ << " " << oid << dendl;
  object_snaps out;
  int r = get_snaps(oid, &out);
  if (r < 0)
    return r;

  clear_snaps(oid, t);

  set<string> to_remove;
  for (set<snapid_t>::iterator i = out.snaps.begin();
       i != out.snaps.end();
       ++i) {
    to_remove.insert(to_raw_key(make_pair(*i, oid)));
  }
  if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
    for (auto& i : to_remove) {
      dout(20) << __func__ << " rm " << i << dendl;
    }
  }
  backend.remove_keys(to_remove, t);
  return 0;
}

int SnapMapper::get_snaps(
  const hobject_t &oid,
  std::set<snapid_t> *snaps)
{
  ceph_assert(check(oid));
  object_snaps out;
  int r = get_snaps(oid, &out);
  if (r < 0)
    return r;
  if (snaps)
    snaps->swap(out.snaps);
  return 0;
}


// -- purged snaps --

string SnapMapper::make_purged_snap_key(int64_t pool, snapid_t last)
{
  char k[80];
  snprintf(k, sizeof(k), "%s_%llu_%016llx", PURGED_SNAP_PREFIX,
	   (unsigned long long)pool, (unsigned long long)last);
  return k;
}

void SnapMapper::make_purged_snap_key_value(
  int64_t pool, snapid_t begin, snapid_t end, map<string,bufferlist> *m)
{
  string k = make_purged_snap_key(pool, end - 1);
  auto& v = (*m)[k];
  ceph::encode(pool, v);
  ceph::encode(begin, v);
  ceph::encode(end, v);
}

int SnapMapper::_lookup_purged_snap(
  CephContext *cct,
  ObjectStore *store,
  ObjectStore::CollectionHandle& ch,
  const ghobject_t& hoid,
  int64_t pool, snapid_t snap,
  snapid_t *begin, snapid_t *end)
{
  string k = make_purged_snap_key(pool, snap);
  auto it = store->get_omap_iterator(ch, hoid);
  it->lower_bound(k);
  if (!it->valid()) {
    dout(20) << __func__ << " pool " << pool << " snap " << snap
	     << " key '" << k << "' lower_bound not found" << dendl;
    return -ENOENT;
  }
  if (it->key().find(PURGED_SNAP_PREFIX) != 0) {
    dout(20) << __func__ << " pool " << pool << " snap " << snap
	     << " key '" << k << "' lower_bound got mismatched prefix '"
	     << it->key() << "'" << dendl;
    return -ENOENT;
  }
  bufferlist v = it->value();
  auto p = v.cbegin();
  int64_t gotpool;
  decode(gotpool, p);
  decode(*begin, p);
  decode(*end, p);
  if (snap < *begin || snap >= *end) {
    dout(20) << __func__ << " pool " << pool << " snap " << snap
	     << " found [" << *begin << "," << *end << "), no overlap" << dendl;
    return -ENOENT;
  }
  return 0;
}

void SnapMapper::record_purged_snaps(
  CephContext *cct,
  ObjectStore *store,
  ObjectStore::CollectionHandle& ch,
  ghobject_t hoid,
  ObjectStore::Transaction *t,
  map<epoch_t,mempool::osdmap::map<int64_t,snap_interval_set_t>> purged_snaps)
{
  dout(10) << __func__ << " purged_snaps " << purged_snaps << dendl;
  map<string,bufferlist> m;
  set<string> rm;
  for (auto& [epoch, bypool] : purged_snaps) {
    // index by (pool, snap)
    for (auto& [pool, snaps] : bypool) {
      for (auto i = snaps.begin();
	   i != snaps.end();
	   ++i) {
	snapid_t begin = i.get_start();
	snapid_t end = i.get_end();
	snapid_t before_begin, before_end;
	snapid_t after_begin, after_end;
	int b = _lookup_purged_snap(cct, store, ch, hoid,
				    pool, begin - 1, &before_begin, &before_end);
	int a = _lookup_purged_snap(cct, store, ch, hoid,
				    pool, end, &after_begin, &after_end);
	if (!b && !a) {
	  dout(10) << __func__
		   << " [" << begin << "," << end << ") - joins ["
		   << before_begin << "," << before_end << ") and ["
		   << after_begin << "," << after_end << ")" << dendl;
	  // erase only the begin record; we'll overwrite the end one
	  rm.insert(make_purged_snap_key(pool, before_end - 1));
	  make_purged_snap_key_value(pool, before_begin, after_end, &m);
	} else if (!b) {
	  dout(10) << __func__
		   << " [" << begin << "," << end << ") - join with earlier ["
		   << before_begin << "," << before_end << ")" << dendl;
	  rm.insert(make_purged_snap_key(pool, before_end - 1));
	  make_purged_snap_key_value(pool, before_begin, end, &m);
	} else if (!a) {
	  dout(10) << __func__
		   << " [" << begin << "," << end << ") - join with later ["
		   << after_begin << "," << after_end << ")" << dendl;
	  // overwrite after record
	  make_purged_snap_key_value(pool, begin, after_end, &m);
	} else {
	  make_purged_snap_key_value(pool, begin, end, &m);
	}
      }
    }
  }
  t->omap_rmkeys(ch->cid, hoid, rm);
  t->omap_setkeys(ch->cid, hoid, m);
  dout(10) << __func__ << " rm " << rm.size() << " keys, set " << m.size()
	   << " keys" << dendl;
}


bool SnapMapper::Scrubber::_parse_p()
{
  if (!psit->valid()) {
    pool = -1;
    return false;
  }
  if (psit->key().find(PURGED_SNAP_PREFIX) != 0) {
    pool = -1;
    return false;
  }
  bufferlist v = psit->value();
  auto p = v.cbegin();
  ceph::decode(pool, p);
  ceph::decode(begin, p);
  ceph::decode(end, p);
  dout(20) << __func__ << " purged_snaps pool " << pool
	   << " [" << begin << "," << end << ")" << dendl;
  psit->next();
  return true;
}

bool SnapMapper::Scrubber::_parse_m()
{
  if (!mapit->valid()) {
    return false;
  }
  if (mapit->key().find(MAPPING_PREFIX) != 0) {
    return false;
  }
  auto v = mapit->value();
  auto p = v.cbegin();
  mapping.decode(p);

  {
    unsigned long long p, s;
    long sh;
    string k = mapit->key();
    int r = sscanf(k.c_str(), "SNA_%lld_%llx.%lx", &p, &s, &sh);
    if (r != 1) {
      shard = shard_id_t::NO_SHARD;
    } else {
      shard = shard_id_t(sh);
    }
  }
  dout(20) << __func__ << " mapping pool " << mapping.hoid.pool
	   << " snap " << mapping.snap
	   << " shard " << shard
	   << " " << mapping.hoid << dendl;
  mapit->next();
  return true;
}

void SnapMapper::Scrubber::run()
{
  dout(10) << __func__ << dendl;

  psit = store->get_omap_iterator(ch, purged_snaps_hoid);
  psit->upper_bound(PURGED_SNAP_PREFIX);
  _parse_p();

  mapit = store->get_omap_iterator(ch, mapping_hoid);
  mapit->upper_bound(MAPPING_PREFIX);

  while (_parse_m()) {
    // advance to next purged_snaps range?
    while (pool >= 0 &&
	   (mapping.hoid.pool > pool ||
	    (mapping.hoid.pool == pool && mapping.snap >= end))) {
      _parse_p();
    }
    if (pool < 0) {
      dout(10) << __func__ << " passed final purged_snaps interval, rest ok"
	       << dendl;
      break;
    }
    if (mapping.hoid.pool < pool ||
	mapping.snap < begin) {
      // ok
      dout(20) << __func__ << " ok " << mapping.hoid
	       << " snap " << mapping.snap
	       << " precedes pool " << pool
	       << " purged_snaps [" << begin << "," << end << ")" << dendl;
    } else {
      assert(mapping.snap >= begin &&
	     mapping.snap < end &&
	     mapping.hoid.pool == pool);
      // invalid
      dout(10) << __func__ << " stray " << mapping.hoid
	       << " snap " << mapping.snap
	       << " in pool " << pool
	       << " shard " << shard
	       << " purged_snaps [" << begin << "," << end << ")" << dendl;
      stray.emplace_back(std::tuple<int64_t,snapid_t,uint32_t,shard_id_t>(
			   pool, mapping.snap, mapping.hoid.get_hash(),
			   shard
			   ));
    }
  }

  dout(10) << __func__ << " end, found " << stray.size() << " stray" << dendl;
  psit = ObjectMap::ObjectMapIterator();
  mapit = ObjectMap::ObjectMapIterator();
}


// -------------------------------------
// legacy conversion/support

string SnapMapper::get_legacy_prefix(snapid_t snap)
{
  char buf[100];
  int len = snprintf(
    buf, sizeof(buf),
    "%.*X_",
    (int)(sizeof(snap)*2), static_cast<unsigned>(snap));
  return LEGACY_MAPPING_PREFIX + string(buf, len);
}

string SnapMapper::to_legacy_raw_key(
  const pair<snapid_t, hobject_t> &in)
{
  return get_legacy_prefix(in.first) + shard_prefix + in.second.to_str();
}

bool SnapMapper::is_legacy_mapping(const string &to_test)
{
  return to_test.substr(0, LEGACY_MAPPING_PREFIX.size()) ==
    LEGACY_MAPPING_PREFIX;
}

int SnapMapper::convert_legacy(
  CephContext *cct,
  ObjectStore *store,
  ObjectStore::CollectionHandle& ch,
  ghobject_t hoid,
  unsigned max)
{
  uint64_t n = 0;

  ObjectMap::ObjectMapIterator iter = store->get_omap_iterator(ch, hoid);
  if (!iter) {
    return -EIO;
  }

  auto start = ceph::mono_clock::now();

  iter->upper_bound(SnapMapper::LEGACY_MAPPING_PREFIX);
  map<string,bufferlist> to_set;
  while (iter->valid()) {
    bool valid = SnapMapper::is_legacy_mapping(iter->key());
    if (valid) {
      SnapMapper::Mapping m;
      bufferlist bl(iter->value());
      auto bp = bl.cbegin();
      decode(m, bp);
      to_set.emplace(
	SnapMapper::get_prefix(m.hoid.pool, m.snap),
	bl);
      ++n;
      iter->next();
    }
    if (!valid || !iter->valid() || to_set.size() >= max) {
      ObjectStore::Transaction t;
      t.omap_setkeys(ch->cid, hoid, to_set);
      int r = store->queue_transaction(ch, std::move(t));
      ceph_assert(r == 0);
      to_set.clear();
      if (!valid) {
	break;
      }
      dout(10) << __func__ << " converted " << n << " keys" << dendl;
    }
  }

  auto end = ceph::mono_clock::now();

  dout(1) << __func__ << " converted " << n << " keys in "
	  << timespan_str(end - start) << dendl;

  // remove the old keys
  {
    ObjectStore::Transaction t;
    string end = SnapMapper::LEGACY_MAPPING_PREFIX;
    ++end[end.size()-1]; // turn _ to whatever comes after _
    t.omap_rmkeyrange(ch->cid, hoid,
		      SnapMapper::LEGACY_MAPPING_PREFIX,
		      end);
    int r = store->queue_transaction(ch, std::move(t));
    ceph_assert(r == 0);
  }
  return 0;
}
Commit	Line	Data
7c673cae FG	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab
	3	/*
	4	* Ceph - scalable distributed file system
	5	*
	6	* Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
	7	*
	8	* This is free software; you can redistribute it and/or
	9	* modify it under the terms of the GNU Lesser General Public
	10	* License version 2.1, as published by the Free Software
	11	* Foundation. See file COPYING.
	12	*
	13	*/
	14
	15	#include "SnapMapper.h"
	16
	17	#define dout_context cct
	18	#define dout_subsys ceph_subsys_osd
	19	#undef dout_prefix
	20	#define dout_prefix *_dout << "snap_mapper."
	21
f67539c2 TL	22	using std::make_pair;
	23	using std::map;
	24	using std::pair;
	25	using std::set;
7c673cae	26	using std::string;
f67539c2 TL	27	using std::vector;
	28
	29	using ceph::bufferlist;
	30	using ceph::decode;
	31	using ceph::encode;
	32	using ceph::timespan_str;
7c673cae	33
9f95a23c TL	34	const string SnapMapper::LEGACY_MAPPING_PREFIX = "MAP_";
9f95a23c TL	35	const string SnapMapper::MAPPING_PREFIX = "SNA_";
7c673cae FG	36	const string SnapMapper::OBJECT_PREFIX = "OBJ_";
7c673cae FG	37
9f95a23c TL	38	const char *SnapMapper::PURGED_SNAP_PREFIX = "PSN_";
	39
	40	/*
	41
	42	We have a bidirectional mapping, (1) from each snap+obj to object,
	43	sorted by snapshot, such that we can enumerate to identify all clones
	44	mapped to a particular snapshot, and (2) from object to snaps, so we
	45	can identify which reverse mappings exist for any given object (and,
	46	e.g., clean up on deletion).
	47
	48	"MAP_"
	49	+ ("%016x" % snapid)
	50	+ "_"
	51	+ (".%x" % shard_id)
	52	+ "_"
	53	+ hobject_t::to_str() ("%llx.%8x.%lx.name...." % pool, hash, snap)
	54	-> SnapMapping::Mapping { snap, hoid }
	55
	56	"SNA_"
	57	+ ("%lld" % poolid)
	58	+ "_"
	59	+ ("%016x" % snapid)
	60	+ "_"
	61	+ (".%x" % shard_id)
	62	+ "_"
	63	+ hobject_t::to_str() ("%llx.%8x.%lx.name...." % pool, hash, snap)
	64	-> SnapMapping::Mapping { snap, hoid }
	65
	66	"OBJ_" +
	67	+ (".%x" % shard_id)
	68	+ hobject_t::to_str()
	69	-> SnapMapper::object_snaps { oid, set<snapid_t> }
	70
	71	*/
	72
7c673cae FG	73	int OSDriver::get_keys(
	74	const std::set<std::string> &keys,
	75	std::map<std::string, bufferlist> *out)
	76	{
11fdf7f2	77	return os->omap_get_values(ch, hoid, keys, out);
7c673cae FG	78	}
	79
	80	int OSDriver::get_next(
	81	const std::string &key,
	82	pair<std::string, bufferlist> *next)
	83	{
	84	ObjectMap::ObjectMapIterator iter =
11fdf7f2	85	os->get_omap_iterator(ch, hoid);
7c673cae FG	86	if (!iter) {
	87	ceph_abort();
	88	return -EINVAL;
	89	}
	90	iter->upper_bound(key);
	91	if (iter->valid()) {
	92	if (next)
	93	*next = make_pair(iter->key(), iter->value());
	94	return 0;
	95	} else {
	96	return -ENOENT;
	97	}
	98	}
	99
9f95a23c	100	string SnapMapper::get_prefix(int64_t pool, snapid_t snap)
7c673cae FG	101	{
	102	char buf[100];
	103	int len = snprintf(
	104	buf, sizeof(buf),
9f95a23c TL	105	"%lld_%.*X_",
	106	(long long)pool,
	107	(int)(sizeof(snap)*2), static_cast<unsigned>(snap));
7c673cae FG	108	return MAPPING_PREFIX + string(buf, len);
	109	}
	110
	111	string SnapMapper::to_raw_key(
	112	const pair<snapid_t, hobject_t> &in)
	113	{
9f95a23c	114	return get_prefix(in.second.pool, in.first) + shard_prefix + in.second.to_str();
7c673cae FG	115	}
	116
	117	pair<string, bufferlist> SnapMapper::to_raw(
	118	const pair<snapid_t, hobject_t> &in)
	119	{
	120	bufferlist bl;
11fdf7f2	121	encode(Mapping(in), bl);
7c673cae FG	122	return make_pair(
	123	to_raw_key(in),
	124	bl);
	125	}
	126
	127	pair<snapid_t, hobject_t> SnapMapper::from_raw(
	128	const pair<std::string, bufferlist> &image)
	129	{
9f95a23c	130	using ceph::decode;
7c673cae FG	131	Mapping map;
7c673cae FG	132	bufferlist bl(image.second);
11fdf7f2 TL	133	auto bp = bl.cbegin();
11fdf7f2 TL	134	decode(map, bp);
7c673cae FG	135	return make_pair(map.snap, map.hoid);
	136	}
	137
	138	bool SnapMapper::is_mapping(const string &to_test)
	139	{
	140	return to_test.substr(0, MAPPING_PREFIX.size()) == MAPPING_PREFIX;
	141	}
	142
	143	string SnapMapper::to_object_key(const hobject_t &hoid)
	144	{
	145	return OBJECT_PREFIX + shard_prefix + hoid.to_str();
	146	}
	147
	148	void SnapMapper::object_snaps::encode(bufferlist &bl) const
	149	{
	150	ENCODE_START(1, 1, bl);
11fdf7f2 TL	151	encode(oid, bl);
11fdf7f2 TL	152	encode(snaps, bl);
7c673cae FG	153	ENCODE_FINISH(bl);
	154	}
	155
11fdf7f2	156	void SnapMapper::object_snaps::decode(bufferlist::const_iterator &bl)
7c673cae FG	157	{
7c673cae FG	158	DECODE_START(1, bl);
11fdf7f2 TL	159	decode(oid, bl);
11fdf7f2 TL	160	decode(snaps, bl);
7c673cae FG	161	DECODE_FINISH(bl);
	162	}
	163
11fdf7f2 TL	164	bool SnapMapper::check(const hobject_t &hoid) const
	165	{
	166	if (hoid.match(mask_bits, match)) {
	167	return true;
	168	}
	169	derr << __func__ << " " << hoid << " mask_bits " << mask_bits
	170	<< " match 0x" << std::hex << match << std::dec << " is false"
	171	<< dendl;
	172	return false;
	173	}
	174
7c673cae FG	175	int SnapMapper::get_snaps(
	176	const hobject_t &oid,
	177	object_snaps *out)
	178	{
11fdf7f2	179	ceph_assert(check(oid));
7c673cae FG	180	set<string> keys;
	181	map<string, bufferlist> got;
	182	keys.insert(to_object_key(oid));
	183	int r = backend.get_keys(keys, &got);
11fdf7f2 TL	184	if (r < 0) {
11fdf7f2 TL	185	dout(20) << __func__ << " " << oid << " got err " << r << dendl;
7c673cae	186	return r;
11fdf7f2 TL	187	}
	188	if (got.empty()) {
	189	dout(20) << __func__ << " " << oid << " got.empty()" << dendl;
7c673cae	190	return -ENOENT;
11fdf7f2	191	}
7c673cae	192	if (out) {
11fdf7f2 TL	193	auto bp = got.begin()->second.cbegin();
11fdf7f2 TL	194	decode(*out, bp);
7c673cae	195	dout(20) << __func__ << " " << oid << " " << out->snaps << dendl;
28e407b8 AA	196	if (out->snaps.empty()) {
28e407b8 AA	197	dout(1) << __func__ << " " << oid << " empty snapset" << dendl;
11fdf7f2	198	ceph_assert(!cct->_conf->osd_debug_verify_snaps);
28e407b8	199	}
7c673cae FG	200	} else {
	201	dout(20) << __func__ << " " << oid << " (out == NULL)" << dendl;
	202	}
	203	return 0;
	204	}
	205
	206	void SnapMapper::clear_snaps(
	207	const hobject_t &oid,
	208	MapCacher::Transaction<std::string, bufferlist> *t)
	209	{
	210	dout(20) << __func__ << " " << oid << dendl;
11fdf7f2	211	ceph_assert(check(oid));
7c673cae FG	212	set<string> to_remove;
7c673cae FG	213	to_remove.insert(to_object_key(oid));
11fdf7f2	214	if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
31f18b77 FG	215	for (auto& i : to_remove) {
	216	dout(20) << __func__ << " rm " << i << dendl;
	217	}
	218	}
7c673cae FG	219	backend.remove_keys(to_remove, t);
	220	}
	221
	222	void SnapMapper::set_snaps(
	223	const hobject_t &oid,
	224	const object_snaps &in,
	225	MapCacher::Transaction<std::string, bufferlist> *t)
	226	{
11fdf7f2	227	ceph_assert(check(oid));
7c673cae FG	228	map<string, bufferlist> to_set;
7c673cae FG	229	bufferlist bl;
11fdf7f2	230	encode(in, bl);
7c673cae FG	231	to_set[to_object_key(oid)] = bl;
7c673cae FG	232	dout(20) << __func__ << " " << oid << " " << in.snaps << dendl;
11fdf7f2	233	if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
31f18b77 FG	234	for (auto& i : to_set) {
	235	dout(20) << __func__ << " set " << i.first << dendl;
	236	}
	237	}
7c673cae FG	238	backend.set_keys(to_set, t);
	239	}
	240
	241	int SnapMapper::update_snaps(
	242	const hobject_t &oid,
	243	const set<snapid_t> &new_snaps,
	244	const set<snapid_t> *old_snaps_check,
	245	MapCacher::Transaction<std::string, bufferlist> *t)
	246	{
	247	dout(20) << __func__ << " " << oid << " " << new_snaps
	248	<< " was " << (old_snaps_check ? *old_snaps_check : set<snapid_t>())
	249	<< dendl;
11fdf7f2	250	ceph_assert(check(oid));
7c673cae FG	251	if (new_snaps.empty())
	252	return remove_oid(oid, t);
	253
	254	object_snaps out;
	255	int r = get_snaps(oid, &out);
11fdf7f2 TL	256	// Tolerate missing keys but not disk errors
11fdf7f2 TL	257	if (r < 0 && r != -ENOENT)
7c673cae FG	258	return r;
7c673cae FG	259	if (old_snaps_check)
11fdf7f2	260	ceph_assert(out.snaps == *old_snaps_check);
7c673cae FG	261
	262	object_snaps in(oid, new_snaps);
	263	set_snaps(oid, in, t);
	264
	265	set<string> to_remove;
	266	for (set<snapid_t>::iterator i = out.snaps.begin();
	267	i != out.snaps.end();
	268	++i) {
	269	if (!new_snaps.count(*i)) {
	270	to_remove.insert(to_raw_key(make_pair(*i, oid)));
	271	}
	272	}
11fdf7f2	273	if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
31f18b77 FG	274	for (auto& i : to_remove) {
	275	dout(20) << __func__ << " rm " << i << dendl;
	276	}
	277	}
7c673cae FG	278	backend.remove_keys(to_remove, t);
	279	return 0;
	280	}
	281
	282	void SnapMapper::add_oid(
	283	const hobject_t &oid,
	284	const set<snapid_t>& snaps,
	285	MapCacher::Transaction<std::string, bufferlist> *t)
	286	{
	287	dout(20) << __func__ << " " << oid << " " << snaps << dendl;
11fdf7f2 TL	288	ceph_assert(!snaps.empty());
11fdf7f2 TL	289	ceph_assert(check(oid));
7c673cae FG	290	{
	291	object_snaps out;
	292	int r = get_snaps(oid, &out);
94b18763 FG	293	if (r != -ENOENT) {
	294	derr << __func__ << " found existing snaps mapped on " << oid
	295	<< ", removing" << dendl;
11fdf7f2	296	ceph_assert(!cct->_conf->osd_debug_verify_snaps);
94b18763 FG	297	remove_oid(oid, t);
94b18763 FG	298	}
7c673cae FG	299	}
	300
	301	object_snaps _snaps(oid, snaps);
	302	set_snaps(oid, _snaps, t);
	303
	304	map<string, bufferlist> to_add;
	305	for (set<snapid_t>::iterator i = snaps.begin();
	306	i != snaps.end();
	307	++i) {
	308	to_add.insert(to_raw(make_pair(*i, oid)));
	309	}
11fdf7f2	310	if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
31f18b77 FG	311	for (auto& i : to_add) {
	312	dout(20) << __func__ << " set " << i.first << dendl;
	313	}
	314	}
7c673cae FG	315	backend.set_keys(to_add, t);
	316	}
	317
	318	int SnapMapper::get_next_objects_to_trim(
	319	snapid_t snap,
	320	unsigned max,
	321	vector<hobject_t> *out)
	322	{
11fdf7f2 TL	323	ceph_assert(out);
11fdf7f2 TL	324	ceph_assert(out->empty());
1d09f67e TL	325
	326	// if max would be 0, we return ENOENT and the caller would mistakenly
	327	// trim the snaptrim queue
	328	ceph_assert(max > 0);
7c673cae FG	329	int r = 0;
	330	for (set<string>::iterator i = prefixes.begin();
	331	i != prefixes.end() && out->size() < max && r == 0;
	332	++i) {
9f95a23c	333	string prefix(get_prefix(pool, snap) + *i);
7c673cae FG	334	string pos = prefix;
	335	while (out->size() < max) {
	336	pair<string, bufferlist> next;
	337	r = backend.get_next(pos, &next);
31f18b77 FG	338	dout(20) << __func__ << " get_next(" << pos << ") returns " << r
31f18b77 FG	339	<< " " << next << dendl;
7c673cae FG	340	if (r != 0) {
	341	break; // Done
	342	}
	343
	344	if (next.first.substr(0, prefix.size()) !=
	345	prefix) {
	346	break; // Done with this prefix
	347	}
	348
11fdf7f2	349	ceph_assert(is_mapping(next.first));
7c673cae	350
31f18b77	351	dout(20) << __func__ << " " << next.first << dendl;
7c673cae	352	pair<snapid_t, hobject_t> next_decoded(from_raw(next));
11fdf7f2 TL	353	ceph_assert(next_decoded.first == snap);
11fdf7f2 TL	354	ceph_assert(check(next_decoded.second));
7c673cae FG	355
	356	out->push_back(next_decoded.second);
	357	pos = next.first;
	358	}
	359	}
	360	if (out->size() == 0) {
	361	return -ENOENT;
	362	} else {
	363	return 0;
	364	}
	365	}
	366
	367
	368	int SnapMapper::remove_oid(
	369	const hobject_t &oid,
	370	MapCacher::Transaction<std::string, bufferlist> *t)
	371	{
	372	dout(20) << __func__ << " " << oid << dendl;
11fdf7f2	373	ceph_assert(check(oid));
7c673cae FG	374	return _remove_oid(oid, t);
	375	}
	376
	377	int SnapMapper::_remove_oid(
	378	const hobject_t &oid,
	379	MapCacher::Transaction<std::string, bufferlist> *t)
	380	{
	381	dout(20) << __func__ << " " << oid << dendl;
	382	object_snaps out;
	383	int r = get_snaps(oid, &out);
	384	if (r < 0)
	385	return r;
	386
	387	clear_snaps(oid, t);
	388
	389	set<string> to_remove;
	390	for (set<snapid_t>::iterator i = out.snaps.begin();
	391	i != out.snaps.end();
	392	++i) {
	393	to_remove.insert(to_raw_key(make_pair(*i, oid)));
	394	}
11fdf7f2	395	if (g_conf()->subsys.should_gather<ceph_subsys_osd, 20>()) {
31f18b77 FG	396	for (auto& i : to_remove) {
	397	dout(20) << __func__ << " rm " << i << dendl;
	398	}
	399	}
7c673cae FG	400	backend.remove_keys(to_remove, t);
	401	return 0;
	402	}
	403
	404	int SnapMapper::get_snaps(
	405	const hobject_t &oid,
	406	std::set<snapid_t> *snaps)
	407	{
11fdf7f2	408	ceph_assert(check(oid));
7c673cae FG	409	object_snaps out;
	410	int r = get_snaps(oid, &out);
	411	if (r < 0)
	412	return r;
	413	if (snaps)
	414	snaps->swap(out.snaps);
	415	return 0;
	416	}
9f95a23c TL	417
	418
	419	// -- purged snaps --
	420
	421	string SnapMapper::make_purged_snap_key(int64_t pool, snapid_t last)
	422	{
	423	char k[80];
	424	snprintf(k, sizeof(k), "%s_%llu_%016llx", PURGED_SNAP_PREFIX,
	425	(unsigned long long)pool, (unsigned long long)last);
	426	return k;
	427	}
	428
	429	void SnapMapper::make_purged_snap_key_value(
	430	int64_t pool, snapid_t begin, snapid_t end, map<string,bufferlist> *m)
	431	{
	432	string k = make_purged_snap_key(pool, end - 1);
	433	auto& v = (*m)[k];
	434	ceph::encode(pool, v);
	435	ceph::encode(begin, v);
	436	ceph::encode(end, v);
	437	}
	438
	439	int SnapMapper::_lookup_purged_snap(
	440	CephContext *cct,
	441	ObjectStore *store,
	442	ObjectStore::CollectionHandle& ch,
	443	const ghobject_t& hoid,
	444	int64_t pool, snapid_t snap,
	445	snapid_t begin, snapid_t end)
	446	{
	447	string k = make_purged_snap_key(pool, snap);
	448	auto it = store->get_omap_iterator(ch, hoid);
	449	it->lower_bound(k);
	450	if (!it->valid()) {
	451	dout(20) << __func__ << " pool " << pool << " snap " << snap
	452	<< " key '" << k << "' lower_bound not found" << dendl;
	453	return -ENOENT;
	454	}
	455	if (it->key().find(PURGED_SNAP_PREFIX) != 0) {
	456	dout(20) << __func__ << " pool " << pool << " snap " << snap
	457	<< " key '" << k << "' lower_bound got mismatched prefix '"
	458	<< it->key() << "'" << dendl;
	459	return -ENOENT;
	460	}
	461	bufferlist v = it->value();
	462	auto p = v.cbegin();
	463	int64_t gotpool;
	464	decode(gotpool, p);
	465	decode(*begin, p);
	466	decode(*end, p);
	467	if (snap < begin \|\| snap >= end) {
	468	dout(20) << __func__ << " pool " << pool << " snap " << snap
	469	<< " found [" << begin << "," << end << "), no overlap" << dendl;
	470	return -ENOENT;
	471	}
	472	return 0;
	473	}
	474
	475	void SnapMapper::record_purged_snaps(
	476	CephContext *cct,
	477	ObjectStore *store,
	478	ObjectStore::CollectionHandle& ch,
	479	ghobject_t hoid,
	480	ObjectStore::Transaction *t,
481	map<epoch_t,mempool::osdmap::map<int64_t,snap_interval_set_t>> purged_snaps)
482	{
483	dout(10) << __func__ << " purged_snaps " << purged_snaps << dendl;
484	map<string,bufferlist> m;
485	set<string> rm;
486	for (auto& [epoch, bypool] : purged_snaps) {
487	// index by (pool, snap)
488	for (auto& [pool, snaps] : bypool) {
489	for (auto i = snaps.begin();
490	i != snaps.end();
491	++i) {
492	snapid_t begin = i.get_start();
493	snapid_t end = i.get_end();
494	snapid_t before_begin, before_end;
495	snapid_t after_begin, after_end;
496	int b = _lookup_purged_snap(cct, store, ch, hoid,
497	pool, begin - 1, &before_begin, &before_end);
498	int a = _lookup_purged_snap(cct, store, ch, hoid,
499	pool, end, &after_begin, &after_end);
500	if (!b && !a) {
501	dout(10) << __func__
502	<< " [" << begin << "," << end << ") - joins ["
503	<< before_begin << "," << before_end << ") and ["
504	<< after_begin << "," << after_end << ")" << dendl;
505	// erase only the begin record; we'll overwrite the end one
506	rm.insert(make_purged_snap_key(pool, before_end - 1));
507	make_purged_snap_key_value(pool, before_begin, after_end, &m);
508	} else if (!b) {
509	dout(10) << __func__
510	<< " [" << begin << "," << end << ") - join with earlier ["
511	<< before_begin << "," << before_end << ")" << dendl;
512	rm.insert(make_purged_snap_key(pool, before_end - 1));
513	make_purged_snap_key_value(pool, before_begin, end, &m);
514	} else if (!a) {
515	dout(10) << __func__
516	<< " [" << begin << "," << end << ") - join with later ["
517	<< after_begin << "," << after_end << ")" << dendl;
518	// overwrite after record
519	make_purged_snap_key_value(pool, begin, after_end, &m);
520	} else {
521	make_purged_snap_key_value(pool, begin, end, &m);
522	}
523	}
524	}
525	}
526	t->omap_rmkeys(ch->cid, hoid, rm);
527	t->omap_setkeys(ch->cid, hoid, m);
528	dout(10) << __func__ << " rm " << rm.size() << " keys, set " << m.size()
529	<< " keys" << dendl;
530	}
531
532
533	bool SnapMapper::Scrubber::_parse_p()
534	{
535	if (!psit->valid()) {
536	pool = -1;
537	return false;
538	}
539	if (psit->key().find(PURGED_SNAP_PREFIX) != 0) {
540	pool = -1;
541	return false;
542	}
543	bufferlist v = psit->value();
544	auto p = v.cbegin();
545	ceph::decode(pool, p);
546	ceph::decode(begin, p);
547	ceph::decode(end, p);
548	dout(20) << __func__ << " purged_snaps pool " << pool
549	<< " [" << begin << "," << end << ")" << dendl;
550	psit->next();
551	return true;
552	}
553
554	bool SnapMapper::Scrubber::_parse_m()
555	{
556	if (!mapit->valid()) {
557	return false;
558	}
559	if (mapit->key().find(MAPPING_PREFIX) != 0) {
560	return false;
561	}
562	auto v = mapit->value();
563	auto p = v.cbegin();
564	mapping.decode(p);
565
566	{
567	unsigned long long p, s;
568	long sh;
569	string k = mapit->key();
570	int r = sscanf(k.c_str(), "SNA_%lld_%llx.%lx", &p, &s, &sh);
571	if (r != 1) {
572	shard = shard_id_t::NO_SHARD;
573	} else {
574	shard = shard_id_t(sh);
575	}
576	}
577	dout(20) << __func__ << " mapping pool " << mapping.hoid.pool
578	<< " snap " << mapping.snap
579	<< " shard " << shard
580	<< " " << mapping.hoid << dendl;
581	mapit->next();
582	return true;
583	}
584
585	void SnapMapper::Scrubber::run()
586	{
587	dout(10) << __func__ << dendl;
588
589	psit = store->get_omap_iterator(ch, purged_snaps_hoid);
590	psit->upper_bound(PURGED_SNAP_PREFIX);
591	_parse_p();
592
593	mapit = store->get_omap_iterator(ch, mapping_hoid);
594	mapit->upper_bound(MAPPING_PREFIX);
595
596	while (_parse_m()) {
597	// advance to next purged_snaps range?
598	while (pool >= 0 &&
599	(mapping.hoid.pool > pool \|\|
600	(mapping.hoid.pool == pool && mapping.snap >= end))) {
601	_parse_p();
602	}
603	if (pool < 0) {
604	dout(10) << __func__ << " passed final purged_snaps interval, rest ok"
605	<< dendl;
606	break;
607	}
608	if (mapping.hoid.pool < pool \|\|
609	mapping.snap < begin) {
610	// ok
611	dout(20) << __func__ << " ok " << mapping.hoid
612	<< " snap " << mapping.snap
613	<< " precedes pool " << pool
614	<< " purged_snaps [" << begin << "," << end << ")" << dendl;
615	} else {
616	assert(mapping.snap >= begin &&
617	mapping.snap < end &&
618	mapping.hoid.pool == pool);
619	// invalid
620	dout(10) << __func__ << " stray " << mapping.hoid
621	<< " snap " << mapping.snap
622	<< " in pool " << pool
623	<< " shard " << shard
624	<< " purged_snaps [" << begin << "," << end << ")" << dendl;
625	stray.emplace_back(std::tuple<int64_t,snapid_t,uint32_t,shard_id_t>(
626	pool, mapping.snap, mapping.hoid.get_hash(),
627	shard
628	));
629	}
630	}
631
632	dout(10) << __func__ << " end, found " << stray.size() << " stray" << dendl;
633	psit = ObjectMap::ObjectMapIterator();
634	mapit = ObjectMap::ObjectMapIterator();
635	}
636
637
638	// -------------------------------------
639	// legacy conversion/support
640
641	string SnapMapper::get_legacy_prefix(snapid_t snap)
642	{
643	char buf[100];
644	int len = snprintf(
645	buf, sizeof(buf),
646	"%.*X_",
647	(int)(sizeof(snap)*2), static_cast<unsigned>(snap));
648	return LEGACY_MAPPING_PREFIX + string(buf, len);
649	}
650
651	string SnapMapper::to_legacy_raw_key(
652	const pair<snapid_t, hobject_t> &in)
653	{
654	return get_legacy_prefix(in.first) + shard_prefix + in.second.to_str();
655	}
656
657	bool SnapMapper::is_legacy_mapping(const string &to_test)
658	{
659	return to_test.substr(0, LEGACY_MAPPING_PREFIX.size()) ==
660	LEGACY_MAPPING_PREFIX;
661	}
662
663	int SnapMapper::convert_legacy(
664	CephContext *cct,
665	ObjectStore *store,
666	ObjectStore::CollectionHandle& ch,
667	ghobject_t hoid,
668	unsigned max)
669	{
670	uint64_t n = 0;
671
672	ObjectMap::ObjectMapIterator iter = store->get_omap_iterator(ch, hoid);
673	if (!iter) {
674	return -EIO;
675	}
676
677	auto start = ceph::mono_clock::now();
678
679	iter->upper_bound(SnapMapper::LEGACY_MAPPING_PREFIX);
680	map<string,bufferlist> to_set;
681	while (iter->valid()) {
682	bool valid = SnapMapper::is_legacy_mapping(iter->key());
683	if (valid) {
684	SnapMapper::Mapping m;
685	bufferlist bl(iter->value());
686	auto bp = bl.cbegin();
687	decode(m, bp);
688	to_set.emplace(
689	SnapMapper::get_prefix(m.hoid.pool, m.snap),
690	bl);
691	++n;
692	iter->next();
693	}
694	if (!valid \|\| !iter->valid() \|\| to_set.size() >= max) {
695	ObjectStore::Transaction t;
696	t.omap_setkeys(ch->cid, hoid, to_set);
697	int r = store->queue_transaction(ch, std::move(t));
698	ceph_assert(r == 0);
699	to_set.clear();
700	if (!valid) {
701	break;
702	}
703	dout(10) << __func__ << " converted " << n << " keys" << dendl;
704	}
705	}
706
707	auto end = ceph::mono_clock::now();
708
709	dout(1) << __func__ << " converted " << n << " keys in "
710	<< timespan_str(end - start) << dendl;
711
712	// remove the old keys
713	{
714	ObjectStore::Transaction t;
715	string end = SnapMapper::LEGACY_MAPPING_PREFIX;
716	++end[end.size()-1]; // turn _ to whatever comes after _
717	t.omap_rmkeyrange(ch->cid, hoid,
718	SnapMapper::LEGACY_MAPPING_PREFIX,
719	end);
720	int r = store->queue_transaction(ch, std::move(t));
721	ceph_assert(r == 0);
722	}
723	return 0;
724	}