Fix Python FiberTask to properly init/reset Python thread state

Reviewed By: alikhtarov

Differential Revision: D2813735

fb-gh-sync-id: ab3c3e18618ed8bf15f478bcfca008786834e65c

diff --git a/folly/experimental/fibers/Fiber.cpp b/folly/experimental/fibers/Fiber.cpp
index d5a8b1e9bac551d669deda7c9690d7eb2f473507..175021534caaa0f27b92a8ba58edb39ad22d1e47 100644 (file)
--- a/folly/experimental/fibers/Fiber.cpp
+++ b/folly/experimental/fibers/Fiber.cpp
@@ -182,20 +182,30 @@ void Fiber::fiberFunc() {
 }
 
 intptr_t Fiber::preempt(State state) {
-  DCHECK_EQ(fiberManager_.activeFiber_, this);
-  DCHECK_EQ(state_, RUNNING);
-  DCHECK_NE(state, RUNNING);
+  intptr_t ret;
 
-  fiberManager_.activeFiber_ = nullptr;
-  state_ = state;
+  auto preemptImpl = [&]() mutable {
+    DCHECK_EQ(fiberManager_.activeFiber_, this);
+    DCHECK_EQ(state_, RUNNING);
+    DCHECK_NE(state, RUNNING);
 
-  recordStackPosition();
+    fiberManager_.activeFiber_ = nullptr;
+    state_ = state;
+
+    recordStackPosition();
+
+    ret = jumpContext(&fcontext_, &fiberManager_.mainContext_, 0);
 
-  auto ret = jumpContext(&fcontext_, &fiberManager_.mainContext_, 0);
+    DCHECK_EQ(fiberManager_.activeFiber_, this);
+    DCHECK_EQ(state_, READY_TO_RUN);
+    state_ = RUNNING;
+  };
 
-  DCHECK_EQ(fiberManager_.activeFiber_, this);
-  DCHECK_EQ(state_, READY_TO_RUN);
-  state_ = RUNNING;
+  if (fiberManager_.preemptRunner_) {
+    fiberManager_.preemptRunner_->run(std::ref(preemptImpl));
+  } else {
+    preemptImpl();
+  }
 
   return ret;
 }

diff --git a/folly/experimental/fibers/FiberManager.cpp b/folly/experimental/fibers/FiberManager.cpp
index e405630db9811e2187bcf249a6dd233585929b33..2a9a2ea1806de9cc26f9864a89f766e16da3b109 100644 (file)
--- a/folly/experimental/fibers/FiberManager.cpp
+++ b/folly/experimental/fibers/FiberManager.cpp
@@ -121,6 +121,10 @@ void FiberManager::setObserver(ExecutionObserver* observer) {
   observer_ = observer;
 }
 
+void FiberManager::setPreemptRunner(InlineFunctionRunner* preemptRunner) {
+  preemptRunner_ = preemptRunner;
+}
+
 void FiberManager::doFibersPoolResizing() {
   while (fibersAllocated_ > maxFibersActiveLastPeriod_ &&
          fibersPoolSize_ > options_.maxFibersPoolSize) {

diff --git a/folly/experimental/fibers/FiberManager.h b/folly/experimental/fibers/FiberManager.h
index 0a74500f6cf9185213f4dde0579b201b8033bd5a..549d3f323ea8dab82f71de18d14b518224ee7b4f 100644 (file)
--- a/folly/experimental/fibers/FiberManager.h
+++ b/folly/experimental/fibers/FiberManager.h
@@ -48,6 +48,16 @@ template <typename T>
 class LocalType {
 };
 
+class InlineFunctionRunner {
+ public:
+  virtual ~InlineFunctionRunner() {}
+
+  /**
+   * func must be executed inline and only once.
+   */
+  virtual void run(std::function<void()> func) = 0;
+};
+
 /**
  * @class FiberManager
  * @brief Single-threaded task execution engine.
@@ -260,6 +270,11 @@ class FiberManager : public ::folly::Executor {
    */
   void setObserver(ExecutionObserver* observer);
 
+  /**
+   * Setup fibers preempt runner.
+   */
+  void setPreemptRunner(InlineFunctionRunner* preemptRunner);
+
   /**
    * Returns an estimate of the number of fibers which are waiting to run (does
    * not include fibers or tasks scheduled remotely).
@@ -387,6 +402,11 @@ class FiberManager : public ::folly::Executor {
    */
   std::function<void()> immediateFunc_;
 
+  /**
+   * Preempt runner.
+   */
+  InlineFunctionRunner* preemptRunner_{nullptr};
+
   /**
    * Fiber's execution observer.
    */