Merge pull request #667 from evoskuil/master

Use thread_local block_arenas in block_memory.

include/bitcoin/node/block_arena.hpp

@@ -26,29 +26,39 @@
 namespace libbitcoin {
 namespace node {
 
-/// Thread safe block memory arena.
+/// Thread UNSAFE linear memory arena.
 class BCN_API block_arena final
   : public arena
 {
 public:
-    DELETE_COPY_MOVE(block_arena);
+    DELETE_COPY(block_arena);
+
+    block_arena(size_t size=zero) NOEXCEPT;
+    block_arena(block_arena&& other) NOEXCEPT;
+    ~block_arena() NOEXCEPT;
+
+    block_arena& operator=(block_arena&& other) NOEXCEPT;
 
     inline std::shared_mutex& get_mutex() NOEXCEPT
     {
         return mutex_;
     }
 
-    block_arena() NOEXCEPT;
-    ~block_arena() NOEXCEPT;
-
 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;
 
-    // This is thread safe.
+    // These are thread safe.
     std::shared_mutex mutex_{};
+    uint8_t* memory_map_;
+    size_t capacity_;
+
+    // This is unprotected, caller must guard.
+    size_t offset_;
+
 };
+
 } // namespace node
 } // namespace libbitcoin
 

include/bitcoin/node/block_memory.hpp

@@ -19,24 +19,41 @@
 #ifndef LIBBITCOIN_NODE_BLOCK_MEMORY_HPP
 #define LIBBITCOIN_NODE_BLOCK_MEMORY_HPP
 
+#include <atomic>
 #include <shared_mutex>
