Merge pull request #578 from evoskuil/master

Optimize transaction writes.

Author: evoskuil

include/bitcoin/database/error.hpp

@@ -117,12 +117,11 @@ enum error_t : uint8_t
     tx_empty,
     tx_tx_allocate,
     tx_input_put,
-    tx_point_allocate,
-    tx_point_put,
-    tx_ins_allocate,
     tx_ins_put,
     tx_output_put,
     tx_outs_put,
+    tx_point_allocate,
+    tx_point_put,
     tx_tx_set,
     tx_address_allocate,
     tx_address_put,

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

@@ -112,6 +112,12 @@ Link CLASS::count() const NOEXCEPT
     return body_.count();
 }
 
+TEMPLATE
+bool CLASS::expand(const Link& count) NOEXCEPT
+{
+    return body_.expand(count);
+}
+
 // query interface
 // ----------------------------------------------------------------------------
 

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

@@ -104,6 +104,12 @@ Link CLASS::count() const NOEXCEPT
     return body_.count();
 }
 
+TEMPLATE
+bool CLASS::expand(const Link& count) NOEXCEPT
+{
+    return body_.expand(count);
+}
+
 // query interface
 // ----------------------------------------------------------------------------
 

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

@@ -89,97 +89,42 @@ code CLASS::set_code(tx_link& tx_fk, const transaction& tx) NOEXCEPT
     if (tx.is_empty())
         return error::tx_empty;
 
+    using ix = linkage<schema::index>;
     const auto& ins = tx.inputs_ptr();
     const auto& ous = tx.outputs_ptr();
-
-    using ix = linkage<schema::index>;
     const auto txs = possible_narrow_cast<tx_link::integer>(one);
     const auto inputs = possible_narrow_cast<ix::integer>(ins->size());
     const auto outputs = possible_narrow_cast<ix::integer>(ous->size());
 
-    // Declare outs record for output accumulation.
-    table::outs::record outs{};
-    outs.out_fks.reserve(outputs);
 
     // ========================================================================
     const auto scope = store_.get_transactor();
 
-    // Allocate single tx record.
+    // Allocate tx record.
     tx_fk = store_.tx.allocate(txs);
     if (tx_fk.is_terminal())
         return error::tx_tx_allocate;
 
-    // Allocate points records.
-    const auto point_fk = store_.point.allocate(inputs);
-    auto point_it = point_fk;
-    if (point_fk.is_terminal())
-        return error::tx_point_allocate;
-
-    // Expand synchronizes keys with point allocation.
-    if (!store_.ins.expand(point_fk + inputs))
-        return error::tx_ins_allocate;
+    // Allocate contiguously and store inputs (script & witness).
+    input_link in_fk{};
+    if (!store_.input.put_link(in_fk, table::input::put_ref{ {}, tx }))
+        return error::tx_input_put;
 
-    // Commit input and ins|point records.
-    for (const auto& in: *ins)
-    {
-        // TODO: preallocate (requires input sizes).
-        input_link input_fk{};
-        if (!store_.input.put_link(input_fk, table::input::put_ref
-        {
-            {},
-            *in
-        }))
-        {
-            return error::tx_input_put;
-        }
+    // Allocate contiguously and store outputs (script, w/parent).
+    output_link out_fk{};
+    if (!store_.output.put_link(out_fk, table::output::put_ref{ {}, tx_fk, tx }))
+        return error::tx_output_put;
 
-        if (!store_.ins.put(point_it++, table::ins::record
-        {
-            {},
-            in->sequence(),
-            input_fk,
-            tx_fk
-        }))
-        {
-            return error::tx_ins_put;
-        }
-    }
+    // Allocate and contiguously store input links (first + tx.input sizes, w/parent & seq).
+    point_link ins_fk{};
+    if (!store_.ins.put_link(ins_fk, table::ins::put_ref{ {}, in_fk, tx_fk, tx }))
+        return error::tx_ins_put;
 
-    // TODO: preallocate (requires output sizes).
-    // Commit output records.
-    for (const auto& out: *ous)
-    {
-        output_link output_fk{};
-        if (!store_.output.put_link(output_fk, table::output::put_ref
-        {
-            {},
-            tx_fk,
-            *out
-        }))
-        {
-            return error::tx_output_put;
-        }
-
-        // Accumulate outputs in order.
-        outs.out_fks.push_back(output_fk);
-    }
-
-    // Commit accumulated outs.
-    const auto outs_fk = store_.outs.put_link(outs);
-    if (outs_fk.is_terminal())
+    // Allocate and contiguously store output links (first + tx.output sizes).
+    point_link outs_fk{};
+    if (!store_.outs.put_link(outs_fk, table::outs::put_ref{ {}, out_fk, tx }))
         return error::tx_outs_put;
 
