Read the batched remote free ring's length word without domesticating it

src/snmalloc/mem/freelist.h

@@ -218,6 +218,27 @@ namespace snmalloc
           return n_tame;
         }
 
+        /**
+         * Read the next field without domesticating it.
+         *
+         * This returns exactly the value that was stored, with only the
+         * free-list encoding undone: a (possibly Wild) pointer in the common
+         * case, or the opaque bit-packed word that BatchedRemoteMessage keeps
+         * in the next field of its free ring.  Callers that want a pointer
+         * must domesticate the result (see read_next); callers reading an
+         * opaque word must not, as a domesticator may legitimately reject
+         * anything that is not an address within the heap.
+         */
+        BQueuePtr<BQueue>
+        read_next_raw(const FreeListKey& key, address_t key_tweak)
+        {
+          return Object::decode_next(
+            address_cast(&this->next_object),
+            this->next_object,
+            key,
+            key_tweak);
+        }
+
         /**
          * Read the next pointer
          */
@@ -228,11 +249,7 @@ namespace snmalloc
         BHeadPtr<BView, BQueue> read_next(
           const FreeListKey& key, address_t key_tweak, Domesticator domesticate)
         {
-          return domesticate(Object::decode_next(
-            address_cast(&this->next_object),
-            this->next_object,
-            key,
-            key_tweak));
+          return domesticate(read_next_raw(key, key_tweak));
         }
 
         /**

src/snmalloc/mem/remoteallocator.h

@@ -112,8 +112,15 @@ namespace snmalloc
       address_t key_tweak,
       Domesticator_queue domesticate)
     {
+      /*
+       * The next field of the free ring holds the bit-packed (displacement,
+       * length) word written by mk_from_freelist_builder, not a pointer, so
+       * it must be read raw rather than through the domesticator (which may
+       * reject anything outside the heap, as FixedRangeConfig's does).  The
+       * pointer derived from it below is domesticated separately.
+       */
       uintptr_t encoded =
-        m->free_ring.read_next(key, key_tweak, domesticate).unsafe_uintptr();
+        m->free_ring.read_next_raw(key, key_tweak).unsafe_uintptr();
 
       uint16_t decoded_size =
         static_cast<uint16_t>(encoded) & bits::mask_bits(MAX_CAPACITY_BITS);
@@ -155,8 +162,9 @@ namespace snmalloc
       address_t key_tweak,
       Domesticator_queue domesticate)
     {
+      // See open_free_ring: this is a bit-packed word, not a pointer.
       uintptr_t encoded =
-        m->free_ring.read_next(key, key_tweak, domesticate).unsafe_uintptr();
+        m->free_ring.read_next_raw(key, key_tweak).unsafe_uintptr();
 
       uint16_t decoded_size =
         static_cast<uint16_t>(encoded) & bits::mask_bits(MAX_CAPACITY_BITS);

src/test/func/fixed_region_remote_dealloc/fixed_region_remote_dealloc.cc

@@ -0,0 +1,87 @@
+#include "test/setup.h"
+
+#include <iostream>
+#include <snmalloc/backend/fixedglobalconfig.h>
+#include <snmalloc/snmalloc.h>
+
+#ifdef assert
+#  undef assert
+#endif
+#define assert please_use_SNMALLOC_ASSERT
+
+using namespace snmalloc;
+
+using CustomGlobals = FixedRangeConfig<PALNoAlloc<DefaultPal>>;
+using FixedAlloc = Allocator<CustomGlobals>;
+
+/**
+ * Regression test for https://github.com/microsoft/snmalloc/issues/876.
+ *
+ * FixedRangeConfig has a range-checking capptr_domesticate, which returns
+ * nullptr for anything outside the fixed region.  BatchedRemoteMessage stores
+ * a bit-packed (displacement, length) word, not a pointer, in the next field
+ * of its free ring; that word used to be passed through the domesticator when
+ * the receiving allocator opened the ring, so every ring decoded as length 0
+ * and every batched remote deallocation was lost.
+ *
+ * Allocator A allocates small objects and allocator B frees them (remotely)
+ * and flushes.  Everything allocated is freed, so the loop must be able to run
+ * well past the point at which the fixed region would be exhausted if the
+ * remote frees never made it back to A's slabs.
+ */
+int main()
+{
+  setup();
+
+  const size_t size = bits::one_at_bit(25); // 32 MiB region
+  auto base = DefaultPal::reserve(size);
+  DefaultPal::notify_using<NoZero>(base, size);
+  std::cout << "Allocated region " << base << " - "
+            << pointer_offset(base, size) << std::endl;
+
+  CustomGlobals::init(nullptr, base, size);
+
+  auto a = get_scoped_allocator<FixedAlloc>();
+  auto b = get_scoped_allocator<FixedAlloc>();
+
+  constexpr size_t object_size = 64;
+  constexpr size_t batch = 64;
+  void* objects[batch];
+
+  // Enough rounds to exhaust the region twice over if the remote frees were
+  // being lost; the bug shows up well before the first exhaustion.
+  const size_t rounds = 2 * (size / (batch * object_size));
+
+  for (size_t round = 0; round < rounds; round++)
+  {
+    for (size_t i = 0; i < batch; i++)
+    {
+      objects[i] = a->alloc(object_size);
+      if (objects[i] == nullptr)
+      {
+        std::cout << "Allocator A returned nullptr in round " << round
+                  << " after " << round * batch
+                  << " remote frees, even though everything allocated so far "
+                     "has been freed."
+                  << std::endl;
+        abort();
+      }
+      SNMALLOC_CHECK(snmalloc::is_owned<CustomGlobals>(objects[i]));
+    }
+
+    // B is not the owner of these objects, so these are remote deallocations
+    // that are batched into rings and pushed onto A's message queue by flush.
+    for (size_t i = 0; i < batch; i++)
+      b->dealloc(objects[i]);
+    b->flush();
+
+    if ((round % 1024) == 0)
+      std::cout << "." << std::flush;
+  }
+
+  std::cout << std::endl
+            << "Completed " << rounds << " rounds of " << batch
+            << " remote frees of " << object_size << "-byte objects"
+            << std::endl;
+  return 0;
+}