Add opaque value get and set to headmap.

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

@@ -132,7 +132,13 @@ code CLASS::reload() NOEXCEPT
 // ----------------------------------------------------------------------------
 
 TEMPLATE
-Link CLASS::at(size_t index) const NOEXCEPT
+memory CLASS::get_memory() const NOEXCEPT
+{
+    return file_.get();
+}
+
+TEMPLATE
+Value CLASS::at(size_t index) const NOEXCEPT
 {
     using namespace system;
 
@@ -147,9 +153,9 @@ Link CLASS::at(size_t index) const NOEXCEPT
     const auto raw = ptr.data();
     if constexpr (aligned)
     {
-        // Reads full padded word (masked by to_link).
+        // Reads full padded word (masked by to_value).
         const auto& head = *pointer_cast<std::atomic<cell>>(raw);
-        return to_link(head.load(std::memory_order_relaxed));
+        return to_value(head.load(load_order));
     }
     else
     {
@@ -160,18 +166,12 @@ Link CLASS::at(size_t index) const NOEXCEPT
         cell_array(value) = head;
         mutex_.unlock_shared();
 
-        return to_link(value);
+        return to_value(value);
     }
 }
 
 TEMPLATE
-memory CLASS::get_memory() const NOEXCEPT
-{
-    return file_.get();
-}
-
-TEMPLATE
-Link CLASS::at(const memory& ptr, size_t index) const NOEXCEPT
+Value CLASS::at(const memory& ptr, size_t index) const NOEXCEPT
 {
     using namespace system;
 
@@ -185,9 +185,9 @@ Link CLASS::at(const memory& ptr, size_t index) const NOEXCEPT
 
     if constexpr (aligned)
     {
-        // Reads full padded word (masked by to_link).
+        // Reads full padded word (masked by to_value).
         const auto& head = *pointer_cast<std::atomic<cell>>(raw);
-        return to_link(head.load(std::memory_order_relaxed));
+        return to_value(head.load(load_order));
     }
     else
     {
@@ -198,7 +198,7 @@ Link CLASS::at(const memory& ptr, size_t index) const NOEXCEPT
         cell_array(value) = head;
         mutex_.unlock_shared();
 
-        return to_link(value);
+        return to_value(value);
     }
 }
 
@@ -221,7 +221,7 @@ Link CLASS::allocate(const Link& bytes) NOEXCEPT
 }
 
 TEMPLATE
