#include "common.hh"

#include <algorithm>
#include <condition_variable>
#include <deque>
#include <mutex>
#include <thread>

#include "task_runner.hh"

namespace {

class TaskRunnerThread : public TaskRunner {
public:
  explicit TaskRunnerThread(size_t threads)
    : threads_(std::max<size_t>(1, threads)) {
    thread_ = std::make_unique<std::thread[]>(threads_);
    for (size_t i = 0; i < threads_; ++i)
      thread_[i] = std::thread(&TaskRunnerThread::thread, this);
  }

  ~TaskRunnerThread() override {
    {
      std::lock_guard<std::mutex> lock(mutex_);
      quit_ = true;
    }
    cond_.notify_all();
    for (size_t i = 0; i < threads_; ++i)
      thread_[i].join();
  }

  void post(std::function<void()> callback) override {
    {
      std::lock_guard<std::mutex> lock(mutex_);
      queue_.push_back(std::move(callback));
    }
    cond_.notify_one();
  }

private:
  void thread() {
    while (true) {
      std::function<void()> callback;
      while (true) {
        std::unique_lock<std::mutex> lock(mutex_);
        if (queue_.empty()) {
          if (quit_)
            return;
          cond_.wait(lock);
        } else {
          callback = std::move(queue_.front());
          queue_.pop_front();
          break;
        }
      }

      callback();
    }
  }

  size_t const threads_;
  bool quit_{false};
  std::condition_variable cond_;
  std::mutex mutex_;
  std::deque<std::function<void()>> queue_;
  std::unique_ptr<std::thread[]> thread_;
};

}  // namespace

std::unique_ptr<TaskRunner> TaskRunner::create(size_t threads) {
  return std::make_unique<TaskRunnerThread>(threads);
}