fetch: HTTP/2 multiplexing, adaptive Phase 1, configurable batch size, request sorting

- Enable HTTP/2 multiplexing (CURLPIPE_MULTIPLEX + HTTP_VERSION_2TLS)
  so all concurrent range requests share a single TCP connection
- Adaptive Phase 1 sizing for contiguous progressions (RLCP, RPCL,
  single-layer LRCP): estimate needed data from tile size and reduce
  level so reusePhase1Data() can skip the Phase 2 round-trip
- Add fetch_batch_size to grk_stream_params / IFetcher::setBatchSize()
  allowing callers to tune HTTP concurrency (default remains 30)
- Sort flat fetch request list by file offset after genCollections()
  for better server-side read coalescing with HTTP/2 streams

Author: Grok Compression

src/lib/core/codestream/decompress/CodeStreamDecompress.cpp

@@ -16,6 +16,7 @@
  */
 
 #include <chrono>
+#include <cmath>
 #include <functional>
 
 #include "TFSingleton.h"
@@ -2066,12 +2067,36 @@ bool CodeStreamDecompress::fetchByTileSelective(
   const auto& allTileParts = cp_.tlmMarkers_->getTileParts();
   uint8_t reduce = cp_.codingParams_.dec_.reduce_;
 
-  // Phase 1: Fetch 4KB header of each tile-part
+  // Determine if the progression order produces contiguous data from the start.
+  // For these cases, Phase 1 can speculatively fetch the estimated needed amount
+  // so that reusePhase1Data() can skip Phase 2 entirely.
+  auto tcp = defaultTcp_.get();
+  bool contiguousProgression = false;
+  switch(tcp->prg_)
+  {
+    case GRK_RLCP:
+    case GRK_RPCL:
+      contiguousProgression = true;
+      break;
+    case GRK_LRCP:
+      contiguousProgression = (tcp->numLayers_ == 1);
+      break;
+    default:
+      break;
+  }
+
+  // Phase 1: fetch tile-part headers (and possibly all needed data for contiguous cases)
   constexpr uint32_t headerFetchSize = 4096;
-  grklog.debug("fetchByTileSelective: Phase 1 starting, %zu tiles, reduce=%u", slated.size(),
-               (unsigned)reduce);
+  grklog.debug("fetchByTileSelective: Phase 1 starting, %zu tiles, reduce=%u, contiguous=%d",
+               slated.size(), (unsigned)reduce, (int)contiguousProgression);
+
+  // Create modified tile-part info for Phase 1 fetch.
+  // For contiguous progressions, estimate the needed data size and fetch that
+  // directly, eliminating the need for a second round-trip in most cases.
+  double pixelRatio = (contiguousProgression && reduce > 0)
+                          ? (1.0 / std::pow(4.0, reduce)) * 2.0
+                          : 0.0;
 
-  // Create modified tile-part info with 4KB lengths for header fetch
   TPSEQ_VEC headerTileParts(allTileParts.size());
   for(auto tileIndex : slated)
   {
@@ -2082,7 +2107,17 @@ bool CodeStreamDecompress::fetchByTileSelective(
     for(size_t i = 0; i < srcParts->size(); ++i)
     {
       auto& part = (*srcParts)[i];
-      uint32_t fetchLen = std::min<uint32_t>((uint32_t)part->length_, headerFetchSize);
+      uint32_t fetchLen;
+      if(contiguousProgression && reduce > 0)
+      {
+        uint64_t estimated = std::max<uint64_t>(
+            headerFetchSize, (uint64_t)(part->length_ * pixelRatio));
+        fetchLen = (uint32_t)std::min<uint64_t>(estimated, part->length_);
+      }
+      else
+      {
+        fetchLen = std::min<uint32_t>((uint32_t)part->length_, headerFetchSize);
+      }
       headerTileParts[tileIndex]->push_back((uint8_t)i, (uint8_t)srcParts->size(), part->offset_,
                                             fetchLen);
     }

src/lib/core/grok.h

@@ -625,6 +625,9 @@ typedef struct _grk_stream_params
   uint32_t max_retry; /* maximum number of retries */
   uint32_t retry_delay; /* delay between retries in seconds */
 
+  /* 13 Fetch concurrency (0 = use default of 30) */
+  uint32_t fetch_batch_size; /* max concurrent HTTP range requests */
+
 } grk_stream_params;
 
 /**

src/lib/core/stream/StreamGenerator.cpp

@@ -160,6 +160,7 @@ IStream* StreamGenerator::createCurlFetchStream(void)
   auth.connect_timeout_ = streamParams_.connect_timeout;
   auth.max_retry_ = streamParams_.max_retry;
   auth.retry_delay_ = streamParams_.retry_delay;
+  auth.fetch_batch_size_ = streamParams_.fetch_batch_size;
   grklog.debug("StreamGenerator: s3_allow_insecure: streamParams=%d, auth=%d",
                (int)streamParams_.s3_allow_insecure, (int)auth.s3_allow_insecure_);
   std::string_view file{streamParams_.file};

src/lib/core/stream/TPFetchSeq.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <algorithm>
 #include <cassert>
 #include <cstring>
 #include <vector>
@@ -114,6 +115,12 @@ struct SharedPtrSeq
     objStore_.clear();
   }
 
+  template<typename Compare>
+  void sort(Compare comp)
+  {
+    std::sort(objStore_.begin(), objStore_.end(), comp);
+  }
+
 private:
   std::vector<std::shared_ptr<T>> objStore_;
 };
@@ -242,6 +249,13 @@ struct TPFetch : public DataSlice
 
 struct TPFetchSeq : SharedPtrSeq<TPFetch>
 {
+  void sortByOffset()
+  {
+    sort([](const std::shared_ptr<TPFetch>& a, const std::shared_ptr<TPFetch>& b) {
+      return a->offset_ < b->offset_;
+    });
+  }
+
   void push_back(uint16_t tileIndex, const std::unique_ptr<TPSeq>& tileParts)
   {
     if(!tileParts)
@@ -289,6 +303,11 @@ struct TPFetchSeq : SharedPtrSeq<TPFetch>
         }
       }
     }
+
+    // Sort requests by file offset so that batched HTTP requests target
+    // sequential file regions, improving server-side read coalescing
+    // and CDN cache efficiency.
+    tilePartFetchFlat->sortByOffset();
   }
 
   /**

src/lib/core/stream/fetchers/CurlFetcher.cpp

@@ -76,6 +76,7 @@ CurlFetcher::CurlFetcher(void) : tileWriteCallback_(tileWriteCallback)
   if(!multi_handle_)
     throw std::runtime_error("Failed to initialize CURL multi handle");
   curl_multi_setopt(multi_handle_, CURLMOPT_MAX_TOTAL_CONNECTIONS, 100L);
+  curl_multi_setopt(multi_handle_, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
   fetchThread_ = std::thread(&CurlFetcher::fetchWorker, this);
 }
 
@@ -105,13 +106,20 @@ void CurlFetcher::notifyThrottleRelease()
   throttleCV_.notify_one();
 }
 
+void CurlFetcher::setBatchSize(size_t batchSize)
+{
+  batchSize_ = std::max<size_t>(1, batchSize);
+}
+
 void CurlFetcher::init(const std::string& path, const FetchAuth& auth)
 {
   auth_ = auth;
   if(auth_.max_retry_ > 0)
     maxRetries_ = auth_.max_retry_;
   if(auth_.retry_delay_ > 0)
     retryDelayMs_ = auth_.retry_delay_ * 1000;
+  if(auth_.fetch_batch_size_ > 0)
+    batchSize_ = auth_.fetch_batch_size_;
   parse(path);
   fetch_total_size();
 }
@@ -561,6 +569,8 @@ CURL* CurlFetcher::configureHandle(uint64_t offset, uint64_t end, FetchResult& r
   curl_easy_setopt(curl, CURLOPT_URL, url_.c_str());
   curl_initiate_retry(curl);
   curl_easy_setopt(curl, CURLOPT_VERBOSE, 0L);
+  curl_easy_setopt(curl, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
+  curl_easy_setopt(curl, CURLOPT_PIPEWAIT, 1L);
   curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, callback);
   curl_easy_setopt(curl, CURLOPT_WRITEDATA, &result);
   curl_easy_setopt(curl, CURLOPT_PRIVATE, &result);

src/lib/core/stream/fetchers/CurlFetcher.h

@@ -94,6 +94,8 @@ class IFetcher
   // fetcher so it can re-check the condition.
   virtual void setFetchThrottle(std::function<bool()> throttle) = 0;
   virtual void notifyThrottleRelease() = 0;
+
+  virtual void setBatchSize(size_t batchSize) = 0;
 };
 
 struct TileFetchContext : public std::enable_shared_from_this<TileFetchContext>
@@ -131,6 +133,7 @@ class CurlFetcher : public IFetcher
 
   void setFetchThrottle(std::function<bool()> throttle) override;
   void notifyThrottleRelease() override;
+  void setBatchSize(size_t batchSize) override;
   void init(const std::string& path, const FetchAuth& auth) override;
   size_t read(uint8_t* buffer, size_t numBytes) override;
   bool seek(uint64_t offset) override;

src/lib/core/stream/fetchers/FetchCommon.h

@@ -79,6 +79,9 @@ struct FetchAuth
   uint32_t max_retry_ = 0;
   uint32_t retry_delay_ = 0;
 
+  // Fetch concurrency (0 = use default of 30)
+  uint32_t fetch_batch_size_ = 0;
+
   FetchAuth() = default;
   FetchAuth(const std::string& u, const std::string& p, const std::string& t,
             const std::vector<std::string>& h = {}, const std::string& r = "",