jobsystem.h

#pragma once

#include <algorithm>
#include <vector>
#include <deque>
#include <array>
#include <thread>
#include <functional>
#include <atomic>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <stdio.h>
#include <string.h>
#include <inttypes.h>

namespace jobsystem
{
    inline uint64_t GetBit(uint64_t n)
    {
        return static_cast<uint64_t>(1) << n;
    }

    inline size_t CountBits(uint64_t n)
    {
        size_t bits = 0;
        while (n)
        {
            bits += n & 1;
            n >>= 1;
        }
        return bits;
    }

    typedef std::function<void()> JobDelegate; // Structure of callbacks that can be requested as jobs.

    typedef uint64_t affinity_t;

    static const affinity_t kAffinityAllBits = static_cast<affinity_t>(~0);

    /**
     * Global system components.
     */
    std::atomic<size_t> s_nextJobId;         // Job ID assignment for debugging / profiling.
    std::mutex s_signalLock;                 // Global mutex for worker signaling.
    std::condition_variable s_signalThreads; // Global condition var for worker signaling.
    std::atomic<size_t> s_activeWorkers;

    inline affinity_t CalculateSafeWorkerAffinity(size_t workerIndex, size_t workerCount)
    {
        affinity_t affinity = kAffinityAllBits; // Set all bits so jobs with affinities out of range can still be processed.
        affinity &= ~(workerCount - 1);         // Wipe bits within valid range.
        affinity |= GetBit(workerIndex);        // Set worker-specific bit.

        return affinity;
    }

    /**
     * Offers access to the state of job.
     * In particular, callers can use the Wait() function to ensure a given job is complete,
     * or Cancel().
     * Note, however, that doing so is not good practice with job systems. If no hardware threads
     * are available to process a given job, you can stall the caller for significant time.
     *
     * Internally, the state manages dependencies as well as atomics describing the status of the job.
     */
    typedef std::shared_ptr<class JobState> JobStatePtr;

    class JobState
    {
    private:
        friend class JobSystemWorker;
        friend class JobManager;

        std::atomic<bool> m_cancel; // Is the job pending cancellation?
        std::atomic<bool> m_ready;  // Has the job been marked as ready for processing?

        std::vector<JobStatePtr> m_dependants; // List of dependent jobs.
        std::atomic<int> m_dependencies;       // Number of outstanding dependencies.

        std::atomic<bool> m_done; // Has the job executed to completion?
        std::condition_variable m_doneSignal;
        std::mutex m_doneMutex;

        affinity_t m_workerAffinity; // Option to limit execution to specific worker threads / cores.

        size_t m_jobId;   // Debug/profiling ID.
        char m_debugChar; // Debug character for profiling display.

        void SetQueued()
        {
            m_done.store(false, std::memory_order_release);
        }

        void SetDone()
        {
            JOBSYSTEM_ASSERT(!IsDone());

            for (const JobStatePtr &dependant : m_dependants)
            {
                dependant->m_dependencies.fetch_sub(1, std::memory_order_relaxed);
            }

            std::lock_guard<std::mutex> lock(m_doneMutex);
            m_done.store(true, std::memory_order_release);
            m_doneSignal.notify_all();
        }

        bool AwaitingCancellation() const
        {
            return m_cancel.load(std::memory_order_relaxed);
        }

    public:
        JobState()
            : m_debugChar(0)
        {
            m_jobId = s_nextJobId++;
            m_workerAffinity = kAffinityAllBits;

            m_dependencies.store(0, std::memory_order_release);
            m_cancel.store(false, std::memory_order_release);
            m_ready.store(false, std::memory_order_release);
            m_done.store(false, std::memory_order_release);
        }

        ~JobState() {}

        JobState &SetReady()
        {
            JOBSYSTEM_ASSERT(!IsDone());

            m_cancel.store(false, std::memory_order_relaxed);
            m_ready.store(true, std::memory_order_release);

            s_signalThreads.notify_all();

            return *this;
        }

        JobState &Cancel()
        {
            JOBSYSTEM_ASSERT(!IsDone());

            m_cancel.store(true, std::memory_order_relaxed);

            return *this;
        }

