#include "rgw_process_env.h"
#include "common/async/yield_context.h"
#include "rgw_arn.h"
+#include "rgw_s3vector_background.h"
#define dout_context g_ceph_context
#define dout_subsys ceph_subsys_rgw
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if(op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
const rgw_bucket bucket_id(s->bucket_tenant, configuration.vector_bucket_name);
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
}
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
std::unique_ptr<rgw::sal::VectorBucket> bucket;
op_ret = driver->load_vector_bucket(this, bucket_id, &bucket, y);
if (op_ret < 0) {
+ if (op_ret == -ENOENT) {
+ rgw::s3vector::notify_session_delete(this, bucket_id.name);
+ }
ldpp_dout(this, 1) << "ERROR: failed to load s3vector bucket " << bucket_id << ". error: " << op_ret << dendl;
return;
}
return nullptr;
}
-
namespace rgw::s3vector {
+ struct LanceDBSessionConnHandle {
+ std::shared_ptr<const LanceDBSession> session_keepalive;
+ LanceDBConnection* conn = nullptr;
+
+ explicit operator bool() const {
+ return conn != nullptr;
+ }
+ };
+
+ struct LanceDBSessionTableHandle {
+ LanceDBSessionConnHandle conn_handle;
+ LanceDBTable* table = nullptr;
+
+ explicit operator bool() const {
+ return table != nullptr;
+ }
+ };
+
// convert LanceDBError to linux error codes
int lancedb_error_to_errno(LanceDBError err) {
switch (err) {
return conn;
}
+ LanceDBSessionConnHandle connect_with_session_handle(DoutPrefixProvider* dpp, const std::string& vector_bucket_name){
+ const auto dbname = fmt::format("/tmp/lancedb/{}", vector_bucket_name);
+ //get shared pointer to session for the bucket, if session doesn't exist, fallback to connect w/o session and trigger session creation for future connections
+ auto session_sp = rgw::s3vector::get_session(dpp, vector_bucket_name);
+ if (!session_sp) {
+ rgw::s3vector::notify_session_create(dpp, vector_bucket_name);
+ return LanceDBSessionConnHandle{
+ .conn = connect(dpp, vector_bucket_name)
+ };
+ }
+ LanceDBConnectBuilder* builder = lancedb_connect(dbname.c_str());
+ builder = lancedb_connect_builder_session(builder, session_sp.get());
+ LanceDBConnection* conn = lancedb_connect_builder_execute(builder);
+ if (!conn) {
+ ldpp_dout(dpp, 1) << "ERROR: s3vector failed to connect using session to: " << dbname << " falling back to connect without session" << dendl;
+ return LanceDBSessionConnHandle{
+ .conn = connect(dpp, vector_bucket_name)
+ }; // fallback to connect without session
+ }
+ return LanceDBSessionConnHandle{
+ .session_keepalive = std::move(session_sp),
+ .conn = conn
+ };
+ }
+
LanceDBTable* open_table(DoutPrefixProvider* dpp, const std::string& vector_bucket_name, const std::string& index_name) {
LanceDBConnection* conn = connect(dpp, vector_bucket_name);
if (!conn) {
return table;
}
+ LanceDBSessionTableHandle open_table_with_session_handle(DoutPrefixProvider* dpp, const std::string& vector_bucket_name, const std::string& index_name) {
+ auto conn_handle = connect_with_session_handle(dpp, vector_bucket_name);
+ if (!conn_handle) {
+ return {};
+ }
+ LanceDBTable* table = lancedb_connection_open_table(conn_handle.conn, index_name.c_str());
+ if (!table) {
+ ldpp_dout(dpp, 1) << "ERROR: s3vector failed to open index: " << index_name << " in: " << vector_bucket_name << dendl;
+ lancedb_connection_free(conn_handle.conn);
+ return {};
+ }
+ return LanceDBSessionTableHandle{
+ .conn_handle = std::move(conn_handle),
+ .table = table
+ };
+ }
+
// get creation time from the first version of a table
// returns 0 on failure
uint64_t get_table_creation_time(const LanceDBTable* table, DoutPrefixProvider* dpp) {
lancedb_connection_free(conn);
return -EIO;
}
+ ldpp_dout(dpp, 20) << "INFO: deleting in-memory session (if it exists) for bucket: " << configuration.vector_bucket_name << dendl;
+ rgw::s3vector::notify_session_delete(dpp, configuration.vector_bucket_name);
lancedb_connection_free(conn);
return 0;
}
int create_vector_bucket(const create_vector_bucket_t& configuration, DoutPrefixProvider* dpp, optional_yield y) {
log_configuration(dpp, "CreateVectorBucket", configuration);
- LanceDBConnection* conn = connect(dpp, configuration.vector_bucket_name);
- if (!conn) {
+ auto conn_handle = connect_with_session_handle(dpp, configuration.vector_bucket_name);
+ if (!conn_handle) {
return -EIO;
}
+ LanceDBConnection* conn = conn_handle.conn;
// verify connectivity by listing table names
char** table_names;
size_t name_count;
int put_vectors(const put_vectors_t& configuration, DoutPrefixProvider* dpp, optional_yield y) {
log_configuration(dpp, "PutVectors", configuration);
- LanceDBTable* table = open_table(dpp, configuration.vector_bucket_name, configuration.index_name);
- if (!table) {
+ auto table_handle = open_table_with_session_handle(dpp, configuration.vector_bucket_name, configuration.index_name);
+ if (!table_handle) {
return -EIO;
}
+ LanceDBTable* table = table_handle.table;
+ LanceDBConnection* conn = table_handle.conn_handle.conn;
if (configuration.vectors.empty()) {
ldpp_dout(dpp, 10) << "WARNING: s3vector no vectors provided" << dendl;
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return 0;
}
unsigned int dimension = 0;
if (int ret = get_vector_dimension(configuration.index_name, table, dpp, dimension); ret < 0) {
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return ret;
}
<< ". error: " << schema_error_message << dendl;
lancedb_free_string(schema_error_message);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return lancedb_error_to_errno(result);
}
<< ". error: " << schema.status().ToString() << dendl;
lancedb_free_arrow_schema(reinterpret_cast<FFI_ArrowSchema*>(c_schema_ptr));
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EINVAL;
}
if (num_rows == 0) {
ldpp_dout(dpp, 1) << "ERROR: s3vector no valid vectors to insert" << dendl;
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EINVAL;
}
ldpp_dout(dpp, 1) << "ERROR: s3vector failed to export record batch to C ABI: " << status.ToString() << dendl;
if (c_schema.release) c_schema.release(&c_schema);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EINVAL;
}
if (c_array.release) c_array.release(&c_array);
if (c_schema.release) c_schema.release(&c_schema);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EINVAL;
}
ldpp_dout(dpp, 1) << "ERROR: s3vector failed to upsert record batch to index. error: " << error_message << dendl;
lancedb_free_string(error_message);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return lancedb_error_to_errno(result);
}
- lancedb_table_free(table);
// we are not failing the operation if we cannot notify the background process on index update
notify_index_update(dpp, configuration.vector_bucket_name, configuration.index_name);
+ lancedb_table_free(table);
+ lancedb_connection_free(conn);
return 0;
}
int get_vectors(const get_vectors_t& configuration, DoutPrefixProvider* dpp, optional_yield y, get_vectors_reply_t& reply) {
log_configuration(dpp, "GetVectors", configuration);
- LanceDBTable* table = open_table(dpp, configuration.vector_bucket_name, configuration.index_name);
- if (!table) {
+ auto table_handle = open_table_with_session_handle(dpp, configuration.vector_bucket_name, configuration.index_name);
+ if (!table_handle) {
return -EIO;
}
+ LanceDBTable* table = table_handle.table;
+ LanceDBConnection* conn = table_handle.conn_handle.conn;
LanceDBQuery* query = lancedb_query_new(table);
if (!query) {
ldpp_dout(dpp, 1) << "ERROR: s3vector failed to create query for index: " << configuration.index_name << dendl;
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EIO;
}
lancedb_free_string(error_message);
lancedb_query_free(query);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return lancedb_error_to_errno(result);
}
lancedb_free_string(error_message);
lancedb_query_free(query);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return lancedb_error_to_errno(result);
}
ldpp_dout(dpp, 1) << "ERROR: s3vector failed to execute query on index: " << configuration.index_name << dendl;
lancedb_query_free(query);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EIO;
}
auto ret = populate_vectors_from_query(dpp, query_result, reply.vectors, configuration.index_name, configuration.return_data, false, false);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return ret;
}
int delete_vectors(const delete_vectors_t& configuration, DoutPrefixProvider* dpp, optional_yield y) {
log_configuration(dpp, "DeleteVectors", configuration);
- LanceDBTable* table = open_table(dpp, configuration.vector_bucket_name, configuration.index_name);
- if (!table) {
+ auto table_handle = open_table_with_session_handle(dpp, configuration.vector_bucket_name, configuration.index_name);
+ if (!table_handle) {
return -EIO;
}
+ LanceDBTable* table = table_handle.table;
+ LanceDBConnection* conn = table_handle.conn_handle.conn;
// build where filter for keys
std::ostringstream oss;
const auto keys_size = configuration.keys.size();
lancedb_free_string(error_message);
}
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return lancedb_error_to_errno(result);
}
int query_vectors(const query_vectors_t& configuration, DoutPrefixProvider* dpp, optional_yield y, query_vectors_reply_t& reply) {
log_configuration(dpp, "QueryVectors", configuration);
- LanceDBTable* table = open_table(dpp, configuration.vector_bucket_name, configuration.index_name);
- if (!table) {
+ auto table_handle = open_table_with_session_handle(dpp, configuration.vector_bucket_name, configuration.index_name);
+ if (!table_handle) {
return -EIO;
}
+ LanceDBTable* table = table_handle.table;
+ LanceDBConnection* conn = table_handle.conn_handle.conn;
unsigned int table_dimension;
if (int ret = get_vector_dimension(configuration.index_name, table, dpp, table_dimension); ret < 0) {
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return ret;
}
if (table_dimension != query_dimension) {
ldpp_dout(dpp, 1) << "ERROR: s3vector query vector dimension (" << query_dimension << ") does not match index: " <<
configuration.index_name << " vector dimension (" << table_dimension << ")" << dendl;
+ lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EINVAL;
}
if (!query) {
ldpp_dout(dpp, 1) << "ERROR: s3vector failed to create vector query for index: " << configuration.index_name << dendl;
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EIO;
}
lancedb_free_string(error_message);
lancedb_vector_query_free(query);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return lancedb_error_to_errno(result);
}
lancedb_free_string(error_message);
lancedb_vector_query_free(query);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return lancedb_error_to_errno(result);
}
lancedb_free_string(error_message);
lancedb_vector_query_free(query);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return lancedb_error_to_errno(result);
}
ldpp_dout(dpp, 1) << "ERROR: s3vector failed to execute query on index: " << configuration.index_name << dendl;
lancedb_vector_query_free(query);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return -EIO;
}
int ret = populate_vectors_from_query(dpp, query_result, reply.vectors, configuration.index_name, false, configuration.return_distance, true);
reply.distance_metric = get_table_distance_metric(table, dpp);
lancedb_table_free(table);
+ lancedb_connection_free(conn);
return ret;
}
}
-
#include <string>
#include <unordered_map>
#include "rgw_sal.h"
+#include "lancedb.h"
#define dout_subsys ceph_subsys_rgw
class Manager : public DoutPrefixProvider {
public:
+ //message_t -> pass in empty index name for session messages (can extend to per table sessions in the future if needed)
using table_name_t = std::pair<std::string, std::string>; // pair of vector bucket name and index name
struct message_t {
enum class Op {
UPDATE,
- REMOVE
+ REMOVE,
+ SESSION_CREATE,
+ SESSION_DELETE
};
message_t(const std::string& bucket_name, const std::string& index_name, Op _type) :
table_name(bucket_name, index_name), type(_type) {}
boost::asio::executor_work_guard<Executor> work_guard;
std::vector<std::thread> workers;
rgw::sal::Driver* const driver;
+ struct LanceDBSessionDeleter {
+ void operator()(LanceDBSession* session) const {
+ if(session) {
+ lancedb_session_free(session);
+ }
+ }
+ };
+ using SessionPtr = std::shared_ptr<LanceDBSession>;
+ ceph::shared_mutex sessions_mutex = ceph::make_shared_mutex("s3vector::Manager::sessions_mutex");
+ std::unordered_map<std::string, SessionPtr> sessions;
std::unordered_map<table_name_t, ceph::coarse_real_time, boost::hash<table_name_t>> tables;
MessageQueue messages;
static constexpr auto idle_sleep = std::chrono::milliseconds(1000); // 1s
return 0;
}
- // process all work items
- void process_tables(boost::asio::yield_context yield) {
- ldpp_dout(this, 5) << "INFO: start processing tables" << dendl;
+ // process all work items for tables and sessions
+ void process_messages(boost::asio::yield_context yield) {
+ ldpp_dout(this, 5) << "INFO: manager started. starting to process messages for background table and session operations" << dendl;
while (!shutdown) {
std::vector<table_name_t> tables_to_process;
const auto message_count = messages.consume_all([&tables_to_process, this](auto message) {
std::unique_ptr<message_t> 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;
- tables_to_process.push_back(table_name);
- return;
- }
- const auto now = ceph::coarse_real_clock::now();
- const auto time_since_last_process = now - it->second;
- if (time_since_last_process > std::chrono::milliseconds(5000)) {
- ldpp_dout(this, 20) << "INFO: will try to process table: " << table_name.first << "." << table_name.second <<
- ". " << time_since_last_process << " passed since last processing" << dendl;
- it->second = now;
- tables_to_process.push_back(table_name);
- } else {
- ldpp_dout(this, 20) << "INFO: will skip processing table: " << table_name.first << "." << table_name.second <<
- ". only " << time_since_last_process << " passed since last processing" << dendl;
+ switch(message->type) {
+ case 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;
+ case message_t::Op::UPDATE:
+ {
+ ldpp_dout(this, 20) << "INFO: received update message for table: " << table_name.first << "." << table_name.second << dendl;
+ 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;
+ tables_to_process.push_back(table_name);
+ return;
+ }
+ const auto now = ceph::coarse_real_clock::now();
+ const auto time_since_last_process = now - it->second;
+ if (time_since_last_process > std::chrono::milliseconds(5000)) {
+ ldpp_dout(this, 20) << "INFO: will try to process table: " << table_name.first << "." << table_name.second <<
+ ". " << time_since_last_process << " passed since last processing" << dendl;
+ it->second = now;
+ tables_to_process.push_back(table_name);
+ } else {
+ ldpp_dout(this, 20) << "INFO: will skip processing table: " << table_name.first << "." << table_name.second <<
+ ". only " << time_since_last_process << " passed since last processing" << dendl;
+ }
+ return;
+ }
+ case message_t::Op::SESSION_CREATE:
+ {
+ ldpp_dout(this, 20) << "INFO: received session create message for bucket: " << table_name.first << dendl;
+ std::unique_lock l(sessions_mutex);
+ if (sessions.find(table_name.first) == sessions.end()) {
+ //create session if not exist, otherwise just ignore
+ //Can define session options in the future if needed, for now just create with default options for cache sizes
+ LanceDBSession* session = lancedb_session_new(nullptr);
+ if (session) {
+ sessions[table_name.first] = SessionPtr(session, LanceDBSessionDeleter());
+ ldpp_dout(this, 20) << "INFO: created session for bucket: " << table_name.first << dendl;
+ }
+ else {
+ ldpp_dout(this, 1) << "ERROR: failed to create session for bucket: " << table_name.first << dendl;
+ }
+ return;
+ }
+ ldpp_dout(this, 20) << "INFO: session already exists for bucket: " << table_name.first << dendl;
+ return;
+ }
+ case message_t::Op::SESSION_DELETE:
+ {
+ ldpp_dout(this, 20) << "INFO: received session delete message for bucket: " << table_name.first << dendl;
+ std::unique_lock l(sessions_mutex);
+ if (sessions.erase(table_name.first) > 0) {
+ ldpp_dout(this, 20) << "INFO: deleted session for bucket: " << table_name.first << dendl;
+ } else {
+ ldpp_dout(this, 20) << "INFO: session doesn't exist for bucket: " << table_name.first << dendl;
+ }
+ return;
+ }
+ default:
+ ldpp_dout(this, 1) << "ERROR: received message with unknown type for bucket: " << table_name.first << " index: " << table_name.second << dendl;
+ return;
}
});
+
tokens_waiter tw(this);
for (const auto& table_name : tables_to_process) {
// start processing a table
// wait for all pending work to finish
tw.async_wait(yield);
}
+
if (message_count == 0) {
// if no messages, sleep for a while before checking again
- ldpp_dout(this, 20) << "INFO: no tables to process" << dendl;
+ ldpp_dout(this, 20) << "INFO: no messages to process" << dendl;
async_sleep(yield, idle_sleep);
}
}
- ldpp_dout(this, 5) << "INFO: manager stopped. done processing all tables" << dendl;
- }
-
+ ldpp_dout(this, 5) << "INFO: manager stopped. done processing all table and session operations" << dendl;
+ }
+
public:
~Manager() {
void init() {
boost::asio::spawn(make_strand(io_context), std::allocator_arg, make_stack_allocator(),
[this](boost::asio::yield_context yield) {
- process_tables(yield);
+ process_messages(yield);
}, [] (std::exception_ptr eptr) {
if (eptr) std::rethrow_exception(eptr);
});
return false;
}
+ bool notify_session(const DoutPrefixProvider* dpp, const std::string& bucket_name, message_t::Op op) {
+ if (shutdown) {
+ ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about session: manager is shutting down" << dendl;
+ return false;
+ }
+ auto message_guard = std::make_unique<message_t>(bucket_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 session" << dendl;
+ return true;
+ }
+ ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about session: queue is full" << dendl;
+ return false;
+ }
+
+ std::shared_ptr<const LanceDBSession> get_session(const std::string& bucket_name) {
+ std::shared_lock l(sessions_mutex);
+ auto it = sessions.find(bucket_name);
+ if (it == sessions.end()) {
+ return nullptr;
+ }
+ return it->second;
+ }
+
Manager(CephContext* _cct, rgw::sal::Driver* _driver) :
cct(_cct),
work_guard(boost::asio::make_work_guard(io_context)),
return s_manager->notify_index(dpp, bucket_name, index_name, Manager::message_t::Op::REMOVE);
}
+std::shared_ptr<const LanceDBSession> get_session(const DoutPrefixProvider* dpp, const std::string& bucket_name) {
+ if (!s_manager) {
+ ldpp_dout(dpp, 1) << "ERROR: failed to get LanceDB session for bucket: manager is not initialized" << dendl;
+ return nullptr;
+ }
+ return s_manager->get_session(bucket_name);
+}
+
+bool notify_session_create(const DoutPrefixProvider* dpp, const std::string& bucket_name) {
+ if (!s_manager) {
+ ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about session creation: manager is not initialized" << dendl;
+ return false;
+ }
+ return s_manager->notify_session(dpp, bucket_name, Manager::message_t::Op::SESSION_CREATE);
+}
+
+bool notify_session_delete(const DoutPrefixProvider* dpp, const std::string& bucket_name) {
+ if (!s_manager) {
+ ldpp_dout(dpp, 1) << "ERROR: failed to notify s3vectors manager about session deletion: manager is not initialized" << dendl;
+ return false;
+ }
+ return s_manager->notify_session(dpp, bucket_name, Manager::message_t::Op::SESSION_DELETE);
+}
+
} // namespace rgw::s3vector
#pragma once
+#include <memory>
#include <string>
namespace rgw::sal {
class Driver;
}
class DoutPrefixProvider;
+struct LanceDBSession;
namespace rgw::s3vector {
bool init(const DoutPrefixProvider* dpp, rgw::sal::Driver* driver);
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);
+ // get LanceDB session for a bucket, returns nullptr if session doesn't exist or manager is not initialized
+ std::shared_ptr<const LanceDBSession> get_session(const DoutPrefixProvider* dpp, const std::string& bucket_name);
+ // notify manager for session creation
+ bool notify_session_create(const DoutPrefixProvider* dpp, const std::string& bucket_name);
+ // notify manager for session deletion
+ bool notify_session_delete(const DoutPrefixProvider* dpp, const std::string& bucket_name);
}
projects / ceph.git / commitdiff