+#include <thread>
 #include <bitcoin/network.hpp>
 #include <bitcoin/node/block_arena.hpp>
 #include <bitcoin/node/define.hpp>
 
 namespace libbitcoin {
 namespace node {
 
-/// Thread safe block memory.
+/// Thread SAFE linear memory allocation and tracking.
 class BCN_API block_memory final
   : public network::memory
 {
 public:
+    DELETE_COPY_MOVE_DESTRUCT(block_memory);
+
+    /// Default allocate each arena to preclude allcation and locking.
+    block_memory(size_t bytes, size_t threads) NOEXCEPT;
+
+    /// Each thread obtains an arena of the same size.
     arena* get_arena() NOEXCEPT override;
+
+    /// Each thread obtains its arena's retainer.
     retainer::ptr get_retainer() NOEXCEPT override;
 
+protected:
+    block_arena* get_block_arena() THROWS;
+
 private:
-    block_arena arena_{};
+    // This is thread safe.
+    std::atomic_size_t count_;
+
+    // This is protected by constructor init and thread_local indexation.
+    std::vector<block_arena> arenas_;
 };
 
 } // namespace node

include/bitcoin/node/settings.hpp

@@ -74,6 +74,7 @@ class BCN_API settings
     /// Properties.
     bool headers_first;
     float allowed_deviation;
+    uint64_t allocation_bytes;
     uint64_t snapshot_bytes;
     uint32_t snapshot_valid;
     uint32_t maximum_height;
@@ -83,6 +84,7 @@ class BCN_API settings
     uint32_t threads;
 
     /// Helpers.
+    virtual size_t allocation() const NOEXCEPT;
     virtual size_t maximum_height_() const NOEXCEPT;
     virtual size_t maximum_concurrency_() const NOEXCEPT;
     virtual network::steady_clock::duration sample_period() const NOEXCEPT;

src/block_arena.cpp

@@ -18,36 +18,89 @@
  */
 #include <bitcoin/node/block_arena.hpp>
 
+#include <stdlib.h>
 #include <shared_mutex>
 #include <bitcoin/system.hpp>
 
 namespace libbitcoin {
+
+template <typename Type, if_unsigned_integer<Type> = true>
+constexpr Type to_aligned(Type value, Type alignment) NOEXCEPT
+{
+    return (value + sub1(alignment)) & ~sub1(alignment);
+}
+
 namespace node {
 
-// TODO: initialize memory.
-block_arena::block_arena() NOEXCEPT
+BC_PUSH_WARNING(NO_MALLOC_OR_FREE)
+BC_PUSH_WARNING(NO_POINTER_ARITHMETIC)
+
+// "If size is zero, the behavior of malloc is implementation-defined. For
+// example, a null pointer may be returned. Alternatively, a non-null pointer
+// may be returned; but such a pointer should not be dereferenced, and should
+// be passed to free to avoid memory leaks."
+// en.cppreference.com/w/c/memory/malloc
+
+block_arena::block_arena(size_t size) NOEXCEPT
+  : memory_map_{ system::pointer_cast<uint8_t>(malloc(size)) },
+    capacity_{ size },
+    offset_{}
+{
+}
+
+block_arena::block_arena(block_arena&& other) NOEXCEPT
+  : memory_map_{ other.memory_map_ },
+    capacity_{ other.capacity_ },
+    offset_{ other.offset_ }
 {
+    // Prevents free(memory_map_) as responsibility is passed to this object.
+    other.memory_map_ = nullptr;
 }
 
-// TODO: block on mutex until exclusive and then free memory.
 block_arena::~block_arena() NOEXCEPT
 {
+    if (!is_null(memory_map_))
+    {
+        std::unique_lock lock(mutex_);
+        free(memory_map_);
+    }
 }
 
-// TODO: if aligned size is insufficient block on mutex until memory remapped.
-void* block_arena::do_allocate(size_t bytes, size_t) THROWS
+block_arena& block_arena::operator=(block_arena&& other) NOEXCEPT
 {
-    BC_PUSH_WARNING(NO_NEW_OR_DELETE)
-    return ::operator new(bytes);
-    BC_POP_WARNING()
+    memory_map_ = other.memory_map_;
+    capacity_ = other.capacity_;
+    offset_ = other.offset_;
+
+    // Prevents free(memory_map_) as responsibility is passed to this object.
+    other.memory_map_ = nullptr;
+    return *this;
+}
+
+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(power2(floored_log2(align)) == align, "align power");
+    BC_ASSERT_MSG(!is_add_overflow(bytes, sub1(align)), "align overflow");
+
+    auto aligned = to_aligned(offset_, align);
+    if (bytes > system::floored_subtract(capacity_, aligned))
+    {
+        std::unique_lock lock(mutex_);
+        aligned = offset_ = zero;
+
+        if (bytes > capacity_)
+            throw allocation_exception();
+    }
+
+    offset_ = aligned + bytes;
+    return memory_map_ + aligned;
 }
 
-// TODO: change to nop.
-void block_arena::do_deallocate(void* ptr, size_t, size_t) NOEXCEPT
+void block_arena::do_deallocate(void*, size_t, size_t) NOEXCEPT
 {
-    BC_PUSH_WARNING(NO_NEW_OR_DELETE)
-    ::operator delete(ptr);
-    BC_POP_WARNING()
 }
 
 bool block_arena::do_is_equal(const arena& other) const NOEXCEPT
@@ -56,5 +109,8 @@ bool block_arena::do_is_equal(const arena& other) const NOEXCEPT
     return &other == this;
 }
 
+BC_POP_WARNING()
+BC_POP_WARNING()
+
 } // namespace node
 } // namespace libbitcoin

src/block_memory.cpp

@@ -19,22 +19,46 @@
 #include <bitcoin/node/block_memory.hpp>
 
 #include <memory>
