I have a Balancer Class in my project which should dynamically increases/decreases the amount of Worker Thread based on the current amount of Messages which are written to an output queue.

For that i have created the following Classes:

Message.h

A message class thats deriving from the Poco::Notification Class because i'm also using the corresponding Poco::NotificationQueue.

class Message : public Poco::Notification {
public:
  explicit Message(const std::string& id);
  Message(const std::string& id, const std::string& content);

  std::string content() const {
    return content_;
  }

  std::string id() const {
    return id_;
  }

  void content(const std::string& content) {
    content_ = content;
  }

  void id(const std::string& id) {
    id_ = id;
  }

  friend bool operator==(const Message& l_message, const Message& r_message);
  friend bool operator!=(const Message& l_message, const Message& r_message);

private:
  std::string id_;
  std::string content_;
};

MessageQueue.h

The MessageQueue.h is just an interface for some queue implementation.

// MessageQueueImpl.h
class Message;

class MessageQueue {

public:
  virtual void enqueue(std::unique_ptr<Message> message) = 0;
  virtual std::unique_ptr<Message> dequeue() = 0;
  virtual int size() const = 0;
  virtual void close() = 0;
};

This is the concrete Implementation of that Interface:

class Message;

class MessageQueueImpl : public MessageQueue {
  class MessageQueuePimpl;
public:
  MessageQueueImpl();
  virtual ~MessageQueueImpl();
  void enqueue(std::unique_ptr<Message> message) override;
  std::unique_ptr<Message> dequeue() override;
  int size() const override;
  void close() override;

private:
  std::unique_ptr<MessageQueuePimpl> pimpl_;
};

// MessageQueueImpl.cpp

class MessageQueueImpl::MessageQueuePimpl {
public:
  void enqueue(std::unique_ptr<Message> message);
  std::unique_ptr<Message> dequeue();
  int size() const;
  void close();
private:
  Poco::NotificationQueue queue_;
};


void MessageQueueImpl::MessageQueuePimpl::enqueue(std::unique_ptr<Message> message) {
  auto raw_message = message.release();
  queue_.enqueueNotification(raw_message);
}

std::unique_ptr<Message> MessageQueueImpl::MessageQueuePimpl::dequeue() {
  auto notification = queue_.waitDequeueNotification();
  return std::unique_ptr<Message>(dynamic_cast<Message*>(notification));
}

int MessageQueueImpl::MessageQueuePimpl::size() const {
  return queue_.size();
}

void MessageQueueImpl::MessageQueuePimpl::close() {
  queue_.wakeUpAll();
}


MessageQueueImpl::MessageQueueImpl() : pimpl_(std::make_unique<MessageQueuePimpl>()) {
}

MessageQueueImpl::~MessageQueueImpl() = default;

void MessageQueueImpl::enqueue(std::unique_ptr<Message> message) {
  pimpl_->enqueue(std::move(message));
}

std::unique_ptr<Message> MessageQueueImpl::dequeue() {
  return pimpl_->dequeue();
}

int MessageQueueImpl::size() const {
  return pimpl_->size();
}

void MessageQueueImpl::close() {
  pimpl_->close();
}

** Balancer.h **

This class uses two of the above described MessageQueues. The first is the input queue and data is written from some other thread into that queue. The idea is that the Balancer starts additional Threads if the size of the output queue is 3 times in a row greater then 5 times the current amount of running workerthreads. If the current output queue size is 5 times in a row less then 5 times the current amount of running workerthreads then a thread is stopped. Also the Balancer makes sure that at least 2 Workers are always running.

Here is the corresponding Code:

class Balancer {
  class BalancerPimpl;

public:
  Balancer(MessageQueue& input_queue, MessageQueue& output_queue, WorkerPtr worker_prototype);
  ~Balancer();
  void run();
  void start();
  void stop();

  int threadPoolSize() const;
  int runningWorker() const;
  size_t workerSize() const;
  int increasingCounter() const;
  int decreasingCounter() const;

private:
  std::unique_ptr<BalancerPimpl> pimpl_;
};

