Improve fast path of handle remote deallocs

This adds some branch predictor and cache hints to the fast path of
processing remote deallocation. It also removes the batching.

src/mem/allocconfig.h

@@ -24,16 +24,6 @@ namespace snmalloc
     1 << 20
 #endif
     ;
-
-  // Handle at most this many object from the remote dealloc queue at a time.
-  static constexpr size_t REMOTE_BATCH =
-#ifdef USE_REMOTE_BATCH
-    REMOTE_BATCH
-#else
-    4096
-#endif
-    ;
-
   enum DecommitStrategy
   {
     /**

src/mem/corealloc.h

@@ -457,8 +457,7 @@ namespace snmalloc
         [local_state](freelist::QueuePtr p) SNMALLOC_FAST_PATH_LAMBDA {
           return capptr_domesticate<SharedStateHandle>(local_state, p);
         };
-      size_t i = 0;
-      auto cb = [this, local_state, &need_post, &i](freelist::HeadPtr msg)
+      auto cb = [this, local_state, &need_post](freelist::HeadPtr msg)
                   SNMALLOC_FAST_PATH_LAMBDA {
 #ifdef SNMALLOC_TRACING
                     std::cout << "Handling remote" << std::endl;
@@ -469,7 +468,7 @@ namespace snmalloc
 
                     handle_dealloc_remote(entry, msg.as_void(), need_post);
 
-                    return (i++ < REMOTE_BATCH);
+                    return true;
                   };
 
       if constexpr (SharedStateHandle::Options.QueueHeadsAreTame)

src/mem/remoteallocator.h

@@ -146,12 +146,21 @@ namespace snmalloc
       Cb cb)
     {
       invariant();
-      freelist::HeadPtr curr = domesticate_head(front);
-      freelist::HeadPtr next = curr->atomic_read_next(key, domesticate_queue);
 
-      while (next != nullptr)
+      // Use back to bound, so we don't handle new entries.
+      auto b = back.load(std::memory_order_relaxed);
+      freelist::HeadPtr curr = domesticate_head(front);
+
+      while (address_cast(curr) != address_cast(b))
       {
-        if (!cb(curr))
+        freelist::HeadPtr next = curr->atomic_read_next(key, domesticate_queue);
+        // We have observed a non-linearisable effect of the queue.
+        // Just go back to allocating normally.
+        if (unlikely(next == nullptr))
+          break;
+        // We want this element next, so start it loading.
+        Aal::prefetch(next.unsafe_ptr());
+        if (unlikely(!cb(curr)))
         {
           /*
            * We've domesticate_queue-d next so that we can read through it, but
@@ -168,7 +177,6 @@ namespace snmalloc
         }
 
         curr = next;
-        next = next->atomic_read_next(key, domesticate_queue);
       }
 
       /*

src/test/func/domestication/domestication.cc

@@ -154,14 +154,12 @@ int main()
    *
    *   - RemoteAllocator::dequeue domesticating the stub's next pointer (p)
    *
-   *   - RemoteAllocator::dequeue domesticating nullptr (p is the last message)
-   *
    *   - Metaslab::alloc_free_list, domesticating the successor object in the
    *     newly minted freelist::Iter (i.e., the thing that would be allocated
    *     after q).
    */
   static constexpr size_t expected_count =
-    snmalloc::CustomGlobals::Options.QueueHeadsAreTame ? 3 : 4;
+    snmalloc::CustomGlobals::Options.QueueHeadsAreTame ? 2 : 3;
   SNMALLOC_CHECK(snmalloc::CustomGlobals::domesticate_count == expected_count);
 
   // Prevent the allocators from going out of scope during the above test