Namespaces
namespace	detail

Classes
struct	InitializationLock

struct	InitializationToken

struct	JobSystemContext

struct	JobSystemCreateOptions
	The runtime configuration for the Job System. More...

struct	JobSystemMemoryRequirements
	The memory requirements for a given configuration `JobSystemCreateOptions`. More...

class	LockedQueue

class	MPMCQueue

struct	Splitter

class	SPMCDeque

class	SPSCQueue

struct	Task

struct	TaskData
	A buffer for user-data you can write to, maybe large enough to store task data inline. More...

union	TaskFnStorage

union	TaskMemoryBlock

struct	TaskPool

struct	TaskPtr

struct	ThreadLocalState

Typedefs
using	WorkerID = std::uint16_t
	The id type of each worker thread. More...

using	TaskFn = void()(Task )
	The signature of the type of function for a single Task. More...

using	TaskHandle = std::uint16_t

using	TaskHandleType = TaskHandle

using	AtomicTaskHandleType = std::atomic< TaskHandle >

using	WorkerIDType = WorkerID

using	AtomicInt32 = std::atomic_int32_t

using	Byte = unsigned char

using	AtomicTaskPtr = std::atomic< TaskPtr >

Enumerations
enum class	QueueType : std::uint8_t { NORMAL = 0 , MAIN = 1 , WORKER = 2 }
	Determines which threads the task will be allowed to run on. More...

enum class	SPMCDequeStatus { SUCCESS , FAILED_RACE , FAILED_SIZE }

Functions
std::size_t	NumSystemThreads () noexcept
	Makes some system calls to grab the number threads / processors on the device. This function can be called by any thread concurrently. More...

InitializationToken	Initialize (const JobSystemMemoryRequirements &memory_requirements={}, void *const memory=nullptr) noexcept
	Sets up the Job system and creates all the worker threads. The thread that calls 'Job::Initialize' is considered the main thread. More...

void	SetupUserThread ()
	Must be called in the callstack of the thread to be setup. More...

const char *	ProcessorArchitectureName () noexcept
	An implementation defined name for the CPU architecture of the device. This function can be called by any thread concurrently. More...

std::uint16_t	NumWorkers () noexcept
	Returns the number of workers created by the system. This function can be called by any thread concurrently. More...

WorkerID	CurrentWorker () noexcept
	The current id of the current thread. This function can be called by any thread concurrently. More...

bool	IsMainThread () noexcept
	Allows for querying if we are currently executing in the main thread. More...

void	Shutdown () noexcept
	This will deallocate any memory used by the system and shutdown any threads created by 'bfJob::initialize'. More...

Task *	TaskMake (const TaskFn function, Task *const parent=nullptr) noexcept
	Creates a new Task that should be later submitted by calling 'TaskSubmit'. More...

TaskData	TaskGetData (Task *const task, const std::size_t alignment) noexcept
	Returns you the user-data buffer you way write to get data into your TaskFn. More...

void	TaskAddContinuation (Task const self, Task const continuation, const QueueType queue=QueueType::NORMAL) noexcept
	A 'continuation' is a task that will be added to a queue after the 'self' Task has finished running. More...

template<typename T >
T *	TaskDataAs (Task *const task) noexcept
	Grabs the user-data pointer as the T you specified. No safety is guaranteed, this is just a dumb cast. More...

template<typename T , typename... Args>
void	TaskEmplaceData (Task *const task, Args &&... args)
	Calls the constructor of T on the user-data buffer. More...

template<typename T >
void	TaskSetData (Task *const task, const T &data)
	Copies 'data' into the user-data buffer by calling the T copy constructor. More...

template<typename T >
void	TaskDestructData (Task *const task)
	Helper for calling destructor on the task's user data. More...

template<typename Closure >
Task *	TaskMake (Closure &&function, Task *const parent=nullptr)
	Creates a new task making a copy of the closure. More...

void	TaskIncRef (Task *const task) noexcept
	Increments the task's ref count preventing it from being garbage collected. More...

void	TaskDecRef (Task *const task) noexcept
	Decrements the task's ref count allow it to be garbage collected. More...

bool	TaskIsDone (const Task *const task) noexcept
	Returns the done status of the task. More...

template<typename ConditionFn >
void	TickMainQueue (ConditionFn &&condition) noexcept
	Runs tasks from the main queue as long as there are tasks available and `condition` returns true. More...

void	TickMainQueue () noexcept
	Runs tasks from the main queue until it is empty. More...

void	TaskSubmit (Task *const self, const QueueType queue=QueueType::NORMAL) noexcept
	Submits the task to the specified queue. More...

void	WaitOnTask (const Task *const task) noexcept
	Waits until the specified `task` is done executing. This function will block but do work while being blocked so there is no wasted time. More...

void	TaskSubmitAndWait (Task *const self, const QueueType queue=QueueType::NORMAL) noexcept
	Same as calling `taskSubmit` followed by `waitOnTask`. More...

void	PauseProcessor () noexcept
	CPU pause instruction to indicate when you are in a spin wait loop. More...

void	YieldTimeSlice () noexcept
	Asks the OS to yield this threads execution to another thread on the current cpu core. More...

template<typename F , typename S >
Task *	ParallelFor (const std::size_t start, const std::size_t count, S &&splitter, F &&fn, Task *parent=nullptr)
	Parallel for algorithm, splits the work up recursively splitting based on the `splitter` passed in. More...

template<typename Splitter , typename Reducer >
Task *	ParallelReduce (const std::size_t start, const std::size_t count, Splitter &&splitter, Reducer &&reduce, Task *parent=nullptr)

template<typename T , typename F , typename S >
Task *	ParallelFor (T const data, const std::size_t count, S &&splitter, F &&fn, Task parent=nullptr)
	Parallel for algorithm, splits the work up recursively splitting based on the `splitter` passed in. This version is a helper for array data. More...

