diff --git a/Makefile b/Makefile index 3a864de4cc3..aa529e8e62a 100644 --- a/Makefile +++ b/Makefile @@ -59,7 +59,7 @@ endef # Fast bootstrap. # wsdb belongs to foundation until disentangled. fast-foundation: barretenberg bb-tests \ - wsdb wsdb-tests lmdblib-tests \ + wsdb wsdb-tests lmdblib-tests kvdb-tests \ l1-contracts l1-contracts-tests \ mock-protocol-circuits \ noir-protocol-circuits noir-protocol-circuits-tests \ @@ -395,7 +395,7 @@ ipc-runtime-cross: ipc-runtime ipc-runtime-cross-arm64-linux ipc-runtime-cross-a # lmdblib and kvdb are barretenberg-free: they build against their own deps # (lmdb, msgpack-c, node-addon-api) only, never bb. -.PHONY: lmdblib kvdb lmdblib-tests wsdb-tests +.PHONY: lmdblib kvdb lmdblib-tests kvdb-tests wsdb-tests lmdblib: $(call build,$@,native-packages/lmdblib) @@ -405,12 +405,16 @@ kvdb: lmdblib wsdb: lmdblib $(call build,$@,native-packages/wsdb) -# Native-package C++ tests (self-contained gtest binaries). kvdb has no C++ tests -# (its NAPI is exercised by yarn-project's kv-store tests). wsdb_tests use no bb -# headers; the bb-header parity/equivalence target is manual (WSDB_BUILD_BB_TESTS). +# Native-package tests. lmdblib and wsdb are self-contained gtest binaries. kvdb has no +# C++ tests: its NAPI protocol is driven from node (ts/test) and exercised further by +# yarn-project's kv-store tests. wsdb_tests use no bb headers; the bb-header +# parity/equivalence target is manual (WSDB_BUILD_BB_TESTS). lmdblib-tests: lmdblib $(call test,$@,native-packages/lmdblib) +kvdb-tests: kvdb + $(call test,$@,native-packages/kvdb) + wsdb-tests: wsdb $(call test,$@,native-packages/wsdb) diff --git a/native-packages/kvdb/.rebuild_patterns b/native-packages/kvdb/.rebuild_patterns index 9cb9042b35b..44afd463a93 100644 --- a/native-packages/kvdb/.rebuild_patterns +++ b/native-packages/kvdb/.rebuild_patterns @@ -3,4 +3,5 @@ ^native-packages/kvdb/\.rebuild_patterns$ ^native-packages/kvdb/cpp/ ^native-packages/kvdb/ts/src/ +^native-packages/kvdb/ts/test/ ^native-packages/lmdblib/ diff --git a/native-packages/kvdb/bootstrap.sh b/native-packages/kvdb/bootstrap.sh index 894e2fb4481..692fc06bf60 100755 --- a/native-packages/kvdb/bootstrap.sh +++ b/native-packages/kvdb/bootstrap.sh @@ -32,6 +32,18 @@ function build { yarn build } +# Emit test commands for the CI test engine: the addon's message protocol, driven over msgpack from node. +function test_cmds { + echo "$hash:TIMEOUT=120s cd native-packages/kvdb/ts && node --test test/*.test.mjs" +} + +# Manual: build then run the tests directly. +function test { + echo_header "kvdb test" + build + test_cmds | filter_test_cmds | parallelize +} + function clean { rm -rf ts/dest ts/build ts/packages ts/node_modules node_modules cpp/build } @@ -47,7 +59,7 @@ function release { (cd ts && retry "deploy_npm ${REF_NAME#v}") } -export -f build_native build clean release +export -f build_native build test_cmds test clean release case "$cmd" in "") diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp index 497b78669a5..03bd02fb0d2 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_message.hpp @@ -28,6 +28,9 @@ enum LMDBStoreMessageType { CLOSE, COPY_STORE, + + START_READ_TX, + CLOSE_READ_TX, }; struct OpenDatabaseRequest { @@ -39,7 +42,9 @@ struct OpenDatabaseRequest { struct GetRequest { lmdblib::KeysVector keys; std::string db; - MSGPACK_DEFINE_MAP(keys, db); + // When set, read through the snapshot of this open read transaction instead of a fresh one + std::optional txId; + MSGPACK_DEFINE_MAP(keys, db, txId); }; struct GetResponse { @@ -51,7 +56,9 @@ struct HasRequest { // std::map> entries; lmdblib::KeyOptionalValuesVector entries; std::string db; - MSGPACK_DEFINE_MAP(entries, db); + // When set, check against the snapshot of this open read transaction instead of a fresh one + std::optional txId; + MSGPACK_DEFINE_MAP(entries, db, txId); }; struct HasResponse { @@ -78,7 +85,9 @@ struct StartCursorRequest { std::optional count; std::optional onePage; std::string db; - MSGPACK_DEFINE_MAP(key, reverse, count, onePage, db); + // When set, iterate over the snapshot of this open read transaction instead of a fresh one + std::optional txId; + MSGPACK_DEFINE_MAP(key, reverse, count, onePage, db, txId); }; struct StartCursorResponse { @@ -139,6 +148,17 @@ struct CopyStoreRequest { MSGPACK_DEFINE_MAP(dstPath, compact); }; +struct StartReadTxResponse { + uint64_t txId; + MSGPACK_DEFINE_MAP(txId); +}; + +struct CloseReadTxRequest { + // Optional only so that a request without it is rejected rather than read as an indeterminate id + std::optional txId; + MSGPACK_DEFINE_MAP(txId); +}; + } // namespace azteclabs::kvdb::lmdb_store MSGPACK_ADD_ENUM(azteclabs::kvdb::lmdb_store::LMDBStoreMessageType) diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp index 7436e37d4a7..d627990ad5a 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.cpp @@ -6,9 +6,11 @@ #include #include #include +#include #include #include #include +#include #include using namespace azteclabs::kvdb; @@ -17,6 +19,9 @@ using namespace azteclabs::kvdb::lmdb_store; const uint64_t DEFAULT_MAP_SIZE = 1024UL * 1024; const uint64_t DEFAULT_MAX_READERS = 16; const uint64_t DEFAULT_CURSOR_PAGE_SIZE = 10; +// Reader permits a new read transaction must leave free, so that requests which open their own short-lived read +// transaction can always make progress while long-lived ones are held open +const uint32_t READ_TX_RESERVED_READERS = 1; LMDBStoreWrapper::LMDBStoreWrapper(const Napi::CallbackInfo& info) : ObjectWrap(info) @@ -91,6 +96,9 @@ LMDBStoreWrapper::LMDBStoreWrapper(const Napi::CallbackInfo& info) _msg_processor.register_handler(LMDBStoreMessageType::CLOSE, this, &LMDBStoreWrapper::close, true); _msg_processor.register_handler(LMDBStoreMessageType::COPY_STORE, this, &LMDBStoreWrapper::copy_store, true); + + _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); } Napi::Value LMDBStoreWrapper::call(const Napi::CallbackInfo& info) @@ -116,6 +124,17 @@ void LMDBStoreWrapper::verify_store() const throw std::runtime_error(azteclabs::lmdblib::format("LMDB store unavailable, was close already called?")); } +ReadTxData LMDBStoreWrapper::get_read_tx(uint64_t id) +{ + std::lock_guard lock(_read_tx_mutex); + auto it = _read_txs.find(id); + if (it == _read_txs.end()) { + throw std::runtime_error( + azteclabs::lmdblib::format("Read transaction ", id, " not found, was it already closed?")); + } + return it->second; +} + BoolResponse LMDBStoreWrapper::open_database(const OpenDatabaseRequest& req) { verify_store(); @@ -123,12 +142,55 @@ BoolResponse LMDBStoreWrapper::open_database(const OpenDatabaseRequest& req) return { true }; } +StartReadTxResponse LMDBStoreWrapper::start_read_tx() +{ + verify_store(); + // Waiting for a permit here could park every libuv worker, starving the CLOSE_READ_TX that would free one + auto tx = _store->try_create_shared_read_transaction(READ_TX_RESERVED_READERS); + if (!tx) { + throw std::runtime_error("No reader slot available to start a read transaction, close an open read " + "transaction or cursor and retry"); + } + ReadTxData data{ std::move(tx), std::make_shared() }; + uint64_t id = data.tx->id(); + { + std::lock_guard lock(_read_tx_mutex); + _read_txs.emplace(id, std::move(data)); + } + return { id }; +} + +BoolResponse LMDBStoreWrapper::close_read_tx(const CloseReadTxRequest& req) +{ + if (!req.txId.has_value()) { + throw std::runtime_error("CLOSE_READ_TX request is missing the read transaction id"); + } + // Moved out so that, if this was the last reference, the transaction is aborted after the registry lock is released + ReadTxData closed; + { + std::lock_guard lock(_read_tx_mutex); + auto it = _read_txs.find(req.txId.value()); + if (it == _read_txs.end()) { + return { false }; + } + closed = std::move(it->second); + _read_txs.erase(it); + } + 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 tx_lock(*data.mtx); + _store->get(keys, vals, req.db, data.tx); + } else { + _store->get(keys, vals, req.db); + } return { vals }; } @@ -136,7 +198,13 @@ HasResponse LMDBStoreWrapper::has(const HasRequest& req) { verify_store(); std::vector exists; - _store->has(req.entries, exists, req.db); + if (req.txId.has_value()) { + ReadTxData data = get_read_tx(req.txId.value()); + std::lock_guard tx_lock(*data.mtx); + _store->has(req.entries, exists, req.db, data.tx); + } else { + _store->has(req.entries, exists, req.db); + } return { exists }; } @@ -148,29 +216,29 @@ 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::LMDBCursor::SharedPtr cursor = _store->create_cursor(tx, req.db); - bool start_ok = cursor->set_at_key(key); + lmdblib::LMDBReadTransaction::SharedPtr tx; + std::shared_ptr tx_mtx; + if (req.txId.has_value()) { + ReadTxData data = get_read_tx(req.txId.value()); + tx = std::move(data.tx); + tx_mtx = std::move(data.mtx); + } else { + tx = _store->create_shared_read_transaction(); + tx_mtx = std::make_shared(); + } - 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(); + lmdblib::LMDBCursor::SharedPtr cursor; + bool start_ok = false; + bool done = false; + lmdblib::KeyDupValuesVector first_page; + { + // Opening a cursor can update the transaction's database state, so it needs the lock as much as reads do + std::lock_guard tx_lock(*tx_mtx); + cursor = _store->create_cursor(tx, req.db); + start_ok = _set_cursor_start(*cursor, key, reverse); + if (start_ok) { + std::tie(done, first_page) = _advance_cursor(*cursor, reverse, page_size); } - - // 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 @@ -178,7 +246,6 @@ StartCursorResponse LMDBStoreWrapper::start_cursor(const StartCursorRequest& req return { std::nullopt, {} }; } - 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 }; @@ -187,12 +254,40 @@ StartCursorResponse LMDBStoreWrapper::start_cursor(const StartCursorRequest& req auto cursor_id = cursor->id(); { std::lock_guard lock(_cursor_mutex); - _cursors[cursor_id] = { cursor, reverse }; + _cursors[cursor_id] = { cursor, reverse, tx_mtx }; } return { cursor_id, first_page }; } +bool LMDBStoreWrapper::_set_cursor_start(const lmdblib::LMDBCursor& cursor, lmdblib::Key& key, bool reverse) +{ + 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 + } + + return start_ok; +} + BoolResponse LMDBStoreWrapper::close_cursor(const CloseCursorRequest& req) { { @@ -212,6 +307,7 @@ AdvanceCursorResponse LMDBStoreWrapper::advance_cursor(const AdvanceCursorReques } uint32_t page_size = req.count.value_or(DEFAULT_CURSOR_PAGE_SIZE); + std::lock_guard tx_lock(*data.txMtx); auto [done, entries] = _advance_cursor(*data.cursor, data.reverse, page_size); return { entries, done }; } @@ -225,6 +321,7 @@ AdvanceCursorCountResponse LMDBStoreWrapper::advance_cursor_count(const AdvanceC data = _cursors.at(req.cursor); } + std::lock_guard tx_lock(*data.txMtx); auto [done, count] = _advance_cursor_count(*data.cursor, data.reverse, req.endKey); return { count, done }; } @@ -267,6 +364,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); diff --git a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp index 33a5eea4139..024531bdcc2 100644 --- a/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp +++ b/native-packages/kvdb/cpp/src/lmdb_store/lmdb_store_wrapper.hpp @@ -1,22 +1,36 @@ #pragma once +#include "lmdb_store/lmdb_store_message.hpp" #include "lmdblib/lmdb_cursor.hpp" #include "lmdblib/lmdb_store.hpp" #include "lmdblib/types.hpp" #include "messaging/dispatcher.hpp" #include "messaging/header.hpp" -#include "lmdb_store/lmdb_store_message.hpp" #include "util/message_processor.hpp" #include #include +#include #include #include +#include namespace azteclabs::kvdb::lmdb_store { +/** + * @brief A read transaction held open on behalf of the JavaScript side, together with the mutex that serializes + * access to it. The environment is opened with `MDB_NOTLS`, so a read transaction may move between the libuv worker + * threads, but it must never be used by two of them at once. + */ +struct ReadTxData { + lmdblib::LMDBReadTransaction::SharedPtr tx; + std::shared_ptr mtx; +}; + struct CursorData { lmdblib::LMDBCursor::SharedPtr cursor; bool reverse; + // Guards the cursor's read transaction, which may be shared with gets and other cursors + std::shared_ptr txMtx; }; /** * @brief Manages the interaction between the JavaScript runtime and the LMDB instance. @@ -38,12 +52,38 @@ class LMDBStoreWrapper : public Napi::ObjectWrap { std::mutex _cursor_mutex; std::unordered_map _cursors; + std::mutex _read_tx_mutex; + std::unordered_map _read_txs; + azteclabs::kvdb::AsyncMessageProcessor _msg_processor; void verify_store() const; + /** + * @brief Returns a copy of the registered read transaction, which keeps it alive even if it is closed concurrently. + * @throws std::runtime_error if no read transaction with this id is open + */ + ReadTxData get_read_tx(uint64_t id); + BoolResponse open_database(const OpenDatabaseRequest& req); + /** + * @brief Opens a read transaction that stays open until CLOSE_READ_TX (or CLOSE) and returns its id. It holds one + * of the environment's reader slots for its whole lifetime. + * @note The free slot is only checked here: a START_CURSOR without txId still waits for a slot and holds it until + * the cursor closes. Callers must keep open read transactions plus open cursors below the reader count, or plain + * reads can wait on every libuv worker at once, leaving no worker to run the close that would free a slot. + * @throws std::runtime_error without waiting if taking a slot would leave none free for other reads + */ + StartReadTxResponse start_read_tx(); + + /** + * @brief Unregisters a read transaction. Cursors already opened against it keep it alive until they are closed. + * @return ok is false if no read transaction with this id was open + * @throws std::runtime_error if the request carries no id + */ + BoolResponse close_read_tx(const CloseReadTxRequest& req); + GetResponse get(const GetRequest& req); HasResponse has(const HasRequest& req); @@ -60,6 +100,8 @@ class LMDBStoreWrapper : public Napi::ObjectWrap { BoolResponse copy_store(const CopyStoreRequest& req); + static bool _set_cursor_start(const lmdblib::LMDBCursor& cursor, lmdblib::Key& key, bool reverse); + static std::pair _advance_cursor(const lmdblib::LMDBCursor& cursor, bool reverse, uint64_t page_size); diff --git a/native-packages/kvdb/ts/package.json b/native-packages/kvdb/ts/package.json index 72caccac35b..5a8ec4d5b91 100644 --- a/native-packages/kvdb/ts/package.json +++ b/native-packages/kvdb/ts/package.json @@ -28,6 +28,7 @@ }, "devDependencies": { "@types/node": "^22.15.17", + "msgpackr": "1.11.2", "typescript": "^5.3.3" } } diff --git a/native-packages/kvdb/ts/test/read_tx.test.mjs b/native-packages/kvdb/ts/test/read_tx.test.mjs new file mode 100644 index 00000000000..54621db9190 --- /dev/null +++ b/native-packages/kvdb/ts/test/read_tx.test.mjs @@ -0,0 +1,311 @@ +// Exercises the addon's read transaction protocol over the same msgpack wire format the kv-store client uses. +// Run after building the addon: `node --test test/read_tx.test.mjs` (AZTEC_KVDB_NAPI_PATH overrides the binary). +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { createRequire } from "node:module"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, describe, it } from "node:test"; + +import { Encoder } from "msgpackr"; + +import { findNapiBinary } from "../dest/index.js"; + +const MSG = { + OPEN_DATABASE: 100, + GET: 101, + HAS: 102, + START_CURSOR: 103, + ADVANCE_CURSOR: 104, + CLOSE_CURSOR: 106, + BATCH: 107, + CLOSE: 109, + START_READ_TX: 111, + CLOSE_READ_TX: 112, +}; + +const napiPath = findNapiBinary(); +assert.ok(napiPath, "kvdb NAPI binary not found, build it first"); +const { LMDBStore } = createRequire(import.meta.url)(napiPath); + +const encoder = new Encoder({ useRecords: false, int64AsType: "bigint" }); +const b = (s) => Buffer.from(s); +const str = (v) => Buffer.from(v).toString(); +const values = (resp) => + resp.values.map((v) => (v == null ? null : v.map(str))); +const keysOf = (entries) => entries.map(([k]) => str(k)); + +let messageId = 1; +function openStore(dir, maxReaders) { + const store = new LMDBStore(dir, 10 * 1024, maxReaders); + const send = async (msgType, value) => { + const msg = { msgType, header: { messageId: messageId++, requestId: 0 } }; + if (value !== undefined) { + msg.value = value; + } + return encoder.unpack(await store.call(encoder.encode(msg))).value; + }; + const batch = (data = [], index = []) => + send(MSG.BATCH, { + batches: new Map([ + ["data", { addEntries: data, removeEntries: [] }], + ["index", { addEntries: index, removeEntries: [] }], + ]), + }); + const startCursor = (key, txId, opts = {}) => + send(MSG.START_CURSOR, { + key: b(key), + reverse: false, + count: 1, + onePage: false, + db: "data", + ...opts, + txId, + }); + return { send, batch, startCursor }; +} + +async function drainCursor(send, start) { + const keys = keysOf(start.entries); + let cursor = start.cursor; + while (cursor != null) { + const next = await send(MSG.ADVANCE_CURSOR, { cursor, count: 1 }); + keys.push(...keysOf(next.entries)); + if (next.done) { + await send(MSG.CLOSE_CURSOR, { cursor }); + cursor = null; + } + } + return keys; +} + +describe("kvdb read transactions", () => { + let dir; + let send; + let batch; + let startCursor; + const maxReaders = 4; + + before(async () => { + dir = mkdtempSync(join(tmpdir(), "kvdb-read-tx-")); + ({ send, batch, startCursor } = openStore(dir, maxReaders)); + await send(MSG.OPEN_DATABASE, { db: "data", uniqueKeys: true }); + await send(MSG.OPEN_DATABASE, { db: "index", uniqueKeys: false }); + await batch( + [ + [b("a"), [b("a1")]], + [b("b"), [b("b1")]], + [b("c"), [b("c1")]], + ], + [[b("k"), [b("v1"), b("v2")]]], + ); + }); + + after(() => rmSync(dir, { recursive: true, force: true })); + + it("serves requests that omit txId or set it to null as before", async () => { + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("a")], db: "data" })), + [["a1"]], + ); + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("a")], db: "data", txId: null })), + [["a1"]], + ); + const legacy = await send(MSG.START_CURSOR, { + key: b("a"), + reverse: false, + count: null, + onePage: null, + db: "data", + }); + assert.equal(legacy.cursor, null); + assert.deepEqual(keysOf(legacy.entries), ["a", "b", "c"]); + }); + + it("reads and iterates a snapshot that later writes do not change", async () => { + const { txId } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", txId); + assert.notEqual(cursor.cursor, null); + + await batch( + [ + [b("a"), [b("a2")]], + [b("d"), [b("d2")]], + ], + [[b("k"), [b("v3")]]], + ); + + assert.deepEqual( + values( + await send(MSG.GET, { keys: [b("a"), b("d")], db: "data", txId }), + ), + [["a1"], null], + ); + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("k")], db: "index", txId })), + [["v1", "v2"]], + ); + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("a"), b("d")], db: "data" })), + [["a2"], ["d2"]], + ); + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("k")], db: "index" })), + [["v1", "v2", "v3"]], + ); + + const entries = [ + [b("d"), null], + [b("a"), [b("a1")]], + ]; + assert.deepEqual(await send(MSG.HAS, { entries, db: "data", txId }), { + exists: [false, true], + }); + assert.deepEqual(await send(MSG.HAS, { entries, db: "data" }), { + exists: [true, false], + }); + assert.deepEqual( + await send(MSG.HAS, { + entries: [[b("k"), [b("v3")]]], + db: "index", + txId, + }), + { exists: [false] }, + ); + + const reverse = await startCursor("z", txId, { reverse: true, count: 100 }); + assert.deepEqual(keysOf(reverse.entries), ["c", "b", "a"]); + assert.deepEqual(await drainCursor(send, cursor), ["a", "b", "c"]); + + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: true }); + }); + + it("keeps cursors usable after their read transaction is closed", async () => { + const { txId } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", txId); + await batch([[b("e"), [b("e3")]]]); + + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: true }); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: false }); + assert.deepEqual(await drainCursor(send, cursor), ["a", "b", "c", "d"]); + }); + + it("rejects unknown and missing read transaction ids", async () => { + const { txId } = await send(MSG.START_READ_TX); + await send(MSG.CLOSE_READ_TX, { txId }); + await assert.rejects( + send(MSG.GET, { keys: [b("a")], db: "data", txId }), + /Read transaction .* not found/, + ); + await assert.rejects( + startCursor("a", 987654), + /Read transaction 987654 not found/, + ); + await assert.rejects( + send(MSG.CLOSE_READ_TX, {}), + /missing the read transaction id/, + ); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId: 987654 }), { + ok: false, + }); + }); + + it("serializes concurrent gets and cursor scans on one read transaction", async () => { + const { txId } = await send(MSG.START_READ_TX); + const expected = keysOf( + (await startCursor("a", txId, { count: 100 })).entries, + ); + const work = []; + for (let i = 0; i < 200; i++) { + if (i % 20 === 0) { + work.push(batch([[b(`n${i}`), [b("x")]]], [[b("k"), [b(`z${i}`)]]])); + } + work.push( + send(MSG.GET, { keys: [b("a"), b("k")], db: "data", txId }).then( + (r) => assert.deepEqual(values(r), [["a2"], null]), + ), + send(MSG.GET, { keys: [b("k")], db: "index", txId }).then((r) => + assert.deepEqual(values(r), [["v1", "v2", "v3"]]), + ), + startCursor("a", txId) + .then((start) => drainCursor(send, start)) + .then((keys) => assert.deepEqual(keys, expected)), + ); + } + await Promise.all(work); + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: true }); + }); + + it("rejects new read transactions instead of waiting when reader slots run out", async () => { + const held = []; + // one slot always stays free for requests that open their own read transaction + for (let i = 0; i < maxReaders - 1; i++) { + held.push((await send(MSG.START_READ_TX)).txId); + } + const attempts = await Promise.allSettled( + Array.from({ length: 8 }, () => send(MSG.START_READ_TX)), + ); + for (const attempt of attempts) { + assert.equal(attempt.status, "rejected"); + assert.match(attempt.reason.message, /No reader slot available/); + } + assert.deepEqual( + values(await send(MSG.GET, { keys: [b("a")], db: "data" })), + [["a2"]], + ); + + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId: held.pop() }), { + ok: true, + }); + held.push((await send(MSG.START_READ_TX)).txId); + for (const txId of held) { + assert.deepEqual(await send(MSG.CLOSE_READ_TX, { txId }), { ok: true }); + } + }); + + it("counts closed read transactions kept alive by cursors against the reader slots", async () => { + const { txId } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", txId); + await send(MSG.CLOSE_READ_TX, { txId }); + + const held = []; + for (let i = 0; i < maxReaders - 2; i++) { + held.push((await send(MSG.START_READ_TX)).txId); + } + await assert.rejects(send(MSG.START_READ_TX), /No reader slot available/); + + await send(MSG.CLOSE_CURSOR, { cursor: cursor.cursor }); + held.push((await send(MSG.START_READ_TX)).txId); + for (const id of held) { + await send(MSG.CLOSE_READ_TX, { txId: id }); + } + }); + + it("closes the store with live read transactions and cursors, releasing the environment", async () => { + const { txId } = await send(MSG.START_READ_TX); + const cursor = await startCursor("a", txId); + assert.notEqual(cursor.cursor, null); + + assert.deepEqual(await send(MSG.CLOSE), { ok: true }); + await assert.rejects( + send(MSG.GET, { keys: [b("a")], db: "data", txId }), + /closed/, + ); + + const reopened = openStore(dir, maxReaders); + await reopened.send(MSG.OPEN_DATABASE, { db: "data", uniqueKeys: true }); + const { txId: fresh } = await reopened.send(MSG.START_READ_TX); + assert.deepEqual( + values( + await reopened.send(MSG.GET, { + keys: [b("d")], + db: "data", + txId: fresh, + }), + ), + [["d2"]], + ); + assert.deepEqual(await reopened.send(MSG.CLOSE), { ok: true }); + }); +}); diff --git a/native-packages/kvdb/yarn.lock b/native-packages/kvdb/yarn.lock index 621e736e478..a071444928a 100644 --- a/native-packages/kvdb/yarn.lock +++ b/native-packages/kvdb/yarn.lock @@ -44,6 +44,7 @@ __metadata: "@aztec-labs/kvdb-linux-arm64": "npm:0.1.0" "@aztec-labs/kvdb-linux-x64": "npm:0.1.0" "@types/node": "npm:^22.15.17" + msgpackr: "npm:1.11.2" tslib: "npm:^2.4.0" typescript: "npm:^5.3.3" dependenciesMeta: @@ -58,6 +59,57 @@ __metadata: languageName: unknown linkType: soft +"@isaacs/fs-minipass@npm:^4.0.0": + version: 4.0.1 + resolution: "@isaacs/fs-minipass@npm:4.0.1" + dependencies: + minipass: "npm:^7.0.4" + checksum: 10c0/c25b6dc1598790d5b55c0947a9b7d111cfa92594db5296c3b907e2f533c033666f692a3939eadac17b1c7c40d362d0b0635dc874cbfe3e70db7c2b07cc97a5d2 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-darwin-arm64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-darwin-arm64@npm:3.0.4" + conditions: os=darwin & cpu=arm64 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-darwin-x64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-darwin-x64@npm:3.0.4" + conditions: os=darwin & cpu=x64 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-linux-arm64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-linux-arm64@npm:3.0.4" + conditions: os=linux & cpu=arm64 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-linux-arm@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-linux-arm@npm:3.0.4" + conditions: os=linux & cpu=arm + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-linux-x64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-linux-x64@npm:3.0.4" + conditions: os=linux & cpu=x64 + languageName: node + linkType: hard + +"@msgpackr-extract/msgpackr-extract-win32-x64@npm:3.0.4": + version: 3.0.4 + resolution: "@msgpackr-extract/msgpackr-extract-win32-x64@npm:3.0.4" + conditions: os=win32 & cpu=x64 + languageName: node + linkType: hard + "@types/node@npm:^22.15.17": version: 22.20.0 resolution: "@types/node@npm:22.20.0" @@ -67,6 +119,216 @@ __metadata: languageName: node linkType: hard +"abbrev@npm:^5.0.0": + version: 5.0.0 + resolution: "abbrev@npm:5.0.0" + checksum: 10c0/8e88f5c798ea4562d28c5a3e9ad69e3879890bc5d695d8f2dffb8609be4c890aacc8f80ef4553fdd2c6a62d70c2ce8bc57b38074e383beb7487bdafa9ed42ea5 + languageName: node + linkType: hard + +"chownr@npm:^3.0.0": + version: 3.0.0 + resolution: "chownr@npm:3.0.0" + checksum: 10c0/43925b87700f7e3893296c8e9c56cc58f926411cce3a6e5898136daaf08f08b9a8eb76d37d3267e707d0dcc17aed2e2ebdf5848c0c3ce95cf910a919935c1b10 + languageName: node + linkType: hard + +"detect-libc@npm:^2.0.1": + version: 2.1.2 + resolution: "detect-libc@npm:2.1.2" + checksum: 10c0/acc675c29a5649fa1fb6e255f993b8ee829e510b6b56b0910666949c80c364738833417d0edb5f90e4e46be17228b0f2b66a010513984e18b15deeeac49369c4 + languageName: node + linkType: hard + +"env-paths@npm:^2.2.0": + version: 2.2.1 + resolution: "env-paths@npm:2.2.1" + checksum: 10c0/285325677bf00e30845e330eec32894f5105529db97496ee3f598478e50f008c5352a41a30e5e72ec9de8a542b5a570b85699cd63bd2bc646dbcb9f311d83bc4 + languageName: node + linkType: hard + +"exponential-backoff@npm:^3.1.1": + version: 3.1.3 + resolution: "exponential-backoff@npm:3.1.3" + checksum: 10c0/77e3ae682b7b1f4972f563c6dbcd2b0d54ac679e62d5d32f3e5085feba20483cf28bd505543f520e287a56d4d55a28d7874299941faf637e779a1aa5994d1267 + languageName: node + linkType: hard + +"fdir@npm:^6.5.0": + version: 6.5.0 + resolution: "fdir@npm:6.5.0" + peerDependencies: + picomatch: ^3 || ^4 + peerDependenciesMeta: + picomatch: + optional: true + checksum: 10c0/e345083c4306b3aed6cb8ec551e26c36bab5c511e99ea4576a16750ddc8d3240e63826cc624f5ae17ad4dc82e68a253213b60d556c11bfad064b7607847ed07f + languageName: node + linkType: hard + +"graceful-fs@npm:^4.2.6": + version: 4.2.11 + resolution: "graceful-fs@npm:4.2.11" + checksum: 10c0/386d011a553e02bc594ac2ca0bd6d9e4c22d7fa8cfbfc448a6d148c59ea881b092db9dbe3547ae4b88e55f1b01f7c4a2ecc53b310c042793e63aa44cf6c257f2 + languageName: node + linkType: hard + +"isexe@npm:^4.0.0": + version: 4.0.0 + resolution: "isexe@npm:4.0.0" + checksum: 10c0/5884815115bceac452877659a9c7726382531592f43dc29e5d48b7c4100661aed54018cb90bd36cb2eaeba521092570769167acbb95c18d39afdccbcca06c5ce + languageName: node + linkType: hard + +"minipass@npm:^7.0.4, minipass@npm:^7.1.2": + version: 7.1.3 + resolution: "minipass@npm:7.1.3" + checksum: 10c0/539da88daca16533211ea5a9ee98dc62ff5742f531f54640dd34429e621955e91cc280a91a776026264b7f9f6735947629f920944e9c1558369e8bf22eb33fbb + languageName: node + linkType: hard + +"minizlib@npm:^3.1.0": + version: 3.1.0 + resolution: "minizlib@npm:3.1.0" + dependencies: + minipass: "npm:^7.1.2" + checksum: 10c0/5aad75ab0090b8266069c9aabe582c021ae53eb33c6c691054a13a45db3b4f91a7fb1bd79151e6b4e9e9a86727b522527c0a06ec7d45206b745d54cd3097bcec + languageName: node + linkType: hard + +"msgpackr-extract@npm:^3.0.2": + version: 3.0.4 + resolution: "msgpackr-extract@npm:3.0.4" + dependencies: + "@msgpackr-extract/msgpackr-extract-darwin-arm64": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-darwin-x64": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-linux-arm": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-linux-arm64": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-linux-x64": "npm:3.0.4" + "@msgpackr-extract/msgpackr-extract-win32-x64": "npm:3.0.4" + node-gyp: "npm:latest" + node-gyp-build-optional-packages: "npm:5.2.2" + dependenciesMeta: + "@msgpackr-extract/msgpackr-extract-darwin-arm64": + optional: true + "@msgpackr-extract/msgpackr-extract-darwin-x64": + optional: true + "@msgpackr-extract/msgpackr-extract-linux-arm": + optional: true + "@msgpackr-extract/msgpackr-extract-linux-arm64": + optional: true + "@msgpackr-extract/msgpackr-extract-linux-x64": + optional: true + "@msgpackr-extract/msgpackr-extract-win32-x64": + optional: true + bin: + download-msgpackr-prebuilds: bin/download-prebuilds.js + checksum: 10c0/582a9d17abbf3019e600e948736695056280ce401fd0235ee2474e95f9952208b9f6cce4d0e355b03b7d3c5630e6c3d11fe5fc27fdedb2311cce48de464338d8 + languageName: node + linkType: hard + +"msgpackr@npm:1.11.2": + version: 1.11.2 + resolution: "msgpackr@npm:1.11.2" + dependencies: + msgpackr-extract: "npm:^3.0.2" + dependenciesMeta: + msgpackr-extract: + optional: true + checksum: 10c0/7d2e81ca82c397b2352d470d6bc8f4a967fe4fe14f8fc1fc9906b23009fdfb543999b1ad29c700b8861581e0b6bf903d6f0fefb69a09375cbca6d4d802e6c906 + languageName: node + linkType: hard + +"node-gyp-build-optional-packages@npm:5.2.2": + version: 5.2.2 + resolution: "node-gyp-build-optional-packages@npm:5.2.2" + dependencies: + detect-libc: "npm:^2.0.1" + bin: + node-gyp-build-optional-packages: bin.js + node-gyp-build-optional-packages-optional: optional.js + node-gyp-build-optional-packages-test: build-test.js + checksum: 10c0/c81128c6f91873381be178c5eddcbdf66a148a6a89a427ce2bcd457593ce69baf2a8662b6d22cac092d24aa9c43c230dec4e69b3a0da604503f4777cd77e282b + languageName: node + linkType: hard + +"node-gyp@npm:latest": + version: 13.1.0 + resolution: "node-gyp@npm:13.1.0" + dependencies: + env-paths: "npm:^2.2.0" + exponential-backoff: "npm:^3.1.1" + graceful-fs: "npm:^4.2.6" + nopt: "npm:^10.0.0" + proc-log: "npm:^7.0.0" + semver: "npm:^7.3.5" + tar: "npm:^7.5.7" + tinyglobby: "npm:^0.2.12" + undici: "npm:^8.4.1" + which: "npm:^7.0.0" + bin: + node-gyp: bin/node-gyp.js + checksum: 10c0/c6693e52e0e111b21f5cd8ff303b41b51f40338adcc79ac5c6a02399eba7039d50dc816dcafcc485870291a641f7cb75db8887a9bdd0de537fe635a3a81cba5b + languageName: node + linkType: hard + +"nopt@npm:^10.0.0": + version: 10.0.1 + resolution: "nopt@npm:10.0.1" + dependencies: + abbrev: "npm:^5.0.0" + bin: + nopt: bin/nopt.js + checksum: 10c0/980d89257f9587f3e1f77877ddbf905d6aa3b738ec33e49a4fa1a059a0dd82eb28063982b150654a7ae9de386f2ead60e56172db7d37cf56de545f7392a2a26a + languageName: node + linkType: hard + +"picomatch@npm:^4.0.4": + version: 4.0.7 + resolution: "picomatch@npm:4.0.7" + checksum: 10c0/beb6ae02c43ae44e84883b90830196d9046b1726ead292adcf7f57945e0bb0d992d68563d87e03b484b6f3c9a5c6defda7523477f047d7f0e663f126cc01787f + languageName: node + linkType: hard + +"proc-log@npm:^7.0.0": + version: 7.0.0 + resolution: "proc-log@npm:7.0.0" + checksum: 10c0/b89c2d862604f35fec795477b0c7e376feab3ba0d4f4d291c4e959567442697cf451ac557d0623c1cc38af45a78128b983410f397a10c5d3a67f76c33de4754b + languageName: node + linkType: hard + +"semver@npm:^7.3.5": + version: 7.8.5 + resolution: "semver@npm:7.8.5" + bin: + semver: bin/semver.js + checksum: 10c0/b1f3127a5be8125a94f37188b361c212466c292c6910adce3ec106cff5dc211ccaedc4739c11bb70fda59d6fc1f040a9bca289f4e093451521a2372e5231fe0c + languageName: node + linkType: hard + +"tar@npm:^7.5.7": + version: 7.5.22 + resolution: "tar@npm:7.5.22" + dependencies: + "@isaacs/fs-minipass": "npm:^4.0.0" + chownr: "npm:^3.0.0" + minipass: "npm:^7.1.2" + minizlib: "npm:^3.1.0" + yallist: "npm:^5.0.0" + checksum: 10c0/1311f6be85a8157ac4c9147bae43e13923d2a1aae15e4aa1bd5239e4e03d2cf53cfe103dde7f35832fbb4c938b042856bc8e9a0afd29abd05e2d1608788c4fea + languageName: node + linkType: hard + +"tinyglobby@npm:^0.2.12": + version: 0.2.17 + resolution: "tinyglobby@npm:0.2.17" + dependencies: + fdir: "npm:^6.5.0" + picomatch: "npm:^4.0.4" + checksum: 10c0/7f7bb0f197c88bc4b20c231e0deca4240ca3bf313a88f5a7fee93a872b84966a4d50220947c0455ad07a60b3b360961c5b7fd979222aeb716a9f99b412002e4c + languageName: node + linkType: hard + "tslib@npm:^2.4.0": version: 2.8.1 resolution: "tslib@npm:2.8.1" @@ -100,3 +362,28 @@ __metadata: checksum: 10c0/c01ed51829b10aa72fc3ce64b747f8e74ae9b60eafa19a7b46ef624403508a54c526ffab06a14a26b3120d055e1104d7abe7c9017e83ced038ea5cf52f8d5e04 languageName: node linkType: hard + +"undici@npm:^8.4.1": + version: 8.11.2 + resolution: "undici@npm:8.11.2" + checksum: 10c0/106f6991178fb7562fd90cdec7993c5f807a9428682306091f0c4b27fc474ec291f868e29972c6252988bf061ad193df943b3b99e775be157a28d92bbb95699d + languageName: node + linkType: hard + +"which@npm:^7.0.0": + version: 7.0.0 + resolution: "which@npm:7.0.0" + dependencies: + isexe: "npm:^4.0.0" + bin: + node-which: bin/which.js + checksum: 10c0/ca0b54f198f78bbc4b7c02e34bda8d335cb352e0adb4cbca1c37b1a957af3a879a82c4c27ca6525bc942f548d8b64f816ef6528360af9f3de55ffb9b979b620d + languageName: node + linkType: hard + +"yallist@npm:^5.0.0": + version: 5.0.0 + resolution: "yallist@npm:5.0.0" + checksum: 10c0/a499c81ce6d4a1d260d4ea0f6d49ab4da09681e32c3f0472dee16667ed69d01dae63a3b81745a24bd78476ec4fcf856114cb4896ace738e01da34b2c42235416 + languageName: node + linkType: hard diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.cpp index 50bcbabf474..b896509f21a 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.cpp @@ -43,6 +43,11 @@ void LMDBEnvironment::wait_for_reader() _readGuard.wait(); } +bool LMDBEnvironment::try_wait_for_reader(uint32_t keepFree) +{ + return _readGuard.try_wait(keepFree); +} + void LMDBEnvironment::release_reader() { _readGuard.release(); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.hpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.hpp index 8b42fb43d09..795247f6c79 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.hpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_environment.hpp @@ -49,6 +49,13 @@ class LMDBEnvironment { void wait_for_reader(); + /** + * @brief Takes a reader permit without blocking. + * @param keepFree How many permits must remain available after this one is taken + * @return false, taking nothing, if fewer than `keepFree + 1` permits are available + */ + bool try_wait_for_reader(uint32_t keepFree = 0); + void release_reader(); void wait_for_writer(); @@ -86,6 +93,16 @@ class LMDBEnvironment { ++_current; } + bool try_wait(uint32_t keepFree) + { + std::unique_lock lock(_lock); + if (static_cast(_current) + keepFree >= _maxAllowed) { + return false; + } + ++_current; + return true; + } + void release() { std::unique_lock lock(_lock); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp index 61d574ab486..3c036ed6b33 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.cpp @@ -113,7 +113,16 @@ void LMDBStore::put(std::vector& data) void LMDBStore::get(KeysVector& keys, OptionalValuesVector& values, const std::string& name) { - get(keys, values, get_database(name)); + LMDBDatabase::SharedPtr db = get_database(name); + get(keys, values, db, 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, @@ -136,10 +145,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) { @@ -179,6 +190,14 @@ void LMDBStore::get(KeysVector& keys, OptionalValuesVector& values, LMDBDatabase } void LMDBStore::has(const KeyOptionalValuesVector& entries, std::vector& results, const std::string& name) +{ + has(entries, results, name, create_shared_read_transaction()); +} + +void LMDBStore::has(const KeyOptionalValuesVector& entries, + std::vector& results, + const std::string& name, + ReadTransaction::SharedPtr tx) { auto string_cmp = [](const Key& a, const Key& b) { return std::lexicographical_compare(a.begin(), a.end(), b.begin(), b.end()); @@ -191,7 +210,7 @@ void LMDBStore::has(const KeyOptionalValuesVector& entries, std::vector& r KeysVector keys(key_set.begin(), key_set.end()); OptionalValuesVector vals; - get(keys, vals, name); + get(keys, vals, name, std::move(tx)); results.reserve(entries.size()); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp index d55d1d5bd40..a4113632a37 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.hpp @@ -48,7 +48,22 @@ class LMDBStore : public LMDBStoreBase { void put(std::vector& data); void get(KeysVector& keys, OptionalValuesVector& values, const std::string& name); + /** + * @brief Reads the given keys through 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 that nothing else uses `tx` + * (including opening or advancing cursors against it) while this call runs. + */ + void get(KeysVector& keys, OptionalValuesVector& values, const std::string& name, ReadTransaction::SharedPtr tx); void has(const KeyOptionalValuesVector& entries, std::vector& results, const std::string& name); + /** + * @brief Checks the given entries through an already open read transaction, like the `get` overload above. + * @note The caller must ensure that nothing else uses `tx` while this call runs. + */ + void has(const KeyOptionalValuesVector& entries, + std::vector& results, + const std::string& name, + ReadTransaction::SharedPtr tx); Cursor::Ptr create_cursor(ReadTransaction::SharedPtr tx, const std::string& dbName); @@ -63,7 +78,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 diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp index e297e6c3681..438381fcdc4 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store.test.cpp @@ -9,7 +9,9 @@ #include #include #include +#include #include +#include #include #include "lmdblib/fixtures.hpp" @@ -229,6 +231,191 @@ 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); + auto addedKey = get_key(1); + auto addedValue = get_value(1, 1); + + KeyOptionalValuesVector toDelete; + KeyDupValuesVector toWrite = { { { key, { original } } } }; + std::vector putDatas = { { toWrite, toDelete, dbName } }; + store->put(putDatas); + + LMDBStore::ReadTransaction::SharedPtr tx = store->create_shared_read_transaction(); + + KeysVector keys = { { key }, { addedKey } }; + OptionalValuesVector snapshot; + store->get(keys, snapshot, dbName, tx); + ASSERT_EQ(snapshot.size(), 2); + EXPECT_EQ(snapshot[0].value(), ValuesVector{ original }); + EXPECT_FALSE(snapshot[1].has_value()); + + // commits while the read transaction is still open + toWrite = { KeyValuesPair{ key, { updated } }, KeyValuesPair{ addedKey, { addedValue } } }; + putDatas = { { toWrite, toDelete, dbName } }; + store->put(putDatas); + + OptionalValuesVector afterWrite; + store->get(keys, afterWrite, dbName, tx); + ASSERT_EQ(afterWrite.size(), 2); + EXPECT_EQ(afterWrite[0].value(), ValuesVector{ original }); + EXPECT_FALSE(afterWrite[1].has_value()); + + OptionalValuesVector latest; + store->get(keys, latest, dbName); + ASSERT_EQ(latest.size(), 2); + EXPECT_EQ(latest[0].value(), ValuesVector{ updated }); + EXPECT_EQ(latest[1].value(), ValuesVector{ addedValue }); +} + +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(); + + // adds a further duplicate to the key once the read transaction is open + KeyOptionalValuesVector toDelete; + KeyDupValuesVector toWrite = { { { get_key(2), { get_value(2, numValues) } } } }; + std::vector putDatas = { { toWrite, toDelete, dbName } }; + store->put(putDatas); + + 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_EQ(values.size(), 1); + ASSERT_TRUE(values[0].has_value()); + EXPECT_EQ(values[0].value(), expected); + + OptionalValuesVector latest; + store->get(keys, latest, dbName); + expected.emplace_back(get_value(2, numValues)); + ASSERT_TRUE(latest[0].has_value()); + EXPECT_EQ(latest[0].value(), expected); +} + +TEST_F(LMDBStoreTest, can_share_a_read_transaction_between_gets_and_cursors) +{ + LMDBStore::Ptr store = create_store(); + const std::string dbName = "Test Database"; + store->open_database(dbName); + + int64_t numKeys = 5; + write_test_data({ dbName }, numKeys, 1, *store); + + LMDBStore::ReadTransaction::SharedPtr tx = store->create_shared_read_transaction(); + LMDBStore::Cursor::Ptr cursor = store->create_cursor(tx, dbName); + + KeyOptionalValuesVector toDelete = { { get_key(0), std::nullopt } }; + KeyDupValuesVector toWrite; + std::vector putDatas = { { toWrite, toDelete, dbName } }; + store->put(putDatas); + + KeysVector keys = { { get_key(0) } }; + OptionalValuesVector values; + store->get(keys, values, dbName, tx); + ASSERT_TRUE(values[0].has_value()); + EXPECT_EQ(values[0].value(), ValuesVector{ get_value(0, 0) }); + + EXPECT_TRUE(cursor->set_at_start()); + KeyDupValuesVector entries; + cursor->read_next(static_cast(numKeys), entries); + ASSERT_EQ(entries.size(), static_cast(numKeys)); + EXPECT_EQ(entries[0].first, get_key(0)); +} + +TEST_F(LMDBStoreTest, try_create_shared_read_transaction_does_not_wait_and_keeps_permits_free) +{ + const uint32_t maxReaders = 3; + LMDBStore::Ptr store = std::make_unique(_directory, _mapSize, maxReaders, 1); + + std::vector held; + for (uint32_t i = 0; i < maxReaders - 1; i++) { + held.push_back(store->try_create_shared_read_transaction(1)); + ASSERT_NE(held.back(), nullptr); + } + EXPECT_EQ(store->try_create_shared_read_transaction(1), nullptr); + + held.push_back(store->try_create_shared_read_transaction()); + ASSERT_NE(held.back(), nullptr); + EXPECT_EQ(store->try_create_shared_read_transaction(), nullptr); + + held.pop_back(); + held.pop_back(); + EXPECT_NE(store->try_create_shared_read_transaction(1), nullptr); +} + +TEST_F(LMDBStoreTest, reader_permit_is_returned_when_a_read_transaction_fails_to_begin) +{ + const uint32_t maxReaders = 2; + int toParent[2]; + int toChild[2]; + ASSERT_EQ(pipe(toParent), 0); + ASSERT_EQ(pipe(toChild), 0); + + // Another process occupies every slot in the environment's reader table, so mdb_txn_begin fails here + pid_t pid = fork(); + ASSERT_GE(pid, 0); + if (pid == 0) { + { + auto env = std::make_shared(_directory, _mapSize, 1, maxReaders); + std::vector> txs; + for (uint32_t i = 0; i < maxReaders; i++) { + env->wait_for_reader(); + txs.push_back(std::make_unique(env)); + } + char c = 'r'; + if (write(toParent[1], &c, 1) != 1 || read(toChild[0], &c, 1) != 1) { + _exit(1); + } + } + _exit(0); + } + + char c = 0; + ASSERT_EQ(read(toParent[0], &c, 1), 1); + + LMDBStore::Ptr store = std::make_unique(_directory, _mapSize, maxReaders, 1); + // more failed attempts than there are permits, so a leaked permit would surface as a nullptr here + EXPECT_THROW(store->create_shared_read_transaction(), std::runtime_error); + for (uint32_t i = 0; i < maxReaders; i++) { + EXPECT_THROW(store->try_create_shared_read_transaction(), std::runtime_error); + } + + ASSERT_EQ(write(toChild[1], &c, 1), 1); + int status = 0; + ASSERT_EQ(waitpid(pid, &status, 0), pid); + ASSERT_TRUE(WIFEXITED(status)); + ASSERT_EQ(WEXITSTATUS(status), 0); + for (int fd : { toParent[0], toParent[1], toChild[0], toChild[1] }) { + close(fd); + } + + std::vector held; + for (uint32_t i = 0; i < maxReaders; i++) { + held.push_back(store->try_create_shared_read_transaction()); + EXPECT_NE(held.back(), nullptr); + } +} + TEST_F(LMDBStoreTest, can_not_read_from_non_existent_database) { LMDBStore::Ptr store = create_store(); diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp index efc85c1de91..acd29e21ced 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.cpp @@ -7,28 +7,60 @@ LMDBStoreBase::LMDBStoreBase( , _environment(std::make_shared(_dbDirectory, mapSizeKb, maxDbs, maxNumReaders, ephemeral)) {} LMDBStoreBase::~LMDBStoreBase() = default; +namespace { +// A transaction returns its permit from its destructor, which never runs if its constructor (mdb_txn_begin) throws, so +// a permit taken before construction has to be given back here +template auto with_permit(LMDBEnvironment& env, void (LMDBEnvironment::*release)(), Make make) +{ + try { + return make(); + } catch (...) { + (env.*release)(); + throw; + } +} +} // namespace + LMDBStoreBase::ReadTransaction::Ptr LMDBStoreBase::create_read_transaction() const { _environment->wait_for_reader(); - return std::make_unique(_environment); + return with_permit(*_environment, &LMDBEnvironment::release_reader, [&] { + return std::make_unique(_environment); + }); } LMDBStoreBase::ReadTransaction::SharedPtr LMDBStoreBase::create_shared_read_transaction() const { _environment->wait_for_reader(); - return std::make_shared(_environment); + return with_permit(*_environment, &LMDBEnvironment::release_reader, [&] { + return std::make_shared(_environment); + }); +} + +LMDBStoreBase::ReadTransaction::SharedPtr LMDBStoreBase::try_create_shared_read_transaction(uint32_t keepFree) const +{ + if (!_environment->try_wait_for_reader(keepFree)) { + return nullptr; + } + return with_permit(*_environment, &LMDBEnvironment::release_reader, [&] { + return std::make_shared(_environment); + }); } LMDBStoreBase::DBCreationTransaction::Ptr LMDBStoreBase::create_db_transaction() const { _environment->wait_for_writer(); - return std::make_unique(_environment); + return with_permit(*_environment, &LMDBEnvironment::release_writer, [&] { + return std::make_unique(_environment); + }); } LMDBStoreBase::WriteTransaction::Ptr LMDBStoreBase::create_write_transaction() const { _environment->wait_for_writer(); - return std::make_unique(_environment); + return with_permit(*_environment, &LMDBEnvironment::release_writer, [&] { + return std::make_unique(_environment); + }); } void LMDBStoreBase::copy_store(const std::string& dstPath, bool compact) diff --git a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.hpp b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.hpp index 234721b6db2..90dfa624cdb 100644 --- a/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.hpp +++ b/native-packages/lmdblib/cpp/src/lmdblib/lmdb_store_base.hpp @@ -20,6 +20,12 @@ class LMDBStoreBase { virtual ~LMDBStoreBase() = 0; ReadTransaction::Ptr create_read_transaction() const; ReadTransaction::SharedPtr create_shared_read_transaction() const; + /** + * @brief Like create_shared_read_transaction, but never waits for a reader permit. + * @param keepFree How many reader permits must remain available once this transaction holds one + * @return nullptr if no permit could be taken + */ + ReadTransaction::SharedPtr try_create_shared_read_transaction(uint32_t keepFree = 0) const; WriteTransaction::Ptr create_write_transaction() const; LMDBDatabaseCreationTransaction::Ptr create_db_transaction() const; void copy_store(const std::string& dstPath, bool compact);