fetch refactor II

Author: Grok Compression

src/lib/core/CMakeLists.txt

@@ -91,6 +91,7 @@ set(GROK_LIBRARY_SRCS
   ${CMAKE_CURRENT_SOURCE_DIR}/stream/MappedFile.cpp
   ${CMAKE_CURRENT_SOURCE_DIR}/stream/StreamIO.cpp
   ${CMAKE_CURRENT_SOURCE_DIR}/stream/StreamGenerator.cpp
+  ${CMAKE_CURRENT_SOURCE_DIR}/stream/fetchers/CurlFetcher.cpp
 
   ${CMAKE_CURRENT_SOURCE_DIR}/plugin/minpf_dynamic_library.cpp
   ${CMAKE_CURRENT_SOURCE_DIR}/plugin/minpf_plugin_manager.cpp

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

@@ -1682,28 +1682,8 @@ bool CodeStreamDecompress::fetchByTile(
   startDecompressConsumer(std::min((uint16_t)TFSingleton::num_threads(),
                                    (uint16_t)((maxRowsAhead_ + 1) * cp_.t_grid_width_)));
 
-  // Back pressure: prevent the fetcher from scheduling more HTTP requests
-  // when either (a) too many tiles are in-flight for decompression, or
-  // (b) the fetcher is too far ahead of the consumer (swath-based release).
   auto numTileCols = cp_.t_grid_width_;
-  fetcher->setFetchThrottle([this]() {
-    // Row-based: don't fetch tiles more than maxRowsAhead_ rows beyond
-    // the last row released by the consumer.
-    if(tileCompletion_)
-    {
-      int32_t lastCleared = tileCompletion_->getLastClearedTileY();
-      int32_t maxAllowed = lastCleared + maxRowsAhead_ + 2;
-      if(maxFetchedTileRow_.load(std::memory_order_acquire) >= maxAllowed)
-        return false;
-    }
-    // In-flight: don't overwhelm the decompress pipeline
-    uint16_t inFlight;
-    {
-      std::lock_guard<std::mutex> lock(decompressThrottleMutex_);
-      inFlight = decompressInFlight_;
-    }
-    return decompressQueue_->size() + inFlight < maxDecompressInFlight_;
-  });
+  installFetchThrottle(fetcher);
 
   fetchByTileFutures_.push_back(fetcher->fetchTiles(
       cp_.tlmMarkers_->getTileParts(), slated, nullptr,
@@ -2053,6 +2033,28 @@ std::vector<std::pair<uint16_t, std::shared_ptr<TPFetchSeq>>>
   return prefetchedTiles;
 }
 
+void CodeStreamDecompress::installFetchThrottle(IFetcher* fetcher)
+{
+  fetcher->setFetchThrottle([this]() {
+    // Row-based: don't fetch tiles more than maxRowsAhead_ rows beyond
+    // the last row released by the consumer.
+    if(tileCompletion_)
+    {
+      int32_t lastCleared = tileCompletion_->getLastClearedTileY();
+      int32_t maxAllowed = lastCleared + maxRowsAhead_ + 2;
+      if(maxFetchedTileRow_.load(std::memory_order_acquire) >= maxAllowed)
+        return false;
+    }
+    // In-flight: don't overwhelm the decompress pipeline
+    uint16_t inFlight;
+    {
+      std::lock_guard<std::mutex> lock(decompressThrottleMutex_);
+      inFlight = decompressInFlight_;
+    }
+    return decompressQueue_->size() + inFlight < maxDecompressInFlight_;
+  });
+}
+
 bool CodeStreamDecompress::fetchByTileSelective(
     std::set<uint16_t>& slated, Rect32 unreducedImageBounds,
     std::function<std::function<void()>(ITileProcessor*)> postGenerator)
@@ -2092,9 +2094,6 @@ bool CodeStreamDecompress::fetchByTileSelective(
   auto phase1Future = phase1Promise.get_future();
   auto remainingTiles = std::make_shared<std::atomic<size_t>>(slated.size());
 
-  // Save slated tiles before Phase 1 fetch (fetchTiles moves the set)
-  auto slatedTiles = slated;
-
   // Phase 1 fetch: get tile-part headers
   fetcher->fetchTiles(headerTileParts, slated, nullptr,
                       [&headerResults, &headerMutex, &remainingTiles,
@@ -2149,7 +2148,7 @@ bool CodeStreamDecompress::fetchByTileSelective(
   selectiveTileParts_.resize(allTileParts.size());
   auto selectiveFetchTiles = std::make_shared<std::set<uint16_t>>();
 
-  for(auto tileIndex : slatedTiles)
+  for(auto tileIndex : slated)
   {
     auto it = headerResults->find(tileIndex);
     if(it == headerResults->end())
@@ -2196,21 +2195,7 @@ bool CodeStreamDecompress::fetchByTileSelective(
 
   if(!selectiveFetchTiles->empty())
   {
-  fetcher->setFetchThrottle([this]() {
-    if(tileCompletion_)
-    {
-      int32_t lastCleared = tileCompletion_->getLastClearedTileY();
-      int32_t maxAllowed = lastCleared + maxRowsAhead_ + 2;
-      if(maxFetchedTileRow_.load(std::memory_order_acquire) >= maxAllowed)
-        return false;
-    }
-    uint16_t inFlight;
-    {
-      std::lock_guard<std::mutex> lock(decompressThrottleMutex_);
-      inFlight = decompressInFlight_;
-    }
-    return decompressQueue_->size() + inFlight < maxDecompressInFlight_;
-  });
+  installFetchThrottle(fetcher);
 
   fetchByTileFutures_.push_back(fetcher->fetchTiles(
       selectiveTileParts_, *selectiveFetchTiles, nullptr,

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

@@ -552,6 +552,15 @@ class CodeStreamDecompress final : public CodeStream, public IDecompressor
       const std::unordered_map<uint16_t, TileHeaderResult>& headerResults,
       const TPSEQ_VEC& allTileParts);
 
+  /**
+   * @brief Install fetch throttle on the given fetcher.
+   *
+   * Back pressure: prevent the fetcher from scheduling more HTTP requests
+   * when either (a) the fetcher is too far ahead of the consumer (swath-based
+   * release), or (b) too many tiles are in-flight for decompression.
+   */
+  void installFetchThrottle(IFetcher* fetcher);
+
   /**
    * @brief Scratch @ref GrkImage for decompressor
    * This image may composite multiple tiles, if needed.

src/lib/core/stream/TPFetchSeq.h

@@ -17,13 +17,17 @@
 
 #pragma once
 
+#include <cassert>
+#include <cstring>
 #include <vector>
 #include <memory>
 #include <set>
 #include <unordered_map>
 #include <atomic>
 
 #include "FetchCommon.h"
+#include "Logger.h"
+#include "IStream.h"
 
 namespace grk
 {
@@ -56,6 +60,15 @@ struct SharedPtrSeq
     return objStore_.begin();
   }
 
+  /**
+   * @brief Returns const iterator to beginning of sequence
+   * @return Const iterator pointing to first shared_ptr
+   */
+  const_iterator begin() const
+  {
+    return objStore_.begin();
+  }
+
   /**
    * @brief Returns const iterator to end of sequence
    * @return Const iterator pointing past the last shared_ptr
@@ -211,6 +224,12 @@ struct TPFetch : public DataSlice
     {
       data_ = std::make_unique<uint8_t[]>(length_);
     }
+    if(fetchOffset_ + chunkLen > length_)
+    {
+      grklog.error("TPFetch::copy: buffer overflow (offset=%zu + len=%zu > capacity=%zu)",
+                   fetchOffset_, chunkLen, (size_t)length_);
+      return;
+    }
     std::memcpy(data_.get() + fetchOffset_, chunk, chunkLen);
     fetchOffset_ += chunkLen;
   }
@@ -249,7 +268,7 @@ struct TPFetchSeq : SharedPtrSeq<TPFetch>
   }
 
   static void
-      genCollections(const TPSEQ_VEC* allTileParts, std::set<uint16_t>& slated,
+      genCollections(const TPSEQ_VEC* allTileParts, const std::set<uint16_t>& slated,
                      std::shared_ptr<TPFetchSeq>& tilePartFetchFlat,
                      std::shared_ptr<std::unordered_map<uint16_t, std::shared_ptr<TPFetchSeq>>>&
                          tilePartFetchByTile)
@@ -272,6 +291,13 @@ struct TPFetchSeq : SharedPtrSeq<TPFetch>
     }
   }
 
+  /**
+   * @brief Atomically increment and return the fetch-completed count.
+   *
+   * When the returned value equals size(), all tile-parts for this tile
+   * have been fetched. Uses uint8_t storage, limiting to 255 tile-parts
+   * per tile (the JPEG 2000 spec allows up to 255 tile-parts per tile).
+   */
   uint8_t incrementFetchCount()
   {
     std::atomic_ref<uint8_t> atomicCount(fetchCount_);