From aecd46bc3a9aae8ad4f29f0c93b3e4c66fb86768 Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Sat, 17 Oct 2026 16:41:29 -0400 Subject: [PATCH 1/6] Add turbo parallel wire block writer. --- .../impl/query/archive/wire_reader.ipp | 83 ++++++++++++++++++- include/bitcoin/database/query.hpp | 8 +- test/query/archive/wire_reader.cpp | 71 ++++++++++++++++ 3 files changed, 159 insertions(+), 3 deletions(-) diff --git a/include/bitcoin/database/impl/query/archive/wire_reader.ipp b/include/bitcoin/database/impl/query/archive/wire_reader.ipp index 070989dc0..e19929074 100644 --- a/include/bitcoin/database/impl/query/archive/wire_reader.ipp +++ b/include/bitcoin/database/impl/query/archive/wire_reader.ipp @@ -19,6 +19,7 @@ #ifndef LIBBITCOIN_DATABASE_QUERY_ARCHIVE_WIRE_READER_IPP #define LIBBITCOIN_DATABASE_QUERY_ARCHIVE_WIRE_READER_IPP +#include #include namespace libbitcoin { @@ -254,8 +255,8 @@ data_chunk CLASS::get_wire_tx(const tx_link& link, bool witness) const NOEXCEPT } TEMPLATE -data_chunk CLASS::get_wire_block(const header_link& link, - bool witness) const NOEXCEPT +data_chunk CLASS::get_wire_block(const header_link& link, bool witness, + bool turbo) const NOEXCEPT { using namespace system; size_t size{}; @@ -263,6 +264,9 @@ data_chunk CLASS::get_wire_block(const header_link& link, return {}; data_chunk data(size); + if (turbo) + return get_wire_block(data, link, witness, turbo) ? data : data_chunk{}; + stream::flip::fast ostream(data); flip::bytes::fast out(ostream); if (!get_wire_block(out, link, witness) || !out) @@ -273,6 +277,81 @@ data_chunk CLASS::get_wire_block(const header_link& link, return data; } +TEMPLATE +template +bool CLASS::get_wire_block(const system::data_slab& out, + const header_link& link, bool witness, bool turbo) const NOEXCEPT +{ + using namespace system; + if (!turbo) + { + stream::out::fast sink{ out }; + Writer writer{ sink }; + return get_wire_block(writer, link, witness) && writer; + } + + const auto txs = to_transactions(link); + if (txs.empty()) + return false; + + // Tx sizes, converted in place to offsets. + std::atomic_bool fault{}; + std::vector offsets(txs.size()); + const auto policy = poolstl::execution::par; + std::transform(policy, txs.cbegin(), txs.cend(), offsets.begin(), + [&](const tx_link& tx) NOEXCEPT + { + size_t bytes{}; + if (!get_tx_size(bytes, tx, witness)) + fault.store(true, std::memory_order_relaxed); + + return bytes; + }); + + const auto last = offsets.back(); + const auto prefix = chain::header::serialized_size() + + variable_size(txs.size()); + std::exclusive_scan(offsets.cbegin(), offsets.cend(), offsets.begin(), + prefix); + + const auto bytes = offsets.back() + last; + const auto scale = out.size() / bytes; + if (fault.load(std::memory_order_relaxed) || is_zero(scale) || + (scale * bytes != out.size())) + return false; + + const auto region = [&](size_t offset, size_t size) NOEXCEPT + { + const auto begin = std::next(out.data(), offset * scale); + return data_slab{ begin, std::next(begin, size * scale) }; + }; + + stream::out::fast head_sink{ region(zero, prefix) }; + Writer head{ head_sink }; + if (!get_wire_header(head, link)) + return false; + + head.write_variable(txs.size()); + if (!head) + return false; + + std::for_each(policy, offsets.cbegin(), offsets.cend(), + [&](const size_t& offset) NOEXCEPT + { + const auto index = to_unsigned(std::distance( + std::as_const(offsets).data(), &offset)); + const auto next = add1(index); + const auto end = next == offsets.size() ? bytes : offsets.at(next); + + stream::out::fast sink{ region(offset, end - offset) }; + Writer writer{ sink }; + if (!get_wire_tx(writer, txs.at(index), witness) || !writer) + fault.store(true, std::memory_order_relaxed); + }); + + return !fault.load(std::memory_order_relaxed); +} + } // namespace database } // namespace libbitcoin diff --git a/include/bitcoin/database/query.hpp b/include/bitcoin/database/query.hpp index 5c40cfb06..5e4eb2f11 100644 --- a/include/bitcoin/database/query.hpp +++ b/include/bitcoin/database/query.hpp @@ -507,7 +507,13 @@ class query data_chunk get_wire_header(const header_link& link) const NOEXCEPT; data_chunk get_wire_tx(const tx_link& link, bool witness) const NOEXCEPT; - data_chunk get_wire_block(const header_link& link, bool witness) const NOEXCEPT; + data_chunk get_wire_block(const header_link& link, bool witness, + bool turbo=false) const NOEXCEPT; + + /// Write to a target sized to the encoding, in parallel with turbo. + template + bool get_wire_block(const system::data_slab& out, const header_link& link, + bool witness, bool turbo=false) const NOEXCEPT; /// Objects. /// ----------------------------------------------------------------------- diff --git a/test/query/archive/wire_reader.cpp b/test/query/archive/wire_reader.cpp index a26df123e..84d4949b5 100644 --- a/test/query/archive/wire_reader.cpp +++ b/test/query/archive/wire_reader.cpp @@ -216,4 +216,75 @@ BOOST_AUTO_TEST_CASE(query_wire_reader__get_wire_block__witness_false__expected) BOOST_CHECK(!store.close(test::events_handler)); } +BOOST_AUTO_TEST_CASE(query_wire_reader__get_wire_block__turbo_witness_true__expected) +{ + using namespace system; + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::store_t store{ settings }; + test::query_t query{ store }; + BOOST_CHECK(!store.create(test::events_handler)); + BOOST_CHECK(test::setup_three_block_witness_store(query)); + BOOST_CHECK_EQUAL(query.get_wire_block(0, true, true), test::genesis.to_data(true)); + BOOST_CHECK_EQUAL(query.get_wire_block(1, true, true), test::block1a.to_data(true)); + BOOST_CHECK_EQUAL(query.get_wire_block(2, true, true), test::block2a.to_data(true)); + BOOST_CHECK(!store.close(test::events_handler)); +} + +BOOST_AUTO_TEST_CASE(query_wire_reader__get_wire_block__turbo_witness_false__expected) +{ + using namespace system; + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::store_t store{ settings }; + test::query_t query{ store }; + BOOST_CHECK(!store.create(test::events_handler)); + BOOST_CHECK(test::setup_three_block_witness_store(query)); + BOOST_CHECK_EQUAL(query.get_wire_block(0, false, true), test::genesis.to_data(false)); + BOOST_CHECK_EQUAL(query.get_wire_block(1, false, true), test::block1a.to_data(false)); + BOOST_CHECK_EQUAL(query.get_wire_block(2, false, true), test::block2a.to_data(false)); + BOOST_CHECK(!store.close(test::events_handler)); +} + +BOOST_AUTO_TEST_CASE(query_wire_reader__get_wire_block__base16_target__expected) +{ + using namespace system; + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::store_t store{ settings }; + test::query_t query{ store }; + BOOST_CHECK(!store.create(test::events_handler)); + BOOST_CHECK(test::setup_three_block_witness_store(query)); + + size_t size{}; + BOOST_CHECK(query.get_block_size(size, 2, true)); + + std::string sequential(two * size, '\0'); + BOOST_CHECK(query.get_wire_block(sequential, 2, true, false)); + BOOST_CHECK_EQUAL(sequential, encode_base16(test::block2a.to_data(true))); + + std::string parallel(two * size, '\0'); + BOOST_CHECK(query.get_wire_block(parallel, 2, true, true)); + BOOST_CHECK_EQUAL(parallel, encode_base16(test::block2a.to_data(true))); + BOOST_CHECK(!store.close(test::events_handler)); +} + +BOOST_AUTO_TEST_CASE(query_wire_reader__get_wire_block__turbo_missized_target__false) +{ + using namespace system; + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::store_t store{ settings }; + test::query_t query{ store }; + BOOST_CHECK(!store.create(test::events_handler)); + BOOST_CHECK(test::setup_three_block_witness_store(query)); + + size_t size{}; + BOOST_CHECK(query.get_block_size(size, 2, true)); + + data_chunk out(add1(size)); + BOOST_CHECK(!query.get_wire_block(out, 2, true, true)); + BOOST_CHECK(!store.close(test::events_handler)); +} + BOOST_AUTO_TEST_SUITE_END() From 715af3d32fb7e175e3f28499be24cbfb348dacd4 Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Sun, 18 Oct 2026 03:39:42 -0400 Subject: [PATCH 2/6] Scan silent rows of confirmed heights only. --- .../database/impl/query/batch/silent.ipp | 131 ++++++++++++++++++ include/bitcoin/database/query.hpp | 8 ++ test/query/batch/silent.cpp | 66 +++++++++ 3 files changed, 205 insertions(+) diff --git a/include/bitcoin/database/impl/query/batch/silent.ipp b/include/bitcoin/database/impl/query/batch/silent.ipp index 3cfcdc89e..7882e0fe2 100644 --- a/include/bitcoin/database/impl/query/batch/silent.ipp +++ b/include/bitcoin/database/impl/query/batch/silent.ipp @@ -76,6 +76,137 @@ bool CLASS::scan_silent(const stopper& cancel, return !cancel; } +// Rows are in validation order, so the rows of confirmed blocks are selected +// by tx link from the range and only the selected rows are scanned. A block's +// txs are a few runs of consecutive links, so membership is a search of runs. +TEMPLATE +bool CLASS::scan_silent(const stopper& cancel, + const system::scan::batch::receiver& keys, size_t first, size_t last, + size_t start, size_t stop, const silent_handler& callback) NOEXCEPT +{ + using namespace system; + using link_t = tx_link::integer; + using run_t = std::pair; + + std_vector runs{}; + const auto top = std::min(stop, get_top_confirmed()); + for (auto height = start; height <= top; ++height) + { + if (cancel) + return false; + + for (const auto tx: to_transactions(to_confirmed(height))) + { + if (!runs.empty() && add1(runs.back().second) == tx) + runs.back().second = tx; + else + runs.emplace_back(tx, tx); + } + } + + std::sort(runs.begin(), runs.end()); + std_vector merged{}; + for (const auto& [low, high]: runs) + { + if (!merged.empty() && low <= add1(merged.back().second)) + merged.back().second = std::max(merged.back().second, high); + else + merged.emplace_back(low, high); + } + + const auto latch = is_pooling() ? get_latch_tx() : max_size_t; + const auto selected = [&](link_t link) NOEXCEPT + { + if (link >= latch) + return true; + + const auto it = std::upper_bound(merged.cbegin(), merged.cend(), link, + [](link_t value, const run_t& run) NOEXCEPT + { + return value < run.first; + }); + + return it != merged.cbegin() && link <= std::prev(it)->second; + }; + + const auto correlate_ptr = store_.scan.correlate.get_memory(); + const auto point_ptr = store_.scan.point.get_memory(); + const auto outputs_ptr = store_.scan.outputs.get_memory(); + const auto prefix_ptr = store_.scan_prefix.get_memory(); + + using correlate_t = table::scan_correlate::span; + using point_t = table::scan_point::span; + using outputs_t = system::scan::batch::outputs_t; + using prefix_t = const system::scan::batch::prefix; + + const auto correlate = pointer_cast( + correlate_ptr.data()); + const auto point = pointer_cast(point_ptr.data()); + const auto outputs = pointer_cast(outputs_ptr.data()); + const auto prefix = pointer_cast(prefix_ptr.data()); + + BC_ASSERT(first <= last && last <= store_.scan.count()); + constexpr size_t chunk = power2(16_size); + const auto policy = poolstl::execution::par_if(store_.turbo()); + std_vector> rows(ceilinged_divide(last - first, chunk)); + std_vector chunks(rows.size()); + std::iota(chunks.begin(), chunks.end(), zero); + std::for_each(policy, chunks.cbegin(), chunks.cend(), + [&](size_t index) NOEXCEPT + { + const auto begin = first + index * chunk; + const auto end = std::min(last, begin + chunk); + for (auto row = begin; row < end && !cancel; ++row) + if (selected(from_little_endian(*std::next(correlate, + row)))) + rows.at(index).push_back(row); + }); + + if (cancel) + return false; + + size_t count{}; + std_vector offsets(rows.size()); + for (size_t index{}; index < rows.size(); ++index) + { + offsets.at(index) = count; + count += rows.at(index).size(); + } + + std_vector correlates(count); + std_vector points(count); + std_vector outs(count); + std::for_each(policy, chunks.cbegin(), chunks.cend(), + [&](size_t index) NOEXCEPT + { + auto at = offsets.at(index); + for (const auto row: rows.at(index)) + { + correlates.at(at) = *std::next(correlate, row); + points.at(at) = *std::next(point, row); + outs.at(at) = *std::next(outputs, row); + ++at; + } + }); + + const auto prefixes = possible_narrow_cast( + store_.scan_prefix.count().value); + + const system::scan::batch batch + { + .correlates = { correlates.data(), count }, + .points = { points.data(), count }, + .outputs = { outs.data(), count }, + .prefixes = { prefix, prefixes } + }; + + // False return only implies canceled. + // Callbacks invoked on caller thread if turbo is false. + const auto minimum = store_.gpu_scan_minimum(); + scan::batch::scan(cancel, batch, keys, minimum, callback, store_.turbo()); + return !cancel; +} + // Rows are allocated zero-filled and a nonzero correlate publishes the row. TEMPLATE size_t CLASS::get_silent_frontier(size_t first) const NOEXCEPT diff --git a/include/bitcoin/database/query.hpp b/include/bitcoin/database/query.hpp index 5e4eb2f11..bed1b6d9e 100644 --- a/include/bitcoin/database/query.hpp +++ b/include/bitcoin/database/query.hpp @@ -795,6 +795,14 @@ class query const system::scan::batch::receiver& keys, size_t first, size_t last, const silent_handler& callback) NOEXCEPT; + /// Invoke callback for each matched tx of rows [first, last) that is + /// confirmed at heights [start, stop] or linked at or above the pool + /// latch, false implies cancel. + bool scan_silent(const stopper& cancel, + const system::scan::batch::receiver& keys, size_t first, + size_t last, size_t start, size_t stop, + const silent_handler& callback) NOEXCEPT; + /// The first silent row at or above first that is not fully written. size_t get_silent_frontier(size_t first) const NOEXCEPT; diff --git a/test/query/batch/silent.cpp b/test/query/batch/silent.cpp index e6efdac1b..918cc1cf9 100644 --- a/test/query/batch/silent.cpp +++ b/test/query/batch/silent.cpp @@ -705,6 +705,72 @@ BOOST_AUTO_TEST_CASE(query_batch_silent__scan_silent__range__expected) BOOST_REQUIRE_EQUAL(links, std::vector{ 43 }); } +BOOST_AUTO_TEST_CASE(query_batch_silent__scan_silent__confirmed_heights__selected) +{ + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::chunk_store store{ settings }; + test::query_accessor query{ store }; + BOOST_REQUIRE(!store.create(test::events_handler)); + BOOST_REQUIRE(query.initialize(test::genesis)); + BOOST_REQUIRE(query.set_silent(42, simple_send())); + BOOST_REQUIRE(query.set_silent(0, simple_send())); + + std::vector links{}; + const auto handler = [&](const code&, tx_link_t tx, const ec_compressed&) NOEXCEPT + { + links.push_back(tx); + }; + + const stopper cancel{}; + BOOST_REQUIRE(query.scan_silent(cancel, get_keys(), 0, 2, 0, 0, handler)); + BOOST_REQUIRE_EQUAL(links, std::vector{ 0 }); +} + +BOOST_AUTO_TEST_CASE(query_batch_silent__scan_silent__heights_above_top__none) +{ + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::chunk_store store{ settings }; + test::query_accessor query{ store }; + BOOST_REQUIRE(!store.create(test::events_handler)); + BOOST_REQUIRE(query.initialize(test::genesis)); + BOOST_REQUIRE(query.set_silent(0, simple_send())); + + size_t calls{}; + const auto handler = [&](const code&, tx_link_t, const ec_compressed&) NOEXCEPT + { + ++calls; + }; + + const stopper cancel{}; + BOOST_REQUIRE(query.scan_silent(cancel, get_keys(), 0, 1, 1, 9, handler)); + BOOST_REQUIRE_EQUAL(calls, 0u); +} + +BOOST_AUTO_TEST_CASE(query_batch_silent__scan_silent__pooling__latched_selected) +{ + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::chunk_store store{ settings }; + test::query_accessor query{ store }; + BOOST_REQUIRE(!store.create(test::events_handler)); + BOOST_REQUIRE(query.initialize(test::genesis)); + query.set_pooling(); + BOOST_REQUIRE(query.set_silent(0, simple_send())); + BOOST_REQUIRE(query.set_silent(42, simple_send())); + + std::vector links{}; + const auto handler = [&](const code&, tx_link_t tx, const ec_compressed&) NOEXCEPT + { + links.push_back(tx); + }; + + const stopper cancel{}; + BOOST_REQUIRE(query.scan_silent(cancel, get_keys(), 0, 2, 1, 1, handler)); + BOOST_REQUIRE_EQUAL(links, std::vector{ 42 }); +} + // get_silent_frontier // ---------------------------------------------------------------------------- From a7fa0d9b5aa8c7d3389aa95e6f65befc3b9b7101 Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Sun, 18 Oct 2026 03:45:23 -0400 Subject: [PATCH 3/6] Read wire txs under one accessor per table. --- .../impl/query/archive/wire_reader.ipp | 103 +++++++++++++----- include/bitcoin/database/query.hpp | 25 +++++ 2 files changed, 103 insertions(+), 25 deletions(-) diff --git a/include/bitcoin/database/impl/query/archive/wire_reader.ipp b/include/bitcoin/database/impl/query/archive/wire_reader.ipp index e19929074..eeaf8e473 100644 --- a/include/bitcoin/database/impl/query/archive/wire_reader.ipp +++ b/include/bitcoin/database/impl/query/archive/wire_reader.ipp @@ -117,50 +117,100 @@ bool CLASS::get_wire_header(bytewriter& sink, TEMPLATE bool CLASS::get_wire_input(bytewriter& sink, const ins_link& link) const NOEXCEPT +{ + return get_wire_input(sink, store_.ins.get_memory(), + store_.ins.sequence.get_memory(), store_.input.get_memory(), link); +} + +TEMPLATE +bool CLASS::get_wire_output(bytewriter& sink, + const output_link& link) const NOEXCEPT +{ + return get_wire_output(sink, store_.output.get_memory(), link); +} + +TEMPLATE +bool CLASS::get_wire_witness(bytewriter& sink, + const ins_link& link) const NOEXCEPT +{ + return get_wire_witness(sink, store_.ins.sequence.get_memory(), + store_.input.get_memory(), link); +} + +TEMPLATE +bool CLASS::get_wire_tx(bytewriter& sink, const tx_link& link, + bool witness) const NOEXCEPT +{ + return get_wire_tx(sink, get_wire_accessors(), link, witness); +} + +// protected +TEMPLATE +CLASS::wire_accessors CLASS::get_wire_accessors() const NOEXCEPT +{ + return + { + store_.tx.get_memory(), + store_.outs.puts.get_memory(), + store_.ins.get_memory(), + store_.ins.sequence.get_memory(), + store_.input.get_memory(), + store_.output.get_memory() + }; +} + +// protected +TEMPLATE +bool CLASS::get_wire_input(bytewriter& sink, const memory& ins, + const memory& sequence, const memory& input, + const ins_link& link) const NOEXCEPT { table::ins_point::wire_point point{ {}, sink }; - if (!store_.ins.get(link, point)) + if (!store_.ins.get(ins, link, point)) return false; - table::ins_sequence::get_input ins{}; + table::ins_sequence::get_input in{}; table::input::wire_script script{ {}, sink }; - if (!store_.ins.sequence.get(link, ins) || - !store_.input.get(ins.input_fk, script)) + if (!store_.ins.sequence.get(sequence, link, in) || + !store_.input.get(input, in.input_fk, script)) return false; - sink.write_4_bytes_little_endian(ins.sequence); + sink.write_4_bytes_little_endian(in.sequence); return true; } +// protected TEMPLATE -bool CLASS::get_wire_output(bytewriter& sink, +bool CLASS::get_wire_output(bytewriter& sink, const memory& output, const output_link& link) const NOEXCEPT { table::output::wire_script out{ {}, sink }; - return store_.output.get(link, out); + return store_.output.get(output, link, out); } +// protected TEMPLATE -bool CLASS::get_wire_witness(bytewriter& sink, - const ins_link& link) const NOEXCEPT +bool CLASS::get_wire_witness(bytewriter& sink, const memory& sequence, + const memory& input, const ins_link& link) const NOEXCEPT { - table::ins_sequence::get_input ins{}; + table::ins_sequence::get_input in{}; table::input::wire_witness wire{ {}, sink }; - return store_.ins.sequence.get(link, ins) - && store_.input.get(ins.input_fk, wire); + return store_.ins.sequence.get(sequence, link, in) + && store_.input.get(input, in.input_fk, wire); } +// protected TEMPLATE -bool CLASS::get_wire_tx(bytewriter& sink, const tx_link& link, - bool witness) const NOEXCEPT +bool CLASS::get_wire_tx(bytewriter& sink, const wire_accessors& ptrs, + const tx_link& link, bool witness) const NOEXCEPT { table::transaction::record tx{}; - if (!store_.tx.get(link, tx)) + if (!store_.tx.get(ptrs.tx, link, tx)) return false; table::outs::record outs{}; outs.out_fks.resize(tx.outs_count); - if (!store_.outs.puts.get(tx.outs_fk, outs)) + if (!store_.outs.puts.get(ptrs.puts, tx.outs_fk, outs)) return false; // Point links are contiguous (computed). @@ -179,18 +229,18 @@ bool CLASS::get_wire_tx(bytewriter& sink, const tx_link& link, sink.write_variable(ins_count); for (auto fk = ins_begin; fk < ins_final; ++fk) - if (!get_wire_input(sink, fk)) + if (!get_wire_input(sink, ptrs.ins, ptrs.sequence, ptrs.input, fk)) return false; sink.write_variable(outs.out_fks.size()); for (const auto& fk: outs.out_fks) - if (!get_wire_output(sink, fk)) + if (!get_wire_output(sink, ptrs.output, fk)) return false; if (witnessed) { for (auto fk = ins_begin; fk < ins_final; ++fk) - if (!get_wire_witness(sink, fk)) + if (!get_wire_witness(sink, ptrs.sequence, ptrs.input, fk)) return false; } @@ -210,8 +260,9 @@ bool CLASS::get_wire_block(bytewriter& sink, const header_link& link, return false; sink.write_variable(txs.size()); + const auto ptrs = get_wire_accessors(); for (const auto& tx_link: txs) - if (!get_wire_tx(sink, tx_link, witness)) + if (!get_wire_tx(sink, ptrs, tx_link, witness)) return false; return true; @@ -294,18 +345,20 @@ bool CLASS::get_wire_block(const system::data_slab& out, if (txs.empty()) return false; - // Tx sizes, converted in place to offsets. + // Tx sizes, converted in place to offsets. Tasks read through the held + // accessors, as a task taking a held lock could wait behind a remap. std::atomic_bool fault{}; std::vector offsets(txs.size()); + const auto ptrs = get_wire_accessors(); const auto policy = poolstl::execution::par; std::transform(policy, txs.cbegin(), txs.cend(), offsets.begin(), [&](const tx_link& tx) NOEXCEPT { - size_t bytes{}; - if (!get_tx_size(bytes, tx, witness)) + table::transaction::get_sizes sizes{}; + if (!store_.tx.get(ptrs.tx, tx, sizes)) fault.store(true, std::memory_order_relaxed); - return bytes; + return witness ? sizes.heavy : sizes.light; }); const auto last = offsets.back(); @@ -345,7 +398,7 @@ bool CLASS::get_wire_block(const system::data_slab& out, stream::out::fast sink{ region(offset, end - offset) }; Writer writer{ sink }; - if (!get_wire_tx(writer, txs.at(index), witness) || !writer) + if (!get_wire_tx(writer, ptrs, txs.at(index), witness) || !writer) fault.store(true, std::memory_order_relaxed); }); diff --git a/include/bitcoin/database/query.hpp b/include/bitcoin/database/query.hpp index bed1b6d9e..a18e6b4e1 100644 --- a/include/bitcoin/database/query.hpp +++ b/include/bitcoin/database/query.hpp @@ -1286,6 +1286,31 @@ class query const allocation& fks,const transaction_view& tx, bool bypass, bool prune) NOEXCEPT; + /// Wire reading (all accessors held across the txs read). + /// ----------------------------------------------------------------------- + + /// Per-table accessors, each holds a shared lock on its storage remap. + struct wire_accessors + { + memory tx; + memory puts; + memory ins; + memory sequence; + memory input; + memory output; + }; + + wire_accessors get_wire_accessors() const NOEXCEPT; + bool get_wire_input(bytewriter& sink, const memory& ins, + const memory& sequence, const memory& input, + const ins_link& link) const NOEXCEPT; + bool get_wire_output(bytewriter& sink, const memory& output, + const output_link& link) const NOEXCEPT; + bool get_wire_witness(bytewriter& sink, const memory& sequence, + const memory& input, const ins_link& link) const NOEXCEPT; + bool get_wire_tx(bytewriter& sink, const wire_accessors& ptrs, + const tx_link& link, bool witness) const NOEXCEPT; + /// History. /// ----------------------------------------------------------------------- From ca60ac2af0cf3b7fa13b4082f65af267d22acacf Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Sun, 18 Oct 2026 07:01:54 -0400 Subject: [PATCH 4/6] Copy wire scripts and outputs without allocation. --- .../impl/query/archive/wire_reader.ipp | 19 ++++++++++--------- .../database/tables/archives/header.hpp | 4 ++-- .../database/tables/archives/input.hpp | 4 ++-- .../database/tables/archives/output.hpp | 2 +- 4 files changed, 15 insertions(+), 14 deletions(-) diff --git a/include/bitcoin/database/impl/query/archive/wire_reader.ipp b/include/bitcoin/database/impl/query/archive/wire_reader.ipp index eeaf8e473..77677968a 100644 --- a/include/bitcoin/database/impl/query/archive/wire_reader.ipp +++ b/include/bitcoin/database/impl/query/archive/wire_reader.ipp @@ -208,12 +208,7 @@ bool CLASS::get_wire_tx(bytewriter& sink, const wire_accessors& ptrs, if (!store_.tx.get(ptrs.tx, link, tx)) return false; - table::outs::record outs{}; - outs.out_fks.resize(tx.outs_count); - if (!store_.outs.puts.get(ptrs.puts, tx.outs_fk, outs)) - return false; - - // Point links are contiguous (computed). + // Point and put links are contiguous (computed). const auto ins_begin = tx.point_fk; const auto ins_count = tx.ins_count; const auto ins_final = ins_begin + ins_count; @@ -232,10 +227,16 @@ bool CLASS::get_wire_tx(bytewriter& sink, const wire_accessors& ptrs, if (!get_wire_input(sink, ptrs.ins, ptrs.sequence, ptrs.input, fk)) return false; - sink.write_variable(outs.out_fks.size()); - for (const auto& fk: outs.out_fks) - if (!get_wire_output(sink, ptrs.output, fk)) + const auto outs_begin = tx.outs_fk; + const auto outs_final = outs_begin + tx.outs_count; + sink.write_variable(tx.outs_count); + for (auto fk = outs_begin; fk < outs_final; ++fk) + { + table::outs::get_output put{}; + if (!store_.outs.puts.get(ptrs.puts, fk, put) || + !get_wire_output(sink, ptrs.output, put.out_fk)) return false; + } if (witnessed) { diff --git a/include/bitcoin/database/tables/archives/header.hpp b/include/bitcoin/database/tables/archives/header.hpp index d757899b9..5ba723ef6 100644 --- a/include/bitcoin/database/tables/archives/header.hpp +++ b/include/bitcoin/database/tables/archives/header.hpp @@ -374,12 +374,12 @@ struct header const auto time_bits_nonce_size = 3u * sizeof(uint32_t); const auto version_size = sizeof(uint32_t); source.skip_bytes(skip_to_version); - sink.write_bytes(source.read_bytes(version_size)); + sink.write_bytes(source, version_size); sink.write_bytes(parent_hash); source.skip_bytes(time_bits_nonce_size); sink.write_bytes(source.read_hash()); source.rewind_bytes(time_bits_nonce_size + schema::hash); - sink.write_bytes(source.read_bytes(time_bits_nonce_size)); + sink.write_bytes(source, time_bits_nonce_size); return source; } diff --git a/include/bitcoin/database/tables/archives/input.hpp b/include/bitcoin/database/tables/archives/input.hpp index 1b020f15a..6b67e1f26 100644 --- a/include/bitcoin/database/tables/archives/input.hpp +++ b/include/bitcoin/database/tables/archives/input.hpp @@ -264,7 +264,7 @@ struct input // script (prefixed) const auto length = source.read_size(); sink.write_variable(length); - sink.write_bytes(source.read_bytes(length)); + sink.write_bytes(source, length); return source; } @@ -288,7 +288,7 @@ struct input { const auto length = source.read_size(); sink.write_variable(length); - sink.write_bytes(source.read_bytes(length)); + sink.write_bytes(source, length); } return source; diff --git a/include/bitcoin/database/tables/archives/output.hpp b/include/bitcoin/database/tables/archives/output.hpp index cb4e7f12c..e78b9784b 100644 --- a/include/bitcoin/database/tables/archives/output.hpp +++ b/include/bitcoin/database/tables/archives/output.hpp @@ -418,7 +418,7 @@ struct output // script (prefixed) const auto length = source.read_size(); sink.write_variable(length); - sink.write_bytes(source.read_bytes(length)); + sink.write_bytes(source, length); return source; } From 302e26c6e6322299c541908f5760bb6a28c0d03e Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Sun, 18 Oct 2026 07:01:55 -0400 Subject: [PATCH 5/6] Write turbo wire blocks in ranges of txs. --- .../impl/query/archive/wire_reader.ipp | 67 ++++++++++++------- test/query/archive/wire_reader.cpp | 41 ++++++++++++ 2 files changed, 84 insertions(+), 24 deletions(-) diff --git a/include/bitcoin/database/impl/query/archive/wire_reader.ipp b/include/bitcoin/database/impl/query/archive/wire_reader.ipp index 77677968a..d45ef5c38 100644 --- a/include/bitcoin/database/impl/query/archive/wire_reader.ipp +++ b/include/bitcoin/database/impl/query/archive/wire_reader.ipp @@ -348,19 +348,16 @@ bool CLASS::get_wire_block(const system::data_slab& out, // Tx sizes, converted in place to offsets. Tasks read through the held // accessors, as a task taking a held lock could wait behind a remap. - std::atomic_bool fault{}; - std::vector offsets(txs.size()); const auto ptrs = get_wire_accessors(); - const auto policy = poolstl::execution::par; - std::transform(policy, txs.cbegin(), txs.cend(), offsets.begin(), - [&](const tx_link& tx) NOEXCEPT - { - table::transaction::get_sizes sizes{}; - if (!store_.tx.get(ptrs.tx, tx, sizes)) - fault.store(true, std::memory_order_relaxed); + std::vector offsets(txs.size()); + for (size_t index{}; index < txs.size(); ++index) + { + table::transaction::get_sizes sizes{}; + if (!store_.tx.get(ptrs.tx, txs.at(index), sizes)) + return false; - return witness ? sizes.heavy : sizes.light; - }); + offsets.at(index) = witness ? sizes.heavy : sizes.light; + } const auto last = offsets.back(); const auto prefix = chain::header::serialized_size() + @@ -370,10 +367,20 @@ bool CLASS::get_wire_block(const system::data_slab& out, const auto bytes = offsets.back() + last; const auto scale = out.size() / bytes; - if (fault.load(std::memory_order_relaxed) || is_zero(scale) || - (scale * bytes != out.size())) + if (is_zero(scale) || (scale * bytes != out.size())) return false; + // Txs are written in ranges, one task each, as each task costs a wakeup, + // and a block of two ranges or fewer is written without tasks. + constexpr size_t range_txs = 256; + const auto ranges = ceilinged_divide(txs.size(), range_txs); + if (ranges <= two) + { + stream::out::fast sink{ out }; + Writer writer{ sink }; + return get_wire_block(writer, link, witness) && writer; + } + const auto region = [&](size_t offset, size_t size) NOEXCEPT { const auto begin = std::next(out.data(), offset * scale); @@ -389,18 +396,30 @@ bool CLASS::get_wire_block(const system::data_slab& out, if (!head) return false; - std::for_each(policy, offsets.cbegin(), offsets.cend(), - [&](const size_t& offset) NOEXCEPT + std::atomic_bool fault{}; + std::vector starts(ranges); + std::iota(starts.begin(), starts.end(), zero); + std::for_each(poolstl::execution::par, starts.cbegin(), starts.cend(), + [&](size_t range) NOEXCEPT { - const auto index = to_unsigned(std::distance( - std::as_const(offsets).data(), &offset)); - const auto next = add1(index); - const auto end = next == offsets.size() ? bytes : offsets.at(next); - - stream::out::fast sink{ region(offset, end - offset) }; - Writer writer{ sink }; - if (!get_wire_tx(writer, ptrs, txs.at(index), witness) || !writer) - fault.store(true, std::memory_order_relaxed); + const auto first = range * range_txs; + const auto stop = std::min(txs.size(), first + range_txs); + for (auto index = first; index < stop; ++index) + { + const auto offset = offsets.at(index); + const auto next = add1(index); + const auto end = next == offsets.size() ? bytes : + offsets.at(next); + + stream::out::fast sink{ region(offset, end - offset) }; + Writer writer{ sink }; + if (!get_wire_tx(writer, ptrs, txs.at(index), witness) || + !writer) + { + fault.store(true, std::memory_order_relaxed); + return; + } + } }); return !fault.load(std::memory_order_relaxed); diff --git a/test/query/archive/wire_reader.cpp b/test/query/archive/wire_reader.cpp index 84d4949b5..1b2be42c2 100644 --- a/test/query/archive/wire_reader.cpp +++ b/test/query/archive/wire_reader.cpp @@ -22,6 +22,21 @@ BOOST_FIXTURE_TEST_SUITE(query_wire_reader_reader_tests, test::directory_setup_fixture) +// A block of enough txs that the turbo writer uses more than one range. +static system::chain::block many_tx_block() NOEXCEPT +{ + using namespace system; + const auto version = test::tx4.version(); + const auto inputs = test::tx4.inputs_ptr(); + const auto outputs = test::tx4.outputs_ptr(); + chain::transactions txs{}; + txs.reserve(1100); + for (uint32_t locktime{}; locktime < 1100; ++locktime) + txs.emplace_back(version, inputs, outputs, locktime); + + return { test::block1a.header(), txs }; +} + // is_witness_match BOOST_AUTO_TEST_CASE(query_wire_reader__is_witness_match__non_segregated__true) @@ -269,6 +284,32 @@ BOOST_AUTO_TEST_CASE(query_wire_reader__get_wire_block__base16_target__expected) BOOST_CHECK(!store.close(test::events_handler)); } +BOOST_AUTO_TEST_CASE(query_wire_reader__get_wire_block__turbo_ranges__expected) +{ + using namespace system; + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::store_t store{ settings }; + test::query_t query{ store }; + BOOST_CHECK(!store.create(test::events_handler)); + BOOST_CHECK(query.initialize(test::genesis)); + + const auto block = many_tx_block(); + BOOST_CHECK(query.set(block, database::context{ 0, 1, 0 }, {}, false, false)); + const auto link = query.to_header(block.hash()); + + BOOST_CHECK_EQUAL(query.get_wire_block(link, true, true), block.to_data(true)); + BOOST_CHECK_EQUAL(query.get_wire_block(link, false, true), block.to_data(false)); + + size_t size{}; + BOOST_CHECK(query.get_block_size(size, link, true)); + + std::string hex(two * size, '\0'); + BOOST_CHECK(query.get_wire_block(hex, link, true, true)); + BOOST_CHECK_EQUAL(hex, encode_base16(block.to_data(true))); + BOOST_CHECK(!store.close(test::events_handler)); +} + BOOST_AUTO_TEST_CASE(query_wire_reader__get_wire_block__turbo_missized_target__false) { using namespace system; From 86c94e5e78decb4ebc09aee335650212820a318b Mon Sep 17 00:00:00 2001 From: Eric Voskuil Date: Sun, 18 Oct 2026 12:17:35 -0400 Subject: [PATCH 6/6] Query silent payment matches of a cut. --- .../database/impl/query/batch/silent.ipp | 73 +++++++++++++++++++ include/bitcoin/database/query.hpp | 17 +++++ include/bitcoin/database/types/type.hpp | 2 + test/query/batch/silent.cpp | 72 ++++++++++++++++++ 4 files changed, 164 insertions(+) diff --git a/include/bitcoin/database/impl/query/batch/silent.ipp b/include/bitcoin/database/impl/query/batch/silent.ipp index 7882e0fe2..11a11efb1 100644 --- a/include/bitcoin/database/impl/query/batch/silent.ipp +++ b/include/bitcoin/database/impl/query/batch/silent.ipp @@ -207,6 +207,67 @@ bool CLASS::scan_silent(const stopper& cancel, return !cancel; } +// Confirmation is strong association, otherwise height is zero (unconfirmed). +// A tx archived only on a stale branch reports as unconfirmed, and one also +// archived there before the store began pooling may report twice (benign). +TEMPLATE +bool CLASS::get_silent_matches(silent_matches& out, const stopper& cancel, + const system::scan::batch::receiver& keys, size_t first, size_t last, + size_t start, size_t stop, bool time) NOEXCEPT +{ + using namespace system; + using link_t = scan::batch::tx_link_t; + constexpr size_t selection_blocks = 10'000; + + std::mutex mutex{}; + std::map found{}; + const auto handler = [&](const code&, link_t link, + const ec_compressed& tweak) NOEXCEPT + { + std::unique_lock lock{ mutex }; + found.emplace(link, tweak); + }; + + const auto top = get_top_confirmed(); + const auto select = !time && (start > top || + std::min(stop, top) - start < selection_blocks) && + first == get_silent_floor(start, false); + + if (!(select ? + scan_silent(cancel, keys, first, last, start, stop, handler) : + scan_silent(cancel, keys, first, last, handler))) + return false; + + out.clear(); + out.reserve(found.size()); + for (const auto& [link, tweak]: found) + { + size_t height{}; + if (!get_tx_height(height, link)) + out.emplace_back(zero, get_tx_key(link), tweak); + else if (!is_silent_cut(link, height, start, stop, time)) + out.emplace_back(height, get_tx_key(link), tweak); + } + + std::sort(out.begin(), out.end(), + [](const silent_match& left, const silent_match& right) NOEXCEPT + { + return sub1(std::get<0>(left)) < sub1(std::get<0>(right)); + }); + + return true; +} + +TEMPLATE +size_t CLASS::get_silent_floor(size_t start, bool time) const NOEXCEPT +{ + auto height = start; + if (time && !get_mtp_height(height, start)) + height = max_size_t; + + return get_scan_floor(height).value; +} + // Rows are allocated zero-filled and a nonzero correlate publishes the row. TEMPLATE size_t CLASS::get_silent_frontier(size_t first) const NOEXCEPT @@ -817,6 +878,18 @@ bool CLASS::is_silent_selected(const tx_link& link, return link >= coinbase || !is_pooled(link); } +TEMPLATE +bool CLASS::is_silent_cut(const tx_link& link, size_t height, size_t start, + size_t stop, bool time) const NOEXCEPT +{ + if (!time) + return height < start || height > stop; + + context ctx{}; + return !get_context(ctx, find_strong(link)) || ctx.mtp < start || + ctx.mtp > stop; +} + // Rows are read in chunks so the guard is not held across a long seek. TEMPLATE void CLASS::seek_scan_start(size_t& row, size_t& last, diff --git a/include/bitcoin/database/query.hpp b/include/bitcoin/database/query.hpp index a18e6b4e1..dc9a14c2c 100644 --- a/include/bitcoin/database/query.hpp +++ b/include/bitcoin/database/query.hpp @@ -803,6 +803,18 @@ class query size_t last, size_t start, size_t stop, const silent_handler& callback) NOEXCEPT; + /// Matches (height, tx hash, tweak) of rows [first, last) confirmed at + /// heights (or mtps if time) [start, stop], ordered by height, then + /// unconfirmed (zero height), false implies cancel. A short height cut + /// scanned from its floor scans only the rows of its confirmed blocks. + bool get_silent_matches(silent_matches& out, const stopper& cancel, + const system::scan::batch::receiver& keys, size_t first, + size_t last, size_t start, size_t stop, bool time) NOEXCEPT; + + /// The scan fk from which a scan covers all blocks at or above start + /// (height, or mtp if time). + size_t get_silent_floor(size_t start, bool time) const NOEXCEPT; + /// The first silent row at or above first that is not fully written. size_t get_silent_frontier(size_t first) const NOEXCEPT; @@ -1237,6 +1249,11 @@ class query bool is_silent_selected(const tx_link& link, const tx_link& coinbase) const NOEXCEPT; + /// The tx confirmed at height is outside heights (or mtps if time) + /// [start, stop]. + bool is_silent_cut(const tx_link& link, size_t height, size_t start, + size_t stop, bool time) const NOEXCEPT; + /// The wire prevouts of the spends of the block's selected txs, where /// silent selects the silent payment eligible txs and otherwise all. bool get_prevouts_(data_chunk& prevouts, std::vector& selected, diff --git a/include/bitcoin/database/types/type.hpp b/include/bitcoin/database/types/type.hpp index 12ebd7315..c4c1f9f0b 100644 --- a/include/bitcoin/database/types/type.hpp +++ b/include/bitcoin/database/types/type.hpp @@ -92,6 +92,8 @@ using threshold = std::pair; using thresholds = std::vector; using confirmables = std::unordered_map; using compact_matches = std::vector>; +using silent_match = std::tuple; +using silent_matches = std::vector; } // namespace database } // namespace libbitcoin diff --git a/test/query/batch/silent.cpp b/test/query/batch/silent.cpp index 918cc1cf9..1df54ccda 100644 --- a/test/query/batch/silent.cpp +++ b/test/query/batch/silent.cpp @@ -771,6 +771,78 @@ BOOST_AUTO_TEST_CASE(query_batch_silent__scan_silent__pooling__latched_selected) BOOST_REQUIRE_EQUAL(links, std::vector{ 42 }); } +// get_silent_matches + +BOOST_AUTO_TEST_CASE(query_batch_silent__get_silent_matches__short_cut_from_floor__confirmed_selected) +{ + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::chunk_store store{ settings }; + test::query_accessor query{ store }; + BOOST_REQUIRE(!store.create(test::events_handler)); + BOOST_REQUIRE(query.initialize(test::genesis)); + BOOST_REQUIRE(query.set_silent(42, simple_send())); + BOOST_REQUIRE(query.set_silent(0, simple_send())); + + silent_matches out{}; + const stopper cancel{}; + const auto genesis = test::genesis.transactions_ptr()->front()->hash(false); + BOOST_REQUIRE(query.get_silent_matches(out, cancel, get_keys(), 0, 2, 0, 0, false)); + BOOST_REQUIRE_EQUAL(out.size(), 1u); + BOOST_REQUIRE_EQUAL(std::get<1>(out.front()), genesis); +} + +BOOST_AUTO_TEST_CASE(query_batch_silent__get_silent_matches__above_floor__unconfirmed) +{ + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::chunk_store store{ settings }; + test::query_accessor query{ store }; + BOOST_REQUIRE(!store.create(test::events_handler)); + BOOST_REQUIRE(query.initialize(test::genesis)); + BOOST_REQUIRE(query.set_silent(0, simple_send())); + BOOST_REQUIRE(query.set_silent(42, simple_send())); + + silent_matches out{}; + const stopper cancel{}; + BOOST_REQUIRE(query.get_silent_matches(out, cancel, get_keys(), 1, 2, 0, 0, false)); + BOOST_REQUIRE_EQUAL(out.size(), 1u); + BOOST_REQUIRE_EQUAL(std::get<0>(out.front()), 0u); +} + +BOOST_AUTO_TEST_CASE(query_batch_silent__get_silent_matches__time_cut__expected) +{ + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::chunk_store store{ settings }; + test::query_accessor query{ store }; + BOOST_REQUIRE(!store.create(test::events_handler)); + BOOST_REQUIRE(query.initialize(test::genesis)); + BOOST_REQUIRE(query.set_silent(0, simple_send())); + + silent_matches included{}; + silent_matches excluded{}; + const stopper cancel{}; + BOOST_REQUIRE(query.get_silent_matches(included, cancel, get_keys(), 0, 1, 0, max_size_t, true)); + BOOST_REQUIRE(query.get_silent_matches(excluded, cancel, get_keys(), 0, 1, max_uint32, max_size_t, true)); + BOOST_REQUIRE_EQUAL(included.size(), 1u); + BOOST_REQUIRE(excluded.empty()); +} + +// get_silent_floor + +BOOST_AUTO_TEST_CASE(query_batch_silent__get_silent_floor__unmapped__zero) +{ + database::settings settings{}; + settings.path = TEST_DIRECTORY; + test::chunk_store store{ settings }; + test::query_accessor query{ store }; + BOOST_REQUIRE(!store.create(test::events_handler)); + BOOST_REQUIRE(query.initialize(test::genesis)); + BOOST_REQUIRE_EQUAL(query.get_silent_floor(0, false), 0u); + BOOST_REQUIRE_EQUAL(query.get_silent_floor(0, true), 0u); +} + // get_silent_frontier // ----------------------------------------------------------------------------