        JobState &AddDependant(JobStatePtr dependant)
        {
            JOBSYSTEM_ASSERT(m_dependants.end() == std::find(m_dependants.begin(), m_dependants.end(), dependant));

            m_dependants.push_back(dependant);

            dependant->m_dependencies.fetch_add(1, std::memory_order_relaxed);

            return *this;
        }

        JobState &SetWorkerAffinity(affinity_t affinity)
        {
            m_workerAffinity = affinity ? affinity : kAffinityAllBits;

            return *this;
        }

        bool IsDone() const
        {
            return m_done.load(std::memory_order_acquire);
        }

        bool Wait(size_t maxWaitMicroseconds = 0)
        {
            if (!IsDone())
            {
                std::unique_lock<std::mutex> lock(m_doneMutex);

                if (maxWaitMicroseconds == 0)
                {
                    m_doneSignal.wait(lock,
                                      [this]()
                                      {
                                          return IsDone();
                                      });
                }
                else
                {
                    m_doneSignal.wait_for(lock, std::chrono::microseconds(maxWaitMicroseconds));
                }
            }

            return IsDone();
        }

        bool AreDependenciesMet() const
        {
            if (!m_ready.load(std::memory_order_acquire))
            {
                return false;
            }

            if (m_dependencies.load(std::memory_order_relaxed) > 0)
            {
                return false;
            }

            return true;
        }

        bool HasDependencies() const
        {
            return (m_dependencies.load(std::memory_order_relaxed) > 0);
        }
    };

    /**
     * Represents an entry in a job queue.
     * - A delegate to invoke
     * - Internal job state
     */
    struct JobQueueEntry
    {
        JobDelegate m_delegate; // Delegate to invoke for the job.
        JobStatePtr m_state;    // Pointer to job state.
    };

    /**
     * Descriptor for a given job worker thread, to be provided by the host application.
     */
    struct JobWorkerDescriptor
    {
        JobWorkerDescriptor(const char *name = "JobSystemWorker", affinity_t cpuAffinity = affinity_t(~0), bool enableWorkSteeling = true)
            : m_name(name), m_cpuAffinity(cpuAffinity), m_enableWorkStealing(enableWorkSteeling)
        {
        }

        std::string m_name;            // Worker name, for debug/profiling displays.
        affinity_t m_cpuAffinity;      // Thread affinity. Defaults to all cores.
        bool m_enableWorkStealing : 1; // Enable queue-sharing between workers?
    };

    /**
     * Job events (for tracking/debugging).
     */
    enum EJobEvent
    {
        eJobEvent_JobPopped,      // A job was popped from a queue.
        eJobEvent_JobStart,       // A job is about to start.
        eJobEvent_JobDone,        // A job just completed.
        eJobEvent_JobRun,         // A job has been completed.
        eJobEvent_JobRunAssisted, // A job has been completed through outside assistance.
        eJobEvent_JobStolen,      // A worker has stolen a job from another worker.
        eJobEvent_WorkerAwoken,   // A worker has been awoken.
        eJobEvent_WorkerUsed,     // A worker has been utilized.
    };

    typedef std::function<void(const JobQueueEntry &job, EJobEvent, uint64_t, size_t)> JobEventObserver; // Delegate definition for job event observation.

    typedef std::deque<JobQueueEntry> JobQueue; // Data structure to represent job queue.

    /**
     * High-res clock based on windows performance counter. Supports STL chrono interfaces.
     */
    using TimePoint = std::chrono::high_resolution_clock::time_point;
    TimePoint ProfileClockNow()
    {
        return std::chrono::high_resolution_clock::now();
    }

    /**
     * Tracking each job's start/end times in a per-worker timeline, for debugging/profiling.
     */
    class ProfilingTimeline
    {
    public:
        struct TimelineEntry
        {
            uint64_t jobId;  // ID of the job that generated this timeline entry.
            TimePoint start; // Job start time.
            TimePoint end;   // Job end time.
            char debugChar;  // Job's debug character for profiling display.