template<typename... F>
Task *	ParallelInvoke (Task *const parent, F &&... fns)
	Invokes each passed in function object in parallel. More...

Variables
static constexpr std::size_t	k_FalseSharingPadSize = std::hardware_destructive_interference_size

static constexpr std::size_t	k_CachelineSize = 64u

static constexpr std::size_t	k_ExpectedTaskSize = std::max(std::size_t(128u), k_CachelineSize)

static constexpr QueueType	k_InvalidQueueType = QueueType(int(QueueType::WORKER) + 1)

static constexpr TaskHandle	NullTaskHandle = std::numeric_limits<TaskHandle>::max()

Class Documentation

◆ Job::InitializationLock

struct Job::InitializationLock

Definition at line 192 of file job_system.cpp.

Class Members
mutex	init_mutex
condition_variable	init_cv
atomic_uint32_t	num_workers_ready

◆ Job::JobSystemContext

struct Job::JobSystemContext

Definition at line 199 of file job_system.cpp.

Class Members
ThreadLocalState *	workers
uint32_t	num_workers
uint32_t	num_owned_workers
atomic_uint32_t	num_user_threads_setup
uint32_t	num_tasks_per_worker
InitializationLock	init_lock
const char *	sys_arch_str
size_t	system_alloc_size
size_t	system_alloc_alignment
bool	needs_delete
atomic_bool	is_running
LockedQueue< TaskPtr >	main_queue
mutex	worker_sleep_mutex
condition_variable	worker_sleep_cv
atomic_uint32_t	num_available_jobs

◆ Job::JobSystemCreateOptions

struct Job::JobSystemCreateOptions

The runtime configuration for the Job System.

Definition at line 85 of file job_api.hpp.

Class Members
uint8_t	num_user_threads	The number of threads not owned by this system but wants access to the Job API (The thread must call Job::SetupUserThread).
uint8_t	num_threads	Use 0 to indicate using the number of cores available on the system.
uint16_t	main_queue_size	Number of tasks in the job system's `QueueType::MAIN` queue. (Must be power of two)
uint16_t	normal_queue_size	Number of tasks in each worker's `QueueType::NORMAL` queue. (Must be power of two)
uint16_t	worker_queue_size	Number of tasks in each worker's `QueueType::WORKER` queue. (Must be power of two)
uint64_t	job_steal_rng_seed	The RNG for work queue stealing will be seeded with this value.

◆ Job::TaskData

struct Job::TaskData

A buffer for user-data you can write to, maybe large enough to store task data inline.

If you store non trivial data remember to manually call it's destructor at the end of the task function.

If you call 'TaskEmplaceData' or 'TaskSetData' and need to update the data once more be sure to destruct the previous contents correctly if the data stored in the buffer is non trivial.

Definition at line 203 of file job_api.hpp.

Class Members
void *	ptr	The start of the buffer you may write to.
size_t	size	The size of the buffer.

◆ Job::TaskMemoryBlock

union Job::TaskMemoryBlock

Definition at line 167 of file job_system.cpp.

Class Members
TaskMemoryBlock *	next
unsigned char	storage[sizeof(Task)]

◆ Job::TaskPool

struct Job::TaskPool

Definition at line 174 of file job_system.cpp.

Class Members
TaskMemoryBlock *	memory
TaskMemoryBlock *	freelist

◆ Job::ThreadLocalState

struct Job::ThreadLocalState

Definition at line 180 of file job_system.cpp.

Class Members
SPMCDeque< TaskPtr >	normal_queue
SPMCDeque< TaskPtr >	worker_queue
TaskPool	task_allocator
TaskHandle *	allocated_tasks
TaskHandleType	num_allocated_tasks
ThreadLocalState *	last_stolen_worker
pcg_state_setseq_64	rng_state
thread	thread_id

Typedef Documentation

◆ WorkerID

using Job::WorkerID = typedef std::uint16_t

The id type of each worker thread.

Definition at line 54 of file job_api.hpp.

◆ TaskFn

using Job::TaskFn = typedef void (*)(Task*)

The signature of the type of function for a single Task.

Definition at line 55 of file job_api.hpp.

◆ TaskHandle

using Job::TaskHandle = typedef std::uint16_t

Definition at line 82 of file job_system.cpp.

◆ TaskHandleType

using Job::TaskHandleType = typedef TaskHandle

Definition at line 83 of file job_system.cpp.

◆ AtomicTaskHandleType

using Job::AtomicTaskHandleType = typedef std::atomic<TaskHandle>

Definition at line 84 of file job_system.cpp.

◆ WorkerIDType

using Job::WorkerIDType = typedef WorkerID

Definition at line 85 of file job_system.cpp.

◆ AtomicInt32

using Job::AtomicInt32 = typedef std::atomic_int32_t

Definition at line 86 of file job_system.cpp.

◆ Byte

using Job::Byte = typedef unsigned char

Definition at line 87 of file job_system.cpp.

◆ AtomicTaskPtr

using Job::AtomicTaskPtr = typedef std::atomic<TaskPtr>

Definition at line 115 of file job_system.cpp.

Enumeration Type Documentation

◆ QueueType

enum class Job::QueueType : std::uint8_t

strong

Determines which threads the task will be allowed to run on.

Enumerator
NORMAL	Tasks in this queue will run on either the main or worker threads.
MAIN	Tasks in this queue will only be run by the main thread.
WORKER	Tasks in this queue will never run on the main thread.

Definition at line 45 of file job_api.hpp.

  {
    NORMAL = 0,  
    MAIN   = 1,  
    WORKER = 2,  
  };

◆ SPMCDequeStatus

enum class Job::SPMCDequeStatus

strong

Enumerator
SUCCESS	Returned from Push, Pop and Steal.
FAILED_RACE	Returned from Pop and Steal.
FAILED_SIZE	Returned from Push, Pop and Steal.