+#include <thread>
 #include <bitcoin/system.hpp>
 
 namespace libbitcoin {
 namespace node {
 
+block_memory::block_memory(size_t bytes, size_t threads) NOEXCEPT
+  : count_{}, arenas_{}
+{
+    arenas_.reserve(threads);
+    for (auto index = zero; index < threads; ++index)
+        arenas_.emplace_back(bytes);
+}
+
 arena* block_memory::get_arena() NOEXCEPT
 {
-    return &arena_;
+    BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
+    return get_block_arena();
+    BC_POP_WARNING()
 }
 
 retainer::ptr block_memory::get_retainer() NOEXCEPT
 {
-    BC_PUSH_WARNING(NO_NEW_OR_DELETE)
-    return std::make_shared<retainer>(arena_.get_mutex());
+    BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
+    return std::make_shared<retainer>(get_block_arena()->get_mutex());
     BC_POP_WARNING()
 }
 
+// protected
+
+block_arena* block_memory::get_block_arena() THROWS
+{
+    static thread_local auto index = count_.fetch_add(one);
+
+    // More threads are requesting an arena than specified at construct.
+    if (index >= arenas_.size())
+        throw allocation_exception();
+
+    return &arenas_.at(index);
+}
+
 } // namespace node
 } // namespace libbitcoin

src/chasers/chaser_check.cpp

@@ -298,7 +298,8 @@ map_ptr chaser_check::get_map() NOEXCEPT
 
 bool chaser_check::set_map(const map_ptr& map) NOEXCEPT
 {
-    BC_ASSERT(stranded());
+    // Called from start.
+    ////BC_ASSERT(stranded());
     BC_ASSERT(map->size() <= messages::max_inventory);
 
     if (map->empty())

src/full_node.cpp

@@ -40,7 +40,7 @@ full_node::full_node(query& query, const configuration& configuration,
     const logger& log) NOEXCEPT
   : p2p(configuration.network, log),
     config_(configuration),
-    memory_(),
+    memory_(configuration.node.allocation(), configuration.network.threads),
     query_(query),
     chaser_block_(*this),
     chaser_header_(*this),

src/parser.cpp

@@ -897,6 +897,11 @@ options_metadata parser::load_settings() THROWS
         value<float>(&configured.node.allowed_deviation),
         "Allowable underperformance standard deviation, defaults to 1.5 (0 disables)."
     )
+    (
+        "node.allocation_bytes",
+        value<uint64_t>(&configured.node.allocation_bytes),
+        "Blocks preallocated memory buffer, defaults to 1'073'741'824."
+    )
     (
         "node.maximum_height",
         value<uint32_t>(&configured.node.maximum_height),

src/settings.cpp

@@ -77,6 +77,7 @@ namespace node {
 settings::settings() NOEXCEPT
   : headers_first{ true },
     allowed_deviation{ 1.5 },
+    allocation_bytes{ 1'073'741'824 },
     snapshot_bytes{ 107'374'182'400 },
     snapshot_valid{ 100'000 },
     maximum_height{ 0 },
@@ -93,6 +94,11 @@ settings::settings(chain::selection) NOEXCEPT
     // TODO: testnet, etc. maximum_concurrency, snapshot_bytes.
 }
 
+size_t settings::allocation() const NOEXCEPT
+{
+    return system::limit<size_t>(allocation_bytes);
+}
+
 size_t settings::maximum_height_() const NOEXCEPT
 {
     return to_bool(maximum_height) ? maximum_height : max_size_t;

test/settings.cpp

@@ -59,10 +59,15 @@ BOOST_AUTO_TEST_CASE(settings__node__default_context__expected)
     BOOST_REQUIRE_EQUAL(node.snapshot_bytes, 107'374'182'400_u64);
     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.maximum_height_(), max_size_t);
     BOOST_REQUIRE_EQUAL(node.maximum_concurrency, 50000_u32);
     BOOST_REQUIRE_EQUAL(node.maximum_concurrency_(), 50000_size);
     BOOST_REQUIRE_EQUAL(node.sample_period_seconds, 10_u16);
+
     BOOST_REQUIRE(node.sample_period() == steady_clock::duration(seconds(10)));
     BOOST_REQUIRE_EQUAL(node.currency_window_minutes, 60_u32);
     BOOST_REQUIRE(node.currency_window() == steady_clock::duration(minutes(60)));