From f54a0bdb714272ea06386dddb859258ecfacc754 Mon Sep 17 00:00:00 2001
From: Lovro Puzar <lovro@fb.com>
Date: Thu, 16 Feb 2017 03:33:51 -0800
Subject: [PATCH] Add AtomicIntrusiveLinkedList::reverseSweep()

Summary: In D4558451 I want to pull elements off an atomic list onto a thread-local (single-consumer) vector and pop to get elements in insertion order.  A sweep that provides elements in LIFO order will avoid a redundant pair of list reversals.

Reviewed By: nbronson

Differential Revision: D4564816

fbshipit-source-id: 38cf50418e6afe0be3eec96ce85d571c65f06d7e
---
 folly/AtomicIntrusiveLinkedList.h   | 41 ++++++++++++++++++++++++-----
 folly/AtomicLinkedList.h            | 22 ++++++++++++++++
 folly/test/AtomicLinkedListTest.cpp | 17 ++++++++++++
 3 files changed, 74 insertions(+), 6 deletions(-)

diff --git a/folly/AtomicIntrusiveLinkedList.h b/folly/AtomicIntrusiveLinkedList.h
index ce5c52e6..c0ee1b14 100644
--- a/folly/AtomicIntrusiveLinkedList.h
+++ b/folly/AtomicIntrusiveLinkedList.h
@@ -18,6 +18,7 @@
 
 #include <atomic>
 #include <cassert>
+#include <utility>
 
 namespace folly {
 
@@ -104,15 +105,31 @@ class AtomicIntrusiveLinkedList {
   void sweep(F&& func) {
     while (auto head = head_.exchange(nullptr)) {
       auto rhead = reverse(head);
-      while (rhead != nullptr) {
-        auto t = rhead;
-        rhead = next(t);
-        next(t) = nullptr;
-        func(t);
-      }
+      unlinkAll(rhead, std::forward<F>(func));
     }
   }
 
+  /**
+   * Similar to sweep() but calls func() on elements in LIFO order.
+   *
+   * func() is called for all elements in the list at the moment
+   * reverseSweep() is called.  Unlike sweep() it does not loop to ensure the
+   * list is empty at some point after the last invocation.  This way callers
+   * can reason about the ordering: elements inserted since the last call to
+   * reverseSweep() will be provided in LIFO order.
+   *
+   * Example: if elements are inserted in the order 1-2-3, the callback is
+   * invoked 3-2-1.  If the callback moves elements onto a stack, popping off
+   * the stack will produce the original insertion order 1-2-3.
+   */
+  template <typename F>
+  void reverseSweep(F&& func) {
+    // We don't loop like sweep() does because the overall order of callbacks
+    // would be strand-wise LIFO which is meaningless to callers.
+    auto head = head_.exchange(nullptr);
+    unlinkAll(head, std::forward<F>(func));
+  }
+
  private:
   std::atomic<T*> head_{nullptr};
 
@@ -132,6 +149,18 @@ class AtomicIntrusiveLinkedList {
     }
     return rhead;
   }
+
+  /* Unlinks all elements in the linked list fragment pointed to by `head',
+   * calling func() on every element */
+  template <typename F>
+  void unlinkAll(T* head, F&& func) {
+    while (head != nullptr) {
+      auto t = head;
+      head = next(t);
+      next(t) = nullptr;
+      func(t);
+    }
+  }
 };
 
 } // namespace folly
diff --git a/folly/AtomicLinkedList.h b/folly/AtomicLinkedList.h
index 022a7123..3ee27bbb 100644
--- a/folly/AtomicLinkedList.h
+++ b/folly/AtomicLinkedList.h
@@ -73,6 +73,28 @@ class AtomicLinkedList {
     });
   }
 
+  /**
+   * Similar to sweep() but calls func() on elements in LIFO order.
+   *
+   * func() is called for all elements in the list at the moment
+   * reverseSweep() is called.  Unlike sweep() it does not loop to ensure the
+   * list is empty at some point after the last invocation.  This way callers
+   * can reason about the ordering: elements inserted since the last call to
+   * reverseSweep() will be provided in LIFO order.
+   *
+   * Example: if elements are inserted in the order 1-2-3, the callback is
+   * invoked 3-2-1.  If the callback moves elements onto a stack, popping off
+   * the stack will produce the original insertion order 1-2-3.
+   */
+  template <typename F>
+  void reverseSweep(F&& func) {
+    list_.reverseSweep([&](Wrapper* wrapperPtr) mutable {
+      std::unique_ptr<Wrapper> wrapper(wrapperPtr);
+
+      func(std::move(wrapper->data));
+    });
+  }
+
  private:
   struct Wrapper {
     explicit Wrapper(T&& t) : data(std::move(t)) {}
diff --git a/folly/test/AtomicLinkedListTest.cpp b/folly/test/AtomicLinkedListTest.cpp
index 008b7100..303261ff 100644
--- a/folly/test/AtomicLinkedListTest.cpp
+++ b/folly/test/AtomicLinkedListTest.cpp
@@ -74,6 +74,23 @@ TEST(AtomicIntrusiveLinkedList, Basic) {
   TestIntrusiveObject::List movedList = std::move(list);
 }
 
+TEST(AtomicIntrusiveLinkedList, ReverseSweep) {
+  TestIntrusiveObject a(1), b(2), c(3);
+  TestIntrusiveObject::List list;
+  list.insertHead(&a);
+  list.insertHead(&b);
+  list.insertHead(&c);
+  size_t next_expected_id = 3;
+  list.reverseSweep([&](TestIntrusiveObject* obj) {
+    EXPECT_EQ(next_expected_id--, obj->id());
+  });
+  EXPECT_TRUE(list.empty());
+  // Test that we can still insert
+  list.insertHead(&a);
+  EXPECT_FALSE(list.empty());
+  list.reverseSweep([](TestIntrusiveObject*) {});
+}
+
 TEST(AtomicIntrusiveLinkedList, Move) {
   TestIntrusiveObject a(1), b(2);
 
-- 
2.34.1