Definition at line 231 of file job_queue.hpp.

  {
    SUCCESS,      
    FAILED_RACE,  
    FAILED_SIZE,  
  };

Function Documentation

◆ NumSystemThreads()

std::size_t Job::NumSystemThreads ( )

noexcept

Makes some system calls to grab the number threads / processors on the device. This function can be called by any thread concurrently.

Can be called even if the job system has not been initialized.

Returns: std::size_t The number threads / processors on the computer.

Definition at line 866 of file job_system.cpp.

{
#if IS_SINGLE_THREADED
  return 1;
#else
  const auto n = std::thread::hardware_concurrency();
  return n != 0 ? n : 1;
#endif
 
#if 0
 
#if IS_WINDOWS
    SYSTEM_INFO sysinfo;
    GetSystemInfo(&sysinfo);
    return sysinfo.dwNumberOfProcessors;
#elif IS_POSIX
    return sysconf(_SC_NPROCESSORS_ONLN) /* * 2*/;
#elif 0  // FreeBSD, MacOS X, NetBSD, OpenBSD
    nt          mib[4];
    int         numCPU;
    std::size_t len = sizeof(numCPU);
 
    /* set the mib for hw.ncpu */
    mib[0] = CTL_HW;
    mib[1] = HW_AVAILCPU;  // alternatively, try HW_NCPU;
 
    /* get the number of CPUs from the system */
    sysctl(mib, 2, &numCPU, &len, NULL, 0);
 
    if (numCPU < 1)
    {
      mib[1] = HW_NCPU;
      sysctl(mib, 2, &numCPU, &len, NULL, 0);
      if (numCPU < 1)
        numCPU = 1;
    }
 
    return numCPU;
#elif 0  // HPUX
    return mpctl(MPC_GETNUMSPUS, NULL, NULL);
#elif 0  // IRIX
    return sysconf(_SC_NPROC_ONLN);
#elif 0  // Objective-C (Mac OS X >=10.5 or iOS)
    NSUInteger a = [[NSProcessInfo processInfo] processorCount];
    NSUInteger b = [[NSProcessInfo processInfo] activeProcessorCount];
 
    return a;
#endif
 
#endif
}

◆ Initialize()

Job::InitializationToken Job::Initialize	(	const JobSystemMemoryRequirements &	memory_requirements = `{}`,
		void *const	memory = `nullptr`
	)

noexcept

Sets up the Job system and creates all the worker threads. The thread that calls 'Job::Initialize' is considered the main thread.

Parameters

memory_requirements	The customization parameters to initialize the system with. To be gotten from `Job::MemRequirementsForConfig`.
memory	Must be `memory_requirements.byte_size` in size and with alignment `memory_requirements.alignment`. If nullptr then the system heap will be used.

Returns: The InitializationToken can be used by other subsystem to verify that the Job System has been initialized.

Definition at line 741 of file job_system.cpp.

{
  JobAssert(g_JobSystem == nullptr, "Already initialized.");
 
  const bool needs_delete = memory == nullptr;
 
  if (!memory)
  {
    memory = ::operator new[](memory_requirements.byte_size, std::align_val_t{memory_requirements.alignment});
  }
 
  JobAssert(memory != nullptr, "memory must be a valid pointer.");
  JobAssert(IsPointerAligned(memory, memory_requirements.alignment), "memory must be a aligned to `memory_requirements.alignment`.");
 
  const JobSystemCreateOptions& options              = memory_requirements.options;
  const std::uint64_t           rng_seed             = options.job_steal_rng_seed;
  const WorkerID                num_threads          = config::WorkerCount(options);
  const WorkerID                owned_threads        = num_threads - options.num_user_threads;
  const std::uint16_t           num_tasks_per_worker = config::NumTasksPerWorker(options);
  const std::uint32_t           total_num_tasks      = config::TotalNumTasks(num_threads, num_tasks_per_worker);
 
  void*                  alloc_ptr        = memory;
  JobSystemContext*      job_system       = LinearAlloc<JobSystemContext>(alloc_ptr, 1u).ptr;
  Span<ThreadLocalState> all_workers      = LinearAlloc<ThreadLocalState>(alloc_ptr, num_threads);
  Span<TaskMemoryBlock>  all_tasks        = LinearAlloc<TaskMemoryBlock>(alloc_ptr, total_num_tasks);
  Span<TaskPtr>          main_tasks_ptrs  = LinearAlloc<TaskPtr>(alloc_ptr, options.main_queue_size);
  Span<AtomicTaskPtr>    worker_task_ptrs = LinearAlloc<AtomicTaskPtr>(alloc_ptr, total_num_tasks);
  Span<TaskHandle>       all_task_handles = LinearAlloc<TaskHandle>(alloc_ptr, total_num_tasks);
 
  job_system->main_queue.Initialize(SpanAlloc(&main_tasks_ptrs, options.main_queue_size), options.main_queue_size);
  job_system->workers           = all_workers.ptr;
  job_system->num_workers       = num_threads;
  job_system->num_owned_workers = owned_threads;
  job_system->num_user_threads_setup.store(0, std::memory_order_relaxed);
  job_system->num_tasks_per_worker = num_tasks_per_worker;
  job_system->sys_arch_str         = "Unknown Arch";
  job_system->num_available_jobs.store(0, std::memory_order_relaxed);
  job_system->needs_delete           = needs_delete;
  job_system->system_alloc_size      = memory_requirements.byte_size;
  job_system->system_alloc_alignment = memory_requirements.alignment;
  job_system->init_lock.num_workers_ready.store(1u, std::memory_order_relaxed);  // Main thread already initialized.
 
#if IS_WINDOWS
  SYSTEM_INFO sysinfo;
  GetSystemInfo(&sysinfo);
 
  switch (sysinfo.wProcessorArchitecture)
  {
    case PROCESSOR_ARCHITECTURE_AMD64:
    {
      job_system->sys_arch_str = "x64 (Intel or AMD)";
      break;
    }
    case PROCESSOR_ARCHITECTURE_ARM:
    {
      job_system->sys_arch_str = "ARM";
      break;
    }
    case PROCESSOR_ARCHITECTURE_ARM64:
    {
      job_system->sys_arch_str = "ARM64";
      break;
    }
    case PROCESSOR_ARCHITECTURE_IA64:
    {
      job_system->sys_arch_str = "Intel Itanium-Based";
      break;
    }
    case PROCESSOR_ARCHITECTURE_INTEL:
    {
      job_system->sys_arch_str = "Intel x86";
      break;
    }
    case PROCESSOR_ARCHITECTURE_UNKNOWN:
    default:
    {
      job_system->sys_arch_str = "Unknown Arch";
      break;
    }
  }
#endif
 
  ThreadLocalState* const main_thread_worker = job_system->workers;
 
  for (std::uint64_t worker_index = 0; worker_index < num_threads; ++worker_index)
  {
    ThreadLocalState* const worker = SpanAlloc(&all_workers, 1u);
 
    worker->normal_queue.Initialize(SpanAlloc(&worker_task_ptrs, options.normal_queue_size), options.normal_queue_size);
    worker->worker_queue.Initialize(SpanAlloc(&worker_task_ptrs, options.worker_queue_size), options.worker_queue_size);
    task_pool::Initialize(&worker->task_allocator, SpanAlloc(&all_tasks, num_tasks_per_worker), num_tasks_per_worker);
    worker->allocated_tasks     = SpanAlloc(&all_task_handles, num_tasks_per_worker);
    worker->num_allocated_tasks = 0u;
    pcg32_srandom_r(&worker->rng_state, worker_index + rng_seed, worker_index * 2u + 1u + rng_seed);
    worker->last_stolen_worker = main_thread_worker;
  }
 
  g_JobSystem     = job_system;
  g_CurrentWorker = main_thread_worker;
 
  std::atomic_thread_fence(std::memory_order_release);
  for (std::uint64_t worker_index = 1; worker_index < owned_threads; ++worker_index)
  {
    worker::InitializeThread(job_system->workers + worker_index);
  }
 
  JobAssert(all_workers.num_elements == 0u, "All elements expected to be allocated out.");
  JobAssert(all_tasks.num_elements == 0u, "All elements expected to be allocated out.");
  JobAssert(main_tasks_ptrs.num_elements == 0u, "All elements expected to be allocated out.");
  JobAssert(worker_task_ptrs.num_elements == 0u, "All elements expected to be allocated out.");
  JobAssert(all_task_handles.num_elements == 0u, "All elements expected to be allocated out.");
 
  return Job::InitializationToken{owned_threads};
}

