Remote I/O egress controls and retryable transport failures

cpp/src/detail/http_retry.cpp

@@ -30,9 +30,21 @@ std::size_t HttpRetryPolicy::max_attempts() const noexcept { return _max_attempt
 
 bool HttpRetryPolicy::is_retryable(CURLcode curl_code, long http_code) const
 {
-  // TODO: Currently the timeout is the only libcurl transport error treated as retryable. Need to
-  // revisit and add more candidates.
-  if (curl_code == CURLE_OPERATION_TIMEDOUT) { return true; }
+  // Transport-level failures that say nothing about the request itself, only that this particular
+  // connection or name lookup did not survive. Reading a multi-TB dataset from S3 at several hundred
+  // Gbit/s makes all of these routine: the peer recycles connections, and the VPC resolver drops
+  // queries once a host exceeds its per-ENI packet allowance. Treating them as fatal fails an entire
+  // table scan for one lost packet, so they are retried with the same backoff as a timeout.
+  switch (curl_code) {
+    case CURLE_OPERATION_TIMEDOUT:
+    case CURLE_COULDNT_RESOLVE_HOST:
+    case CURLE_COULDNT_CONNECT:
+    case CURLE_RECV_ERROR:
+    case CURLE_SEND_ERROR:
+    case CURLE_PARTIAL_FILE:
+    case CURLE_GOT_NOTHING: return true;
+    default: break;
+  }
   return std::find(_retryable_status_codes.begin(),
                    _retryable_status_codes.end(),
                    static_cast<int>(http_code)) != _retryable_status_codes.end();
@@ -68,27 +80,28 @@ RetryOutcome HttpRetryPolicy::evaluate(CURLcode curl_code,
     return {RetryDecision::FATAL, std::chrono::milliseconds{0}, ss.str()};
   }
 
+  // A transport failure carries no HTTP status, so reporting one would be misleading.
+  auto const reason = [&] {
+    std::stringstream rs;
+    if (curl_code == CURLE_HTTP_RETURNED_ERROR) {
+      rs << "Got HTTP code " << http_code << ".";
+    } else {
+      rs << "Transport error: " << curl_easy_strerror(curl_code) << ".";
+    }
+    return rs.str();
+  }();
+
   if (attempt >= _max_attempts) {
     std::stringstream ss;
     ss << "KvikIO: HTTP request reached maximum number of attempts (" << _max_attempts
-       << "). Reason: ";
-    if (curl_code == CURLE_OPERATION_TIMEDOUT) {
-      ss << "Operation timed out.";
-    } else {
-      ss << "Got HTTP code " << http_code << ".";
-    }
+       << "). Reason: " << reason;
     return {RetryDecision::EXHAUSTED, std::chrono::milliseconds{0}, ss.str()};
   }
 
   auto const delay_ms = backoff_for(attempt);
   std::stringstream ss;
-  if (curl_code == CURLE_OPERATION_TIMEDOUT) {
-    ss << "KvikIO: Timeout error. Retrying after " << delay_ms.count() << "ms (attempt " << attempt
-       << " of " << _max_attempts << ").";
-  } else {
-    ss << "KvikIO: Got HTTP code " << http_code << ". Retrying after " << delay_ms.count()
-       << "ms (attempt " << attempt << " of " << _max_attempts << ").";
-  }
+  ss << "KvikIO: " << reason << " Retrying after " << delay_ms.count() << "ms (attempt " << attempt
+     << " of " << _max_attempts << ").";
   return {RetryDecision::RETRY, delay_ms, ss.str()};
 }
 

cpp/src/detail/remote_callback.cpp

@@ -3,11 +3,15 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include <algorithm>
 #include <cstddef>
+#include <cstdint>
+#include <cstdlib>
 #include <cstring>
 #include <string>
 
 #include <curl/curl.h>
+#include <immintrin.h>
 
 #include <kvikio/detail/nvtx.hpp>
 #include <kvikio/detail/remote_callback.hpp>
@@ -20,6 +24,75 @@ void CallbackContext::reset_for_retry() noexcept
   overflow_error = false;
 }
 