            std::string description; // Timeline entry description.
        };

        typedef std::vector<TimelineEntry> TimelineEntries;

        TimelineEntries m_entries; //< List of timeline entries for this thread.
    };

    /**
     * Represents a worker thread.
     * - Owns a job queue
     * - Implements work-stealing from other workers
     */
    class JobSystemWorker
    {
        friend class JobManager;

    public:
        JobSystemWorker(const JobWorkerDescriptor &desc, const JobEventObserver &eventObserver)
            : m_allWorkers(nullptr), m_workerCount(0), m_workerIndex(0), m_desc(desc), m_eventObserver(eventObserver), m_hasShutDown(false)
        {
        }

        void Start(size_t index, JobSystemWorker **allWorkers, size_t workerCount)
        {
            m_allWorkers = allWorkers;
            m_workerCount = workerCount;
            m_workerIndex = index;

            m_thread = std::thread(&JobSystemWorker::WorkerThreadProc, this);
        }

        void Shutdown()
        {
            m_stop.store(true, std::memory_order_relaxed);

            while (!m_hasShutDown.load(std::memory_order_acquire))
            {
                s_signalThreads.notify_all();

                std::this_thread::sleep_for(std::chrono::microseconds(100));
            }

            if (m_hasShutDown.load(std::memory_order_acquire))
            {
                m_thread.join();
            }
        }

        JobStatePtr PushJob(JobDelegate delegate)
        {
            JobQueueEntry entry = {delegate, std::make_shared<JobState>()};
            entry.m_state->SetQueued();

            {
                std::lock_guard<std::mutex> queueLock(m_queueLock);
                m_queue.insert(m_queue.begin(), entry);
            }

            return entry.m_state;
        }

    private:
        void NotifyEventObserver(const JobQueueEntry &job, EJobEvent event, uint64_t workerIndex, size_t jobId = 0)
        {
#ifdef JOBSYSTEM_ENABLE_PROFILING

            if (m_eventObserver)
            {
                m_eventObserver(job, event, workerIndex, jobId);
            }

#endif // JOBSYSTEM_ENABLE_PROFILING
        }

        bool PopJobFromQueue(JobQueue &queue, JobQueueEntry &job, bool &hasUnsatisfiedDependencies, affinity_t workerAffinity)
        {
            for (auto jobIter = queue.begin(); jobIter != queue.end();)
            {
                const JobQueueEntry &candidate = (*jobIter);

                if ((workerAffinity & candidate.m_state->m_workerAffinity) != 0)
                {
                    if (candidate.m_state->AwaitingCancellation())
                    {
                        candidate.m_state->SetDone();
                        jobIter = queue.erase(jobIter);

                        continue;
                    }
                    else if (candidate.m_state->AreDependenciesMet())
                    {
                        job = candidate;
                        queue.erase(jobIter);

                        NotifyEventObserver(job, eJobEvent_JobPopped, m_workerIndex);

                        return true;
                    }
                }

                hasUnsatisfiedDependencies = true;
                ++jobIter;
            }

            return false;
        }

        bool PopNextJob(JobQueueEntry &job, bool &hasUnsatisfiedDependencies, bool useWorkStealing, affinity_t workerAffinity)
        {
            bool foundJob = false;

            {
                std::lock_guard<std::mutex> queueLock(m_queueLock);
                foundJob = PopJobFromQueue(m_queue, job, hasUnsatisfiedDependencies, workerAffinity);
            }

            if (!foundJob && useWorkStealing)
            {
                for (size_t i = 0; foundJob == false && i < m_workerCount; ++i)
                {
                    JOBSYSTEM_ASSERT(m_allWorkers[i]);
                    JobSystemWorker &worker = *m_allWorkers[i];

                    {
                        std::lock_guard<std::mutex> queueLock(worker.m_queueLock);
                        foundJob = PopJobFromQueue(worker.m_queue, job, hasUnsatisfiedDependencies, workerAffinity);
                    }
                }

                if (foundJob)
                {
                    NotifyEventObserver(job, eJobEvent_JobStolen, m_workerIndex);
                }
            }

            return foundJob;
        }

