support multiple decompressions from a single codec instance

Author: Grok Compression

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

@@ -163,7 +163,11 @@ bool CodeStreamDecompress::decompress(grk_plugin_tile* tile)
     for(auto tileIndex : slated)
     {
       auto cacheEntry = tileCache_->get(tileIndex);
-      if(cacheEntry && cacheEntry->processor->getImage() && !cacheEntry->dirty_)
+      if(!cacheEntry)
+        continue;
+      auto proc = cacheEntry->processor;
+      if(proc->isBestEffortDecompressed() ||
+         (proc->getImage() && (!cacheEntry->dirty_ || proc->allSOTMarkersParsed())))
         tileCompletion_->complete(tileIndex);
     }
   }
@@ -183,7 +187,10 @@ bool CodeStreamDecompress::decompressImpl(std::set<uint16_t> slated)
   // Filter out fully cached tiles from slated
   std::erase_if(slated, [this](uint16_t index) {
     auto cacheEntry = tileCache_->get(index);
-    return cacheEntry && cacheEntry->processor->getImage() && !cacheEntry->dirty_;
+    if(!cacheEntry)
+      return false;
+    auto proc = cacheEntry->processor;
+    return proc->isBestEffortDecompressed() || (proc->getImage() && !cacheEntry->dirty_);
   });
   if(slated.empty())
     return true;
@@ -385,6 +392,7 @@ void CodeStreamDecompress::decompressSequentialPrepare(void)
 {
   stream_->seek(markerCache_->getTileStreamStart() + MARKER_BYTES);
   markerParser_.setSOT();
+  tileCache_->resetSOTParsing();
   if(cp_.plmMarkers_)
     cp_.plmMarkers_->rewind();
   stream_->memAdvise(stream_->tell(), 0, GrkAccessPattern::ACCESS_RANDOM);
@@ -400,6 +408,11 @@ void CodeStreamDecompress::decompressSequential(void)
     {
       if(!sequentialParseAndSchedule(true))
       {
+        // If we've exhausted all slated tiles' markers or the codestream
+        // ran out of data (truncated image), stop gracefully.
+        if(tileCache_->allSlatedSOTMarkersParsed(tilesToDecompress_.getSlatedTiles()) ||
+           markerParser_.endOfCodeStream() || stream_->numBytesLeft() == 0)
+          break;
         success_ = false;
         break;
       }
@@ -443,6 +456,19 @@ void CodeStreamDecompress::decompressSequential(void)
     if(tileCache_->allSlatedSOTMarkersParsed(tilesToDecompress_.getSlatedTiles()))
       break;
   }
+
+  // Mark tiles that were never parsed (missing from truncated codestream)
+  // as complete so wait() doesn't hang.  Tiles that WERE parsed are still
+  // being decompressed asynchronously and must not be completed here.
+  if(tileCompletion_)
+  {
+    for(auto tileIndex : tilesToDecompress_.getSlatedTiles())
+    {
+      auto cached = tileCache_->get(tileIndex);
+      if(!cached || !cached->processor || !cached->processor->allSOTMarkersParsed())
+        tileCompletion_->complete(tileIndex);
+    }
+  }
 }
 
 bool CodeStreamDecompress::sequentialParseAndSchedule(bool multiTile)
@@ -468,7 +494,7 @@ bool CodeStreamDecompress::sequentialParseAndSchedule(bool multiTile)
       if(!processed)
         return false;
     }
-    catch(const CorruptSOTMarkerException& csme)
+    catch([[maybe_unused]] const CorruptSOTMarkerException& csme)
     {
       return false;
     }
@@ -832,6 +858,7 @@ std::function<void()> CodeStreamDecompress::postMultiTile(ITileProcessor* tilePr
       return;
     }
     tileProcessor->post_decompressT2T1(scratchImage_.get());
+    tileProcessor->setBestEffortDecompressed();
     numTilesDecompressed_++;
     auto tileImage = tileProcessor->getImage();
     if(!cp_.codingParams_.dec_.skipAllocateComposite_ && scratchImage_->has_multiple_tiles &&

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

@@ -188,6 +188,8 @@ void CodeStreamDecompress::postReadHeader(void)
   // set up tile completion based on tile and image bounds
   if(headerRead_)
   {
+    setDecompressRegion(RectD(cp_.dw_x0, cp_.dw_y0, cp_.dw_x1, cp_.dw_y1));
+
     if(cp_.asynchronous_ && cp_.simulate_synchronous_)
     {
       auto bounds = headerImage_->getBounds();
@@ -198,8 +200,6 @@ void CodeStreamDecompress::postReadHeader(void)
           },
           [this]() { scheduleTileBatch(); }, tilesToDecompress_.getSlatedTileRect());
     }
-
-    setDecompressRegion(RectD(cp_.dw_x0, cp_.dw_y0, cp_.dw_x1, cp_.dw_y1));
   }
 }
 
@@ -370,6 +370,15 @@ bool CodeStreamDecompress::readSOT(uint8_t* headerData, uint16_t headerSize)
                : true;
   }
 
+  auto cached = tileCache_->get(tileIndex);
+  if(cached && cached->processor && cached->processor->isBestEffortDecompressed())
+  {
+    currTileIndex_ = -1;
+    return currTilePartInfo_.tilePartLength_
+               ? stream_->skip((int64_t)(currTilePartInfo_.tilePartLength_ - sotMarkerSegmentLen))
+               : true;
+  }
+
   auto processor = getTileProcessor(tileIndex);
   if(!processor->readSOT(this->stream_, headerData, headerSize, currTilePartInfo_, false))
     return false;

