Fix head race condition introduced by optimizations.

Makefile.am

@@ -186,6 +186,7 @@ include_bitcoin_database_locks_HEADERS = \
 include_bitcoin_database_memorydir = ${includedir}/bitcoin/database/memory
 include_bitcoin_database_memory_HEADERS = \
     include/bitcoin/database/memory/accessor.hpp \
+    include/bitcoin/database/memory/finalizer.hpp \
     include/bitcoin/database/memory/map.hpp \
     include/bitcoin/database/memory/memory.hpp \
     include/bitcoin/database/memory/reader.hpp \

builds/msvc/vs2022/libbitcoin-database/libbitcoin-database.vcxproj

@@ -103,6 +103,7 @@
     <ClInclude Include="..\..\..\..\include\bitcoin\database\locks\interprocess_lock.hpp" />
     <ClInclude Include="..\..\..\..\include\bitcoin\database\locks\locks.hpp" />
     <ClInclude Include="..\..\..\..\include\bitcoin\database\memory\accessor.hpp" />
+    <ClInclude Include="..\..\..\..\include\bitcoin\database\memory\finalizer.hpp" />
     <ClInclude Include="..\..\..\..\include\bitcoin\database\memory\interfaces\memory.hpp" />
     <ClInclude Include="..\..\..\..\include\bitcoin\database\memory\interfaces\storage.hpp" />
     <ClInclude Include="..\..\..\..\include\bitcoin\database\memory\map.hpp" />

builds/msvc/vs2022/libbitcoin-database/libbitcoin-database.vcxproj.filters

@@ -152,6 +152,9 @@
     <ClInclude Include="..\..\..\..\include\bitcoin\database\memory\accessor.hpp">
       <Filter>include\bitcoin\database\memory</Filter>
     </ClInclude>
+    <ClInclude Include="..\..\..\..\include\bitcoin\database\memory\finalizer.hpp">
+      <Filter>include\bitcoin\database\memory</Filter>
+    </ClInclude>
     <ClInclude Include="..\..\..\..\include\bitcoin\database\memory\interfaces\memory.hpp">
       <Filter>include\bitcoin\database\memory\interfaces</Filter>
     </ClInclude>

include/bitcoin/database.hpp

@@ -32,6 +32,7 @@
 #include <bitcoin/database/locks/interprocess_lock.hpp>
 #include <bitcoin/database/locks/locks.hpp>
 #include <bitcoin/database/memory/accessor.hpp>
+#include <bitcoin/database/memory/finalizer.hpp>
 #include <bitcoin/database/memory/map.hpp>
 #include <bitcoin/database/memory/memory.hpp>
 #include <bitcoin/database/memory/reader.hpp>

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

@@ -147,21 +147,26 @@ TEMPLATE
 Link CLASS::first(const Key& key) const NOEXCEPT
 {
     ////return it(key).self();
-    return first(manager_.get(), head_.top(key), key);
+    return first(get_memory(), head_.top(key), key);
 }
 
 TEMPLATE
 typename CLASS::iterator CLASS::it(const Key& key) const NOEXCEPT
 {
-    return { manager_.get(), head_.top(key), key };
+    return { get_memory(), head_.top(key), key };
 }
-
 TEMPLATE
 Link CLASS::allocate(const Link& size) NOEXCEPT
 {
     return manager_.allocate(size);
 }
 