        void SetThreadName(const char *name)
        {
            (void)name;

            pthread_setname_np(pthread_self(), name);
        }

        void WorkerThreadProc()
        {
            SetThreadName(m_desc.m_name.c_str());

            cpu_set_t cpuset;
            CPU_ZERO(&cpuset);
            for (size_t i = 0; i < sizeof(m_desc.m_cpuAffinity) * 8; ++i)
            {
                if ((1 << i) & m_desc.m_cpuAffinity)
                {
                    CPU_SET(i, &cpuset);
                }
            }

            const affinity_t workerAffinity = CalculateSafeWorkerAffinity(m_workerIndex, m_workerCount);

            while (true)
            {
                JobQueueEntry job;
                {
                    std::unique_lock<std::mutex> signalLock(s_signalLock);

                    bool hasUnsatisfiedDependencies;

                    while (!m_stop.load(std::memory_order_relaxed) &&
                           !PopNextJob(job, hasUnsatisfiedDependencies, m_desc.m_enableWorkStealing, workerAffinity))
                    {
                        s_signalThreads.wait(signalLock);
                        NotifyEventObserver(job, eJobEvent_WorkerAwoken, m_workerIndex);
                    }
                }

                if (m_stop.load(std::memory_order_relaxed))
                {
                    m_hasShutDown.store(true, std::memory_order_release);

                    break;
                }

                s_activeWorkers.fetch_add(1, std::memory_order_acq_rel);
                {
                    NotifyEventObserver(job, eJobEvent_WorkerUsed, m_workerIndex);

                    NotifyEventObserver(job, eJobEvent_JobStart, m_workerIndex, job.m_state->m_jobId);
                    job.m_delegate();
                    NotifyEventObserver(job, eJobEvent_JobDone, m_workerIndex);

                    job.m_state->SetDone();

                    NotifyEventObserver(job, eJobEvent_JobRun, m_workerIndex);

                    s_signalThreads.notify_one();
                }
                s_activeWorkers.fetch_sub(1, std::memory_order_acq_rel);
            }
        }

        std::thread m_thread;            // Thread instance for worker.
        std::atomic<bool> m_stop;        // Has a stop been requested?
        std::atomic<bool> m_hasShutDown; // Has the worker completed shutting down?

        mutable std::mutex m_queueLock; // Mutex to guard worker queue.
        JobQueue m_queue;               // Queue containing requested jobs.

        JobSystemWorker **m_allWorkers; // Pointer to array of all workers, for queue-sharing / work-stealing.
        size_t m_workerCount;           // Number of total workers (size of m_allWorkers array).
        size_t m_workerIndex;           // This worker's index within m_allWorkers.

        JobEventObserver m_eventObserver; // Observer of job-related events occurring on this worker.
        JobWorkerDescriptor m_desc;       // Descriptor/configuration of this worker.
    };

    /**
     * Descriptor for configuring the job manager.
     * - Contains descriptor for each worker
     */
    struct JobManagerDescriptor
    {
        std::vector<JobWorkerDescriptor> m_workers; // Configurations for all workers that should be spawned by JobManager.
    };

