threadpool.hh

Go to the documentation of this file.

/*
 * Scroom - Generic viewer for 2D data
 * Copyright (C) 2009-2022 Kees-Jan Dijkzeul
 *
 * SPDX-License-Identifier: LGPL-2.1
 */
 
#pragma once
 
#ifdef HAVE_CONFIG_H
#  include <config.h>
#endif
 
#include <queue>
#include <vector>
 
#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
 
#include <scroom/semaphore.hh>
 
enum
{
  PRIO_HIGHEST = 100,
  PRIO_HIGHER,
  PRIO_HIGH,
  PRIO_NORMAL,
  PRIO_LOW,
  PRIO_LOWER,
  PRIO_LOWEST,
};
 
namespace Scroom::Detail::ThreadPool
{
  class QueueImpl;
} // namespace Scroom::Detail::ThreadPool
 
class ThreadPool
{
public:
  class WeakQueue;
 
  class Queue
  {
  public:
    using Ptr     = boost::shared_ptr<Queue>;
    using WeakPtr = boost::weak_ptr<Queue>;
 
  public:
    static Ptr create();
 
    static Ptr createAsync();
 
    ~Queue();
    boost::shared_ptr<Scroom::Detail::ThreadPool::QueueImpl> get();
    boost::shared_ptr<WeakQueue>                             getWeak();
 
  private:
    Queue();
    Queue(const Queue&)            = delete;
    Queue(Queue&&)                 = delete;
    Queue& operator=(const Queue&) = delete;
    Queue& operator=(Queue&&)      = delete;
 
  private:
    boost::shared_ptr<WeakQueue> weak;
  };
 
  class WeakQueue
  {
  public:
    using Ptr     = boost::shared_ptr<WeakQueue>;
    using WeakPtr = boost::weak_ptr<WeakQueue>;
 
  public:
    ~WeakQueue() = default;
 
    WeakQueue(const WeakQueue&)            = delete;
    WeakQueue(WeakQueue&&)                 = delete;
    WeakQueue& operator=(const WeakQueue&) = delete;
    WeakQueue& operator=(WeakQueue&&)      = delete;
 
    static Ptr                                               create();
    boost::shared_ptr<Scroom::Detail::ThreadPool::QueueImpl> get();
 
  private:
    WeakQueue();
 
  private:
    boost::shared_ptr<Scroom::Detail::ThreadPool::QueueImpl> qi;
  };
 
private:
  struct Job
  {
    boost::shared_ptr<Scroom::Detail::ThreadPool::QueueImpl> queue;
    boost::function<void()>                                  fn;
 
    Job() = default;
    Job(boost::function<void()> fn, const WeakQueue::Ptr& queue);
  };
 
public:
  using Ptr       = boost::shared_ptr<ThreadPool>;
  using ThreadPtr = boost::shared_ptr<boost::thread>;
 
private:
  class PrivateData
  {
  public:
    using Ptr = boost::shared_ptr<PrivateData>;
 
  public:
    unsigned int              jobcount{0}; 
    boost::mutex              mut;         
    bool                      alive{true}; 
    boost::condition_variable cond;        
    std::map<int, std::queue<Job>> jobs;
 
    bool completeAllJobsBeforeDestruction;
 
    Queue::Ptr defaultQueue;
 
  private:
    explicit PrivateData(bool completeAllJobsBeforeDestruction);
 
  public:
    static Ptr create(bool completeAllJobsBeforeDestruction);
  };
 
  std::list<ThreadPtr> threads; 
  PrivateData::Ptr     priv;
 
private:
  static void work(const PrivateData::Ptr& priv);
 
  static void do_one(const PrivateData::Ptr& priv);
 
  static Queue::Ptr defaultQueue();
  static const int  defaultPriority;
 
public:
  explicit ThreadPool(bool completeAllJobsBeforeDestruction = false);
 
  explicit ThreadPool(int count, bool completeAllJobsBeforeDestruction = false);
 
  static ThreadPool::Ptr create(bool completeAllJobsBeforeDestruction = false);
 
  static ThreadPool::Ptr create(int count, bool completeAllJobsBeforeDestruction = false);
 
  ThreadPool(const ThreadPool&)           = delete;
  ThreadPool(ThreadPool&&)                = delete;
  ThreadPool operator=(const ThreadPool&) = delete;
  ThreadPool operator=(ThreadPool&&)      = delete;
 
  ~ThreadPool();
 
  void schedule(boost::function<void()> const& fn, int priority = defaultPriority, const Queue::Ptr& queue = defaultQueue());
 
  void schedule(boost::function<void()> const& fn, const Queue::Ptr& queue);
 
  template <typename T>
  void schedule(boost::shared_ptr<T> fn, int priority = defaultPriority, const Queue::Ptr& queue = defaultQueue());
 
  template <typename T>
  void schedule(boost::shared_ptr<T> fn, const Queue::Ptr& queue);
 
  void schedule(boost::function<void()> const& fn, int priority, const WeakQueue::Ptr& queue);
 
  void schedule(boost::function<void()> const& fn, const WeakQueue::Ptr& queue);
 
  template <typename T>
  void schedule(boost::shared_ptr<T> fn, int priority, WeakQueue::Ptr queue);
 
  template <typename T>
  void schedule(boost::shared_ptr<T> fn, WeakQueue::Ptr queue);
 
  template <typename R>
  boost::unique_future<R>
    schedule(boost::function<R()> const& fn, int priority = defaultPriority, const Queue::Ptr& queue = defaultQueue());
 
  template <typename R>
  boost::unique_future<R> schedule(boost::function<R()> const& fn, const Queue::Ptr& queue);
 
  template <typename R, typename T>
  boost::unique_future<R>
    schedule(boost::shared_ptr<T> fn, int priority = defaultPriority, const Queue::Ptr& queue = defaultQueue());
 
  template <typename R, typename T>
  boost::unique_future<R> schedule(boost::shared_ptr<T> fn, const Queue::Ptr& queue);
 
  template <typename R>
  boost::unique_future<R> schedule(boost::function<R()> const& fn, int priority, WeakQueue::Ptr queue);
 
  template <typename R>
  boost::unique_future<R> schedule(boost::function<R()> const& fn, WeakQueue::Ptr queue);
 
  template <typename R, typename T>
  boost::unique_future<R> schedule(boost::shared_ptr<T> fn, int priority, WeakQueue::Ptr queue);
 
  template <typename R, typename T>
  boost::unique_future<R> schedule(boost::shared_ptr<T> fn, WeakQueue::Ptr queue);
 
  ThreadPtr add();
 
  std::vector<ThreadPtr> add(int count);
};
 
class QueueJumper
{
public:
  using Ptr = boost::shared_ptr<QueueJumper>;
 
private:
  boost::mutex mut;
  bool         inQueue{true};
  bool         isSet{false};
 
  boost::function<void()> fn;
 
protected:
  QueueJumper() = default;
 
public:
  static Ptr create();
 
  bool setWork(boost::function<void()> const& fn);
 
  void operator()();
};
 
ThreadPool::Ptr CpuBound();
 
ThreadPool::Ptr Sequentially();
 
#include <scroom/impl/threadpoolimpl.hh>