store: add support for aux data in hg import queue

Summary:
As a first step towards enabling aux data fetching, we need to allow aux data
to be batch fetched.

Reviewed By: chadaustin

Differential Revision: D44110105

fbshipit-source-id: f9c3f55ffdc5554784b5f38457ef675689664263

Author: xavierd

eden/fs/config/EdenConfig.h

@@ -667,6 +667,15 @@ class EdenConfig : private ConfigSettingManager {
       1,
       this};
 
+  /**
+   * Controls the max number of blob metadata import requests we batch in
+   * HgBackingStore
+   */
+  ConfigSetting<uint32_t> importBatchSizeBlobMeta{
+      "hg:import-batch-size-blobmeta",
+      1024,
+      this};
+
   /**
    * Whether fetching trees should fall back on an external hg importer process.
    */

eden/fs/store/hg/HgImportRequest.cpp

@@ -55,4 +55,12 @@ std::shared_ptr<HgImportRequest> HgImportRequest::makeTreeImportRequest(
   return makeRequest<TreeImport>(priority, cause, hash, proxyHash);
 }
 
+std::shared_ptr<HgImportRequest> HgImportRequest::makeBlobMetaImportRequest(
+    const ObjectId& hash,
+    const HgProxyHash& proxyHash,
+    ImportPriority priority,
+    ObjectFetchContext::Cause cause) {
+  return makeRequest<BlobMetaImport>(priority, cause, hash, proxyHash);
+}
+
 } // namespace facebook::eden

eden/fs/store/hg/HgImportRequest.h

@@ -56,6 +56,18 @@ class HgImportRequest {
     std::vector<folly::Promise<Response>> promises;
   };
 
+  struct BlobMetaImport {
+    using Response = std::unique_ptr<BlobMetadata>;
+    BlobMetaImport(ObjectId hash, HgProxyHash proxyHash)
+        : hash{std::move(hash)}, proxyHash{std::move(proxyHash)} {}
+
+    ObjectId hash;
+    HgProxyHash proxyHash;
+
+    // See the comment above for BlobImport::promises
+    std::vector<folly::Promise<Response>> promises;
+  };
+
   /**
    * Allocate a blob request.
    */
@@ -74,6 +86,12 @@ class HgImportRequest {
       ImportPriority priority,
       ObjectFetchContext::Cause cause);
 
