Line data    Source code

       1             : #include "Tools/Thread/Thread_pool/Standard/Thread_pool_standard.hpp"
       2             : 
       3             : using namespace spu;
       4             : using namespace spu::tools;
       5             : 
       6         791 : Thread_pool_standard::Thread_pool_standard(const size_t n_threads)
       7             :   : Thread_pool(n_threads)
       8         791 :   , pool(n_threads)
       9         791 :   , mtx(n_threads)
      10         791 :   , cnd(n_threads)
      11        1582 :   , barrier(n_threads)
      12             : {
      13         791 : }
      14             : 
      15           0 : Thread_pool_standard::Thread_pool_standard(const size_t n_threads, std::function<void(const size_t)>& func_init)
      16           0 :   : Thread_pool_standard(n_threads)
      17             : {
      18           0 :     this->set_func_init(func_init);
      19           0 :     this->init();
      20           0 : }
      21             : 
      22           0 : Thread_pool_standard::Thread_pool_standard(const Thread_pool_standard& other)
      23             :   : Thread_pool(other)
      24           0 :   , pool(other.n_threads)
      25           0 :   , mtx(other.n_threads)
      26           0 :   , cnd(other.n_threads)
      27           0 :   , barrier(other.barrier)
      28             : {
      29           0 : }
      30             : 
      31             : void
      32         791 : Thread_pool_standard::init(const bool async)
      33             : {
      34         791 :     if (this->initialized)
      35             :     {
      36           0 :         std::stringstream message;
      37           0 :         message << "This pool of threads has already been initialized and cannot be initialized twice.";
      38           0 :         throw tools::runtime_error(__FILE__, __LINE__, __func__, message.str());
      39           0 :     }
      40             : 
      41        6470 :     for (size_t tid = 0; tid < n_threads; tid++)
      42        5679 :         this->pool[tid] = std::thread(&Thread_pool_standard::_start_thread, this, tid);
      43             : 
      44         791 :     if (!async)
      45             :     {
      46         791 :         this->barrier.wait();
      47         791 :         this->initialized = true;
      48             :     }
      49         791 : }
      50             : 
      51        1582 : Thread_pool_standard::~Thread_pool_standard()
      52             : {
      53             :     // stop the threads pool
      54         791 :     this->stop_threads = true;
      55        6470 :     for (size_t tid = 0; tid < this->n_threads; tid++)
      56             :     {
      57        5679 :         std::lock_guard<std::mutex> lock(this->mtx[tid]);
      58        5679 :         this->cnd[tid].notify_one();
      59        5679 :     }
      60             : 
      61        6470 :     for (size_t tid = 0; tid < n_threads; tid++)
      62        5679 :         this->pool[tid].join();
      63        1582 : }
      64             : 
      65             : void
      66         564 : Thread_pool_standard::run(const bool async)
      67             : {
      68         564 :     if (!this->initialized)
      69             :     {
      70           0 :         std::stringstream message;
      71           0 :         message << "This pool of threads cannot be run because it has not been initialized.";
      72           0 :         throw tools::runtime_error(__FILE__, __LINE__, __func__, message.str());
      73           0 :     }
      74             : 
      75        5736 :     for (size_t tid = 0; tid < n_threads; tid++)
      76             :     {
      77        5172 :         std::lock_guard<std::mutex> lock(this->mtx[tid]);
      78        5176 :         this->cnd[tid].notify_one();
      79        5177 :     }
      80             : 
      81         564 :     if (!async) this->barrier.wait();
      82         564 : }
      83             : 
      84             : void
      85         564 : Thread_pool_standard::run(std::function<void(const size_t)>& set_func_exec, const bool async)
      86             : {
      87         564 :     this->set_func_exec(set_func_exec);
      88         564 :     this->run(async);
      89         564 : }
      90             : 
      91             : void
      92         554 : Thread_pool_standard::wait()
      93             : {
      94         554 :     this->barrier.wait();
      95         561 :     this->initialized = true;
      96         561 : }
      97             : 
      98             : void
      99        5625 : Thread_pool_standard::_start_thread(const size_t tid)
     100             : {
     101        5625 :     this->func_init(tid);
     102             : 
     103       15847 :     while (!this->stop_threads)
     104             :     {
     105       10661 :         std::unique_lock<std::mutex> lock(this->mtx[tid]);
     106       10733 :         this->barrier.arrive();
     107       10690 :         this->cnd[tid].wait(lock);
     108             : 
     109        9934 :         if (!this->stop_threads) this->func_exec(tid);
     110       10071 :     }
     111             : 
     112        5186 :     this->func_deinit(tid);
     113        4156 : }