+TEMPLATE
+memory_ptr CLASS::get_memory() const NOEXCEPT
+{
+    return manager_.get();
+}
+
 TEMPLATE
 Key CLASS::get_key(const Link& link) NOEXCEPT
 {
@@ -178,7 +183,7 @@ template <typename Element, if_equal<Element::size, Size>>
 bool CLASS::find(const Key& key, Element& element) const NOEXCEPT
 {
     // This override avoids duplicated memory_ptr construct in get(first()).
-    const auto ptr = manager_.get();
+    const auto ptr = get_memory();
     return read(ptr, first(ptr, head_.top(key), key), element);
 }
 
@@ -187,7 +192,7 @@ template <typename Element, if_equal<Element::size, Size>>
 bool CLASS::get(const Link& link, Element& element) const NOEXCEPT
 {
     // This override is the normal form.
-    return read(manager_.get(), link, element);
+    return read(get_memory(), link, element);
 }
 
 // static
@@ -219,7 +224,7 @@ bool CLASS::set(const Link& link, const Element& element) NOEXCEPT
         return false;
 
     iostream stream{ *ptr };
-    flipper sink{ stream };
+    finalizer sink{ stream };
     sink.skip_bytes(index_size);
 
     if constexpr (!is_slab) { BC_DEBUG_ONLY(sink.set_limit(Size);) }
@@ -269,16 +274,23 @@ bool CLASS::put_link(Link& link, const Key& key,
         return false;
 
     iostream stream{ *ptr };
-    flipper sink{ stream };
+    finalizer sink{ stream };
     sink.skip_bytes(Link::size);
     sink.write_bytes(key);
 
+    // The finalizer provides deferred index commit following serialization.
+    // Because the lambda captures ptr and is in turn held as a member of sink,
+    // ptr is guaranteed in scope until sink destruction, which follows flush.
+    // It may be possible to instead assign ptr to a sink member, but not being
+    // referenced would make it subject to optimization (volatile might work).
+    sink.set_finalizer([this, link, index = head_.index(key), ptr]() NOEXCEPT
+    {
+        auto& next = unsafe_array_cast<uint8_t, Link::size>(ptr->begin());
+        return head_.push(link, next, index);
+    });
+
     if constexpr (!is_slab) { BC_DEBUG_ONLY(sink.set_limit(Size * count);) }
-    if (!element.to_data(sink))
-        return false;
-
-    auto& next = unsafe_array_cast<uint8_t, Link::size>(ptr->begin());
-    return head_.push(link, next, head_.index(key));
+    return element.to_data(sink) && sink.finalize();
 }
 
 TEMPLATE
@@ -300,16 +312,23 @@ bool CLASS::put(const Link& link, const Key& key,
         return false;
 
     iostream stream{ *ptr };
-    flipper sink{ stream };
+    finalizer sink{ stream };
     sink.skip_bytes(Link::size);
     sink.write_bytes(key);
 
-    if constexpr (!is_slab) { BC_DEBUG_ONLY(sink.set_limit(Size * count);) }
-    if (!element.to_data(sink))
-        return false;
+    // The finalizer provides deferred index commit following serialization.
+    // Because the lambda captures ptr and is in turn held as a member of sink,
+    // ptr is guaranteed in scope until sink destruction, which follows flush.
+    // It may be possible to instead assign ptr to a sink member, but not being
+    // referenced would make it subject to optimization (volatile might work).
+    sink.set_finalizer([this, link, index = head_.index(key), ptr]() NOEXCEPT
+    {
+        auto& next = unsafe_array_cast<uint8_t, Link::size>(ptr->begin());
+        return head_.push(link, next, index);
+    });
 
-    auto& next = unsafe_array_cast<uint8_t, Link::size>(ptr->begin());
-    return head_.push(link, next, head_.index(key));
+    if constexpr (!is_slab) { BC_DEBUG_ONLY(sink.set_limit(Size * count);) }
+    return element.to_data(sink) && sink.finalize();
 }
 
 TEMPLATE
@@ -328,6 +347,15 @@ bool CLASS::commit(const Link& link, const Key& key) NOEXCEPT
     return head_.push(link, next, head_.index(key));
 }
 
+TEMPLATE
+Link CLASS::commit_link(const Link& link, const Key& key) NOEXCEPT
+{
+    if (!commit(link, key))
+        return {};
+
+    return link;
+}
+
 // protected/static
 // ----------------------------------------------------------------------------
 
