run flush on reclaim thread

Summary:
As discovered in T177120945, it seems like there is some regression in p99 latency of insert
requests with navy-async. The reason for this might be the fact that we are running the flush in
line if issued for insert request on NavyThread. However, since the flush could take long, this
could cause unnecessarily high latency, particularly, if flushes are executed multiple times on the
same NavyThread as part of insert requests back-to-back. Since the flush can be executed async, this
change makes the flush to be run by the reclaim threads.

In doing so, this change also adds a feature to get the reference to the current NavyThread via
global function.

Reviewed By: haowu14

Differential Revision: D53491112

fbshipit-source-id: 309d43c9fc23343d006344661f618f03c9d30843

Author: Jaesoo Lee

cachelib/navy/CMakeLists.txt

@@ -33,6 +33,7 @@ add_library (cachelib_navy
   common/Device.cpp
   common/FdpNvme.cpp
   common/Hash.cpp
+  common/NavyThread.cpp
   common/SizeDistribution.cpp
   common/Types.cpp
   driver/Driver.cpp

cachelib/navy/block_cache/Allocator.cpp

@@ -22,6 +22,7 @@
 #include <cassert>
 #include <utility>
 
+#include "cachelib/navy/common/NavyThread.h"
 #include "cachelib/navy/common/Utils.h"
 
 namespace facebook::cachelib::navy {
@@ -84,7 +85,7 @@ std::tuple<RegionDescriptor, uint32_t, RelAddress> Allocator::allocateWith(
   // picked ended up being full.
   XDCHECK(!rid.valid());
 
-  if (canWait && !folly::fibers::onFiber()) {
+  if (canWait && !getCurrentNavyThread()) {
     // Waiting on main thread could cause indefinite blocking, so do not wait
     canWait = false;
   }

cachelib/navy/block_cache/Region.cpp

@@ -16,6 +16,8 @@
 
 #include "cachelib/navy/block_cache/Region.h"
 
+#include "cachelib/navy/common/NavyThread.h"
+
 namespace facebook::cachelib::navy {
 
 bool Region::readyForReclaim(bool wait) {
@@ -50,7 +52,7 @@ RegionDescriptor Region::openForRead() {
   std::unique_lock<TimedMutex> l{lock_};
   if (flags_ & kBlockAccess) {
     // Region is currently in reclaim, retry later
-    if (folly::fibers::onFiber()) {
+    if (getCurrentNavyThread()) {
       // If we are on fiber, we can just sleep here
       cond_.wait(l);
     }

cachelib/navy/block_cache/RegionManager.cpp

@@ -63,6 +63,7 @@ RegionManager::RegionManager(uint32_t numRegions,
     auto name = fmt::format("region_manager_{}", i);
     workers_.emplace_back(
         std::make_unique<NavyThread>(name, NavyThread::Options(stackSize)));
+    workerSet_.insert(workers_.back().get());
     workers_.back()->addTaskRemote(
         [name]() { XLOGF(INFO, "{} started", name); });
   }
@@ -263,7 +264,7 @@ void RegionManager::doFlush(RegionId rid, bool async) {
   getRegion(rid).setPendingFlush();
   numInMemBufWaitingFlush_.inc();
 
-  if (!async || folly::fibers::onFiber()) {
+  if (!async || isOnWorker()) {
     doFlushInternal(rid);
   } else {
     getNextWorker().addTaskRemote([this, rid]() { doFlushInternal(rid); });

cachelib/navy/block_cache/RegionManager.h

@@ -273,6 +273,15 @@ class RegionManager {
     return *(workers_[numReclaimScheduled_.add_fetch(1) % workers_.size()]);
   }
 
+  bool isOnWorker() {
+    auto* thread = getCurrentNavyThread();
+    if (!thread) {
+      return false;
+    }
+
+    return workerSet_.count(thread) > 0;
+  }
+
   void doReclaim();
   void doFlushInternal(RegionId rid);
 
@@ -313,6 +322,7 @@ class RegionManager {
   // The thread that runs the flush and reclaim. For Navy-async thread mode, the
   // async flushes will be run in-line on fiber by the async NavyThread itself
   std::vector<std::unique_ptr<NavyThread>> workers_;
+  std::unordered_set<NavyThread*> workerSet_;
   mutable AtomicCounter numReclaimScheduled_;
 
   const RegionEvictCallback evictCb_;

cachelib/navy/common/NavyThread.cpp

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "cachelib/navy/common/NavyThread.h"
+
+namespace facebook::cachelib::navy {
+
+static thread_local NavyThread* currentNavyThread_ = nullptr;
+
+NavyThread* getCurrentNavyThread() { return currentNavyThread_; }
+
+NavyThread::NavyThread(folly::StringPiece name, Options options) {
+  th_ = std::make_unique<folly::ScopedEventBaseThread>(name.str());
+
+  folly::fibers::FiberManager::Options opts;
+  opts.stackSize =
+      options.stackSize ? options.stackSize : Options::kDefaultStackSize;
+  auto& eb = *th_->getEventBase();
+  fm_ = &folly::fibers::getFiberManager(eb, opts);
+
+  eb.runInEventBaseThreadAndWait([this]() { currentNavyThread_ = this; });
+}
+
+NavyThread::~NavyThread() { th_.reset(); }
+
+} // namespace facebook::cachelib::navy

cachelib/navy/common/NavyThread.h

@@ -31,6 +31,12 @@ namespace facebook {
 namespace cachelib {
 namespace navy {
 
+class NavyThread;
+
+// @return the current NavyThread context if running on NavyThread. nullptr
+// otherwise.
+NavyThread* getCurrentNavyThread();
+
 /**
  * NavyThread is a wrapper class that wraps folly::ScopedEventBaseThread and
  * FiberManager. The purpose of NavyThread is to start a new thread running
@@ -58,16 +64,9 @@ class NavyThread {
   /**
    * Initializes with current EventBaseManager and passed-in thread name.
    */
-  explicit NavyThread(folly::StringPiece name, Options options = Options()) {
-    th_ = std::make_unique<folly::ScopedEventBaseThread>(name.str());
-
-    folly::fibers::FiberManager::Options opts;
-    opts.stackSize =
-        options.stackSize ? options.stackSize : Options::kDefaultStackSize;
-    fm_ = &folly::fibers::getFiberManager(*th_->getEventBase(), opts);
-  }
+  explicit NavyThread(folly::StringPiece name, Options options = Options());
 
-  ~NavyThread() { th_.reset(); }
+  ~NavyThread();
 
   /**
    * Add the passed-in task to the FiberManager.