workerpool.h

Go to the documentation of this file.

#ifndef ION_BASE_WORKERPOOL_H_
#define ION_BASE_WORKERPOOL_H_

#include <atomic>
#include <functional>

#include "ion/base/stlalloc/allocset.h"
#include "ion/port/mutex.h"
#include "ion/port/semaphore.h"
#include "ion/port/threadutils.h"

namespace ion {
namespace base {

class WorkerPool : public ion::base::Allocatable {
 public:
  class Worker {
   public:
    virtual void DoWork() = 0;

    virtual const std::string& GetName() const = 0;

   protected:
    virtual ~Worker() {}
  };

  explicit WorkerPool(Worker* worker);
  ~WorkerPool() override;

  const std::string& GetName() const { return worker_->GetName(); }

  void Suspend();
  void Resume();
  bool IsSuspended() const;

  void ResizeThreadPool(size_t thread_count);

  port::Semaphore* GetWorkSemaphore() { return &work_sema_; }

 protected:
  virtual void ThreadEntryPoint();

 private:
  void KillAllThreads();

  static void Wait(port::Semaphore* sema);
  static void Post(port::Semaphore* sema);

  Worker* const worker_;
  ion::base::AllocSet<ion::port::ThreadId> threads_;
  port::Semaphore work_sema_;
  port::Semaphore active_threads_sema_;
  std::atomic<bool> suspended_;
  std::atomic<bool> killing_;
  std::atomic<bool> slow_path_;
  std::function<bool()> spawn_func_;
  mutable ion::port::Mutex mutex_;

  DISALLOW_IMPLICIT_CONSTRUCTORS(WorkerPool);
};

}  // namespace base
}  // namespace ion

#endif  // ION_BASE_WORKERPOOL_H_