-template <typename Element>
+template <typename Element, if_equal<Element::size, Cell>>
 bool CLASS::get(const Link& link, Element& element) const NOEXCEPT
 {
     using namespace system;
@@ -250,7 +250,7 @@ bool CLASS::get(const Link& link, Element& element) const NOEXCEPT
 
 // NOT WRITER-WRITER THREAD SAFE (the element is read-write).
 TEMPLATE
-template <typename Element>
+template <typename Element, if_equal<Element::size, Cell>>
 bool CLASS::put(const Link& link, const Element& element) NOEXCEPT
 {
     using namespace system;
@@ -317,6 +317,82 @@ bool CLASS::push(const Link& link) NOEXCEPT
     return file_.allocate(bucket_size) != storage::eof;
 }
 
+// cell
+// ----------------------------------------------------------------------------
+
+TEMPLATE
+template <typename Element, if_equal<Element::size, Cell>,
+    if_not_equal<Element::size, max_size_t>>
+bool CLASS::get(size_t index, Element& element) const NOEXCEPT
+{
+    using namespace system;
+    static_assert(!is_slab);
+
+    // Buckets at or above the logical size are unallocated (count publishes).
+    if (index >= count())
+        return false;
+
+    const auto ptr = file_.get(link_to_position(index));
+    if (!ptr)
+        return false;
+
+    const auto raw = ptr.data();
+    cell value{};
+    if constexpr (aligned)
+    {
+        const auto& head = *pointer_cast<std::atomic<cell>>(raw);
+        value = head.load(load_order);
+    }
+    else
+    {
+        const auto& head = cell_array(raw);
+
+        mutex_.lock_shared();
+        cell_array(value) = head;
+        mutex_.unlock_shared();
+    }
+
+    iostream stream{ pointer_cast<uint8_t>(&value), cell_size };
+    reader source{ stream };
+    return element.from_data(source);
+}
+
+TEMPLATE
+template <typename Element, if_equal<Element::size, Cell>,
+    if_not_equal<Element::size, max_size_t>>
+bool CLASS::set(size_t index, const Element& element) NOEXCEPT
+{
+    using namespace system;
+    static_assert(!is_slab);
+    static_assert(aligned, "set requires an aligned cell");
+
+    // Allocate as necessary and fill allocations.
+    const auto position = link_to_position(index);
+    const auto ptr = file_.get_filled(position, bucket_size, fill);
+    if (!ptr)
+        return false;
+
+    file_.prepare(position, bucket_size);
+    auto& top = *pointer_cast<std::atomic<cell>>(ptr.data());
+    auto head = top.load(std::memory_order_acquire);
+    auto written = true;
+    cell update{};
+    do
+    {
+        auto copy = element;
+        iostream current{ pointer_cast<uint8_t>(&head), cell_size };
+        reader source{ current };
+        iostream replacement{ pointer_cast<uint8_t>(&update), cell_size };
+        flipper sink{ replacement };
+        written = copy.from_data(source) && copy.to_data(sink);
+    }
+    while (written && !top.compare_exchange_weak(head, update,
+        std::memory_order_release, std::memory_order_acquire));
+
+    file_.mark(position, bucket_size);
+    return written;
+}
+
 } // namespace database
 } // namespace libbitcoin
 

include/bitcoin/database/primitives/headmap.hpp