src/lib/core/codestream/markers/MarkerParser.cpp

@@ -89,6 +89,7 @@ uint16_t MarkerParser::currId(void)
 void MarkerParser::setSOT(void)
 {
   currMarkerId_ = SOT;
+  foundEOC_ = false;
 }
 
 bool MarkerParser::readSOTorEOC(void)

src/lib/core/tile_processor/ITileProcessor.h

@@ -310,6 +310,11 @@ struct ITileProcessor
    */
   virtual bool allSOTMarkersParsed() = 0;
 
+  /**
+   * @brief Reset SOT parsing state so the tile can be re-parsed from the codestream.
+   */
+  virtual void resetSOTParsing() = 0;
+
   /**
    * @brief Sets the processor to truncated state if not all tile parts are parsed
    */
@@ -326,6 +331,13 @@ struct ITileProcessor
    * @return true if initialized, false otherwise
    */
   virtual bool isInitialized(void) = 0;
+
+  /**
+   * @brief Check if tile was decompressed on a best-effort basis (may have been truncated or
+   * errored). Such tiles should not be re-decompressed on codec reuse.
+   */
+  virtual bool isBestEffortDecompressed(void) = 0;
+  virtual void setBestEffortDecompressed(void) = 0;
 };
 
 } // namespace grk

src/lib/core/tile_processor/TileCache.h

@@ -128,18 +128,12 @@ class TileCache
     if(!cache_[tile_index])
     {
       cache_[tile_index] = new TileCacheEntry(processor);
-      grklog.debug("Added ITileProcessor at tile index %u, address %p", tile_index, processor);
     }
     else
     {
       if(cache_[tile_index]->processor)
-      {
-        grklog.debug("Removing ITileProcessor at tile index %u, address %p", tile_index,
-                     cache_[tile_index]->processor);
-        delete cache_[tile_index]->processor; // Delete old processor
-      }
+        delete cache_[tile_index]->processor;
       cache_[tile_index]->processor = processor;
-      grklog.debug("Added ITileProcessor at tile index %u, address %p", tile_index, processor);
     }
     return cache_[tile_index];
   }
@@ -153,12 +147,18 @@ class TileCache
 
   void release(uint16_t tileIndex)
   {
-    // Release the tile processor if it exists
     if(cache_[tileIndex] && cache_[tileIndex]->processor)
-    {
-      grklog.debug("Releasing ITileProcessor at tile index %u, address %p", tileIndex,
-                   cache_[tileIndex]->processor);
       cache_[tileIndex]->processor->release(GRK_TILE_CACHE_NONE);
+  }
+
+  void resetSOTParsing()
+  {
+    for(auto* entry : cache_)
+    {
+      if(!entry || !entry->processor)
+        continue;
+      if(!entry->processor->isBestEffortDecompressed())
+        entry->processor->resetSOTParsing();
     }
   }
 

src/lib/core/tile_processor/TileProcessor.cpp

@@ -245,6 +245,22 @@ bool TileProcessor::isInitialized(void)
   return initialized_;
 }
 
+bool TileProcessor::isBestEffortDecompressed(void)
+{
+  return bestEffortDecompressed_;
+}
+
+void TileProcessor::setBestEffortDecompressed(void)
+{
+  bestEffortDecompressed_ = true;
+}
+
+void TileProcessor::resetSOTParsing()
+{
+  numSOTsParsed_ = 0;
+  tcp_->tilePartCounter_ = 0;
+}
+
 bool TileProcessor::init(void)
 {
   if(!tile_)

src/lib/core/tile_processor/TileProcessor.h

@@ -336,6 +336,10 @@ struct TileProcessor : virtual public ITileProcessor
 
   bool isInitialized(void) override;
 
+  bool isBestEffortDecompressed(void) override;
+  void setBestEffortDecompressed(void) override;
+  void resetSOTParsing() override;
+
 protected:
   /**
    * @brief header @ref GrkImage
@@ -467,6 +471,12 @@ struct TileProcessor : virtual public ITileProcessor
    */
   bool truncated_ = false;
 
+  /**
+   * @brief true if tile was decompressed on a best-effort basis
+   * (may have been truncated or errored). Not re-decompressed on codec reuse.
+   */
+  bool bestEffortDecompressed_ = false;
+
   /**
    * @brief @ref GrkImage for this tile
    *

tests/CMakeLists.txt

@@ -32,6 +32,9 @@ target_link_libraries(compare_dump_files ${GROK_CORE_NAME} spdlog::spdlog)
 add_executable(j2k_random_tile_access j2k_random_tile_access.cpp GrkRandomTileAccess.cpp)
 target_link_libraries(j2k_random_tile_access ${GROK_CORE_NAME} spdlog::spdlog)
 
+add_executable(j2k_multi_region_decompress j2k_multi_region_decompress.cpp GrkMultiRegionDecompress.cpp)
+target_link_libraries(j2k_multi_region_decompress ${GROK_CORE_NAME} spdlog::spdlog)
+
 add_executable(compare_raw_files compare_raw_files.cpp GrkCompareRawFiles.cpp)
 target_link_libraries(compare_raw_files ${GROK_CORE_NAME} spdlog::spdlog)