// Balancer.cpp
class Balancer::BalancerPimpl {
public:
  BalancerPimpl(MessageQueue& input_queue, MessageQueue& output_queue, WorkerPtr worker_prototype);

  void run();
  void start();
  void stop();

  int threadPoolSize() const;
  int runningWorker() const;
  size_t workerSize() const;
  int increasingCounter() const;
  int decreasingCounter() const;
private:
  using WorkerList = std::vector<WorkerPtr>;

  static const int LIMIT_MULTIPLIKATOR;
  static const int INCREASING_LIMIT;
  static const int DECREASING_LIMIT;
  static const int THREADPOOL_MIN_CAPACITY;

  void addWorker();
  int outputQueueUpperLimit() const;
  int outputQueueLowerLimit() const;
  bool outputQueueAboveUpperLimit() const;
  bool outputQueueBelowLowerLimit() const;
  void adjustThreadPool();
  void startNewThread();
  void stopThread();
  bool threadsAvailable();

  MessageQueue& input_queue_;
  MessageQueue& output_queue_;
  WorkerPtr worker_prototype_;
  Poco::ThreadPool balancer_pool_;
  bool is_running_{ true };
  WorkerList worker_list_;
  int increasing_counter_{ 0 };
  int decreasing_counter_{ 0 };
  int message_count_from_input_queue_{ 0 };
  int message_count_to_output_queue_{ 0 };
};

const int Balancer::BalancerPimpl::LIMIT_MULTIPLIKATOR{ 5 };
const int Balancer::BalancerPimpl::INCREASING_LIMIT{ 3 };
const int Balancer::BalancerPimpl::DECREASING_LIMIT{ 5 };
const int Balancer::BalancerPimpl::THREADPOOL_MIN_CAPACITY{ 2 };

Balancer::BalancerPimpl::BalancerPimpl(MessageQueue& input_queue, MessageQueue& output_queue, WorkerPtr worker_prototype) : input_queue_{ input_queue }, output_queue_{ output_queue }, worker_prototype_{ std::move(worker_prototype) }, balancer_pool_{ THREADPOOL_MIN_CAPACITY, 16, 1 } {
  for (int i = 0; i < balancer_pool_.capacity(); ++i) {
    addWorker();
  }
}

void Balancer::BalancerPimpl::run() {
  while (is_running_) {
    adjustThreadPool();
    auto message = input_queue_.dequeue();
    ++message_count_from_input_queue_;
    if (message) {
      output_queue_.enqueue(std::move(message));
      ++message_count_to_output_queue_;
    }
  }
}

void Balancer::BalancerPimpl::start() {
  for (int i = 0; i < THREADPOOL_MIN_CAPACITY; i++) {
    worker_list_[i]->start();
    balancer_pool_.start(*worker_list_[i]);
  }
}

void Balancer::BalancerPimpl::stop() {
  for (auto& worker : worker_list_) {
    if (!worker->stopped()) {
      worker->stop();
    }
  }
  is_running_ = false;
  output_queue_.close();
  balancer_pool_.collect();
  balancer_pool_.stopAll();
  balancer_pool_.joinAll();
}

int Balancer::BalancerPimpl::runningWorker() const {
  return std::count_if(worker_list_.begin(), worker_list_.end(), [](const WorkerPtr& worker) {
    return !worker->stopped();
  });
}

int Balancer::BalancerPimpl::threadPoolSize() const {
  return balancer_pool_.capacity();
}

size_t Balancer::BalancerPimpl::workerSize() const {
  return worker_list_.size();
}

int Balancer::BalancerPimpl::increasingCounter() const {
  return increasing_counter_;
}

int Balancer::BalancerPimpl::decreasingCounter() const {
  return decreasing_counter_;
}

void Balancer::BalancerPimpl::addWorker() {
  auto worker_thread = WorkerPtr(worker_prototype_->clone());
  worker_list_.push_back(std::move(worker_thread));
}

int Balancer::BalancerPimpl::outputQueueUpperLimit() const {
  return runningWorker() * LIMIT_MULTIPLIKATOR;
}