    /**
     * Manages job workers, and acts as the primary interface to the job queue.
     */
    class JobManager
    {
    private:
        void Observer(const JobQueueEntry &job, EJobEvent event, uint64_t workerIndex, size_t jobId = 0)
        {
#ifdef JOBSYSTEM_ENABLE_PROFILING
            switch (event)
            {
            case eJobEvent_JobRun:
            {
                ++m_jobsRun;
            }
            break;

            case eJobEvent_JobStolen:
            {
                ++m_jobsStolen;
            }
            break;

            case eJobEvent_JobRunAssisted:
            {
                ++m_jobsAssisted;
                ++m_jobsRun;
            }
            break;

            case eJobEvent_WorkerAwoken:
            {
                m_awokenMask |= GetBit(workerIndex);
            }
            break;

            case eJobEvent_WorkerUsed:
            {
                m_usedMask |= GetBit(workerIndex);
            }
            break;

            case eJobEvent_JobStart:
            {
                ProfilingTimeline &timeline = workerIndex < m_workers.size() ? m_timelines[workerIndex] : m_timelines[m_workers.size()];
                ProfilingTimeline::TimelineEntry entry;
                entry.jobId = jobId;
                entry.start = ProfileClockNow();
                entry.debugChar = job.m_state ? job.m_state->m_debugChar : 0;
                timeline.m_entries.push_back(entry);
            }
            break;

            case eJobEvent_JobDone:
            {
                ProfilingTimeline &timeline = workerIndex < m_workers.size() ? m_timelines[workerIndex] : m_timelines[m_workers.size()];
                ProfilingTimeline::TimelineEntry &entry = timeline.m_entries.back();
                entry.end = ProfileClockNow();
            }
            break;

            case eJobEvent_JobPopped:
            {
                if (!m_hasPushedJob)
                {
                    m_firstJobTime = ProfileClockNow();
                    m_hasPushedJob = true;
                }
            }
            break;
            }
#endif // JOBSYSTEM_ENABLE_PROFILING
        }

    public:
        JobManager()
            : m_jobsRun(0), m_jobsStolen(0), m_usedMask(0), m_awokenMask(0), m_nextRoundRobinWorkerIndex(0), m_timelines(nullptr), m_firstJobTime()
        {
        }

        ~JobManager()
        {
            DumpProfilingResults();

            JoinWorkersAndShutdown();
        }

        bool Create(const JobManagerDescriptor &desc)
        {
            JoinWorkersAndShutdown();

            m_desc = desc;

            const size_t workerCount = desc.m_workers.size();
            m_workers.reserve(workerCount);

#ifdef JOBSYSTEM_ENABLE_PROFILING

            m_timelines = new ProfilingTimeline[workerCount + 1];
            m_hasPushedJob = false;

#endif // JOBSYSTEM_ENABLE_PROFILING

            const JobEventObserver observer = std::bind(
                &JobManager::Observer, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4);

            // Create workers. We don't spawn the threads yet.
            for (size_t i = 0; i < workerCount; ++i)
            {
                const JobWorkerDescriptor &workerDesc = desc.m_workers[i];

                JobSystemWorker *worker = new JobSystemWorker(workerDesc, observer);
                m_workers.push_back(worker);
            }

            // Start the workers (includes spawning threads). Each worker maintains
            // understanding of what other workers exist, for work-stealing purposes.
            for (size_t i = 0; i < workerCount; ++i)
            {
                m_workers[i]->Start(i, &m_workers[0], workerCount);
            }

            return !m_workers.empty();
        }

        JobStatePtr AddJob(JobDelegate delegate, char debugChar = 0)
        {
            JobStatePtr state = nullptr;

            // \todo - workers should maintain a tls pointer to themselves, so we can push
            // directly into our own queue.

            if (!m_workers.empty())
            {
                // Add round-robin style. Note that work-stealing helps load-balance,
                // if it hasn't been disabled. If it has we may need to consider a
                // smarter scheme here.
                state = m_workers[m_nextRoundRobinWorkerIndex]->PushJob(delegate);
                state->m_debugChar = debugChar;

                m_nextRoundRobinWorkerIndex = (m_nextRoundRobinWorkerIndex + 1) % m_workers.size();
            }

            return state;
        }

        void AssistUntilJobDone(JobStatePtr state)
        {
            JOBSYSTEM_ASSERT(state->m_ready.load(std::memory_order_acquire));

            const affinity_t workerAffinity = kAffinityAllBits;

            // Steal jobs from workers until the specified job is done.
            while (!state->IsDone())
            {
                JOBSYSTEM_ASSERT(!m_workers.empty());

                JobQueueEntry job;
                bool hasUnsatisfiedDependencies;

                if (m_workers[0]->PopNextJob(job, hasUnsatisfiedDependencies, true, workerAffinity))
                {
                    Observer(job, eJobEvent_JobStart, m_workers.size(), job.m_state->m_jobId);
                    job.m_delegate();
                    Observer(job, eJobEvent_JobDone, m_workers.size());

                    job.m_state->SetDone();

                    Observer(job, eJobEvent_JobRunAssisted, 0);

                    s_signalThreads.notify_one();
                }
            }
        }

