ceph/src/crimson/os/seastore/transaction_manager.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 "include/denc.h"
   5 #include "include/intarith.h"
   6
   7 #include "crimson/common/log.h"
   8
   9 #include "crimson/os/seastore/transaction_manager.h"
  10 #include "crimson/os/seastore/segment_manager.h"
  11 #include "crimson/os/seastore/journal.h"
  12
  13 namespace {
  14   seastar::logger& logger() {
  15     return crimson::get_logger(ceph_subsys_filestore);
  16   }
  17 }
  18
  19 namespace crimson::os::seastore {
  20
  21 TransactionManager::TransactionManager(
  22   SegmentManager &segment_manager,
  23   SegmentCleaner &segment_cleaner,
  24   Journal &journal,
  25   Cache &cache,
  26   LBAManager &lba_manager)
  27   : segment_manager(segment_manager),
  28     segment_cleaner(segment_cleaner),
  29     cache(cache),
  30     lba_manager(lba_manager),
  31     journal(journal)
  32 {}
  33
  34 TransactionManager::mkfs_ertr::future<> TransactionManager::mkfs()
  35 {
  36   return journal.open_for_write().safe_then([this](auto addr) {
  37     logger().debug("TransactionManager::mkfs: about to do_with");
  38     segment_cleaner.set_journal_head(addr);
  39     return seastar::do_with(
  40       create_transaction(),
  41       [this](auto &transaction) {
  42         logger().debug("TransactionManager::mkfs: about to cache.mkfs");
  43         cache.init();
  44         return cache.mkfs(*transaction
  45         ).safe_then([this, &transaction] {
  46           return lba_manager.mkfs(*transaction);
  47         }).safe_then([this, &transaction] {
  48           logger().debug("TransactionManager::mkfs: about to submit_transaction");
  49           return submit_transaction(std::move(transaction)).handle_error(
  50             crimson::ct_error::eagain::handle([] {
  51               ceph_assert(0 == "eagain impossible");
  52               return mkfs_ertr::now();
  53             }),
  54             mkfs_ertr::pass_further{}
  55           );
  56         });
  57       });
  58   }).safe_then([this] {
  59     return journal.close();
  60   });
  61 }
  62
  63 TransactionManager::mount_ertr::future<> TransactionManager::mount()
  64 {
  65   cache.init();
  66   return journal.replay([this](auto seq, auto paddr, const auto &e) {
  67     return cache.replay_delta(seq, paddr, e);
  68   }).safe_then([this] {
  69     return journal.open_for_write();
  70   }).safe_then([this](auto addr) {
  71     segment_cleaner.set_journal_head(addr);
  72     return seastar::do_with(
  73       make_weak_transaction(),
  74       [this](auto &t) {
  75         return cache.init_cached_extents(*t, [this](auto &t, auto &e) {
  76           return lba_manager.init_cached_extent(t, e);
  77         }).safe_then([this, &t] {
  78           assert(segment_cleaner.debug_check_space(
  79                    *segment_cleaner.get_empty_space_tracker()));
  80           return lba_manager.scan_mapped_space(
  81             *t,
  82             [this](paddr_t addr, extent_len_t len) {
  83               logger().debug("TransactionManager::mount: marking {}~{} used",
  84                            addr,
  85                            len);
  86               segment_cleaner.mark_space_used(
  87                 addr,
  88                 len ,
  89                 /* init_scan = */ true);
  90             });
  91         });
  92       });
  93   }).safe_then([this] {
  94     segment_cleaner.complete_init();
  95   }).handle_error(
  96     mount_ertr::pass_further{},
  97     crimson::ct_error::all_same_way([] {
  98       ceph_assert(0 == "unhandled error");
  99       return mount_ertr::now();
 100     }));
 101 }
 102
 103 TransactionManager::close_ertr::future<> TransactionManager::close() {
 104   return cache.close(
 105   ).safe_then([this] {
 106     return journal.close();
 107   });
 108 }
 109
 110 TransactionManager::ref_ret TransactionManager::inc_ref(
 111   Transaction &t,
 112   LogicalCachedExtentRef &ref)
 113 {
 114   return lba_manager.incref_extent(t, ref->get_laddr()).safe_then([](auto r) {
 115     return r.refcount;
 116   }).handle_error(
 117     ref_ertr::pass_further{},
 118     ct_error::all_same_way([](auto e) {
 119       ceph_assert(0 == "unhandled error, TODO");
 120     }));
 121 }
 122
 123 TransactionManager::ref_ret TransactionManager::inc_ref(
 124   Transaction &t,
 125   laddr_t offset)
 126 {
 127   return lba_manager.incref_extent(t, offset).safe_then([](auto result) {
 128     return result.refcount;
 129   });
 130 }
 131
 132 TransactionManager::ref_ret TransactionManager::dec_ref(
 133   Transaction &t,
 134   LogicalCachedExtentRef &ref)
 135 {
 136   return lba_manager.decref_extent(t, ref->get_laddr()
 137   ).safe_then([this, &t, ref](auto ret) {
 138     if (ret.refcount == 0) {
 139       logger().debug(
 140         "TransactionManager::dec_ref: extent {} refcount 0",
 141         *ref);
 142       cache.retire_extent(t, ref);
 143     }
 144     return ret.refcount;
 145   });
 146 }
 147
 148 TransactionManager::ref_ret TransactionManager::dec_ref(
 149   Transaction &t,
 150   laddr_t offset)
 151 {
 152   return lba_manager.decref_extent(t, offset
 153   ).safe_then([this, offset, &t](auto result) -> ref_ret {
 154     if (result.refcount == 0) {
 155       logger().debug(
 156         "TransactionManager::dec_ref: offset {} refcount 0",
 157         offset);
 158       return cache.retire_extent_if_cached(t, result.addr).safe_then([] {
 159         return ref_ret(
 160           ref_ertr::ready_future_marker{},
 161           0);
 162       });
 163     } else {
 164       return ref_ret(
 165         ref_ertr::ready_future_marker{},
 166         result.refcount);
 167     }
 168   });
 169 }
 170
 171 TransactionManager::submit_transaction_ertr::future<>
 172 TransactionManager::submit_transaction(
 173   TransactionRef t)
 174 {
 175   logger().debug("TransactionManager::submit_transaction");
 176   return segment_cleaner.do_immediate_work(*t
 177   ).safe_then([this, t=std::move(t)]() mutable -> submit_transaction_ertr::future<> {
 178     auto record = cache.try_construct_record(*t);
 179     if (!record) {
 180       return crimson::ct_error::eagain::make();
 181     }
 182
 183     return journal.submit_record(std::move(*record)
 184     ).safe_then([this, t=std::move(t)](auto p) mutable {
 185       auto [addr, journal_seq] = p;
 186       segment_cleaner.set_journal_head(journal_seq);
 187       cache.complete_commit(*t, addr, journal_seq, &segment_cleaner);
 188       lba_manager.complete_transaction(*t);
 189       auto to_release = t->get_segment_to_release();
 190       if (to_release != NULL_SEG_ID) {
 191         segment_cleaner.mark_segment_released(to_release);
 192         return segment_manager.release(to_release);
 193       } else {
 194         return SegmentManager::release_ertr::now();
 195       }
 196     }).handle_error(
 197       submit_transaction_ertr::pass_further{},
 198       crimson::ct_error::all_same_way([](auto e) {
 199         ceph_assert(0 == "Hit error submitting to journal");
 200       }));
 201   });
 202 }
 203
 204 TransactionManager::get_next_dirty_extents_ret
 205 TransactionManager::get_next_dirty_extents(journal_seq_t seq)
 206 {
 207   return cache.get_next_dirty_extents(seq);
 208 }
 209
 210 TransactionManager::rewrite_extent_ret TransactionManager::rewrite_extent(
 211   Transaction &t,
 212   CachedExtentRef extent)
 213 {
 214   {
 215     auto updated = cache.update_extent_from_transaction(t, extent);
 216     if (!updated) {
 217       logger().debug(
 218         "{}: {} is already retired, skipping",
 219         __func__,
 220         *extent);
 221       return rewrite_extent_ertr::now();
 222     }
 223     extent = updated;
 224   }
 225
 226   if (extent->get_type() == extent_types_t::ROOT) {
 227     logger().debug(
 228       "{}: marking root {} for rewrite",
 229       __func__,
 230       *extent);
 231     cache.duplicate_for_write(t, extent);
 232     return rewrite_extent_ertr::now();
 233   }
 234   return lba_manager.rewrite_extent(t, extent);
 235 }
 236
 237 TransactionManager::get_extent_if_live_ret TransactionManager::get_extent_if_live(
 238   Transaction &t,
 239   extent_types_t type,
 240   paddr_t addr,
 241   laddr_t laddr,
 242   segment_off_t len)
 243 {
 244   CachedExtentRef ret;
 245   auto status = cache.get_extent_if_cached(t, addr, &ret);
 246   if (status != Transaction::get_extent_ret::ABSENT) {
 247     return get_extent_if_live_ret(
 248       get_extent_if_live_ertr::ready_future_marker{},
 249       ret);
 250   }
 251
 252   if (is_logical_type(type)) {
 253     return lba_manager.get_mapping(
 254       t,
 255       laddr,
 256       len).safe_then([=, &t](lba_pin_list_t pins) {
 257         ceph_assert(pins.size() <= 1);
 258         if (pins.empty()) {
 259           return get_extent_if_live_ret(
 260             get_extent_if_live_ertr::ready_future_marker{},
 261             CachedExtentRef());
 262         }
 263
 264         auto pin = std::move(pins.front());
 265         pins.pop_front();
 266         ceph_assert(pin->get_laddr() == laddr);
 267         ceph_assert(pin->get_length() == (extent_len_t)len);
 268         if (pin->get_paddr() == addr) {
 269           return cache.get_extent_by_type(
 270             t,
 271             type,
 272             addr,
 273             laddr,
 274             len).safe_then(
 275               [this, pin=std::move(pin)](CachedExtentRef ret) mutable {
 276                 auto lref = ret->cast<LogicalCachedExtent>();
 277                 if (!lref->has_pin()) {
 278                   lref->set_pin(std::move(pin));
 279                   lba_manager.add_pin(lref->get_pin());
 280                 }
 281                 return get_extent_if_live_ret(
 282                   get_extent_if_live_ertr::ready_future_marker{},
 283                   ret);
 284               });
 285         } else {
 286           return get_extent_if_live_ret(
 287             get_extent_if_live_ertr::ready_future_marker{},
 288             CachedExtentRef());
 289         }
 290       });
 291   } else {
 292     logger().debug(
 293       "TransactionManager::get_extent_if_live: non-logical extent {}",
 294       addr);
 295     return lba_manager.get_physical_extent_if_live(
 296       t,
 297       type,
 298       addr,
 299       laddr,
 300       len);
 301   }
 302 }
 303
 304 TransactionManager::~TransactionManager() {}
 305
 306 }