feat support resume to Master Fiber

tile/fiber/detail/fiber.cc

@@ -12,26 +12,58 @@ namespace tile {
 namespace fiber {
 namespace detail {
 
+#define SET_PREV_FCTX(new_fctx)                                                \
+  do {                                                                         \
+    if (PrevFiber()) {                                                         \
+      PrevFiber()->ctx_->fctx = (new_fctx);                                    \
+    }                                                                          \
+  } while (0)
+
 static thread_local Fiber *tls_current_fiber = nullptr;
 static thread_local Fiber *tls_master_fiber = nullptr;
+static thread_local Fiber *tls_prev_fiber = nullptr;
 
 constexpr std::size_t kStackSize = 128 * 1024; // 128k
 constexpr std::size_t kAlignSize = 16;
 
+struct alignas(hardware_destructive_interference_size) Fiber::FiberContext {
+  fcontext_t fctx;
+  std::aligned_storage<kStackSize, kAlignSize>::type stack;
+  std::function<void()> proc;
+};
+
+// static void SetPrevFiber(Fiber *fiber) noexcept { tls_prev_fiber = fiber; }
+static Fiber *PrevFiber() noexcept { return tls_prev_fiber; }
+
 void FiberEntry(fcontext_transfer_t t) {
-  // TILE_LOG_INFO("FiberEntry creater {}, proc {}", t.fctx, t.data);
+  SET_PREV_FCTX(t.fctx);
+
   std::function<void()> *fn = static_cast<std::function<void()> *>(t.data);
   TILE_CHECK_NE(t.data, nullptr);
   TILE_CHECK_NE(t.fctx, nullptr);
+
+  Fiber *self;
+
   try {
+    // From Resume()
     t = jump_fcontext(t.fctx, nullptr);
+    self = Fiber::Current();
+    SET_PREV_FCTX(t.fctx);
+
     (*fn)();
   } catch (const std::exception &e) {
-    // TILE_LOG_ERROR("Exception caught in fiber: {}", e.what());
+    TILE_LOG_ERROR("Exception caught in fiber: {}", e.what());
   }
   TILE_CHECK_NE(t.fctx, nullptr);
-  // TILE_LOG_INFO("FiberEntry End. Resume to {}", t.fctx);
+
-  jump_fcontext(t.fctx, nullptr);
+  if (self == Fiber::MasterFiber()) {
+    // TILE_LOG_INFO("FiberEntry End. Resume to {}", t.fctx);
+    jump_fcontext(t.fctx, nullptr);
+  } else {
+    // TILE_LOG_INFO("FiberEntry End. Resume to {}",
+    //               Fiber::MasterFiber()->ctx_->fctx);
+    Fiber::MasterFiber()->Resume();
+  }
 }
 
 fcontext_t CreateFiber(void *stack, std::size_t stack_size,
@@ -43,14 +75,11 @@ fcontext_t CreateFiber(void *stack, std::size_t stack_size,
   return jump_fcontext(fctx, fn).fctx;
 }
 
-struct alignas(hardware_destructive_interference_size) Fiber::FiberContext {
-  fcontext_t fctx;
-  std::aligned_storage<kStackSize, kAlignSize>::type stack;
-  std::function<void()> proc;
-};
-
 Fiber *Fiber::Current() noexcept { return tls_current_fiber; }
-void Fiber::SetCurrent(Fiber *fiber) noexcept { tls_current_fiber = fiber; }
+void Fiber::SetCurrent(Fiber *fiber) noexcept {
+  tls_prev_fiber = tls_current_fiber;
+  tls_current_fiber = fiber;
+}
 
 Fiber *Fiber::MasterFiber() noexcept { return tls_master_fiber; }
 void Fiber::SetMasterFiber(Fiber *fiber) noexcept { tls_master_fiber = fiber; }
@@ -62,13 +91,10 @@ std::unique_ptr<Fiber> Fiber::Create(std::function<void()> proc) noexcept {
 
 Fiber::Fiber(std::function<void()> proc)
     : ctx_(object_pool::Get<FiberContext>().Leak()) {
+  TILE_CHECK(proc);
 
   ctx_->proc = std::move(proc);
-  if (ctx_->proc) {
+  ctx_->fctx = CreateFiber(&ctx_->stack, kStackSize, &ctx_->proc);
-    ctx_->fctx = CreateFiber(&ctx_->stack, kStackSize, &ctx_->proc);
-  } else {
-    ctx_->fctx = nullptr;
-  }
 }
 
 Fiber::~Fiber() {
@@ -78,18 +104,32 @@ Fiber::~Fiber() {
 }
 
 void Fiber::Resume() {
+  TILE_CHECK(state_ != FiberState::Suspended);
+  TILE_CHECK(state_ != FiberState::Terminated);
   TILE_CHECK_NE(ctx_->fctx, nullptr);
+
   auto caller = Current();
   TILE_CHECK_NE(caller, this, "Can't `Resume()` self");
-  SetCurrent(this);
+  if (caller != Fiber::MasterFiber() && this != Fiber::MasterFiber()) {
-  // TILE_LOG_INFO("Resume before proc: {}", fmt::ptr(&proc_));
+    caller->state_ = FiberState::Suspended;
-  ctx_->fctx =
+  } else if (caller) {
-      jump_fcontext(internal::Exchange(ctx_->fctx, nullptr), nullptr).fctx;
+    caller->state_ = FiberState::Ready;
-  // TILE_LOG_INFO("Resume after  proc: {}", fmt::ptr(&proc_));
+  }
-  SetCurrent(caller);
-}
 
-void Fiber::Yield() {}
+  // TILE_LOG_INFO("Resume from {} to {}", fmt::ptr(caller), fmt::ptr(this));
+  SetCurrent(this);
+  this->state_ = FiberState::Running;
+  auto caller_fctx =
+      jump_fcontext(internal::Exchange(ctx_->fctx, nullptr), nullptr).fctx;
+  if (Current() == this) {
+    // fiber terminated
+    state_ = FiberState::Terminated;
+  } else {
+    state_ = FiberState::Ready;
+  }
+
+  // SET_PREV_FCTX(caller_fctx);
+}
 
 } // namespace detail
 } // namespace fiber

tile/fiber/detail/fiber.h

@@ -15,15 +15,15 @@ struct fcontext_transfer;
 namespace tile {
 namespace fiber {
 namespace detail {
-void RunProc(void *arg);
-void RunProc1(struct fcontext_transfer);
 
 class Scheduler;
 
 class alignas(hardware_destructive_interference_size) Fiber {
 public:
-  enum FiberState {
+  enum class FiberState {
-    Ready,
+    Runnable,
+    Running,
+    Suspended,
     Waiting,
     Terminated,
   };
@@ -50,6 +50,8 @@ public:
 private:
   TILE_FRIEND_TEST(Fiber, Base);
   friend Scheduler;
+
+  friend void FiberEntry(struct fcontext_transfer);
   struct FiberContext;
   friend class ::tile::PoolTraits<FiberContext>;
 
@@ -57,7 +59,7 @@ private:
 
 private:
   std::unique_ptr<FiberContext> ctx_;
-  FiberState state_{Ready};
+  FiberState state_{FiberState::Runnable};
 };
 
 } // namespace detail

tile/fiber/detail/fiber_test.cc

@@ -7,7 +7,7 @@ namespace fiber {
 namespace detail {
 
 TEST(Fiber, Base) {
-  constexpr auto kMaxCnt = 100 * 1000;
+  constexpr auto kMaxCnt = 1; // 100 * 1000;
   int cnt = 0;
   int resume_cnt = 0;
 
@@ -24,22 +24,26 @@ TEST(Fiber, Base) {
     ASSERT_EQ(cnt, 0);
     while (cnt < kMaxCnt) {
       std::unique_ptr<Fiber> worker_fiber = Fiber::Create([&] {
-        ASSERT_EQ(Fiber::Current(), worker_fiber.get());
+        while (cnt < kMaxCnt) {
-        ++cnt;
+          ASSERT_EQ(Fiber::Current(), worker_fiber.get());
+          ++cnt;
+          master_fiber->Resume();
+        }
       });
 
-      int old = cnt;
+      while (cnt < kMaxCnt) {
-      worker_fiber->Resume();
+        int old = cnt;
-      ASSERT_EQ(old + 1, cnt);
+        worker_fiber->Resume();
-      ASSERT_EQ(Fiber::Current(), master_fiber.get());
+        ASSERT_EQ(old + 1, cnt);
+        ASSERT_EQ(Fiber::Current(), master_fiber.get());
+      }
     }
 
     ASSERT_EQ(cnt, kMaxCnt);
   });
 
   Fiber::SetMasterFiber(master_fiber.get());
-  // Fiber::SetCurrent(master_fiber.get());
+  Fiber::SetCurrent(nullptr);
-  // master_fiber->Resume();
   master_fiber->Resume();
 }
 

tile/fiber/detail/scheduler.h

@@ -0,0 +1,15 @@
+#ifndef TILE_FIBER_DETAIL_SCHEDULER_H
+#define TILE_FIBER_DETAIL_SCHEDULER_H
+
+#pragma once
+
+namespace tile {
+namespace fiber {
+namespace detail {
+
+class Scheduler {};
+} // namespace detail
+} // namespace fiber
+} // namespace tile
+
+#endif // TILE_FIBER_DETAIL_SCHEDULER_H