CodeStreamDecompress.h

Go to the documentation of this file.

/*
 *    Copyright (C) 2016-2026 Grok Image Compression Inc.
 *
 *    This source code is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This source code is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 */
 
#pragma once
 
#include "ConcurrentQueue.h"
#include "TFSingleton.h"
#include "CompressedChunkCache.h"
#include <map>
 
namespace grk
{

class CodeStreamDecompress final : public CodeStream, public IDecompressor
{
public:
  explicit CodeStreamDecompress(IStream* stream);

  ~CodeStreamDecompress()
  {
    if(decompressQueue_)
      decompressQueue_->close();
    if(decompressConsumer_.joinable())
      decompressConsumer_.join();
    if(decompressWorker_.joinable())
      decompressWorker_.join();
    TFSingleton::get().wait_for_all();
  }
 
  void init(grk_decompress_parameters* param) override;
 
  void setBandCallback(grk_io_band_callback callback, void* user_data) override;
  grk_io_band_callback getBandCallback() const override
  {
    return ioBandCallback_;
  }
  void* getBandUserData() const override
  {
    return ioBandUserData_;
  }
 
  grk_progression_state getProgressionState(uint16_t tile_index) override;
 
  bool setProgressionState(grk_progression_state state) override;

  ITileProcessor* getTileProcessor(uint16_t tile_index);

  void initTilesToDecompress(Rect16 region);
 
  void setNumComponents(uint16_t numComps);

  bool initDefaultTCP();
 
  bool readHeader(grk_header_info* header_info) override;

  bool needsHeaderRead(void) const;
 
  GrkImage* getImage(uint16_t tile_index, bool wait) override;
 
  GrkImage* getImage(void) override;
 
  GrkImage* getCompositeNoWait(void);

  GrkImage* getHeaderImage(void);
 
  bool decompress(grk_plugin_tile* tile) override;
 
  bool decompressTile(uint16_t tile_index) override;

  bool postProcess(GrkImage* img);
 
  void dump(uint32_t flag, FILE* outputFileStream) override;
 
  void wait(grk_wait_swath* swath) override;
 
  void scheduleSwathCopy(const grk_wait_swath* swath, grk_swath_buffer* buf) override;
 
  void waitSwathCopy() override;

  void setPostPostProcess(std::function<bool(GrkImage*)> func);
 
protected:
  bool setDecompressRegion(RectD region);

  bool decompressImpl(std::set<uint16_t> pendingTiles);
 
  bool decompressTileImpl(uint16_t tile_index);

  void dumpMainHeader(FILE* outputFileStream);

  void dumpImageHeader(GrkImage* image, bool dev_dump_flag, FILE* outputFileStream);

  void dumpTileHeader(TileCodingParams* defaultTile, uint32_t numcomps, FILE* outputFileStream);

  void dumpImageComponentHeader(grk_image_comp* comp, bool dev_dump_flag, FILE* outputFileStream);
 
private:
  void postReadHeader(void);
 
  void onRowCompleted(uint16_t tileIndexBegin);
  void scheduleTileBatch();
  static constexpr int32_t maxRowsAhead_ = 2;
 
  void wait(uint16_t tile_index);
  void decompressSequentialPrepare(void);
 
  bool schedule(ITileProcessor* tileProcessor, bool multiTile);

  bool sequentialParseAndSchedule(bool multiTile);

  bool readUNK(void);

  void differentialUpdate(GrkImage* scratch);

  bool readCBD(uint8_t* headerData, uint16_t headerSize);

  bool readHeaderProcedure(void);
  bool readSOC();
  bool readSIZ(uint8_t* headerData, uint16_t headerSize);
  bool readCAP(uint8_t* headerData, uint16_t headerSize);

  bool readCRG(uint8_t* headerData, uint16_t headerSize);
  bool readTLM(uint8_t* headerData, uint16_t headerSize);
  bool readPLM(uint8_t* headerData, uint16_t headerSize);

  bool readPPM(uint8_t* headerData, uint16_t headerSize);

  bool readSOT(uint8_t* headerData, uint16_t headerSize);

  bool mergePpm(CodingParams* p_cp);

  bool activateScratch(bool singleTile, GrkImage* scratch);

  std::function<void()> postMultiTile(ITileProcessor* tileProcessor);
  std::function<void()> postMultiTile(void);
 
  std::function<void()> postSingleTile(ITileProcessor* tileProcessor);
 
  std::function<bool()>
      genDecompressTileTLMTask(ITileProcessor* tileProcessor,
                               const std::shared_ptr<TPFetchSeq>& tilePartFetchSeq,
                               Rect32 unreducedImageBounds,
                               std::function<std::function<void()>(ITileProcessor*)> postGenerator);
 
