#include "crimson/os/seastore/extent_placement_manager.h"
-#include "crimson/os/seastore/segment_cleaner.h"
-
-namespace {
- seastar::logger& logger() {
- return crimson::get_logger(ceph_subsys_seastore_tm);
- }
-}
-
SET_SUBSYS(seastore_tm);
namespace crimson::os::seastore {
SegmentedAllocator::SegmentedAllocator(
SegmentProvider& sp,
SegmentManager& sm)
- : segment_provider(sp),
- segment_manager(sm)
{
std::generate_n(
std::back_inserter(writers),
crimson::common::get_conf<uint64_t>(
"seastore_init_rewrite_segments_num_per_device"),
[&] {
- return Writer{
- segment_provider,
- segment_manager};
- });
+ return Writer{sp, sm};
+ }
+ );
}
SegmentedAllocator::Writer::write_iertr::future<>
ool_record_t& record)
{
LOG_PREFIX(SegmentedAllocator::Writer::_write);
- record.set_base(allocated_to);
+ record.set_base(segment_allocator.get_written_to());
auto record_size = record.get_encoded_record_length();
- allocated_to += record_size.get_encoded_length();
- segment_provider.update_segment_avail_bytes(
- paddr_t::make_seg_paddr(
- current_segment->get_segment_id(),
- allocated_to));
bufferlist bl = record.encode(
- current_segment->get_segment_id(),
- 0);
+ segment_allocator.get_segment_id(),
+ segment_allocator.get_nonce());
+ assert(bl.length() == record_size.get_encoded_length());
DEBUGT(
"written {} extents, {} bytes to segment {} at {}",
t,
record.get_num_extents(),
bl.length(),
- current_segment->get_segment_id(),
+ segment_allocator.get_segment_id(),
record.get_base());
// account transactional ool writes before write()
}
return trans_intr::make_interruptible(
- current_segment->write(record.get_base(), bl
- ).safe_then([FNAME, &t, base=record.get_base(), cs=current_segment] {
- DEBUGT("written {} {}", t, cs->get_segment_id(), base);
- })
+ segment_allocator.write(bl).discard_result()
);
}
LOG_PREFIX(SegmentedAllocator::Writer::do_write);
assert(!extents.empty());
if (roll_promise.has_value()) {
- return roll_promise->get_shared_future(
- ).then([this, &t, &extents] {
+ return trans_intr::make_interruptible(
+ roll_promise->get_shared_future()
+ ).then_interruptible([this, &t, &extents] {
return do_write(t, extents);
});
}
- assert(current_segment);
+ assert(segment_allocator.can_write());
- ool_record_t record(segment_manager.get_block_size());
+ ool_record_t record(segment_allocator.get_block_size());
for (auto it = extents.begin(); it != extents.end();) {
auto& extent = *it;
auto wouldbe_length = record.get_wouldbe_encoded_record_length(extent);
- if (_needs_roll(wouldbe_length)) {
+ if (segment_allocator.needs_roll(wouldbe_length)) {
// reached the segment end, write and roll
assert(!roll_promise.has_value());
roll_promise = seastar::shared_promise<>();
auto num_extents = record.get_num_extents();
- DEBUGT(
- "end of segment, writing {} extents to segment {} at {}",
- t,
- num_extents,
- current_segment->get_segment_id(),
- allocated_to);
+ DEBUGT("end of segment, writing {} extents", t, num_extents);
return (num_extents ?
_write(t, record) :
write_iertr::now()
).si_then([this] {
- return roll_segment();
+ return segment_allocator.roll();
}).finally([this] {
roll_promise->set_value();
roll_promise.reset();
it = extents.erase(it);
}
- DEBUGT(
- "writing {} extents to segment {} at {}",
- t,
- record.get_num_extents(),
- current_segment->get_segment_id(),
- allocated_to);
+ DEBUGT("writing {} extents", t, record.get_num_extents());
return _write(t, record);
}
return write_iertr::now();
}
return seastar::with_gate(write_guard, [this, &t, &extents] {
- if (!roll_promise.has_value() && !current_segment) {
+ if (!roll_promise.has_value() &&
+ !segment_allocator.can_write()) {
roll_promise = seastar::shared_promise<>();
return trans_intr::make_interruptible(
- roll_segment().finally([this] {
- roll_promise->set_value();
- roll_promise.reset();
- })
- ).si_then([this, &t, &extents] {
+ segment_allocator.open().discard_result()
+ ).finally([this] {
+ roll_promise->set_value();
+ roll_promise.reset();
+ }).si_then([this, &t, &extents] {
return do_write(t, extents);
});
}
});
}
-bool SegmentedAllocator::Writer::_needs_roll(seastore_off_t length) const {
- return allocated_to + length > current_segment->get_write_capacity();
-}
-
-SegmentedAllocator::Writer::init_segment_ertr::future<>
-SegmentedAllocator::Writer::init_segment(Segment& segment) {
- bufferptr bp(
- ceph::buffer::create_page_aligned(
- segment_manager.get_block_size()));
- bp.zero();
- auto header =segment_header_t{
- OOL_SEG_SEQ,
- segment.get_segment_id(),
- NO_DELTAS, 0};
- logger().debug("SegmentedAllocator::Writer::init_segment: initting {}, {}",
- segment.get_segment_id(),
- header);
- ceph::bufferlist bl;
- encode(header, bl);
- bl.cbegin().copy(bl.length(), bp.c_str());
- bl.clear();
- bl.append(bp);
- allocated_to = segment_manager.get_block_size();
- return segment.write(0, bl).handle_error(
- crimson::ct_error::input_output_error::pass_further{},
- crimson::ct_error::assert_all{
- "Invalid error when initing segment"}
- );
-}
-
-SegmentedAllocator::Writer::roll_segment_ertr::future<>
-SegmentedAllocator::Writer::roll_segment() {
- LOG_PREFIX(SegmentedAllocator::Writer::roll_segment);
- DEBUG("start");
- assert(roll_promise.has_value());
- return [this, FNAME] {
- if (current_segment) {
- auto seg_to_close = std::move(current_segment);
- if (write_guard.is_closed()) {
- DEBUG("write_guard is closed, should be stopping");
- return seg_to_close->close(
- ).safe_then([seg_to_close=std::move(seg_to_close)] {});
- } else {
- DEBUG("rolling OOL segment, close {} ...", seg_to_close->get_segment_id());
- (void) seastar::with_gate(write_guard,
- [this, seg_to_close=std::move(seg_to_close)]() mutable
- {
- return seg_to_close->close(
- ).safe_then([this, seg_to_close=std::move(seg_to_close)] {
- segment_provider.close_segment(seg_to_close->get_segment_id());
- });
- });
- return Segment::close_ertr::now();
- }
- } else {
- DEBUG("rolling OOL segment, no current ...");
- return Segment::close_ertr::now();
- }
- }().safe_then([this] {
- auto new_segment_id = segment_provider.get_segment(
- segment_manager.get_device_id(), OOL_SEG_SEQ);
- return segment_manager.open(new_segment_id);
- }).safe_then([this, FNAME](auto segref) {
- DEBUG("opened new segment: {}", segref->get_segment_id());
- return init_segment(*segref
- ).safe_then([segref=std::move(segref), this, FNAME] {
- assert(!current_segment);
- current_segment = segref;
- DEBUG("inited new segment: {}", segref->get_segment_id());
- });
- }).handle_error(
- roll_segment_ertr::pass_further{},
- crimson::ct_error::all_same_way([] { ceph_assert(0 == "TODO"); })
- );
-}
-
}
#include "seastar/core/shared_future.hh"
#include "crimson/os/seastore/cached_extent.h"
+#include "crimson/os/seastore/journal/segment_allocator.h"
#include "crimson/os/seastore/logging.h"
-#include "crimson/os/seastore/segment_manager.h"
#include "crimson/os/seastore/transaction.h"
namespace crimson::os::seastore {
class SegmentedAllocator : public ExtentAllocator {
class Writer : public ExtentOolWriter {
public:
- Writer(
- SegmentProvider& sp,
- SegmentManager& sm)
- : segment_provider(sp),
- segment_manager(sm)
- {}
+ Writer(SegmentProvider& sp, SegmentManager& sm)
+ : segment_allocator(segment_type_t::OOL, sp, sm) {}
+
Writer(Writer &&) = default;
write_iertr::future<> write(
stop_ertr::future<> stop() final {
return write_guard.close().then([this] {
- if (current_segment) {
- return current_segment->close();
- } else {
- return Segment::close_ertr::now();
- }
+ return segment_allocator.close();
});
}
+
private:
write_iertr::future<> do_write(
Transaction& t,
std::list<LogicalCachedExtentRef>& extent);
- bool _needs_roll(seastore_off_t length) const;
-
write_iertr::future<> _write(
Transaction& t,
ool_record_t& record);
- using roll_segment_ertr = crimson::errorator<
- crimson::ct_error::input_output_error>;
- roll_segment_ertr::future<> roll_segment();
-
- using init_segment_ertr = crimson::errorator<
- crimson::ct_error::input_output_error>;
- init_segment_ertr::future<> init_segment(Segment& segment);
-
- SegmentProvider& segment_provider;
- SegmentManager& segment_manager;
- SegmentRef current_segment;
- seastore_off_t allocated_to = 0;
+ journal::SegmentAllocator segment_allocator;
std::optional<seastar::shared_promise<>> roll_promise;
seastar::gate write_guard;
};
});
}
private:
- SegmentProvider& segment_provider;
- SegmentManager& segment_manager;
std::vector<Writer> writers;
};
projects / ceph.git / commitdiff