Parallelize set_strong in the confirmation path.

include/bitcoin/database/impl/primitives/linkage.ipp

@@ -73,6 +73,25 @@ inline CLASS CLASS::operator++(int) NOEXCEPT
     return self;
 }
 
+TEMPLATE
+template <typename Integer, if_unsigned_integer<Integer>>
+constexpr CLASS& CLASS::operator+=(Integer offset) NOEXCEPT
+{
+    // An actual row cannot be terminal, so the sum must be less than terminal.
+    BC_ASSERT(!system::is_add_overflow<integer>(value, offset));
+    BC_ASSERT(system::is_lesser(system::add<integer>(value, offset), terminal));
+    value = system::add<integer>(value, offset);
+    return *this;
+}
+
+TEMPLATE
+template <typename Integer, if_unsigned_integer<Integer>>
+constexpr CLASS CLASS::operator+(Integer offset) const NOEXCEPT
+{
+    auto self = *this;
+    return self += offset;
+}
+
 TEMPLATE
 constexpr CLASS::operator CLASS::integer() const NOEXCEPT
 {

include/bitcoin/database/impl/query/archive/chain_writer.ipp

@@ -225,8 +225,8 @@ code CLASS::set_code(const tx_link& tx_fk, const transaction& tx,
             // See outs::put_ref.
             // Calculate next corresponding output fk from serialized size.
             // (variable_size(value) + (value + script)) - (value - parent)
-            out_fk.value += (variable_size(output->value()) +
-                output->serialized_size() - value_parent);
+            out_fk += variable_size(output->value()) +
+                output->serialized_size() - value_parent;
         }
     }
 
@@ -409,7 +409,8 @@ code CLASS::set_code(const block& block, const header_link& key,
     constexpr auto positive = true;
 
     // Transactor assures cannot be restored without txs, as required to unset.
-    if (strong && !set_strong(key, txs, tx_fks, positive))
+    // Sequential write, as archival is already concurrent across blocks.
+    if (strong && !set_strong(key, txs, tx_fks, positive, false))
         return error::txs_confirm;
 
     // Header link is the key for the txs table.

include/bitcoin/database/impl/query/archive/wire_writer.ipp

@@ -164,9 +164,8 @@ code CLASS::set_code(std::vector<point>& twins, const accessors& ptrs,
                 table::address::record{ {}, out_fk }))
                 return error::tx_address_put;
 
-            out_fk.value += possible_narrow_cast<output_link::integer>(
-                tx_link::size + variable_size(value) + variable_size(bytes) +
-                bytes);
+            out_fk += tx_link::size + variable_size(value) +
+                variable_size(bytes) + bytes;
         }
 
         BC_ASSERT(osource);
