improve thrift mononoke backing store implementation

Summary:
This diff tries to improve the Thrift Mononoke backing store implementation by:

* Re-creating thrift client for every request since thrift has builtin connection pooling
* Using Eden's CPU thread pool for processing data fetched from remote
* Calling thrift methods within correct executor

Reviewed By: wez

Differential Revision: D15170818

fbshipit-source-id: c8be70755a851f63fb62e4d4f36833703283565e

eden/fs/store/hg/HgBackingStore.cpp

@@ -401,12 +401,11 @@ std::unique_ptr<MononokeThriftBackingStore>
 HgBackingStore::initializeThriftMononokeBackingStore() {
   auto edenConfig = config_->getEdenConfig();
   auto tierName = edenConfig->getMononokeTierName();
-  auto executor = folly::getIOExecutor();
 
   XLOG(DBG2) << "Initializing thrift Mononoke backing store for repository "
              << repoName_ << ", using tier " << tierName;
   return std::make_unique<MononokeThriftBackingStore>(
-      tierName, repoName_, executor);
+      tierName, repoName_, serverThreadPool_);
 }
 #endif
 
@@ -544,7 +543,7 @@ Future<unique_ptr<Tree>> HgBackingStore::importTreeImpl(
         })
         .thenError([this, manifestNode, edenTreeID, ownedPath, writeBatch](
                        const folly::exception_wrapper& ex) mutable {
-          XLOG(WARN) << "got exception from MononokeHttpBackingStore: "
+          XLOG(WARN) << "got exception from Mononoke backing store: "
                      << ex.what();
           return fetchTreeFromHgCacheOrImporter(
               manifestNode,

eden/fs/store/mononoke/MononokeThriftBackingStore.cpp

@@ -53,20 +53,18 @@ TreeEntryType treeEntryTypeFromMononoke(MononokeFileType type) {
 MononokeThriftBackingStore::MononokeThriftBackingStore(
     std::string serviceName,
     std::string repo,
-    std::shared_ptr<folly::Executor> executor)
-    : MononokeThriftBackingStore(
-          /*client=*/servicerouter::cpp2::getClientFactory()
-              .getClientUnique<MononokeAPIServiceAsyncClient>(serviceName),
-          /*repo=*/std::move(repo),
-          /*executor=*/std::move(executor)) {}
+    folly::Executor* executor)
+    : serviceName_(std::move(serviceName)),
+      repo_(std::move(repo)),
+      executor_(executor) {}
 
 MononokeThriftBackingStore::MononokeThriftBackingStore(
-    std::unique_ptr<MononokeAPIServiceAsyncClient> client,
+    std::unique_ptr<MononokeAPIServiceAsyncClient> testClient,
     std::string repo,
-    std::shared_ptr<folly::Executor> executor)
+    folly::Executor* executor)
     : repo_(std::move(repo)),
-      client_(std::move(client)),
-      executor_(std::move(executor)) {}
+      executor_(executor),
+      testClient_(std::move(testClient)) {}
 
 MononokeThriftBackingStore::~MononokeThriftBackingStore() {}
 
@@ -82,8 +80,11 @@ folly::Future<std::unique_ptr<Tree>> MononokeThriftBackingStore::getTree(
   params.set_repo(repo_);
   params.set_tree_hash(treeHash);
 
-  return client_->semifuture_get_tree(params)
-      .via(executor_.get())
+  return withClient([params = std::move(params)](
+                        MononokeAPIServiceAsyncClient* client) {
+           return client->semifuture_get_tree(params);
+         })
+      .via(folly::getCPUExecutor().get())
       .thenValue([id](const MononokeDirectory&& response) {
         auto& files = response.get_files();
 
@@ -121,8 +122,11 @@ folly::Future<std::unique_ptr<Blob>> MononokeThriftBackingStore::getBlob(
   params.set_repo(repo_);
   params.set_blob_hash(blobHash);
 
-  return client_->semifuture_get_blob(params)
-      .via(executor_.get())
+  return withClient([params = std::move(params)](
+                        MononokeAPIServiceAsyncClient* client) {
+           return client->semifuture_get_blob(params);
+         })
+      .via(folly::getCPUExecutor().get())
       .thenValue([id](const MononokeBlob&& response) {
         return std::make_unique<Blob>(id, std::move(*response.get_content()));
       });
@@ -139,11 +143,30 @@ MononokeThriftBackingStore::getTreeForCommit(const Hash& commitID) {
   params.set_repo(repo_);
   params.set_revision(rev);
 
-  return client_->semifuture_get_changeset(params)
-      .via(executor_.get())
+  return withClient([params = std::move(params)](
+                        MononokeAPIServiceAsyncClient* client) {
+           return client->semifuture_get_changeset(params);
+         })
+      .via(executor_)
       .thenValue([this](const MononokeChangeset&& response) {
         return getTree(Hash(response.get_manifest().get_hash()));
       });
 }
+
+template <typename Func>
+std::invoke_result_t<Func, MononokeAPIServiceAsyncClient*>
+MononokeThriftBackingStore::withClient(Func&& func) {
+  return folly::via(executor_, [this, func = std::forward<Func>(func)]() {
+    if (testClient_) {
+      return func(testClient_.get());
+    }
+
+    auto client =
+        servicerouter::cpp2::getClientFactory()
+            .getClientUnique<MononokeAPIServiceAsyncClient>(serviceName_);
+    return func(client.get());
+  });
+}
+
 } // namespace eden
 } // namespace facebook

eden/fs/store/mononoke/MononokeThriftBackingStore.h

@@ -8,11 +8,12 @@
  *
  */
 #pragma once
-
-#include <scm/mononoke/apiserver/gen-cpp2/MononokeAPIServiceAsyncClient.h>
-
 #include "eden/fs/store/BackingStore.h"
 
+namespace scm::mononoke::apiserver::thrift {
+class MononokeAPIServiceAsyncClient;
+}
+
 namespace facebook {
 namespace eden {
 class MononokeThriftBackingStore : public BackingStore {
@@ -20,13 +21,14 @@ class MononokeThriftBackingStore : public BackingStore {
   MononokeThriftBackingStore(
       std::string tierName,
       std::string repo,
-      std::shared_ptr<folly::Executor> executor);
+      folly::Executor* executor);
+
   MononokeThriftBackingStore(
       std::unique_ptr<
           scm::mononoke::apiserver::thrift::MononokeAPIServiceAsyncClient>
-          client,
+          testClient,
       std::string repo,
-      std::shared_ptr<folly::Executor> executor);
+      folly::Executor* executor);
 
   virtual ~MononokeThriftBackingStore() override;
 
@@ -40,11 +42,19 @@ class MononokeThriftBackingStore : public BackingStore {
   MononokeThriftBackingStore(MononokeThriftBackingStore&&) = delete;
   MononokeThriftBackingStore& operator=(MononokeThriftBackingStore&&) = delete;
 
+  template <typename Func>
+  std::invoke_result_t<
+      Func,
+      scm::mononoke::apiserver::thrift::MononokeAPIServiceAsyncClient*>
+  withClient(Func&& func);
+
+  std::string serviceName_;
   std::string repo_;
+  folly::Executor* executor_;
+
   std::unique_ptr<
       scm::mononoke::apiserver::thrift::MononokeAPIServiceAsyncClient>
-      client_;
-  std::shared_ptr<folly::Executor> executor_;
+      testClient_;
 };
 } // namespace eden
 } // namespace facebook

eden/fs/store/mononoke/test/ThriftTest.cpp

@@ -165,7 +165,7 @@ class MononokeThriftTest : public ::testing::Test {
   MononokeThriftBackingStore store = MononokeThriftBackingStore(
       runner.newClient<MononokeAPIServiceAsyncClient>(),
       "fbsource",
-      folly::getIOExecutor());
+      folly::getIOExecutor().get());
 };
 
 TEST_F(MononokeThriftTest, getBlob) {