ThreadPool5.hxx

Go to the documentation of this file.

/*--------------------------------------------------------------------------*\
 |                                                                          |
 |  Copyright (C) 2020                                                      |
 |                                                                          |
 |         , __                 , __                                        |
 |        /|/  \               /|/  \                                       |
 |         | __/ _   ,_         | __/ _   ,_                                |
 |         |   \|/  /  |  |   | |   \|/  /  |  |   |                        |
 |         |(__/|__/   |_/ \_/|/|(__/|__/   |_/ \_/|/                       |
 |                           /|                   /|                        |
 |                           \|                   \|                        |
 |                                                                          |
 |      Enrico Bertolazzi                                                   |
 |      Dipartimento di Ingegneria Industriale                              |
 |      Università degli Studi di Trento                                    |
 |      email: enrico.bertolazzi@unitn.it                                   |
 |                                                                          |
\*--------------------------------------------------------------------------*/
 
//
// file: ThreadPool5.hxx
//
 
namespace Utils {

 
  /*\
   |   _____ _                        _ ____             _
   |  |_   _| |__  _ __ ___  __ _  __| |  _ \ ___   ___ | |
   |    | | | '_ \| '__/ _ \/ _` |/ _` | |_) / _ \ / _ \| |
   |    | | | | | | | |  __/ (_| | (_| |  __/ (_) | (_) | |
   |    |_| |_| |_|_|  \___|\__,_|\__,_|_|   \___/ \___/|_|
  \*/
  class ThreadPool5 : public ThreadPoolBase {
 
    using FUN = ThreadPoolBase::FUN;
    //using PFUN = ThreadPoolBase::PFUN;
 
    /*\
     |  __        __         _
     |  \ \      / /__  _ __| | _____ _ __
     |   \ \ /\ / / _ \| '__| |/ / _ \ '__|
     |    \ V  V / (_) | |  |   <  __/ |
     |     \_/\_/ \___/|_|  |_|\_\___|_|
    \*/
    class Worker {
    
    
      FUN             m_push_worker;
      bool            m_active{true};    
      UTILS_SEMAPHORE m_is_running;      
      std::thread     m_running_thread;  
      FUN             m_job;             

      void
      worker_loop() {
        m_is_running.red(); // block computation
        while ( m_active ) {
          m_is_running.wait(); // wait signal to start computation
          // ----------------------------------------
          m_job();
          // ----------------------------------------
          m_is_running.red(); // block computation
          m_push_worker();    // worker ready for a new computation
        }
      }
 
    public:
 
      Worker() { m_is_running.red(); }

      explicit
      Worker( ThreadPool5 * tp, unsigned id ) {
        m_is_running.red();
        m_push_worker    = [tp,id]()->void { tp->push_worker( id ); };
        m_running_thread = std::thread( &Worker::worker_loop, this );
      }

      ~Worker() {
        wait();               // if running task wait it terminate
        m_active = false;     // deactivate computation
        m_job    = [](){};    // dummy task
        m_is_running.green(); // start computation (exiting loop)
        if ( m_running_thread.joinable() ) m_running_thread.join(); // wait thread for exiting
      }

      void wait() { m_is_running.wait_red(); }

      void
      exec( FUN && fun ) {
        m_is_running.wait_red();
        m_job = std::move(fun); // cambia funzione da eseguire
        m_is_running.green();   // activate computation
      }
 
    };
 
    // =========================================================================
    // =========================================================================
    // =========================================================================
 
    // Vector of workers managed by the thread pool.
    std::vector<Worker>     m_workers;
    std::list<unsigned>     m_queue;
    std::mutex              m_queue_mutex; 
    std::condition_variable m_queue_cond;  

    void
    push_worker( unsigned id ) {
      std::unique_lock<std::mutex> lock(m_queue_mutex);
      m_queue.emplace_back(id);
      m_queue_cond.notify_one();
    }

    unsigned
    pop_worker() {
      std::unique_lock<std::mutex> lock(m_queue_mutex);
      while ( m_queue.empty() ) m_queue_cond.wait( lock );
      unsigned id{ m_queue.back() }; m_queue.pop_back();
      return id;
    }
 
  public:

    ThreadPool5(
      unsigned nthread = std::max(
        unsigned(1),
        unsigned(std::thread::hardware_concurrency()-1)
      )
    )
    : ThreadPoolBase()
    , m_workers( size_t(nthread) )
    {
      m_queue.clear();
      unsigned id{0};
      for ( Worker & w : m_workers ) { new (&w) Worker( this, id ); ++id; }
      while ( id-- > 0 ) push_worker( id );
    }

    virtual
    ~ThreadPool5() {
      wait();
      m_workers.clear();
    }

    void
    exec( FUN && fun ) override {
      // cerca prima thread libera
      m_workers[pop_worker()].exec( std::move(fun) );
    }

    void
    wait() override
    { for ( auto & w : m_workers ) w.wait(); }

    unsigned
    thread_count() const override
    { return unsigned(m_workers.size()); }

    static char const * Name() { return "ThreadPool5"; }
 
    char const * name() const override { return Name(); }
 
  };

 
}
 
//
// eof: ThreadPool5.hxx
//