+// Benchmarking escape hatch: accept and account for every byte without
+// delivering it. The point is to separate the cost of moving bytes off the
+// network from the cost of landing them in a caller buffer, which at high
+// concurrency is a stream of writes to memory far larger than L3 and so is not
+// free. Byte accounting is unchanged, so a throughput figure measured with this
+// on is still an honest count of bytes received -- but the destination holds
+// garbage, which is why it is off unless explicitly asked for.
+//
+// Read once: this callback runs per libcurl buffer, millions of times a second.
+static bool const discard_payload = [] {
+  auto const* env = std::getenv("KVIKIO_REMOTE_IO_DISCARD");
+  return env != nullptr && env[0] == '1';
+}();
+
+// Non-temporal copy for the receive path.
+//
+// libcurl delivers a buffer at a time, 16 KiB by default, so every memcpy here
+// is far below glibc's non-temporal threshold and uses ordinary stores. Ordinary
+// stores to a destination much larger than last-level cache fetch each line
+// first (read-for-ownership), so a destination byte costs two DRAM accesses
+// rather than one, and at hundreds of Gbps the receive path becomes
+// memory-bandwidth-bound rather than CPU-bound. Streaming stores skip the fetch.
+//
+// Only worthwhile when the destination is not read again soon, which is exactly
+// the case for a buffer that is about to be handed to a GPU or a consumer
+// thread. Off by default because it is the wrong choice for a small destination
+// that stays in cache.
+static bool const nt_copy = [] {
+  auto const* env = std::getenv("KVIKIO_REMOTE_IO_NT_COPY");
+  return env != nullptr && env[0] == '1';
+}();
+
+namespace {
+
+#if defined(__AVX512F__) || defined(__AVX2__)
+void copy_nontemporal(char* dst, char const* src, std::size_t nbytes)
+{
+  // Streaming stores need a 32-byte aligned destination, so copy the leading
+  // partial line normally and resume streaming once aligned.
+  constexpr std::size_t kVec = 32;
+  auto const misaligned      = reinterpret_cast<std::uintptr_t>(dst) % kVec;
+  if (misaligned != 0) {
+    auto const head = std::min(nbytes, kVec - misaligned);
+    std::memcpy(dst, src, head);
+    dst += head;
+    src += head;
+    nbytes -= head;
+  }
+  while (nbytes >= kVec) {
+    _mm256_stream_si256(reinterpret_cast<__m256i*>(dst),
+                        _mm256_loadu_si256(reinterpret_cast<__m256i const*>(src)));
+    dst += kVec;
+    src += kVec;
+    nbytes -= kVec;
+  }
+  if (nbytes != 0) { std::memcpy(dst, src, nbytes); }
+  // Streaming stores are weakly ordered with respect to everything else, so the
+  // writes must be fenced before the buffer is handed on.
+  _mm_sfence();
+}
+#else
+void copy_nontemporal(char* dst, char const* src, std::size_t nbytes)
+{
+  std::memcpy(dst, src, nbytes);
+}
+#endif
+
+}  // namespace
+
 std::size_t callback_host_memory(char* data, std::size_t size, std::size_t nmemb, void* context)
 {
   KVIKIO_NVTX_FUNC_RANGE();
@@ -30,7 +103,13 @@ std::size_t callback_host_memory(char* data, std::size_t size, std::size_t nmemb
     return CURL_WRITEFUNC_ERROR;
   }
   KVIKIO_NVTX_FUNC_RANGE(nbytes);
-  std::memcpy(ctx->buf + ctx->offset, data, nbytes);
+  if (!discard_payload) {
+    if (nt_copy) {
+      copy_nontemporal(ctx->buf + ctx->offset, data, nbytes);
+    } else {
+      std::memcpy(ctx->buf + ctx->offset, data, nbytes);
+    }
+  }
   ctx->offset += nbytes;
   return nbytes;
 }

cpp/src/shim/libcurl.cpp

@@ -3,9 +3,11 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include <algorithm>
 #include <chrono>
 #include <cstddef>
 #include <cstdint>
+#include <cstdlib>
 #include <cstring>
 #include <functional>
 #include <memory>
@@ -15,6 +17,7 @@
 #include <vector>
 
 #include <curl/curl.h>
+#include <openssl/ssl.h>
 
 #include <kvikio/defaults.hpp>
 #include <kvikio/detail/http_retry.hpp>
