desc: CPU cores on which POSIX threads alienized to seastar will run in cpuset(7) format
flags:
- startup
+- name: crimson_test_workload
+ type: bool
+ level: dev
+ desc: enable test workload
+ default: false
# Seastore options
level: dev
desc: select write through policy when data length is greater than this value.
default: 512_K
+- name: seastore_test_workload_write_through_probability
+ type: float
+ level: dev
+ desc: the probability of write the incoming data to slow devices
+ default: 0.5
+- name: seastore_test_workload_2Q_promote_probability
+ type: float
+ level: dev
+ desc: the probability of promote the extents to the faster devices
+ default: 0.5
+- name: seastore_test_workload_force_prcess_background_tasks_period
+ type: uint
+ level: dev
+ desc: the seconds of period for force process background tasks
+ default: 5
return static_cast<std::size_t>(ret);
}
-seastar::future<> JournalTrimmerImpl::trim() {
+seastar::future<> JournalTrimmerImpl::trim(bool force) {
return seastar::when_all(
- [this] {
- if (should_trim_alloc()) {
+ [this, force] {
+ if (force || should_trim_alloc()) {
return trim_alloc(
).handle_error(
crimson::ct_error::assert_all(
return seastar::now();
}
},
- [this] {
- if (should_start_trim_dirty()) {
+ [this, force] {
+ if (force || should_start_trim_dirty()) {
return trim_dirty(
).handle_error(
crimson::ct_error::assert_all(
Transaction &t,
CachedExtentRef extent) = 0;
+ virtual promote_extent_ret promote_extents_from_disk(
+ Transaction &t,
+ paddr_t paddr) = 0;
+
/**
* demote_region
*
reserved_usage -= usage;
}
- seastar::future<> trim();
+ seastar::future<> trim(bool force);
static JournalTrimmerImplRef create(
store_index_t store_index,
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
// vim: ts=8 sw=2 sts=2 expandtab
+#include "crimson/common/coroutine.h"
#include "crimson/os/seastore/backref/btree_backref_manager.h"
SET_SUBSYS(seastore_backref);
});
}
+BtreeBackrefManager::scan_device_ret
+BtreeBackrefManager::scan_device(
+ Transaction &t,
+ paddr_t paddr,
+ scan_device_func_t &f)
+{
+ LOG_PREFIX(BtreeBackrefManager::scan_device);
+ auto c = get_context(t);
+ auto croot = co_await cache.get_root(t);
+ auto btree = BackrefBtree(croot);
+ auto iter = co_await btree.lower_bound(c, paddr);
+ while (!iter.is_end()) {
+ auto key = iter.get_key();
+ auto bentry = cache.get_cached_backref_entry(key);
+ if (bentry) {
+ assert(bentry->paddr == key);
+ DEBUGT("found in cache: {} {}", t, bentry->paddr, bentry->laddr);
+ }
+ if (bentry && bentry->laddr == L_ADDR_NULL) {
+ DEBUGT("{} is removed", t, bentry->paddr);
+ iter = co_await iter.next(c);
+ continue;
+ }
+ if (key.get_device_id() == paddr.get_device_id()) {
+ auto val = iter.get_val();
+ if (bentry && bentry->laddr != val.laddr) {
+ DEBUGT("{} changed from {} to {}",
+ t, bentry->paddr, val.laddr, bentry->laddr);
+ iter = co_await iter.next(c);
+ continue;
+ }
+ DEBUGT("scanned {}, {}", t, key, val.laddr);
+ auto ret = co_await f(key, val.len, val.type, val.laddr);
+ if (ret == seastar::stop_iteration::yes) {
+ break;
+ }
+ } else if (key.get_device_id() > paddr.get_device_id()) {
+ break;
+ }
+ iter = co_await iter.next(c);
+ }
+ co_return;
+}
+
base_iertr::future<> _init_cached_extent(
op_context_t c,
const CachedExtentRef &e,
Transaction &t,
scan_mapped_space_func_t &&f) final;
+ scan_device_ret scan_device(
+ Transaction &t,
+ paddr_t paddr,
+ scan_device_func_t &f) final;
+
init_cached_extent_ret init_cached_extent(
Transaction &t,
CachedExtentRef e) final;
Transaction &t,
scan_mapped_space_func_t &&f) = 0;
+ using scan_device_ret = base_iertr::future<>;
+ using scan_device_func_t = std::function<
+ base_iertr::future<seastar::stop_iteration>(
+ paddr_t, extent_len_t, extent_types_t, laddr_t)>;
+ virtual scan_device_ret scan_device(
+ Transaction &t,
+ paddr_t start,
+ scan_device_func_t &f) = 0;
+
virtual ~BackrefManager() {}
};
delta_based_overwrite_enabled(
crimson::common::get_conf<Option::size_t>(
"seastore_data_delta_based_overwrite") > 0),
+ force_backref(crimson::common::get_conf<bool>("crimson_test_workload")),
pinboard(create_extent_pinboard(
crimson::common::get_conf<Option::size_t>(
"seastore_cachepin_size_pershard"),
}
bool can_drop_backref() const {
- return epm.is_pure_rbm();
+ return epm.is_pure_rbm() && !force_backref;
}
void update_read_ratio(Transaction &t) {
return res;
}
+ std::optional<backref_entry_t> get_cached_backref_entry(paddr_t addr) {
+ auto it = backref_entry_mset.lower_bound(
+ addr,
+ backref_entry_t::cmp_t());
+ if (it == backref_entry_mset.end()) {
+ return std::nullopt;
+ }
+ while (it->paddr == addr) {
+ auto &backref_entry = *it;
+ ++it;
+ if (it->paddr != addr || it == backref_entry_mset.end()) {
+ return backref_entry;
+ }
+ }
+ return std::nullopt;
+ }
+
const backref_entry_mset_t& get_backref_entry_mset() {
return backref_entry_mset;
}
transaction_id_t next_id = 0;
+ const bool force_backref = false;
+
/**
* dirty
*
prior.last_committed_crc = extent.last_committed_crc;
prior.dirty_from = extent.dirty_from;
prior.length = extent.length;
- prior.loaded_length = extent.loaded_length;
- prior.buffer_space = std::move(extent.buffer_space);
+ // XXX: at present, zero loaded_length extents here
+ // must have been created by promoting/demoting them.
+ if (likely(extent.loaded_length != 0)) {
+ assert(prior.loaded_length == extent.loaded_length);
+ prior.buffer_space = std::move(extent.buffer_space);
+ }
// XXX: We can go ahead and change the prior's version because
// transactions don't hold a local view of the version field,
// unlike FixedKVLeafNode::modifications
class ExtentPromoter {
public:
- ExtentPromoter(size_t promotion_size, ExtentPlacementManager &epm)
- : promotion_size(promotion_size), epm(epm) {}
+ ExtentPromoter(
+ size_t promotion_size,
+ ExtentPlacementManager &epm)
+ : promotion_size(promotion_size),
+ epm(epm),
+ test_workload(crimson::common::get_conf<bool>("crimson_test_workload"))
+ {}
~ExtentPromoter() {
clear();
}
size_t get_promotion_size() const {
- return current_contents;
+ if (unlikely(test_workload)) {
+ return current_contents >= promotion_size
+ ? current_contents
+ : promotion_size;
+ } else {
+ return current_contents;
+ }
}
void set_background_callback(BackgroundListener *l) {
}
bool should_run_promote() const {
- return enabled() && current_contents >= promotion_size;
+ return enabled() &&
+ (current_contents >= promotion_size
+ || test_workload);
}
std::size_t get_promoted_size() const {
std::size_t promote_size = 0;
std::list<CachedExtentRef> extents;
DEBUGT("start promote", t);
- for (auto &extent : list) {
- DEBUGT("promote {} to the hot tier", t, extent);
- ceph_assert(extent.is_stable_clean());
- ceph_assert(extent.get_pin_state() == extent_pin_state_t::PendingPromote);
- promote_size += extent.get_length();
- t.add_to_read_set(&extent);
- extents.emplace_back(&extent);
- }
- for (auto &extent : extents) {
- remove_extent(*extent, extent_pin_state_t::Fresh);
- }
- for (auto &extent : extents) {
- co_await trans_intr::make_interruptible(extent->wait_io());
- co_await ecb->promote_extent(t, extent);
+ if (current_contents < promotion_size && test_workload) {
+ auto id = epm.get_cold_device_id();
+ paddr_t start = P_ADDR_NULL;
+ if (device_id_to_paddr_type(id) == paddr_types_t::SEGMENT) {
+ start = paddr_t::make_seg_paddr(id, 0, 0);
+ } else {
+ start = paddr_t::make_blk_paddr(id, 0);
+ }
+ co_await ecb->promote_extents_from_disk(t, start);
+ } else {
+ for (auto &extent : list) {
+ DEBUGT("promote {} to the hot tier", t, extent);
+ ceph_assert(extent.is_stable_clean());
+ ceph_assert(extent.get_pin_state() == extent_pin_state_t::PendingPromote);
+ promote_size += extent.get_length();
+ t.add_to_read_set(&extent);
+ extents.emplace_back(&extent);
+ }
+ for (auto &extent : extents) {
+ remove_extent(*extent, extent_pin_state_t::Fresh);
+ }
+ for (auto &extent : extents) {
+ co_await trans_intr::make_interruptible(extent->wait_io());
+ co_await ecb->promote_extent(t, extent);
+ }
}
// existing extents in lru will be retired after transaction submitted
co_await ecb->submit_transaction_direct(t);
size_t promoted_count;
size_t promoted_size;
+ bool test_workload = false;
};
class ExtentPinboardLRU : public ExtentPinboard {
: warm_in(warm_in_capacity),
warm_out(warm_out_capacity),
hot(hot_capacity),
- promoter(promotion_size, epm)
+ promoter(promotion_size, epm),
+ test_workload(crimson::common::get_conf<bool>("crimson_test_workload")),
+ TwoQ_promote_probability(crimson::common::get_conf<double>(
+ "seastore_test_workload_2Q_promote_probability"))
{
LOG_PREFIX(ExtentPinboardTwoQ::ExtentPinboardTwoQ);
INFO("created, warm_in_capacity=0x{:x}B, warm_out_capacity=0x{:x}B, "
// to the warm out queue.
continue;
}
+ if (promoter.should_promote_extent(*extent)
+ && test_workload
+ && (double(std::rand() % 100) / 100.0) <= TwoQ_promote_probability) {
+ promoter.add_extent(*extent);
+ continue;
+ }
auto lext = extent->cast<LogicalCachedExtent>();
auto laddr = lext->get_laddr();
auto end = extent->get_last_touch_end();
// hit and miss indicates if an extent is linked when touching it
uint64_t hit = 0;
uint64_t miss = 0;
+ bool test_workload = false;
+ double TwoQ_promote_probability = 0;
};
void ExtentPinboardTwoQ::get_stats(
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
// vim: ts=8 sw=2 sts=2 expandtab expandtab
+#include <chrono>
+#include <seastar/core/sleep.hh>
+
#include "crimson/os/seastore/extent_placement_manager.h"
#include "crimson/common/errorator-utils.h"
{
assert(is_running());
while (is_running()) {
- if (background_should_run()) {
+ if (background_should_run()
+ || force_run_background()) {
log_state("run(background)");
co_await do_background_cycle();
// Edge-triggered: yield only when a blocked IO was actually woken, so the
if (cold_cleaner) {
cold_cleaner->maybe_adjust_thresholds();
}
+ if (unlikely(test_workload && force_process_state != ForceProcessState::STOP)) {
+ last_process_state = force_process_state;
+ force_process_state = ForceProcessState::STOP;
+ set_next_arm_timepoint();
+ }
} else {
log_state("run(block)");
assert(!blocking_background);
+ if (unlikely(test_workload)) {
+ set_next_arm_timepoint();
+ }
blocking_background = seastar::promise<>();
co_await blocking_background->get_future();
// After waking (typically because arm_blocking_io_and_wake() kicked us),
}
}
+ bool force_trim = false;
+ bool should_abort_cleaner_usage = true;
+ if (unlikely(test_workload && force_process_state == ForceProcessState::TRIM)) {
+ if (!proceed_trim) {
+ should_abort_cleaner_usage = false;
+ }
+ proceed_trim = true;
+ force_trim = true;
+ }
+
if (proceed_trim) {
DEBUG("started trimming...");
- return trimmer->trim(
- ).finally([this, trim_usage, FNAME] {
+ return trimmer->trim(force_trim
+ ).finally([this, trim_usage, should_abort_cleaner_usage, FNAME] {
DEBUG("finished trimming");
- abort_cleaner_usage(trim_usage, {true, true});
+ if (should_abort_cleaner_usage) {
+ abort_cleaner_usage(trim_usage, {true, true});
+ }
});
} else {
assert(!proceed_trim);
proceed_demote = true;
}
+ bool abort_cold_cleaner_usage = true;
+ if (unlikely(test_workload && force_process_state == ForceProcessState::CLEAN)) {
+ if (!proceed_clean_main) {
+ abort_cold_cleaner_usage = false;
+ }
+ proceed_clean_main = main_cleaner->can_clean_space();
+ if (has_cold_tier()) {
+ proceed_clean_cold = cold_cleaner->can_clean_space();
+ }
+ if (logical_bucket) {
+ proceed_demote = logical_bucket->could_demote();
+ }
+ }
+
if (!proceed_clean_main && !proceed_clean_cold && !proceed_demote) {
ceph_abort_msg("no background process will start");
}
return seastar::when_all(
- [this, FNAME, proceed_clean_main,
+ [this, FNAME, proceed_clean_main, abort_cold_cleaner_usage,
should_clean_main_for_trim, main_cold_usage] {
if (!proceed_clean_main) {
return seastar::now();
crimson::ct_error::assert_all(
"do_background_cycle encountered invalid error in main clean_space"
)
- ).finally([this, main_cold_usage, FNAME] {
+ ).finally([this, main_cold_usage, abort_cold_cleaner_usage, FNAME] {
DEBUG("finished clean main");
- abort_cold_usage(main_cold_usage, true);
+ if (abort_cold_cleaner_usage) {
+ abort_cold_usage(main_cold_usage, true);
+ }
});
},
[this, FNAME, proceed_clean_cold,
ceph_assert(!blocking_promote);
blocking_promote = seastar::promise<>();
return blocking_promote->get_future();
+ }).then([this] {
+ if (unlikely(test_workload)) {
+ return seastar::sleep(std::chrono::seconds(
+ force_process_half_life));
+ }
+ return seastar::now();
}).then([] {
return seastar::stop_iteration::no;
});
max_data_allocation_size(crimson::common::get_conf<Option::size_t>(
"seastore_max_data_allocation_size")),
write_through_size(crimson::common::get_conf<Option::size_t>(
- "seastore_write_through_size"))
+ "seastore_write_through_size")),
+ test_workload(crimson::common::get_conf<bool>("crimson_test_workload")),
+ write_through_probability(crimson::common::get_conf<double>(
+ "seastore_test_workload_write_through_probability"))
{
LOG_PREFIX(ExtentPlacementManager::ExtentPlacementManager);
devices_by_id.resize(DEVICE_ID_MAX, nullptr);
}
write_policy_t get_write_policy(extent_types_t type, extent_len_t length) const {
- if (has_cold_tier() && length >= write_through_size && is_data_type(type)) {
- return write_policy_t::WRITE_THROUGH;
+ if (has_cold_tier() && is_data_type(type)) {
+ if (length >= write_through_size
+ || (test_workload
+ && (double(std::rand() % 100) / 100.0) <=
+ write_through_probability))
+ {
+ return write_policy_t::WRITE_THROUGH;
+ }
}
return write_policy_t::WRITE_BACK;
}
return hot_tier_generations - 1;
}
+ device_id_t get_cold_device_id() const {
+ return background_process.get_cold_device_id();
+ }
+
private:
rewrite_gen_t adjust_generation(
data_category_t category,
get_conf<Option::size_t>("seastore_logical_bucket_capacity"),
get_conf<Option::size_t>("seastore_logical_bucket_proceed_size_per_cycle"));
}
+ LOG_PREFIX(BackgroundProcess::init);
+ test_workload = crimson::common::get_conf<bool>("crimson_test_workload");
+ force_process_half_life = crimson::common::get_conf<uint64_t>(
+ "seastore_test_workload_force_prcess_background_tasks_period");
+ force_background_timer.set_callback([this] { wake_half_life(); });
+ write_through_probability = crimson::common::get_conf<double>(
+ "seastore_test_workload_write_through_probability");
+ SUBINFO(seastore_epm, "crimson test workload supported, enabled: {}", test_workload);
+ if (test_workload) {
+ set_next_arm_timepoint();
+ }
}
LogicalBucket *get_logical_bucket() {
}
}
+
+ device_id_t get_cold_device_id() const {
+ assert(has_cold_tier());
+ return *cold_cleaner->get_device_ids().begin();
+ }
+
seastar::future<> reserve_projected_usage(io_usage_t usage);
void release_projected_usage(const io_usage_t &usage) {
state_t state = state_t::STOP;
eviction_state_t eviction_state;
+ enum class ForceProcessState : uint8_t{
+ STOP,
+ TRIM,
+ CLEAN,
+ };
+ bool test_workload = false;
+ double write_through_probability = 0;
+ ForceProcessState force_process_state = ForceProcessState::STOP;
+ ForceProcessState last_process_state = ForceProcessState::STOP;
+ seastar::timer<seastar::steady_clock_type> force_background_timer;
+ int force_process_half_life;
+
+ void set_next_arm_timepoint() {
+ assert(test_workload);
+ force_background_timer.rearm(
+ seastar::steady_clock_type::now() +
+ std::chrono::seconds(force_process_half_life));
+ }
+
+ void wake_half_life() {
+ assert(test_workload);
+ if (last_process_state == ForceProcessState::TRIM) {
+ force_process_state = ForceProcessState::CLEAN;
+ } else {
+ force_process_state = ForceProcessState::TRIM;
+ }
+
+ do_wake_background();
+ }
+
+ bool force_run_background() const {
+ return test_workload && force_process_state != ForceProcessState::STOP
+ && (logical_bucket && logical_bucket->could_demote());
+ }
friend class ::transaction_manager_test_t;
};
SegmentSeqAllocatorRef ool_segment_seq_allocator;
extent_len_t max_data_allocation_size = 0;
std::size_t write_through_size;
+ bool test_workload = false;
+ double write_through_probability = 0;
friend class ::transaction_manager_test_t;
friend class Cache;
co_await rdir.close();
}
+ if (sds.empty() && crimson::common::get_conf<bool>("crimson_test_workload")) {
+ // lbc test workload enabled while no secondary devices indicated, create one
+ std::string path = fmt::format("{}/block.1", root);
+ co_await seastar::make_directory(path);
+ DeviceRef sec_dev = co_await Device::make_device(path, dtype, btype);
+ auto p_sec_dev = sec_dev.get();
+ secondaries.emplace_back(std::move(sec_dev));
+ co_await p_sec_dev->start(store_shard_nums);
+ magic_t magic = (magic_t)std::rand();
+ device_id_t id = 0x1;
+ sds.emplace(id, device_spec_t{magic, dtype, btype, id});
+ co_await p_sec_dev->mkfs(
+ device_config_t::create_secondary(new_osd_fsid, id, dtype, btype, magic)
+ ).handle_error(crimson::ct_error::assert_all("not possible"));
+ co_await set_secondaries();
+ }
device_id_t id = 0;
device_type_t d_type = device->get_device_type();
backend_type_t b_type = device->get_backend_type();
t, *mapping.direct_cursor, std::move(promoted_extents));
}
+TransactionManager::promote_extent_ret
+TransactionManager::promote_extents_from_disk(
+ Transaction &t,
+ paddr_t paddr)
+{
+ using scan_device_func_t = BackrefManager::scan_device_func_t;
+ std::size_t size = 0;
+ scan_device_func_t func = [this, &t, &size](
+ paddr_t paddr, extent_len_t length, extent_types_t type, laddr_t laddr)
+ -> base_iertr::future<seastar::stop_iteration> {
+ if (type != extent_types_t::OBJECT_DATA_BLOCK) {
+ co_return seastar::stop_iteration::no;
+ }
+ auto cursor = co_await lba_manager->get_cursor(t, laddr
+ ).handle_error_interruptible(
+ crimson::ct_error::enoent::assert_failure(),
+ crimson::ct_error::pass_further_all{}
+ );
+ if (cursor->is_end() ||
+ !cursor->get_paddr().is_absolute() ||
+ !cache->is_on_cold_tier(cursor->get_paddr())) {
+ // the mapping has been modified and the extent is
+ // either removed or already on the hot tier, skip it.
+ co_return seastar::stop_iteration::no;
+ }
+ assert(cursor->is_direct());
+ assert(!cursor->has_shadow_paddr());
+ auto extent = co_await read_cursor_by_type(t, std::move(cursor), type);
+ if (extent->is_stable_dirty()) {
+ // dirty extents shouldn't be promoted as is in
+ // the real world
+ co_return seastar::stop_iteration::no;
+ }
+ auto &pinboard = *cache->get_extent_pinboard();
+ pinboard.remove(*extent);
+ extent->set_pin_state(extent_pin_state_t::Promoting);
+ co_await promote_extent(t, extent);
+ size += length;
+ if (size >= crimson::common::get_conf<
+ Option::size_t>("seastore_cache_promotion_size")) {
+ co_return seastar::stop_iteration::yes;
+ } else {
+ co_return seastar::stop_iteration::no;
+ }
+ };
+ co_await backref_manager->scan_device(t, paddr, func);
+}
+
TransactionManager::rewrite_extents_ret TransactionManager::rewrite_extents(
Transaction &t,
std::vector<CachedExtentRef> &extents,
store_index,
*backref_manager, trimmer_config,
backend_type, roll_start, roll_size,
- !pure_rbm_backend);
+ !pure_rbm_backend
+ || crimson::common::get_conf<bool>("crimson_test_workload")
+ );
AsyncCleanerRef cleaner;
JournalRef journal;
using ExtentCallbackInterface::promote_extent_ret;
promote_extent_ret promote_extent(
Transaction &t,
- CachedExtentRef extent);
+ CachedExtentRef extent) final;
+
+ promote_extent_ret promote_extents_from_disk(
+ Transaction &t,
+ paddr_t paddr) final;
using ExtentCallbackInterface::demote_region_res_t;
using ExtentCallbackInterface::demote_region_ret;
if (run_clean) {
return epm->run_background_work_until_halt();
} else {
- return epm->background_process.trimmer->trim();
+ return epm->background_process.trimmer->trim(false);
}
}).handle_error(
crimson::ct_error::assert_all(
projects / ceph-ci.git / commitdiff