@@ -217,7 +216,6 @@ code CLASS::set_code(const block_view& block, const header_link& key,
     using out_t = output_link::integer;
     using ins_t = ins_link::integer;
     using outs_t = outs_link::integer;
-    using address_t = address_link::integer;
 
     if (key.is_terminal())
         return error::txs_header;
@@ -324,11 +322,11 @@ code CLASS::set_code(const block_view& block, const header_link& key,
             tx.output_table_size();
 
         fks.tx_fk++;
-        fks.ins_fk.value  += possible_narrow_cast<ins_t>(tx.inputs());
-        fks.outs_fk.value += possible_narrow_cast<outs_t>(tx.outputs());
-        fks.in_fk.value   += possible_narrow_cast<in_t>(tx.input_table_size());
-        fks.out_fk.value  += possible_narrow_cast<out_t>(out_bytes);
-        fks.ad_fk.value   += possible_narrow_cast<address_t>(tx.outputs());
+        fks.ins_fk  += tx.inputs();
+        fks.outs_fk += tx.outputs();
+        fks.in_fk   += tx.input_table_size();
+        fks.out_fk  += out_bytes;
+        fks.ad_fk   += tx.outputs();
     }
 
     // Release all accessors (subsequent writes allocate).
@@ -349,7 +347,8 @@ code CLASS::set_code(const block_view& block, const header_link& key,
     constexpr auto positive = true;
 
     // Transactor assures cannot be restored without txs, as required to unset.
-    if (strong && !set_strong(key, txs, tx_fks, positive))
+    // Sequential write, as archival is already concurrent across blocks.
+    if (strong && !set_strong(key, txs, tx_fks, positive, false))
         return error::txs_confirm;
 
     // Header link is the key for the txs table.

include/bitcoin/database/impl/query/consensus/consensus_strong.ipp

@@ -19,6 +19,7 @@
 #ifndef LIBBITCOIN_DATABASE_QUERY_CONSENSUS_STRONG_IPP
 #define LIBBITCOIN_DATABASE_QUERY_CONSENSUS_STRONG_IPP
 
+#include <algorithm>
 #include <bitcoin/database/define.hpp>
 
 namespace libbitcoin {
@@ -107,27 +108,33 @@ header_link CLASS::find_strong(const hash_digest& tx_hash) const NOEXCEPT
 // protected
 TEMPLATE
 bool CLASS::set_strong(const header_link& link, size_t count,
-    const tx_link& first_fk, bool positive) NOEXCEPT
+    const tx_link& first_fk, bool positive, bool parallel) NOEXCEPT
 {
     using namespace system;
     using link_t = table::strong_tx::link;
     using element_t = table::strong_tx::record;
+    const auto policy = poolstl::execution::par_if(parallel);
 
     // Preallocate all strong_tx records for the block and reuse memory ptr.
+    // Terminal allocation must not be indexed (would wrap to valid records).
     const auto records = possible_narrow_cast<link_t::integer>(count);
-    auto record = store_.strong_tx.allocate(records);
-    const auto ptr = store_.strong_tx.get_memory();
-    const auto end = first_fk + count;
+    const auto record = store_.strong_tx.allocate(records);
+    if (record.is_terminal())
+        return false;
 
-    // Contiguous tx links.
-    for (auto fk = first_fk; fk < end; ++fk)
-        if (!store_.strong_tx.put(ptr, record++, fk, element_t
-            {
-                {},
-                table::strong_tx::merge(positive, link)
-            })) return false;
+    const auto ptr = store_.strong_tx.get_memory();
 
-    return true;
+    // All records of the block share one element value.
+    const element_t element{ {}, table::strong_tx::merge(positive, link) };
+
+    // Contiguous tx links to contiguous records.
+    return std::all_of(policy, poolstl::iota_iter<size_t>(zero),
+        poolstl::iota_iter<size_t>(count),
+        [&](size_t index) NOEXCEPT
+        {
+            return store_.strong_tx.put(ptr, record + index, first_fk + index,
+                element);
+        });
 }
 
 TEMPLATE
@@ -144,7 +151,7 @@ bool CLASS::set_strong(const header_link& link) NOEXCEPT
     const auto scope = get_transactor();
 
     // Clean allocation failure (e.g. disk full).
-    return set_strong(link, txs.number, txs.coinbase_fk , true);
+    return set_strong(link, txs.number, txs.coinbase_fk, true, true);
     // ========================================================================
 }
 
@@ -162,7 +169,7 @@ bool CLASS::set_unstrong(const header_link& link) NOEXCEPT
     const auto scope = get_transactor();
 
     // Clean allocation failure (e.g. disk full).
-    return set_strong(link, txs.number, txs.coinbase_fk, false);
+    return set_strong(link, txs.number, txs.coinbase_fk, false, true);
     // ========================================================================
 }
 

include/bitcoin/database/impl/query/height.ipp

@@ -85,7 +85,7 @@ bool CLASS::push_confirmed(const header_link& link, bool strong) NOEXCEPT
     const auto scope = get_transactor();
 
     // This reservation guard assumes no concurrent writes to the table.
-    if (strong && !set_strong(link, txs.number, txs.coinbase_fk, true))
+    if (strong && !set_strong(link, txs.number, txs.coinbase_fk, true, true))
         return false;
 
     const table::height::record confirmed{ {}, link };
@@ -110,7 +110,7 @@ bool CLASS::pop_confirmed() NOEXCEPT
     const auto scope = get_transactor();
 
     // Clean single allocation failure.
-    if (!set_strong(link, txs.number, txs.coinbase_fk, false))
+    if (!set_strong(link, txs.number, txs.coinbase_fk, false, true))
         return false;
 
     ///////////////////////////////////////////////////////////////////////////

include/bitcoin/database/primitives/arrayhead.hpp

@@ -96,7 +96,7 @@ class arrayhead
         using namespace system;
         BC_ASSERT(!is_multiply_overflow<size_t>(index, bucket_size));
         BC_ASSERT(!is_add_overflow<size_t>(bucket_size, index * bucket_size));
-        return possible_narrow_cast<size_t>(add1(index) * bucket_size);
+        return possible_narrow_cast<size_t>(add1<size_t>(index) * bucket_size);
     }
 
     // These are thread safe.

include/bitcoin/database/primitives/hashhead.hpp

@@ -131,7 +131,7 @@ class hashhead
         using namespace system;
         BC_ASSERT(!is_multiply_overflow<size_t>(index, cell_size));
         BC_ASSERT(!is_add_overflow<size_t>(cell_size, index * cell_size));
