From dff001265d1a5e23c092dd09d5cc37689843683f Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Tue, 18 Aug 2026 09:32:16 +0200 Subject: [PATCH 1/2] Add counters for what a remote read costs before it reads --- cpp/CMakeLists.txt | 1 + cpp/include/kvikio/shim/libcurl.hpp | 17 ++ cpp/include/kvikio/statistics/counters.hpp | 165 ++++++++++++++++++ cpp/include/kvikio/statistics/summary.hpp | 18 +- cpp/src/detail/multi_poll_reactor.cpp | 3 + cpp/src/hdfs.cpp | 8 +- cpp/src/remote_handle.cpp | 11 +- cpp/src/shim/libcurl.cpp | 29 ++++ cpp/src/statistics/counters.cpp | 185 +++++++++++++++++++++ cpp/src/statistics/summary.cpp | 52 ++++-- cpp/tests/CMakeLists.txt | 2 + cpp/tests/test_counters.cpp | 81 +++++++++ cpp/tests/test_summary.cpp | 47 +++++- docs/source/statistics.rst | 9 +- python/kvikio/kvikio/_lib/statistics.pyx | 36 +++- python/kvikio/kvikio/statistics.py | 27 ++- python/kvikio/tests/test_http_io.py | 39 +++++ python/kvikio/tests/test_statistics.py | 24 +++ 18 files changed, 718 insertions(+), 36 deletions(-) create mode 100644 cpp/include/kvikio/statistics/counters.hpp create mode 100644 cpp/src/statistics/counters.cpp create mode 100644 cpp/tests/test_counters.cpp diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt index 1b9e5b8992..d3a8f5dce4 100644 --- a/cpp/CMakeLists.txt +++ b/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" diff --git a/cpp/include/kvikio/shim/libcurl.hpp b/cpp/include/kvikio/shim/libcurl.hpp index ae32f1f2ed..e6f45d34d8 100644 --- a/cpp/include/kvikio/shim/libcurl.hpp +++ b/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. * diff --git a/cpp/include/kvikio/statistics/counters.hpp b/cpp/include/kvikio/statistics/counters.hpp new file mode 100644 index 0000000000..b827e3b29c --- /dev/null +++ b/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 +#include +#include + +#include +#include +#include + +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 +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 diff --git a/cpp/include/kvikio/statistics/summary.hpp b/cpp/include/kvikio/statistics/summary.hpp index 056aae2f6b..7defa33054 100644 --- a/cpp/include/kvikio/statistics/summary.hpp +++ b/cpp/include/kvikio/statistics/summary.hpp @@ -15,6 +15,7 @@ #include #include +#include /** * @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 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. * diff --git a/cpp/src/detail/multi_poll_reactor.cpp b/cpp/src/detail/multi_poll_reactor.cpp index 8c65e8f290..8a9b2d41c5 100644 --- a/cpp/src/detail/multi_poll_reactor.cpp +++ b/cpp/src/detail/multi_poll_reactor.cpp @@ -28,6 +28,7 @@ #include #include #include +#include #include 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); + detail::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()) { diff --git a/cpp/src/hdfs.cpp b/cpp/src/hdfs.cpp index aff9236049..eb135f9e8e 100644 --- a/cpp/src/hdfs.cpp +++ b/cpp/src/hdfs.cpp @@ -7,11 +7,13 @@ #include #include +#include #include #include #include #include #include +#include namespace kvikio { @@ -94,8 +96,10 @@ std::size_t WebHdfsEndpoint::get_file_size() std::string response; curl.setopt(CURLOPT_WRITEDATA, static_cast(&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); diff --git a/cpp/src/remote_handle.cpp b/cpp/src/remote_handle.cpp index 20f828a7c0..13083b97ac 100644 --- a/cpp/src/remote_handle.cpp +++ b/cpp/src/remote_handle.cpp @@ -32,6 +32,7 @@ #include #include #include +#include #include 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(&file_size)); curl.setopt(CURLOPT_HEADERFUNCTION, callback_header); - curl.perform(); + { + detail::ScopedTimer const probe{detail::count_remote_size_probe}; + curl.perform(); + } return file_size; } diff --git a/cpp/src/shim/libcurl.cpp b/cpp/src/shim/libcurl.cpp index a5d0259112..fab0490e2b 100644 --- a/cpp/src/shim/libcurl.cpp +++ b/cpp/src/shim/libcurl.cpp @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -24,6 +25,7 @@ #include #include #include +#include #include namespace kvikio { @@ -122,6 +124,31 @@ std::string CurlHandle::error_message() const return std::string{_errbuf}; } +namespace detail { + +void count_http_connection_of(CURL* easy) noexcept +{ + auto const micros = [](curl_off_t us) { return std::chrono::microseconds{us}; }; + + long connections{0}; + if (curl_easy_getinfo(easy, CURLINFO_NUM_CONNECTS, &connections) != CURLE_OK) { return; } + // Zero means the connection was reused, so nothing was paid here. + if (connections <= 0) { return; } + + curl_off_t namelookup{0}; + curl_off_t connect{0}; + curl_off_t appconnect{0}; + curl_easy_getinfo(easy, CURLINFO_NAMELOOKUP_TIME_T, &namelookup); + curl_easy_getinfo(easy, CURLINFO_CONNECT_TIME_T, &connect); + curl_easy_getinfo(easy, CURLINFO_APPCONNECT_TIME_T, &appconnect); + + auto const tcp = connect > namelookup ? micros(connect - namelookup) : Duration::zero(); + auto const tls = appconnect > connect ? micros(appconnect - connect) : Duration::zero(); + count_http_connection(static_cast(connections), micros(namelookup), tcp, tls); +} + +} // namespace detail + void CurlHandle::clear_error_message() noexcept { _errbuf[0] = 0; } void CurlHandle::perform() { perform({}); } @@ -134,6 +161,7 @@ void CurlHandle::perform(std::function const& on_retry) for (std::size_t attempt = 1;; ++attempt) { clear_error_message(); auto const curl_code = curl_easy_perform(handle()); + detail::count_http_connection_of(handle()); long http_code = 0; // We had an error. Is it retryable? @@ -145,6 +173,7 @@ void CurlHandle::perform(std::function const& on_retry) case detail::RetryDecision::SUCCESS: return; case detail::RetryDecision::RETRY: KVIKIO_LOG_WARN(outcome.message); + detail::count_http_retry(outcome.delay_ms); if (on_retry) { on_retry(); } std::this_thread::sleep_for(outcome.delay_ms); break; diff --git a/cpp/src/statistics/counters.cpp b/cpp/src/statistics/counters.cpp new file mode 100644 index 0000000000..078ef0166d --- /dev/null +++ b/cpp/src/statistics/counters.cpp @@ -0,0 +1,185 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace kvikio { + +namespace { + +/** + * @brief The counters themselves. + * + * Relaxed throughout. Each is a running total that nothing else is ordered against, and a reading + * taken while another thread is incrementing is a snapshot of a moving target either way. + */ +struct Atomics { + std::atomic remote_size_probes{0}; + std::atomic remote_size_probing_ns{0}; + std::atomic http_connections{0}; + std::atomic http_dns_ns{0}; + std::atomic http_tcp_ns{0}; + std::atomic http_tls_ns{0}; + std::atomic http_retries{0}; + std::atomic http_retry_backoff_ns{0}; +}; + +/** + * @brief The one set of counters. + * + * Intentionally leaked, since a thread pool can be torn down during static destruction and must + * still find somewhere to count. + */ +Atomics& atomics() noexcept +{ + static auto* instance = new Atomics{}; + return *instance; +} + +/// Add a duration to a counter of nanoseconds. +void add(std::atomic& counter, Duration duration) noexcept +{ + counter.fetch_add(duration.count(), std::memory_order_relaxed); +} + +} // namespace + +namespace detail { + +void count_remote_size_probe(Duration probing) noexcept +{ + auto& all = atomics(); + all.remote_size_probes.fetch_add(1, std::memory_order_relaxed); + add(all.remote_size_probing_ns, probing); +} + +void count_http_connection(std::uint64_t connections, + Duration dns, + Duration tcp, + Duration tls) noexcept +{ + auto& all = atomics(); + all.http_connections.fetch_add(connections, std::memory_order_relaxed); + add(all.http_dns_ns, dns); + add(all.http_tcp_ns, tcp); + add(all.http_tls_ns, tls); +} + +void count_http_retry(Duration backoff) noexcept +{ + auto& all = atomics(); + all.http_retries.fetch_add(1, std::memory_order_relaxed); + add(all.http_retry_backoff_ns, backoff); +} + +} // namespace detail + +namespace statistics { + +Counters counters() noexcept +{ + auto& all = atomics(); + Counters ret; + ret.remote_size_probes = all.remote_size_probes.load(std::memory_order_relaxed); + ret.remote_size_probing = Duration{all.remote_size_probing_ns.load(std::memory_order_relaxed)}; + ret.http_connections = all.http_connections.load(std::memory_order_relaxed); + ret.http_dns = Duration{all.http_dns_ns.load(std::memory_order_relaxed)}; + ret.http_tcp = Duration{all.http_tcp_ns.load(std::memory_order_relaxed)}; + ret.http_tls = Duration{all.http_tls_ns.load(std::memory_order_relaxed)}; + ret.http_retries = all.http_retries.load(std::memory_order_relaxed); + ret.http_retry_backoff = Duration{all.http_retry_backoff_ns.load(std::memory_order_relaxed)}; + return ret; +} + +namespace { +// Saturating, since a counter only grows but the two readings need not be ordered. +template +[[nodiscard]] constexpr T spent(T mine, T previous) noexcept +{ + return mine > previous ? mine - previous : T{}; +} +} // namespace + +Counters Counters::since(Counters const& previous) const noexcept +{ + // Every counter is named below. Adding one to the struct changes its size, which fails here + // until it is named too. + static_assert(sizeof(Counters) == 8 * sizeof(std::uint64_t), + "a counter was added or removed: difference it below and correct this size"); + return Counters{ + .remote_size_probes = spent(remote_size_probes, previous.remote_size_probes), + .remote_size_probing = spent(remote_size_probing, previous.remote_size_probing), + .http_connections = spent(http_connections, previous.http_connections), + .http_dns = spent(http_dns, previous.http_dns), + .http_tcp = spent(http_tcp, previous.http_tcp), + .http_tls = spent(http_tls, previous.http_tls), + .http_retries = spent(http_retries, previous.http_retries), + .http_retry_backoff = spent(http_retry_backoff, previous.http_retry_backoff), + }; +} + +bool Counters::empty() const noexcept +{ + return remote_size_probes == 0 && http_connections == 0 && http_retries == 0; +} + +std::string Counters::to_json() const +{ + std::ostringstream os; + os.imbue(std::locale::classic()); + os << "{\"remote_size_probes\": " << remote_size_probes + << ", \"remote_size_probing_ns\": " << remote_size_probing.count() + << ", \"http_connections\": " << http_connections << ", \"http_dns_ns\": " << http_dns.count() + << ", \"http_tcp_ns\": " << http_tcp.count() << ", \"http_tls_ns\": " << http_tls.count() + << ", \"http_retries\": " << http_retries + << ", \"http_retry_backoff_ns\": " << http_retry_backoff.count() << "}"; + return os.str(); +} + +std::string Counters::report(ReportRows rows) const +{ + std::ostringstream os; + // The same width the summary's rows use, so the two read as one report. + auto const row = [&os](char const* label, std::string const& value) { + os << " " << std::left << std::setw(21) << label << value << "\n"; + }; + + // All of it or none of it. A zero here is an answer, `0 retries` or nothing spent on TLS, once + // the run has done any remote I/O at all. + if (rows != ReportRows::ALL && empty()) { return os.str(); } + + { + std::ostringstream value; + value << remote_size_probes << " probes, " << detail::format_duration(remote_size_probing); + row("http size probes", value.str()); + } + { + std::ostringstream value; + value << http_connections << " connections, " << detail::format_duration(http_dns) << " dns, " + << detail::format_duration(http_tcp) << " tcp, " << detail::format_duration(http_tls) + << " tls"; + row("http handshake", value.str()); + } + { + std::ostringstream value; + value << http_retries << " retries, " << detail::format_duration(http_retry_backoff) + << " backoff"; + row("http retries", value.str()); + } + return os.str(); +} + +} // namespace statistics +} // namespace kvikio diff --git a/cpp/src/statistics/summary.cpp b/cpp/src/statistics/summary.cpp index 145b0bda9a..c64abcf472 100644 --- a/cpp/src/statistics/summary.cpp +++ b/cpp/src/statistics/summary.cpp @@ -55,7 +55,7 @@ constexpr std::array serial_magic{ /// Bumped whenever a field is added, removed or reordered. The size below catches most of that on /// its own, but not two fields of the same width swapping places. -constexpr std::uint32_t serial_version = 1; +constexpr std::uint32_t serial_version = 2; /// Written in the writer's byte order rather than in the header's, so that a reader whose order /// differs sees it scrambled and refuses the payload instead of reinterpreting it. @@ -160,7 +160,8 @@ std::string Summary::to_json() const << ", \"busy_ns\": " << busy.count() << ", \"total_duration_ns\": " << total_duration.count() << ", \"busy_bytes_per_sec\": " << busy_bytes_per_sec() << ", \"busy_fraction\": " << busy_fraction() - << ", \"mean_duration_ns\": " << mean_duration().count() << ", \"by_backend\": {"; + << ", \"mean_duration_ns\": " << mean_duration().count() + << ", \"counters\": " << counters.to_json() << ", \"by_backend\": {"; for (std::size_t i = 0; i < num_io_backends; ++i) { auto const& totals = by_backend[i]; if (i != 0) { os << ", "; } @@ -174,7 +175,7 @@ std::string Summary::to_json() const return os.str(); } -std::string Summary::report() const +std::string Summary::report(ReportRows rows) const { std::ostringstream os; // The width is that of the widest label below, so the values line up. @@ -210,14 +211,12 @@ std::string Summary::report() const detail::format_nbytes(bytes_written), " written)"); row("errors", num_errors); - // Every backend gets a row, including the ones that carried nothing, since "no GDS here" is an - // answer as often as the bytes are, and a report of a fixed shape can be compared with another. for (std::size_t i = 0; i < num_io_backends; ++i) { auto const& totals = by_backend[i]; auto const label = std::string{"backend "} + std::string{kvikio::to_string(static_cast(i))}; if (totals.num_ops == 0) { - row(label, "unused"); + if (rows == ReportRows::ALL) { row(label, "unused"); } continue; } std::ostringstream value; @@ -232,6 +231,12 @@ std::string Summary::report() const if (totals.num_errors != 0) { value << ", " << totals.num_errors << " failed"; } row(label, value.str()); } + // Last, since these belong to no single operation rather than to what the run asked for. A span + // that did remote I/O gets the rows even when every counter is zero, which is what a reused + // connection looks like. + auto const remote = by_backend[static_cast(IoBackend::REMOTE_HTTP)].num_ops != 0 || + by_backend[static_cast(IoBackend::REMOTE_HDFS)].num_ops != 0; + os << counters.report(remote ? ReportRows::ALL : rows); return os.str(); } @@ -244,10 +249,11 @@ SummaryMonitor::SummaryMonitor(Callback on_destruction) : _on_destruction{std::m // rather than by only one of them, which would leave the tracker's in-flight count unbalanced. _registration = register_monitor(this); std::lock_guard const lock{_mutex}; - auto const t = detail::now(); - _totals.start = t; - _totals.anchor = ClockAnchor::now(); - _registered = t; + auto const t = detail::now(); + _totals.start = t; + _totals.anchor = ClockAnchor::now(); + _registered = t; + _counters_at_start = statistics::counters(); _busy.reset(t); } @@ -352,8 +358,16 @@ void SummaryMonitor::on_finish(Observation const& observation) noexcept Summary SummaryMonitor::get() const { + // Read before the lock, since the counters take no lock of their own and this one guards the + // rest. + auto const current_counters = statistics::counters(); + std::lock_guard const lock{_mutex}; Summary ret = _totals; + // The counters are the whole process's and go on rising, so a stopped monitor reports the ones + // it stopped at rather than whatever has happened since. + ret.counters = + (_stopped_end != TimePoint{} ? _counters_at_stop : current_counters).since(_counters_at_start); // Measured to the same instant the reading is stamped with, inside the lock, so that busy time // can never run past the end of the span it is divided by and `busy_fraction()` stays <= 1. ret.end = _stopped_end != TimePoint{} ? _stopped_end : detail::now(); @@ -398,6 +412,7 @@ Summary Summary::since(Summary const& previous) const .bytes_written = subtract(bytes_written, previous.bytes_written), .num_errors = subtract(num_errors, previous.num_errors), .by_backend = backends, + .counters = counters.since(previous.counters), .total_duration = subtract(total_duration, previous.total_duration), .busy = subtract(busy, previous.busy)}; } @@ -407,11 +422,12 @@ Summary SummaryMonitor::since(Summary const& previous) const { return get().sinc void SummaryMonitor::reset() { std::lock_guard const lock{_mutex}; - auto const t = detail::now(); - auto const anchor = _totals.anchor; - _totals = Summary{}; - _totals.start = t; - _totals.anchor = anchor; + auto const t = detail::now(); + auto const anchor = _totals.anchor; + _totals = Summary{}; + _totals.start = t; + _totals.anchor = anchor; + _counters_at_start = statistics::counters(); // Keeps `busy <= wall()` across the reset: an operation already in flight contributes only the // part of its span that follows it. _busy.reset(t); @@ -426,8 +442,12 @@ void SummaryMonitor::stop() _registration = 0; // Stamped after the drain, so every operation that was going to be counted ends at or before it. // From here on this is the end of the measured span, however long the monitor lives on. + // Stamped after the drain, like the span's end, so it holds everything that was going to be + // counted. + auto const final_counters = statistics::counters(); std::lock_guard const lock{_mutex}; - _stopped_end = detail::now(); + _counters_at_stop = final_counters; + _stopped_end = detail::now(); } } // namespace statistics diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt index f4e62d0ad3..cfc757dbf7 100644 --- a/cpp/tests/CMakeLists.txt +++ b/cpp/tests/CMakeLists.txt @@ -153,6 +153,8 @@ kvikio_add_test(NAME SUMMARY_TEST SOURCES test_summary.cpp) kvikio_add_test(NAME STRING_UTILS_TEST SOURCES test_string_utils.cpp) +kvikio_add_test(NAME COUNTERS_TEST SOURCES test_counters.cpp) + kvikio_add_test(NAME MMAP_TEST SOURCES test_mmap.cpp) kvikio_add_test(NAME CONCURRENT_REQUEST_LIMITER_TEST SOURCES test_concurrent_request_limiter.cpp) diff --git a/cpp/tests/test_counters.cpp b/cpp/tests/test_counters.cpp new file mode 100644 index 0000000000..a6f552767a --- /dev/null +++ b/cpp/tests/test_counters.cpp @@ -0,0 +1,81 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include +#include +#include +#include +#include + +#include + +#include + +using kvikio::statistics::Counters; +using kvikio::statistics::counters; +using kvikio::statistics::ReportRows; + +TEST(CountersTest, every_counter_is_differenced_and_none_of_them_wrap) +{ + // A reading of the live counters differences against itself, whatever the process has done. + auto const now = counters(); + EXPECT_TRUE(now.since(now).empty()); + + // `since()` names each counter, so this checks that none was missed. Every field is eight bytes + // wide, which is what lets the whole struct be filled a word at a time. + constexpr auto words = sizeof(Counters) / sizeof(std::uint64_t); + std::array earlier{}; + std::array later{}; + for (std::size_t i = 0; i < words; ++i) { + earlier[i] = i + 1; + later[i] = (i + 1) * 10; + } + + auto const spent = std::bit_cast(later).since(std::bit_cast(earlier)); + auto const got = std::bit_cast>(spent); + for (std::size_t i = 0; i < words; ++i) { + EXPECT_EQ(got[i], later[i] - earlier[i]) << "counter " << i << " was not differenced"; + } + + // Backwards, every counter saturates rather than wrapping. + EXPECT_TRUE(std::bit_cast(earlier).since(std::bit_cast(later)).empty()); + + // And the words really are the named counters, not some other layout. + Counters one; + one.http_retries = 7; + EXPECT_EQ(one.since(Counters{}).http_retries, 7); + EXPECT_EQ(one.since(Counters{}).http_connections, 0); +} + +TEST(CountersTest, a_reading_formats_only_what_was_paid_for) +{ + Counters nothing; + EXPECT_TRUE(nothing.empty()); + EXPECT_EQ(nothing.report(), "") << "a reading of nothing should say nothing"; + // Unless everything is asked for, which is how to see what can be recorded at all. + EXPECT_NE(nothing.report(ReportRows::ALL).find("http"), std::string::npos); + + Counters reading; + reading.remote_size_probes = 12; + reading.remote_size_probing = std::chrono::milliseconds{600}; + reading.http_connections = 128; + reading.http_dns = std::chrono::milliseconds{40}; + reading.http_tcp = std::chrono::milliseconds{900}; + reading.http_tls = std::chrono::milliseconds{1900}; + reading.http_retries = 7; + reading.http_retry_backoff = std::chrono::milliseconds{1200}; + + auto const text = reading.report(); + EXPECT_NE(text.find("http size probes 12 probes, 600 ms"), std::string::npos); + EXPECT_NE(text.find("128 connections, 40 ms dns, 900 ms tcp, 1.90 s tls"), std::string::npos); + EXPECT_NE(text.find("http retries 7 retries, 1.20 s backoff"), std::string::npos); + + auto const json = reading.to_json(); + EXPECT_NE(json.find("\"http_retries\": 7"), std::string::npos); + EXPECT_NE(json.find("\"http_retry_backoff_ns\": 1200000000"), std::string::npos); + // The JSON schema is the same whatever was used, unlike the report. + EXPECT_NE(nothing.to_json().find("\"http_connections\": 0"), std::string::npos); +} diff --git a/cpp/tests/test_summary.cpp b/cpp/tests/test_summary.cpp index 1332a41f40..07f901205d 100644 --- a/cpp/tests/test_summary.cpp +++ b/cpp/tests/test_summary.cpp @@ -56,6 +56,7 @@ class SummaryTest : public testing::Test { std::vector _data; }; +using kvikio::statistics::ReportRows; using kvikio::statistics::Summary; using kvikio::statistics::SummaryMonitor; @@ -95,6 +96,26 @@ TEST_F(SummaryTest, a_monitor_counts_the_calls_it_spans) EXPECT_GT(s.mean_duration(), kvikio::Duration::zero()); } +TEST_F(SummaryTest, a_span_that_did_remote_io_reports_the_http_rows_even_when_they_are_zero) +{ + SummaryMonitor const monitor; + { + kvikio::detail::LogicalObservationRecorder remote{ + IoBackend::REMOTE_HTTP, TransferDirection::READ, MemoryKind::HOST, 0, 1024}; + remote.finish(1024); + } + + auto const s = monitor.get(); + // Nothing was probed, connected or retried inside the span, which is what perfect reuse of a + // connection opened before it looks like, and is worth a row rather than a silence. + EXPECT_TRUE(s.counters.empty()); + auto const text = s.report(); + EXPECT_NE(text.find("http handshake 0 connections"), std::string::npos); + EXPECT_NE(text.find("http retries 0 retries"), std::string::npos); + // A span that never left the machine still gets none of them. + EXPECT_EQ(Summary{}.report().find("http"), std::string::npos); +} + TEST_F(SummaryTest, the_backends_that_carried_the_work_are_counted_apart) { SummaryMonitor const monitor; @@ -127,8 +148,9 @@ TEST_F(SummaryTest, the_backends_that_carried_the_work_are_counted_apart) auto const text = s.report(); EXPECT_NE(text.find("backend GDS 1 KiB in 1 ops"), std::string::npos); EXPECT_NE(text.find("backend POSIX 768 B in 2 ops"), std::string::npos); - // A backend that carried nothing says so, since that is the answer as often as the bytes are. - EXPECT_NE(text.find("backend MMAP unused"), std::string::npos); + // A backend the run never reached is left out, and says so when every row is asked for. + EXPECT_EQ(text.find("backend MMAP"), std::string::npos); + EXPECT_NE(s.report(ReportRows::ALL).find("backend MMAP unused"), std::string::npos); EXPECT_NE(s.to_json().find("\"by_backend\""), std::string::npos); } @@ -669,6 +691,27 @@ TEST_F(SummaryTest, bytes_that_are_not_a_summary_are_refused) EXPECT_THROW(std::ignore = Summary::deserialize(other_endian), std::invalid_argument); } +TEST_F(SummaryTest, the_internal_costs_are_those_of_the_span) +{ + // Everything counted is remote, so a run against a local file owes nothing and says so by + // leaving the rows out. + std::vector buffer(_data.size()); + SummaryMonitor monitor; + { + kvikio::FileHandle f{_filepath, "r"}; + f.pread(buffer.data(), nbytes(), 0).get(); + } + + auto const s = monitor.get(); + EXPECT_TRUE(s.counters.empty()) << "a local read owes none of what is counted"; + EXPECT_EQ(s.report().find("size probes"), std::string::npos); + // The JSON schema does not vary, unlike the report. + EXPECT_NE(s.to_json().find("\"counters\""), std::string::npos); + + monitor.reset(); + EXPECT_TRUE(monitor.get().counters.empty()); +} + TEST_F(SummaryTest, the_report_is_human_readable) { std::vector buffer(_data.size()); diff --git a/docs/source/statistics.rst b/docs/source/statistics.rst index 4e17a35dc6..2d169f52fd 100644 --- a/docs/source/statistics.rst +++ b/docs/source/statistics.rst @@ -27,10 +27,11 @@ by a person: bytes 3.12 KiB of 3.12 KiB requested (2.34 KiB read, 800 B written) errors 0 backend POSIX 3.12 KiB in 5 ops, 5.66 ms, 565.50 kB/s - backend GDS unused - backend MMAP unused - backend REMOTE_HTTP unused - backend REMOTE_HDFS unused + +A report holds what the run used, so a backend it never reached and a subsystem it never +touched are left out. To print every row whatever the run did:: + + print(summary.report(all_rows=True)) Busy time and bandwidth ----------------------- diff --git a/python/kvikio/kvikio/_lib/statistics.pyx b/python/kvikio/kvikio/_lib/statistics.pyx index 903276362d..2f6bc24b7c 100644 --- a/python/kvikio/kvikio/_lib/statistics.pyx +++ b/python/kvikio/kvikio/_lib/statistics.pyx @@ -42,6 +42,24 @@ cdef extern from "" namespace "kvikio" nogil: const size_t num_io_backends +cdef extern from "" nogil: + cdef cppclass cpp_Counters "kvikio::statistics::Counters": + uint64_t remote_size_probes + cpp_Duration remote_size_probing + uint64_t http_connections + cpp_Duration http_dns + cpp_Duration http_tcp + cpp_Duration http_tls + uint64_t http_retries + cpp_Duration http_retry_backoff + + +cdef extern from "" namespace "kvikio::statistics" nogil: + cdef enum cpp_ReportRows "kvikio::statistics::ReportRows": + USED "kvikio::statistics::ReportRows::USED" + ALL "kvikio::statistics::ReportRows::ALL" + + cdef extern from "" nogil: cdef cppclass cpp_BackendTotals "kvikio::statistics::Summary::BackendTotals": uint64_t num_ops @@ -64,13 +82,14 @@ cdef extern from "" nogil: cpp_Duration busy cpp_Duration total_duration cpp_BackendTotals[5] by_backend + cpp_Counters counters cpp_Duration wall() except + double busy_bytes_per_sec() except + double busy_fraction() except + cpp_Duration mean_duration() except + cpp_Summary since(const cpp_Summary& previous) except + string to_json() except + - string report() except + + string report(cpp_ReportRows rows) except + cdef cppclass cpp_SummaryMonitor "kvikio::statistics::SummaryMonitor": cpp_SummaryMonitor() except + @@ -162,6 +181,16 @@ cdef class Summary: } for i, name in enumerate(_BACKEND_NAMES) }, + "counters": { + "remote_size_probes": self._handle.counters.remote_size_probes, + "remote_size_probing_ns": self._handle.counters.remote_size_probing.count(), + "http_connections": self._handle.counters.http_connections, + "http_dns_ns": self._handle.counters.http_dns.count(), + "http_tcp_ns": self._handle.counters.http_tcp.count(), + "http_tls_ns": self._handle.counters.http_tls.count(), + "http_retries": self._handle.counters.http_retries, + "http_retry_backoff_ns": self._handle.counters.http_retry_backoff.count(), + }, "wall_ns": self._handle.wall().count(), "busy_bytes_per_sec": self._handle.busy_bytes_per_sec(), "busy_fraction": self._handle.busy_fraction(), @@ -174,8 +203,9 @@ cdef class Summary: def to_json(self) -> str: return self._handle.to_json().decode() - def report(self) -> str: - return self._handle.report().decode() + def report(self, all_rows: bool = False) -> str: + cdef cpp_ReportRows rows = ALL if all_rows else USED + return self._handle.report(rows).decode() def serialize(self) -> bytes: return kvikio_summary_to_bytes(self._handle) diff --git a/python/kvikio/kvikio/statistics.py b/python/kvikio/kvikio/statistics.py index 64579fbdc6..c0991a1137 100644 --- a/python/kvikio/kvikio/statistics.py +++ b/python/kvikio/kvikio/statistics.py @@ -91,9 +91,18 @@ class BackendTotals(TypedDict): backend. There is no per-backend busy time, that being a union over wall time which two backends running at once would both claim. - Excluded from :func:`hash` as the only unhashable field, and only from that. It - still takes part in ``==``, so two summaries that differ here are unequal, they - merely share a hash bucket. + Excluded from :func:`hash`, as an unhashable field, and only from that. It still + takes part in ``==``, so two summaries that differ here are unequal, they merely + share a hash bucket. + """ + + counters: dict[str, int] = field(hash=False) + """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. + + Excluded from :func:`hash` for the same reason as :attr:`by_backend`. """ wall_ns: int @@ -215,17 +224,23 @@ def __reduce__(self): # The C++ handle cannot be pickled, so a summary travels as its bytes. return (Summary.deserialize, (self.serialize(),)) - def report(self) -> str: + def report(self, all_rows: bool = False) -> str: """Format a human-readable report of every field Byte counts, durations and rates are scaled to readable units. Use :meth:`to_json` instead when the output is going to be parsed. + Parameters + ---------- + all_rows + Print every row, including the backends the run never reached and the + subsystems it never touched. + Returns ------- The report, one field per line, newline-terminated. """ - return self._handle.report() + return self._handle.report(all_rows) def __str__(self) -> str: return self.report() @@ -289,7 +304,7 @@ def since(self, previous: Summary) -> Summary: Parameters ---------- - previous : Summary + previous An earlier reading from this monitor. Returns diff --git a/python/kvikio/tests/test_http_io.py b/python/kvikio/tests/test_http_io.py index 6b855d4d78..2306c023c4 100644 --- a/python/kvikio/tests/test_http_io.py +++ b/python/kvikio/tests/test_http_io.py @@ -160,6 +160,45 @@ def test_file_size(http_server, tmpdir): assert f.nbytes() == a.nbytes +def test_a_failed_size_probe_is_still_counted(http_server): + """A probe that threw cost the same round trip as one that returned.""" + monitor = kvikio.SummaryMonitor() + with pytest.raises(RuntimeError): + kvikio.RemoteFile.open_http(f"{http_server}/not-there") + + spent = monitor.get().counters + assert spent["remote_size_probes"] == 1, "a failed probe was not counted" + assert spent["remote_size_probing_ns"] > 0 + + +def test_counters_count_what_the_endpoint_cost(http_server, tmpdir): + """The http counters are the only ones a local read cannot exercise.""" + a = np.arange(1000) + a.tofile(tmpdir / "a") + + monitor = kvikio.SummaryMonitor() + with kvikio.RemoteFile.open_http(f"{http_server}/a") as f: + b = np.empty_like(a) + assert f.read(b) == a.nbytes + spent = monitor.get().counters + + # Opening the file asks how big it is, which is a round trip of its own. + assert spent["remote_size_probes"] >= 1 + assert spent["remote_size_probing_ns"] > 0 + # Something had to be connected to, and nothing was turned away. + assert spent["http_connections"] >= 1 + assert spent["http_tcp_ns"] > 0 + assert spent["http_retries"] == 0 + assert spent["http_retry_backoff_ns"] == 0 + + # Stopping freezes them, so a stopped summary does not keep growing with the process. + monitor.stop() + stopped = monitor.get().counters + with kvikio.RemoteFile.open_http(f"{http_server}/a") as f: + f.read(np.empty_like(a)) + assert monitor.get().counters == stopped, "a stopped monitor kept counting" + + @pytest.mark.parametrize("size", [10, 100, 1000]) @pytest.mark.parametrize("nthreads", [1, 3]) @pytest.mark.parametrize("tasksize", [99, 999]) diff --git a/python/kvikio/tests/test_statistics.py b/python/kvikio/tests/test_statistics.py index 684f7b2322..49acc35900 100644 --- a/python/kvikio/tests/test_statistics.py +++ b/python/kvikio/tests/test_statistics.py @@ -254,3 +254,27 @@ def test_derived_values(a_file): assert summary.busy_bytes_per_sec * summary.busy_fraction == pytest.approx( whole_span, rel=0.01 ) + + +def test_counters_are_those_of_the_span(a_file): + path, nbytes = a_file + buffer = np.empty(nbytes // 8, dtype="u8") + + monitor = kvikio.SummaryMonitor() + with kvikio.CuFile(path, "r") as f: + f.read(buffer) + + # Everything counted is remote, so a local read owes none of it. + spent = monitor.get().counters + assert "http_connections" in spent + assert all(value == 0 for value in spent.values()) + + # The report leaves them out, and the JSON keeps its shape either way. + report = str(monitor.get()) + assert report.startswith("KvikIO I/O summary") + assert "size probes" not in report + assert json.loads(monitor.get().to_json())["counters"]["http_retries"] == 0 + assert "size probes" in monitor.get().report(all_rows=True) + + monitor.reset() + assert all(value == 0 for value in monitor.get().counters.values()) From 3c3f0afc093cd1681254d4b409aa53f110f80b58 Mon Sep 17 00:00:00 2001 From: "Mads R. B. Kristensen" Date: Fri, 28 Aug 2026 08:16:13 +0200 Subject: [PATCH 2/2] Based on review by @kingcrimsontianyu --- cpp/src/detail/multi_poll_reactor.cpp | 2 +- cpp/src/shim/libcurl.cpp | 28 +++++++++++++++++---------- cpp/src/statistics/counters.cpp | 5 ++++- 3 files changed, 23 insertions(+), 12 deletions(-) diff --git a/cpp/src/detail/multi_poll_reactor.cpp b/cpp/src/detail/multi_poll_reactor.cpp index 8a9b2d41c5..bf05fa859d 100644 --- a/cpp/src/detail/multi_poll_reactor.cpp +++ b/cpp/src/detail/multi_poll_reactor.cpp @@ -354,7 +354,7 @@ void MultiPollReactor::io_thread_main() std::runtime_error); auto transfer = std::move(it->second); _in_flight.erase(it); - detail::count_http_connection_of(easy); + count_http_connection_of(easy); std::exception_ptr transfer_err; try { diff --git a/cpp/src/shim/libcurl.cpp b/cpp/src/shim/libcurl.cpp index fab0490e2b..3ddfbe1335 100644 --- a/cpp/src/shim/libcurl.cpp +++ b/cpp/src/shim/libcurl.cpp @@ -125,26 +125,34 @@ std::string CurlHandle::error_message() const } namespace detail { +namespace { +/// A libcurl timing in microseconds, or zero if it could not be read. libcurl fills the value in +/// only when the call succeeds, so a failed one leaves the phase uncounted rather than garbage. +[[nodiscard]] curl_off_t timing_of(CURL* easy, CURLINFO info) noexcept +{ + curl_off_t value{0}; + if (curl_easy_getinfo(easy, info, &value) != CURLE_OK) { return 0; } + return value; +} +} // namespace void count_http_connection_of(CURL* easy) noexcept { - auto const micros = [](curl_off_t us) { return std::chrono::microseconds{us}; }; + using std::chrono::microseconds; long connections{0}; if (curl_easy_getinfo(easy, CURLINFO_NUM_CONNECTS, &connections) != CURLE_OK) { return; } // Zero means the connection was reused, so nothing was paid here. if (connections <= 0) { return; } - curl_off_t namelookup{0}; - curl_off_t connect{0}; - curl_off_t appconnect{0}; - curl_easy_getinfo(easy, CURLINFO_NAMELOOKUP_TIME_T, &namelookup); - curl_easy_getinfo(easy, CURLINFO_CONNECT_TIME_T, &connect); - curl_easy_getinfo(easy, CURLINFO_APPCONNECT_TIME_T, &appconnect); + auto const namelookup = timing_of(easy, CURLINFO_NAMELOOKUP_TIME_T); + auto const connect = timing_of(easy, CURLINFO_CONNECT_TIME_T); + auto const appconnect = timing_of(easy, CURLINFO_APPCONNECT_TIME_T); - auto const tcp = connect > namelookup ? micros(connect - namelookup) : Duration::zero(); - auto const tls = appconnect > connect ? micros(appconnect - connect) : Duration::zero(); - count_http_connection(static_cast(connections), micros(namelookup), tcp, tls); + auto const tcp = connect > namelookup ? microseconds{connect - namelookup} : Duration::zero(); + auto const tls = appconnect > connect ? microseconds{appconnect - connect} : Duration::zero(); + count_http_connection( + static_cast(connections), microseconds{namelookup}, tcp, tls); } } // namespace detail diff --git a/cpp/src/statistics/counters.cpp b/cpp/src/statistics/counters.cpp index 078ef0166d..aef5ff1350 100644 --- a/cpp/src/statistics/counters.cpp +++ b/cpp/src/statistics/counters.cpp @@ -132,7 +132,10 @@ Counters Counters::since(Counters const& previous) const noexcept bool Counters::empty() const noexcept { - return remote_size_probes == 0 && http_connections == 0 && http_retries == 0; + return remote_size_probes == 0 && remote_size_probing == Duration::zero() && + http_connections == 0 && http_dns == Duration::zero() && http_tcp == Duration::zero() && + http_tls == Duration::zero() && http_retries == 0 && + http_retry_backoff == Duration::zero(); } std::string Counters::to_json() const