ct_threadpool.h

/*
 * Developer : Alexey Zakharov (alexey.zakharov@vectioneer.com)
 * All rights reserved. Copyright (c) 2020 VECTIONEER.
 */
 
#ifndef MOTORCORTEX_CORE_CT_THREADPOOL_H
#define MOTORCORTEX_CORE_CT_THREADPOOL_H
 
#include "ct_tasksched.h"
#include "utl_fncwrapper.h"
#include "utl_tsqueue.h"
 
#include <atomic>
#include <future>
#include <thread>
#include <vector>
 
namespace mcx::container {
 
class JoinThreads {
public:
  explicit JoinThreads(std::vector<std::thread>& threads) : threads_(threads) {}
 
  ~JoinThreads() {
    for (auto& thread : threads_) {
      if (thread.joinable())
        thread.join();
    }
  }
 
private:
  std::vector<std::thread>& threads_;
};
 
class ThreadPool {
 
public:
  explicit ThreadPool(size_t thread_count = std::thread::hardware_concurrency());
 
  ~ThreadPool();
 
  bool setName(std::string name);
 
  std::string getName() const;
 
  bool setTaskSched(TaskSched task_sched, int task_prio = 0);
 
  std::tuple<TaskSched, int> getTaskSched() const;
 
  bool setCpuAffinity(const std::vector<uint32_t>& cpu_affinity_list);
 
  std::vector<uint32_t> getCpuAffinity() const;
 
  template <typename FunctionType>
  std::future<typename std::result_of<FunctionType()>::type> submit(FunctionType f) {
    typedef typename std::result_of<FunctionType()>::type result_type;
    std::packaged_task<result_type()> task(std::move(f));
    std::future<result_type> res(task.get_future());
    work_queue_.push(std::move(task));
    return res;
  }
 
  bool empty() const;
 
private:
  void workerThread();
 
  static constexpr size_t MAX_THREAD_NAME_LEN = 16;
  std::vector<std::thread> threads_;
  std::vector<std::thread::native_handle_type> ids_;
  ThreadSafeQueue<FunctionWrapper> work_queue_;
  std::atomic_bool done_{};
  std::atomic_int counter_{};
  JoinThreads joiner_;
};
 
} // namespace mcx::container
 
#endif // MOTORCORTEX_CORE_CT_THREADPOOL_H