-        return possible_narrow_cast<size_t>(add1(index) * cell_size);
+        return possible_narrow_cast<size_t>(add1<size_t>(index) * cell_size);
     }
 
     // These are thread safe.

include/bitcoin/database/primitives/linkage.hpp

@@ -56,6 +56,12 @@ struct linkage
     inline self& operator++() NOEXCEPT;
     inline self operator++(int) NOEXCEPT;
 
+    /// Offset addition operators (records, or bytes for slab links).
+    template <typename Integer, if_unsigned_integer<Integer> = true>
+    constexpr self& operator+=(Integer offset) NOEXCEPT;
+    template <typename Integer, if_unsigned_integer<Integer> = true>
+    constexpr self operator+(Integer offset) const NOEXCEPT;
+
     /// Integral and array cast operators.
     constexpr operator integer() const NOEXCEPT;
     inline operator bytes() const NOEXCEPT;

include/bitcoin/database/primitives/nohead.hpp

@@ -84,7 +84,7 @@ class nohead
         using namespace system;
         BC_ASSERT(!is_multiply_overflow<size_t>(index, bucket_size));
         BC_ASSERT(!is_add_overflow<size_t>(bucket_size, index * bucket_size));
-        return possible_narrow_cast<size_t>(add1(index) * bucket_size);
+        return possible_narrow_cast<size_t>(add1<size_t>(index) * bucket_size);
     }
 
     // These are thread safe.

include/bitcoin/database/query.hpp

@@ -837,7 +837,7 @@ class query
 
     /// Support set_strong and set_unstrong writers.
     bool set_strong(const header_link& link, size_t count,
-        const tx_link& first_fk, bool positive) NOEXCEPT;
+        const tx_link& first_fk, bool positive, bool parallel) NOEXCEPT;
 
     /// Get all tx links for any point of block that is also in duplicate table.
     bool get_doubles(tx_links& out, const block& block) const NOEXCEPT;

test/primitives/linkage.cpp

@@ -110,6 +110,17 @@ static_assert(linkage<6>{ 42 } == 42_u64);
 static_assert(linkage<7>{ 42 } == 42_u64);
 static_assert(linkage<8>{ 42 } == 42_u64);
 
+// Offset addition (integer() cast operator for equality test).
+// Size 0 excluded, as its terminal is zero (no row is valid).
+static_assert(linkage<1>{ 40 } + 2u == 42_u8);
+static_assert(linkage<2>{ 40 } + 2u == 42_u16);
+static_assert(linkage<3>{ 40 } + 2u == 42_u32);
+static_assert(linkage<4>{ 40 } + 2u == 42_u32);
+static_assert(linkage<5>{ 40 } + 2u == 42_u64);
+static_assert(linkage<6>{ 40 } + 2u == 42_u64);
+static_assert(linkage<7>{ 40 } + 2u == 42_u64);
+static_assert(linkage<8>{ 40 } + 2u == 42_u64);
+
 // assign integer
 
 BOOST_AUTO_TEST_CASE(linkage__assign_integer__1__expected)
@@ -352,6 +363,39 @@ BOOST_AUTO_TEST_CASE(linkage__increment__8__expected)
     BOOST_REQUIRE_EQUAL(instance, 0x4201020304050602u);
 }
 
+// add-assign
+
+BOOST_AUTO_TEST_CASE(linkage__add_assign__4__expected)
+{
+    linkage<4> instance{ 0x42010200 };
+    instance += 1u;
+    BOOST_REQUIRE_EQUAL(instance, 0x42010201u);
+
+    instance += 0x10_size;
+    BOOST_REQUIRE_EQUAL(instance, 0x42010211u);
+}
+
+BOOST_AUTO_TEST_CASE(linkage__add_assign__2__expected)
+{
+    linkage<2> instance{ 0x4200 };
+    instance += 2u;
+    BOOST_REQUIRE_EQUAL(instance, 0x4202u);
+}
+
+BOOST_AUTO_TEST_CASE(linkage__add_assign__5__expected)
+{
+    linkage<5> instance{ 0x4201020300 };
+    instance += 0x100u;
+    BOOST_REQUIRE_EQUAL(instance, 0x4201020400u);
+}
+
+BOOST_AUTO_TEST_CASE(linkage__add_assign__8__expected)
+{
+    linkage<8> instance{ 0x4201020304050600 };
+    instance += 0x42u;
+    BOOST_REQUIRE_EQUAL(instance, 0x4201020304050642u);
+}
+
 // cast bytes
 
 BOOST_AUTO_TEST_CASE(linkage__bytes__0__expected)