-    // Commit accumulated points.
-    point_it = point_fk;
-    for (const auto& in: *ins)
-    {
-        const auto key = table::point::compose(in->point());
-        if (!store_.point.put(point_it++, key, table::point::record{}))
-        {
-            return error::tx_point_put;
-        }
-    }
-
     // Create tx record.
     // Commit is deferred for point/address index consistency.
     if (!store_.tx.set(tx_fk, table::transaction::record_put_ref
@@ -188,38 +133,50 @@ code CLASS::set_code(tx_link& tx_fk, const transaction& tx) NOEXCEPT
         tx,
         inputs,
         outputs,
-        point_fk,
+        ins_fk,
         outs_fk
     }))
     {
         return error::tx_tx_set;
     }
 
-    // Commit address index records.
+    // Commit points (hashmap).
+    {
+        // Expand synchronizes keys with ins_fk, entries dropped into same offset.
+        // Allocate contiguous points (at sequential keys matching ins_fk).
+        if (!store_.point.expand(ins_fk + inputs))
+            return error::tx_point_allocate;
+
+        // This must be set after tx.set and before tx.commit, since searchable and
+        // produces an association to tx.link, and is also an integral part of tx.
+        const auto ptr = store_.point.get_memory();
+        for (const auto& in: *ins)
+        {
+            if (!store_.point.put(ptr, ins_fk++,
+                table::point::compose(in->point()), table::point::record{}))
+                return error::tx_point_put;
+        }
+    }
+
+    // Commit address index records (hashmap).
     if (address_enabled())
     {
         auto ad_fk = store_.address.allocate(outputs);
         if (ad_fk.is_terminal())
             return error::tx_address_allocate;
 
-        auto output = ous->begin();
         const auto ptr = store_.address.get_memory();
-
-        for (auto out_fk: outs.out_fks)
+        for (const auto& output: *ous)
         {
-            const auto key = (*output++)->script().hash();
-            if (!store_.address.put(ptr, ad_fk++, key, table::address::record
-            {
-                {},
-                out_fk
-            }))
-            {
+            if (!store_.address.put(ptr, ad_fk++, output->script().hash(),
+                table::address::record{ {}, out_fk }))
                 return error::tx_address_put;
-            }
+
+            out_fk.value += output->serialized_size();
         }
     }
 
-    // Commit tx to search.
+    // Commit tx to search (hashmap).
     // tx.get_hash() assumes cached or is not thread safe.
     return store_.tx.commit(tx_fk, tx.get_hash(false)) ?
         error::success : error::tx_tx_commit;

include/bitcoin/database/primitives/arraymap.hpp

@@ -73,6 +73,9 @@ class arraymap
     /// Count of records (or body file bytes if slab).
     Link count() const NOEXCEPT;
 
+    /// Increase count as neccesary to specified.
+    bool expand(const Link& count) NOEXCEPT;
+
     /// Errors.
     /// -----------------------------------------------------------------------
 

include/bitcoin/database/primitives/hashmap.hpp

@@ -74,6 +74,9 @@ class hashmap
     /// Count of records (or body file bytes if slab).
     Link count() const NOEXCEPT;
 
+    /// Increase count as neccesary to specified.
+    bool expand(const Link& count) NOEXCEPT;
+
     /// Errors.
     /// -----------------------------------------------------------------------
 

include/bitcoin/database/tables/archives/input.hpp

