Merge pull request #45 from facebookresearch/constrained-lambdas

fbshipit-source-id: 6d910ea36fd6b20f97723517f5cedc472fb1ba8d

folly/experimental/pushmi/benchmarks/PushmiBenchmarks.cpp

@@ -18,6 +18,8 @@
 
 using namespace pushmi::aliases;
 
+struct inline_executor_none;
+
 template<class R>
 struct countdown {
   countdown(int& c)
@@ -41,6 +43,15 @@ using countdownsingle = countdown<mi::make_single_fn>;
 using countdownflowsingle = countdown<mi::make_flow_single_fn>;
 using countdownmany = countdown<mi::make_many_fn>;
 
+struct countdownnone {
+  countdownnone(int& c)
+      : counter(&c) {}
+
+  int* counter;
+
+  void operator()() const;
+};
+
 struct inline_executor {
     using properties = mi::property_set<mi::is_sender<>, mi::is_single<>>;
     template<class Out>
@@ -49,21 +60,29 @@ struct inline_executor {
     }
 };
 
+struct inline_executor_none {
+    using properties = mi::property_set<mi::is_sender<>, mi::is_none<>>;
+    template<class Out>
+    void submit(Out out) {
+      ::mi::set_done(out);
+    }
+};
+
+void countdownnone::operator()() const {
+  if (--*counter > 0) {
+    inline_executor_none{} | op::submit(mi::make_none(*this));
+  }
+}
+
+template<class CancellationFactory>
 struct inline_executor_flow_single {
+    CancellationFactory cf;
+
     using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_single<>>;
     template<class Out>
     void submit(Out out) {
 
-      // boolean cancellation
-      bool stop = false;
-      auto set_stop = [](auto& e) {
-        auto stop = e.lockPointerToDual();
-        if (!!stop) {
-          *stop = true;
-        }
-        e.unlockPointerToDual();
-      };
-      auto tokens = mi::entangle(stop, set_stop);
+      auto tokens = cf();
 
       using Stopper = decltype(tokens.second);
       struct Data : mi::none<> {
@@ -73,16 +92,19 @@ struct inline_executor_flow_single {
       auto up = mi::MAKE(none)(
           Data{std::move(tokens.second)},
           [](auto& data, auto e) noexcept {
-            data.stopper.t(data.stopper);
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
           },
           [](auto& data) {
-            data.stopper.t(data.stopper);
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
           });
 
     // pass reference for cancellation.
     ::mi::set_starting(out, std::move(up));
 
-    if (!tokens.first.t) {
+    auto both = lock_both(tokens.first);
+    if (!!both.first && !*(both.first)) {
       ::mi::set_value(out, *this);
     } else {
       // cancellation is not an error
@@ -91,6 +113,45 @@ struct inline_executor_flow_single {
   }
 };
 
+struct shared_cancellation_factory{
+  auto operator()(){
+    // boolean cancellation
+    mi::moving_atomic<bool> stop = false;
+    auto set_stop = [](auto& stop) {
+      if (!!stop) {
+        stop->store(true);
+      }
+    };
+    return mi::shared_entangle(std::move(stop), set_stop);
+  }
+};
+using inline_executor_flow_single_shared = inline_executor_flow_single<shared_cancellation_factory>;
+
+struct entangled_cancellation_factory{
+  auto operator()(){
+    // boolean cancellation
+    mi::moving_atomic<bool> stop = false;
+    auto set_stop = [](auto& stop) {
+      if (!!stop) {
+        stop->store(true);
+      }
+    };
+    return mi::entangle(std::move(stop), set_stop);
+  }
+};
+using inline_executor_flow_single_entangled = inline_executor_flow_single<entangled_cancellation_factory>;
+
+struct inline_executor_flow_single_ignore {
+    using properties = mi::property_set<mi::is_sender<>, mi::is_flow<>, mi::is_single<>>;
+    template<class Out>
+    void submit(Out out) {
+      // pass reference for cancellation.
+      ::mi::set_starting(out, mi::none<>{});
+
+      ::mi::set_value(out, *this);
+    }
+};
+
 struct inline_executor_many {
     using properties = mi::property_set<mi::is_sender<>, mi::is_many<>>;
     template<class Out>
@@ -104,6 +165,18 @@ struct inline_executor_many {
 #define concept Concept
 #include <nonius/nonius.h++>
 
+NONIUS_BENCHMARK("inline 10 none", [](nonius::chronometer meter){
+  int counter = 0;
+  auto ie = inline_executor_none{};
+  using IE = decltype(ie);
+  countdownnone none{counter};
+  meter.measure([&]{
+    counter = 10;
+    ie | op::submit(mi::make_none(none));
+    return counter;
+  });
+})
+
 NONIUS_BENCHMARK("inline 10 single", [](nonius::chronometer meter){
   int counter = 0;
   auto ie = inline_executor{};
@@ -116,26 +189,50 @@ NONIUS_BENCHMARK("inline 10 single", [](nonius::chronometer meter){
   });
 })
 
-NONIUS_BENCHMARK("inline 10 flow_single", [](nonius::chronometer meter){
+NONIUS_BENCHMARK("inline 10 many", [](nonius::chronometer meter){
   int counter = 0;
-  auto ie = inline_executor_flow_single{};
+  auto ie = inline_executor_many{};
+  using IE = decltype(ie);
+  countdownmany many{counter};
+  meter.measure([&]{
+    counter = 10;
+    ie | op::submit(mi::make_many(many));
+    return counter;
+  });
+})
+
+NONIUS_BENCHMARK("inline 10 flow_single shared", [](nonius::chronometer meter){
+  int counter = 0;
+  auto ie = inline_executor_flow_single_shared{};
   using IE = decltype(ie);
   countdownflowsingle flowsingle{counter};
   meter.measure([&]{
     counter = 10;
-    ie | op::submit(mi::make_flow_single(on_value(flowsingle)));
+    ie | op::submit(mi::make_flow_single(flowsingle));
     return counter;
   });
 })
 
-NONIUS_BENCHMARK("inline 10 many", [](nonius::chronometer meter){
+NONIUS_BENCHMARK("inline 10 flow_single entangle", [](nonius::chronometer meter){
   int counter = 0;
-  auto ie = inline_executor_many{};
+  auto ie = inline_executor_flow_single_entangled{};
   using IE = decltype(ie);
-  countdownmany many{counter};
+  countdownflowsingle flowsingle{counter};
   meter.measure([&]{
     counter = 10;
-    ie | op::submit(mi::make_many(many));
+    ie | op::submit(mi::make_flow_single(flowsingle));
+    return counter;
+  });
+})
+
+NONIUS_BENCHMARK("inline 10 flow_single ignore cancellation", [](nonius::chronometer meter){
+  int counter = 0;
+  auto ie = inline_executor_flow_single_ignore{};
+  using IE = decltype(ie);
+  countdownflowsingle flowsingle{counter};
+  meter.measure([&]{
+    counter = 10;
+    ie | op::submit(mi::make_flow_single(flowsingle));
     return counter;
   });
 })

folly/experimental/pushmi/include/pushmi.h

@@ -1205,43 +1205,40 @@ namespace detail {
 
 namespace pushmi {
 
-#if 0
 
-template <class T, class Dual>
-struct entangled {
-  T t;
-  entangled<Dual, T>* dual;
-
-  ~entangled() {
-    if (!!dual) {
-      dual->dual = nullptr;
-    }
-  }
-  explicit entangled(T t) : t(std::move(t)), dual(nullptr) {}
-  entangled(entangled&& o) : t(std::move(o.t)), dual(o.dual) {
-    o.dual = nullptr;
-    if (!!dual) {
-      dual->dual = this;
-    }
-  }
-
-  entangled() = delete;
-  entangled(const entangled&) = delete;
-  entangled& operator=(const entangled&) = delete;
-  entangled& operator=(entangled&&) = delete;
-
-  Dual* lockPointerToDual() {
-    if (!!dual) {
-      return std::addressof(dual->t);
-    }
-    return nullptr;
-  }
-
-  void unlockPointerToDual() {
-  }
-};
-
-#else
+// template <class T, class Dual>
+// struct entangled {
+//   T t;
+//   entangled<Dual, T>* dual;
+//
+//   ~entangled() {
+//     if (!!dual) {
+//       dual->dual = nullptr;
+//     }
+//   }
+//   explicit entangled(T t) : t(std::move(t)), dual(nullptr) {}
+//   entangled(entangled&& o) : t(std::move(o.t)), dual(o.dual) {
+//     o.dual = nullptr;
+//     if (!!dual) {
+//       dual->dual = this;
+//     }
+//   }
+//
+//   entangled() = delete;
+//   entangled(const entangled&) = delete;
+//   entangled& operator=(const entangled&) = delete;
+//   entangled& operator=(entangled&&) = delete;
+//
+//   Dual* lockPointerToDual() {
+//     if (!!dual) {
+//       return std::addressof(dual->t);
+//     }
+//     return nullptr;
+//   }
+//
+//   void unlockPointerToDual() {
+//   }
+// };
 
 // This class can be used to keep a pair of values with pointers to each other
 // in sync, even when both objects are allowed to move. Ordinarily you'd do this
@@ -1284,7 +1281,7 @@ struct entangled {
     // have to hold *both* locks to write any of either entangled object's
     // metadata, but need only one to read it.
     int expected = kUnlocked;
-    if (!stateMachine.compare_exchange_weak(expected, kLocked)) {
+    if (!stateMachine.compare_exchange_weak(expected, kLocked, std::memory_order_seq_cst, std::memory_order_relaxed)) {
       return false;
     }
     // Having *either* object local-locked protects the data in both objects.
@@ -1293,7 +1290,7 @@ struct entangled {
       return true;
     }
     expected = kUnlocked;
-    if (dual->stateMachine.compare_exchange_strong(expected, kLocked)) {
+    if (dual->stateMachine.compare_exchange_strong(expected, kLocked, std::memory_order_seq_cst)) {
       return true;
     }
     // We got here, and so hit the race; we're deadlocked if we stick to
@@ -1305,15 +1302,15 @@ struct entangled {
     if ((uintptr_t)this < (uintptr_t)dual) {
       // I get to win the race. I'll acquire the locks, but have to make sure
       // my memory stays valid until the other thread acknowledges its loss.
-      while (stateMachine.load() != kLockedAndLossAcknowledged) {
+      while (stateMachine.load(std::memory_order_relaxed) != kLockedAndLossAcknowledged) {
         // Spin.
       }
-      stateMachine.store(kLocked);
+      stateMachine.store(kLocked, std::memory_order_relaxed);
       return true;
     } else {
       // I lose the race, but have to coordinate with the winning thread, so
       // that it knows that I'm not about to try to touch it's data
-      dual->stateMachine.store(kLockedAndLossAcknowledged);
+      dual->stateMachine.store(kLockedAndLossAcknowledged, std::memory_order_relaxed);
       return false;
     }
   }
@@ -1336,9 +1333,9 @@ struct entangled {
     // other object so long as its locked. Going in the other order could let
     // another thread incorrectly think we're going down the deadlock-avoidance
     // path in tryLock().
-    stateMachine.store(kUnlocked);
+    stateMachine.store(kUnlocked, std::memory_order_release);
     if (dual != nullptr) {
-      dual->stateMachine.store(kUnlocked);
+      dual->stateMachine.store(kUnlocked, std::memory_order_release);
     }
   }
 
@@ -1392,11 +1389,13 @@ struct entangled {
     unlockBoth();
   }
 };
-#endif
+
+template <class First, class Second>
+using entangled_pair = std::pair<entangled<First, Second>, entangled<Second, First>>;
 
 template <class First, class Second>
 auto entangle(First f, Second s)
-    -> std::pair<entangled<First, Second>, entangled<Second, First>> {
+    -> entangled_pair<First, Second> {
   entangled<First, Second> ef(std::move(f));
   entangled<Second, First> es(std::move(s));
   ef.dual = std::addressof(es);
@@ -1404,6 +1403,85 @@ auto entangle(First f, Second s)
   return {std::move(ef), std::move(es)};
 }
 
+template <class T, class Dual>
+struct locked_entangled_pair : std::pair<T*, Dual*> {
+  entangled<T, Dual>* e;
+  ~locked_entangled_pair(){if (!!e) {e->unlockBoth();}}
+  explicit locked_entangled_pair(entangled<T, Dual>& e) : e(std::addressof(e)){
+    this->e->lockBoth();
+    this->first = std::addressof(this->e->t);
+    this->second = !!this->e->dual ? std::addressof(this->e->dual->t) : nullptr;
+  }
+  locked_entangled_pair() = delete;
+  locked_entangled_pair(const locked_entangled_pair&) = delete;
+  locked_entangled_pair& operator=(const locked_entangled_pair&) = delete;
+  locked_entangled_pair(locked_entangled_pair&& o) : std::pair<T*, Dual*>(o), e(o.e) {o.e = nullptr;};
+  locked_entangled_pair& operator=(locked_entangled_pair&& o){
+    static_cast<std::pair<T*, Dual*>&>(*this) = static_cast<std::pair<T*, Dual*>&&>(o);
+    e = o.e;
+    o.e = nullptr;
+    return *this;
+  }
+};
+
+template <class T, class Dual>
+locked_entangled_pair<T, Dual> lock_both(entangled<T, Dual>& e){
+  return locked_entangled_pair<T, Dual>{e};
+}
+
+template <class T, class Dual>
+struct shared_entangled : std::shared_ptr<T> {
+  Dual* dual;
+
+  template<class P>
+  explicit shared_entangled(std::shared_ptr<P>& p, T& t, Dual& d) : std::shared_ptr<T>(p, std::addressof(t)), dual(std::addressof(d)){
+  }
+  shared_entangled() = delete;
+  shared_entangled(const shared_entangled&) = delete;
+  shared_entangled& operator=(const shared_entangled&) = delete;
+  shared_entangled(shared_entangled&&) = default;
+  shared_entangled& operator=(shared_entangled&&) = default;
+};
+
+template <class First, class Second>
+using shared_entangled_pair = std::pair<shared_entangled<First, Second>, shared_entangled<Second, First>>;
+
+template <class First, class Second>
+auto shared_entangle(First f, Second s)
+    -> shared_entangled_pair<First, Second> {
+  auto p = std::make_shared<std::pair<First, Second>>(std::move(f), std::move(s));
+  shared_entangled<First, Second> ef(p, p->first, p->second);
+  shared_entangled<Second, First> es(p, p->second, p->first);
+  return {std::move(ef), std::move(es)};
+}
+
+template <class T, class Dual>
+struct locked_shared_entangled_pair : std::pair<T*, Dual*> {
+  shared_entangled<T, Dual> e;
+  explicit locked_shared_entangled_pair(shared_entangled<T, Dual>& e) : e(std::move(e)){
+    this->first = this->e.get();
+    this->second = this->e.dual;
+  }
+  locked_shared_entangled_pair() = delete;
+  locked_shared_entangled_pair(const locked_shared_entangled_pair&) = delete;
+  locked_shared_entangled_pair& operator=(const locked_shared_entangled_pair&) = delete;
+  locked_shared_entangled_pair(locked_shared_entangled_pair&&) = default;
+  locked_shared_entangled_pair& operator=(locked_shared_entangled_pair&&) = default;
+};
+
+
+template <class T, class Dual>
+locked_shared_entangled_pair<T, Dual> lock_both(shared_entangled<T, Dual>& e){
+  return locked_shared_entangled_pair<T, Dual>{e};
+}
+
+// external synchronization required for move and delete
+template<class T>
+struct moving_atomic : std::atomic<T> {
+  using std::atomic<T>::atomic;
+  moving_atomic(moving_atomic&& o) : std::atomic<T>(o.load()) {}
+};
+
 } // namespace pushmi
 //#pragma once
 // Copyright (c) 2018-present, Facebook, Inc.
@@ -1441,8 +1519,8 @@ void set_next(S& s, V&& v) noexcept(noexcept(s.next((V&&) v))) {
 
 PUSHMI_TEMPLATE (class S, class Up)
   (requires requires (std::declval<S&>().starting(std::declval<Up>())))
-void set_starting(S& s, Up up) noexcept(noexcept(s.starting(std::move(up)))) {
-  s.starting(std::move(up));
+void set_starting(S& s, Up&& up) noexcept(noexcept(s.starting((Up&&) up))) {
+  s.starting((Up&&) up);
 }
 
 PUSHMI_TEMPLATE (class SD, class Out)
@@ -1515,9 +1593,9 @@ void set_next(std::reference_wrapper<S> s, V&& v) noexcept(
 }
 PUSHMI_TEMPLATE (class S, class Up)
   (requires requires ( set_starting(std::declval<S&>(), std::declval<Up>()) ))
-void set_starting(std::reference_wrapper<S> s, Up up) noexcept(
-  noexcept(set_starting(s.get(), std::move(up)))) {
-  set_starting(s.get(), std::move(up));
+void set_starting(std::reference_wrapper<S> s, Up&& up) noexcept(
+  noexcept(set_starting(s.get(), (Up&&) up))) {
+  set_starting(s.get(), (Up&&) up);
 }
 PUSHMI_TEMPLATE (class SD, class Out)
   (requires requires ( submit(std::declval<SD&>(), std::declval<Out>()) ))
@@ -1604,10 +1682,10 @@ struct set_starting_fn {
       set_starting(std::declval<S&>(), std::declval<Up>()),
       set_error(std::declval<S&>(), std::current_exception())
     ))
-  void operator()(S&& s, Up up) const
-      noexcept(noexcept(set_starting(s, std::move(up)))) {
+  void operator()(S&& s, Up&& up) const
+      noexcept(noexcept(set_starting(s, (Up&&) up))) {
     try {
-      set_starting(s, std::move(up));
+      set_starting(s, (Up&&) up);
     } catch (...) {
       set_error(s, std::current_exception());
     }
@@ -2361,8 +2439,8 @@ struct passDStrtF {
     (requires requires (
       ::pushmi::set_starting(std::declval<Data&>(), std::declval<Up>())
     ) && Receiver<Data>)
-  void operator()(Data& out, Up up) const {
-    ::pushmi::set_starting(out, std::move(up));
+  void operator()(Data& out, Up&& up) const {
+    ::pushmi::set_starting(out, (Up&&) up);
   }
 };
 

folly/experimental/pushmi/include/pushmi/boosters.h

@@ -117,8 +117,8 @@ struct passDStrtF {
     (requires requires (
       ::pushmi::set_starting(std::declval<Data&>(), std::declval<Up>())
     ) && Receiver<Data>)
-  void operator()(Data& out, Up up) const {
-    ::pushmi::set_starting(out, std::move(up));
+  void operator()(Data& out, Up&& up) const {
+    ::pushmi::set_starting(out, (Up&&) up);
   }
 };
 

folly/experimental/pushmi/include/pushmi/entangle.h

@@ -8,43 +8,40 @@
 
 namespace pushmi {
 
-#if 0
 
-template <class T, class Dual>
-struct entangled {
-  T t;
-  entangled<Dual, T>* dual;
-
-  ~entangled() {
-    if (!!dual) {
-      dual->dual = nullptr;
-    }
-  }
-  explicit entangled(T t) : t(std::move(t)), dual(nullptr) {}
-  entangled(entangled&& o) : t(std::move(o.t)), dual(o.dual) {
-    o.dual = nullptr;
-    if (!!dual) {
-      dual->dual = this;
-    }
-  }
-
-  entangled() = delete;
-  entangled(const entangled&) = delete;
-  entangled& operator=(const entangled&) = delete;
-  entangled& operator=(entangled&&) = delete;
-
-  Dual* lockPointerToDual() {
-    if (!!dual) {
-      return std::addressof(dual->t);
-    }
-    return nullptr;
-  }
-
-  void unlockPointerToDual() {
-  }
-};
-
-#else
+// template <class T, class Dual>
+// struct entangled {
+//   T t;
+//   entangled<Dual, T>* dual;
+//
+//   ~entangled() {
+//     if (!!dual) {
+//       dual->dual = nullptr;
+//     }
+//   }
+//   explicit entangled(T t) : t(std::move(t)), dual(nullptr) {}
+//   entangled(entangled&& o) : t(std::move(o.t)), dual(o.dual) {
+//     o.dual = nullptr;
+//     if (!!dual) {
+//       dual->dual = this;
+//     }
+//   }
+//
+//   entangled() = delete;
+//   entangled(const entangled&) = delete;
+//   entangled& operator=(const entangled&) = delete;
+//   entangled& operator=(entangled&&) = delete;
+//
+//   Dual* lockPointerToDual() {
+//     if (!!dual) {
+//       return std::addressof(dual->t);
+//     }
+//     return nullptr;
+//   }
+//
+//   void unlockPointerToDual() {
+//   }
+// };
 
 // This class can be used to keep a pair of values with pointers to each other
 // in sync, even when both objects are allowed to move. Ordinarily you'd do this
@@ -87,7 +84,7 @@ struct entangled {
     // have to hold *both* locks to write any of either entangled object's
     // metadata, but need only one to read it.
     int expected = kUnlocked;
-    if (!stateMachine.compare_exchange_weak(expected, kLocked)) {
+    if (!stateMachine.compare_exchange_weak(expected, kLocked, std::memory_order_seq_cst, std::memory_order_relaxed)) {
       return false;
     }
     // Having *either* object local-locked protects the data in both objects.
@@ -96,7 +93,7 @@ struct entangled {
       return true;
     }
     expected = kUnlocked;
-    if (dual->stateMachine.compare_exchange_strong(expected, kLocked)) {
+    if (dual->stateMachine.compare_exchange_strong(expected, kLocked, std::memory_order_seq_cst)) {
       return true;
     }
     // We got here, and so hit the race; we're deadlocked if we stick to
@@ -108,15 +105,15 @@ struct entangled {
     if ((uintptr_t)this < (uintptr_t)dual) {
       // I get to win the race. I'll acquire the locks, but have to make sure
       // my memory stays valid until the other thread acknowledges its loss.
-      while (stateMachine.load() != kLockedAndLossAcknowledged) {
+      while (stateMachine.load(std::memory_order_relaxed) != kLockedAndLossAcknowledged) {
         // Spin.
       }
-      stateMachine.store(kLocked);
+      stateMachine.store(kLocked, std::memory_order_relaxed);
       return true;
     } else {
       // I lose the race, but have to coordinate with the winning thread, so
       // that it knows that I'm not about to try to touch it's data
-      dual->stateMachine.store(kLockedAndLossAcknowledged);
+      dual->stateMachine.store(kLockedAndLossAcknowledged, std::memory_order_relaxed);
       return false;
     }
   }
@@ -139,9 +136,9 @@ struct entangled {
     // other object so long as its locked. Going in the other order could let
     // another thread incorrectly think we're going down the deadlock-avoidance
     // path in tryLock().
-    stateMachine.store(kUnlocked);
+    stateMachine.store(kUnlocked, std::memory_order_release);
     if (dual != nullptr) {
-      dual->stateMachine.store(kUnlocked);
+      dual->stateMachine.store(kUnlocked, std::memory_order_release);
     }
   }
 
@@ -195,11 +192,13 @@ struct entangled {
     unlockBoth();
   }
 };
-#endif
+
+template <class First, class Second>
+using entangled_pair = std::pair<entangled<First, Second>, entangled<Second, First>>;
 
 template <class First, class Second>
 auto entangle(First f, Second s)
-    -> std::pair<entangled<First, Second>, entangled<Second, First>> {
+    -> entangled_pair<First, Second> {
   entangled<First, Second> ef(std::move(f));
   entangled<Second, First> es(std::move(s));
   ef.dual = std::addressof(es);
@@ -207,4 +206,83 @@ auto entangle(First f, Second s)
   return {std::move(ef), std::move(es)};
 }
 
+template <class T, class Dual>
+struct locked_entangled_pair : std::pair<T*, Dual*> {
+  entangled<T, Dual>* e;
+  ~locked_entangled_pair(){if (!!e) {e->unlockBoth();}}
+  explicit locked_entangled_pair(entangled<T, Dual>& e) : e(std::addressof(e)){
+    this->e->lockBoth();
+    this->first = std::addressof(this->e->t);
+    this->second = !!this->e->dual ? std::addressof(this->e->dual->t) : nullptr;
+  }
+  locked_entangled_pair() = delete;
+  locked_entangled_pair(const locked_entangled_pair&) = delete;
+  locked_entangled_pair& operator=(const locked_entangled_pair&) = delete;
+  locked_entangled_pair(locked_entangled_pair&& o) : std::pair<T*, Dual*>(o), e(o.e) {o.e = nullptr;};
+  locked_entangled_pair& operator=(locked_entangled_pair&& o){
+    static_cast<std::pair<T*, Dual*>&>(*this) = static_cast<std::pair<T*, Dual*>&&>(o);
+    e = o.e;
+    o.e = nullptr;
+    return *this;
+  }
+};
+
+template <class T, class Dual>
+locked_entangled_pair<T, Dual> lock_both(entangled<T, Dual>& e){
+  return locked_entangled_pair<T, Dual>{e};
+}
+
+template <class T, class Dual>
+struct shared_entangled : std::shared_ptr<T> {
+  Dual* dual;
+
+  template<class P>
+  explicit shared_entangled(std::shared_ptr<P>& p, T& t, Dual& d) : std::shared_ptr<T>(p, std::addressof(t)), dual(std::addressof(d)){
+  }
+  shared_entangled() = delete;
+  shared_entangled(const shared_entangled&) = delete;
+  shared_entangled& operator=(const shared_entangled&) = delete;
+  shared_entangled(shared_entangled&&) = default;
+  shared_entangled& operator=(shared_entangled&&) = default;
+};
+
+template <class First, class Second>
+using shared_entangled_pair = std::pair<shared_entangled<First, Second>, shared_entangled<Second, First>>;
+
+template <class First, class Second>
+auto shared_entangle(First f, Second s)
+    -> shared_entangled_pair<First, Second> {
+  auto p = std::make_shared<std::pair<First, Second>>(std::move(f), std::move(s));
+  shared_entangled<First, Second> ef(p, p->first, p->second);
+  shared_entangled<Second, First> es(p, p->second, p->first);
+  return {std::move(ef), std::move(es)};
+}
+
+template <class T, class Dual>
+struct locked_shared_entangled_pair : std::pair<T*, Dual*> {
+  shared_entangled<T, Dual> e;
+  explicit locked_shared_entangled_pair(shared_entangled<T, Dual>& e) : e(std::move(e)){
+    this->first = this->e.get();
+    this->second = this->e.dual;
+  }
+  locked_shared_entangled_pair() = delete;
+  locked_shared_entangled_pair(const locked_shared_entangled_pair&) = delete;
+  locked_shared_entangled_pair& operator=(const locked_shared_entangled_pair&) = delete;
+  locked_shared_entangled_pair(locked_shared_entangled_pair&&) = default;
+  locked_shared_entangled_pair& operator=(locked_shared_entangled_pair&&) = default;
+};
+
+
+template <class T, class Dual>
+locked_shared_entangled_pair<T, Dual> lock_both(shared_entangled<T, Dual>& e){
+  return locked_shared_entangled_pair<T, Dual>{e};
+}
+
+// external synchronization required for move and delete
+template<class T>
+struct moving_atomic : std::atomic<T> {
+  using std::atomic<T>::atomic;
+  moving_atomic(moving_atomic&& o) : std::atomic<T>(o.load()) {}
+};
+
 } // namespace pushmi

folly/experimental/pushmi/include/pushmi/extension_points.h

@@ -34,8 +34,8 @@ void set_next(S& s, V&& v) noexcept(noexcept(s.next((V&&) v))) {
 
 PUSHMI_TEMPLATE (class S, class Up)
   (requires requires (std::declval<S&>().starting(std::declval<Up>())))
-void set_starting(S& s, Up up) noexcept(noexcept(s.starting(std::move(up)))) {
-  s.starting(std::move(up));
+void set_starting(S& s, Up&& up) noexcept(noexcept(s.starting((Up&&) up))) {
+  s.starting((Up&&) up);
 }
 
 PUSHMI_TEMPLATE (class SD, class Out)
@@ -108,9 +108,9 @@ void set_next(std::reference_wrapper<S> s, V&& v) noexcept(
 }
 PUSHMI_TEMPLATE (class S, class Up)
   (requires requires ( set_starting(std::declval<S&>(), std::declval<Up>()) ))
-void set_starting(std::reference_wrapper<S> s, Up up) noexcept(
-  noexcept(set_starting(s.get(), std::move(up)))) {
-  set_starting(s.get(), std::move(up));
+void set_starting(std::reference_wrapper<S> s, Up&& up) noexcept(
+  noexcept(set_starting(s.get(), (Up&&) up))) {
+  set_starting(s.get(), (Up&&) up);
 }
 PUSHMI_TEMPLATE (class SD, class Out)
   (requires requires ( submit(std::declval<SD&>(), std::declval<Out>()) ))
@@ -197,10 +197,10 @@ struct set_starting_fn {
       set_starting(std::declval<S&>(), std::declval<Up>()),
       set_error(std::declval<S&>(), std::current_exception())
     ))
-  void operator()(S&& s, Up up) const
-      noexcept(noexcept(set_starting(s, std::move(up)))) {
+  void operator()(S&& s, Up&& up) const
+      noexcept(noexcept(set_starting(s, (Up&&) up))) {
     try {
-      set_starting(s, std::move(up));
+      set_starting(s, (Up&&) up);
     } catch (...) {
       set_error(s, std::current_exception());
     }

folly/experimental/pushmi/test/FlowTest.cpp

@@ -28,15 +28,13 @@ SCENARIO("flow single immediate cancellation", "[flow][deferred]") {
   GIVEN("A flow single deferred") {
     auto f = mi::MAKE(flow_single_deferred)([&](auto out) {
       // boolean cancellation
-      bool stop = false;
-      auto set_stop = [](auto& e) {
-        auto stop = e.lockPointerToDual();
+      mi::moving_atomic<bool> stop = false;
+      auto set_stop = [](auto& stop) {
         if (!!stop) {
-          *stop = true;
+          stop->store(true);
         }
-        e.unlockPointerToDual();
       };
-      auto tokens = mi::entangle(stop, set_stop);
+      auto tokens = mi::shared_entangle(std::move(stop), set_stop);
 
       using Stopper = decltype(tokens.second);
       struct Data : mi::none<> {
@@ -47,18 +45,21 @@ SCENARIO("flow single immediate cancellation", "[flow][deferred]") {
           Data{std::move(tokens.second)},
           [&](auto& data, auto e) noexcept {
             signals += 100000;
-            data.stopper.t(data.stopper);
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
           },
           [&](auto& data) {
             signals += 10000;
-            data.stopper.t(data.stopper);
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
           });
 
       // pass reference for cancellation.
       ::mi::set_starting(out, std::move(up));
 
       // check boolean to select signal
-      if (!tokens.first.t) {
+      auto both = lock_both(tokens.first);
+      if (!!both.first && !*(both.first)) {
         ::mi::set_value(out, 42);
       } else {
         // cancellation is not an error
@@ -109,15 +110,13 @@ SCENARIO("flow single cancellation trampoline", "[flow][deferred]") {
   GIVEN("A flow single deferred") {
     auto f = mi::MAKE(flow_single_deferred)([&](auto out) {
       // boolean cancellation
-      bool stop = false;
-      auto set_stop = [](auto& e) {
-        auto stop = e.lockPointerToDual();
+      mi::moving_atomic<bool> stop = false;
+      auto set_stop = [](auto& stop) {
         if (!!stop) {
-          *stop = true;
+          stop->store(true);
         }
-        e.unlockPointerToDual();
       };
-      auto tokens = mi::entangle(stop, set_stop);
+      auto tokens = mi::shared_entangle(std::move(stop), set_stop);
 
       using Stopper = decltype(tokens.second);
       struct Data : mi::none<> {
@@ -128,11 +127,13 @@ SCENARIO("flow single cancellation trampoline", "[flow][deferred]") {
           Data{std::move(tokens.second)},
           [&](auto& data, auto e) noexcept {
             signals += 100000;
-            data.stopper.t(data.stopper);
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
           },
           [&](auto& data) {
             signals += 10000;
-            data.stopper.t(data.stopper);
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
           });
 
       tr |
@@ -149,7 +150,8 @@ SCENARIO("flow single cancellation trampoline", "[flow][deferred]") {
                     [out = std::move(out),
                      stoppee = std::move(stoppee)](auto) mutable {
                       // check boolean to select signal
-                      if (!stoppee.t) {
+                      auto both = lock_both(stoppee);
+                      if (!!both.first && !*(both.first)) {
                         ::mi::set_value(out, 42);
                       } else {
                         // cancellation is not an error
@@ -211,7 +213,177 @@ struct moving_atomic : std::atomic<T> {
   moving_atomic(moving_atomic&& o) : std::atomic<T>(o.load()) {}
 };
 
-SCENARIO("flow single cancellation new thread", "[flow][deferred]") {
+SCENARIO("flow single shared cancellation new thread", "[flow][deferred]") {
+  auto nt = mi::new_thread();
+  using NT = decltype(nt);
+  std::atomic<int> signals{0};
+  auto at = nt.now() + 200ms;
+
+  GIVEN("A flow single deferred") {
+    auto f = mi::MAKE(flow_single_deferred)([&](auto out) {
+      // boolean cancellation
+      mi::moving_atomic<bool> stop = false;
+      auto set_stop = [](auto& stop) {
+        if (!!stop) {
+          stop->store(true);
+        }
+      };
+      auto tokens = mi::shared_entangle(std::move(stop), set_stop);
+
+      using Stopper = decltype(tokens.second);
+      struct Data : mi::none<> {
+        explicit Data(Stopper stopper) : stopper(std::move(stopper)) {}
+        Stopper stopper;
+      };
+      auto up = mi::MAKE(none)(
+          Data{std::move(tokens.second)},
+          [&](auto& data, auto e) noexcept {
+            signals += 100000;
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
+          },
+          [&](auto& data) {
+            signals += 10000;
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
+          });
+
+      // make all the signals come from the same thread
+      nt |
+          op::submit([stoppee = std::move(tokens.first),
+                      up = std::move(up),
+                      out = std::move(out),
+                      at](auto nt) mutable {
+            // pass reference for cancellation.
+            ::mi::set_starting(out, std::move(up));
+
+            // submit work to happen later
+            nt |
+                op::submit_at(
+                    at,
+                    [stoppee = std::move(stoppee),
+                     out = std::move(out)](auto) mutable {
+                      // check boolean to select signal
+                      auto both = lock_both(stoppee);
+                      if (!!both.first && !*(both.first)) {
+                        ::mi::set_value(out, 42);
+                      } else {
+                        // cancellation is not an error
+                        ::mi::set_done(out);
+                      }
+                    });
+          });
+    });
+
+    WHEN("submit is applied and cancels the producer early") {
+      {
+      f |
+          op::blocking_submit(
+              mi::on_value([&](int) { signals += 100; }),
+              mi::on_error([&](auto) noexcept { signals += 1000; }),
+              mi::on_done([&]() { signals += 1; }),
+              // stop producer before it is scheduled to run
+              mi::on_starting([&](auto up) {
+                signals += 10;
+                nt |
+                    op::submit_at(
+                        at - 50ms, [up = std::move(up)](auto) mutable {
+                          ::mi::set_done(up);
+                        });
+              }));
+      }
+
+      // make sure that the completion signal arrives
+      std::this_thread::sleep_for(100ms);
+
+      THEN(
+          "the starting, up.done and out.done signals are each recorded once") {
+        REQUIRE(signals == 10011);
+      }
+    }
+
+    WHEN("submit is applied and cancels the producer late") {
+      {
+      f |
+          op::blocking_submit(
+              mi::on_value([&](int) { signals += 100; }),
+              mi::on_error([&](auto) noexcept { signals += 1000; }),
+              mi::on_done([&]() { signals += 1; }),
+              // do not stop producer before it is scheduled to run
+              mi::on_starting([&](auto up) {
+                signals += 10;
+                nt |
+                    op::submit_at(
+                        at + 50ms, [up = std::move(up)](auto) mutable {
+                          ::mi::set_done(up);
+                        });
+              }));
+      }
+
+      std::this_thread::sleep_for(100ms);
+
+      THEN(
+          "the starting, up.done and out.value signals are each recorded once") {
+        REQUIRE(signals == 10110);
+      }
+    }
+
+    WHEN("submit is applied and cancels the producer at the same time") {
+      // count known results
+      int total = 0;
+      int cancellostrace = 0; // 10110
+      int cancelled = 0; // 10011
+
+      for (;;) {
+        signals = 0;
+        // set completion time to be in 100ms
+        at = nt.now() + 100ms;
+        {
+        f |
+            op::blocking_submit(
+                mi::on_value([&](int) { signals += 100; }),
+                mi::on_error([&](auto) noexcept { signals += 1000; }),
+                mi::on_done([&]() { signals += 1; }),
+                // stop producer at the same time that it is scheduled to run
+                mi::on_starting([&](auto up) {
+                  signals += 10;
+                  nt | op::submit_at(at, [up = std::move(up)](auto) mutable {
+                    ::mi::set_done(up);
+                  });
+                }));
+        }
+
+        // make sure any cancellation signal has completed
+        std::this_thread::sleep_for(10ms);
+
+        // accumulate known signals
+        ++total;
+        cancellostrace += signals == 10110;
+        cancelled += signals == 10011;
+
+        if (total != cancellostrace + cancelled) {
+          // display the unrecognized signals recorded
+          REQUIRE(signals == -1);
+        }
+        if (total >= 100) {
+          // too long, abort and show the signals distribution
+          WARN(
+              "total " << total << ", cancel-lost-race " << cancellostrace
+                       << ", cancelled " << cancelled);
+          break;
+        }
+        if (cancellostrace > 4 && cancelled > 4) {
+          // yay all known outcomes were observed!
+          break;
+        }
+        // try again
+        continue;
+      }
+    }
+  }
+}
+
+SCENARIO("flow single entangled cancellation new thread", "[flow][deferred]") {
   auto nt = mi::new_thread();
   using NT = decltype(nt);
   std::atomic<int> signals{0};
@@ -220,15 +392,13 @@ SCENARIO("flow single cancellation new thread", "[flow][deferred]") {
   GIVEN("A flow single deferred") {
     auto f = mi::MAKE(flow_single_deferred)([&](auto out) {
       // boolean cancellation
-      moving_atomic<bool> stop = false;
-      auto set_stop = [](auto& e) {
-        auto stop = e.lockPointerToDual();
+      mi::moving_atomic<bool> stop = false;
+      auto set_stop = [](auto& stop) {
         if (!!stop) {
           stop->store(true);
         }
-        e.unlockPointerToDual();
       };
-      auto tokens = mi::entangle(std::move(stop), std::move(set_stop));
+      auto tokens = mi::entangle(std::move(stop), set_stop);
 
       using Stopper = decltype(tokens.second);
       struct Data : mi::none<> {
@@ -239,11 +409,13 @@ SCENARIO("flow single cancellation new thread", "[flow][deferred]") {
           Data{std::move(tokens.second)},
           [&](auto& data, auto e) noexcept {
             signals += 100000;
-            data.stopper.t(data.stopper);
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
           },
           [&](auto& data) {
             signals += 10000;
-            data.stopper.t(data.stopper);
+            auto both = lock_both(data.stopper);
+            (*(both.first))(both.second);
           });
 
       // make all the signals come from the same thread
@@ -262,7 +434,8 @@ SCENARIO("flow single cancellation new thread", "[flow][deferred]") {
                     [stoppee = std::move(stoppee),
                      out = std::move(out)](auto) mutable {
                       // check boolean to select signal
-                      if (!stoppee.t.load()) {
+                      auto both = lock_both(stoppee);
+                      if (!!both.first && !*(both.first)) {
                         ::mi::set_value(out, 42);
                       } else {
                         // cancellation is not an error