hazptr: Support obj batch bypass to domain

Summary:
This diff adds support for optionally setting an obj batch such that objects retired to the batch bypass the batch and are directly pushed to the domain.
This is useful in cases where the benefits of earlier reclamation outweigh the benefits of amortizing the cost of pushing objects to the domain.

Reviewed By: davidtgoldblatt

Differential Revision: D18114444

fbshipit-source-id: a1af88dfa1b601827b2f71e700fb1dfcb08e513c

folly/synchronization/HazptrObj.h

@@ -299,12 +299,17 @@ class hazptr_obj_batch {
   SharedList l_;
   Atom<int> count_;
   bool active_;
+  bool bypass_to_domain_;
   Atom<bool> pushed_to_domain_tagged_;
 
  public:
   /** Constructor */
   hazptr_obj_batch() noexcept
-      : l_(), count_(0), active_(true), pushed_to_domain_tagged_{false} {}
+      : l_(),
+        count_(0),
+        active_(true),
+        bypass_to_domain_(false),
+        pushed_to_domain_tagged_{false} {}
 
   /** Not copyable or moveable */
   hazptr_obj_batch(const hazptr_obj_batch& o) = delete;
@@ -338,6 +343,11 @@ class hazptr_obj_batch {
     DCHECK(l_.empty());
   }
 
+  /** set_bypass_to_domain : Not thread-safe */
+  void set_bypass_to_domain() {
+    bypass_to_domain_ = true;
+  }
+
  private:
   friend class hazptr_obj<Atom>;
 
@@ -360,13 +370,19 @@ class hazptr_obj_batch {
 
   /** push_obj */
   void push_obj(Obj* obj) {
-    if (active_ && default_hazptr_domain_alive<Atom>()) {
+    obj->set_next(nullptr);
+    if (!active_ || !default_hazptr_domain_alive<Atom>()) {
+      reclaim_list(obj);
+    } else if (bypass_to_domain_) {
+      if (obj->tagged()) {
+        pushed_to_domain_tagged_.store(true, std::memory_order_relaxed);
+      }
+      hazptr_obj_list<Atom> l(obj, obj, 1);
+      hazptr_domain_push_list<Atom>(l);
+    } else {
       l_.push(obj);
       inc_count();
       check_threshold_push();
-    } else {
-      obj->set_next(nullptr);
-      reclaim_list(obj);
     }
   }
 
@@ -385,9 +401,6 @@ class hazptr_obj_batch {
 
   /** check_threshold_push */
   void check_threshold_push() {
-    if (!default_hazptr_domain_alive<Atom>()) {
-      return;
-    }
     auto c = count();
     while (c >= kThreshold) {
       if (cas_count(c, 0)) {

folly/synchronization/test/HazptrTest.cpp

@@ -814,6 +814,7 @@ template <template <typename> class Atom = std::atomic>
 void batch_test() {
   int num = 10001;
   using NodeT = Node<Atom>;
+  // untagged
   c_.clear();
   {
     hazptr_obj_batch<Atom> batch;
@@ -827,8 +828,9 @@ void batch_test() {
     DSched::join(thr);
   }
   ASSERT_EQ(c_.ctors(), num);
-  //  ASSERT_GT(c_.dtors(), 0);
   hazptr_cleanup<Atom>();
+  ASSERT_EQ(c_.dtors(), num);
+  // tagged
   c_.clear();
   {
     hazptr_obj_batch<Atom> batch;
@@ -842,8 +844,23 @@ void batch_test() {
     DSched::join(thr);
   }
   ASSERT_EQ(c_.ctors(), num);
-  ASSERT_GT(c_.dtors(), 0);
-  hazptr_cleanup<Atom>();
+  ASSERT_EQ(c_.dtors(), num);
+  // bypass to domain
+  c_.clear();
+  {
+    hazptr_obj_batch<Atom> batch;
+    batch.set_bypass_to_domain();
+    auto thr = DSched::thread([&]() {
+      for (int i = 0; i < num; ++i) {
+        auto p = new NodeT;
+        p->set_batch_tag(&batch);
+        p->retire();
+      }
+    });
+    DSched::join(thr);
+  }
+  ASSERT_EQ(c_.ctors(), num);
+  ASSERT_EQ(c_.dtors(), num);
 }
 
 template <template <typename> class Atom = std::atomic>