@@ -101,20 +101,37 @@ struct input
     {
         inline link count() const NOEXCEPT
         {
-            return system::possible_narrow_cast<link::integer>(
-                input.script().serialized_size(true) +
-                input.witness().serialized_size(true));
+            using namespace system;
+            static constexpr auto sequence_point_size = sizeof(uint32_t) +
+                chain::point::serialized_size();
+
+            const auto& ins = *tx_.inputs_ptr();
+            const auto other = ins.size() * sequence_point_size;
+            const auto inputs = std::accumulate(ins.begin(), ins.end(), zero,
+                [](size_t total, const auto& in) NOEXCEPT
+                {
+                    // sizes cached, so this is free.
+                    return total + in->serialized_size(true);
+                });
+
+            // (script + witness + sequence + point) - (sequence + point)
+            return possible_narrow_cast<link::integer>(inputs - other);
         }
 
         inline bool to_data(flipper& sink) const NOEXCEPT
         {
-            input.script().to_data(sink, true);
-            input.witness().to_data(sink, true);
+            const auto& ins = *tx_.inputs_ptr();
+            std::for_each(ins.begin(), ins.end(), [&](const auto& in) NOEXCEPT
+            {
+                in->script().to_data(sink, true);
+                in->witness().to_data(sink, true);
+            });
+
             BC_ASSERT(!sink || sink.get_write_position() == count());
             return sink;
         }
 
-        const system::chain::input& input{};
+        const system::chain::transaction& tx_{};
     };
 };
 

include/bitcoin/database/tables/archives/ins.hpp

@@ -97,6 +97,42 @@ struct ins
         uint32_t sequence{};
         in::integer input_fk{};
     };
+
+    struct put_ref
+      : public schema::ins
+    {
+        link count() const NOEXCEPT
+        {
+            return system::possible_narrow_cast<link::integer>(tx_.inputs());
+        }
+
+        inline bool to_data(flipper& sink) const NOEXCEPT
+        {
+            using namespace system;
+            static constexpr auto sequence_point_size = sizeof(uint32_t) +
+                chain::point::serialized_size();
+
+            auto in_fk = input_fk;
+            const auto& ins = *tx_.inputs_ptr();
+            std::for_each(ins.begin(), ins.end(), [&](const auto& in) NOEXCEPT
+            {
+                sink.write_little_endian<uint32_t>(in->sequence());
+                sink.write_little_endian<in::integer, in::size>(in_fk);
+                sink.write_little_endian<tx::integer, tx::size>(parent_fk);
+
+                // Calculate next corresponding input fk from serialized size.
+                // (script + witness + sequence + point) - (sequence + point)
+                in_fk += (in->serialized_size(true) - sequence_point_size);
+            });
+
+            BC_ASSERT(!sink || sink.get_write_position() == count() * minrow);
+            return sink;
+        }
+
+        const in::integer input_fk{};
+        const tx::integer parent_fk{};
+        const system::chain::transaction& tx_{};
+    };
 };
 
 } // namespace table

include/bitcoin/database/tables/archives/output.hpp

@@ -150,23 +150,42 @@ struct output
     {
         inline link count() const NOEXCEPT
         {
-            return system::possible_narrow_cast<link::integer>(
-                tx::size +
-                variable_size(output.value()) +
-                output.script().serialized_size(true));
+            using namespace system;
+            static_assert(tx::size <= sizeof(uint64_t));
+            static constexpr auto value_parent_difference = sizeof(uint64_t) -
+                tx::size;
+
+            const auto& outs = *tx_.outputs_ptr();
+            const auto other = outs.size() * value_parent_difference;
+            const auto outputs = std::accumulate(outs.begin(), outs.end(), zero,
+                [](size_t total, const auto& out) NOEXCEPT
+                {
+                    // size cached, so this is free, includes sizeof(value).
+                    return total + variable_size(out->value()) +
+                        out->serialized_size();
+                });
+
+            // Converts value from fixed size wire encoding to variable.
+            // (variable_size(value) + (value + script)) - (value - parent)
+            return possible_narrow_cast<link::integer>(outputs - other);
         }
 
         inline bool to_data(flipper& sink) const NOEXCEPT
         {
-            sink.write_little_endian<tx::integer, tx::size>(parent_fk);
-            sink.write_variable(output.value());
-            output.script().to_data(sink, true);
+            const auto& outs = *tx_.outputs_ptr();
+            std::for_each(outs.begin(), outs.end(), [&](const auto& out) NOEXCEPT
+            {
+                sink.write_little_endian<tx::integer, tx::size>(parent_fk);
+                sink.write_variable(out->value());
+                out->script().to_data(sink, true);
+            });
+
             BC_ASSERT(!sink || sink.get_write_position() == count());
             return sink;
         }
 
-        tx::integer parent_fk{};
-        const system::chain::output& output{};
+        const tx::integer parent_fk{};
+        const system::chain::transaction& tx_{};
     };
 };