Add performance and diagnostic counters for remote I/O

cpp/CMakeLists.txt

@@ -170,6 +170,7 @@ set(SOURCES
     "src/logger.cpp"
     "src/mmap.cpp"
     "src/observation.cpp"
+    "src/statistics/counters.cpp"
     "src/statistics/summary.cpp"
     "src/detail/bounce_buffer_cache.cpp"
     "src/detail/concurrent_request_limiter.cpp"

cpp/include/kvikio/shim/libcurl.hpp

@@ -211,6 +211,23 @@ __attribute__((noinline)) inline std::string fix_conda_file_path_hack(std::strin
 }
 }  // namespace detail
 
+namespace detail {
+
+/**
+ * @brief Record what opening a connection cost a finished transfer.
+ *
+ * What resolving, connecting and shaking hands took, which a transfer that reused a connection
+ * paid nothing for.
+ *
+ * libcurl measures this whether or not anybody asks, and reports the phases cumulatively from the
+ * start of the transfer, so they are differenced here.
+ *
+ * @param easy The handle the transfer ran on, after it completed.
+ */
+void count_http_connection_of(CURL* easy) noexcept;
+
+}  // namespace detail
+
 /**
  * @brief Create a new curl handle.
  *

cpp/include/kvikio/statistics/counters.hpp

@@ -0,0 +1,165 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+#include <utility>
+
+#include <kvikio/detail/observation_recorder.hpp>
+#include <kvikio/observation.hpp>
+#include <kvikio/shim/utils.hpp>
+
+namespace KVIKIO_EXPORT kvikio {
+namespace statistics {
+
+/**
+ * @brief Which rows `report()` prints.
+ */
+enum class ReportRows : std::uint8_t {
+  /// Only the backends and subsystems the run used.
+  USED,
+  /// Every row, including the ones the run never used.
+  ALL,
+};
+
+/**
+ * @brief Totals for work that belongs to no single operation.
+ *
+ * An `Observation` records one operation, and part of what I/O costs does not fit there, either
+ * because nothing correlates it with one operation or because it is shared between many.
+ *
+ * These counters count from the moment the process starts, whether or not anybody is watching, so
+ * there is nothing to enable and nothing to switch off. Reading them is `counters()`, and the cost
+ * of an interval is the difference between two readings.
+ */
+struct Counters {
+  /// Remote file sizes asked for, which is an HTTP round trip each.
+  std::uint64_t remote_size_probes{};
+  /// Time spent waiting for them.
+  Duration remote_size_probing{};
+
+  /// Connections libcurl opened, as opposed to reused. Against the requests a run made, this is
+  /// whether connections are being reused at all, which against a TLS endpoint dominates
+  /// everything else.
+  std::uint64_t http_connections{};
+  /// Time those connections spent resolving, connecting, and shaking hands. libcurl measures
+  /// these whether or not anybody asks, so reading them costs nothing.
+  Duration http_dns{};
+  Duration http_tcp{};
+  Duration http_tls{};
+
+  /// Requests the endpoint turned away with a retryable error, and the time spent sleeping before
+  /// trying again. A request that was retried twice counts twice. Nothing else records this, since
+  /// a retry that eventually succeeds is reported as a success.
+  std::uint64_t http_retries{};
+  Duration http_retry_backoff{};
+
+  /**
+   * @brief The difference between this reading and an earlier one.
+   *
+   * @param previous An earlier reading.
+   * @return What was spent in between, saturating at zero.
+   */
+  [[nodiscard]] Counters since(Counters const& previous) const noexcept;
+
+  /**
+   * @brief Whether anything was counted at all.
+   *
+   * @return True if every counter is zero.
+   */
+  [[nodiscard]] bool empty() const noexcept;
+
+  /**
+   * @brief Serialise to JSON.
+   *
+   * @return A JSON object as a string.
+   */
+  [[nodiscard]] std::string to_json() const;
+
+  /**
+   * @brief Format a human-readable report.
+   *
+   * Grouped by subsystem, and a group the run never touched is left out, so an empty reading
+   * formats as an empty string.
+   *
+   * @code
+   *   http size probes     12 probes, 600 ms
+   *   http handshake       128 connections, 40 ms dns, 900 ms tcp, 1.90 s tls
+   * @endcode
+   *
+   * @param rows Which rows to print.
+   * @return The report, newline-terminated, or empty.
+   */
+  [[nodiscard]] std::string report(ReportRows rows = ReportRows::USED) const;
+};
+
+/**
+ * @brief Every counter as it stands now.
+ *
+ * @return The running totals.
+ */
+[[nodiscard]] Counters counters() noexcept;
+
+}  // namespace statistics
+
+namespace detail {
+
+/**
+ * @brief Times a scope and records the duration however the scope is left.
+ *
+ * @code
+ * detail::ScopedTimer const probe{detail::count_remote_size_probe};
+ * curl.perform();  // Counted whether it returns or throws.
+ * @endcode
+ *
+ * @tparam Record What to hand the duration to. A `count_*()` below, or a lambda for one that
+ * takes more than a duration.
+ */
+template <typename Record>
+class ScopedTimer {
+ public:
+  explicit ScopedTimer(Record record) noexcept : _record{std::move(record)}, _started{now()} {}
+
+  ~ScopedTimer() noexcept { _record(now() - _started); }
+
+  ScopedTimer(ScopedTimer const&)            = delete;
+  ScopedTimer& operator=(ScopedTimer const&) = delete;
+
+ private:
+  Record _record;
+  TimePoint _started;
+};
+
+/**
+ * @brief Record asking a remote endpoint how big a file is.
+ *
+ * @param probing How long the round trip took.
+ */
+void count_remote_size_probe(Duration probing) noexcept;
+
+/**
+ * @brief Record what a finished HTTP transfer spent getting connected.
+ *
+ * @param connections Connections opened, which is zero when one was reused.
+ * @param dns Time resolving the name.
+ * @param tcp Time establishing the transport connection.
+ * @param tls Time shaking hands, or zero without TLS.
+ */
+void count_http_connection(std::uint64_t connections,
+                           Duration dns,
+                           Duration tcp,
+                           Duration tls) noexcept;
+
+/**
+ * @brief Record an HTTP request that hit a retryable error and will be tried again.
+ *
+ * @param backoff How long the next attempt waits before it goes out.
+ */
+void count_http_retry(Duration backoff) noexcept;
+
+}  // namespace detail
+}  // namespace KVIKIO_EXPORT kvikio