@@ -340,7 +368,7 @@ bool CLASS::read(const memory_ptr& ptr, const Link& link,
         return false;
 
     using namespace system;
-    const auto start = iterator::link_to_position(link);
+    const auto start = manager::link_to_position(link);
     if (is_limited<ptrdiff_t>(start))
         return false;
 
@@ -371,7 +399,7 @@ Link CLASS::first(const memory_ptr& ptr, Link link, const Key& key) NOEXCEPT
     while (!link.is_terminal())
     {
         // get element offset (fault)
-        const auto offset = ptr->offset(iterator::link_to_position(link));
+        const auto offset = ptr->offset(manager::link_to_position(link));
         if (is_null(offset))
             return {};
 

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

@@ -23,9 +23,6 @@
 #include <bitcoin/system.hpp>
 #include <bitcoin/database/define.hpp>
 
-// Heads are not subject to resize/remap and therefore do not require memory
-// smart pointer with shared remap lock. Using get_raw() saves that allocation.
-
 namespace libbitcoin {
 namespace database {
 
@@ -103,7 +100,7 @@ bool CLASS::get_body_count(Link& count) const NOEXCEPT
     if (!ptr)
         return false;
 
-    count = array_cast<Link::size>(ptr->begin());
+    count = array_cast<Link::size>(*ptr);
     return true;
 }
 
@@ -114,7 +111,7 @@ bool CLASS::set_body_count(const Link& count) NOEXCEPT
     if (!ptr)
         return false;
 
-    array_cast<Link::size>(ptr->begin()) = count;
+    array_cast<Link::size>(*ptr) = count;
     return true;
 }
 
@@ -127,11 +124,11 @@ Link CLASS::top(const Key& key) const NOEXCEPT
 TEMPLATE
 Link CLASS::top(const Link& index) const NOEXCEPT
 {
-    const auto ptr = file_.get_raw(offset(index));
-    if (is_null(ptr))
-        return {};
+    const auto ptr = file_.get(offset(index));
+    if (!ptr)
+        return Link::terminal;
 
-    const auto& head = array_cast<Link::size>(ptr);
+    const auto& head = array_cast<Link::size>(*ptr);
 
     mutex_.lock_shared();
     const auto top = head;
@@ -148,11 +145,11 @@ bool CLASS::push(const bytes& current, bytes& next, const Key& key) NOEXCEPT
 TEMPLATE
 bool CLASS::push(const bytes& current, bytes& next, const Link& index) NOEXCEPT
 {
-    const auto ptr = file_.get_raw(offset(index));
-    if (is_null(ptr))
+    const auto ptr = file_.get(offset(index));
+    if (!ptr)
         return false;
 
-    auto& head = array_cast<Link::size>(ptr);
+    auto& head = array_cast<Link::size>(*ptr);
 
     mutex_.lock();
     next = head;

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

@@ -71,13 +71,13 @@ Link CLASS::to_match(Link link) const NOEXCEPT
     while (!link.is_terminal())
     {
         // get element offset (fault)
-        const auto offset = memory_->offset(link_to_position(link));
+        const auto offset = memory_->offset(manager::link_to_position(link));
         if (is_null(offset))
             return {};
 
         // element key matches (found)
         const auto key_ptr = std::next(offset, Link::size);
-        if (is_zero(std::memcmp(key_.data(), key_ptr, key_size)))
+        if (is_zero(std::memcmp(key_.data(), key_ptr, array_count<Key>)))
             return std::move(link);
 
         // set next element link (loop)
@@ -93,7 +93,7 @@ Link CLASS::to_next(Link link) const NOEXCEPT
     while (!link.is_terminal())
     {
         // get element offset (fault)
-        auto offset = memory_->offset(link_to_position(link));
+        auto offset = memory_->offset(manager::link_to_position(link));
         if (is_null(offset))
             return {};
 
@@ -103,42 +103,19 @@ Link CLASS::to_next(Link link) const NOEXCEPT
             return std::move(link);
 
         // get next element offset (fault)
-        offset = memory_->offset(link_to_position(link));
+        offset = memory_->offset(manager::link_to_position(link));
         if (is_null(offset))
             return {};
 
         // next element key matches (found)
         const auto key_ptr = std::next(offset, Link::size);
-        if (is_zero(std::memcmp(key_.data(), key_ptr, key_size)))
+        if (is_zero(std::memcmp(key_.data(), key_ptr, array_count<Key>)))
             return std::move(link);
     }
 
     return std::move(link);
 }
 
-// private
-// ----------------------------------------------------------------------------
-
-TEMPLATE
-constexpr size_t CLASS::link_to_position(const Link& link) NOEXCEPT
-{
-    const auto value = system::possible_narrow_cast<size_t>(link.value);
-
-    if constexpr (is_slab)
-    {
-        // Slab implies link/key incorporated into size.
-        return value;
-    }
-    else
-    {
-        // Record implies link/key independent of Size.
-        constexpr auto element_size = Link::size + array_count<Key> + Size;
-        BC_ASSERT(!system::is_multiply_overflow(value, element_size));
-
-        return value * element_size;
-    }
-}
-
 } // namespace database
 } // namespace libbitcoin
 

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

@@ -105,7 +105,7 @@ code CLASS::reload() NOEXCEPT
     return file_.load();
 }
 
-// private
+// static
 // ----------------------------------------------------------------------------
 
 TEMPLATE
@@ -128,12 +128,13 @@ constexpr size_t CLASS::link_to_position(const Link& link) NOEXCEPT
     }
     else
     {
-        // No key implies no linked list.
+        // No key implies no linked list (not a hashmap), so record array.
         BC_ASSERT(!is_multiply_overflow(value, Size));
         return value * Size;
     }
 }
 
+// private
 TEMPLATE
 constexpr typename Link::integer CLASS::cast_link(size_t link) NOEXCEPT
 {
@@ -146,6 +147,7 @@ constexpr typename Link::integer CLASS::cast_link(size_t link) NOEXCEPT
     return link >= terminal ? terminal : possible_narrow_cast<integer>(link);
 }
 
+// private
 TEMPLATE
 constexpr Link CLASS::position_to_link(size_t position) NOEXCEPT
 {