add new thread cancellation test

fbshipit-source-id: 9d12db7b6b8ea4bc2ba87de78a35046884c30e3d

folly/experimental/pushmi/include/pushmi.h

@@ -1330,9 +1330,9 @@ void set_stopping(S& s) noexcept(noexcept(s.stopping())) {
   s.stopping();
 }
 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(up))) {
-  s.starting(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));
 }
 
 PUSHMI_TEMPLATE (class SD, class Out)
@@ -1404,10 +1404,10 @@ void set_stopping(std::reference_wrapper<S> s) noexcept(
   set_stopping(s.get());
 }
 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(), up))) {
-  set_starting(s.get(), 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));
 }
 PUSHMI_TEMPLATE (class SD, class Out)
   (requires requires ( submit(std::declval<SD&>(), std::declval<Out>()) ))
@@ -1485,13 +1485,13 @@ struct set_stopping_fn {
 struct set_starting_fn {
   PUSHMI_TEMPLATE (class S, class Up)
     (requires requires (
-      set_starting(std::declval<S&>(), std::declval<Up&>()),
+      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, up))) {
+  void operator()(S&& s, Up up) const
+      noexcept(noexcept(set_starting(s, std::move(up)))) {
     try {
-      set_starting(s, up);
+      set_starting(s, std::move(up));
     } catch (...) {
       set_error(s, std::current_exception());
     }
@@ -2004,8 +2004,8 @@ PUSHMI_CONCEPT_DEF(
     class PE = std::exception_ptr,
     class E = PE)
   (concept FlowNoneReceiver)(N, Up, PE, E),
-    requires(N& n, Up& up) (
-      ::pushmi::set_starting(n, up)
+    requires(N& n, Up&& up) (
+      ::pushmi::set_starting(n, (Up &&) up)
     ) &&
     FlowReceiver<N> &&
     Receiver<Up> &&
@@ -2188,7 +2188,7 @@ struct ignoreStpF {
 
 struct ignoreStrtF {
   template <class Up>
-  void operator()(Up&) {}
+  void operator()(Up&&) {}
 };
 
 
@@ -2240,10 +2240,10 @@ struct passDStpF {
 struct passDStrtF {
   PUSHMI_TEMPLATE(class Up, class Data)
     (requires requires (
-      ::pushmi::set_starting(std::declval<Data&>(), std::declval<Up&>())
+      ::pushmi::set_starting(std::declval<Data&>(), std::declval<Up>())
     ) && Receiver<Data>)
-  void operator()(Data& out, Up& up) const {
-    ::pushmi::set_starting(out, up);
+  void operator()(Data& out, Up up) const {
+    ::pushmi::set_starting(out, std::move(up));
   }
 };
 
@@ -4045,13 +4045,13 @@ class flow_single<V, PE, E> {
     static void s_error(data&, E) noexcept { std::terminate(); }
     static void s_value(data&, V) {}
     static void s_stopping(data&) noexcept {}
-    static void s_starting(data&, any_none<PE>&) {}
+    static void s_starting(data&, any_none<PE>) {}
     void (*op_)(data&, data*) = vtable::s_op;
     void (*done_)(data&) = vtable::s_done;
     void (*error_)(data&, E) noexcept = vtable::s_error;
     void (*value_)(data&, V) = vtable::s_value;
     void (*stopping_)(data&) noexcept = vtable::s_stopping;
-    void (*starting_)(data&, any_none<PE>&) = vtable::s_starting;
+    void (*starting_)(data&, any_none<PE>) = vtable::s_starting;
   };
   static constexpr vtable const noop_ {};
   vtable const* vptr_ = &noop_;
@@ -4075,8 +4075,8 @@ class flow_single<V, PE, E> {
       static void stopping(data& src) noexcept {
         ::pushmi::set_stopping(*static_cast<Wrapped*>(src.pobj_));
       }
-      static void starting(data& src, any_none<PE>& up) {
-        ::pushmi::set_starting(*static_cast<Wrapped*>(src.pobj_), up);
+      static void starting(data& src, any_none<PE> up) {
+        ::pushmi::set_starting(*static_cast<Wrapped*>(src.pobj_), std::move(up));
       }
     };
     static const vtable vtbl{s::op, s::done, s::error, s::value, s::stopping, s::starting};
@@ -4106,8 +4106,8 @@ class flow_single<V, PE, E> {
       static void stopping(data& src) noexcept {
         ::pushmi::set_stopping(*static_cast<Wrapped*>((void*)src.buffer_));
       }
-      static void starting(data& src, any_none<PE>& up) {
-        ::pushmi::set_starting(*static_cast<Wrapped*>((void*)src.buffer_), up);
+      static void starting(data& src, any_none<PE> up) {
+        ::pushmi::set_starting(*static_cast<Wrapped*>((void*)src.buffer_), std::move(up));
       }
     };
     static const vtable vtbl{s::op, s::done, s::error, s::value, s::stopping, s::starting};
@@ -4150,8 +4150,8 @@ public:
   void stopping() noexcept {
     vptr_->stopping_(data_);
   }
-  void starting(any_none<PE>& up) {
-    vptr_->starting_(data_, up);
+  void starting(any_none<PE> up) {
+    vptr_->starting_(data_, std::move(up));
   }
 };
 
@@ -4219,9 +4219,9 @@ class flow_single<VF, EF, DF, StpF, StrtF> {
     stpf_();
   }
   PUSHMI_TEMPLATE(class Up)
-    (requires Receiver<Up, is_none<>> && Invocable<StrtF&, Up&>)
-  void starting(Up& up) {
-    strtf_(up);
+    (requires Receiver<Up, is_none<>> && Invocable<StrtF&, Up&&>)
+  void starting(Up&& up) {
+    strtf_( (Up &&) up);
   }
 };
 
@@ -4291,9 +4291,9 @@ class flow_single<Data, DVF, DEF, DDF, DStpF, DStrtF> {
     stpf_(data_);
   }
   PUSHMI_TEMPLATE (class Up)
-    (requires Invocable<DStrtF&, Data&, Up&>)
-  void starting(Up& up) {
-    strtf_(data_, up);
+    (requires Invocable<DStrtF&, Data&, Up&&>)
+  void starting(Up&& up) {
+    strtf_(data_, (Up &&) up);
   }
 };
 
@@ -4936,6 +4936,9 @@ class trampoline {
       auto item = std::move(pending(pending_store).front());
       pending(pending_store).pop_front();
       auto& when = std::get<0>(item);
+      if (when > trampoline<E>::now()) {
+        std::this_thread::sleep_until(when);
+      }
       auto& what = std::get<1>(item);
       any_time_executor_ref<error_type, time_point> anythis{that};
       ::pushmi::set_value(what, anythis);

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

@@ -62,7 +62,7 @@ struct ignoreStpF {
 
 struct ignoreStrtF {
   template <class Up>
-  void operator()(Up&) {}
+  void operator()(Up&&) {}
 };
 
 
@@ -114,10 +114,10 @@ struct passDStpF {
 struct passDStrtF {
   PUSHMI_TEMPLATE(class Up, class Data)
     (requires requires (
-      ::pushmi::set_starting(std::declval<Data&>(), std::declval<Up&>())
+      ::pushmi::set_starting(std::declval<Data&>(), std::declval<Up>())
     ) && Receiver<Data>)
-  void operator()(Data& out, Up& up) const {
-    ::pushmi::set_starting(out, up);
+  void operator()(Data& out, Up up) const {
+    ::pushmi::set_starting(out, std::move(up));
   }
 };
 

folly/experimental/pushmi/include/pushmi/concepts.h

@@ -271,8 +271,8 @@ PUSHMI_CONCEPT_DEF(
     class PE = std::exception_ptr,
     class E = PE)
   (concept FlowNoneReceiver)(N, Up, PE, E),
-    requires(N& n, Up& up) (
-      ::pushmi::set_starting(n, up)
+    requires(N& n, Up&& up) (
+      ::pushmi::set_starting(n, (Up &&) up)
     ) &&
     FlowReceiver<N> &&
     Receiver<Up> &&

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

+#pragma once
+// Copyright (c) 2018-present, Facebook, Inc.
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+#include "forwards.h"
+
+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
+
+// 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
+// with a heap-allocated, refcounted, control block (or do something else using
+// external storage, like a lock table chosen by the current addresses of both
+// objects).
+// Another thing you could do is have locks, and a backoff strategy for dealing
+// with deadlock: lock locally, trylock your dual, if the trylock fails,
+// unlock yourself, wait a little while (giving a thread touching the dual a
+// chance to acquire the local lock), and repeat. That's kind of ugly.
+// This algorithm (which, to be clear, is untested and I haven't really even
+// thought through carefully) provides the same guarantees, but without using
+// external storage or backoff-based deadlock recovery.
+
+template <class T, class Dual>
+struct entangled {
+  // must be constructed first so that the other.lockBoth() in the move
+  // constructor is run before moving other.t and other.dual
+  std::atomic<int> stateMachine;
+
+  T t;
+  // In a couple places, we can save on some atomic ops by making this atomic,
+  // and adding a "dual == null" fast-path without locking.
+  entangled<Dual, T>* dual;
+
+  const static int kUnlocked = 0;
+  const static int kLocked = 1;
+  const static int kLockedAndLossAcknowledged = 2;
+
+  // Note: *not* thread-safe; it's a bug for two threads to concurrently call
+  // lockBoth() on the same entangled (just as it's a bug for two threads to
+  // concurrently move from the same object).
+  // However, calling lockBoth() on two entangled objects at once is
+  // thread-safe.
+  // Note also that this may wait indefinitely; it's not the usual non-blocking
+  // tryLock().
+  bool tryLockBoth() {
+    // Try to acquire the local lock. We have to start locally, since local
+    // addresses are the only ones we know are safe at first. The rule is, you
+    // 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)) {
+      return false;
+    }
+    // Having *either* object local-locked protects the data in both objects.
+    // Once we hold our lock, no control data can change, in either object.
+    if (dual == nullptr) {
+      return true;
+    }
+    expected = kUnlocked;
+    if (dual->stateMachine.compare_exchange_strong(expected, kLocked)) {
+      return true;
+    }
+    // We got here, and so hit the race; we're deadlocked if we stick to
+    // locking. Revert to address-ordering. Note that address-ordering would
+    // not be safe on its own, because of the lifetime issues involved; the
+    // addresses here are only stable *because* we know both sides are locked,
+    // and because of the invariant that you must hold both locks to modify
+    // either piece of data.
+    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) {
+        // Spin.
+      }
+      stateMachine.store(kLocked);
+      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);
+      return false;
+    }
+  }
+
+  void lockBoth() {
+    while (!tryLockBoth()) {
+      // Spin. But, note that all the unbounded spinning in tryLockBoth can be
+      // straightforwardly futex-ified. There's a potentialy starvation issue
+      // here, but note that it can be dealt with by adding a "priority" bit to
+      // the state machine (i.e. if my priority bit is set, the thread for whom
+      // I'm the local member of the pair gets to win the race, rather than
+      // using address-ordering).
+    }
+  }
+
+  void unlockBoth() {
+    // Note that unlocking locally and then remotely is the right order. There
+    // are no concurrent accesses to this object (as an API constraint -- lock
+    // and unlock are not thread safe!), and no other thread will touch the
+    // 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);
+    if (dual != nullptr) {
+      dual->stateMachine.store(kUnlocked);
+    }
+  }
+
+  entangled() = delete;
+  entangled(const entangled&) = delete;
+  entangled& operator=(const entangled&) = delete;
+  entangled& operator=(entangled&&) = delete;
+
+  explicit entangled(T t)
+      : t(std::move(t)), dual(nullptr), stateMachine(kUnlocked) {}
+  entangled(entangled&& other)
+      : stateMachine((other.lockBoth(), kLocked)),
+        t(std::move(other.t)),
+        dual(std::move(other.dual)) {
+    // Note that, above, we initialized stateMachine to the locked state; the
+    // address of this object hasn't escaped yet, and won't (until we unlock
+    // the dual), so it doesn't *really* matter, but it's conceptually helpful
+    // to maintain that invariant.
+
+    // Update our dual's data.
+    if (dual != nullptr) {
+      dual->dual = this;
+    }
+
+    // Update other's data.
+    other.dual = nullptr;
+    // unlock other so that its destructor can complete
+    other.stateMachine.store(kUnlocked);
+
+    // We locked on other, but will unlock on *this. The locking protocol
+    // ensured that no accesses to other will occur after lock() returns, and
+    // since then we updated dual's dual to be us.
+    unlockBoth();
+  }
+
+  ~entangled() {
+    lockBoth();
+    if (dual != nullptr) {
+      dual->dual = nullptr;
+    }
+    unlockBoth();
+  }
+
+  // Must unlock later even if dual is nullptr. This is fixable.
+  Dual* lockPointerToDual() {
+    lockBoth();
+    return !!dual ? std::addressof(dual->t) : nullptr;
+  }
+
+  void unlockPointerToDual() {
+    unlockBoth();
+  }
+};
+#endif
+
+template <class First, class Second>
+auto entangle(First f, Second s)
+    -> std::pair<entangled<First, Second>, entangled<Second, First>> {
+  entangled<First, Second> ef(std::move(f));
+  entangled<Second, First> es(std::move(s));
+  ef.dual = std::addressof(es);
+  es.dual = std::addressof(ef);
+  return {std::move(ef), std::move(es)};
+}
+
+} // namespace pushmi

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

@@ -33,9 +33,9 @@ void set_stopping(S& s) noexcept(noexcept(s.stopping())) {
   s.stopping();
 }
 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(up))) {
-  s.starting(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));
 }
 
 PUSHMI_TEMPLATE (class SD, class Out)
@@ -107,10 +107,10 @@ void set_stopping(std::reference_wrapper<S> s) noexcept(
   set_stopping(s.get());
 }
 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(), up))) {
-  set_starting(s.get(), 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));
 }
 PUSHMI_TEMPLATE (class SD, class Out)
   (requires requires ( submit(std::declval<SD&>(), std::declval<Out>()) ))
@@ -188,13 +188,13 @@ struct set_stopping_fn {
 struct set_starting_fn {
   PUSHMI_TEMPLATE (class S, class Up)
     (requires requires (
-      set_starting(std::declval<S&>(), std::declval<Up&>()),
+      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, up))) {
+  void operator()(S&& s, Up up) const
+      noexcept(noexcept(set_starting(s, std::move(up)))) {
     try {
-      set_starting(s, up);
+      set_starting(s, std::move(up));
     } catch (...) {
       set_error(s, std::current_exception());
     }

folly/experimental/pushmi/include/pushmi/flow_single.h

@@ -25,13 +25,13 @@ class flow_single<V, PE, E> {
     static void s_error(data&, E) noexcept { std::terminate(); }
     static void s_value(data&, V) {}
     static void s_stopping(data&) noexcept {}
-    static void s_starting(data&, any_none<PE>&) {}
+    static void s_starting(data&, any_none<PE>) {}
     void (*op_)(data&, data*) = vtable::s_op;
     void (*done_)(data&) = vtable::s_done;
     void (*error_)(data&, E) noexcept = vtable::s_error;
     void (*value_)(data&, V) = vtable::s_value;
     void (*stopping_)(data&) noexcept = vtable::s_stopping;
-    void (*starting_)(data&, any_none<PE>&) = vtable::s_starting;
+    void (*starting_)(data&, any_none<PE>) = vtable::s_starting;
   };
   static constexpr vtable const noop_ {};
   vtable const* vptr_ = &noop_;
@@ -55,8 +55,8 @@ class flow_single<V, PE, E> {
       static void stopping(data& src) noexcept {
         ::pushmi::set_stopping(*static_cast<Wrapped*>(src.pobj_));
       }
-      static void starting(data& src, any_none<PE>& up) {
-        ::pushmi::set_starting(*static_cast<Wrapped*>(src.pobj_), up);
+      static void starting(data& src, any_none<PE> up) {
+        ::pushmi::set_starting(*static_cast<Wrapped*>(src.pobj_), std::move(up));
       }
     };
     static const vtable vtbl{s::op, s::done, s::error, s::value, s::stopping, s::starting};
@@ -86,8 +86,8 @@ class flow_single<V, PE, E> {
       static void stopping(data& src) noexcept {
         ::pushmi::set_stopping(*static_cast<Wrapped*>((void*)src.buffer_));
       }
-      static void starting(data& src, any_none<PE>& up) {
-        ::pushmi::set_starting(*static_cast<Wrapped*>((void*)src.buffer_), up);
+      static void starting(data& src, any_none<PE> up) {
+        ::pushmi::set_starting(*static_cast<Wrapped*>((void*)src.buffer_), std::move(up));
       }
     };
     static const vtable vtbl{s::op, s::done, s::error, s::value, s::stopping, s::starting};
@@ -130,8 +130,8 @@ public:
   void stopping() noexcept {
     vptr_->stopping_(data_);
   }
-  void starting(any_none<PE>& up) {
-    vptr_->starting_(data_, up);
+  void starting(any_none<PE> up) {
+    vptr_->starting_(data_, std::move(up));
   }
 };
 
@@ -199,9 +199,9 @@ class flow_single<VF, EF, DF, StpF, StrtF> {
     stpf_();
   }
   PUSHMI_TEMPLATE(class Up)
-    (requires Receiver<Up, is_none<>> && Invocable<StrtF&, Up&>)
-  void starting(Up& up) {
-    strtf_(up);
+    (requires Receiver<Up, is_none<>> && Invocable<StrtF&, Up&&>)
+  void starting(Up&& up) {
+    strtf_( (Up &&) up);
   }
 };
 
@@ -271,9 +271,9 @@ class flow_single<Data, DVF, DEF, DDF, DStpF, DStrtF> {
     stpf_(data_);
   }
   PUSHMI_TEMPLATE (class Up)
-    (requires Invocable<DStrtF&, Data&, Up&>)
-  void starting(Up& up) {
-    strtf_(data_, up);
+    (requires Invocable<DStrtF&, Data&, Up&&>)
+  void starting(Up&& up) {
+    strtf_(data_, (Up &&) up);
   }
 };
 

folly/experimental/pushmi/include/pushmi/trampoline.h

@@ -225,6 +225,9 @@ class trampoline {
       auto item = std::move(pending(pending_store).front());
       pending(pending_store).pop_front();
       auto& when = std::get<0>(item);
+      if (when > trampoline<E>::now()) {
+        std::this_thread::sleep_until(when);
+      }
       auto& what = std::get<1>(item);
       any_time_executor_ref<error_type, time_point> anythis{that};
       ::pushmi::set_value(what, anythis);

folly/experimental/pushmi/test/FlowTest.cpp

@@ -8,37 +8,57 @@ using namespace std::literals;
 #include "pushmi/flow_single_deferred.h"
 #include "pushmi/o/submit.h"
 
-#include "pushmi/trampoline.h"
+#include "pushmi/entangle.h"
 #include "pushmi/new_thread.h"
+#include "pushmi/trampoline.h"
 
 using namespace pushmi::aliases;
 
 #if __cpp_deduction_guides >= 201703
 #define MAKE(x) x MAKE_
-#define MAKE_(...) {__VA_ARGS__}
+#define MAKE_(...) \
+  { __VA_ARGS__ }
 #else
-#define MAKE(x) make_ ## x
+#define MAKE(x) make_##x
 #endif
 
-SCENARIO( "flow single immediate cancellation", "[flow][deferred]" ) {
-
+SCENARIO("flow single immediate cancellation", "[flow][deferred]") {
   int signals = 0;
 
-  GIVEN( "A flow single deferred" ) {
-
-    auto f = mi::MAKE(flow_single_deferred)([&](auto out){
-
-      // boolean cancellation - on stack
+  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();
+        if (!!stop) {
+          *stop = true;
+        }
+        e.unlockPointerToDual();
+      };
+      auto tokens = mi::entangle(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)(
-        [&](auto e) noexcept {signals += 1000000; stop = true;},
-        [&](){signals += 100000; stop = true;});
+          Data{std::move(tokens.second)},
+          [&](auto& data, auto e) noexcept {
+            signals += 1000000;
+            data.stopper.t(data.stopper);
+          },
+          [&](auto& data) {
+            signals += 100000;
+            data.stopper.t(data.stopper);
+          });
 
       // pass reference for cancellation.
-      ::mi::set_starting(out, up);
+      ::mi::set_starting(out, std::move(up));
 
       // check boolean to select signal
-      if (!stop) {
+      if (!tokens.first.t) {
         ::mi::set_value(out, 42);
       } else {
         // cancellation is not an error
@@ -49,76 +69,320 @@ SCENARIO( "flow single immediate cancellation", "[flow][deferred]" ) {
       ::mi::set_stopping(out);
     });
 
-    WHEN( "submit is applied and cancels the producer" ) {
-
-      f | op::submit(
-          mi::on_value([&](int){ signals = 100; }),
-          mi::on_error([&](auto) noexcept { signals = 1000; }),
-          mi::on_done([&](){signals += 1;}),
-          mi::on_stopping([&](){signals += 10000;}),
+    WHEN("submit is applied and cancels the producer") {
+      f |
+          op::submit(
+              mi::on_value([&](int) { signals += 100; }),
+              mi::on_error([&](auto) noexcept { signals += 1000; }),
+              mi::on_done([&]() { signals += 1; }),
+              mi::on_stopping([&]() { signals += 10000; }),
               // immediately stop producer
-          mi::on_starting([&](auto up){ signals = 10; ::mi::set_done(up); }));
+              mi::on_starting([&](auto up) {
+                signals += 10;
+                ::mi::set_done(up);
+              }));
 
-      THEN( "the starting, up.done, out.done out.stopping signals are each recorded once" ) {
+      THEN(
+          "the starting, up.done, out.done and out.stopping signals are each recorded once") {
         REQUIRE(signals == 110011);
       }
     }
+
+    WHEN("submit is applied and cancels the producer late") {
+      f |
+          op::submit(
+              mi::on_value([&](int) { signals += 100; }),
+              mi::on_error([&](auto) noexcept { signals += 1000; }),
+              mi::on_done([&]() { signals += 1; }),
+              mi::on_stopping([&]() { signals += 10000; }),
+              // do not stop producer before it is scheduled to run
+              mi::on_starting([&](auto up) { signals += 10; }));
+
+      THEN(
+          "the starting, out.value and out.stopping signals are each recorded once") {
+        REQUIRE(signals == 10110);
+      }
+    }
   }
 }
 
-SCENARIO( "flow single cancellation", "[flow][deferred]" ) {
-
-  auto tr = v::trampoline();
+SCENARIO("flow single cancellation trampoline", "[flow][deferred]") {
+  auto tr = mi::trampoline();
   using TR = decltype(tr);
   int signals = 0;
 
-  GIVEN( "A flow single deferred" ) {
-
-    auto f = mi::MAKE(flow_single_deferred)([&](auto out){
-
-      // boolean cancellation - on stack
+  GIVEN("A flow single deferred") {
+    auto f = mi::MAKE(flow_single_deferred)([&](auto out) {
+      // boolean cancellation
       bool stop = false;
-      auto up = mi::MAKE(none)(
-        [&](auto e) noexcept {signals += 1000000; stop = true;},
-        [&](){signals += 100000; stop = true;});
+      auto set_stop = [](auto& e) {
+        auto stop = e.lockPointerToDual();
+        if (!!stop) {
+          *stop = true;
+        }
+        e.unlockPointerToDual();
+      };
+      auto tokens = mi::entangle(stop, set_stop);
 
-      // blocking the stack to keep 'up' alive
-      tr | op::blocking_submit([&](auto tr){
+      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 += 1000000;
+            data.stopper.t(data.stopper);
+          },
+          [&](auto& data) {
+            signals += 100000;
+            data.stopper.t(data.stopper);
+          });
 
+      tr |
+          op::submit([out = std::move(out),
+                      up = std::move(up),
+                      stoppee = std::move(tokens.first)](auto tr) mutable {
             // pass reference for cancellation.
-        ::mi::set_starting(out, up);
+            ::mi::set_starting(out, std::move(up));
 
-        tr | op::submit_after(
+            // submit work to happen later
+            tr |
+                op::submit_after(
                     100ms,
-          [&](auto) {
+                    [out = std::move(out),
+                     stoppee = std::move(stoppee)](auto) mutable {
                       // check boolean to select signal
-            if (!stop) {
+                      if (!stoppee.t) {
                         ::mi::set_value(out, 42);
                       } else {
                         // cancellation is not an error
                         ::mi::set_done(out);
                       }
-            // I want to get rid of this signal it makes usage harder and
-            // messes up r-value qualifing done, error and value.
+                      // I want to get rid of this signal it makes usage harder
+                      // and messes up r-value qualifing done, error and value.
                       ::mi::set_stopping(out);
+                    });
+          });
+    });
+
+    WHEN("submit is applied and cancels the producer") {
+      f |
+          op::submit(
+              mi::on_value([&](int) { signals += 100; }),
+              mi::on_error([&](auto) noexcept { signals += 1000; }),
+              mi::on_done([&]() { signals += 1; }),
+              mi::on_stopping([&]() { signals += 10000; }),
+              // stop producer before it is scheduled to run
+              mi::on_starting([&](auto up) {
+                signals += 10;
+                tr | op::submit_after(50ms, [up = std::move(up)](auto) mutable {
+                  ::mi::set_done(up);
+                });
+              }));
+
+      THEN(
+          "the starting, up.done, out.done and out.stopping signals are each recorded once") {
+        REQUIRE(signals == 110011);
+      }
     }
-        );
+
+    WHEN("submit is applied and cancels the producer late") {
+      f |
+          op::submit(
+              mi::on_value([&](int) { signals += 100; }),
+              mi::on_error([&](auto) noexcept { signals += 1000; }),
+              mi::on_done([&]() { signals += 1; }),
+              mi::on_stopping([&]() { signals += 10000; }),
+              // do not stop producer before it is scheduled to run
+              mi::on_starting([&](auto up) {
+                signals += 10;
+                tr |
+                    op::submit_after(250ms, [up = std::move(up)](auto) mutable {
+                      ::mi::set_done(up);
                     });
+              }));
+
+      THEN(
+          "the starting, up.done, out.value and out.stopping signals are each recorded once") {
+        REQUIRE(signals == 110110);
+      }
+    }
+  }
+}
+
+// copies the value of the atomic during move. requires external Synchronization
+// to make the copy safe. entangle() provides the needed external
+// Synchronization
+template <class T>
+struct moving_atomic : std::atomic<T> {
+  using std::atomic<T>::atomic;
+  moving_atomic(moving_atomic&& o) : std::atomic<T>(o.load()) {}
+};
+
+SCENARIO("flow single 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
+      moving_atomic<bool> stop = false;
+      auto set_stop = [](auto& e) {
+        auto stop = e.lockPointerToDual();
+        if (!!stop) {
+          stop->store(true);
+        }
+        e.unlockPointerToDual();
+      };
+      auto tokens = mi::entangle(std::move(stop), std::move(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 += 1000000;
+            data.stopper.t(data.stopper);
+          },
+          [&](auto& data) {
+            signals += 100000;
+            data.stopper.t(data.stopper);
           });
 
-    WHEN( "submit is applied and cancels the producer" ) {
+      // 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
+                      if (!stoppee.t.load()) {
+                        ::mi::set_value(out, 42);
+                      } else {
+                        // cancellation is not an error
+                        ::mi::set_done(out);
+                      }
+                      // I want to get rid of this signal it makes usage harder
+                      // and messes up r-value qualifing done, error and value.
+                      ::mi::set_stopping(out);
+                    });
+          });
+    });
 
-      f | op::submit(
-          mi::on_value([&](int){ signals = 100; }),
-          mi::on_error([&](auto) noexcept { signals = 1000; }),
-          mi::on_done([&](){signals += 1;}),
-          mi::on_stopping([&](){signals += 10000;}),
+    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; }),
+              mi::on_stopping([&]() { signals += 10000; }),
               // stop producer before it is scheduled to run
-          mi::on_starting([&](auto up){ signals = 10; tr | op::submit_after(50ms, [up](auto) mutable {::mi::set_done(up);}); }));
+              mi::on_starting([&](auto up) {
+                signals += 10;
+                nt |
+                    op::submit_at(
+                        at - 50ms, [up = std::move(up)](auto) mutable {
+                          ::mi::set_done(up);
+                        });
+              }));
 
-      THEN( "the starting, up.done, out.done out.stopping signals are each recorded once" ) {
+      THEN(
+          "the starting, up.done, out.done and out.stopping signals are each recorded once") {
         REQUIRE(signals == 110011);
       }
     }
+
+    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; }),
+              mi::on_stopping([&]() { signals += 10000; }),
+              // 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, out.value and out.stopping signals are each recorded once") {
+        REQUIRE(signals == 110110);
+      }
+    }
+
+    WHEN("submit is applied and cancels the producer at the same time") {
+      // count known results
+      int total = 0;
+      int cancellostrace = 0; // 110110
+      int cancelled = 0; // 110011
+
+      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; }),
+                mi::on_stopping([&]() { signals += 10000; }),
+                // 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 == 110110;
+        cancelled += signals == 110011;
+
+        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 && !!cancelled) {
+          // yay all known outcomes were observed!
+          break;
+        }
+        // try again
+        continue;
+      }
+    }
   }
 }