memory back presssure: fix for grk_decompress decompress to disk

Author: Grok Compression

.vscode/launch.json

@@ -139,15 +139,15 @@
             "name": "SPOT6",
             "type": "cppdbg",
             "request": "launch",
-            //"program": "/usr/bin/time",
-            "program": "${workspaceFolder}/build/bin/core_decompress",
+            "program": "/usr/bin/time",
+            //"program": "${workspaceFolder}/build/bin/grk_decompress",
             "args": [
-                // "-v",
-                // "${workspaceFolder}/build/bin/core_decompress",
+                "-v",
+                "${workspaceFolder}/build/bin/grk_decompress",
                 "-i",
                 "$HOME/temp/IMG_SPOT6_PMS_201305251604372_ORT_816009101_R1C1.JP2",
-                // "-o",
-                // "$HOME/temp/spot6.tif"
+                "-o",
+                "$HOME/temp/spot6.tif"
             ],
             "cwd": "${workspaceFolder}",
             "environment": [

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

@@ -15,6 +15,7 @@
  *
  */
 
+#include <chrono>
 #include <functional>
 
 #include "TFSingleton.h"
@@ -258,7 +259,15 @@ bool CodeStreamDecompress::decompress(grk_plugin_tile* tile)
           uint32_t yBegin = 0;
           uint32_t yEnd = std::min(tileGlobalYEnd, regionY1) - std::max(tileGlobalYBegin, regionY0);
           if(yEnd <= yBegin)
+          {
+            // Empty tile row (e.g. zero pixels after resolution reduction).
+            // Still advance nextBandTileY_ so backpressure unblocks.
+            std::lock_guard<std::mutex> lock(bandOrderMutex_);
+            if(tileY == nextBandTileY_)
+              nextBandTileY_ = tileY + 1;
+            bandDrainCV_.notify_one();
             return;
+          }
 
           uint16_t tileX0 = tileIndexBegin % numTileCols;
           uint16_t numSlatedCols = (uint16_t)tilesToDecompress_.getSlatedTileRect().width();
@@ -288,6 +297,16 @@ bool CodeStreamDecompress::decompress(grk_plugin_tile* tile)
 
             if(!ioBandCallback_(band.yBegin, band.yEnd, scratchImage_.get(), ioBandUserData_))
               success_ = false;
+
+            // Release tile processors for this completed row — compositing is done,
+            // so tile image data is no longer needed.
+            for(uint16_t col = 0; col < band.numCols; col++)
+            {
+              uint16_t tileIndex = nextBandTileY_ * numTileCols + (band.tileX0 + col);
+              tileCache_->releaseForSwath(tileIndex);
+            }
+            MemoryManager::releaseFreedPages();
+
             pendingBands_.erase(nextBandTileY_);
 
             // Advance strip buffer for the next tile row
@@ -310,6 +329,8 @@ bool CodeStreamDecompress::decompress(grk_plugin_tile* tile)
 
             nextBandTileY_ = nextTileY;
           }
+          // Wake the parser thread so it can schedule more tiles
+          bandDrainCV_.notify_one();
         },
         nullptr, tilesToDecompress_.getSlatedTileRect());
   }
@@ -825,6 +846,20 @@ bool CodeStreamDecompress::sequentialParseAndSchedule(bool multiTile)
     return true;
   }
 
+  // Backpressure for strip-based band callback: block the parser thread if
+  // this tile's row is too far ahead of the row currently being drained.
+  // Without this, all tiles decompress into memory before any row is written.
+  // Use wait_for with a timeout so we periodically re-check success_ even if
+  // the Taskflow task graph skips the post callback on error.
+  if(ioBandCallback_ && tileCompletion_)
+  {
+    uint16_t numTileCols = tileCompletion_->getNumTileCols();
+    uint16_t tileY = tileProcessor->getIndex() / numTileCols;
+    std::unique_lock<std::mutex> lock(bandOrderMutex_);
+    while(!(tileY < nextBandTileY_ + 2 || !success_))
+      bandDrainCV_.wait_for(lock, std::chrono::milliseconds(100));
+  }
+
   // schedule
   return schedule(tileProcessor, multiTile);
 }
