JobQueue.h

Go to the documentation of this file.

// Copyright © 2008-2023 Pioneer Developers. See AUTHORS.txt for details
// Licensed under the terms of the GPL v3. See licenses/GPL-3.txt
 
#ifndef JOBQUEUE_H
#define JOBQUEUE_H
 
#include "SDL_thread.h"
#include "core/TaskGraph.h"
#include <atomic>
#include <cassert>
#include <deque>
#include <set>
#include <string>
#include <vector>
 
static const uint32_t MAX_THREADS = 64;
 
class JobClient;
class JobQueue;
 
// represents a single unit of work that you want done
// subclass and implement:
//
// OnRun: called from worker thread, and does the actual stuff you want done.
//        store all your data in the object itself.
//
// OnFinish: called from the main thread once the worker completes the job.
//           this is where you deliver the results from the worker
//
// OnCancel: optional. called from the main thread to tell the job that its
//           results are not wanted. it should arrange for OnRun to return
//           as quickly as possible. OnFinish will not be called for the job
class Job {
public:
    // This is the RAII handle for a queued Job. A job is cancelled when the
    // Job::Handle is destroyed. There is at most one Job::Handle for each Job
    // (non-queued Jobs have no handle). Job::Handle is not copyable only
    // moveable.
    class Handle {
    public:
        Handle() :
            m_id(++s_nextId),
            m_job(nullptr),
            m_queue(nullptr),
            m_client(nullptr) {}
        Handle(Handle &&other);
        Handle &operator=(Handle &&other);
        ~Handle();
 
        Handle(const Handle &) = delete;
        Handle &operator=(const Handle &) = delete;
 
        bool HasJob() const { return m_job != nullptr; }
        Job *GetJob() const { return m_job; }
 
        bool operator<(const Handle &other) const { return m_id < other.m_id; }
 
    private:
        friend class Job;
        friend class AsyncJobQueue;
        friend class SyncJobQueue;
        friend class TaskGraphJobQueueImpl;
 
        Handle(Job *job, JobQueue *queue, JobClient *client);
        void Unlink();
 
        static Uint64 s_nextId;
 
        Uint64 m_id;
        Job *m_job;
        JobQueue *m_queue;
        JobClient *m_client;
    };
 
public:
    Job() :
        cancelled(false),
        m_handle(nullptr) {}
    virtual ~Job();
 
    Job(const Job &) = delete;
    Job &operator=(const Job &) = delete;
 
    virtual void OnRun() = 0;
    virtual void OnFinish() = 0;
    virtual void OnCancel() {}
 
private:
    friend class AsyncJobQueue;
    friend class SyncJobQueue;
    friend class JobRunner;
 
    // TaskGraph stuff
    friend class TaskGraph;
    friend class TaskGraphJobQueueImpl;
 
    void UnlinkHandle();
    const Handle *GetHandle() const { return m_handle; }
    void SetHandle(Handle *handle) { m_handle.store(handle, std::memory_order_release); }
    void ClearHandle() { m_handle = nullptr; }
 
    std::atomic<bool> cancelled;
    std::atomic<Handle *> m_handle;
};
 
// the queue management class. create one from the main thread, and feed your
// jobs do it. it will take care of the rest
class JobQueue {
public:
    JobQueue() = default;
    JobQueue(const JobQueue &) = delete;
    JobQueue &operator=(const JobQueue &) = delete;
    virtual ~JobQueue() {}
 
    // call from the main thread to add a job to the queue. the job should be
    // allocated with new. the queue will delete it once its its completed
    virtual Job::Handle Queue(Job *job, JobClient *client = nullptr) = 0;
 
    // Call from the main thread to cancel a job.
    // The job will not be run if it is not already executing, and OnFinished
    // will not be called for the job. OnCancel will be called for the job
    // when it is eventually processed in a FinishJobs() call
    virtual void Cancel(Job *job) = 0;
 
    // call from the main loop. this will call OnFinish for any finished jobs,
    // and then delete all finished and cancelled jobs. returns the number of
    // finished jobs (not cancelled)
    virtual Uint32 FinishJobs() = 0;
};
 
class SyncJobQueue : public JobQueue {
public:
    SyncJobQueue() = default;
    virtual ~SyncJobQueue();
 
    virtual Job::Handle Queue(Job *job, JobClient *client = nullptr) override;
    virtual void Cancel(Job *job) override;
    virtual Uint32 FinishJobs() override;
 
    Uint32 RunJobs(Uint32 count = 1);
 
private:
    std::deque<Job *> m_queue;
    std::deque<Job *> m_finished;
};
 
// JobClient is an abstraction to allow transparent management of job handles
class JobClient {
public:
    virtual void Order(Job *job) = 0;
    virtual void RemoveJob(Job::Handle *handle) = 0;
    virtual ~JobClient() {}
};
 
// JobSet provides an interface for "fire and forget" jobs - call Order with your job,
// and JobSet will keep the handle alive until the job has finished.
class JobSet : public JobClient {
public:
    JobSet(JobQueue *queue) :
        m_queue(queue) {}
    JobSet(JobSet &&other) :
        m_queue(other.m_queue),
        m_jobs(std::move(other.m_jobs)) { other.m_queue = nullptr; }
    JobSet &operator=(JobSet &&other)
    {
        m_queue = other.m_queue;
        m_jobs = std::move(other.m_jobs);
        other.m_queue = nullptr;
        return *this;
    }
 
    JobSet(const JobSet &) = delete;
    JobSet &operator=(const JobSet &other) = delete;
 
    virtual void Order(Job *job)
    {
        auto x = m_jobs.insert(m_queue->Queue(job, this));
        (void)x; // suppress unused variable warning
        assert(x.second);
    }
    virtual void RemoveJob(Job::Handle *handle) { m_jobs.erase(*handle); }
 
    bool IsEmpty() const { return m_jobs.empty(); }
 
private:
    JobQueue *m_queue;
    std::set<Job::Handle> m_jobs;
};
 
#endif