Referenced by Initialize().

◆ SetupUserThread()

void Job::SetupUserThread ( )

Must be called in the callstack of the thread to be setup.

Sets up the state needed to be able to use the job system from this thread. The job system will not start up until all user threads have been setup.

Warning: Must never be called by either a thread setup by this system or the main thread.

Definition at line 856 of file job_system.cpp.

{
  Job::JobSystemContext* const job_system     = g_JobSystem;
  const std::uint32_t          user_thread_id = job_system->num_owned_workers + job_system->num_user_threads_setup.fetch_add(1, std::memory_order_relaxed);
 
  JobAssert(user_thread_id < job_system->num_workers, "Too many calls to `SetupUserThread`.");
 
  worker::WorkerThreadSetup(job_system->workers + user_thread_id);
}

References g_JobSystem, JobAssert, Job::JobSystemContext::num_owned_workers, Job::JobSystemContext::num_user_threads_setup, and Job::JobSystemContext::workers.

◆ ProcessorArchitectureName()

const char * Job::ProcessorArchitectureName ( )

noexcept

An implementation defined name for the CPU architecture of the device. This function can be called by any thread concurrently.

Returns: const char* Nul terminated name for the CPU architecture of the device.

Definition at line 923 of file job_system.cpp.

{
  return g_JobSystem->sys_arch_str;
}

References g_JobSystem, and Job::JobSystemContext::sys_arch_str.

◆ NumWorkers()

std::uint16_t Job::NumWorkers ( )

noexcept

Returns the number of workers created by the system. This function can be called by any thread concurrently.

Returns: std::size_t The number of workers created by the system.

Definition at line 918 of file job_system.cpp.

{
  return std::uint16_t(g_JobSystem->num_workers);
}

References g_JobSystem, and Job::JobSystemContext::num_workers.

Referenced by Job::Splitter::EvenSplit(), and TaskSubmit().

◆ CurrentWorker()

WorkerID Job::CurrentWorker ( )

noexcept

The current id of the current thread. This function can be called by any thread concurrently.

The main thread will always be 0.

Returns: WorkerID The current id of the current thread.

Definition at line 928 of file job_system.cpp.

{
  JobAssert(g_CurrentWorker != nullptr, "This thread was not created by the job system.");
  return WorkerID(g_CurrentWorker - g_JobSystem->workers);
}

References g_CurrentWorker, g_JobSystem, JobAssert, and Job::JobSystemContext::workers.

Referenced by WaitOnTask().

◆ IsMainThread()

bool Job::IsMainThread ( )

noexcept

Allows for querying if we are currently executing in the main thread.

Returns: True if we are in the main thread, false otherwise.

Warning: Must only be called from a thread registered with the job system.

Definition at line 934 of file job_system.cpp.

{
  return worker::IsMainThread(g_CurrentWorker);
}

References g_CurrentWorker, and IsMainThread().

Referenced by IsMainThread(), and Job::detail::mainQueueTryRunTask().

◆ Shutdown()

void Job::Shutdown ( )

noexcept

This will deallocate any memory used by the system and shutdown any threads created by 'bfJob::initialize'.