        void AssistUntilDone()
        {
            JOBSYSTEM_ASSERT(!m_workers.empty());

            // Steal and run jobs from workers until all queues are exhausted.

            const affinity_t workerAffinity = kAffinityAllBits;

            JobQueueEntry job;
            bool foundBusyWorker = true;

            while (foundBusyWorker)
            {
                foundBusyWorker = false;

                for (JobSystemWorker *worker : m_workers)
                {
                    if (worker->PopNextJob(job, foundBusyWorker, false, workerAffinity))
                    {
                        Observer(job, eJobEvent_JobStart, m_workers.size(), job.m_state->m_jobId);
                        job.m_delegate();
                        Observer(job, eJobEvent_JobDone, m_workers.size());

                        job.m_state->SetDone();

                        Observer(job, eJobEvent_JobRunAssisted, 0);

                        foundBusyWorker = true;
                        s_signalThreads.notify_one();
                        break;
                    }
                }
            }

            for (JobSystemWorker *worker : m_workers)
            {
                if (!worker->m_queue.empty())
                {
                    JOBSYSTEM_ASSERT(0);
                }
            }
        }

        void JoinWorkersAndShutdown(bool finishJobs = false)
        {
            if (finishJobs)
            {
                AssistUntilDone();
            }

            // Tear down each worker. Un-popped jobs may still reside in the queues at this point
            // if finishJobs = false.
            // Don't destruct workers yet, in case someone's in the process of work-stealing.
            for (size_t i = 0, n = m_workers.size(); i < n; ++i)
            {
                JOBSYSTEM_ASSERT(m_workers[i]);
                m_workers[i]->Shutdown();
            }

            // Destruct all workers.
            std::for_each(m_workers.begin(), m_workers.end(), [](JobSystemWorker *worker)
                          { delete worker; });
            m_workers.clear();

#ifdef JOBSYSTEM_ENABLE_PROFILING

            delete[] m_timelines;
            m_timelines = nullptr;

#endif // JOBSYSTEM_ENABLE_PROFILING
        }

    private:
        size_t m_nextRoundRobinWorkerIndex; // Index of the worker to receive the next requested job, round-robin style.

        std::atomic<unsigned int> m_jobsRun;      // Counter to track # of jobs run.
        std::atomic<unsigned int> m_jobsAssisted; // Counter to track # of jobs run via external Assist*().
        std::atomic<unsigned int> m_jobsStolen;   // Counter to track # of jobs stolen from another worker's queue.
        std::atomic<affinity_t> m_usedMask;       // Mask with bits set according to the IDs of the jobs that have executed jobs.
        std::atomic<affinity_t> m_awokenMask;     // Mask with bits set according to the IDs of the jobs that have been awoken at least once.

    private:
        JobManagerDescriptor m_desc; // Descriptor/configuration of the job manager.

        bool m_hasPushedJob;            // For profiling - has a job been pushed yet?
        TimePoint m_firstJobTime;       // For profiling - when was the first job pushed?
        ProfilingTimeline *m_timelines; // For profiling - a ProfilingTimeline entry for each worker, plus an additional entry to represent the Assist thread.

        std::vector<JobSystemWorker *> m_workers; // Storage for worker instances.

