Merge pull request #669 from evoskuil/master

Enable chaser_populate.

include/bitcoin/node/block_arena.hpp

@@ -27,6 +27,7 @@ namespace libbitcoin {
 namespace node {
 
 /// Thread UNSAFE linear memory arena.
+/// Caller must manage capacity to ensure buffer is not overflowed.
 class BCN_API block_arena final
   : public arena
 {
@@ -45,16 +46,20 @@ class BCN_API block_arena final
         return mutex_;
     }
 
+    void* require(size_t bytes) NOEXCEPT override;
+
 private:
     void* do_allocate(size_t bytes, size_t align) THROWS override;
     void do_deallocate(void* ptr, size_t bytes, size_t align) NOEXCEPT override;
     bool do_is_equal(const arena& other) const NOEXCEPT override;
-    size_t do_get_capacity() const NOEXCEPT override;
 
-    // These are thread safe.
+    // Number of bytes remaining to be allocated.
+    size_t capacity() const NOEXCEPT;
+
+    // These are thread safe (set only construct).
     std::shared_mutex mutex_{};
     uint8_t* memory_map_;
-    size_t capacity_;
+    size_t size_;
 
     // This is unprotected, caller must guard.
     size_t offset_;

include/bitcoin/node/full_node.hpp

@@ -173,7 +173,7 @@ class BCN_API full_node
     chaser_block chaser_block_;
     chaser_header chaser_header_;
     chaser_check chaser_check_;
-    ////chaser_populate chaser_populate_;
+    chaser_populate chaser_populate_;
     chaser_validate chaser_validate_;
     chaser_confirm chaser_confirm_;
     chaser_transaction chaser_transaction_;

src/block_arena.cpp

@@ -22,6 +22,8 @@
 #include <bitcoin/system.hpp>
 
 namespace libbitcoin {
+
+BC_DEBUG_ONLY(constexpr auto max_align = alignof(std::max_align_t);)
 
 template <typename Type, if_unsigned_integer<Type> = true>
 constexpr Type to_aligned(Type value, Type alignment) NOEXCEPT
@@ -42,14 +44,14 @@ BC_PUSH_WARNING(NO_POINTER_ARITHMETIC)
 
 block_arena::block_arena(size_t size) NOEXCEPT
   : memory_map_{ system::pointer_cast<uint8_t>(malloc(size)) },
-    capacity_{ size },
+    size_{ size },
     offset_{}
 {
 }
 
 block_arena::block_arena(block_arena&& other) NOEXCEPT
   : memory_map_{ other.memory_map_ },
-    capacity_{ other.capacity_ },
+    size_{ other.size_ },
     offset_{ other.offset_ }
 {
     // Prevents free(memory_map_) as responsibility is passed to this object.
@@ -68,38 +70,47 @@ block_arena::~block_arena() NOEXCEPT
 block_arena& block_arena::operator=(block_arena&& other) NOEXCEPT
 {
     memory_map_ = other.memory_map_;
-    capacity_ = other.capacity_;
+    size_ = other.size_;
     offset_ = other.offset_;
 
     // Prevents free(memory_map_) as responsibility is passed to this object.
     other.memory_map_ = nullptr;
     return *this;
 }
 
+// private
+size_t block_arena::capacity() const NOEXCEPT
+{
+    return system::floored_subtract(size_, offset_);
+}
+
+// Bytes includes any expected alignment.
+void* block_arena::require(size_t bytes) NOEXCEPT
+{
+    if (bytes > capacity())
+    {
+        std::unique_lock lock{ mutex_ };
+        offset_ = zero;
+    }
+
+    return memory_map_ + offset_;
+}
+
 void* block_arena::do_allocate(size_t bytes, size_t align) THROWS
 {
     using namespace system;
     BC_ASSERT_MSG(is_nonzero(align), "align zero");
-    BC_ASSERT_MSG(align <= alignof(std::max_align_t), "align overflow");
+    BC_ASSERT_MSG(align <= max_align, "align overflow");
     BC_ASSERT_MSG(power2(floored_log2(align)) == align, "align power");
-    BC_ASSERT_MSG(!is_add_overflow(bytes, sub1(align)), "align overflow");
-
-    auto aligned_offset = to_aligned(offset_, align);
-    auto padding = aligned_offset - offset_;
-    auto allocation = padding + bytes;
+    BC_ASSERT_MSG(!is_add_overflow(bytes, sub1(align)), "alignment overflow");
 
-    // Wraps if allocation would overflow.
-    if (allocation > get_capacity())
-    {
-        // Block until arena retainers are all released.
-        std::unique_lock lock(mutex_);
-        aligned_offset = offset_ = zero;
-        allocation = bytes;
+    const auto aligned_offset = to_aligned(offset_, align);
+    const auto padding = aligned_offset - offset_;
+    const auto allocation = padding + bytes;
 
-        // Throws if necessary allocation exceeds buffer.
-        if (bytes > capacity_)
-            throw allocation_exception();
-    }
+    ////BC_ASSERT_MSG(allocation <= capacity(), "buffer overflow");
+    if (allocation > capacity())
+        throw allocation_exception{};
 
     offset_ += allocation;
     return memory_map_ + aligned_offset;
@@ -115,11 +126,6 @@ bool block_arena::do_is_equal(const arena& other) const NOEXCEPT
     return &other == this;
 }
 
-size_t block_arena::do_get_capacity() const NOEXCEPT
-{
-    return system::floored_subtract(capacity_, offset_);
-}
-
 BC_POP_WARNING()
 BC_POP_WARNING()
 

src/block_memory.cpp

@@ -53,7 +53,9 @@ block_arena* block_memory::get_block_arena() const THROWS
         std::memory_order_relaxed);
 
     // More threads are requesting an arena than specified at construct.
-    BC_ASSERT(index < arenas_.size());
+    ////BC_ASSERT(index < arenas_.size());
+    if (index >= arenas_.size())
+        throw allocation_exception{};
 
     return &arenas_.at(index);
 }

src/chasers/chaser_populate.cpp

@@ -102,10 +102,10 @@ void chaser_populate::do_populate(const block::cptr& block,
     /*bool*/ ////archive().populate(*block);
 
     // Use all closure parameters to ensure they aren't optimized out.
+    // Captured block is not deleted until closure over this method deletes.
     if (block->is_valid() && is_nonzero(height) &&
         link != header_link::terminal)
     {
-        // Sends notification and deletes captured block in creating strand.
         // Notify coincident with delete ensures there is no cleanup backlog.
         complete(error::success);
     }

src/full_node.cpp

@@ -45,7 +45,7 @@ full_node::full_node(query& query, const configuration& configuration,
     chaser_block_(*this),
     chaser_header_(*this),
     chaser_check_(*this),
-    ////chaser_populate_(*this),
+    chaser_populate_(*this),
     chaser_validate_(*this),
     chaser_confirm_(*this),
     chaser_transaction_(*this),
@@ -85,7 +85,7 @@ void full_node::do_start(const result_handler& handler) NOEXCEPT
         chaser_header_.stopping(ec);
         chaser_block_.stopping(ec);
         chaser_check_.stopping(ec);
-        ////chaser_populate_.stopping(ec);
+        chaser_populate_.stopping(ec);
         chaser_validate_.stopping(ec);
         chaser_confirm_.stopping(ec);
         chaser_transaction_.stopping(ec);
@@ -99,7 +99,7 @@ void full_node::do_start(const result_handler& handler) NOEXCEPT
             chaser_header_.start() :
             chaser_block_.start()))) ||
         ((ec = chaser_check_.start())) ||
-        ////((ec = chaser_populate_.start())) ||
+        ((ec = chaser_populate_.start())) ||
         ((ec = chaser_validate_.start())) ||
         ((ec = chaser_confirm_.start())) ||
         ((ec = chaser_transaction_.start())) ||
@@ -245,10 +245,11 @@ void full_node::unsubscribe_events(object_key key) NOEXCEPT
 // Blocks.
 // ----------------------------------------------------------------------------
 
-void full_node::populate(const chain::block::cptr&, const header_link&, size_t,
-    network::result_handler&&) NOEXCEPT
+void full_node::populate(const chain::block::cptr& block,
+    const header_link& link, size_t height,
+    network::result_handler&& complete) NOEXCEPT
 {
-    ////chaser_populate_.populate(block, link, height, std::move(complete));
+    chaser_populate_.populate(block, link, height, std::move(complete));
 }
 
 void full_node::validate(const chain::block::cptr& block,

src/parser.cpp

@@ -900,7 +900,7 @@ options_metadata parser::load_settings() THROWS
     (
         "node.allocation_bytes",
         value<uint64_t>(&configured.node.allocation_bytes),
-        "Blocks preallocated memory buffer, defaults to 1'073'741'824."
+        "Preallocated block buffer for each network thread, defaults to 30'000'000."
     )
     (
         "node.maximum_height",

src/protocols/protocol_block_in_31800.cpp

@@ -370,13 +370,9 @@ bool protocol_block_in_31800::handle_receive_block(const code& ec,
     LOGP("Downloaded block [" << encode_hash(hash) << ":" << height
         << "] from [" << authority() << "].");
 
-    // protocol bind captures self, keeping channel alive until closure delete.
-    ////populate(block, link, height, BIND(complete, _1, block, height));
-    notify(ec, chase::checked, height);
-    fire(events::block_archived, height /*block->get_retainer()->allocation()*/);
-
-    ////LOGA("Height: " << height << " size: " << size
-    ////    << " bytes: " << block->get_retainer()->allocation());
+    populate(block, link, height, BIND(complete, _1, block, height));
+    ////notify(ec, chase::checked, height);
+    ////fire(events::block_archived, height);
 
     count(size);
     map_->erase(it);
@@ -396,6 +392,7 @@ void protocol_block_in_31800::complete(const code& ec,
     {
         notify(ec, chase::checked, height);
         fire(events::block_archived, height);
+        ////const auto bytes = block->get_retainer()->allocation();
     }
 }
 

src/settings.cpp

@@ -77,7 +77,7 @@ namespace node {
 settings::settings() NOEXCEPT
   : headers_first{ true },
     allowed_deviation{ 1.5 },
-    allocation_bytes{ 1'073'741'824 },
+    allocation_bytes{ 30'000'000 },
     snapshot_bytes{ 107'374'182'400 },
     snapshot_valid{ 100'000 },
     maximum_height{ 0 },

test/settings.cpp

@@ -60,8 +60,8 @@ BOOST_AUTO_TEST_CASE(settings__node__default_context__expected)
     BOOST_REQUIRE_EQUAL(node.snapshot_valid, 100'000_u32);
     BOOST_REQUIRE_EQUAL(node.maximum_height, 0_u32);
 
-    BOOST_REQUIRE_EQUAL(node.allocation(), system::limit<size_t>(1'073'741'824_u64));
-    BOOST_REQUIRE_EQUAL(node.allocation_bytes, 1'073'741'824_u64);
+    BOOST_REQUIRE_EQUAL(node.allocation(), system::limit<size_t>(30'000'000_u64));
+    BOOST_REQUIRE_EQUAL(node.allocation_bytes, 30'000'000_u64);
 
     BOOST_REQUIRE_EQUAL(node.maximum_height_(), max_size_t);
     BOOST_REQUIRE_EQUAL(node.maximum_concurrency, 50000_u32);