Merge pull request #668 from evoskuil/master

Add retainer allocation param, add arena::get_capacity().

include/bitcoin/node/block_arena.hpp

@@ -39,6 +39,7 @@ class BCN_API block_arena final
 
     block_arena& operator=(block_arena&& other) NOEXCEPT;
 
+    /// Get memory block retainer mutex.
     inline std::shared_mutex& get_mutex() NOEXCEPT
     {
         return mutex_;
@@ -48,6 +49,7 @@ class BCN_API block_arena final
     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.
     std::shared_mutex mutex_{};

include/bitcoin/node/block_memory.hpp

@@ -20,8 +20,6 @@
 #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>
@@ -43,17 +41,17 @@ class BCN_API block_memory final
     arena* get_arena() NOEXCEPT override;
 
     /// Each thread obtains its arena's retainer.
-    retainer::ptr get_retainer() NOEXCEPT override;
+    retainer::ptr get_retainer(size_t allocation=zero) NOEXCEPT override;
 
 protected:
-    block_arena* get_block_arena() THROWS;
+    block_arena* get_block_arena() const THROWS;
 
 private:
     // This is thread safe.
-    std::atomic_size_t count_;
+    mutable std::atomic_size_t count_;
 
     // This is protected by constructor init and thread_local indexation.
-    std::vector<block_arena> arenas_;
+    mutable std::vector<block_arena> arenas_;
 };
 
 } // namespace node

include/bitcoin/node/full_node.hpp

@@ -166,10 +166,10 @@ class BCN_API full_node
 
     // These are thread safe.
     const configuration& config_;
-    block_memory memory_;
     query& query_;
 
     // These are protected by strand.
+    block_memory memory_;
     chaser_block chaser_block_;
     chaser_header chaser_header_;
     chaser_check chaser_check_;

include/bitcoin/node/sessions/attach.hpp

@@ -106,6 +106,7 @@ class attach
     network::channel::ptr create_channel(const network::socket::ptr& socket,
         bool quiet) NOEXCEPT override
     {
+        // This memory arena is NOT thread safe.
         return std::make_shared<network::channel>(session::get_memory(),
             network::session::log, socket, network::session::settings(),
             network::session::create_key(), quiet);

src/block_arena.cpp

@@ -18,7 +18,6 @@
  */
 #include <bitcoin/node/block_arena.hpp>
 
-#include <stdlib.h>
 #include <shared_mutex>
 #include <bitcoin/system.hpp>
 
@@ -85,18 +84,25 @@ void* block_arena::do_allocate(size_t bytes, size_t align) THROWS
     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))
+    auto aligned_offset = to_aligned(offset_, align);
+    auto padding = aligned_offset - offset_;
+    auto allocation = padding + bytes;
+
+    // Wraps if allocation would overflow.
+    if (allocation > get_capacity())
     {
+        // Block until arena retainers are all released.
         std::unique_lock lock(mutex_);
-        aligned = offset_ = zero;
+        aligned_offset = offset_ = zero;
+        allocation = bytes;
 
+        // Throws if necessary allocation exceeds buffer.
         if (bytes > capacity_)
             throw allocation_exception();
     }
 
-    offset_ = aligned + bytes;
-    return memory_map_ + aligned;
+    offset_ += allocation;
+    return memory_map_ + aligned_offset;
 }
 
 void block_arena::do_deallocate(void*, size_t, size_t) NOEXCEPT
@@ -109,6 +115,11 @@ 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

@@ -18,13 +18,15 @@
  */
 #include <bitcoin/node/block_memory.hpp>
 
+#include <atomic>
 #include <memory>
-#include <thread>
 #include <bitcoin/system.hpp>
 
 namespace libbitcoin {
 namespace node {
 
+BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
+
 block_memory::block_memory(size_t bytes, size_t threads) NOEXCEPT
   : count_{}, arenas_{}
 {
@@ -35,30 +37,28 @@ block_memory::block_memory(size_t bytes, size_t threads) NOEXCEPT
 
 arena* block_memory::get_arena() NOEXCEPT
 {
-    BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
     return get_block_arena();
-    BC_POP_WARNING()
 }
 
-retainer::ptr block_memory::get_retainer() NOEXCEPT
+retainer::ptr block_memory::get_retainer(size_t allocation) NOEXCEPT
 {
-    BC_PUSH_WARNING(NO_THROW_IN_NOEXCEPT)
-    return std::make_shared<retainer>(get_block_arena()->get_mutex());
-    BC_POP_WARNING()
+    return std::make_shared<retainer>(get_block_arena()->get_mutex(),
+        allocation);
 }
 
 // protected
-
-block_arena* block_memory::get_block_arena() THROWS
+block_arena* block_memory::get_block_arena() const THROWS
 {
-    static thread_local auto index = count_.fetch_add(one);
+    static thread_local auto index = count_.fetch_add(one,
+        std::memory_order_relaxed);
 
     // More threads are requesting an arena than specified at construct.
-    if (index >= arenas_.size())
-        throw allocation_exception();
+    BC_ASSERT(index < arenas_.size());
 
     return &arenas_.at(index);
 }
 
+BC_POP_WARNING()
+
 } // namespace node
 } // namespace libbitcoin

src/full_node.cpp

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

src/protocols/protocol_block_in_31800.cpp

@@ -373,7 +373,10 @@ bool protocol_block_in_31800::handle_receive_block(const code& ec,
     // 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);
+    fire(events::block_archived, height /*block->get_retainer()->allocation()*/);
+
+    ////LOGA("Height: " << height << " size: " << size
+    ////    << " bytes: " << block->get_retainer()->allocation());
 
     count(size);
     map_->erase(it);