C++ Dynamic Balancer on a Threadpool

Question

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.

Answer 1

This is not helpful

Perhaps you have the idea that a thread is consuming resources, so it is better to not have more running than necessary. While each thread uses some memory (mainly because each thread needs its own stack), if a thread is blocked in a waitDequeueNotification() call, it doesn't consume CPU resources. So no performance is lost if you have a bunch of extra threads that are just blocked all the time. However, you are losing performance by adding/removing threads on demand, and having the Balancer shuffle messages from the input_queue_ to the output_queue_. So this means it is a net loss to use your balancer.

If you are not going to balance, the question is then how big should the thread pool be? Assuming the work each thread has to do is CPU-bound, then using as many threads as there are logical CPU cores is usually best. You could use Poco::Environment::processorCount() for this, or just std::thread::hardware_concurrency() from the standard library.

If it's not CPU-bound, but for example memory bound or disk I/O bound, then throwing more threads at the problem might not help at all. If that is so, also consider the case where messages come in faster than the threads can process them. The output queue will grow in size, you add more threads but it doesn't help, and the output queue will grow even further, and you add even more threads, and so on. At some point adding more threads will even make things worse.

Finally, your balancer also assumes the messages come in at a regular pace. However, what if they come in batches, like 100 messages at once, then nothing for a second. Your balancer will then increase the number of threads a lot, then after the first batch it doesn't get any messages for a second, so adjustThreadPool() is also not called during that second. Then when the first DECREASING_LIMIT messages arrives of the next batch, it will decrease the size of the thread pool, even though it will need to increase the size later in the batch. So your balancer is behaving badly for irregularly spaces messages.

Both Impl and Pimpl?

I think it's overkill to have an abstract base class and a pointer-to-implementation in the derived class. Since the derived class can already be easily hidden, there is no need to use the pimpl idiom as well. The extra indirection will hurt performance.

You can make a free function createMessageQueueImpl() that returns a pointer to a MessageQueue. So in a header file you would just have:

std::unique_ptr<MessageQueue> createMessageQueueImpl();

And then in a source file, you can have both the definition of class MessageQueueImpl and the aforementioned function:

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

std::unique_ptr<MessageQueue> createMessageQueueImpl() {
    return std::make_unique<MessageQueueImpl>();
}

Avoid unnecessary public members

Your Balancer exposes several public member functions which I think should be made private, or possibly even removed completely. For example, why are increasingCounter() and decreasingCounter() public members? Calling this will just interfere with the balancers own counting. Why would you want a caller to be able to get the size of the pool or the number of running workers? Maybe you used this for debugging purposes, but I don't see a use for this in production code.

Why are there both run() and start()? What happens if someone calls run() before every having called start(), and then just one message is sent to the queue. Since no threads are running, this message will never be processed. This also brings me to:

Thread safety issues

run() never terminates on its own, so it looks like stop() has to be called from another thread. That means you will have race conditions because there is no mutex protecting is_running_ and other member variables. For example, run() could just have finished executing:

while (is_running_) {

When is_running_ was still true. Then another thread calls stop() which runs in its entirety, and then the thread running run() continues with the body of the while-statement. This will make it call input_queue_.dequeue(), but what if no more messages are being received at this point? The thread will hang indefinitely.