Implements protection on remote messages queues

This extends the freelist protection to the remote message queues. They
effectively perform doubly linked list entries for the message queue
with the enqueue operation first linking in the previous pointer, and
then then atomically setting the next.  This ensures that the visible
states always satisfy the invariant that the forward and backward
pointers are correct for any visisble object.

There is a key_global that is used for all remote deallocations. The
remote cache uses the same protection to build the temporary lists
before forwarding to the next allocator.

The mpscq is integrated into the remoteallocator as it is no longer
a reusable datastructure, but a special purpose implementation.

src/mem/remoteallocator.h

@@ -1,6 +1,5 @@
 #pragma once
 
-#include "../ds/mpscq.h"
 #include "../mem/allocconfig.h"
 #include "../mem/freelist.h"
 #include "../mem/metaslab.h"
@@ -9,59 +8,102 @@
 #include <array>
 #include <atomic>
 
-#ifdef CHECK_CLIENT
-#  define SNMALLOC_DONT_CACHE_ALLOCATOR_PTR
-#endif
-
 namespace snmalloc
 {
-  /*
-   * A region of memory destined for a remote allocator's dealloc() via the
-   * message passing system.  This structure is placed at the beginning of
-   * the allocation itself when it is queued for sending.
-   */
-  struct Remote
-  {
-    using alloc_id_t = size_t;
-    union
-    {
-      CapPtr<Remote, CBAlloc> non_atomic_next;
-      AtomicCapPtr<Remote, CBAlloc> next{nullptr};
-    };
-
-    constexpr Remote() : next(nullptr) {}
-
-    /** Zero out a Remote tracking structure, return pointer to object base */
-    template<capptr_bounds B>
-    SNMALLOC_FAST_PATH static CapPtr<void, B> clear(CapPtr<Remote, B> self)
-    {
-      pal_zero<Pal>(self, sizeof(Remote));
-      return self.as_void();
-    }
-  };
-
-  static_assert(
-    sizeof(Remote) <= MIN_ALLOC_SIZE,
-    "Needs to be able to fit in smallest allocation.");
-
   // Remotes need to be aligned enough that all the
   // small size classes can fit in the bottom bits.
   static constexpr size_t REMOTE_MIN_ALIGN = bits::min<size_t>(
     CACHELINE_SIZE, bits::next_pow2_const(NUM_SIZECLASSES + 1));
 
+  /**
+   * Global key for all remote lists.
+   */
+  inline static FreeListKey key_global(0xdeadbeef);
+
   struct alignas(REMOTE_MIN_ALIGN) RemoteAllocator
   {
-    using alloc_id_t = Remote::alloc_id_t;
+    using alloc_id_t = address_t;
+
     // Store the message queue on a separate cacheline. It is mutable data that
     // is read by other threads.
+    alignas(CACHELINE_SIZE) AtomicCapPtr<FreeObject, CBAlloc> back{nullptr};
+    // Store the two ends on different cache lines as access by different
+    // threads.
+    alignas(CACHELINE_SIZE) CapPtr<FreeObject, CBAlloc> front{nullptr};
 
-    MPSCQ<Remote, CapPtrCBAlloc, AtomicCapPtrCBAlloc> message_queue;
+    constexpr RemoteAllocator() = default;
+
+    void invariant()
+    {
+      SNMALLOC_ASSERT(back != nullptr);
+      SNMALLOC_ASSERT(front != nullptr);
+    }
+
+    void init(CapPtr<FreeObject, CBAlloc> stub)
+    {
+      stub->atomic_store_null();
+      front = stub;
+      back.store(stub, std::memory_order_relaxed);
+      invariant();
+    }
+
+    CapPtr<FreeObject, CBAlloc> destroy()
+    {
+      CapPtr<FreeObject, CBAlloc> fnt = front;
+      back.store(nullptr, std::memory_order_relaxed);
+      front = nullptr;
+      return fnt;
+    }
+
+    inline bool is_empty()
+    {
+      CapPtr<FreeObject, CBAlloc> bk = back.load(std::memory_order_relaxed);
+
+      return bk == front;
+    }
+
+    void enqueue(
+      CapPtr<FreeObject, CBAlloc> first,
+      CapPtr<FreeObject, CBAlloc> last,
+      FreeListKey& key)
+    {
+      // Pushes a list of messages to the queue. Each message from first to
+      // last should be linked together through their next pointers.
+      invariant();
+      last->atomic_store_null();
+
+      // exchange needs to be a release, so nullptr in next is visible.
+      CapPtr<FreeObject, CBAlloc> prev =
+        back.exchange(last, std::memory_order_release);
+
+      prev->atomic_store_next(first, key);
+    }
+
+    CapPtr<FreeObject, CBAlloc> peek()
+    {
+      return front;
+    }
+
+    std::pair<CapPtr<FreeObject, CBAlloc>, bool> dequeue(FreeListKey& key)
+    {
+      // Returns the front message, or null if not possible to return a message.
+      invariant();
+      CapPtr<FreeObject, CBAlloc> first = front;
+      CapPtr<FreeObject, CBAlloc> next = first->atomic_read_next(key);
+
+      if (next != nullptr)
+      {
+        front = next;
+        invariant();
+        return {first, true};
+      }
+
+      return {nullptr, false};
+    }
 
     alloc_id_t trunc_id()
     {
-      return static_cast<alloc_id_t>(
-               reinterpret_cast<uintptr_t>(&message_queue)) &
-        ~SIZECLASS_MASK;
+      return address_cast(this);
     }
   };
 } // namespace snmalloc