job_api.hpp

Go to the documentation of this file.

/******************************************************************************/
/******************************************************************************/
#ifndef JOB_API_HPP
#define JOB_API_HPP
 
#include "job_assert.hpp"  // JobAssert
 
#include <cstdint>  // sized integer types
#include <new>      // placement new
#include <utility>  // forward, move
 
namespace Job
{
  // Fwd Declarations
 
  struct Task;
 
  // Enums
 
  enum class QueueType : std::uint8_t
  {
    NORMAL = 0,  
    MAIN   = 1,  
    WORKER = 2,  
  };
 
  // Type Aliases
 
  using WorkerID = std::uint16_t;    
  using TaskFn   = void (*)(Task*);  
 
  // Private
 
  namespace detail
  {
    QueueType taskQType(const Task* task) noexcept;
    void*     taskGetPrivateUserData(Task* const task, const std::size_t alignment) noexcept;
    void*     taskReservePrivateUserData(Task* const task, const std::size_t num_bytes, const std::size_t alignment) noexcept;
    bool      mainQueueTryRunTask(void) noexcept;
  }  // namespace detail
 
  std::size_t NumSystemThreads() noexcept;
 
  // Main System API
 
  struct JobSystemCreateOptions
  {
    std::uint8_t  num_user_threads   = 0;     
    std::uint8_t  num_threads        = 0;     
    std::uint16_t main_queue_size    = 256;   
    std::uint16_t normal_queue_size  = 1024;  
    std::uint16_t worker_queue_size  = 32;    
    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;
 
  void SetupUserThread();
 
  const char* ProcessorArchitectureName() noexcept;
 
  std::uint16_t NumWorkers() noexcept;
 
  WorkerID CurrentWorker() noexcept;
 
  bool IsMainThread() noexcept;
 
  void Shutdown() noexcept;
 
  // Task API
 
  struct TaskData
  {
    void*       ptr;   
    std::size_t size;  
  };
 
  Task* TaskMake(const TaskFn function, Task* const parent = nullptr) noexcept;
 
  TaskData TaskGetData(Task* const task, const std::size_t alignment) noexcept;
 
  void TaskAddContinuation(Task* const self, Task* const continuation, const QueueType queue = QueueType::NORMAL) noexcept;
 
  template<typename T>
  T* TaskDataAs(Task* const task) noexcept;
 
  template<typename T, typename... Args>
  void TaskEmplaceData(Task* const task, Args&&... args);
 
  template<typename T>
  void TaskSetData(Task* const task, const T& data);
 
  template<typename T>
  void TaskDestructData(Task* const task);
 
  template<typename Closure>
  Task* TaskMake(Closure&& function, Task* const parent = nullptr);
 
  void TaskIncRef(Task* const task) noexcept;
 
  void TaskDecRef(Task* const task) noexcept;
 
  bool TaskIsDone(const Task* const task) noexcept;
 
  template<typename ConditionFn>
  void TickMainQueue(ConditionFn&& condition) noexcept
  {
    do
    {
      if (!detail::mainQueueTryRunTask())
      {
        break;
      }
    } while (condition());
  }
 
  inline void TickMainQueue() noexcept
  {
    TickMainQueue([]() { return true; });
  }
 
  void TaskSubmit(Task* const self, const QueueType queue = QueueType::NORMAL) noexcept;
 
  void WaitOnTask(const Task* const task) noexcept;
 
  void TaskSubmitAndWait(Task* const self, const QueueType queue = QueueType::NORMAL) noexcept;
 
  void PauseProcessor() noexcept;
 
  void YieldTimeSlice() noexcept;
 
  // Template Function Implementation //
 
  template<typename T>
  T* TaskDataAs(Task* const task) noexcept
  {
    const TaskData data = TaskGetData(task, alignof(T));
 
    return data.size >= sizeof(T) ? static_cast<T*>(data.ptr) : nullptr;
  }
 