        void DumpProfilingResults()
        {
#ifdef JOBSYSTEM_ENABLE_PROFILING

            AssistUntilDone();

            auto now = ProfileClockNow();
            auto totalNS = std::chrono::duration_cast<std::chrono::nanoseconds>(now - m_firstJobTime).count();

            std::this_thread::sleep_for(std::chrono::milliseconds(10));

            const size_t workerCount = m_workers.size();

            printf(
                "\n[Job System Statistics]\n"
                "Jobs Run:       %8d\n" // May be < jobs submitted
                "Jobs Stolen:    %8d\n"
                "Jobs Assisted:  %8d\n"
                "Workers Used:   %8lu\n"
                "Workers Awoken: %8lu\n",
                m_jobsRun.load(std::memory_order_acquire),
                m_jobsStolen.load(std::memory_order_acquire),
                m_jobsAssisted.load(std::memory_order_acquire),
                CountBits(m_usedMask.load(std::memory_order_acquire)),
                CountBits(m_awokenMask.load(std::memory_order_acquire)));

            printf("\n[Worker Profiling Results]\n%.3f total ms\n\nTimeline (approximated):\n\n", double(totalNS) / 1000000);

            const char *busySymbols = "abcdefghijklmn";
            const size_t busySymbolCount = strlen(busySymbols);

            for (size_t workerIndex = 0; workerIndex < workerCount + 1; ++workerIndex)
            {
                ProfilingTimeline &timeline = m_timelines[workerIndex];

                const char *name = (workerIndex < workerCount) ? m_workers[workerIndex]->m_desc.m_name.c_str() : "[Assist]";

                const size_t bufferSize = 200;
                char buffer[bufferSize];
                snprintf(buffer, sizeof(buffer), "%20s: ", name);

                const size_t nameLen = strlen(buffer);
                const size_t remaining = bufferSize - nameLen - 2;

                for (size_t nameIdx = nameLen; nameIdx < bufferSize - 2; ++nameIdx)
                {
                    buffer[nameIdx] = '-';
                }

                buffer[bufferSize - 2] = '\n';
                buffer[bufferSize - 1] = 0;

                for (ProfilingTimeline::TimelineEntry &entry : timeline.m_entries)
                {
                    const auto startNs = std::chrono::duration_cast<std::chrono::nanoseconds>(entry.start - m_firstJobTime).count();
                    const auto endNs = std::chrono::duration_cast<std::chrono::nanoseconds>(entry.end - m_firstJobTime).count();

                    const double startPercent = (double(startNs) / double(totalNS));
                    const double endPercent = (double(endNs) / double(totalNS));

                    const char jobCharacter = (entry.debugChar != 0) ? entry.debugChar : busySymbols[entry.jobId % busySymbolCount];

                    const size_t startIndex = nameLen + std::min<size_t>(remaining - 1, size_t(startPercent * double(remaining)));
                    size_t endIndex = nameLen + std::min<size_t>(remaining - 1, size_t(endPercent * double(remaining)));

                    size_t shift = 0;

                    while (buffer[startIndex + shift] != '-' && startIndex + shift < bufferSize - 3 && endIndex + shift < bufferSize - 3)
                    {
                        ++shift;
                    }

                    endIndex -= std::min<size_t>(endIndex - startIndex, size_t(shift));

                    for (size_t i = startIndex + shift; i <= endIndex + shift; ++i)
                    {
                        JOBSYSTEM_ASSERT(i < bufferSize - 2);
                        buffer[i] = jobCharacter;
                    }
                }

                printf("%s", buffer);
            }

            printf("\n");

#endif // JOBSYSTEM_ENABLE_PROFILING
        }
    };

    /**
     * Helper for building complex job/dependency chains logically.
     *
     * e.g.
     *
     * jobsystem::JobManager jobManager;
     * ...
     * jobsystem::JobChainBuilder<128>(jobManager)
     *  .Do(something, 'a')
     *  .Then()
     *  .Do(somethingAfterThat, 'b')
     *  .Then()
     *  .Together()
     *      .Do(parallelThing1, 'c')
     *      .Do(parallelThing2, 'd')
     *      .Do(parallelThing3, 'e')
     *  .Close()
     *  .Then()
     *  .Do(finalThing, 'F')
     *  .Go()
     * .WaitForAll();
     **                                    --- parallelThing1 ---
     *                                   /                       \
     *  something -> somethingAfterThat -> --- parallelThing2 ---- -> finalThing
     *                                   \                       /
     *                                     --- parallelThing3 ---
     * etc...
     *
     */
    template <size_t MaxJobNodes = 256>
    class JobChainBuilder
    {
    public:
        struct Node
        {
            Node() : isGroup(false), groupDependency(nullptr) {}
            ~Node() {}