@@ -110,6 +113,65 @@ CurlHandle::CurlHandle(LibCurl::UniqueHandlePtr handle,
   auto const verbose = getenv_or("KVIKIO_REMOTE_VERBOSE", false);
   if (verbose) { setopt(CURLOPT_VERBOSE, 1L); }
 
+  // Receive buffer size, left at libcurl's CURL_MAX_WRITE_SIZE (16 KiB) default.
+  //
+  // Raising this looks obviously right and measures worse. At 16 KiB a 400G-class
+  // NIC costs millions of recv() calls per second, and the syscall entry plus
+  // hardened-usercopy check are visible in the profile. But libcurl allocates one
+  // of these buffers per easy handle, and there is one handle per in-flight
+  // sub-range, so the size multiplies by the concurrency: 4096 transfers hold
+  // 64 MiB at 16 KiB and 4 GiB at 1 MiB. The smaller total stays inside L3 and
+  // wins by more than the saved syscalls are worth. Measured on a g7e.48xlarge
+  // across four cards at 4096-way concurrency: 862 Gbps at 16 KiB, 745 at
+  // 256 KiB, 727 at 1 MiB.
+  //
+  // Exposed anyway because the trade-off inverts at low concurrency, where a few
+  // fat transfers do prefer a large buffer. Zero leaves libcurl's default.
+  //
+  // These three options are read once rather than per handle, because a handle is
+  // constructed per sub-range transfer and getenv() would then sit on the hot
+  // path.
+  static long const buffer_size = [] {
+    auto const requested = getenv_or("KVIKIO_REMOTE_IO_BUFFER_SIZE", ssize_t{0});
+    if (requested <= 0) { return 0L; }
+    return static_cast<long>(std::clamp(requested, ssize_t{1024}, ssize_t{CURL_MAX_READ_SIZE}));
+  }();
+  if (buffer_size > 0) { setopt(CURLOPT_BUFFERSIZE, buffer_size); }
+
+  // Bind every connection to one interface. The `if!<name>` form makes libcurl
+  // use SO_BINDTODEVICE, which pins egress to a chosen NIC on a host with several
+  // cards on one subnet. Without it the kernel picks by route metric and every
+  // connection leaves through the same card.
+  static std::string const interface_opt = [] {
+    auto const* env = std::getenv("KVIKIO_REMOTE_IO_INTERFACE");
+    return std::string{env == nullptr ? "" : env};
+  }();
+  if (!interface_opt.empty()) { setopt(CURLOPT_INTERFACE, interface_opt.c_str()); }
+
+  // Spread connections over every address the resolver returns. S3 publishes many
+  // front-end addresses, but libcurl connects to the first one that answers, so
+  // without shuffling a whole process piles onto a single endpoint.
+  static bool const shuffle_dns = getenv_or("KVIKIO_REMOTE_IO_DNS_SHUFFLE", false);
+  if (shuffle_dns) { setopt(CURLOPT_DNS_SHUFFLE_ADDRESSES, 1L); }
+
+  // Kernel TLS receive. Userspace TLS costs two touches of every byte: the
+  // kernel copies ciphertext out to OpenSSL, then OpenSSL decrypts. At
+  // multi-hundred-Gbps the copy alone is the largest single entry in the
+  // profile. kTLS moves the record layer into the kernel so the payload is
+  // decrypted on the way out to the caller.
+  //
+  // Off by default: it needs the `tls` module live, silently falls back to
+  // userspace when the negotiated cipher is unsupported, and changes what is
+  // being measured. Confirm it actually engaged by watching TlsCurrRxSw in
+  // /proc/net/tls_stat rather than trusting the flag.
+  static bool const enable_ktls = getenv_or("KVIKIO_REMOTE_IO_KTLS", false);
+  if (enable_ktls) {
+    setopt(CURLOPT_SSL_CTX_FUNCTION, +[](CURL*, void* ssl_ctx, void*) -> CURLcode {
+      SSL_CTX_set_options(static_cast<SSL_CTX*>(ssl_ctx), SSL_OP_ENABLE_KTLS);
+      return CURLE_OK;
+    });
+  }
+
   detail::set_up_ca_paths(*this);
 }