  template<typename T, typename... Args>
  void TaskEmplaceData(Task* const task, Args&&... args)
  {
    const TaskData data = TaskGetData(task, alignof(T));
 
    JobAssert(data.size >= sizeof(T), "Attempting to store an object too large to fit within a task's storage buffer.");
 
    new (data.ptr) T(std::forward<Args>(args)...);
  }
 
  template<typename T>
  void TaskSetData(Task* const task, const T& data)
  {
    TaskEmplaceData<T>(task, data);
  }
 
  template<typename T>
  void TaskDestructData(Task* const task)
  {
    TaskDataAs<T>(task)->~T();
  }
 
  template<typename Closure>
  Task* TaskMake(Closure&& function, Task* const parent)
  {
    Task* const task = TaskMake(
     +[](Task* const task) -> void {
       Closure& function = *static_cast<Closure*>(detail::taskGetPrivateUserData(task, alignof(Closure)));
       function(task);
       function.~Closure();
     },
     parent);
    void* const private_data = detail::taskReservePrivateUserData(task, sizeof(Closure), alignof(Closure));
    ::new (private_data) Closure(std::forward<Closure>(function));
 
    return task;
  }
 
  // 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>
  Task* ParallelFor(const std::size_t start, const std::size_t count, S&& splitter, F&& fn, Task* parent = nullptr)
  {
    return TaskMake(
     [=, splitter = std::move(splitter), fn = std::move(fn)](Task* const task) {
       if (count > 1u && splitter(count))
       {
         const std::size_t left_count    = count / 2;
         const std::size_t right_count   = count - left_count;
         const QueueType   parent_q_type = detail::taskQType(task);
 
         TaskSubmit(ParallelFor(start, left_count, splitter, fn, task), parent_q_type);
         TaskSubmit(ParallelFor(start + left_count, right_count, splitter, fn, task), parent_q_type);
       }
       else
       {
         for (std::size_t offset = 0u; offset < count; ++offset)
         {
           fn(task, start + offset);
         }
       }
     },
     parent);
  }
 
  template<typename Splitter, typename Reducer>
  Task* ParallelReduce(const std::size_t start, const std::size_t count, Splitter&& splitter, Reducer&& reduce, Task* parent = nullptr)
  {
    return TaskMake(
     [=, splitter = std::move(splitter), reduce = std::move(reduce)](Task* const task) {
       // 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
 
       const QueueType parent_q_type = detail::taskQType(task);
       std::size_t     count_left    = count;
       while (count_left > 1)
       {
         const std::size_t stride = count_left / 2;
 
         const auto ReduceRange = [stride, &reduce](Task* const sub_task, const std::size_t index) {
           reduce(sub_task, index, index + stride);
         };
 
         TaskSubmitAndWait(ParallelFor(start, stride, splitter, ReduceRange, nullptr), parent_q_type);
 
         if ((count_left & 1) != 0)
         {
           reduce(task, start, start + count_left - 1);
         }
 
         count_left = stride;
       }
     },
     parent);
  }
 
  template<typename T, typename F, typename S>
  Task* ParallelFor(T* const data, const std::size_t count, S&& splitter, F&& fn, Task* parent = nullptr)
  {
    return ParallelFor(
     std::size_t(0), count, std::move(splitter), [data, fn = std::move(fn)](Task* const task, const std::size_t index) {
       fn(task, data + index);
     },
     parent);
  }
 
  template<typename... F>
  Task* ParallelInvoke(Task* const parent, F&&... fns)
  {
    return TaskMake(
     [=](Task* const parent_task) mutable {
       const QueueType parent_q_type = detail::taskQType(parent_task);
       (TaskSubmit(TaskMake(std::move(fns), parent_task), parent_q_type), ...);
     },
     parent);
  }
}  // namespace Job
 
#endif  // JOB_API_HPP
 
/******************************************************************************/
/*
  MIT License
 
  Copyright (c) 2020-2025 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.
*/
/******************************************************************************/