cpp/include/kvikio/statistics/summary.hpp

@@ -15,6 +15,7 @@
 
 #include <kvikio/observation.hpp>
 #include <kvikio/shim/utils.hpp>
+#include <kvikio/statistics/counters.hpp>
 
 /**
  * @brief KvikIO namespace.
@@ -77,6 +78,14 @@ struct Summary {
   /// whether a read reaches cuFile or falls back to POSIX, so this is where that shows.
   std::array<BackendTotals, num_io_backends> by_backend{};
 
+  /**
+   * @brief The work in the span that belongs to no single operation.
+   *
+   * The counters run for the life of the process, and this is the part of them that falls inside
+   * the span.
+   */
+  Counters counters{};
+
   /// The operations' durations added up, every operation counted.
   Duration total_duration{};
 
@@ -188,9 +197,11 @@ struct Summary {
    *   ...
    * @endcode
    *
+   * @param rows Which rows to print. Under `ReportRows::USED` a backend the run never reached and a
+   * counter group it never touched are left out.
    * @return The report, one field per line, newline-terminated.
    */
-  [[nodiscard]] std::string report() const;
+  [[nodiscard]] std::string report(ReportRows rows = ReportRows::USED) const;
 };
 
 /**
@@ -355,6 +366,11 @@ class SummaryMonitor final : private kvikio::Monitor {
   /// notifications that take it.
   std::mutex _stopping;
 
+  /// The internal counters as they stood when the span began, so the reading is a difference,
+  /// and as they stood when it ended, so a stopped summary does not keep growing.
+  Counters _counters_at_start{};
+  Counters _counters_at_stop{};
+
   /**
    * @brief What every reading is a copy of.
    *

cpp/src/detail/multi_poll_reactor.cpp

@@ -28,6 +28,7 @@
 #include <kvikio/remote_handle.hpp>
 #include <kvikio/shim/cuda.hpp>
 #include <kvikio/shim/libcurl.hpp>
+#include <kvikio/statistics/counters.hpp>
 #include <kvikio/utils.hpp>
 
 namespace kvikio::detail {
@@ -353,6 +354,7 @@ void MultiPollReactor::io_thread_main()
                       std::runtime_error);
         auto transfer = std::move(it->second);
         _in_flight.erase(it);
+        count_http_connection_of(easy);
 
         std::exception_ptr transfer_err;
         try {
@@ -397,6 +399,7 @@ void MultiPollReactor::io_thread_main()
 
             if (outcome.decision == RetryDecision::RETRY) {
               KVIKIO_LOG_WARN(outcome.message);
+              count_http_retry(outcome.delay_ms);
               auto const ready_at = std::chrono::steady_clock::now() + outcome.delay_ms;
               // If a shorter backoff appears
               if (earliest_ready_at.has_value()) {

cpp/src/hdfs.cpp

@@ -7,11 +7,13 @@
 
 #include <kvikio/detail/env.hpp>
 #include <kvikio/detail/nvtx.hpp>
+#include <kvikio/detail/observation_recorder.hpp>
 #include <kvikio/detail/remote_callback.hpp>
 #include <kvikio/error.hpp>
 #include <kvikio/hdfs.hpp>
 #include <kvikio/remote_handle.hpp>
 #include <kvikio/shim/libcurl.hpp>
+#include <kvikio/statistics/counters.hpp>
 
 namespace kvikio {
 
@@ -94,8 +96,10 @@ std::size_t WebHdfsEndpoint::get_file_size()
   std::string response;
   curl.setopt(CURLOPT_WRITEDATA, static_cast<void*>(&response));
   curl.setopt(CURLOPT_WRITEFUNCTION, detail::callback_get_string_response);
-
-  curl.perform([&response] { response.clear(); });
+  {
+    detail::ScopedTimer const probe{detail::count_remote_size_probe};
+    curl.perform([&response] { response.clear(); });
+  }
 
   long http_status_code{};
   curl.getinfo(CURLINFO_RESPONSE_CODE, &http_status_code);

cpp/src/remote_handle.cpp

@@ -32,6 +32,7 @@
 #include <kvikio/hdfs.hpp>
 #include <kvikio/remote_handle.hpp>
 #include <kvikio/shim/libcurl.hpp>
+#include <kvikio/statistics/counters.hpp>
 #include <kvikio/utils.hpp>
 
 namespace kvikio {
@@ -169,7 +170,10 @@ std::size_t get_file_size_using_head_impl(RemoteEndpoint& endpoint, std::string
   endpoint.setopt(curl);
   curl.setopt(CURLOPT_NOBODY, 1L);
   curl.setopt(CURLOPT_FOLLOWLOCATION, 1L);
-  curl.perform();
+  {
+    detail::ScopedTimer const probe{detail::count_remote_size_probe};
+    curl.perform();
+  }
   curl_off_t cl;
   curl.getinfo(CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &cl);
   KVIKIO_EXPECT(
@@ -646,7 +650,10 @@ std::size_t S3EndpointWithPresignedUrl::get_file_size()
   curl.setopt(CURLOPT_HEADERDATA, static_cast<void*>(&file_size));
   curl.setopt(CURLOPT_HEADERFUNCTION, callback_header);
 
-  curl.perform();
+  {
+    detail::ScopedTimer const probe{detail::count_remote_size_probe};
+    curl.perform();
+  }
   return file_size;
 }