@@ -29,13 +29,15 @@
 namespace libbitcoin {
 namespace database {
 
-/// A head-only array of links indexed by position (no body, no count cell).
+/// A head-only array of values indexed by position (no body, no count cell).
 /// The storage logical size is the count, providing count atomicity and
 /// bucket publication as with bodies. Content restores with the heads.
-/// A maximal cell is a slab, in which the link is a byte offset and elements
-/// are serialized, as with bodies (the link accessors do not apply).
-template <class Link, size_t Cell = Link::size,
-    if_not_greater<Link::size, Cell> = true>
+/// A value is a link (masked to terminal) or an opaque integral, of no more
+/// bytes than the cell or the link.
+/// A cell size of max_size_t is a slab, in which the link is a byte offset
+/// and elements are serialized as with bodies (cell accessors do not apply).
+template <class Link, size_t Cell = Link::size, class Value = Link,
+    if_not_greater<value_size<Value>(), std::min(Cell, Link::size)> = true>
 class headmap
 {
 public:
@@ -88,28 +90,42 @@ class headmap
     /// Query interface.
     /// -----------------------------------------------------------------------
 
-    /// Return link at index, terminal if at or above count.
-    Link at(size_t index) const NOEXCEPT;
-
     /// Head accessor (shared lock), for at(ptr, index) over many indexes.
     memory get_memory() const NOEXCEPT;
-    Link at(const memory& ptr, size_t index) const NOEXCEPT;
+
+    /// Return value at index, default (terminal link) if not allocated.
+    Value at(size_t index) const NOEXCEPT;
+
+    /// Return value at index using accessor, default if not allocated.
+    Value at(const memory& ptr, size_t index) const NOEXCEPT;
 
     /// Append link at count into reserved capacity (single writer).
     bool push(const Link& link) NOEXCEPT;
 
+    /// Get element at index, false if not allocated or deserialize error.
+    template <typename Element, if_equal<Element::size, Cell> = true,
+        if_not_equal<Element::size, max_size_t> = true>
+    bool get(size_t index, Element& element) const NOEXCEPT;
+
+    /// Set element at index, extending count as necessary (filled empty).
+    /// The element reads the current cell and writes its replacement, and is
+    /// rerun from a copy for each failed exchange.
+    template <typename Element, if_equal<Element::size, Cell> = true,
+        if_not_equal<Element::size, max_size_t> = true>
+    bool set(size_t index, const Element& element) NOEXCEPT;
+
     /// Slab.
     /// -----------------------------------------------------------------------
 
     /// Allocate slab bytes at returned link (follow with put).
     Link allocate(const Link& bytes) NOEXCEPT;
 
     /// Get element at link, false if deserialize error.
-    template <typename Element>
+    template <typename Element, if_equal<Element::size, Cell> = true>
     bool get(const Link& link, Element& element) const NOEXCEPT;
 
     /// Put element at allocated link, false if serialize error.
-    template <typename Element>
+    template <typename Element, if_equal<Element::size, Cell> = true>
     bool put(const Link& link, const Element& element) NOEXCEPT;
 
 private:
@@ -121,6 +137,16 @@ class headmap
     static_assert(is_nonzero(Link::size));
     static_assert(std::atomic<cell>::is_always_lock_free);
 
+    // Links are ordered by count publication, opaque values by the cell.
+    static constexpr auto is_link = is_linkage<Value>;
+    static constexpr auto cell_size = system::possible_narrow_sign_cast<
+        ptrdiff_t>(bucket_size);
+    static constexpr auto fill = is_link ? system::bit_all<uint8_t> : uint8_t{};
+    static constexpr auto load_order = is_link ? std::memory_order_relaxed :
+        std::memory_order_acquire;
+    static constexpr auto store_order = is_link ? std::memory_order_relaxed :
+        std::memory_order_release;
+
     INLINE static auto& cell_array(memory::iterator it) NOEXCEPT
     {
         return system::unsafe_array_cast<uint8_t, bucket_size>(it);
@@ -132,11 +158,14 @@ class headmap
         return cell_array(system::pointer_cast<uint8_t>(&value));
     }
 
-    INLINE static constexpr Link to_link(cell value) NOEXCEPT
+    INLINE static constexpr Value to_value(cell value) NOEXCEPT
     {
         using namespace system;
-        return possible_narrow_cast<typename Link::integer>(
-            bit_and<cell>(Link::terminal, value));
+        if constexpr (is_link)
+            return possible_narrow_cast<typename Value::integer>(
+                bit_and<cell>(Value::terminal, value));
+        else
+            return possible_narrow_cast<Value>(value);
     }
 
     // Byte offset of bucket index within head file.
@@ -162,12 +191,16 @@ class headmap
 template <typename Schema>
 using head_map = headmap<typename Schema::link, Schema::cell>;
 
+template <typename Schema>
+using value_map = headmap<typename Schema::link, Schema::cell,
+    typename Schema::value>;
+
 } // namespace database
 } // namespace libbitcoin
 
-#define TEMPLATE template <class Link, size_t Cell, \
-    if_not_greater<Link::size, Cell> If>
-#define CLASS headmap<Link, Cell, If>
+#define TEMPLATE template <class Link, size_t Cell, class Value, \
+    if_not_greater<value_size<Value>(), std::min(Cell, Link::size)> If>
+#define CLASS headmap<Link, Cell, Value, If>
 
 #include <bitcoin/database/impl/primitives/headmap.ipp>
 

include/bitcoin/database/primitives/linkage.hpp

@@ -88,6 +88,23 @@ bool operator!=(const linkage<Size, Bits, If>& left,
     return !(left == right);
 }
 
+/// True if the type is a linkage.
+template <typename Type>
+constexpr bool is_linkage = false;
+
+template <size_t Size, size_t Bits, bool If>
+constexpr bool is_linkage<linkage<Size, Bits, If>> = true;
+
+/// Serialized bytes of a linkage or integral value.
+template <typename Value>
+constexpr size_t value_size() NOEXCEPT
+{
+    if constexpr (is_linkage<Value>)
+        return Value::size;
+    else
+        return sizeof(Value);
+}
+
 } // namespace database
 } // namespace libbitcoin