From: Yuval Lifshitz Date: Thu, 23 Apr 2026 17:09:18 +0000 (+0000) Subject: rgw/s3vectors: support index removal from background X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=3d536e58e4ae14b655cce96bb1bd02a1f6fe68af;p=ceph.git rgw/s3vectors: support index removal from background also. fix token waited issue Signed-off-by: Yuval Lifshitz --- diff --git a/src/rgw/rgw_s3vector.cc b/src/rgw/rgw_s3vector.cc index 0985115db3a..2d67c748b20 100644 --- a/src/rgw/rgw_s3vector.cc +++ b/src/rgw/rgw_s3vector.cc @@ -455,6 +455,8 @@ namespace rgw::s3vector { lancedb_connection_free(conn); return lancedb_error_to_errno(result); } + // we are not failing the operation if we cannot notify the background process on index removal + notify_index_remove(dpp, configuration.vector_bucket_name, configuration.index_name); return 0; } diff --git a/src/rgw/rgw_s3vector_background.cc b/src/rgw/rgw_s3vector_background.cc index 95e228504e2..349a77919e3 100644 --- a/src/rgw/rgw_s3vector_background.cc +++ b/src/rgw/rgw_s3vector_background.cc @@ -24,13 +24,25 @@ namespace rgw::s3vector { class Manager : public DoutPrefixProvider { +public: + using table_name_t = std::pair; // pair of vector bucket name and index name + struct message_t { + enum class Op { + UPDATE, + REMOVE + }; + message_t(const std::string& bucket_name, const std::string& index_name, Op _type) : + table_name(bucket_name, index_name), type(_type) {} + const table_name_t table_name; + const Op type; + }; +private: // use mmap/mprotect to allocate 128k coroutine stacks auto make_stack_allocator() { return boost::context::protected_fixedsize_stack{128*1024}; } - using table_name_t = std::pair; // pair of vector bucket name and index name - using MessageQueue = boost::lockfree::queue>; + using MessageQueue = boost::lockfree::queue>; using Executor = boost::asio::io_context::executor_type; bool shutdown = false; CephContext* const cct; @@ -121,8 +133,13 @@ class Manager : public DoutPrefixProvider { while (!shutdown) { std::vector tables_to_process; const auto message_count = messages.consume_all([&tables_to_process, this](auto message) { - std::unique_ptr message_guard(message); - const auto table_name = std::move(*message); + std::unique_ptr message_guard(message); + const auto table_name = std::move(message->table_name); + if (message->type == message_t::Op::REMOVE) { + ldpp_dout(this, 20) << "INFO: received remove message for table: " << table_name.first << "." << table_name.second << dendl; + tables.erase(table_name); + return; + } auto [it, inserted] = tables.emplace(table_name, ceph::coarse_real_clock::now()); if (inserted) { ldpp_dout(this, 20) << "INFO: will try to process new table: " << table_name.first << "." << table_name.second << dendl; @@ -146,7 +163,7 @@ class Manager : public DoutPrefixProvider { // start processing a table tokens_waiter::token token(&tw); boost::asio::spawn(make_strand(io_context), std::allocator_arg, make_stack_allocator(), - [this, table_name](boost::asio::yield_context yield) { + [this, token = std::move(token), table_name](boost::asio::yield_context yield) { const int rc = process_table(table_name, yield); if (rc < 0) { ldpp_dout(this, 1) << "ERROR: failed to process table: " << table_name.first << "." << table_name.second << " with error code: " << rc << dendl; @@ -218,18 +235,18 @@ public: ldpp_dout(this, 10) << "INfO: started manager" << dendl; } - bool notify_index_update(const DoutPrefixProvider* dpp, const std::string& bucket_name, const std::string& index_name) { + bool notify_index(const DoutPrefixProvider* dpp, const std::string& bucket_name, const std::string& index_name, message_t::Op op) { if (shutdown) { - ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about index update: manager is shutting down" << dendl; + ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about index: manager is shutting down" << dendl; return false; } - auto message_guard = std::make_unique(bucket_name, index_name); + auto message_guard = std::make_unique(bucket_name, index_name, op); if (messages.push(message_guard.get())) { std::ignore = message_guard.release(); // ownership transferred to the queue - ldpp_dout(dpp, 20) << "INFO: notified s3vectors manager about index update" << dendl; + ldpp_dout(dpp, 20) << "INFO: notified s3vectors manager about index" << dendl; return true; } - ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about index update: queue is full" << dendl; + ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about index: queue is full" << dendl; return false; } @@ -272,7 +289,15 @@ bool notify_index_update(const DoutPrefixProvider* dpp, const std::string& bucke ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about table update: manager is not initialized" << dendl; return false; } - return s_manager->notify_index_update(dpp, bucket_name, index_name); + return s_manager->notify_index(dpp, bucket_name, index_name, Manager::message_t::Op::UPDATE); +} + +bool notify_index_remove(const DoutPrefixProvider* dpp, const std::string& bucket_name, const std::string& index_name) { + if (!s_manager) { + ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about table remove: manager is not initialized" << dendl; + return false; + } + return s_manager->notify_index(dpp, bucket_name, index_name, Manager::message_t::Op::REMOVE); } } // namespace rgw::s3vector diff --git a/src/rgw/rgw_s3vector_background.h b/src/rgw/rgw_s3vector_background.h index 4758b4ec15f..b6c37c3f7f0 100644 --- a/src/rgw/rgw_s3vector_background.h +++ b/src/rgw/rgw_s3vector_background.h @@ -17,5 +17,7 @@ namespace rgw::s3vector { void resume(const DoutPrefixProvider* dpp, rgw::sal::Driver* driver); // update whenever new vectors are added to an index bool notify_index_update(const DoutPrefixProvider* dpp, const std::string& bucket_name, const std::string& index_name); + // update whenever a index is removed + bool notify_index_remove(const DoutPrefixProvider* dpp, const std::string& bucket_name, const std::string& index_name); }