From 73f9c7293fb77552eddeac2a55cda2e4a3e48bfa Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Thu, 22 Jan 2026 02:00:31 +0300 Subject: [PATCH] Initialize blockstore after loading pool configuration to pre-shard the DB correctly on start --- src/blockstore/blockstore.h | 3 ++ src/blockstore/blockstore_impl.cpp | 11 +++--- src/blockstore/blockstore_impl.h | 3 +- src/blockstore/blockstore_init.cpp | 11 +----- src/blockstore/v1/impl.cpp | 5 +++ src/blockstore/v1/impl.h | 3 ++ src/client/etcd_state_client.cpp | 27 +++++++++++--- src/osd/osd.cpp | 56 ++++++++++++++++++++---------- src/osd/osd.h | 5 ++- src/osd/osd_cluster.cpp | 14 ++++++-- 10 files changed, 95 insertions(+), 43 deletions(-) diff --git a/src/blockstore/blockstore.h b/src/blockstore/blockstore.h index d4f9f7ef..7edde450 100644 --- a/src/blockstore/blockstore.h +++ b/src/blockstore/blockstore.h @@ -183,6 +183,9 @@ public: // Update configuration virtual void parse_config(blockstore_config_t & config) = 0; + // Reshard database for a pool + virtual void reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size) = 0; + // Event loop virtual void loop() = 0; diff --git a/src/blockstore/blockstore_impl.cpp b/src/blockstore/blockstore_impl.cpp index 9225e5ea..6064a703 100644 --- a/src/blockstore/blockstore_impl.cpp +++ b/src/blockstore/blockstore_impl.cpp @@ -23,6 +23,7 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i * dsk.open_meta(); dsk.open_journal(); dsk.calc_lengths(); + dsk.check_lengths(); } catch (std::exception & e) { @@ -31,16 +32,13 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i * } meta_superblock = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size); memset(meta_superblock, 0, dsk.meta_block_size); -} - -void blockstore_impl_t::init() -{ flusher = new journal_flusher_t(this); if (dsk.inmemory_journal) { buffer_area = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.journal_len); } heap = new blockstore_heap_t(&dsk, buffer_area, log_level); + ringloop->wakeup(); } blockstore_impl_t::~blockstore_impl_t() @@ -394,3 +392,8 @@ std::string blockstore_impl_t::get_op_diag(blockstore_op_t *op) snprintf(buf, sizeof(buf), "state=%d", priv->op_state); return std::string(buf); } + +void blockstore_impl_t::reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size) +{ + heap->reshard(pool, pg_count, pg_stripe_size); +} diff --git a/src/blockstore/blockstore_impl.h b/src/blockstore/blockstore_impl.h index c667e4dd..927c3b75 100644 --- a/src/blockstore/blockstore_impl.h +++ b/src/blockstore/blockstore_impl.h @@ -142,7 +142,6 @@ public: int metadata_buf_size; blockstore_init_meta* metadata_init_reader; - void init(); void check_wait(blockstore_op_t *op); void init_op(blockstore_op_t *op); @@ -190,6 +189,8 @@ public: void parse_config(blockstore_config_t & config); void parse_config(blockstore_config_t & config, bool init); + void reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size); + // Event loop void loop(); diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp index 54e20a13..eac1e1f2 100644 --- a/src/blockstore/blockstore_init.cpp +++ b/src/blockstore/blockstore_init.cpp @@ -72,7 +72,6 @@ resume_1: } if (is_zero((uint64_t*)bs->meta_superblock, bs->dsk.meta_block_size)) { - bs->dsk.check_lengths(); { blockstore_meta_header_v3_t *hdr = (blockstore_meta_header_v3_t *)bs->meta_superblock; hdr->zero = 0; @@ -141,7 +140,7 @@ resume_1: hdr->bitmap_granularity != bs->dsk.bitmap_granularity || hdr->data_csum_type != bs->dsk.data_csum_type || hdr->csum_block_size != bs->dsk.csum_block_size || - hdr->meta_area_size > bs->dsk.meta_area_size) + hdr->meta_area_size != bs->dsk.meta_area_size) { printf( "Configuration stored in metadata superblock" @@ -154,15 +153,7 @@ resume_1: ); exit(1); } - bs->dsk.meta_area_size = hdr->meta_area_size; - if (bs->dsk.meta_format != hdr->version) - { - bs->dsk.meta_format = hdr->version; - bs->dsk.calc_lengths(); - } - bs->dsk.check_lengths(); } - bs->init(); bs->heap->start_load(((blockstore_meta_header_v3_t *)bs->meta_superblock)->completed_lsn); if (bs->dsk.inmemory_journal) { diff --git a/src/blockstore/v1/impl.cpp b/src/blockstore/v1/impl.cpp index 0b1b33a7..42f819ef 100644 --- a/src/blockstore/v1/impl.cpp +++ b/src/blockstore/v1/impl.cpp @@ -807,4 +807,9 @@ std::string blockstore_impl_t::get_op_diag(blockstore_op_t *op) return std::string(buf); } +void blockstore_impl_t::reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size) +{ + reshard_clean_db(pool, pg_count, pg_stripe_size); +} + } // namespace v1 diff --git a/src/blockstore/v1/impl.h b/src/blockstore/v1/impl.h index 5ef97f0b..d342c65c 100644 --- a/src/blockstore/v1/impl.h +++ b/src/blockstore/v1/impl.h @@ -288,6 +288,9 @@ public: void parse_config(blockstore_config_t & config); void parse_config(blockstore_config_t & config, bool init); + // Reshard database for a pool + void reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size); + // Event loop void loop(); diff --git a/src/client/etcd_state_client.cpp b/src/client/etcd_state_client.cpp index fa553757..b52b6b9d 100644 --- a/src/client/etcd_state_client.cpp +++ b/src/client/etcd_state_client.cpp @@ -568,9 +568,19 @@ void etcd_state_client_t::start_ws_keepalive() void etcd_state_client_t::load_global_config() { - etcd_call("/kv/range", json11::Json::object { - { "key", base64_encode(etcd_prefix+"/config/global") } - }, etcd_quick_timeout, max_etcd_attempts, 0, [this](std::string err, json11::Json data) + json11::Json::object req = { { "success", json11::Json::array { + json11::Json::object { + { "request_range", json11::Json::object { + { "key", base64_encode(etcd_prefix+"/config/global") }, + } } + }, + json11::Json::object { + { "request_range", json11::Json::object { + { "key", base64_encode(etcd_prefix+"/config/pools") }, + } } + }, + } } }; + etcd_txn(req, etcd_quick_timeout, max_etcd_attempts, 0, [this](std::string err, json11::Json data) { if (err != "") { @@ -588,10 +598,12 @@ void etcd_state_client_t::load_global_config() } return; } + json11::Json config_kv = data["responses"][0]["response_range"]["kvs"][0]; + json11::Json pools_kv = data["responses"][1]["response_range"]["kvs"][0]; json11::Json::object global_config; - if (data["kvs"].array_items().size() > 0) + if (!config_kv.is_null()) { - auto kv = parse_etcd_kv(data["kvs"][0]); + auto kv = parse_etcd_kv(config_kv); if (kv.value.is_object()) { global_config = kv.value.object_items(); @@ -608,6 +620,11 @@ void etcd_state_client_t::load_global_config() global_bitmap_granularity = DEFAULT_BITMAP_GRANULARITY; } global_immediate_commit = parse_immediate_commit(global_config["immediate_commit"].string_value(), IMMEDIATE_ALL); + if (!pools_kv.is_null()) + { + auto kv = parse_etcd_kv(pools_kv); + parse_state(kv); + } on_load_config_hook(global_config); }); } diff --git a/src/osd/osd.cpp b/src/osd/osd.cpp index ffd6f6f2..650ce7e3 100644 --- a/src/osd/osd.cpp +++ b/src/osd/osd.cpp @@ -31,25 +31,6 @@ osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop) // FIXME: Use timerfd_interval based directly on io_uring this->tfd = epmgr->tfd; - if (!json_is_true(this->config["disable_blockstore"])) - { - auto bs_cfg = json_to_string_map(this->config); - this->bs = blockstore_i::create(bs_cfg, ringloop, tfd); - // Wait for blockstore initialisation before actually starting OSD logic - // to prevent peering timeouts during restart with filled databases - while (!bs->is_started()) - { - ringloop->loop(); - if (bs->is_started()) - break; - ringloop->wait(); - } - // Autosync based on the number of unstable writes to prevent stalls due to insufficient journal space - uint64_t max_autosync = bs->get_journal_size() / bs->get_block_size() / 2; - if (autosync_writes > max_autosync) - autosync_writes = max_autosync; - } - if (json_is_true(this->config["osd_memlock"])) { // Lock all OSD memory if requested @@ -117,6 +98,7 @@ osd_t::~osd_t() autosync_timer_id = -1; } ringloop->unregister_consumer(&consumer); + ringloop->unregister_consumer(&init_consumer); delete epmgr; if (bs) delete bs; @@ -131,6 +113,42 @@ osd_t::~osd_t() free(zero_buffer); } +void osd_t::init_blockstore(std::function on_init) +{ + if (!json_is_true(this->config["disable_blockstore"])) + { + auto bs_cfg = json_to_string_map(this->config); + this->bs = blockstore_i::create(bs_cfg, ringloop, tfd); + // Pre-configure pool PG shards + for (auto & pool_item: st_cli.pool_config) + { + bs->reshard(pool_item.first, pool_item.second.pg_count, pool_item.second.pg_stripe_size); + } + // Autosync based on the number of unstable writes to prevent stalls due to insufficient journal space + uint64_t max_autosync = bs->get_journal_size() / bs->get_block_size() / 2; + if (autosync_writes > max_autosync) + autosync_writes = max_autosync; + if (on_init) + { + init_consumer.loop = [this, on_init]() + { + // Wait for blockstore initialisation before actually starting OSD logic + // to prevent peering timeouts during restart with filled databases + if (bs->is_started()) + { + ringloop->set_immediate([this, on_init] { init_consumer.loop = NULL; on_init(); }); + ringloop->unregister_consumer(&init_consumer); + } + }; + ringloop->register_consumer(&init_consumer); + } + } + else if (on_init) + { + on_init(); + } +} + void osd_t::parse_config(bool init) { config = msgr.merge_configs(cli_config, file_config, etcd_global_config, etcd_osd_config); diff --git a/src/osd/osd.h b/src/osd/osd.h index f314643a..cefc40db 100644 --- a/src/osd/osd.h +++ b/src/osd/osd.h @@ -158,6 +158,7 @@ class osd_t json11::Json self_state; bool loading_peer_config = false; std::set pg_state_dirty; + bool etcd_global_config_loaded = false; bool pg_config_applied = false; bool etcd_reporting_pg_state = false; bool etcd_reporting_stats = false; @@ -206,7 +207,7 @@ class osd_t void *zero_buffer = NULL; uint64_t zero_buffer_size = 0; uint32_t bs_block_size, bs_bitmap_granularity, clean_entry_bitmap_size; - ring_loop_t *ringloop; + ring_loop_t *ringloop = NULL; timerfd_manager_t *tfd = NULL; epoll_manager_t *epmgr = NULL; @@ -218,6 +219,7 @@ class osd_t int rdmacm_port = 0; #endif ring_consumer_t consumer; + ring_consumer_t init_consumer; // op statistics osd_op_stats_t prev_stats, prev_report_stats; @@ -241,6 +243,7 @@ class osd_t // cluster connection void parse_config(bool init); + void init_blockstore(std::function on_init); void init_cluster(); void on_change_osd_state_hook(osd_num_t peer_osd); void on_change_backfillfull_hook(pool_id_t pool_id); diff --git a/src/osd/osd_cluster.cpp b/src/osd/osd_cluster.cpp index 23b8f661..768de27f 100644 --- a/src/osd/osd_cluster.cpp +++ b/src/osd/osd_cluster.cpp @@ -18,6 +18,7 @@ void osd_t::init_cluster() { if (!st_cli.address_count()) { + init_blockstore(NULL); if (run_primary) { // Test version of clustering code with 1 pool, 1 PG and 2 peers @@ -422,7 +423,10 @@ void osd_t::on_change_osd_state_hook(osd_num_t peer_osd) void osd_t::on_change_pool_config_hook() { - apply_pg_locks_localize_only(); + if (etcd_global_config_loaded) + { + apply_pg_locks_localize_only(); + } } void osd_t::apply_pg_locks_localize_only() @@ -484,9 +488,13 @@ void osd_t::on_load_config_hook(json11::Json::object & global_config) { etcd_global_config = global_config; parse_config(true); - bind_socket(); - acquire_lease(); st_cli.on_load_config_hook = [this](json11::Json::object & cfg) { on_reload_config_hook(cfg); }; + etcd_global_config_loaded = true; + init_blockstore([this]() + { + bind_socket(); + acquire_lease(); + }); } void osd_t::on_reload_config_hook(json11::Json::object & global_config)