ceph/src/crimson/os/alienstore/alien_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 "alien_collection.h"
   5 #include "alien_store.h"
   6 #include "alien_log.h"
   7
   8 #include <algorithm>
   9 #include <iterator>
  10 #include <map>
  11 #include <string_view>
  12 #include <boost/algorithm/string/trim.hpp>
  13 #include <boost/iterator/counting_iterator.hpp>
  14 #include <fmt/format.h>
  15 #include <fmt/ostream.h>
  16
  17 #include <seastar/core/alien.hh>
  18 #include <seastar/core/future-util.hh>
  19 #include <seastar/core/reactor.hh>
  20 #include <seastar/core/resource.hh>
  21
  22 #include "common/ceph_context.h"
  23 #include "global/global_context.h"
  24 #include "include/Context.h"
  25 #include "os/ObjectStore.h"
  26 #include "os/Transaction.h"
  27
  28 #include "crimson/common/config_proxy.h"
  29 #include "crimson/common/log.h"
  30 #include "crimson/os/futurized_store.h"
  31
  32 using std::map;
  33 using std::set;
  34 using std::string;
  35
  36 namespace {
  37
  38 seastar::logger& logger()
  39 {
  40   return crimson::get_logger(ceph_subsys_alienstore);
  41 }
  42
  43 class OnCommit final: public Context
  44 {
  45   const int cpuid;
  46   seastar::alien::instance &alien;
  47   seastar::promise<> &alien_done;
  48 public:
  49   OnCommit(
  50     int id,
  51     seastar::promise<> &done,
  52     seastar::alien::instance &alien,
  53     ceph::os::Transaction& txn)
  54     : cpuid(id),
  55       alien(alien),
  56       alien_done(done) {
  57   }
  58
  59   void finish(int) final {
  60     return seastar::alien::submit_to(alien, cpuid, [this] {
  61       alien_done.set_value();
  62       return seastar::make_ready_future<>();
  63     }).wait();
  64   }
  65 };
  66 }
  67
  68 namespace crimson::os {
  69
  70 using crimson::common::get_conf;
  71
  72 AlienStore::AlienStore(const std::string& type,
  73                        const std::string& path,
  74                        const ConfigValues& values)
  75   : type(type),
  76     path{path},
  77     values(values)
  78 {
  79 }
  80
  81 AlienStore::~AlienStore()
  82 {
  83 }
  84
  85 seastar::future<> AlienStore::start()
  86 {
  87   cct = std::make_unique<CephContext>(
  88     CEPH_ENTITY_TYPE_OSD,
  89     CephContext::create_options { CODE_ENVIRONMENT_UTILITY, 0,
  90       [](const ceph::logging::SubsystemMap* subsys_map) {
  91         return new ceph::logging::CnLog(subsys_map, seastar::engine().alien(), seastar::this_shard_id());
  92       }
  93     }
  94   );
  95   g_ceph_context = cct.get();
  96   cct->_conf.set_config_values(values);
  97   cct->_log->start();
  98
  99   store = ObjectStore::create(cct.get(), type, path);
 100   if (!store) {
 101     ceph_abort_msgf("unsupported objectstore type: %s", type.c_str());
 102   }
 103   auto cpu_cores = seastar::resource::parse_cpuset(
 104     get_conf<std::string>("crimson_alien_thread_cpu_cores"));
 105   // cores except the first "N_CORES_FOR_SEASTAR" ones will
 106   // be used for alien threads scheduling:
 107   //    [0, N_CORES_FOR_SEASTAR) are reserved for seastar reactors
 108   //    [N_CORES_FOR_SEASTAR, ..] are assigned to alien threads.
 109   if (!cpu_cores.has_value()) {
 110     seastar::resource::cpuset cpuset;
 111     std::copy(boost::counting_iterator<unsigned>(N_CORES_FOR_SEASTAR),
 112               boost::counting_iterator<unsigned>(sysconf(_SC_NPROCESSORS_ONLN)),
 113               std::inserter(cpuset, cpuset.end()));
 114     if (cpuset.empty()) {
 115       logger().error("{}: unable to get nproc: {}", __func__, errno);
 116     } else {
 117       cpu_cores = cpuset;
 118     }
 119   }
 120   const auto num_threads =
 121     get_conf<uint64_t>("crimson_alien_op_num_threads");
 122   tp = std::make_unique<crimson::os::ThreadPool>(num_threads, 128, cpu_cores);
 123   return tp->start();
 124 }
 125
 126 seastar::future<> AlienStore::stop()
 127 {
 128   if (!tp) {
 129     // not really started yet
 130     return seastar::now();
 131   }
 132   return tp->submit([this] {
 133     for (auto [cid, ch]: coll_map) {
 134       static_cast<AlienCollection*>(ch.get())->collection.reset();
 135     }
 136     store.reset();
 137     cct.reset();
 138     g_ceph_context = nullptr;
 139
 140   }).then([this] {
 141     return tp->stop();
 142   });
 143 }
 144
 145 AlienStore::mount_ertr::future<> AlienStore::mount()
 146 {
 147   logger().debug("{}", __func__);
 148   assert(tp);
 149   return tp->submit([this] {
 150     return store->mount();
 151   }).then([] (const int r) -> mount_ertr::future<> {
 152     if (r != 0) {
 153       return crimson::stateful_ec{
 154         std::error_code(-r, std::generic_category()) };
 155     } else {
 156       return mount_ertr::now();
 157     }
 158   });
 159 }
 160
 161 seastar::future<> AlienStore::umount()
 162 {
 163   logger().info("{}", __func__);
 164   if (!tp) {
 165     // not really started yet
 166     return seastar::now();
 167   }
 168   return op_gate.close().then([this] {
 169     return tp->submit([this] {
 170       return store->umount();
 171     });
 172   }).then([] (int r) {
 173     assert(r == 0);
 174     return seastar::now();
 175   });
 176 }
 177
 178 AlienStore::mkfs_ertr::future<> AlienStore::mkfs(uuid_d osd_fsid)
 179 {
 180   logger().debug("{}", __func__);
 181   store->set_fsid(osd_fsid);
 182   assert(tp);
 183   return tp->submit([this] {
 184     return store->mkfs();
 185   }).then([] (int r) -> mkfs_ertr::future<> {
 186     if (r != 0) {
 187       return crimson::stateful_ec{
 188         std::error_code(-r, std::generic_category()) };
 189     } else {
 190       return mkfs_ertr::now();
 191     }
 192   });
 193 }
 194
 195 seastar::future<std::tuple<std::vector<ghobject_t>, ghobject_t>>
 196 AlienStore::list_objects(CollectionRef ch,
 197                         const ghobject_t& start,
 198                         const ghobject_t& end,
 199                         uint64_t limit) const
 200 {
 201   logger().debug("{}", __func__);
 202   assert(tp);
 203   return do_with_op_gate(std::vector<ghobject_t>(), ghobject_t(),
 204                          [=, this] (auto &objects, auto &next) {
 205     objects.reserve(limit);
 206     return tp->submit(ch->get_cid().hash_to_shard(tp->size()),
 207       [=, this, &objects, &next] {
 208       auto c = static_cast<AlienCollection*>(ch.get());
 209       return store->collection_list(c->collection, start, end,
 210                                     store->get_ideal_list_max(),
 211                                     &objects, &next);
 212     }).then([&objects, &next] (int r) {
 213       assert(r == 0);
 214       return seastar::make_ready_future<
 215         std::tuple<std::vector<ghobject_t>, ghobject_t>>(
 216           std::move(objects), std::move(next));
 217     });
 218   });
 219 }
 220
 221 seastar::future<CollectionRef> AlienStore::create_new_collection(const coll_t& cid)
 222 {
 223   logger().debug("{}", __func__);
 224   assert(tp);
 225   return tp->submit([this, cid] {
 226     return store->create_new_collection(cid);
 227   }).then([this, cid] (ObjectStore::CollectionHandle c) {
 228     CollectionRef ch;
 229     auto cp = coll_map.find(c->cid);
 230     if (cp == coll_map.end()) {
 231       ch = new AlienCollection(c);
 232       coll_map[c->cid] = ch;
 233     } else {
 234       ch = cp->second;
 235       auto ach = static_cast<AlienCollection*>(ch.get());
 236       if (ach->collection != c) {
 237         ach->collection = c;
 238       }
 239     }
 240     return seastar::make_ready_future<CollectionRef>(ch);
 241   });
 242
 243 }
 244
 245 seastar::future<CollectionRef> AlienStore::open_collection(const coll_t& cid)
 246 {
 247   logger().debug("{}", __func__);
 248   assert(tp);
 249   return tp->submit([this, cid] {
 250     return store->open_collection(cid);
 251   }).then([this] (ObjectStore::CollectionHandle c) {
 252     if (!c) {
 253       return seastar::make_ready_future<CollectionRef>();
 254     }
 255     CollectionRef ch;
 256     auto cp = coll_map.find(c->cid);
 257     if (cp == coll_map.end()){
 258       ch = new AlienCollection(c);
 259       coll_map[c->cid] = ch;
 260     } else {
 261       ch = cp->second;
 262       auto ach = static_cast<AlienCollection*>(ch.get());
 263       if (ach->collection != c){
 264         ach->collection = c;
 265       }
 266     }
 267     return seastar::make_ready_future<CollectionRef>(ch);
 268   });
 269 }
 270
 271 seastar::future<std::vector<coll_core_t>> AlienStore::list_collections()
 272 {
 273   logger().debug("{}", __func__);
 274   assert(tp);
 275
 276   return do_with_op_gate(std::vector<coll_t>{}, [this] (auto &ls) {
 277     return tp->submit([this, &ls] {
 278       return store->list_collections(ls);
 279     }).then([&ls] (int r) -> seastar::future<std::vector<coll_core_t>> {
 280       assert(r == 0);
 281       std::vector<coll_core_t> ret;
 282       ret.resize(ls.size());
 283       std::transform(
 284         ls.begin(), ls.end(), ret.begin(),
 285         [](auto p) { return std::make_pair(p, NULL_CORE); });
 286       return seastar::make_ready_future<std::vector<coll_core_t>>(std::move(ret));
 287     });
 288   });
 289 }
 290
 291 AlienStore::read_errorator::future<ceph::bufferlist>
 292 AlienStore::read(CollectionRef ch,
 293                  const ghobject_t& oid,
 294                  uint64_t offset,
 295                  size_t len,
 296                  uint32_t op_flags)
 297 {
 298   logger().debug("{}", __func__);
 299   assert(tp);
 300   return do_with_op_gate(ceph::bufferlist{}, [=, this] (auto &bl) {
 301     return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this, &bl] {
 302       auto c = static_cast<AlienCollection*>(ch.get());
 303       return store->read(c->collection, oid, offset, len, bl, op_flags);
 304     }).then([&bl] (int r) -> read_errorator::future<ceph::bufferlist> {
 305       if (r == -ENOENT) {
 306         return crimson::ct_error::enoent::make();
 307       } else if (r == -EIO) {
 308         return crimson::ct_error::input_output_error::make();
 309       } else {
 310         return read_errorator::make_ready_future<ceph::bufferlist>(
 311           std::move(bl));
 312       }
 313     });
 314   });
 315 }
 316
 317 AlienStore::read_errorator::future<ceph::bufferlist>
 318 AlienStore::readv(CollectionRef ch,
 319                   const ghobject_t& oid,
 320                   interval_set<uint64_t>& m,
 321                   uint32_t op_flags)
 322 {
 323   logger().debug("{}", __func__);
 324   assert(tp);
 325   return do_with_op_gate(ceph::bufferlist{},
 326     [this, ch, oid, &m, op_flags](auto& bl) {
 327     return tp->submit(ch->get_cid().hash_to_shard(tp->size()),
 328       [this, ch, oid, &m, op_flags, &bl] {
 329       auto c = static_cast<AlienCollection*>(ch.get());
 330       return store->readv(c->collection, oid, m, bl, op_flags);
 331     }).then([&bl](int r) -> read_errorator::future<ceph::bufferlist> {
 332       if (r == -ENOENT) {
 333         return crimson::ct_error::enoent::make();
 334       } else if (r == -EIO) {
 335         return crimson::ct_error::input_output_error::make();
 336       } else {
 337         return read_errorator::make_ready_future<ceph::bufferlist>(
 338           std::move(bl));
 339       }
 340     });
 341   });
 342 }
 343
 344 AlienStore::get_attr_errorator::future<ceph::bufferlist>
 345 AlienStore::get_attr(CollectionRef ch,
 346                      const ghobject_t& oid,
 347                      std::string_view name) const
 348 {
 349   logger().debug("{}", __func__);
 350   assert(tp);
 351   return do_with_op_gate(ceph::bufferlist{}, std::string{name},
 352                          [=, this] (auto &value, const auto& name) {
 353     return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this, &value, &name] {
 354       // XXX: `name` isn't a `std::string_view` anymore! it had to be converted
 355       // to `std::string` for the sake of extending life-time not only of
 356       // a _ptr-to-data_ but _data_ as well. Otherwise we would run into a use-
 357       // after-free issue.
 358       auto c = static_cast<AlienCollection*>(ch.get());
 359       return store->getattr(c->collection, oid, name.c_str(), value);
 360     }).then([oid, &value](int r) -> get_attr_errorator::future<ceph::bufferlist> {
 361       if (r == -ENOENT) {
 362         return crimson::ct_error::enoent::make();
 363       } else if (r == -ENODATA) {
 364         return crimson::ct_error::enodata::make();
 365       } else {
 366         return get_attr_errorator::make_ready_future<ceph::bufferlist>(
 367           std::move(value));
 368       }
 369     });
 370   });
 371 }
 372
 373 AlienStore::get_attrs_ertr::future<AlienStore::attrs_t>
 374 AlienStore::get_attrs(CollectionRef ch,
 375                       const ghobject_t& oid)
 376 {
 377   logger().debug("{}", __func__);
 378   assert(tp);
 379   return do_with_op_gate(attrs_t{}, [=, this] (auto &aset) {
 380     return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this, &aset] {
 381       auto c = static_cast<AlienCollection*>(ch.get());
 382       const auto r = store->getattrs(c->collection, oid, aset);
 383       return r;
 384     }).then([&aset] (int r) -> get_attrs_ertr::future<attrs_t> {
 385       if (r == -ENOENT) {
 386         return crimson::ct_error::enoent::make();
 387       } else {
 388         return get_attrs_ertr::make_ready_future<attrs_t>(std::move(aset));
 389       }
 390     });
 391   });
 392 }
 393
 394 auto AlienStore::omap_get_values(CollectionRef ch,
 395                                  const ghobject_t& oid,
 396                                  const set<string>& keys)
 397   -> read_errorator::future<omap_values_t>
 398 {
 399   logger().debug("{}", __func__);
 400   assert(tp);
 401   return do_with_op_gate(omap_values_t{}, [=, this] (auto &values) {
 402     return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this, &values] {
 403       auto c = static_cast<AlienCollection*>(ch.get());
 404       return store->omap_get_values(c->collection, oid, keys,
 405                                     reinterpret_cast<map<string, bufferlist>*>(&values));
 406     }).then([&values] (int r) -> read_errorator::future<omap_values_t> {
 407       if (r == -ENOENT) {
 408         return crimson::ct_error::enoent::make();
 409       } else {
 410         assert(r == 0);
 411         return read_errorator::make_ready_future<omap_values_t>(
 412           std::move(values));
 413       }
 414     });
 415   });
 416 }
 417
 418 auto AlienStore::omap_get_values(CollectionRef ch,
 419                                  const ghobject_t &oid,
 420                                  const std::optional<string> &start)
 421   -> read_errorator::future<std::tuple<bool, omap_values_t>>
 422 {
 423   logger().debug("{} with_start", __func__);
 424   assert(tp);
 425   return do_with_op_gate(omap_values_t{}, [=, this] (auto &values) {
 426     return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this, &values] {
 427       auto c = static_cast<AlienCollection*>(ch.get());
 428       return store->omap_get_values(c->collection, oid, start,
 429                                     reinterpret_cast<map<string, bufferlist>*>(&values));
 430     }).then([&values] (int r)
 431       -> read_errorator::future<std::tuple<bool, omap_values_t>> {
 432       if (r == -ENOENT) {
 433         return crimson::ct_error::enoent::make();
 434       } else if (r < 0){
 435         logger().error("omap_get_values(start): {}", r);
 436         return crimson::ct_error::input_output_error::make();
 437       } else {
 438         return read_errorator::make_ready_future<std::tuple<bool, omap_values_t>>(
 439           true, std::move(values));
 440       }
 441     });
 442   });
 443 }
 444
 445 seastar::future<> AlienStore::do_transaction_no_callbacks(
 446   CollectionRef ch,
 447   ceph::os::Transaction&& txn)
 448 {
 449   logger().debug("{}", __func__);
 450   auto id = seastar::this_shard_id();
 451   auto done = seastar::promise<>();
 452   return do_with_op_gate(
 453     std::move(txn),
 454     std::move(done),
 455     [this, ch, id] (auto &txn, auto &done) {
 456         AlienCollection* alien_coll = static_cast<AlienCollection*>(ch.get());
 457         // moving the `ch` is crucial for buildability on newer S* versions.
 458         return alien_coll->with_lock([this, ch=std::move(ch), id, &txn, &done] {
 459           assert(tp);
 460           return tp->submit(ch->get_cid().hash_to_shard(tp->size()),
 461             [this, ch, id, &txn, &done, &alien=seastar::engine().alien()] {
 462             txn.register_on_commit(new OnCommit(id, done, alien, txn));
 463             auto c = static_cast<AlienCollection*>(ch.get());
 464             return store->queue_transaction(c->collection, std::move(txn));
 465           });
 466         }).then([&done] (int r) {
 467           assert(r == 0);
 468           return done.get_future();
 469         });
 470     });
 471 }
 472
 473 seastar::future<> AlienStore::inject_data_error(const ghobject_t& o)
 474 {
 475   logger().debug("{}", __func__);
 476   assert(tp);
 477   return seastar::with_gate(op_gate, [=, this] {
 478     return tp->submit([o, this] {
 479       return store->inject_data_error(o);
 480     });
 481   });
 482 }
 483
 484 seastar::future<> AlienStore::inject_mdata_error(const ghobject_t& o)
 485 {
 486   logger().debug("{}", __func__);
 487   assert(tp);
 488   return seastar::with_gate(op_gate, [=, this] {
 489     return tp->submit([=, this] {
 490       return store->inject_mdata_error(o);
 491     });
 492   });
 493 }
 494
 495 seastar::future<> AlienStore::write_meta(const std::string& key,
 496                                          const std::string& value)
 497 {
 498   logger().debug("{}", __func__);
 499   assert(tp);
 500   return seastar::with_gate(op_gate, [=, this] {
 501     return tp->submit([=, this] {
 502       return store->write_meta(key, value);
 503     }).then([] (int r) {
 504       assert(r == 0);
 505       return seastar::make_ready_future<>();
 506     });
 507   });
 508 }
 509
 510 seastar::future<std::tuple<int, std::string>>
 511 AlienStore::read_meta(const std::string& key)
 512 {
 513   logger().debug("{}", __func__);
 514   assert(tp);
 515   return seastar::with_gate(op_gate, [this, key] {
 516     return tp->submit([this, key] {
 517       std::string value;
 518       int r = store->read_meta(key, &value);
 519       if (r > 0) {
 520         value.resize(r);
 521         boost::algorithm::trim_right_if(value,
 522           [] (unsigned char c) {return isspace(c);});
 523       } else {
 524         value.clear();
 525       }
 526       return std::make_pair(r, value);
 527     }).then([] (auto entry) {
 528       return seastar::make_ready_future<std::tuple<int, std::string>>(
 529         std::move(entry));
 530     });
 531   });
 532 }
 533
 534 uuid_d AlienStore::get_fsid() const
 535 {
 536   logger().debug("{}", __func__);
 537   return store->get_fsid();
 538 }
 539
 540 seastar::future<store_statfs_t> AlienStore::stat() const
 541 {
 542   logger().info("{}", __func__);
 543   assert(tp);
 544   return do_with_op_gate(store_statfs_t{}, [this] (store_statfs_t &st) {
 545     return tp->submit([this, &st] {
 546       return store->statfs(&st, nullptr);
 547     }).then([&st] (int r) {
 548       assert(r == 0);
 549       return seastar::make_ready_future<store_statfs_t>(std::move(st));
 550     });
 551   });
 552 }
 553
 554 unsigned AlienStore::get_max_attr_name_length() const
 555 {
 556   logger().info("{}", __func__);
 557   return 256;
 558 }
 559
 560 seastar::future<struct stat> AlienStore::stat(
 561   CollectionRef ch,
 562   const ghobject_t& oid)
 563 {
 564   assert(tp);
 565   return do_with_op_gate((struct stat){}, [this, ch, oid](auto& st) {
 566     return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [this, ch, oid, &st] {
 567       auto c = static_cast<AlienCollection*>(ch.get());
 568       store->stat(c->collection, oid, &st);
 569       return st;
 570     });
 571   });
 572 }
 573
 574 auto AlienStore::omap_get_header(CollectionRef ch,
 575                                  const ghobject_t& oid)
 576   -> get_attr_errorator::future<ceph::bufferlist>
 577 {
 578   assert(tp);
 579   return do_with_op_gate(ceph::bufferlist(), [=, this](auto& bl) {
 580     return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this, &bl] {
 581       auto c = static_cast<AlienCollection*>(ch.get());
 582       return store->omap_get_header(c->collection, oid, &bl);
 583     }).then([&bl](int r) -> get_attr_errorator::future<ceph::bufferlist> {
 584       if (r == -ENOENT) {
 585         return crimson::ct_error::enoent::make();
 586       } else if (r < 0) {
 587         logger().error("omap_get_header: {}", r);
 588         ceph_assert(0 == "impossible");
 589       } else {
 590         return get_attr_errorator::make_ready_future<ceph::bufferlist>(
 591           std::move(bl));
 592       }
 593     });
 594   });
 595 }
 596
 597 AlienStore::read_errorator::future<std::map<uint64_t, uint64_t>> AlienStore::fiemap(
 598   CollectionRef ch,
 599   const ghobject_t& oid,
 600   uint64_t off,
 601   uint64_t len)
 602 {
 603   assert(tp);
 604   return do_with_op_gate(std::map<uint64_t, uint64_t>(), [=, this](auto& destmap) {
 605     return tp->submit(ch->get_cid().hash_to_shard(tp->size()), [=, this, &destmap] {
 606       auto c = static_cast<AlienCollection*>(ch.get());
 607       return store->fiemap(c->collection, oid, off, len, destmap);
 608     }).then([&destmap](int r)
 609       -> read_errorator::future<std::map<uint64_t, uint64_t>> {
 610       if (r == -ENOENT) {
 611         return crimson::ct_error::enoent::make();
 612       } else {
 613         return read_errorator::make_ready_future<std::map<uint64_t, uint64_t>>(
 614           std::move(destmap));
 615       }
 616     });
 617   });
 618 }
 619
 620 }