pool.h

Go to the documentation of this file.

// Copyright (c) 2013-2025  Made to Order Software Corp.  All Rights Reserved
//
// https://snapwebsites.org/project/cppthread
// contact@m2osw.com
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License along
// with this program; if not, write to the Free Software Foundation, Inc.,
// 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#pragma once
 
// self
//
#include    <cppthread/exception.h>
#include    <cppthread/fifo.h>
#include    <cppthread/thread.h>
 
 
 
namespace cppthread
{
 
 
 
template<class W, class ...A>
class pool
{
public:
    typedef std::shared_ptr<pool<W, A...>>      pointer_t;
    typedef typename W::work_load_type          work_load_type;
    typedef fifo<work_load_type>                worker_fifo_t;
 
private:
    class worker_thread_t
    {
    public:
        typedef std::shared_ptr<worker_thread_t>    pointer_t;
        typedef std::vector<pointer_t>              vector_t;
 
        worker_thread_t(std::string const & name
                      , std::size_t i
                      , typename worker_fifo_t::pointer_t in
                      , typename worker_fifo_t::pointer_t out
                      , A... args)
            : f_worker(name + " (worker #" + std::to_string(i) + ")"
                     , i
                     , in
                     , out
                     , args...)
            , f_thread(std::make_shared<thread>(name, &f_worker))
        {
            f_thread->start();
        }
 
        W & get_worker()
        {
            return f_worker;
        }
 
        W const & get_worker() const
        {
            return f_worker;
        }
 
    private:
        W                       f_worker;   // runner before thread; this is safe
        thread::pointer_t       f_thread;
    };
 
 
public:
    pool(
              std::string const & name
            , std::size_t pool_size
            , typename worker_fifo_t::pointer_t in
            , typename worker_fifo_t::pointer_t out
            , A... args)
        : f_name(name)
        , f_in(in)
        , f_out(out)
    {
        if(pool_size == 0)
        {
            throw out_of_range("the pool size must be a positive number (1 or more).");
        }
        if(pool_size > 1'000)
        {
            throw out_of_range("pool size too large (we accept up to 1000 at this time, which is already very very large).");
        }
        f_workers.reserve(pool_size);
        for(std::size_t i(0); i < pool_size; ++i)
        {
            f_workers.push_back(std::make_shared<worker_thread_t>(
                          f_name
                        , i
                        , f_in
                        , f_out
                        , args...));
        }
    }
 
    ~pool()
    {
        stop(false);
        wait();
    }
 
    std::size_t size() const
    {
        return f_workers.size();
    }
 
    W & get_worker(int i)
    {
        if(static_cast<std::size_t>(i) >= f_workers.size())
        {
            throw out_of_range("cppthread::pool::get_worker() called with an index out of bounds.");
        }
        return f_workers[i]->get_worker();
    }
 
    W const & get_worker(int i) const
    {
        if(static_cast<std::size_t>(i) >= f_workers.size())
        {
            throw out_of_range("cppthread::pool::get_worker() const called with an index out of bounds.");
        }
        return f_workers[i]->get_worker();
    }
 
    void push_back(work_load_type const & v)
    {
        f_in->push_back(v);
    }
 
    bool pop_front(work_load_type & v, std::int64_t usecs)
    {
        if(f_in->is_done())
        {
            usecs = 0;
        }
        if(f_out != nullptr)
        {
            return f_out->pop_front(v, usecs);
        }
        return false;
    }
 
    void stop(bool immediate)
    {
        if(!f_in->is_done())
        {
            f_in->done(immediate);
        }
    }
 
    void wait()
    {
        f_workers.clear();
    }
 
 
private:
    typedef typename worker_thread_t::vector_t  workers_t;
 
    std::string const                   f_name;
    typename worker_fifo_t::pointer_t   f_in;
    typename worker_fifo_t::pointer_t   f_out;
    workers_t                           f_workers = workers_t();
};
 
 
 
} // namespace cppthread
// vim: ts=4 sw=4 et