JobManager.h

/*
 *  Copyright (C) 2005-2018 Team Kodi
 *  This file is part of Kodi - https://kodi.tv
 *
 *  SPDX-License-Identifier: GPL-2.0-or-later
 *  See LICENSES/README.md for more information.
 */

#pragma once

#include "Job.h"
#include "threads/CriticalSection.h"
#include "threads/Thread.h"

#include <queue>
#include <string>
#include <vector>

class CJobManager;

class CJobWorker : public CThread
{
public:
  explicit CJobWorker(CJobManager *manager);
  ~CJobWorker() override;

  void Process() override;
private:
  CJobManager  *m_jobManager;
};

template<typename F>
class CLambdaJob : public CJob
{
public:
  CLambdaJob(F&& f) : m_f(std::forward<F>(f)) {}
  bool DoWork() override
  {
    m_f();
    return true;
  }
  bool operator==(const CJob *job) const override
  {
    return this == job;
  };
private:
  F m_f;
};

class CJobQueue: public IJobCallback
{
  class CJobPointer
  {
  public:
    explicit CJobPointer(CJob *job)
    {
      m_job = job;
      m_id = 0;
    };
    void CancelJob();
    void FreeJob()
    {
      delete m_job;
      m_job = NULL;
    };
    bool operator==(const CJob *job) const
    {
      if (m_job)
        return *m_job == job;
      return false;
    };
    CJob *m_job;
    unsigned int m_id;
  };
public:
  CJobQueue(bool lifo = false, unsigned int jobsAtOnce = 1, CJob::PRIORITY priority = CJob::PRIORITY_LOW);

  ~CJobQueue() override;

  bool AddJob(CJob *job);

  template<typename F>
  void Submit(F&& f)
  {
    AddJob(new CLambdaJob<F>(std::forward<F>(f)));
  }

  void CancelJob(const CJob *job);

  void CancelJobs();

  bool IsProcessing() const;

  void OnJobComplete(unsigned int jobID, bool success, CJob *job) override;

  void OnJobAbort(unsigned int jobID, CJob* job) override;

protected:
  bool QueueEmpty() const;

private:
  void OnJobNotify(CJob* job);
  void QueueNextJob();

  typedef std::deque<CJobPointer> Queue;
  typedef std::vector<CJobPointer> Processing;
  Queue m_jobQueue;
  Processing m_processing;

  unsigned int m_jobsAtOnce;
  CJob::PRIORITY m_priority;
  mutable CCriticalSection m_section;
  bool m_lifo;
};

class CJobManager final
{
  class CWorkItem
  {
  public:
    CWorkItem(CJob *job, unsigned int id, CJob::PRIORITY priority, IJobCallback *callback)
    {
      m_job = job;
      m_id = id;
      m_callback = callback;
      m_priority = priority;
    }
    bool operator==(unsigned int jobID) const
    {
      return m_id == jobID;
    };
    bool operator==(const CJob *job) const
    {
      return m_job == job;
    };
    void FreeJob()
    {
      delete m_job;
      m_job = NULL;
    };
    void Cancel()
    {
      m_callback = NULL;
    };
    CJob         *m_job;
    unsigned int  m_id;
    IJobCallback *m_callback;
    CJob::PRIORITY m_priority;
  };

public:
  CJobManager();

  unsigned int AddJob(CJob *job, IJobCallback *callback, CJob::PRIORITY priority = CJob::PRIORITY_LOW);

  template<typename F>
  void Submit(F&& f, CJob::PRIORITY priority = CJob::PRIORITY_LOW)
  {
    AddJob(new CLambdaJob<F>(std::forward<F>(f)), nullptr, priority);
  }

  template<typename F>
  void Submit(F&& f, IJobCallback *callback, CJob::PRIORITY priority = CJob::PRIORITY_LOW)
  {
    AddJob(new CLambdaJob<F>(std::forward<F>(f)), callback, priority);
  }

  void CancelJob(unsigned int jobID);

  void CancelJobs();

  void Restart();

  int IsProcessing(const std::string &type) const;

  void PauseJobs();

  void UnPauseJobs();

  bool IsProcessing(const CJob::PRIORITY &priority) const;

protected:
  friend class CJobWorker;
  friend class CJob;
  friend class CJobQueue;

  CJob* GetNextJob();

  void  OnJobComplete(bool success, CJob *job);

  bool  OnJobProgress(unsigned int progress, unsigned int total, const CJob *job) const;

private:
  CJobManager(const CJobManager&) = delete;
  CJobManager const& operator=(CJobManager const&) = delete;

  CJob *PopJob();

  void StartWorkers(CJob::PRIORITY priority);
  void RemoveWorker(const CJobWorker *worker);
  static unsigned int GetMaxWorkers(CJob::PRIORITY priority);

  unsigned int m_jobCounter;

  typedef std::deque<CWorkItem>    JobQueue;
  typedef std::vector<CWorkItem>   Processing;
  typedef std::vector<CJobWorker*> Workers;

  JobQueue   m_jobQueue[CJob::PRIORITY_DEDICATED + 1];
  bool       m_pauseJobs;
  Processing m_processing;
  Workers    m_workers;

  mutable CCriticalSection m_section;
  CEvent           m_jobEvent;
  bool             m_running;
};