scheduling: in single threaded mode, use one inline executor per calling thread

Author: Grok Compression

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

@@ -18,6 +18,7 @@
 #include <chrono>
 #include <cmath>
 #include <functional>
+#include <optional>
 
 #include "TFSingleton.h"
 #include "grk_fseek.h"
@@ -132,6 +133,12 @@ CodeStreamDecompress::CodeStreamDecompress(IStream* stream)
   tileMarkerParsers_.resize(TFSingleton::num_threads());
   std::generate(tileMarkerParsers_.begin(), tileMarkerParsers_.end(),
                 []() { return std::make_unique<MarkerParser>(); });
+
+  // Single-threaded mode: give this codec its own inline (0-worker) executor so
+  // concurrent decodes run on their own thread instead of contending on the
+  // global singleton.  Multi-threaded mode keeps using the shared pool.
+  if(TFSingleton::isSingleThreaded())
+    localExecutor_ = std::make_unique<tf::Executor>(0);
 }
 void CodeStreamDecompress::init(grk_decompress_parameters* parameters)
 {
@@ -166,6 +173,12 @@ void CodeStreamDecompress::setBandCallback(grk_io_band_callback callback, void*
 
 bool CodeStreamDecompress::decompress(grk_plugin_tile* tile)
 {
+  // Route all scheduling/wavelet work onto this codec's own executor while
+  // decoding (single-threaded mode only; no-op when localExecutor_ is null).
+  std::optional<TFSingleton::ScopedExecutor> scopedExec;
+  if(localExecutor_)
+    scopedExec.emplace(localExecutor_.get(), 1);
+
   current_plugin_tile = tile;
 
   // GPU plugin T2-only fast path: skip heavy infrastructure
@@ -928,6 +941,12 @@ bool CodeStreamDecompress::sequentialParseAndSchedule(bool multiTile)
 
 bool CodeStreamDecompress::decompressTile(uint16_t tileIndex)
 {
+  // Route all scheduling/wavelet work onto this codec's own executor while
+  // decoding (single-threaded mode only; no-op when localExecutor_ is null).
+  std::optional<TFSingleton::ScopedExecutor> scopedExec;
+  if(localExecutor_)
+    scopedExec.emplace(localExecutor_.get(), 1);
+
   multiTileComposite_->postReadHeader(&cp_);
 
   // 1. sanity check on tile index

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

@@ -53,7 +53,12 @@ class CodeStreamDecompress final : public CodeStream, public IDecompressor
       decompressConsumer_.join();
     if(decompressWorker_.joinable())
       decompressWorker_.join();
-    TFSingleton::get().wait_for_all();
+    // Wait on this codec's own executor when it owns one (single-threaded mode);
+    // the decode scope has already exited so the thread-local override is gone.
+    if(localExecutor_)
+      localExecutor_->wait_for_all();
+    else
+      TFSingleton::get().wait_for_all();
   }
 
   void init(grk_decompress_parameters* param) override;
@@ -590,6 +595,16 @@ class CodeStreamDecompress final : public CodeStream, public IDecompressor
 
   std::vector<std::unique_ptr<MarkerParser>> tileMarkerParsers_;
 
+  /**
+   * @brief Per-codec inline executor, owned only in single-threaded mode.
+   *
+   * When set, the decode runs on this executor (activated via
+   * TFSingleton::ScopedExecutor for the duration of decompress/decompressTile)
+   * instead of the process-global singleton, so concurrent single-threaded
+   * decodes are independent and lock-free.  Null in multi-threaded mode.
+   */
+  std::unique_ptr<tf::Executor> localExecutor_;
+
   /**
    * @brief global HT flag
    *

src/lib/core/scheduling/CodecScheduler.cpp

@@ -61,6 +61,9 @@ struct ITileProcessor;
 std::unique_ptr<tf::Executor> TFSingleton::instance_ = nullptr;
 std::mutex TFSingleton::mutex_;
 size_t TFSingleton::numThreads_;
+thread_local tf::Executor* TFSingleton::tlsExec_ = nullptr;
+thread_local size_t TFSingleton::tlsNumThreads_ = 0;
+thread_local bool TFSingleton::tlsActive_ = false;
 
 namespace grk
 {

src/lib/core/scheduling/TFSingleton.h

@@ -52,9 +52,15 @@ class TFSingleton
   /**
    * @brief Gets current instance of the Singleton (creates with full hardware concurrency if null)
    * @return Taskflow Executor
+   *
+   * When a per-codec executor is active on this thread (see ScopedExecutor)
+   * it is returned lock-free, so concurrent single-threaded decodes do not
+   * contend on mutex_.  Otherwise the process-global executor is used.
    */
   static tf::Executor& get(void)
   {
+    if(tlsActive_)
+      return *tlsExec_;
     std::lock_guard<std::mutex> lock(mutex_);
     if(!instance_)
     {
@@ -73,6 +79,8 @@ class TFSingleton
    */
   static size_t num_threads()
   {
+    if(tlsActive_)
+      return tlsNumThreads_;
     std::lock_guard<std::mutex> lock(mutex_);
     return numThreads_;
   }
@@ -82,6 +90,8 @@ class TFSingleton
    */
   static bool isSingleThreaded()
   {
+    if(tlsActive_)
+      return tlsNumThreads_ == 1;
     std::lock_guard<std::mutex> lock(mutex_);
     return numThreads_ == 1;
   }
@@ -107,6 +117,39 @@ class TFSingleton
     return (id >= 0) ? (uint32_t)id : 0u;
   }
 
+  /**
+   * @brief Scoped override that makes get()/num_threads()/workerId() resolve to a
+   * caller-owned executor on the current thread instead of the global singleton.
+   *
+   * Used in single-threaded mode so each codec runs on its own inline executor:
+   * concurrent decodes become independent and lock-free.  Saves and restores the
+   * previous thread-local values, so nesting (e.g. overviews) is safe.
+   */
+  class ScopedExecutor
+  {
+  public:
+    ScopedExecutor(tf::Executor* exec, size_t numThreads)
+        : prevExec_(tlsExec_), prevNumThreads_(tlsNumThreads_), prevActive_(tlsActive_)
+    {
+      tlsExec_ = exec;
+      tlsNumThreads_ = numThreads;
+      tlsActive_ = true;
+    }
+    ~ScopedExecutor()
+    {
+      tlsExec_ = prevExec_;
+      tlsNumThreads_ = prevNumThreads_;
+      tlsActive_ = prevActive_;
+    }
+    ScopedExecutor(const ScopedExecutor&) = delete;
+    ScopedExecutor& operator=(const ScopedExecutor&) = delete;
+
+  private:
+    tf::Executor* prevExec_;
+    size_t prevNumThreads_;
+    bool prevActive_;
+  };
+
 private:
   // Deleted copy constructor and assignment operator
   TFSingleton(const TFSingleton&) = delete;
@@ -130,4 +173,19 @@ class TFSingleton
    * @brief total number of threads
    */
   static size_t numThreads_;
+
+  /**
+   * @brief Per-thread override executor (set by ScopedExecutor), or nullptr.
+   */
+  static thread_local tf::Executor* tlsExec_;
+
+  /**
+   * @brief Thread count reported while a ScopedExecutor override is active.
+   */
+  static thread_local size_t tlsNumThreads_;
+
+  /**
+   * @brief True while a ScopedExecutor override is active on this thread.
+   */
+  static thread_local bool tlsActive_;
 };