xbmc/utils/JobManager.h

   1 /*
   2  *  Copyright (C) 2005-2018 Team Kodi
   3  *  This file is part of Kodi - https://kodi.tv
   4  *
   5  *  SPDX-License-Identifier: GPL-2.0-or-later
   6  *  See LICENSES/README.md for more information.
   7  */
   8
   9 #pragma once
  10
  11 #include "Job.h"
  12 #include "threads/CriticalSection.h"
  13 #include "threads/Thread.h"
  14
  15 #include <queue>
  16 #include <string>
  17 #include <vector>
  18
  19 class CJobManager;
  20
  21 class CJobWorker : public CThread
  22 {
  23 public:
  24   explicit CJobWorker(CJobManager *manager);
  25   ~CJobWorker() override;
  26
  27   void Process() override;
  28 private:
  29   CJobManager  *m_jobManager;
  30 };
  31
  32 template<typename F>
  33 class CLambdaJob : public CJob
  34 {
  35 public:
  36   CLambdaJob(F&& f) : m_f(std::forward<F>(f)) {}
  37   bool DoWork() override
  38   {
  39     m_f();
  40     return true;
  41   }
  42   bool operator==(const CJob *job) const override
  43   {
  44     return this == job;
  45   };
  46 private:
  47   F m_f;
  48 };
  49
  50 /*!
  51  \ingroup jobs
  52  \brief Job Queue class to handle a queue of unique jobs to be processed sequentially
  53
  54  Holds a queue of jobs to be processed sequentially, either first in,first out
  55  or last in, first out.  Jobs are unique, so queueing multiple copies of the same job
  56  (based on the CJob::operator==) will not add additional jobs.
  57
  58  Classes should subclass this class and override OnJobCallback should they require
  59  information from the job.
  60
  61  \sa CJob and IJobCallback
  62  */
  63 class CJobQueue: public IJobCallback
  64 {
  65   class CJobPointer
  66   {
  67   public:
  68     explicit CJobPointer(CJob *job)
  69     {
  70       m_job = job;
  71       m_id = 0;
  72     };
  73     void CancelJob();
  74     void FreeJob()
  75     {
  76       delete m_job;
  77       m_job = NULL;
  78     };
  79     bool operator==(const CJob *job) const
  80     {
  81       if (m_job)
  82         return *m_job == job;
  83       return false;
  84     };
  85     CJob *m_job;
  86     unsigned int m_id;
  87   };
  88 public:
  89   /*!
  90    \brief CJobQueue constructor
  91    \param lifo whether the queue should be processed last in first out or first in first out.  Defaults to false (first in first out)
  92    \param jobsAtOnce number of jobs at once to process.  Defaults to 1.
  93    \param priority priority of this queue.
  94    \sa CJob
  95    */
  96   CJobQueue(bool lifo = false, unsigned int jobsAtOnce = 1, CJob::PRIORITY priority = CJob::PRIORITY_LOW);
  97
  98   /*!
  99    \brief CJobQueue destructor
 100    Cancels any in-process jobs, and destroys the job queue.
 101    \sa CJob
 102    */
 103   ~CJobQueue() override;
 104
 105   /*!
 106    \brief Add a job to the queue
 107    On completion of the job, destruction of the job queue or in case the job could not be added successfully, the CJob object will be destroyed.
 108    \param job a pointer to the job to add. The job should be subclassed from CJob.
 109    \return True if the job was added successfully, false otherwise.
 110    In case of failure, the passed CJob object will be deleted before returning from this method.
 111    \sa CJob
 112    */
 113   bool AddJob(CJob *job);
 114
 115   /*!
 116    \brief Add a function f to this job queue
 117    */
 118   template<typename F>
 119   void Submit(F&& f)
 120   {
 121     AddJob(new CLambdaJob<F>(std::forward<F>(f)));
 122   }
 123
 124   /*!
 125    \brief Cancel a job in the queue
 126    Cancels a job in the queue. Any job currently being processed may complete after this
 127    call has completed, but OnJobComplete will not be performed. If the job is only queued
 128    then it will be removed from the queue and deleted.
 129    \param job a pointer to the job to cancel. The job should be subclassed from CJob.
 130    \sa CJob
 131    */
 132   void CancelJob(const CJob *job);
 133
 134   /*!
 135    \brief Cancel all jobs in the queue
 136    Removes all jobs from the queue. Any job currently being processed may complete after this
 137    call has completed, but OnJobComplete will not be performed.
 138    \sa CJob
 139    */
 140   void CancelJobs();
 141
 142   /*!
 143    \brief Check whether the queue is processing a job
 144    */
 145   bool IsProcessing() const;
 146
 147   /*!
 148    \brief The callback used when a job completes.
 149
 150    CJobQueue implementation will cleanup the internal processing queue and then queue the next
 151    job at the job manager, if any.
 152
 153    \param jobID the unique id of the job (as retrieved from CJobManager::AddJob)
 154    \param success the result from the DoWork call
 155    \param job the job that has been processed.
 156    \sa CJobManager, IJobCallback and CJob
 157    */
 158   void OnJobComplete(unsigned int jobID, bool success, CJob *job) override;
 159
 160   /*!
 161    \brief The callback used when a job will be aborted.
 162
 163    CJobQueue implementation will cleanup the internal processing queue and then queue the next
 164    job at the job manager, if any.
 165
 166    \param jobID the unique id of the job (as retrieved from CJobManager::AddJob)
 167    \param job the job that has been aborted.
 168    \sa CJobManager, IJobCallback and CJob
 169    */
 170   void OnJobAbort(unsigned int jobID, CJob* job) override;
 171
 172 protected:
 173   /*!
 174    \brief Returns if we still have jobs waiting to be processed
 175    NOTE: This function does not take into account the jobs that are currently processing
 176    */
 177   bool QueueEmpty() const;
 178
 179 private:
 180   void OnJobNotify(CJob* job);
 181   void QueueNextJob();
 182
 183   typedef std::deque<CJobPointer> Queue;
 184   typedef std::vector<CJobPointer> Processing;
 185   Queue m_jobQueue;
 186   Processing m_processing;
 187
 188   unsigned int m_jobsAtOnce;
 189   CJob::PRIORITY m_priority;
 190   mutable CCriticalSection m_section;
 191   bool m_lifo;
 192 };
 193
 194 /*!
 195  \ingroup jobs
 196  \brief Job Manager class for scheduling asynchronous jobs.
 197
 198  Controls asynchronous job execution, by allowing clients to add and cancel jobs.
 199  Should be accessed via CServiceBroker::GetJobManager().  Jobs are allocated based
 200  on priority levels.  Lower priority jobs are executed only if there are sufficient
 201  spare worker threads free to allow for higher priority jobs that may arise.
 202
 203  \sa CJob and IJobCallback
 204  */
 205 class CJobManager final
 206 {
 207   class CWorkItem
 208   {
 209   public:
 210     CWorkItem(CJob *job, unsigned int id, CJob::PRIORITY priority, IJobCallback *callback)
 211     {
 212       m_job = job;
 213       m_id = id;
 214       m_callback = callback;
 215       m_priority = priority;
 216     }
 217     bool operator==(unsigned int jobID) const
 218     {
 219       return m_id == jobID;
 220     };
 221     bool operator==(const CJob *job) const
 222     {
 223       return m_job == job;
 224     };
 225     void FreeJob()
 226     {
 227       delete m_job;
 228       m_job = NULL;
 229     };
 230     void Cancel()
 231     {
 232       m_callback = NULL;
 233     };
 234     CJob         *m_job;
 235     unsigned int  m_id;
 236     IJobCallback *m_callback;
 237     CJob::PRIORITY m_priority;
 238   };
 239
 240 public:
 241   CJobManager();
 242
 243   /*!
 244    \brief Add a job to the threaded job manager.
 245    On completion or abort of the job or in case the job could not be added successfully, the CJob object will be destroyed.
 246    \param job a pointer to the job to add. The job should be subclassed from CJob
 247    \param callback a pointer to an IJobCallback instance to receive job progress and completion notices.
 248    \param priority the priority that this job should run at.
 249    \return On success, a unique identifier for this job, to be used with other interaction, 0 otherwise.
 250    In case of failure, the passed CJob object will be deleted before returning from this method.
 251    \sa CJob, IJobCallback, CancelJob()
 252    */
 253   unsigned int AddJob(CJob *job, IJobCallback *callback, CJob::PRIORITY priority = CJob::PRIORITY_LOW);
 254
 255   /*!
 256    \brief Add a function f to this job manager for asynchronously execution.
 257    */
 258   template<typename F>
 259   void Submit(F&& f, CJob::PRIORITY priority = CJob::PRIORITY_LOW)
 260   {
 261     AddJob(new CLambdaJob<F>(std::forward<F>(f)), nullptr, priority);
 262   }
 263
 264   /*!
 265    \brief Add a function f to this job manager for asynchronously execution.
 266    */
 267   template<typename F>
 268   void Submit(F&& f, IJobCallback *callback, CJob::PRIORITY priority = CJob::PRIORITY_LOW)
 269   {
 270     AddJob(new CLambdaJob<F>(std::forward<F>(f)), callback, priority);
 271   }
 272
 273   /*!
 274    \brief Cancel a job with the given id.
 275    \param jobID the id of the job to cancel, retrieved previously from AddJob()
 276    \sa AddJob()
 277    */
 278   void CancelJob(unsigned int jobID);
 279
 280   /*!
 281    \brief Cancel all remaining jobs, preparing for shutdown
 282    Should be called prior to destroying any objects that may be being used as callbacks
 283    \sa CancelJob(), AddJob()
 284    */
 285   void CancelJobs();
 286
 287   /*!
 288    \brief Re-start accepting jobs again
 289    Called after calling CancelJobs() to allow this manager to accept more jobs
 290    \throws std::logic_error if the manager was not previously cancelled
 291    \sa CancelJobs()
 292    */
 293   void Restart();
 294
 295   /*!
 296    \brief Checks to see if any jobs of a specific type are currently processing.
 297    \param type Job type to search for
 298    \return Number of matching jobs
 299    */
 300   int IsProcessing(const std::string &type) const;
 301
 302   /*!
 303    \brief Suspends queueing of jobs with priority PRIORITY_LOW_PAUSABLE until unpaused
 304    Useful to (for ex) stop queuing thumb jobs during video start/playback.
 305    Does not affect currently processing jobs, use IsProcessing to see if any need to be waited on
 306    \sa UnPauseJobs()
 307    */
 308   void PauseJobs();
 309
 310   /*!
 311    \brief Resumes queueing of (previously paused) jobs with priority PRIORITY_LOW_PAUSABLE
 312    \sa PauseJobs()
 313    */
 314   void UnPauseJobs();
 315
 316   /*!
 317    \brief Checks to see if any jobs with specific priority are currently processing.
 318    \param priority to search for
 319    \return true if processing jobs, else returns false
 320    */
 321   bool IsProcessing(const CJob::PRIORITY &priority) const;
 322
 323 protected:
 324   friend class CJobWorker;
 325   friend class CJob;
 326   friend class CJobQueue;
 327
 328   /*!
 329    \brief Get a new job to process. Blocks until a new job is available, or a timeout has occurred.
 330    \sa CJob
 331    */
 332   CJob* GetNextJob();
 333
 334   /*!
 335    \brief Callback from CJobWorker after a job has completed.
 336    Calls IJobCallback::OnJobComplete(), and then destroys job.
 337    \param job a pointer to the calling subclassed CJob instance.
 338    \param success the result from the DoWork call
 339    \sa IJobCallback, CJob
 340    */
 341   void  OnJobComplete(bool success, CJob *job);
 342
 343   /*!
 344    \brief Callback from CJob to report progress and check for cancellation.
 345    Checks for cancellation, and calls IJobCallback::OnJobProgress().
 346    \param progress amount of processing performed to date, out of total.
 347    \param total total amount of processing.
 348    \param job pointer to the calling subclassed CJob instance.
 349    \return true if the job has been cancelled, else returns false.
 350    \sa IJobCallback, CJob
 351    */
 352   bool  OnJobProgress(unsigned int progress, unsigned int total, const CJob *job) const;
 353
 354 private:
 355   CJobManager(const CJobManager&) = delete;
 356   CJobManager const& operator=(CJobManager const&) = delete;
 357
 358   /*! \brief Pop a job off the job queue and add to the processing queue ready to process
 359    \return the job to process, NULL if no jobs are available
 360    */
 361   CJob *PopJob();
 362
 363   void StartWorkers(CJob::PRIORITY priority);
 364   void RemoveWorker(const CJobWorker *worker);
 365   static unsigned int GetMaxWorkers(CJob::PRIORITY priority);
 366
 367   unsigned int m_jobCounter;
 368
 369   typedef std::deque<CWorkItem>    JobQueue;
 370   typedef std::vector<CWorkItem>   Processing;
 371   typedef std::vector<CJobWorker*> Workers;
 372
 373   JobQueue   m_jobQueue[CJob::PRIORITY_DEDICATED + 1];
 374   bool       m_pauseJobs;
 375   Processing m_processing;
 376   Workers    m_workers;
 377
 378   mutable CCriticalSection m_section;
 379   CEvent           m_jobEvent;
 380   bool             m_running;
 381 };