threadpool.hh

Go to the documentation of this file.

/*
 * Scroom - Generic viewer for 2D data
 * Copyright (C) 2009-2026 Kees-Jan Dijkzeul
 *
 * SPDX-License-Identifier: LGPL-2.1
 */
 
#pragma once
 
#include <memory>
#include <queue>
#include <vector>
 
#include <boost/function.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 = std::shared_ptr<Queue>;
    using WeakPtr = std::weak_ptr<Queue>;
 
  public:
    static Ptr create();
 
    static Ptr createAsync();
 
    ~Queue();
    std::shared_ptr<Scroom::Detail::ThreadPool::QueueImpl> get();
    std::shared_ptr<WeakQueue> getWeak();
 
  private:
    Queue();
    Queue(const Queue&) = delete;
    Queue(Queue&&) = delete;
    Queue& operator=(const Queue&) = delete;
    Queue& operator=(Queue&&) = delete;
 
  private:
    std::shared_ptr<WeakQueue> weak;
  };
 
  class WeakQueue
  {
  public:
    using Ptr = std::shared_ptr<WeakQueue>;
    using WeakPtr = std::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();
    std::shared_ptr<Scroom::Detail::ThreadPool::QueueImpl> get();
 
  private:
    WeakQueue();
 
  private:
    std::shared_ptr<Scroom::Detail::ThreadPool::QueueImpl> qi;
  };
 
private:
  struct Job
  {
    std::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 = std::shared_ptr<ThreadPool>;
  using ThreadPtr = std::shared_ptr<boost::thread>;
 
private:
  class PrivateData
  {
  public:
    using Ptr = std::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(std::shared_ptr<T> fn, int priority = defaultPriority, const Queue::Ptr& queue = defaultQueue());
 
  template <typename T>
  void schedule(std::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(std::shared_ptr<T> fn, int priority, const WeakQueue::Ptr& queue);
 
  template <typename T>
  void schedule(std::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(const std::shared_ptr<T>& fn, int priority = defaultPriority, const Queue::Ptr& queue = defaultQueue());
 
  template <typename R, typename T>
  boost::unique_future<R> schedule(std::shared_ptr<T> fn, const Queue::Ptr& queue);
 
  template <typename R>
  boost::unique_future<R> schedule(boost::function<R()> const& fn, int priority, const 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(const std::shared_ptr<T>& fn, int priority, const WeakQueue::Ptr& queue);
 
  template <typename R, typename T>
  boost::unique_future<R> schedule(std::shared_ptr<T> fn, WeakQueue::Ptr queue);
 
  ThreadPtr add();
 
  std::vector<ThreadPtr> add(int count);
};
 
class QueueJumper
{
public:
  using Ptr = std::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>