Warning: This function may only be called by the main thread.

Definition at line 939 of file job_system.cpp.

{
  JobAssert(g_JobSystem != nullptr, "Cannot shutdown when never initialized.");
 
  static_assert(std::is_trivially_destructible_v<TaskMemoryBlock>, "TaskMemoryBlock's destructor not called.");
  static_assert(std::is_trivially_destructible_v<TaskPtr>, "TaskPtr's destructor not called.");
  static_assert(std::is_trivially_destructible_v<AtomicTaskPtr>, "AtomicTaskPtr's destructor not called.");
  static_assert(std::is_trivially_destructible_v<TaskHandle>, "TaskHandle's destructor not called.");
 
  JobSystemContext* const job_system  = g_JobSystem;
  const std::uint32_t     num_workers = job_system->num_owned_workers;
 
  // Incase all threads are not initialized by the time shutdown is called.
  while (job_system->is_running.load(std::memory_order_relaxed) != true) {}
 
  {
    std::unique_lock<std::mutex> lock(job_system->worker_sleep_mutex);
    job_system->is_running.store(false, std::memory_order_relaxed);
  }
 
  // Allow one last update loop to allow them to end.
  system::WakeUpAllWorkers();
 
  for (std::uint32_t i = 0; i < num_workers; ++i)
  {
    ThreadLocalState* const worker = job_system->workers + i;
 
    if (i != 0)
    {
      worker::ShutdownThread(worker);
    }
 
    worker->~ThreadLocalState();
  }
 
  const bool needs_delete = job_system->needs_delete;
 
  job_system->~JobSystemContext();
  g_CurrentWorker = nullptr;
  g_JobSystem     = nullptr;
 
  if (needs_delete)
  {
    ::operator delete[](job_system, job_system->system_alloc_size, std::align_val_t{job_system->system_alloc_alignment});
  }
}

References g_CurrentWorker, g_JobSystem, Job::JobSystemContext::is_running, JobAssert, Job::JobSystemContext::needs_delete, Job::JobSystemContext::num_owned_workers, Job::JobSystemContext::system_alloc_size, Job::JobSystemContext::worker_sleep_mutex, and Job::JobSystemContext::workers.

◆ TaskMake() [1/2]

Task * Job::TaskMake	(	const TaskFn	function,
		Task *const	parent = `nullptr`
	)

noexcept

Creates a new Task that should be later submitted by calling 'TaskSubmit'.

Parameters

function	The function you want run by the scheduler.
parent	An optional parent Task used in conjunction with 'WaitOnTask' to force dependencies.

Returns: Task* The newly created task.

Definition at line 986 of file job_system.cpp.

{
  const WorkerID          worker_id            = worker::GetCurrentID();
  ThreadLocalState* const worker               = system::GetWorker(worker_id);
  const std::uint32_t     max_tasks_per_worker = g_JobSystem->num_tasks_per_worker;
 
  if (worker->num_allocated_tasks == max_tasks_per_worker)
  {
    worker::GarbageCollectAllocatedTasks(worker);
 
    if (worker->num_allocated_tasks == max_tasks_per_worker)
    {
      // While we cannot allocate do some work.
      system::WakeUpAllWorkers();
      while (worker->num_allocated_tasks == max_tasks_per_worker)
      {
        worker::TryRunTask(worker);
        worker::GarbageCollectAllocatedTasks(worker);
      }
    }
  }
 
  JobAssert(worker->num_allocated_tasks < max_tasks_per_worker, "Too many tasks allocated.");
 
  Task* const      task     = task_pool::AllocateTask(&worker->task_allocator, worker_id, function, task::PointerToTaskPtr(parent));
  const TaskHandle task_hdl = task_pool::TaskToIndex(worker->task_allocator, task);
 
  if (parent)
  {
    parent->num_unfinished_tasks.fetch_add(1u, std::memory_order_release);
  }
 
  worker->allocated_tasks[worker->num_allocated_tasks++] = task_hdl;
 
  return task;
}

References g_JobSystem, JobAssert, and Job::JobSystemContext::num_tasks_per_worker.

Referenced by ParallelFor(), ParallelInvoke(), ParallelReduce(), and TaskMake().

◆ TaskGetData()

TaskData Job::TaskGetData	(	Task *const	task,
		const std::size_t	alignment
	)

noexcept

Returns you the user-data buffer you way write to get data into your TaskFn.

Parameters

task	The task whose user-data you want to grab.
alignment	The required alignment the returned pointer must have.

Returns: TaskData The user-data buffer you may read and write.

Definition at line 1023 of file job_system.cpp.

{
  Byte* const       user_storage_start = static_cast<Byte*>(AlignPointer(task->user_data + task->user_data_start, alignment));
  const Byte* const user_storage_end   = std::end(task->user_data);
 
  if (user_storage_start <= user_storage_end)
  {
    const std::size_t user_storage_size = user_storage_end - user_storage_start;
 
    return {user_storage_start, user_storage_size};
  }
 
  return {nullptr, 0u};
}

Referenced by TaskDataAs(), and TaskEmplaceData().

◆ TaskAddContinuation()

void Job::TaskAddContinuation	(	Task *const	self,
		Task *const	continuation,
		const QueueType	queue = `QueueType::NORMAL`
	)

noexcept

A 'continuation' is a task that will be added to a queue after the 'self' Task has finished running.

Continuations will be added to the same queue as the queue from the task that submits it.

Parameters

self	The task to add the 'continuation' to.
continuation	The Task to run after 'self' has finished. This task must not have already been submitted to a queue.
queue	The queue you want the task to run on.

Definition at line 1038 of file job_system.cpp.