  void decompressSequential(const std::set<uint16_t>& pendingTiles);
  void decompressTLM(const std::set<uint16_t>& pendingTiles);
 
  bool startTLMDecompress(std::set<uint16_t>& pendingTiles);
  bool startSequentialDecompress(std::set<uint16_t>& pendingTiles);
 
  bool doTileBatching(void);

  MarkerParser markerParser_;

  TileWindow tilesToDecompress_;

  std::unique_ptr<MarkerCache> markerCache_;

  ITileProcessor* currTileProcessor_ = nullptr;
 
  TilePartInfo currTilePartInfo_;
  int32_t currTileIndex_ = -1;

  std::unique_ptr<TileCodingParams> defaultTcp_;

  bool headerError_ = false;
  bool headerRead_ = false;

  std::unique_ptr<GrkImage, RefCountedDeleter<GrkImage>> multiTileComposite_;

  Rect32 region_;
 
  std::function<void()> postMulti_;

  std::unique_ptr<TileCache> tileCache_;

  grk_io_pixels_callback ioBufferCallback_ = nullptr;

  void* ioUserData_ = nullptr;

  grk_io_band_callback ioBandCallback_ = nullptr;
  void* ioBandUserData_ = nullptr;
 
  // band ordering for incremental writes
  std::mutex bandOrderMutex_;
  std::condition_variable bandDrainCV_;
  uint16_t nextBandTileY_ = 0;
  struct PendingBand_
  {
    uint32_t yBegin, yEnd;
    uint16_t tileX0, numCols;
  };
  std::map<uint16_t, PendingBand_> pendingBands_;

  grk_io_register_reclaim_callback grkRegisterReclaimCallback_ = nullptr;

  std::function<bool(GrkImage*)> postPostProcess_;

  TileFutureManager decompressTileFutureManager_;

  TileFutureManager swathCopyFutureManager_;
 
  std::shared_ptr<TPFetchSeq> tilePartFetchFlat_;
  std::shared_ptr<std::unordered_map<uint16_t, std::shared_ptr<TPFetchSeq>>> tilePartFetchByTile_;
  std::vector<std::future<bool>> fetchByTileFutures_;
 
  bool fetchByTile(std::set<uint16_t>& slated, Rect32 unreducedImageBounds,
                   std::function<std::function<void()>(ITileProcessor*)> postGenerator);

  std::unique_ptr<GrkImage, RefCountedDeleter<GrkImage>> scratchImage_;
 
  std::unique_ptr<GrkImage, std::function<void(GrkImage*)>> activeImage_;

  std::atomic<bool> success_{true};

  std::atomic<uint32_t> numTilesDecompressed_{0};

  CoderPool coderPool_;
 
  std::vector<std::unique_ptr<MarkerParser>> tileMarkerParsers_;

  bool isHT_ = false;
 
  std::unique_ptr<TileCompletion> tileCompletion_;
 
  std::thread decompressWorker_;
 
  std::function<bool(std::set<uint16_t>&)> decompressStart_;
 
  std::shared_ptr<ChunkBuffer<>> chunkBuffer_;
 
  // Compressed chunk cache: stores fetched compressed data for re-decompression
  std::unique_ptr<CompressedChunkCache> compressedChunkCache_;
 
  // Tile batching
 
  std::mutex batchTileQueueMutex_;
  std::condition_variable batchTileQueueCondition_;
  uint16_t batchTileInitialRows_ = 2;
  uint16_t batchTileNextRows_ = 2;
  uint16_t batchTileHeadroomIncrement(uint16_t numRows, uint16_t tilesLeft);
 
  // batch TLM
  std::queue<uint16_t> batchTileQueueTLM_;
 
  // producer-consumer decompression pipeline
  void startDecompressConsumer(uint16_t maxInFlight);
  std::unique_ptr<ConcurrentQueue<std::function<void()>>> decompressQueue_;
  std::thread decompressConsumer_;
  std::mutex decompressThrottleMutex_;
  std::condition_variable decompressThrottleCV_;
  uint16_t decompressInFlight_ = 0;
  uint16_t maxDecompressInFlight_ = 0;
 
  // Row-based fetch throttle: highest tile row that has been fully fetched
  std::atomic<int32_t> maxFetchedTileRow_{-1};
 
  // batch sequential
  bool batchDequeueSequential(void);
  std::queue<ITileProcessor*> batchTileQueueSequential_;
  uint16_t batchTileScheduleHeadroomSequential_ = 0;
  uint16_t batchTileUnscheduledSequential_ = 0;
  int32_t batchTileScheduledRows_ = 0; // Highest row index we've scheduled up to
};
 
} // namespace grk