job_api.hpp

Go to the documentation of this file.

/******************************************************************************/
/******************************************************************************/
#ifndef JOB_API_HPP
#define JOB_API_HPP
 
#include <atomic>   // atomic_uint64_t
#include <cstdint>  // sized integer types
#include <new>      // placement new
#include <utility>  // forward, move
 
#ifndef JOB_SYS_ASSERTIONS
#define JOB_SYS_ASSERTIONS 1  
#endif
 
#if JOB_SYS_ASSERTIONS
#define JobAssert(expr, msg) (::job::internal::AssertHandler)((expr), __FILE__, __LINE__, msg)
#else
#define JobAssert(expr, msg)
#endif
 
namespace job
{
  using WorkerID = std::uint16_t;  
 
  struct Counter
  {
    std::atomic_uint64_t unfinished_tasks = 0u;
  };
 
  struct Ctx
  {
    Counter*    task_counter   = nullptr;  
    WorkerID    current_worker = 0u;       
    const char* task_name      = 0u;       
    std::size_t index          = 0u;       
  };
 
  enum class QueueMode : std::uint8_t
  {
    Default,     
    WorkerOnly,  
  };
 
  std::size_t NumSystemThreads() noexcept;
 
  struct JobSystemCreateOptions
  {
    std::uint16_t num_threads        = 0;     
    std::uint16_t normal_queue_size  = 1024;  
    std::uint16_t worker_queue_size  = 512;   
    std::uint64_t job_steal_rng_seed = 0u;    
  };
 
  struct JobSystemMemoryRequirements
  {
    JobSystemCreateOptions options;    
    std::size_t            byte_size;  
    std::size_t            alignment;  
 
    JobSystemMemoryRequirements(const JobSystemCreateOptions& options = {}) noexcept;
  };
 
  void Initialize(const JobSystemMemoryRequirements& memory_requirements = {}, void* const memory = nullptr) noexcept;
 
  const char* ProcessorArchitectureName() noexcept;
 
  std::uint16_t NumWorkers() noexcept;
 
  WorkerID CurrentWorker() noexcept;
 
  bool IsMainThread() noexcept;
 
  void Shutdown() noexcept;
 
  template<typename Closure>
  void Dispatch(const char* const name, Counter* const counter, const Closure& Callback, const QueueMode queue = QueueMode::Default) noexcept;
 
  void WaitOn(const Counter& counter) noexcept;
 
  void PauseProcessor() noexcept;
 
  void YieldTimeSlice() noexcept;
 
  // Template Function Implementation //
 
  namespace internal
  {
    struct PrivateCtx : public Ctx
    {
      void*             src_user_data   = nullptr;
      std::atomic_bool* is_ready_for_gc = nullptr;
 
      void ReleaseTaskToPool() const;
    };
 
    using JobFn = void (*)(const PrivateCtx& ctx);
 
#if JOB_SYS_ASSERTIONS
    void AssertHandler(const bool condition, const char* const filename, const int line_number, const char* const msg);
#endif
 
    void DispatchImpl(const char* const name,
                      Counter* const    counter,
                      const QueueMode   queue,
                      const JobFn       func,
                      const std::size_t user_data_size,
                      const std::size_t user_data_alignment,
                      const void* const user_data,
                      void (*InitUserData)(void* const user_data, const void* const in_user_data)) noexcept;
  }
 
  template<typename Closure>
  void Dispatch(const char* const name, Counter* const counter, const Closure& Callback, const QueueMode queue) noexcept
  {
    const internal::JobFn ErasedCallback = +[](const internal::PrivateCtx& ctx) -> void {
      alignas(Closure) unsigned char local_user_data[sizeof(Closure)];
      Closure* const                 src_user_data  = static_cast<Closure*>(ctx.src_user_data);
      Closure* const                 typed_callback = ::new (local_user_data) Closure(std::move(*src_user_data));
 
      src_user_data->~Closure();
      ctx.ReleaseTaskToPool();
 
      (*typed_callback)(static_cast<const job::Ctx&>(ctx));
      typed_callback->~Closure();
    };
 
    internal::DispatchImpl(name, counter, queue, ErasedCallback, sizeof(Closure), alignof(Closure), &Callback, +[](void* const dst_user_data, const void* const src_user_data) -> void { ::new (dst_user_data) Closure(*static_cast<const Closure*>(src_user_data)); });
  }
 