{
  JobAssert(self->q_type == k_InvalidQueueType, "The parent task should not have already been submitted to a queue.");
  JobAssert(continuation->q_type == k_InvalidQueueType, "A continuation must not have already been submitted to a queue or already added as a continuation.");
  JobAssert(continuation->next_continuation.isNull(), "A continuation must not have already been added to another task.");
 
  const TaskPtr new_head          = task::PointerToTaskPtr(continuation);
  continuation->q_type            = queue;
  continuation->next_continuation = self->first_continuation.load(std::memory_order_relaxed);
 
  while (!std::atomic_compare_exchange_strong(&self->first_continuation, &continuation->next_continuation, new_head))
  {
  }
}

References JobAssert, and k_InvalidQueueType.

◆ TaskDataAs()

template<typename T >

T * Job::TaskDataAs ( Task *const task )

noexcept

Grabs the user-data pointer as the T you specified. No safety is guaranteed, this is just a dumb cast.

Template Parameters

T	The type you want to receive the user-data buffer as.

Parameters

task	The task whose data you are retrieving.

Returns: T& The user-data buffer casted as a T.

Definition at line 479 of file job_api.hpp.

  {
    const TaskData data = TaskGetData(task, alignof(T));
 
    return data.size >= sizeof(T) ? static_cast<T*>(data.ptr) : nullptr;
  }

References Job::TaskData::ptr, Job::TaskData::size, and TaskGetData().

◆ TaskEmplaceData()

template<typename T , typename... Args>

void Job::TaskEmplaceData	(	Task *const	task,
		Args &&...	args
	)

Calls the constructor of T on the user-data buffer.

Template Parameters

T	The type of T you want constructed in-place into the user-data buffer.
Args	The Argument types passed into the T constructor.

Parameters

task	The task whose user-data buffer is affected.
args	The arguments passed into the constructor of the user-data buffer casted as a T.

Definition at line 487 of file job_api.hpp.

  {
    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)...);
  }

References JobAssert, Job::TaskData::ptr, Job::TaskData::size, and TaskGetData().

◆ TaskSetData()

template<typename T >

void Job::TaskSetData	(	Task *const	task,
		const T &	data
	)

Copies 'data' into the user-data buffer by calling the T copy constructor.

Template Parameters

T	The data type that will be emplaced into the user-data buffer.

Parameters

task	The task whose user-data buffer is affected.
data	The data copied into the user-data buffer.

Definition at line 497 of file job_api.hpp.

  {
    TaskEmplaceData<T>(task, data);
  }

◆ TaskDestructData()

template<typename T >

void Job::TaskDestructData ( Task *const task )

Helper for calling destructor on the task's user data.

Template Parameters

T	The expected type of the data, called ~T on the user data buffer.

Parameters

task	The task whose user-data buffer is affected.

Definition at line 503 of file job_api.hpp.

  {
    TaskDataAs<T>(task)->~T();
  }

◆ TaskMake() [2/2]

template<typename Closure >

Task * Job::TaskMake	(	Closure &&	function,
		Task *const	parent = `nullptr`
	)

Creates a new task making a copy of the closure.

Template Parameters

Closure The type of the callable.

Parameters

function	The non pointer callable you want to store.
parent	An optional parent Task used in conjunction with 'WaitOnTask' to force dependencies.

Returns: Task* The newly created task.

Definition at line 509 of file job_api.hpp.

  {
    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;
  }

References Job::detail::taskGetPrivateUserData(), TaskMake(), and Job::detail::taskReservePrivateUserData().

◆ TaskIncRef()

void Job::TaskIncRef ( Task *const task )

noexcept

Increments the task's ref count preventing it from being garbage collected.

This function should be called before taskSubmit.

Parameters

task	The task's who's ref count should be incremented.

Definition at line 1053 of file job_system.cpp.

{
  const auto old_ref_count = task->ref_count.fetch_add(1, std::memory_order_relaxed);
 
  JobAssert(old_ref_count >= std::int16_t(1) || task->q_type == k_InvalidQueueType, "First call to taskIncRef should not happen after the task has been submitted.");
  (void)old_ref_count;
}

References JobAssert, and k_InvalidQueueType.

◆ TaskDecRef()

void Job::TaskDecRef ( Task *const task )

noexcept

Decrements the task's ref count allow it to be garbage collected.

Parameters

task	The task's who's ref count should be decremented.

Definition at line 1061 of file job_system.cpp.

{
  const auto old_ref_count = task->ref_count.fetch_sub(1, std::memory_order_relaxed);
 
  JobAssert(old_ref_count >= 0, "taskDecRef: Called too many times.");
  (void)old_ref_count;
}

References JobAssert.

◆ TaskIsDone()

bool Job::TaskIsDone ( const Task *const task )

noexcept

Returns the done status of the task.

This is only safe to call after submitting the task if you have an active reference to the task through a call to TaskIncRef.

Parameters

task	The task to check whether or not it's done.

Returns: true - The task is done running. false - The task is still running.

See also: TaskIncRef

Definition at line 1069 of file job_system.cpp.

{
  return task->num_unfinished_tasks.load(std::memory_order_acquire) == -1;
}

Referenced by WaitOnTask().

◆ TickMainQueue() [1/2]

template<typename ConditionFn >

void Job::TickMainQueue ( ConditionFn && condition )

noexcept

Runs tasks from the main queue as long as there are tasks available and condition returns true.

This function is not required to be called since the main queue will be evaluated during other calls to this API but allows for an easy way to flush the main queue guaranteeing a minimum latency.

Template Parameters

ConditionFn The type of the callable. Must be callable like: bool operator()(void);.

Parameters

condition The function object indicating if the main queue should continue being evaluated. Will be called after a task has been completed.

Warning: Must only be called from the main thread.

Definition at line 397 of file job_api.hpp.

  {
    do
    {
      if (!detail::mainQueueTryRunTask())
      {
        break;
      }
    } while (condition());
  }

References Job::detail::mainQueueTryRunTask().

Referenced by TickMainQueue().

◆ TickMainQueue() [2/2]