+  static std::shared_ptr<HgImportRequest> makeBlobMetaImportRequest(
+      const ObjectId& hash,
+      const HgProxyHash& proxyHash,
+      ImportPriority priority,
+      ObjectFetchContext::Cause cause);
+
   /**
    * Implementation detail of the make*Request functions from above. Do not use
    * directly.
@@ -147,10 +165,11 @@ class HgImportRequest {
   HgImportRequest(const HgImportRequest&) = delete;
   HgImportRequest& operator=(const HgImportRequest&) = delete;
 
-  using Request = std::variant<BlobImport, TreeImport>;
+  using Request = std::variant<BlobImport, TreeImport, BlobMetaImport>;
   using Response = std::variant<
       folly::Promise<std::unique_ptr<Blob>>,
-      folly::Promise<std::unique_ptr<Tree>>>;
+      folly::Promise<std::unique_ptr<Tree>>,
+      folly::Promise<std::unique_ptr<BlobMetadata>>>;
 
   Request request_;
   ImportPriority priority_;

eden/fs/store/hg/HgImportRequestQueue.cpp

@@ -33,6 +33,14 @@ folly::Future<std::unique_ptr<Tree>> HgImportRequestQueue::enqueueTree(
       std::move(request));
 }
 
+folly::Future<std::unique_ptr<BlobMetadata>>
+HgImportRequestQueue::enqueueBlobMeta(
+    std::shared_ptr<HgImportRequest> request) {
+  return enqueue<
+      std::unique_ptr<BlobMetadata>,
+      HgImportRequest::BlobMetaImport>(std::move(request));
+}
+
 template <typename Ret, typename ImportType>
 folly::Future<Ret> HgImportRequestQueue::enqueue(
     std::shared_ptr<HgImportRequest> request) {
@@ -41,6 +49,10 @@ folly::Future<Ret> HgImportRequestQueue::enqueue(
   std::vector<std::shared_ptr<HgImportRequest>>* queue;
   if constexpr (std::is_same_v<ImportType, HgImportRequest::BlobImport>) {
     queue = &state->blobQueue;
+  } else if constexpr (std::is_same_v<
+                           ImportType,
+                           HgImportRequest::BlobMetaImport>) {
+    queue = &state->blobMetaQueue;
   } else {
     static_assert(std::is_same_v<ImportType, HgImportRequest::TreeImport>);
     queue = &state->treeQueue;
@@ -97,19 +109,26 @@ HgImportRequestQueue::combineAndClearRequestQueues() {
   auto state = state_.lock();
   auto treeQSz = state->treeQueue.size();
   auto blobQSz = state->blobQueue.size();
+  auto blobMetaQSz = state->blobMetaQueue.size();
   XLOGF(
       DBG5,
-      "combineAndClearRequestQueues: tree queue size = {}, blob queue size = {}",
+      "combineAndClearRequestQueues: tree queue size = {}, blob queue size = {}, blob metadata queue size = {}",
       treeQSz,
-      blobQSz);
+      blobQSz,
+      blobMetaQSz);
   auto res = std::move(state->treeQueue);
   res.insert(
       res.end(),
       std::make_move_iterator(state->blobQueue.begin()),
       std::make_move_iterator(state->blobQueue.end()));
+  res.insert(
+      res.end(),
+      std::make_move_iterator(state->blobMetaQueue.begin()),
+      std::make_move_iterator(state->blobMetaQueue.end()));
   state->treeQueue.clear();
   state->blobQueue.clear();
-  XCHECK_EQ(res.size(), treeQSz + blobQSz);
+  state->blobMetaQueue.clear();
+  XCHECK_EQ(res.size(), treeQSz + blobQSz + blobMetaQSz);
   return res;
 }
 
@@ -122,6 +141,7 @@ std::vector<std::shared_ptr<HgImportRequest>> HgImportRequestQueue::dequeue() {
     if (!state->running) {
       state->treeQueue.clear();
       state->blobQueue.clear();
+      state->blobMetaQueue.clear();
       return std::vector<std::shared_ptr<HgImportRequest>>();
     }
 
@@ -137,6 +157,15 @@ std::vector<std::shared_ptr<HgImportRequest>> HgImportRequestQueue::dequeue() {
       queue = &state->treeQueue;
     }
 
+    if (!state->blobMetaQueue.empty()) {
+      auto priority = state->blobMetaQueue.front()->getPriority();
+      if (!queue || priority > highestPriority) {
+        queue = &state->blobMetaQueue;
+        count = config_->getEdenConfig()->importBatchSizeBlobMeta.getValue();
+        highestPriority = priority;
+      }
+    }
+
     if (!state->blobQueue.empty()) {
       auto priority = state->blobQueue.front()->getPriority();
       if (!queue || priority > highestPriority) {

eden/fs/store/hg/HgImportRequestQueue.h

@@ -42,6 +42,14 @@ class HgImportRequestQueue {
   folly::Future<std::unique_ptr<Tree>> enqueueTree(
       std::shared_ptr<HgImportRequest> request);
 
+  /**
+   * Enqueue an aux data request to the queue.
+   *
+   * Return a future that will complete when the aux data request completes.
+   */
+  folly::Future<std::unique_ptr<BlobMetadata>> enqueueBlobMeta(
+      std::shared_ptr<HgImportRequest> request);
+
   /**
    * Returns a list of requests from the queue. It returns an empty list while
    * the queue is being destructed. This function will block when there is no
@@ -87,6 +95,10 @@ class HgImportRequestQueue {
     if constexpr (std::is_same_v<T, Tree>) {
       auto* treeImport = import->getRequest<HgImportRequest::TreeImport>();
       promises = &treeImport->promises;
+    } else if constexpr (std::is_same_v<T, BlobMetadata>) {
+      auto* blobMetaImport =
+          import->getRequest<HgImportRequest::BlobMetaImport>();
+      promises = &blobMetaImport->promises;
     } else {
       static_assert(
           std::is_same_v<T, Blob>,
@@ -130,6 +142,7 @@ class HgImportRequestQueue {
     bool running = true;
     std::vector<std::shared_ptr<HgImportRequest>> treeQueue;
     std::vector<std::shared_ptr<HgImportRequest>> blobQueue;
+    std::vector<std::shared_ptr<HgImportRequest>> blobMetaQueue;
 
     /**
      * Map of a ObjectId to an element in the queue. Any changes to this type