Skip to content

feat(kv-store): expose read-only lmdb transactions #25280

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
spalladino wants to merge 1 commit into from
Open

feat(kv-store): expose read-only lmdb transactions #25280

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions barretenberg/cpp/src/barretenberg/lmdblib/lmdb_store.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,15 @@ void LMDBStore::put(std::vector<PutData>& data)

void LMDBStore::get(KeysVector& keys, OptionalValuesVector& values, const std::string& name)
{
get(keys, values, get_database(name));
get(keys, values, get_database(name), create_shared_read_transaction());
}

void LMDBStore::get(KeysVector& keys,
OptionalValuesVector& values,
const std::string& name,
ReadTransaction::SharedPtr tx)
{
get(keys, values, get_database(name), std::move(tx));
}

void LMDBStore::put(KeyDupValuesVector& toWrite,
Expand All @@ -136,10 +144,12 @@ void LMDBStore::put(KeyDupValuesVector& toWrite,
}
}
}
void LMDBStore::get(KeysVector& keys, OptionalValuesVector& values, LMDBDatabase::SharedPtr db)
void LMDBStore::get(KeysVector& keys,
OptionalValuesVector& values,
LMDBDatabase::SharedPtr db,
ReadTransaction::SharedPtr tx)
{
values.reserve(keys.size());
ReadTransaction::SharedPtr tx = create_read_transaction();
if (!db->duplicate_keys_permitted()) {
const LMDBDatabase& dbRef = *db;
for (auto& k : keys) {
Expand Down
9 changes: 8 additions & 1 deletion barretenberg/cpp/src/barretenberg/lmdblib/lmdb_store.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,13 @@ class LMDBStore : public LMDBStoreBase {

void put(std::vector<PutData>& data);
void get(KeysVector& keys, OptionalValuesVector& values, const std::string& name);
/**
* @brief Reads the given keys against an already open read transaction, so the values observed belong to that
* transaction's snapshot rather than to whatever is committed at call time.
* @note LMDB read transactions are not thread safe. The caller must ensure no other operation runs against `tx`
* concurrently.
*/
void get(KeysVector& keys, OptionalValuesVector& values, const std::string& name, ReadTransaction::SharedPtr tx);
void has(const KeyOptionalValuesVector& entries, std::vector<bool>& results, const std::string& name);

Cursor::Ptr create_cursor(ReadTransaction::SharedPtr tx, const std::string& dbName);
Expand All @@ -63,7 +70,7 @@ class LMDBStore : public LMDBStoreBase {
KeyOptionalValuesVector& toDelete,
const LMDBDatabase& db,
LMDBWriteTransaction& tx);
void get(KeysVector& keys, OptionalValuesVector& values, LMDBDatabase::SharedPtr db);
void get(KeysVector& keys, OptionalValuesVector& values, LMDBDatabase::SharedPtr db, ReadTransaction::SharedPtr tx);
// Returns the database of the given name
Database::SharedPtr get_database(const std::string& name);
// Returns all databases
Expand Down
62 changes: 62 additions & 0 deletions barretenberg/cpp/src/barretenberg/lmdblib/lmdb_store.test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,6 +232,68 @@ TEST_F(LMDBStoreTest, can_read_from_database)
EXPECT_EQ(data[0].value(), ValuesVector{ expected });
}

TEST_F(LMDBStoreTest, reads_against_a_shared_read_transaction_see_a_stable_snapshot)
{
LMDBStore::Ptr store = create_store();
const std::string dbName = "Test Database";
store->open_database(dbName);

auto key = get_key(0);
auto original = get_value(0, 1);
auto updated = get_value(0, 2);

KeyOptionalValuesVector toDelete;
KeyDupValuesVector toWrite = { { { key, { original } } } };
std::vector<LMDBStore::PutData> putDatas = { { toWrite, toDelete, dbName } };
store->put(putDatas);

LMDBStore::ReadTransaction::SharedPtr tx = store->create_shared_read_transaction();

KeysVector keys = { { key } };
OptionalValuesVector snapshot;
store->get(keys, snapshot, dbName, tx);
EXPECT_EQ(snapshot[0].value(), ValuesVector{ original });

// overwrite the key in a new write transaction that commits while the read transaction is still open
toWrite = { { { key, { updated } } } };
putDatas = { { toWrite, toDelete, dbName } };
store->put(putDatas);

// the held transaction still sees the value as of the moment it was created
OptionalValuesVector afterWrite;
store->get(keys, afterWrite, dbName, tx);
EXPECT_EQ(afterWrite[0].value(), ValuesVector{ original });

// whereas a fresh read sees the new value
OptionalValuesVector latest;
store->get(keys, latest, dbName);
EXPECT_EQ(latest[0].value(), ValuesVector{ updated });
}

TEST_F(LMDBStoreTest, can_read_duplicates_against_a_shared_read_transaction)
{
LMDBStore::Ptr store = create_store();
const std::string dbName = "Test Database";
store->open_database(dbName, true);

int64_t numKeys = 5;
int64_t numValues = 3;
write_test_data({ dbName }, numKeys, numValues, *store);

LMDBStore::ReadTransaction::SharedPtr tx = store->create_shared_read_transaction();

KeysVector keys = { { get_key(2) } };
OptionalValuesVector values;
store->get(keys, values, dbName, tx);

ValuesVector expected;
for (int64_t i = 0; i < numValues; i++) {
expected.emplace_back(get_value(2, i));
}
ASSERT_TRUE(values[0].has_value());
EXPECT_EQ(values[0].value(), expected);
}

TEST_F(LMDBStoreTest, can_not_read_from_non_existent_database)
{
LMDBStore::Ptr store = create_store();
Expand Down
21 changes: 19 additions & 2 deletions barretenberg/cpp/src/barretenberg/nodejs_module/lmdb_store/lmdb_store_message.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@ enum LMDBStoreMessageType {

CLOSE,
COPY_STORE,

START_READ_TX,
CLOSE_READ_TX,
};

struct OpenDatabaseRequest {
Expand All @@ -39,7 +42,9 @@ struct OpenDatabaseRequest {
struct GetRequest {
lmdblib::KeysVector keys;
std::string db;
SERIALIZATION_FIELDS(keys, db);
// When set, read against the snapshot of the read transaction with this id instead of opening a fresh one
std::optional<uint64_t> txId;
SERIALIZATION_FIELDS(keys, db, txId);
};

struct GetResponse {
Expand Down Expand Up @@ -78,7 +83,9 @@ struct StartCursorRequest {
std::optional<uint32_t> count;
std::optional<bool> onePage;
std::string db;
SERIALIZATION_FIELDS(key, reverse, count, onePage, db);
// When set, iterate against the snapshot of the read transaction with this id instead of opening a fresh one
std::optional<uint64_t> txId;
SERIALIZATION_FIELDS(key, reverse, count, onePage, db, txId);
};

struct StartCursorResponse {
Expand Down Expand Up @@ -139,6 +146,16 @@ struct CopyStoreRequest {
SERIALIZATION_FIELDS(dstPath, compact);
};

struct StartReadTxResponse {
uint64_t tx;
SERIALIZATION_FIELDS(tx);
};

struct CloseReadTxRequest {
uint64_t tx;
SERIALIZATION_FIELDS(tx);
};

} // namespace bb::nodejs::lmdb_store

MSGPACK_ADD_ENUM(bb::nodejs::lmdb_store::LMDBStoreMessageType)
129 changes: 102 additions & 27 deletions barretenberg/cpp/src/barretenberg/nodejs_module/lmdb_store/lmdb_store_wrapper.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,11 @@
#include <chrono>
#include <cstdint>
#include <memory>
#include <mutex>
#include <optional>
#include <ratio>
#include <stdexcept>
#include <tuple>
#include <utility>

using namespace bb::nodejs;
Expand Down Expand Up @@ -74,6 +76,9 @@ LMDBStoreWrapper::LMDBStoreWrapper(const Napi::CallbackInfo& info)

_msg_processor.register_handler(LMDBStoreMessageType::OPEN_DATABASE, this, &LMDBStoreWrapper::open_database);

_msg_processor.register_handler(LMDBStoreMessageType::START_READ_TX, this, &LMDBStoreWrapper::start_read_tx);
_msg_processor.register_handler(LMDBStoreMessageType::CLOSE_READ_TX, this, &LMDBStoreWrapper::close_read_tx);

_msg_processor.register_handler(LMDBStoreMessageType::GET, this, &LMDBStoreWrapper::get);
_msg_processor.register_handler(LMDBStoreMessageType::HAS, this, &LMDBStoreWrapper::has);

Expand Down Expand Up @@ -116,19 +121,62 @@ void LMDBStoreWrapper::verify_store() const
throw std::runtime_error(format("LMDB store unavailable, was close already called?"));
}

// Returned by value: the copied shared_ptrs keep the transaction and its mutex alive even if the entry is erased by
// a concurrent CLOSE_READ_TX.
ReadTxData LMDBStoreWrapper::get_read_tx(uint64_t id)
{
std::lock_guard<std::mutex> lock(_read_tx_mutex);
auto it = _read_txs.find(id);
if (it == _read_txs.end()) {
throw std::runtime_error(format("Read transaction ", id, " not found, was it already closed?"));
}
return it->second;
}

BoolResponse LMDBStoreWrapper::open_database(const OpenDatabaseRequest& req)
{
verify_store();
_store->open_database(req.db, !req.uniqueKeys.value_or(true));
return { true };
}

StartReadTxResponse LMDBStoreWrapper::start_read_tx()
{
verify_store();
// This consumes one of the environment's reader slots until the matching CLOSE_READ_TX arrives, and pins the
// pages the snapshot references. The JS side caps how many of these can be open at once.
auto tx = _store->create_shared_read_transaction();
uint64_t id = _next_read_tx_id++;
{
std::lock_guard<std::mutex> lock(_read_tx_mutex);
_read_txs[id] = { tx, std::make_shared<std::mutex>() };
}
return { id };
}

BoolResponse LMDBStoreWrapper::close_read_tx(const CloseReadTxRequest& req)
{
{
std::lock_guard<std::mutex> lock(_read_tx_mutex);
// Cursors opened against this transaction hold their own reference, so the underlying transaction is only
// aborted once the last of them is closed too.
_read_txs.erase(req.tx);
}
return { true };
}

GetResponse LMDBStoreWrapper::get(const GetRequest& req)
{
verify_store();
lmdblib::OptionalValuesVector vals;
lmdblib::KeysVector keys = req.keys;
_store->get(keys, vals, req.db);
if (req.txId.has_value()) {
ReadTxData data = get_read_tx(req.txId.value());
std::lock_guard<std::mutex> tx_lock(*data.mtx);
_store->get(keys, vals, req.db, data.tx);
} else {
_store->get(keys, vals, req.db);
}
return { vals };
}

Expand All @@ -148,37 +196,56 @@ StartCursorResponse LMDBStoreWrapper::start_cursor(const StartCursorRequest& req
bool one_page = req.onePage.value_or(false);
lmdblib::Key key = req.key;

auto tx = _store->create_shared_read_transaction();
lmdblib::LMDBReadTransaction::SharedPtr tx;
std::shared_ptr<std::mutex> tx_mtx;
if (req.txId.has_value()) {
// Iterate over the snapshot the client already holds open, rather than over whatever is committed now
ReadTxData data = get_read_tx(req.txId.value());
tx = data.tx;
tx_mtx = data.mtx;
} else {
tx = _store->create_shared_read_transaction();
tx_mtx = std::make_shared<std::mutex>();
}

lmdblib::LMDBCursor::SharedPtr cursor = _store->create_cursor(tx, req.db);
bool start_ok = cursor->set_at_key(key);

if (!start_ok) {
// we couldn't find exactly the requested key. Find the next biggest one.
start_ok = cursor->set_at_key_gte(key);
// if we found a key that's greater _and_ we want to go in reverse order
// then we're actually outside the requested bounds, we need to go back one position
if (start_ok && reverse) {
lmdblib::KeyDupValuesVector entries;
// read_prev returns `true` if there's nothing more to read
// turn this into a "not ok" because there's nothing in the db for this cursor to read
start_ok = !cursor->read_prev(1, entries);
} else if (!start_ok && reverse) {
// we couldn't find a key greater than our starting point _and_ we want to go in reverse..
// then we start at the end of the database (the client requested to start at a key greater than anything in
// the DB)
start_ok = cursor->set_at_end();

bool done = false;
lmdblib::KeyDupValuesVector first_page;
{
// Never hold _cursor_mutex while taking a transaction mutex: advance_cursor takes them in this order too
std::lock_guard<std::mutex> tx_lock(*tx_mtx);
bool start_ok = cursor->set_at_key(key);

if (!start_ok) {
// we couldn't find exactly the requested key. Find the next biggest one.
start_ok = cursor->set_at_key_gte(key);
// if we found a key that's greater _and_ we want to go in reverse order
// then we're actually outside the requested bounds, we need to go back one position
if (start_ok && reverse) {
lmdblib::KeyDupValuesVector entries;
// read_prev returns `true` if there's nothing more to read
// turn this into a "not ok" because there's nothing in the db for this cursor to read
start_ok = !cursor->read_prev(1, entries);
} else if (!start_ok && reverse) {
// we couldn't find a key greater than our starting point _and_ we want to go in reverse..
// then we start at the end of the database (the client requested to start at a key greater than
// anything in the DB)
start_ok = cursor->set_at_end();
}

// in case we're iterating in ascending order and we can't find the exact key or one that's greater than it
// then that means theren's nothing in the DB for the cursor to read
}

// in case we're iterating in ascending order and we can't find the exact key or one that's greater than it
// then that means theren's nothing in the DB for the cursor to read
}
// we couldn't find a starting position
if (!start_ok) {
return { std::nullopt, {} };
}

// we couldn't find a starting position
if (!start_ok) {
return { std::nullopt, {} };
std::tie(done, first_page) = _advance_cursor(*cursor, reverse, page_size);
}

auto [done, first_page] = _advance_cursor(*cursor, reverse, page_size);
// cursor finished after reading a single page or client only wanted the first page
if (done || one_page) {
return { std::nullopt, first_page };
Expand All @@ -187,7 +254,7 @@ StartCursorResponse LMDBStoreWrapper::start_cursor(const StartCursorRequest& req
auto cursor_id = cursor->id();
{
std::lock_guard<std::mutex> lock(_cursor_mutex);
_cursors[cursor_id] = { cursor, reverse };
_cursors[cursor_id] = { cursor, reverse, tx_mtx };
}

return { cursor_id, first_page };
Expand All @@ -212,6 +279,7 @@ AdvanceCursorResponse LMDBStoreWrapper::advance_cursor(const AdvanceCursorReques
}

uint32_t page_size = req.count.value_or(DEFAULT_CURSOR_PAGE_SIZE);
std::lock_guard<std::mutex> tx_lock(*data.txMtx);
auto [done, entries] = _advance_cursor(*data.cursor, data.reverse, page_size);
return { entries, done };
}
Expand All @@ -225,6 +293,7 @@ AdvanceCursorCountResponse LMDBStoreWrapper::advance_cursor_count(const AdvanceC
data = _cursors.at(req.cursor);
}

std::lock_guard<std::mutex> tx_lock(*data.txMtx);
auto [done, count] = _advance_cursor_count(*data.cursor, data.reverse, req.endKey);
return { count, done };
}
Expand Down Expand Up @@ -267,6 +336,12 @@ BoolResponse LMDBStoreWrapper::close()
_cursors.clear();
}

{
// and all of the read transactions still held open on behalf of the JS side
std::lock_guard read_txs(_read_tx_mutex);
_read_txs.clear();
}

// and finally close the database handle
_store.reset(nullptr);

Expand Down
Loading
Loading