store: allow batch of aux data to be fetched

Summary:
Nothing is being enqueued just yet, but this teaches EdenFS how to fetch
batches of aux data.

In the case where the aux data cannot be fetched, a nullptr auxdata is returned
to the higher level which will then trigger a blob fetch to compute aux data.

Reviewed By: chadaustin

Differential Revision: D44110104

fbshipit-source-id: 53df1496addf3a9dae521ffcdba5b060b8fce16a

Author: xavierd

eden/fs/config/EdenConfig.h

@@ -686,6 +686,14 @@ class EdenConfig : private ConfigSettingManager {
    */
   ConfigSetting<bool> hgBlobFetchFallback{"hg:blob-fetch-fallback", true, this};
 
+  /**
+   * Whether fetching blob metadata should fall back to fetching blobs.
+   */
+  ConfigSetting<bool> hgBlobMetaFetchFallback{
+      "hg:blobmeta-fetch-fallback",
+      true,
+      this};
+
   // [backingstore]
 
   /**

eden/fs/store/BackingStore.h

@@ -120,6 +120,21 @@ class BackingStore : public RootIdCodec, public ObjectIdCodec {
       const ObjectId& id,
       const ObjectFetchContextPtr& context) = 0;
 
+  /**
+   * Return value of the getBlobMetadata method.
+   */
+  struct GetBlobMetaResult {
+    /**
+     * The retrieved blob metadata.
+     *
+     * Setting this to a nullptr will make the ObjectStore fallback to fetching
+     * the blob and computing the blob metadata from it.
+     */
+    std::unique_ptr<BlobMetadata> blobMeta;
+    /** The fetch origin of the blob metadata. */
+    ObjectFetchContext::Origin origin;
+  };
+
   /**
    * Fetch blob metadata if available in a local cache. Returns nullptr if not
    * locally available.

eden/fs/store/hg/HgDatapackStore.cpp

@@ -258,6 +258,55 @@ std::unique_ptr<BlobMetadata> HgDatapackStore::getLocalBlobMetadata(
   return nullptr;
 }
 
+void HgDatapackStore::getBlobMetadataBatch(
+    const std::vector<std::shared_ptr<HgImportRequest>>& importRequests) {
+  size_t count = importRequests.size();
+
+  std::vector<sapling::NodeId> requests;
+  requests.reserve(count);
+  for (const auto& importRequest : importRequests) {
+    requests.emplace_back(
+        importRequest->getRequest<HgImportRequest::BlobMetaImport>()
+            ->proxyHash.byteHash());
+  }
+
+  std::vector<RequestMetricsScope> requestsWatches;
+  requestsWatches.reserve(count);
+  for (auto i = 0ul; i < count; i++) {
+    requestsWatches.emplace_back(&liveBatchedBlobMetaWatches_);
+  }
+
+  store_.getBlobMetadataBatch(
+      folly::range(requests),
+      false,
+      [&](size_t index,
+          folly::Try<std::shared_ptr<sapling::FileAuxData>> auxTry) {
+        if (auxTry.hasException() &&
+            config_->getEdenConfig()->hgBlobMetaFetchFallback.getValue()) {
+          // The caller will fallback to fetching the blob.
+          // TODO: Remove this.
+          return;
+        }
+
+        XLOGF(DBG9, "Imported aux={}", folly::hexlify(requests[index]));
+        auto& importRequest = importRequests[index];
+        importRequest->getPromise<std::unique_ptr<BlobMetadata>>()->setWith(
+            [&]() -> folly::Try<std::unique_ptr<BlobMetadata>> {
+              if (auxTry.hasException()) {
+                return folly::Try<std::unique_ptr<BlobMetadata>>{
+                    std::move(auxTry).exception()};
+              }
+
+              auto& aux = auxTry.value();
+              return folly::Try{std::make_unique<BlobMetadata>(
+                  Hash20{aux->content_sha1}, aux->total_size)};
+            });
+
+        // Make sure that we're stopping this watch.
+        requestsWatches[index].reset();
+      });
+}
+
 void HgDatapackStore::flush() {
   store_.flush();
 }

eden/fs/store/hg/HgDatapackStore.h

@@ -72,6 +72,14 @@ class HgDatapackStore {
    */
   std::unique_ptr<BlobMetadata> getLocalBlobMetadata(const HgProxyHash& id);
 
+  /**
+   * Fetch multiple aux data at once.
+   *
+   * This function returns when all the aux data have been fetched.
+   */
+  void getBlobMetadataBatch(
+      const std::vector<std::shared_ptr<HgImportRequest>>& requests);
+
   /**
    * Flush any pending writes to disk.
    *

eden/fs/store/hg/HgQueuedBackingStore.cpp

@@ -242,6 +242,44 @@ void HgQueuedBackingStore::processTreeImportRequests(
   }
 }
 
+void HgQueuedBackingStore::processBlobMetaImportRequests(
+    std::vector<std::shared_ptr<HgImportRequest>>&& requests) {
+  folly::stop_watch<std::chrono::milliseconds> watch;
+
+  for (auto& request : requests) {
+    auto* blobMetaImport =
+        request->getRequest<HgImportRequest::BlobMetaImport>();
+
+    traceBus_->publish(HgImportTraceEvent::start(
+        request->getUnique(),
+        HgImportTraceEvent::BLOBMETA,
+        blobMetaImport->proxyHash,
+        request->getPriority().getClass(),
+        request->getCause()));
+
+    XLOGF(DBG4, "Processing blob meta request for {}", blobMetaImport->hash);
+  }
+
+  backingStore_->getDatapackStore().getBlobMetadataBatch(requests);
+
+  {
+    for (auto& request : requests) {
+      auto* promise = request->getPromise<std::unique_ptr<BlobMetadata>>();
+      if (promise->isFulfilled()) {
+        stats_->addDuration(
+            &HgBackingStoreStats::fetchBlobMetadata, watch.elapsed());
+        continue;
+      }
+
+      // The code waiting on the promise will fallback to fetching the Blob to
+      // compute the blob metadata. We can't trigger a blob fetch here without
+      // the risk of running into a deadlock: if all import thread are in this
+      // code path, there are no free importer to fetch blobs.
+      promise->setValue(nullptr);
+    }
+  }
+}
+
 void HgQueuedBackingStore::processRequest() {
   folly::setThreadName("hgqueue");
   for (;;) {
@@ -257,6 +295,8 @@ void HgQueuedBackingStore::processRequest() {
       processBlobImportRequests(std::move(requests));
     } else if (first->isType<HgImportRequest::TreeImport>()) {
       processTreeImportRequests(std::move(requests));
+    } else if (first->isType<HgImportRequest::BlobMetaImport>()) {
+      processBlobMetaImportRequests(std::move(requests));
     }
   }
 }
@@ -523,6 +563,84 @@ HgQueuedBackingStore::getBlobImpl(
       });
 }
 
+folly::SemiFuture<BackingStore::GetBlobMetaResult>
+HgQueuedBackingStore::getBlobMetadata(
+    const ObjectId& id,
+    const ObjectFetchContextPtr& context) {
+  DurationScope scope{stats_, &HgBackingStoreStats::getBlobMetadata};
+
+  HgProxyHash proxyHash;
+  try {
+    proxyHash =
+        HgProxyHash::load(localStore_.get(), id, "getBlobMetadata", *stats_);
+  } catch (const std::exception&) {
+    logMissingProxyHash();
+    throw;
+  }
+
+  logBackingStoreFetch(
+      *context,
+      folly::Range{&proxyHash, 1},
+      ObjectFetchContext::ObjectType::BlobMetadata);
+
+  if (auto metadata =
+          backingStore_->getDatapackStore().getLocalBlobMetadata(proxyHash)) {
+    stats_->increment(&ObjectStoreStats::getLocalBlobMetadataFromBackingStore);
+    return folly::makeSemiFuture(GetBlobMetaResult{
+        std::move(metadata), ObjectFetchContext::Origin::FromDiskCache});
+  }
+
+  return getBlobMetadataImpl(id, proxyHash, context)
+      .deferEnsure([scope = std::move(scope)] {});
+}
+
+folly::SemiFuture<BackingStore::GetBlobMetaResult>
+HgQueuedBackingStore::getBlobMetadataImpl(
+    const ObjectId& id,
+    const HgProxyHash& proxyHash,
+    const ObjectFetchContextPtr& context) {
+  auto getBlobMetaFuture = folly::makeFutureWith([&] {
+    XLOG(DBG4) << "make blob meta import request for " << proxyHash.path()
+               << ", hash is:" << id;
+
+    auto request = HgImportRequest::makeBlobMetaImportRequest(
+        id, proxyHash, context->getPriority(), context->getCause());
+    auto unique = request->getUnique();
+
+    auto importTracker =
+        std::make_unique<RequestMetricsScope>(&pendingImportBlobMetaWatches_);
+    traceBus_->publish(HgImportTraceEvent::queue(
+        unique,
+        HgImportTraceEvent::BLOBMETA,
+        proxyHash,
+        context->getPriority().getClass(),
+        context->getCause()));
+
+    return queue_.enqueueBlobMeta(std::move(request))
+        .ensure([this,
+                 unique,
+                 proxyHash,
+                 context = context.copy(),
+                 importTracker = std::move(importTracker)]() {
+          traceBus_->publish(HgImportTraceEvent::finish(
+              unique,
+              HgImportTraceEvent::BLOBMETA,
+              proxyHash,
+              context->getPriority().getClass(),
+              context->getCause()));
+        });
+  });
+
+  return std::move(getBlobMetaFuture)
+      .thenTry([this, id](folly::Try<std::unique_ptr<BlobMetadata>>&& result) {
+        this->queue_.markImportAsFinished<BlobMetadata>(id, result);
+        auto blobMeta = std::move(result).value();
+        stats_->increment(&ObjectStoreStats::getBlobMetadataFromBackingStore);
+        return GetBlobMetaResult{
+            std::move(blobMeta), ObjectFetchContext::Origin::FromNetworkFetch};
+      });
+}
+
 ImmediateFuture<std::unique_ptr<Tree>> HgQueuedBackingStore::getRootTree(
     const RootId& rootId,
     const ObjectFetchContextPtr& /*context*/) {

eden/fs/store/hg/HgQueuedBackingStore.h

@@ -171,6 +171,9 @@ class HgQueuedBackingStore final : public BackingStore {
   folly::SemiFuture<GetBlobResult> getBlob(
       const ObjectId& id,
       const ObjectFetchContextPtr& context) override;
+  folly::SemiFuture<GetBlobMetaResult> getBlobMetadata(
+      const ObjectId& id,
+      const ObjectFetchContextPtr& context);
 
   FOLLY_NODISCARD virtual folly::SemiFuture<folly::Unit> prefetchBlobs(
       ObjectIdRange ids,
@@ -218,7 +221,7 @@ class HgQueuedBackingStore final : public BackingStore {
       std::vector<std::shared_ptr<HgImportRequest>>&& requests);
   void processTreeImportRequests(
       std::vector<std::shared_ptr<HgImportRequest>>&& requests);
-  void processPrefetchRequests(
+  void processBlobMetaImportRequests(
       std::vector<std::shared_ptr<HgImportRequest>>&& requests);
 
   /**
@@ -244,6 +247,18 @@ class HgQueuedBackingStore final : public BackingStore {
       const ObjectId& id,
       const ObjectFetchContextPtr& context) override;
 
+  /**
+   * Fetch the blob metadata from Mercurial.
+   *
+   * For latency sensitive context, the caller is responsible for checking if
+   * the blob metadata is present locally, as this function will always push
+   * the request at the end of the queue.
+   */
+  folly::SemiFuture<GetBlobMetaResult> getBlobMetadataImpl(
+      const ObjectId& id,
+      const HgProxyHash& proxyHash,
+      const ObjectFetchContextPtr& context);
+
   /**
    * Fetch a tree from Mercurial.
    *

eden/fs/telemetry/EdenStats.h

@@ -312,6 +312,7 @@ struct HgBackingStoreStats : StatsGroup<HgBackingStoreStats> {
   Duration fetchBlob{"store.hg.fetch_blob_us"};
   Duration importBlob{"store.hg.import_blob_us"};
   Duration getBlobMetadata{"store.hg.get_blob_metadata_us"};
+  Duration fetchBlobMetadata{"store.hg.get_blob_metadata_us"};
   Counter loadProxyHash{"store.hg.load_proxy_hash"};
   Counter auxMetadataMiss{"store.hg.aux_metadata_miss"};
 };