  // Parallel Algorithms API
 
  struct Splitter
  {
    static Splitter EvenSplit(const std::size_t total_num_items, std::size_t num_groups_per_thread = 1u)
    {
      if (num_groups_per_thread < 1u)
      {
        num_groups_per_thread = 1u;
      }
 
      return Splitter{(total_num_items / num_groups_per_thread) / NumWorkers()};
    }
 
    static constexpr Splitter MaxItemsPerTask(const std::size_t max_items)
    {
      return Splitter{max_items};
    }
 
    template<typename T>
    static constexpr Splitter MaxDataSize(const std::size_t max_data_size)
    {
      return Splitter{max_data_size / sizeof(T)};
    }
 
    std::size_t max_count = 0u;
 
    constexpr bool operator()(const std::size_t count) const { return count > max_count; }
  };
 
  template<typename F, typename S>
  void ParallelFor(const char* const name, Counter* const counter, const std::size_t start, const std::size_t count, S&& splitter, F&& fn, const QueueMode queue = QueueMode::Default)
  {
    job::Dispatch(name, counter, [=, splitter = std::forward<S>(splitter), fn = std::forward<F>(fn)](const job::Ctx& ctx) -> void {
      if (count > 1u && splitter(count))
      {
        const std::size_t left_count  = count / 2;
        const std::size_t right_count = count - left_count;
 
        job::ParallelFor(ctx.task_name, ctx.task_counter, start + 0, left_count, splitter, fn, queue);
        job::ParallelFor(ctx.task_name, ctx.task_counter, start + left_count, right_count, splitter, fn, queue);
      }
      else
      {
        job::Ctx child_ctx{ctx};
        for (std::size_t offset = 0u; offset < count; ++offset)
        {
          child_ctx.index = start + offset;
          fn(child_ctx);
        }
      } }, queue);
  }
 
  template<typename T, typename F, typename S>
  void ParallelFor(const char* const name, Counter* const counter, T* const data, const std::size_t count, S&& splitter, F&& fn, const QueueMode queue = QueueMode::Default)
  {
    return job::ParallelFor(name, counter, std::size_t(0), count, std::forward<S>(splitter), [=](const job::Ctx& ctx) { fn(ctx, data + ctx.index); }, queue);
  }
 
  template<typename... F>
  void ParallelInvoke(const char* const name, Counter* const counter, const QueueMode queue, F&&... fns)
  {
    (job::Dispatch(name, counter, std::forward<F>(fns), queue), ...);
  }
 
  template<typename Splitter, typename Reducer>
  void ParallelReduce(const char* const name, Counter* const counter, const std::size_t start, const std::size_t count, Splitter&& splitter, Reducer&& reduce, const QueueMode queue = QueueMode::Default)
  {
    const auto ParallelReduce_Impl = [=, splitter = std::forward<Splitter>(splitter), reduce = std::forward<Reducer>(reduce)](const job::Ctx& ctx) -> void {
      // NOTE(SR):
      //   Could also have a stride that increases each step.
      //   This would be bad for Cuda GPU (Shared Memory Bank Conflict)
      //   But good on CPU with better locality.
      //   https://developer.download.nvidia.com/assets/cuda/files/reduction.pdf
 
      std::size_t count_left = count;
 
      while (count_left > 1)
      {
        const std::size_t stride = count_left / 2;
 
        const auto ReduceRange = [stride, &reduce](const job::Ctx& ctx, const std::size_t index) -> void {
          reduce(ctx, index, index + stride);
        };
 
        Counter c{};
        ParallelFor(name, &c, start, stride, splitter, ReduceRange, queue);
        WaitOn(c);
 
        if ((count_left & 1) != 0)
        {
          reduce(ctx, start, start + count_left - 1);
        }
 
        count_left = stride;
      }
    };
 
    job::Dispatch(name, counter, ParallelReduce_Impl, queue);
  }
}
 
#endif  // JOB_API_HPP
 
/******************************************************************************/
/*
  MIT License
 
  Copyright (c) 2020-2026 Shareef Abdoul-Raheem
 
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
 
  The above copyright notice and this permission notice shall be included in all
  copies or substantial portions of the Software.
 
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  SOFTWARE.
*/
/******************************************************************************/