void Job::TickMainQueue ( )

inlinenoexcept

Runs tasks from the main queue until it is empty.

This function is not required to be called since the main queue will be evaluated during other calls to this API but allows for an easy way to flush the main queue guaranteeing a minimum latency.

Warning: Must only be called from the main thread.

Definition at line 418 of file job_api.hpp.

  {
    TickMainQueue([]() { return true; });
  }

References TickMainQueue().

◆ TaskSubmit()

void Job::TaskSubmit	(	Task *const	self,
		const QueueType	queue = `QueueType::NORMAL`
	)

noexcept

Submits the task to the specified queue.

The Task is not required to have been created on the same thread that submits.

You may now wait on this task using 'WaitOnTask'.

Parameters

self	The task to submit.
queue	The queue you want the task to run on.

Definition at line 1074 of file job_system.cpp.

{
  JobAssert(self->q_type == k_InvalidQueueType, "A task cannot be submitted to a queue multiple times.");
 
  const WorkerID num_workers = NumWorkers();
 
  // If we only have one thread running using the worker queue is invalid.
  if (num_workers == 1u && queue == QueueType::WORKER)
  {
    queue = QueueType::NORMAL;
  }
 
  ThreadLocalState* const worker   = worker::GetCurrent();
  const TaskPtr           task_ptr = task::PointerToTaskPtr(self);
 
  self->q_type = queue;
 
  switch (queue)
  {
    case QueueType::NORMAL:
    {
      task::SubmitQPushHelper(task_ptr, worker, &worker->normal_queue);
      break;
    }
    case QueueType::MAIN:
    {
      LockedQueue<TaskPtr>* const main_queue = &g_JobSystem->main_queue;
 
      // NOTE(SR):
      //   The only way `main_queue` will be emptied
      //   is by the main thread, so there is a chance
      //   that if it does not get flushed frequently
      //   enough then we have a this thread spinning indefinitely.
      //
      while (!main_queue->Push(task_ptr))
      {
        // If we could not push to the queue then just do some work.
        worker::TryRunTask(worker);
      }
      break;
    }
    case QueueType::WORKER:
    {
      task::SubmitQPushHelper(task_ptr, worker, &worker->worker_queue);
      break;
    }
    default:
#if defined(__GNUC__)  // GCC, Clang, ICC
      __builtin_unreachable();
#elif defined(_MSC_VER)  // MSVC
      __assume(false);
#endif
      break;
  }
 
  if (queue != QueueType::MAIN)
  {
    const std::int32_t num_pending_jobs = g_JobSystem->num_available_jobs.fetch_add(1, std::memory_order_relaxed);
 
    if (num_pending_jobs >= num_workers)
    {
      system::WakeUpAllWorkers();
    }
    else
    {
      system::WakeUpOneWorker();
    }
  }
}

References g_JobSystem, JobAssert, k_InvalidQueueType, MAIN, Job::JobSystemContext::main_queue, NORMAL, Job::JobSystemContext::num_available_jobs, NumWorkers(), Job::LockedQueue< T >::Push(), and WORKER.

Referenced by ParallelFor(), ParallelInvoke(), and TaskSubmitAndWait().

◆ WaitOnTask()

void Job::WaitOnTask ( const Task *const task )

noexcept

Waits until the specified task is done executing. This function will block but do work while being blocked so there is no wasted time.

You may only call this function with a task created on the current 'Worker'.

It is a logic error to call this function on a task that has not been submitted (TaskSubmit).

Parameters

task	The task to wait to finish executing.

Definition at line 1144 of file job_system.cpp.

{
  const WorkerID worker_id = CurrentWorker();
 
  JobAssert(task->q_type != k_InvalidQueueType, "The Task must be submitted to a queue before you wait on it.");
  JobAssert(task->owning_worker == worker_id, "You may only call this function with a task created on the current 'Worker'.");
 
  system::WakeUpAllWorkers();
 
  ThreadLocalState* const worker = system::GetWorker(worker_id);
 
  while (!TaskIsDone(task))
  {
    worker::TryRunTask(worker);
  }
}

References CurrentWorker(), JobAssert, k_InvalidQueueType, and TaskIsDone().

Referenced by TaskSubmitAndWait().

◆ TaskSubmitAndWait()

void Job::TaskSubmitAndWait	(	Task *const	self,
		const QueueType	queue = `QueueType::NORMAL`
	)

noexcept

Same as calling taskSubmit followed by waitOnTask.

Parameters

self	The task to submit and wait to finish executing.
queue	The queue you want the task to run on.

Definition at line 1161 of file job_system.cpp.

{
  TaskSubmit(self, queue);
  WaitOnTask(self);
}

References TaskSubmit(), and WaitOnTask().

Referenced by ParallelReduce().

◆ PauseProcessor()

void Job::PauseProcessor ( )

noexcept

CPU pause instruction to indicate when you are in a spin wait loop.

Definition at line 1198 of file job_system.cpp.

{
  NativePause();
}

References NativePause.

Referenced by Job::MPMCQueue::Commit().

◆ YieldTimeSlice()

void Job::YieldTimeSlice ( )

noexcept

Asks the OS to yield this threads execution to another thread on the current cpu core.

Definition at line 1205 of file job_system.cpp.

{
  // Windows : SwitchToThread()
  // Linux   : sched_yield()
  std::this_thread::yield();
}

◆ ParallelFor() [1/2]

template<typename F , typename S >

Task * Job::ParallelFor	(	const std::size_t	start,
		const std::size_t	count,
		S &&	splitter,
		F &&	fn,
		Task *	parent = `nullptr`
	)

Parallel for algorithm, splits the work up recursively splitting based on the splitter passed in.

Assumes all callable objects passed in can be invoked on multiple threads at the same time.

Template Parameters

F	Type of function object passed in. Must be callable like: fn(Task* task, std::size_t index_range)
S	Callable splitter, must be callable like: splitter(std::size_t count)