@@ -1113,6 +1148,10 @@ std::function<void()> CodeStreamDecompress::postMultiTile(ITileProcessor* tilePr
     if(!success_)
     {
       releaseThrottle();
+      // Always mark tile as complete so row callbacks can fire and
+      // backpressure unblocks, even when the decompress failed.
+      if(tileCompletion_)
+        tileCompletion_->complete(tileProcessor->getIndex());
       return;
     }
     tileProcessor->post_decompressT2T1(scratchImage_.get());

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

@@ -412,6 +412,7 @@ class CodeStreamDecompress final : public CodeStream, public IDecompressor
 
   // band ordering for incremental writes
   std::mutex bandOrderMutex_;
+  std::condition_variable bandDrainCV_;
   uint16_t nextBandTileY_ = 0;
   struct PendingBand_
   {

src/lib/core/util/GrkImage.cpp

@@ -444,8 +444,11 @@ bool GrkImage::isOpacity(uint16_t compno) const
 }
 bool GrkImage::isPostProcessNoOp(void) const
 {
-  // applyColour: palette or channel_definition requires processing
-  if(meta && (meta->color.palette || meta->color.channel_definition))
+  // applyColour: palette requires processing; channel_definition only blocks
+  // if it needs component data swaps (type-only metadata is applied early)
+  if(meta && meta->color.palette)
+    return false;
+  if(needsChannelDefinitionSwap())
     return false;
   // applyColourManagement: ICC profile that must be applied
   if(meta && meta->color.icc_profile_buf)
@@ -504,6 +507,11 @@ bool GrkImage::isPostProcessNoOp(void) const
 }
 void GrkImage::postReadHeader(CodingParams* cp)
 {
+  // Apply channel definition types (e.g. alpha) early so format writers
+  // can emit correct metadata (TIFF EXTRASAMPLES, etc.) before strip-based
+  // decompress starts.  Component data swaps are deferred to postProcess.
+  applyChannelDefinitionTypes();
+
   uint8_t prec = comps[0].prec;
   if(precision)
     prec = precision->prec;
@@ -596,6 +604,44 @@ void GrkImage::allocPalette(uint8_t num_channels, uint16_t num_entries)
   ((GrkImageMeta*)meta)->allocPalette(num_channels, num_entries);
 }
 
+bool GrkImage::needsChannelDefinitionSwap() const
+{
+  if(!meta || !meta->color.channel_definition)
+    return false;
+  auto info = meta->color.channel_definition->descriptions;
+  uint16_t n = meta->color.channel_definition->num_channel_descriptions;
+  for(uint16_t i = 0; i < n; ++i)
+  {
+    if(info[i].typ != GRK_CHANNEL_TYPE_COLOUR)
+      continue;
+    if(info[i].asoc == GRK_CHANNEL_ASSOC_WHOLE_IMAGE)
+      continue;
+    uint16_t channel = info[i].channel;
+    uint16_t asoc = info[i].asoc;
+    if(channel >= numcomps || asoc > numcomps)
+      continue;
+    uint16_t asoc_index = (uint16_t)(asoc - 1);
+    if(channel != asoc_index)
+      return true;
+  }
+  return false;
+}
+
+void GrkImage::applyChannelDefinitionTypes()
+{
+  if(!meta || !meta->color.channel_definition)
+    return;
+  auto info = meta->color.channel_definition->descriptions;
+  uint16_t n = meta->color.channel_definition->num_channel_descriptions;
+  for(uint16_t i = 0; i < n; ++i)
+  {
+    uint16_t channel = info[i].channel;
+    if(channel >= numcomps)
+      continue;
+    comps[channel].type = (GRK_CHANNEL_TYPE)info[i].typ;
+  }
+}
+
 void GrkImage::apply_channel_definition()
 {
   if(channel_definition_applied)

src/lib/core/util/GrkImage.h

@@ -123,6 +123,8 @@ class GrkImage : public grk_image
 
   bool check_color(uint16_t signalledNumComps);
   void apply_channel_definition(void);
+  void applyChannelDefinitionTypes(void);
+  bool needsChannelDefinitionSwap(void) const;
   void allocPalette(uint8_t num_channels, uint16_t num_entries);
   uint32_t width(void) const;
   uint32_t height(void) const;