            Node *groupDependency;
            JobStatePtr job;
            bool isGroup;
        };

        Node *AllocNode()
        {
            if (m_nextNodeIndex >= MaxJobNodes)
                return nullptr;

            Node *node = &m_nodePool[m_nextNodeIndex++];
            *node = Node();

            return node;
        }

        JobChainBuilder(JobManager &manager)
            : mgr(manager)
        {
            Reset();

            // Push a sentinel (root) node.
            m_stack.push_back(AllocNode());
        }

        void Reset()
        {
            m_allJobs.clear();
            m_stack.clear();

            m_last = nullptr;
            m_dependency = nullptr;
            m_nextNodeIndex = 0;
            m_failed = false;
        }

        JobChainBuilder &Together(char debugChar = 0)
        {
            if (Node *item = AllocNode())
            {
                item->isGroup = true;
                item->groupDependency = m_dependency;

                item->job = mgr.AddJob([]() {}, debugChar);

                m_allJobs.push_back(item->job);

                m_last = item;
                m_dependency = nullptr;

                m_stack.push_back(item);
            }
            else
            {
                Fail();
            }

            return *this;
        }

        JobChainBuilder &Do(JobDelegate delegate, char debugChar = 0)
        {
            Node *owner = m_stack.back();

            if (Node *item = AllocNode())
            {
                item->job = mgr.AddJob(delegate, debugChar);

                m_allJobs.push_back(item->job);

                if (m_dependency)
                {
                    m_dependency->job->AddDependant(item->job);
                    m_dependency = nullptr;
                }

                if (owner && owner->isGroup)
                {
                    item->job->AddDependant(owner->job);

                    if (owner->groupDependency)
                    {
                        owner->groupDependency->job->AddDependant(item->job);
                    }
                }

                m_last = item;
            }
            else
            {
                Fail();
            }

            return *this;
        }

        JobChainBuilder &Then()
        {
            m_dependency = m_last;
            m_last = (m_dependency) ? m_dependency->groupDependency : nullptr;

            return *this;
        }

        JobChainBuilder &Close()
        {
            if (!m_stack.empty())
            {
                Node *owner = m_stack.back();
                if (owner->isGroup)
                {
                    m_last = owner;
                }
            }

            m_dependency = nullptr;

            if (m_stack.size() > 1)
            {
                m_stack.pop_back();
            }

            return *this;
        }

        JobChainBuilder &Go()
        {
            if (m_allJobs.empty())
            {
                return *this;
            }

            Then();
            Do([]() {}, 'J');
            m_joinJob = m_allJobs.back();

            for (JobStatePtr &job : m_allJobs)
            {
                job->SetReady();
            }

            return *this;
        }

        void Fail()
        {
            for (JobStatePtr &job : m_allJobs)
            {
                job->Cancel();
            }

            m_allJobs.clear();
            m_failed = true;
        }

        bool Failed() const
        {
            return m_failed;
        }

        void WaitForAll()
        {
            if (m_joinJob)
            {
                m_joinJob->Wait();
            }
        }

        void AssistAndWaitForAll()
        {
            mgr.AssistUntilJobDone(m_joinJob);
        }

        JobManager &mgr; // Job manager to submit jobs to.

        Node m_nodePool[MaxJobNodes]; // Pool of chain nodes (on the stack). The only necessary output of this system is jobs. Nodes are purely internal.
        size_t m_nextNodeIndex;       // Next free item in the pool.

        std::vector<Node *> m_stack;        // Internal stack to track groupings.
        std::vector<JobStatePtr> m_allJobs; // All jobs created by the builder, to be readied on completion.

        Node *m_last;       // Last job to be pushed, to handle setting up dependencies after Then() calls.
        Node *m_dependency; // Any job promoted to a dependency for the next job, as dicated by Then().

        JobStatePtr m_joinJob; // Final join job that callers can wait on to complete the batch.

        bool m_failed; // Did an error occur during creation of the DAG?
    };

} // namespace jobsystem