JobQueue.h

Go to the documentation of this file.

// Copyright © 2008-2021 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 <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;
 
        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;
 
    void UnlinkHandle();
    const Handle *GetHandle() const { return m_handle; }
    void SetHandle(Handle *handle) { m_handle = handle; }
    void ClearHandle() { m_handle = nullptr; }
 
    bool cancelled;
    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;
 
    // numRunners is the number of jobs to run in parallel. right now its the
    // same as the number of threads, but there's no reason that it has to be
    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. one of three things will happen
    //
    // - the job hasn't run yet. it will never be run, and neither OnFinished nor
    //   OnCancel will be called. the job will be deleted on the next call to
    //   FinishJobs
    //
    // - the job has finished. neither onFinished not onCancel will be called.
    //   the job will be deleted on the next call to FinishJobs
    //
    // - the job is running. OnCancel will be called
    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;
};
 
// the queue management class. create one from the main thread, and feed your
// jobs do it. it will take care of the rest
class AsyncJobQueue : public JobQueue {
public:
    // numRunners is the number of jobs to run in parallel. right now its the
    // same as the number of threads, but there's no reason that it has to be
    AsyncJobQueue(Uint32 numRunners);
    virtual ~AsyncJobQueue();
 
    // 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) override;
 
    // call from the main thread to cancel a job. one of three things will happen
    //
    // - the job hasn't run yet. it will never be run, and neither OnFinished nor
    //   OnCancel will be called. the job will be deleted on the next call to
    //   FinishJobs
    //
    // - the job has finished. neither onFinished not onCancel will be called.
    //   the job will be deleted on the next call to FinishJobs
    //
    // - the job is running. OnCancel will be called
    virtual void Cancel(Job *job) override;
 
    // 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() override;
 
private:
    // a runner wraps a single thread, and calls into the queue when its ready for
    // a new job. no user-servicable parts inside!
    class JobRunner {
    public:
        JobRunner(AsyncJobQueue *jq, const uint8_t idx);
        ~JobRunner();
        SDL_mutex *GetQueueDestroyingLock();
        void SetQueueDestroyed();
 
    private:
        static int Trampoline(void *);
        void Main();
 
        AsyncJobQueue *m_jobQueue;
 
        Job *m_job;
        SDL_mutex *m_jobLock;
        SDL_mutex *m_queueDestroyingLock;
 
        SDL_Thread *m_threadId;
 
        uint8_t m_threadIdx;
        std::string m_threadName;
        bool m_queueDestroyed;
    };
 
    Job *GetJob();
    void Finish(Job *job, const uint8_t threadIdx);
 
    std::deque<Job *> m_queue;
    SDL_mutex *m_queueLock;
    SDL_cond *m_queueWaitCond;
 
    std::deque<Job *> m_finished[MAX_THREADS];
    SDL_mutex *m_finishedLock[MAX_THREADS];
 
    std::vector<JobRunner *> m_runners;
    // scratch space for finishing jobs on main thread
    std::vector<Job *> m_finishedScratch;
 
    bool m_shutdown;
};
 
class SyncJobQueue : public JobQueue {
public:
    SyncJobQueue() = default;
    virtual ~SyncJobQueue();
 
    // 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) override;
 
    // call from the main thread to cancel a job. one of three things will happen
    //
    // - the job hasn't run yet. it will never be run, and neither OnFinished nor
    //   OnCancel will be called. the job will be deleted on the next call to
    //   FinishJobs
    //
    // - the job has finished. neither onFinished not onCancel will be called.
    //   the job will be deleted on the next call to FinishJobs
    //
    // - the job is running. OnCancel will be called
    virtual void Cancel(Job *job) override;
 
    // 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() 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