dedup

component/ipfs_url_loader.cc

@@ -80,8 +80,8 @@ void ipfs::IpfsUrlLoader::StartRequest(
     me->root_ = cid_str;
     me->api_->SetLoaderFactory(*(me->lower_loader_factory_));
     auto whendone = [me](IpfsRequest const& req, ipfs::Response const& res) {
-      VLOG(1) << "whendone(" << req.path().to_string() << ',' << res.status_
-              << ',' << res.body_.size() << "B mime=" << res.mime_ << ')';
+      LOG(INFO) << "whendone(" << req.path().to_string() << ',' << res.status_
+                << ',' << res.body_.size() << "B mime=" << res.mime_ << ')';
       if (!res.body_.empty()) {
         me->ReceiveBlockBytes(res.body_);
       }
@@ -125,7 +125,9 @@ void ipfs::IpfsUrlLoader::BlocksComplete(std::string mime_type) {
   }
   complete_ = true;
   auto head = network::mojom::URLResponseHead::New();
-  head->mime_type = mime_type;
+  if (mime_type.size()) {
+    head->mime_type = mime_type;
+  }
   std::uint32_t byte_count = partial_block_.size();
   VLOG(1) << "Calling WriteData(" << byte_count << ")";
   pipe_prod_->WriteData(partial_block_.data(), &byte_count,
@@ -143,7 +145,9 @@ void ipfs::IpfsUrlLoader::BlocksComplete(std::string mime_type) {
   auto status_line = base::StringPrintf("HTTP/1.1 %d %s", status_, reason);
   VLOG(1) << "Returning with status line '" << status_line << "'.\n";
   head->headers->ReplaceStatusLine(status_line);
-  head->headers->SetHeader("Content-Type", mime_type);
+  if (mime_type.size()) {
+    head->headers->SetHeader("Content-Type", mime_type);
+  }
   head->headers->SetHeader("Access-Control-Allow-Origin", "*");
   head->was_fetched_via_spdy = false;
   for (auto& [n, v] : additional_outgoing_headers_) {
@@ -153,9 +157,8 @@ void ipfs::IpfsUrlLoader::BlocksComplete(std::string mime_type) {
   VLOG(1) << "Calling PopulateParsedHeaders";
   head->parsed_headers =
       network::PopulateParsedHeaders(head->headers.get(), GURL{original_url_});
-  VLOG(1) << "Sending response for " << original_url_ << " with mime type "
-          << head->mime_type << " @" << (void*)(this)
-      ;
+  LOG(INFO) << "Sending response for " << original_url_ << " with mime type "
+            << head->mime_type << " and status line " << status_line;
   client_->OnReceiveResponse(std::move(head), std::move(pipe_cons_),
                              absl::nullopt);
   client_->OnComplete(network::URLLoaderCompletionStatus{});

library/CMakeLists.txt

@@ -3,7 +3,9 @@ project(ipfs_client
     DESCRIPTION "Code related to using IPFS http gateways to resolve ipfs:// and ipns:// URLs"
     LANGUAGES CXX
 )
-
+if(NOT CXX_VERSION)
+    set(CXX_VERSION 20)
+endif()
 if(NOT SETUP_CMAKE_INCLUDED)
     include(../cmake/setup.cmake)
 endif()

library/include/ipfs_client/context_api.h

@@ -21,6 +21,7 @@ struct HttpRequestDescription {
   std::string accept;
   std::optional<std::size_t> max_response_size;
   bool operator==(HttpRequestDescription const&) const;
+  bool operator<(HttpRequestDescription const&) const;
 };
 
 /**

library/src/ipfs_client/context_api.cc

@@ -6,3 +6,7 @@ bool ipfs::HttpRequestDescription::operator==(
     HttpRequestDescription const& r) const {
   return url == r.url && accept == r.accept;
 }
+bool ipfs::HttpRequestDescription::operator<(
+    HttpRequestDescription const& r) const {
+  return url == r.url ? accept < r.accept : url < r.url;
+}

library/src/ipfs_client/gw/deduplicating_requestor.cc

@@ -0,0 +1,41 @@
+#include "deduplicating_requestor.h"
+
+#include <ipfs_client/gw/gateway_request.h>
+
+#include "log_macros.h"
+
+using Self = ipfs::gw::DeduplicatingRequestor;
+
+std::string_view Self::name() const {
+  return "dedup";
+}
+auto Self::handle(ipfs::gw::Requestor::RequestPtr r) -> HandleOutcome {
+  auto d = r->describe_http();
+  if (!d.has_value()) {
+    return HandleOutcome::NOT_HANDLED;
+  }
+  auto& k = d.value();
+  auto it = seen_.find(k);
+  if (seen_.end() == it) {
+    seen_.emplace(k, r);
+    return HandleOutcome::NOT_HANDLED;
+  }
+  auto& w = it->second;
+  auto old = w.lock();
+  if (old == r) {
+    LOG(INFO)
+        << name()
+        << " has seen the EXACT same request pass by (same object in memory).";
+    return HandleOutcome::NOT_HANDLED;
+  }
+  if (old) {
+    LOG(INFO) << "Dedup squashed a new version of the request "
+              << old->debug_string() << " in " << r->debug_string();
+    return HandleOutcome::DONE;
+  } else {
+    LOG(INFO) << r->debug_string()
+              << " has occurred before, but the old copy is dead and gone.";
+    it->second = r;
+    return HandleOutcome::NOT_HANDLED;
+  }
+}

library/src/ipfs_client/gw/deduplicating_requestor.h

@@ -0,0 +1,19 @@
+#ifndef IPFS_DEDUPLICATING_REQUESTOR_H_
+#define IPFS_DEDUPLICATING_REQUESTOR_H_
+
+#include <ipfs_client/gw/requestor.h>
+
+#include <ipfs_client/context_api.h>
+
+#include <map>
+
+namespace ipfs::gw {
+class DeduplicatingRequestor final : public Requestor {
+  std::string_view name() const override;
+  HandleOutcome handle(RequestPtr) override;
+
+  std::map<HttpRequestDescription, std::weak_ptr<GatewayRequest>> seen_;
+};
+}  // namespace ipfs::gw
+
+#endif  // IPFS_DEDUPLICATING_REQUESTOR_H_

library/src/ipfs_client/gw/default_requestor.cc

@@ -1,6 +1,7 @@
 #include <ipfs_client/gw/default_requestor.h>
 
 #include <ipfs_client/gw/block_request_splitter.h>
+#include <ipfs_client/gw/deduplicating_requestor.h>
 #include <ipfs_client/gw/dnslink_requestor.h>
 #include <ipfs_client/gw/gateway_http_requestor.h>
 #include <ipfs_client/gw/inline_request_handler.h>
@@ -12,6 +13,7 @@ auto ipfs::gw::default_requestor(ipfs::GatewayList gws,
                                  std::shared_ptr<ContextApi> api)
     -> std::shared_ptr<Requestor> {
   auto result = std::make_shared<gw::InlineRequestHandler>();
+  result->or_else(std::make_shared<DeduplicatingRequestor>());
   if (early) {
     result->or_else(early);
     early->api(api);

library/src/ipfs_client/gw/gateway_request.cc

@@ -4,6 +4,7 @@
 #include <ipfs_client/ipld/ipns_name.h>
 
 #include <ipfs_client/dag_block.h>
+#include <ipfs_client/ipfs_request.h>
 #include <ipfs_client/ipns_record.h>
 #include <ipfs_client/orchestrator.h>
 #include <ipfs_client/response.h>
@@ -198,6 +199,9 @@ std::string Self::debug_string() const {
   if (!path.empty()) {
     oss << ' ' << path;
   }
+  if (dependent) {
+    oss << " for=" << dependent->path().to_string();
+  }
   oss << " plel=" << parallel << '}';
   return oss.str();
 }

library/src/ipfs_client/gw/requestor_pool.cc

@@ -51,16 +51,17 @@ auto Self::check(ipfs::gw::RequestPtr req, std::size_t start) -> HandleOutcome {
     }
   }
   if (req->parallel > 0) {
-    VLOG(2) << req->parallel << " requestors have picked up the task.";
+    LOG(INFO) << req->parallel << " requestors have picked up the task "
+              << req->debug_string();
     return O::PENDING;
   }
   if (next_retry < pool.size()) {
-    VLOG(1) << "No requestors are available for " << req->debug_string()
-            << " right now, will retry at index " << next_retry;
+    LOG(INFO) << "No requestors are available for " << req->debug_string()
+              << " right now, will retry at index " << next_retry;
     waiting.emplace(req, next_retry);
     return O::PENDING;
   }
-  VLOG(1) << "Have exhausted all requestors in pool looking for "
-          << req->debug_string();
+  LOG(INFO) << "Have exhausted all requestors in pool looking for "
+            << req->debug_string();
   return O::NOT_HANDLED;
 }