UnboundedQueue: Use hazptr_obj_batch

Summary: Manage retirement of removed segments using hazptr_obj_batch in order to reduce the chances of fragmenting related segments across thread local lists of retired objects of many threads, which could lead to unnecessarily high memory usage.

Reviewed By: djwatson

Differential Revision: D6686697

fbshipit-source-id: 0d786c0f9e0bac2c44183ed3da21619e1feb3d52

diff --git a/folly/concurrency/UnboundedQueue.h b/folly/concurrency/UnboundedQueue.h
index 22e94e076832962350beca3301a12d79628eed53..1947cdae210e1b61b81b6038b33576255a2fd59b 100644 (file)
--- a/folly/concurrency/UnboundedQueue.h
+++ b/folly/concurrency/UnboundedQueue.h
@@ -219,6 +219,7 @@ class UnboundedQueue {
   struct Consumer {
     Atom<Segment*> head;
     Atom<Ticket> ticket;
+    folly::hazptr::hazptr_obj_batch batch;
   };
   struct Producer {
     Atom<Segment*> tail;
@@ -524,8 +525,22 @@ class UnboundedQueue {
       // segment may incorrectly set head back.
       asm_volatile_pause();
     }
+    /* ***IMPORTANT*** prepReclaimSegment() must be called after
+     * confirming that head() is up-to-date and before calling
+     * setHead() to be thread-safe. */
+    /* ***IMPORTANT*** Segment s cannot be retired before the call to
+     * setHead(s). This is why prep_retire_refcounted(), which is
+     * called by prepReclaimSegment() does not retire objects, it
+     * merely adds the object to the batch and returns a private batch
+     * structure of a list of objects that can be retired later, if
+     * there are enough objects for amortizing the cost of updating
+     * the domain structure. */
+    auto res = prepReclaimSegment(s);
     setHead(next);
-    reclaimSegment(s);
+    /* Now it is safe to retire s. */
+    /* ***IMPORTANT*** The destructor of res automatically calls
+     * retire_all(), which retires to the domain any objects moved to
+     * res from batch in the call to prepReclaimSegment(). */
   }
 
   /** reclaimSegment */
@@ -537,6 +552,17 @@ class UnboundedQueue {
     }
   }
 
+  /** prepReclaimSegment */
+  folly::hazptr::hazptr_obj_batch prepReclaimSegment(Segment* s) noexcept {
+    if (SPSC) {
+      delete s;
+      /*Return an empty result; nothing more to do for this segment */
+      return folly::hazptr::hazptr_obj_batch();
+    } else {
+      return c_.batch.prep_retire_refcounted(s);
+    }
+  }
+
   FOLLY_ALWAYS_INLINE size_t index(Ticket t) const noexcept {
     return (t * Stride) & (SegmentSize - 1);
   }

diff --git a/folly/concurrency/test/DynamicBoundedQueueTest.cpp b/folly/concurrency/test/DynamicBoundedQueueTest.cpp
index bce25ec0478ea2a05107a3dd6a639d888bcb79c0..290df3daffa550bc1d671e846f1d4d06f8e7100b 100644 (file)
--- a/folly/concurrency/test/DynamicBoundedQueueTest.cpp
+++ b/folly/concurrency/test/DynamicBoundedQueueTest.cpp
@@ -100,7 +100,7 @@ TEST(DynamicBoundedQueue, size) {
   }
   {
     folly::DynamicBoundedQueue<uint64_t, false, false, false, 7, 4> q(10);
-    ASSERT_EQ(sizeof(q), 80);
+    ASSERT_EQ(sizeof(q), 80 + sizeof(folly::hazptr::hazptr_obj_batch));
   }
 }