Parameters

start	Start index for the range to be parallelized.
count	`start` + count defines the end range.
splitter	Callable splitter, must be callable like: splitter(std::size_t count)
fn	Function object must be callable like: fn(Job::Task* const task, const std::size_t index)
parent	Parent task to add this task as a child of.

Returns: The new task holding the work of the parallel for.

Definition at line 608 of file job_api.hpp.

  {
    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);
  }

References ParallelFor(), TaskMake(), Job::detail::taskQType(), and TaskSubmit().

Referenced by ParallelFor(), and ParallelReduce().

◆ ParallelReduce()

template<typename Splitter , typename Reducer >

Task * Job::ParallelReduce	(	const std::size_t	start,
		const std::size_t	count,
		Splitter &&	splitter,
		Reducer &&	reduce,
		Task *	parent = `nullptr`
	)

Definition at line 633 of file job_api.hpp.

  {
    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);
  }

References ParallelFor(), TaskMake(), Job::detail::taskQType(), and TaskSubmitAndWait().

◆ ParallelFor() [2/2]

template<typename T , typename F , typename S >

Task * Job::ParallelFor	(	T *const	data,
		const std::size_t	count,
		S &&	splitter,
		F &&	fn,
		Task *	parent = `nullptr`
	)

Parallel for algorithm, splits the work up recursively splitting based on the splitter passed in. This version is a helper for array data.

Assumes all callable objects passed in can be invoked on multiple threads at the same time.

Template Parameters

T	Type of the array to process.
F	Type of function object passed in. Must be callable like: fn(Task* task, IndexRange index_range)
S	Callable splitter, must be callable like: splitter(std::size_t count)

Parameters

data	The start of the array to process.
count	The number of elements in the `data` array.
splitter	Callable splitter, must be callable like: splitter(std::size_t count)
fn	Function object must be callable like: fn(job::Task* task, T* data_start, const std::size_t num_items)
parent	Parent task to add this task as a child of.

Returns: The new task holding the work of the parallel for.

Definition at line 702 of file job_api.hpp.

  {
    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);
  }

References ParallelFor().

◆ ParallelInvoke()

template<typename... F>

Task * Job::ParallelInvoke	(	Task *const	parent,
		F &&...	fns
	)

Invokes each passed in function object in parallel.

Template Parameters

...F	The function objects types. Must be callable like: fn(Task* task)

Parameters

parent	Parent task to add this task as a child of.
...fns	Function objects must be callable like: fn(Task* task)

Returns: The new task holding the work of the parallel invoke.

Definition at line 729 of file job_api.hpp.

  {
    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);
  }

References TaskMake(), Job::detail::taskQType(), and TaskSubmit().

Variable Documentation

◆ k_FalseSharingPadSize

constexpr std::size_t Job::k_FalseSharingPadSize = std::hardware_destructive_interference_size

staticconstexpr

Definition at line 30 of file job_queue.hpp.

◆ k_CachelineSize

constexpr std::size_t Job::k_CachelineSize = 64u

staticconstexpr

Definition at line 74 of file job_system.cpp.

◆ k_ExpectedTaskSize

constexpr std::size_t Job::k_ExpectedTaskSize = std::max(std::size_t(128u), k_CachelineSize)

staticconstexpr

Definition at line 77 of file job_system.cpp.

◆ k_InvalidQueueType

constexpr QueueType Job::k_InvalidQueueType = QueueType(int(QueueType::WORKER) + 1)

staticconstexpr

Definition at line 78 of file job_system.cpp.

Referenced by TaskAddContinuation(), TaskIncRef(), TaskSubmit(), and WaitOnTask().

◆ NullTaskHandle

constexpr TaskHandle Job::NullTaskHandle = std::numeric_limits<TaskHandle>::max()

staticconstexpr

Definition at line 89 of file job_system.cpp.

Referenced by Job::TaskPtr::isNull(), and Job::TaskPtr::TaskPtr().

Namespaces

Classes

Typedefs

Enumerations

Functions

Variables

Class Documentation

◆ Job::InitializationLock

◆ Job::JobSystemContext

◆ Job::JobSystemCreateOptions

◆ Job::TaskData

◆ Job::TaskMemoryBlock

◆ Job::TaskPool

◆ Job::ThreadLocalState

Typedef Documentation

◆ WorkerID

◆ TaskFn

◆ TaskHandle

◆ TaskHandleType

◆ AtomicTaskHandleType

◆ WorkerIDType

◆ AtomicInt32

◆ Byte

◆ AtomicTaskPtr

Enumeration Type Documentation

◆ QueueType

◆ SPMCDequeStatus

Function Documentation

◆ NumSystemThreads()

◆ Initialize()

◆ SetupUserThread()

◆ ProcessorArchitectureName()

◆ NumWorkers()

◆ CurrentWorker()

◆ IsMainThread()

◆ Shutdown()

◆ TaskMake() [1/2]

◆ TaskGetData()

◆ TaskAddContinuation()

◆ TaskDataAs()

◆ TaskEmplaceData()

◆ TaskSetData()

◆ TaskDestructData()

◆ TaskMake() [2/2]

◆ TaskIncRef()

◆ TaskDecRef()

◆ TaskIsDone()

◆ TickMainQueue() [1/2]

◆ TickMainQueue() [2/2]

◆ TaskSubmit()

◆ WaitOnTask()

◆ TaskSubmitAndWait()

◆ PauseProcessor()

◆ YieldTimeSlice()

◆ ParallelFor() [1/2]

◆ ParallelReduce()

◆ ParallelFor() [2/2]

◆ ParallelInvoke()

Variable Documentation

◆ k_FalseSharingPadSize

◆ k_CachelineSize

◆ k_ExpectedTaskSize

◆ k_InvalidQueueType

◆ NullTaskHandle