int Balancer::BalancerPimpl::outputQueueLowerLimit() const {
  return (runningWorker() - 1) * LIMIT_MULTIPLIKATOR;
}

bool Balancer::BalancerPimpl::outputQueueAboveUpperLimit() const {
  return output_queue_.size() >= outputQueueUpperLimit();
}

bool Balancer::BalancerPimpl::outputQueueBelowLowerLimit() const {
  return output_queue_.size() < outputQueueLowerLimit();
}

void Balancer::BalancerPimpl::adjustThreadPool() {
  if (threadsAvailable() && outputQueueAboveUpperLimit()) {
    ++increasing_counter_;
    if (increasing_counter_ == INCREASING_LIMIT) {
      startNewThread();
      increasing_counter_ = 0;
    }

  } else if (threadsAvailable() && increasing_counter_ > 0) {
    --increasing_counter_;
  }
  if (runningWorker() > THREADPOOL_MIN_CAPACITY && outputQueueBelowLowerLimit()) {
    ++decreasing_counter_;

    if (decreasing_counter_ == DECREASING_LIMIT) {
      stopThread();
      decreasing_counter_ = 0;
    }
  } else if (runningWorker() > THREADPOOL_MIN_CAPACITY && decreasing_counter_ > 0) {
    --decreasing_counter_;
  }
}

void Balancer::BalancerPimpl::startNewThread() {
  auto stopped_thread = std::find_if(worker_list_.begin(), worker_list_.end(), [](const WorkerPtr& worker) {
    return worker->stopped();
  });
  if (stopped_thread != worker_list_.end()) {
    (*stopped_thread)->start();
    balancer_pool_.start(**stopped_thread);
  }
}

void Balancer::BalancerPimpl::stopThread() {
  auto running_thread = std::find_if(worker_list_.rbegin(), worker_list_.rend(), [](const WorkerPtr& worker) {
    return !worker->stopped();
  });

  (*running_thread)->stop();
}

bool Balancer::BalancerPimpl::threadsAvailable() {
  return balancer_pool_.available() != 0;
}

Balancer::Balancer(MessageQueue& input_queue, MessageQueue& output_queue, WorkerPtr worker_prototype) : pimpl_{ std::make_unique<BalancerPimpl>(input_queue, output_queue, std::move(worker_prototype)) } {}

Balancer::~Balancer() {
  stop();
}

void Balancer::run() {
  pimpl_->run();
}

void Balancer::start() {
  pimpl_->start();
}

void Balancer::stop() {
  pimpl_->stop();
}

int Balancer::threadPoolSize() const {
  return pimpl_->threadPoolSize();
}

int Balancer::runningWorker() const {
  return pimpl_->runningWorker();
}

size_t Balancer::workerSize() const {
  return pimpl_->workerSize();
}

int Balancer::increasingCounter() const {
  return pimpl_->increasingCounter();
}

int Balancer::decreasingCounter() const {
  return pimpl_->decreasingCounter();
}

** Worker.hpp ** The WorkerPtris just a smart pointer to the Worker Prototype. During the startup of the Balancer, 16 clones are created from that Prototype.

class CORE_EXPORT Worker : public Poco::Runnable {

public:
  enum class State {
    STARTING,
    RUNNING,
    STOPPING,
    STOPPED
  };
  virtual ~Worker() = default;
  Worker() = default;
  Worker(const Worker& other) {}

  void start();

  void run() override;

  Worker* clone();

  void stop();

  bool stopped();

  State state() const;

private:


  virtual void startImpl() = 0;
  virtual void runImpl() = 0;
  virtual Worker* cloneImpl() = 0;
  virtual void stopImpl() = 0;

  State state_{ State::STOPPED };
};
using WorkerPtr = std::unique_ptr<Worker>;

The Code it self works when i run it. But from the testing point of view i have some issues to really test that code. I often had to wait a specific amount of time in the test and often on one day the tests succeed on the other they fail because the amount of time is to short. I already thought about extracting the handling of the threadpool (starting/stopping threads) into it's own class and add them in the kind of a strategy to the Balancer. But i'm not quite sure what would be the best. So i'm open for any suggestions.