ceph/src/crimson/os/cyanstore/cyan_store.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "cyan_store.h"
   5
   6 #include <boost/algorithm/string/trim.hpp>
   7 #include <fmt/format.h>
   8 #include <fmt/ostream.h>
   9
  10 #include "common/safe_io.h"
  11 #include "os/Transaction.h"
  12
  13 #include "crimson/common/buffer_io.h"
  14 #include "crimson/common/config_proxy.h"
  15 #include "cyan_collection.h"
  16 #include "cyan_object.h"
  17
  18 namespace {
  19   seastar::logger& logger() {
  20     return crimson::get_logger(ceph_subsys_filestore);
  21   }
  22 }
  23
  24 using std::string;
  25 using crimson::common::local_conf;
  26
  27 namespace crimson::os {
  28
  29 using ObjectRef = boost::intrusive_ptr<Object>;
  30
  31 CyanStore::CyanStore(const std::string& path)
  32   : path{path}
  33 {}
  34
  35 CyanStore::~CyanStore() = default;
  36
  37 template <const char* MsgV>
  38 struct singleton_ec : std::error_code {
  39   singleton_ec()
  40     : error_code(42, this_error_category{}) {
  41   };
  42 private:
  43   struct this_error_category : std::error_category {
  44     const char* name() const noexcept final {
  45       // XXX: we could concatenate with MsgV at compile-time but the burden
  46       // isn't worth the benefit.
  47       return "singleton_ec";
  48     }
  49     std::string message([[maybe_unused]] const int ev) const final {
  50       assert(ev == 42);
  51       return MsgV;
  52     }
  53   };
  54 };
  55
  56 CyanStore::mount_ertr::future<> CyanStore::mount()
  57 {
  58   static const char read_file_errmsg[]{"read_file"};
  59   ceph::bufferlist bl;
  60   std::string fn = path + "/collections";
  61   std::string err;
  62   if (int r = bl.read_file(fn.c_str(), &err); r < 0) {
  63     return crimson::stateful_ec{ singleton_ec<read_file_errmsg>() };
  64   }
  65
  66   std::set<coll_t> collections;
  67   auto p = bl.cbegin();
  68   ceph::decode(collections, p);
  69
  70   for (auto& coll : collections) {
  71     std::string fn = fmt::format("{}/{}", path, coll);
  72     ceph::bufferlist cbl;
  73     if (int r = cbl.read_file(fn.c_str(), &err); r < 0) {
  74       return crimson::stateful_ec{ singleton_ec<read_file_errmsg>() };
  75     }
  76     boost::intrusive_ptr<Collection> c{new Collection{coll}};
  77     auto p = cbl.cbegin();
  78     c->decode(p);
  79     coll_map[coll] = c;
  80     used_bytes += c->used_bytes();
  81   }
  82   return mount_ertr::now();
  83 }
  84
  85 seastar::future<> CyanStore::umount()
  86 {
  87   return seastar::do_with(std::set<coll_t>{}, [this](auto& collections) {
  88     return seastar::do_for_each(coll_map, [&collections, this](auto& coll) {
  89       auto& [col, ch] = coll;
  90       collections.insert(col);
  91       ceph::bufferlist bl;
  92       ceph_assert(ch);
  93       ch->encode(bl);
  94       std::string fn = fmt::format("{}/{}", path, col);
  95       return crimson::write_file(std::move(bl), fn);
  96     }).then([&collections, this] {
  97       ceph::bufferlist bl;
  98       ceph::encode(collections, bl);
  99       std::string fn = fmt::format("{}/collections", path);
 100       return crimson::write_file(std::move(bl), fn);
 101     });
 102   });
 103 }
 104
 105 CyanStore::mkfs_ertr::future<> CyanStore::mkfs(uuid_d new_osd_fsid)
 106 {
 107   static const char read_meta_errmsg[]{"read_meta"};
 108   static const char parse_fsid_errmsg[]{"failed to parse fsid"};
 109   static const char match_ofsid_errmsg[]{"unmatched osd_fsid"};
 110   return read_meta("fsid").then([=](auto&& ret) -> mkfs_ertr::future<> {
 111     auto& [r, fsid_str] = ret;
 112     if (r == -ENOENT) {
 113       if (new_osd_fsid.is_zero()) {
 114         osd_fsid.generate_random();
 115       } else {
 116         osd_fsid = new_osd_fsid;
 117       }
 118       return write_meta("fsid", fmt::format("{}", osd_fsid));
 119     } else if (r < 0) {
 120       return crimson::stateful_ec{ singleton_ec<read_meta_errmsg>() };
 121     } else {
 122       logger().info("mkfs already has fsid {}", fsid_str);
 123       if (!osd_fsid.parse(fsid_str.c_str())) {
 124         return crimson::stateful_ec{ singleton_ec<parse_fsid_errmsg>() };
 125       } else if (osd_fsid != new_osd_fsid) {
 126         logger().error("on-disk fsid {} != provided {}", osd_fsid, new_osd_fsid);
 127         return crimson::stateful_ec{ singleton_ec<match_ofsid_errmsg>() };
 128       } else {
 129         return mkfs_ertr::now();
 130       }
 131     }
 132   }).safe_then([this]{
 133     std::string fn = path + "/collections";
 134     ceph::bufferlist bl;
 135     std::set<coll_t> collections;
 136     ceph::encode(collections, bl);
 137     return crimson::write_file(std::move(bl), fn);
 138   }).safe_then([this] {
 139     return write_meta("type", "memstore");
 140   });
 141 }
 142
 143 seastar::future<store_statfs_t> CyanStore::stat() const
 144 {
 145   logger().debug("{}", __func__);
 146   store_statfs_t st;
 147   st.total = crimson::common::local_conf().get_val<Option::size_t>("memstore_device_bytes");
 148   st.available = st.total - used_bytes;
 149   return seastar::make_ready_future<store_statfs_t>(std::move(st));
 150 }
 151
 152 seastar::future<std::tuple<std::vector<ghobject_t>, ghobject_t>>
 153 CyanStore::list_objects(CollectionRef ch,
 154                         const ghobject_t& start,
 155                         const ghobject_t& end,
 156                         uint64_t limit) const
 157 {
 158   auto c = static_cast<Collection*>(ch.get());
 159   logger().debug("{} {} {} {} {}",
 160                  __func__, c->get_cid(), start, end, limit);
 161   std::vector<ghobject_t> objects;
 162   objects.reserve(limit);
 163   ghobject_t next = ghobject_t::get_max();
 164   for (const auto& [oid, obj] :
 165          boost::make_iterator_range(c->object_map.lower_bound(start),
 166                                     c->object_map.end())) {
 167     std::ignore = obj;
 168     if (oid >= end || objects.size() >= limit) {
 169       next = oid;
 170       break;
 171     }
 172     objects.push_back(oid);
 173   }
 174   return seastar::make_ready_future<std::tuple<std::vector<ghobject_t>, ghobject_t>>(
 175     std::make_tuple(std::move(objects), next));
 176 }
 177
 178 seastar::future<CollectionRef> CyanStore::create_new_collection(const coll_t& cid)
 179 {
 180   auto c = new Collection{cid};
 181   new_coll_map[cid] = c;
 182   return seastar::make_ready_future<CollectionRef>(c);
 183 }
 184
 185 seastar::future<CollectionRef> CyanStore::open_collection(const coll_t& cid)
 186 {
 187   return seastar::make_ready_future<CollectionRef>(_get_collection(cid));
 188 }
 189
 190 seastar::future<std::vector<coll_t>> CyanStore::list_collections()
 191 {
 192   std::vector<coll_t> collections;
 193   for (auto& coll : coll_map) {
 194     collections.push_back(coll.first);
 195   }
 196   return seastar::make_ready_future<std::vector<coll_t>>(std::move(collections));
 197 }
 198
 199 CyanStore::read_errorator::future<ceph::bufferlist> CyanStore::read(
 200   CollectionRef ch,
 201   const ghobject_t& oid,
 202   uint64_t offset,
 203   size_t len,
 204   uint32_t op_flags)
 205 {
 206   auto c = static_cast<Collection*>(ch.get());
 207   logger().debug("{} {} {} {}~{}",
 208                 __func__, c->get_cid(), oid, offset, len);
 209   if (!c->exists) {
 210     return crimson::ct_error::enoent::make();
 211   }
 212   ObjectRef o = c->get_object(oid);
 213   if (!o) {
 214     return crimson::ct_error::enoent::make();
 215   }
 216   if (offset >= o->get_size())
 217     return read_errorator::make_ready_future<ceph::bufferlist>();
 218   size_t l = len;
 219   if (l == 0 && offset == 0)  // note: len == 0 means read the entire object
 220     l = o->get_size();
 221   else if (offset + l > o->get_size())
 222     l = o->get_size() - offset;
 223   return read_errorator::make_ready_future<ceph::bufferlist>(o->read(offset, l));
 224 }
 225
 226 CyanStore::read_errorator::future<ceph::bufferlist> CyanStore::readv(
 227   CollectionRef ch,
 228   const ghobject_t& oid,
 229   interval_set<uint64_t>& m,
 230   uint32_t op_flags)
 231 {
 232   return seastar::do_with(ceph::bufferlist{},
 233     [this, ch, oid, &m, op_flags](auto& bl) {
 234     return crimson::do_for_each(m,
 235       [this, ch, oid, op_flags, &bl](auto& p) {
 236       return read(ch, oid, p.first, p.second, op_flags)
 237       .safe_then([&bl](auto ret) {
 238         bl.claim_append(ret);
 239       });
 240     }).safe_then([&bl] {
 241       return read_errorator::make_ready_future<ceph::bufferlist>(std::move(bl));
 242     });
 243   });
 244 }
 245
 246
 247 CyanStore::get_attr_errorator::future<ceph::bufferlist> CyanStore::get_attr(
 248   CollectionRef ch,
 249   const ghobject_t& oid,
 250   std::string_view name) const
 251 {
 252   auto c = static_cast<Collection*>(ch.get());
 253   logger().debug("{} {} {}",
 254                 __func__, c->get_cid(), oid);
 255   auto o = c->get_object(oid);
 256   if (!o) {
 257     return crimson::ct_error::enoent::make();
 258   }
 259   if (auto found = o->xattr.find(name); found != o->xattr.end()) {
 260     return get_attr_errorator::make_ready_future<ceph::bufferlist>(found->second);
 261   } else {
 262     return crimson::ct_error::enodata::make();
 263   }
 264 }
 265
 266 CyanStore::get_attrs_ertr::future<CyanStore::attrs_t> CyanStore::get_attrs(
 267   CollectionRef ch,
 268   const ghobject_t& oid)
 269 {
 270   auto c = static_cast<Collection*>(ch.get());
 271   logger().debug("{} {} {}",
 272                  __func__, c->get_cid(), oid);
 273   auto o = c->get_object(oid);
 274   if (!o) {
 275     return crimson::ct_error::enoent::make();
 276   }
 277   return get_attrs_ertr::make_ready_future<attrs_t>(o->xattr);
 278 }
 279
 280 auto CyanStore::omap_get_values(CollectionRef ch,
 281                                 const ghobject_t& oid,
 282                                 const omap_keys_t& keys)
 283   -> read_errorator::future<omap_values_t>
 284 {
 285   auto c = static_cast<Collection*>(ch.get());
 286   logger().debug("{} {} {}", __func__, c->get_cid(), oid);
 287   auto o = c->get_object(oid);
 288   if (!o) {
 289     return crimson::ct_error::enoent::make();
 290   }
 291   omap_values_t values;
 292   for (auto& key : keys) {
 293     if (auto found = o->omap.find(key); found != o->omap.end()) {
 294       values.insert(*found);
 295     }
 296   }
 297   return seastar::make_ready_future<omap_values_t>(std::move(values));
 298 }
 299
 300 auto
 301 CyanStore::omap_get_values(CollectionRef ch,
 302                            const ghobject_t &oid,
 303                            const std::optional<string> &start)
 304   -> read_errorator::future<std::tuple<bool, omap_values_t>>
 305 {
 306   auto c = static_cast<Collection*>(ch.get());
 307   logger().debug("{} {} {}", __func__, c->get_cid(), oid);
 308   auto o = c->get_object(oid);
 309   if (!o) {
 310     return crimson::ct_error::enoent::make();
 311   }
 312   omap_values_t values;
 313   for (auto i = start ? o->omap.upper_bound(*start) : o->omap.begin();
 314        values.size() < MAX_KEYS_PER_OMAP_GET_CALL && i != o->omap.end();
 315        ++i) {
 316     values.insert(*i);
 317   }
 318   return seastar::make_ready_future<std::tuple<bool, omap_values_t>>(
 319     std::make_tuple(true, std::move(values)));
 320 }
 321
 322 auto
 323 CyanStore::omap_get_header(CollectionRef ch,
 324                            const ghobject_t& oid)
 325   -> read_errorator::future<ceph::bufferlist>
 326 {
 327   auto c = static_cast<Collection*>(ch.get());
 328   auto o = c->get_object(oid);
 329   if (!o) {
 330     return crimson::ct_error::enoent::make();
 331   }
 332
 333   return read_errorator::make_ready_future<ceph::bufferlist>(
 334     o->omap_header);
 335 }
 336
 337 seastar::future<> CyanStore::do_transaction(CollectionRef ch,
 338                                             ceph::os::Transaction&& t)
 339 {
 340   using ceph::os::Transaction;
 341   int r = 0;
 342   try {
 343     auto i = t.begin();
 344     while (i.have_op()) {
 345       r = 0;
 346       switch (auto op = i.decode_op(); op->op) {
 347       case Transaction::OP_NOP:
 348         break;
 349       case Transaction::OP_REMOVE:
 350       {
 351         coll_t cid = i.get_cid(op->cid);
 352         ghobject_t oid = i.get_oid(op->oid);
 353         r = _remove(cid, oid);
 354         if (r == -ENOENT) {
 355           r = 0;
 356         }
 357       }
 358       break;
 359       case Transaction::OP_TOUCH:
 360       {
 361         coll_t cid = i.get_cid(op->cid);
 362         ghobject_t oid = i.get_oid(op->oid);
 363         r = _touch(cid, oid);
 364       }
 365       break;
 366       case Transaction::OP_WRITE:
 367       {
 368         coll_t cid = i.get_cid(op->cid);
 369         ghobject_t oid = i.get_oid(op->oid);
 370         uint64_t off = op->off;
 371         uint64_t len = op->len;
 372         uint32_t fadvise_flags = i.get_fadvise_flags();
 373         ceph::bufferlist bl;
 374         i.decode_bl(bl);
 375         r = _write(cid, oid, off, len, bl, fadvise_flags);
 376       }
 377       break;
 378       case Transaction::OP_ZERO:
 379       {
 380         coll_t cid = i.get_cid(op->cid);
 381         ghobject_t oid = i.get_oid(op->oid);
 382         uint64_t off = op->off;
 383         uint64_t len = op->len;
 384         r = _zero(cid, oid, off, len);
 385       }
 386       break;
 387       case Transaction::OP_TRUNCATE:
 388       {
 389         coll_t cid = i.get_cid(op->cid);
 390         ghobject_t oid = i.get_oid(op->oid);
 391         uint64_t off = op->off;
 392         r = _truncate(cid, oid, off);
 393       }
 394       break;
 395       case Transaction::OP_SETATTR:
 396       {
 397         coll_t cid = i.get_cid(op->cid);
 398         ghobject_t oid = i.get_oid(op->oid);
 399         std::string name = i.decode_string();
 400         ceph::bufferlist bl;
 401         i.decode_bl(bl);
 402         std::map<std::string, bufferlist> to_set;
 403         to_set.emplace(name, std::move(bl));
 404         r = _setattrs(cid, oid, std::move(to_set));
 405       }
 406       break;
 407       case Transaction::OP_SETATTRS:
 408       {
 409         coll_t cid = i.get_cid(op->cid);
 410         ghobject_t oid = i.get_oid(op->oid);
 411         std::map<std::string, bufferlist> aset;
 412         i.decode_attrset(aset);
 413         r = _setattrs(cid, oid, std::move(aset));
 414       }
 415       break;
 416       case Transaction::OP_RMATTR:
 417       {
 418         coll_t cid = i.get_cid(op->cid);
 419         ghobject_t oid = i.get_oid(op->oid);
 420         std::string name = i.decode_string();
 421         r = _rm_attr(cid, oid, name);
 422       }
 423       break;
 424       case Transaction::OP_RMATTRS:
 425       {
 426         coll_t cid = i.get_cid(op->cid);
 427         ghobject_t oid = i.get_oid(op->oid);
 428         r = _rm_attrs(cid, oid);
 429       }
 430       break;
 431       case Transaction::OP_MKCOLL:
 432       {
 433         coll_t cid = i.get_cid(op->cid);
 434         r = _create_collection(cid, op->split_bits);
 435       }
 436       break;
 437       case Transaction::OP_SETALLOCHINT:
 438       {
 439         r = 0;
 440       }
 441       break;
 442       case Transaction::OP_OMAP_CLEAR:
 443       {
 444         coll_t cid = i.get_cid(op->cid);
 445         ghobject_t oid = i.get_oid(op->oid);
 446         r = _omap_clear(cid, oid);
 447       }
 448       break;
 449       case Transaction::OP_OMAP_SETKEYS:
 450       {
 451         coll_t cid = i.get_cid(op->cid);
 452         ghobject_t oid = i.get_oid(op->oid);
 453         std::map<std::string, ceph::bufferlist> aset;
 454         i.decode_attrset(aset);
 455         r = _omap_set_values(cid, oid, std::move(aset));
 456       }
 457       break;
 458       case Transaction::OP_OMAP_SETHEADER:
 459       {
 460         const coll_t &cid = i.get_cid(op->cid);
 461         const ghobject_t &oid = i.get_oid(op->oid);
 462         ceph::bufferlist bl;
 463         i.decode_bl(bl);
 464         r = _omap_set_header(cid, oid, bl);
 465       }
 466       break;
 467       case Transaction::OP_OMAP_RMKEYS:
 468       {
 469         const coll_t &cid = i.get_cid(op->cid);
 470         const ghobject_t &oid = i.get_oid(op->oid);
 471         omap_keys_t keys;
 472         i.decode_keyset(keys);
 473         r = _omap_rmkeys(cid, oid, keys);
 474       }
 475       break;
 476       case Transaction::OP_OMAP_RMKEYRANGE:
 477       {
 478         const coll_t &cid = i.get_cid(op->cid);
 479         const ghobject_t &oid = i.get_oid(op->oid);
 480         string first, last;
 481         first = i.decode_string();
 482         last = i.decode_string();
 483         r = _omap_rmkeyrange(cid, oid, first, last);
 484       }
 485       break;
 486       case Transaction::OP_COLL_HINT:
 487       {
 488         ceph::bufferlist hint;
 489         i.decode_bl(hint);
 490         // ignored
 491         break;
 492       }
 493       default:
 494         logger().error("bad op {}", static_cast<unsigned>(op->op));
 495         abort();
 496       }
 497       if (r < 0) {
 498         break;
 499       }
 500     }
 501   } catch (std::exception &e) {
 502     logger().error("{} got exception {}", __func__, e);
 503     r = -EINVAL;
 504   }
 505   if (r < 0) {
 506     logger().error(" transaction dump:\n");
 507     JSONFormatter f(true);
 508     f.open_object_section("transaction");
 509     t.dump(&f);
 510     f.close_section();
 511     std::stringstream str;
 512     f.flush(str);
 513     logger().error("{}", str.str());
 514     ceph_assert(r == 0);
 515   }
 516   for (auto i : {
 517       t.get_on_applied(),
 518       t.get_on_commit(),
 519       t.get_on_applied_sync()}) {
 520     if (i) {
 521       i->complete(0);
 522     }
 523   }
 524   return seastar::now();
 525 }
 526
 527 int CyanStore::_remove(const coll_t& cid, const ghobject_t& oid)
 528 {
 529   logger().debug("{} cid={} oid={}",
 530                 __func__, cid, oid);
 531   auto c = _get_collection(cid);
 532   if (!c)
 533     return -ENOENT;
 534
 535   auto i = c->object_hash.find(oid);
 536   if (i == c->object_hash.end())
 537     return -ENOENT;
 538   used_bytes -= i->second->get_size();
 539   c->object_hash.erase(i);
 540   c->object_map.erase(oid);
 541   return 0;
 542 }
 543
 544 int CyanStore::_touch(const coll_t& cid, const ghobject_t& oid)
 545 {
 546   logger().debug("{} cid={} oid={}",
 547                 __func__, cid, oid);
 548   auto c = _get_collection(cid);
 549   if (!c)
 550     return -ENOENT;
 551
 552   c->get_or_create_object(oid);
 553   return 0;
 554 }
 555
 556 int CyanStore::_write(const coll_t& cid, const ghobject_t& oid,
 557                       uint64_t offset, size_t len, const ceph::bufferlist& bl,
 558                       uint32_t fadvise_flags)
 559 {
 560   logger().debug("{} {} {} {} ~ {}",
 561                 __func__, cid, oid, offset, len);
 562   assert(len == bl.length());
 563
 564   auto c = _get_collection(cid);
 565   if (!c)
 566     return -ENOENT;
 567
 568   ObjectRef o = c->get_or_create_object(oid);
 569   if (len > 0 && !local_conf()->memstore_debug_omit_block_device_write) {
 570     const ssize_t old_size = o->get_size();
 571     o->write(offset, bl);
 572     used_bytes += (o->get_size() - old_size);
 573   }
 574
 575   return 0;
 576 }
 577
 578 int CyanStore::_zero(const coll_t& cid, const ghobject_t& oid,
 579                      uint64_t offset, size_t len)
 580 {
 581   logger().debug("{} {} {} {} ~ {}",
 582                 __func__, cid, oid, offset, len);
 583
 584   ceph::buffer::list bl;
 585   bl.append_zero(len);
 586   return _write(cid, oid, offset, len, bl, 0);
 587 }
 588
 589 int CyanStore::_omap_clear(
 590   const coll_t& cid,
 591   const ghobject_t& oid)
 592 {
 593   logger().debug("{} {} {}", __func__, cid, oid);
 594
 595   auto c = _get_collection(cid);
 596   if (!c) {
 597     return -ENOENT;
 598   }
 599   ObjectRef o = c->get_object(oid);
 600   if (!o) {
 601     return -ENOENT;
 602   }
 603   o->omap.clear();
 604   o->omap_header.clear();
 605   return 0;
 606 }
 607
 608 int CyanStore::_omap_set_values(
 609   const coll_t& cid,
 610   const ghobject_t& oid,
 611   std::map<std::string, ceph::bufferlist> &&aset)
 612 {
 613   logger().debug(
 614     "{} {} {} {} keys",
 615     __func__, cid, oid, aset.size());
 616
 617   auto c = _get_collection(cid);
 618   if (!c)
 619     return -ENOENT;
 620
 621   ObjectRef o = c->get_or_create_object(oid);
 622   for (auto&& [key, val]: aset) {
 623     o->omap.insert_or_assign(std::move(key), std::move(val));
 624   }
 625   return 0;
 626 }
 627
 628 int CyanStore::_omap_set_header(
 629   const coll_t& cid,
 630   const ghobject_t& oid,
 631   const ceph::bufferlist &header)
 632 {
 633   logger().debug(
 634     "{} {} {} {} bytes",
 635     __func__, cid, oid, header.length());
 636
 637   auto c = _get_collection(cid);
 638   if (!c)
 639     return -ENOENT;
 640
 641   ObjectRef o = c->get_or_create_object(oid);
 642   o->omap_header = header;
 643   return 0;
 644 }
 645
 646 int CyanStore::_omap_rmkeys(
 647   const coll_t& cid,
 648   const ghobject_t& oid,
 649   const omap_keys_t& aset)
 650 {
 651   logger().debug(
 652     "{} {} {} {} keys",
 653     __func__, cid, oid, aset.size());
 654
 655   auto c = _get_collection(cid);
 656   if (!c)
 657     return -ENOENT;
 658
 659   ObjectRef o = c->get_or_create_object(oid);
 660   for (auto &i: aset) {
 661     o->omap.erase(i);
 662   }
 663   return 0;
 664 }
 665
 666 int CyanStore::_omap_rmkeyrange(
 667   const coll_t& cid,
 668   const ghobject_t& oid,
 669   const std::string &first,
 670   const std::string &last)
 671 {
 672   logger().debug(
 673     "{} {} {} first={} last={}",
 674     __func__, cid, oid, first, last);
 675
 676   auto c = _get_collection(cid);
 677   if (!c)
 678     return -ENOENT;
 679
 680   ObjectRef o = c->get_or_create_object(oid);
 681   for (auto i = o->omap.lower_bound(first);
 682        i != o->omap.end() && i->first <= last;
 683        o->omap.erase(i++));
 684   return 0;
 685 }
 686
 687 int CyanStore::_truncate(const coll_t& cid, const ghobject_t& oid, uint64_t size)
 688 {
 689   logger().debug("{} cid={} oid={} size={}",
 690                 __func__, cid, oid, size);
 691   auto c = _get_collection(cid);
 692   if (!c)
 693     return -ENOENT;
 694
 695   ObjectRef o = c->get_object(oid);
 696   if (!o)
 697     return -ENOENT;
 698   if (local_conf()->memstore_debug_omit_block_device_write)
 699     return 0;
 700   const ssize_t old_size = o->get_size();
 701   int r = o->truncate(size);
 702   used_bytes += (o->get_size() - old_size);
 703   return r;
 704 }
 705
 706 int CyanStore::_setattrs(const coll_t& cid, const ghobject_t& oid,
 707                          std::map<std::string,bufferlist>&& aset)
 708 {
 709   logger().debug("{} cid={} oid={}",
 710                 __func__, cid, oid);
 711   auto c = _get_collection(cid);
 712   if (!c)
 713     return -ENOENT;
 714
 715   ObjectRef o = c->get_object(oid);
 716   if (!o)
 717     return -ENOENT;
 718   for (auto&& [key, val]: aset) {
 719     o->xattr.insert_or_assign(std::move(key), std::move(val));
 720   }
 721   return 0;
 722 }
 723
 724 int CyanStore::_rm_attr(const coll_t& cid, const ghobject_t& oid,
 725                         std::string_view name)
 726 {
 727   logger().debug("{} cid={} oid={} name={}", __func__, cid, oid, name);
 728   auto c = _get_collection(cid);
 729   if (!c) {
 730     return -ENOENT;
 731   }
 732   ObjectRef o = c->get_object(oid);
 733   if (!o) {
 734     return -ENOENT;
 735   }
 736   auto i = o->xattr.find(name);
 737   if (i == o->xattr.end()) {
 738     return -ENODATA;
 739   }
 740   o->xattr.erase(i);
 741   return 0;
 742 }
 743
 744 int CyanStore::_rm_attrs(const coll_t& cid, const ghobject_t& oid)
 745 {
 746   logger().debug("{} cid={} oid={}", __func__, cid, oid);
 747   auto c = _get_collection(cid);
 748   if (!c) {
 749     return -ENOENT;
 750   }
 751   ObjectRef o = c->get_object(oid);
 752   if (!o) {
 753     return -ENOENT;
 754   }
 755   o->xattr.clear();
 756   return 0;
 757 }
 758
 759 int CyanStore::_create_collection(const coll_t& cid, int bits)
 760 {
 761   auto result = coll_map.try_emplace(cid);
 762   if (!result.second)
 763     return -EEXIST;
 764   auto p = new_coll_map.find(cid);
 765   assert(p != new_coll_map.end());
 766   result.first->second = p->second;
 767   result.first->second->bits = bits;
 768   new_coll_map.erase(p);
 769   return 0;
 770 }
 771
 772 boost::intrusive_ptr<Collection> CyanStore::_get_collection(const coll_t& cid)
 773 {
 774   auto cp = coll_map.find(cid);
 775   if (cp == coll_map.end())
 776     return {};
 777   return cp->second;
 778 }
 779
 780 seastar::future<> CyanStore::write_meta(const std::string& key,
 781                                         const std::string& value)
 782 {
 783   std::string v = value;
 784   v += "\n";
 785   if (int r = safe_write_file(path.c_str(), key.c_str(),
 786                               v.c_str(), v.length(), 0600);
 787       r < 0) {
 788     throw std::runtime_error{fmt::format("unable to write_meta({})", key)};
 789   }
 790   return seastar::make_ready_future<>();
 791 }
 792
 793 seastar::future<std::tuple<int, std::string>>
 794 CyanStore::read_meta(const std::string& key)
 795 {
 796   std::string fsid(4096, '\0');
 797   int r = safe_read_file(path.c_str(), key.c_str(), fsid.data(), fsid.size());
 798   if (r > 0) {
 799     fsid.resize(r);
 800     // drop trailing newlines
 801     boost::algorithm::trim_right_if(fsid,
 802                                     [](unsigned char c) {return isspace(c);});
 803   } else {
 804     fsid.clear();
 805   }
 806   return seastar::make_ready_future<std::tuple<int, std::string>>(
 807     std::make_tuple(r, fsid));
 808 }
 809
 810 uuid_d CyanStore::get_fsid() const
 811 {
 812   return osd_fsid;
 813 }
 814
 815 unsigned CyanStore::get_max_attr_name_length() const
 816 {
 817   // arbitrary limitation exactly like in the case of MemStore.
 818   return 256;
 819 }
 820
 821 seastar::future<FuturizedStore::OmapIteratorRef> CyanStore::get_omap_iterator(
 822     CollectionRef ch,
 823     const ghobject_t& oid)
 824 {
 825   auto c = static_cast<Collection*>(ch.get());
 826   auto o = c->get_object(oid);
 827   if (!o) {
 828     throw std::runtime_error(fmt::format("object does not exist: {}", oid));
 829   }
 830   return seastar::make_ready_future<FuturizedStore::OmapIteratorRef>(
 831             new CyanStore::CyanOmapIterator(o));
 832 }
 833
 834 seastar::future<std::map<uint64_t, uint64_t>>
 835 CyanStore::fiemap(
 836     CollectionRef ch,
 837     const ghobject_t& oid,
 838     uint64_t off,
 839     uint64_t len)
 840 {
 841   auto c = static_cast<Collection*>(ch.get());
 842
 843   ObjectRef o = c->get_object(oid);
 844   if (!o) {
 845     throw std::runtime_error(fmt::format("object does not exist: {}", oid));
 846   }
 847   std::map<uint64_t, uint64_t> m{{0, o->get_size()}};
 848   return seastar::make_ready_future<std::map<uint64_t, uint64_t>>(std::move(m));
 849 }
 850
 851 seastar::future<struct stat>
 852 CyanStore::stat(
 853   CollectionRef ch,
 854   const ghobject_t& oid)
 855 {
 856   auto c = static_cast<Collection*>(ch.get());
 857   auto o = c->get_object(oid);
 858   if (!o) {
 859     throw std::runtime_error(fmt::format("object does not exist: {}", oid));
 860   }
 861   struct stat st;
 862   st.st_size = o->get_size();
 863   return seastar::make_ready_future<struct stat>(std::move(st));
 864 }
 865
 866 seastar::future<> CyanStore::CyanOmapIterator::seek_to_first()
 867 {
 868   iter = obj->omap.begin();
 869   return seastar::make_ready_future<>();
 870 }
 871
 872 seastar::future<> CyanStore::CyanOmapIterator::upper_bound(const std::string& after)
 873 {
 874   iter = obj->omap.upper_bound(after);
 875   return seastar::make_ready_future<>();
 876 }
 877
 878 seastar::future<> CyanStore::CyanOmapIterator::lower_bound(const std::string &to)
 879 {
 880   iter = obj->omap.lower_bound(to);
 881   return seastar::make_ready_future<>();
 882 }
 883
 884 bool CyanStore::CyanOmapIterator::valid() const
 885 {
 886   return iter != obj->omap.end();
 887 }
 888
 889 seastar::future<> CyanStore::CyanOmapIterator::next()
 890 {
 891   ++iter;
 892   return seastar::make_ready_future<>();
 893 }
 894
 895 }