thread_pool.hpp

Go to the documentation of this file.

#ifndef FSTTHREADPOOL_H
#define FSTTHREADPOOL_H
 /* 
=========================================================================
Copyright and license:
  * FST4 software (with the exception of any bundled or externally linked
    third-party components) is Copyright (C) 2000-present Institute of
    Information Theory and Automation of the Czech Academy of Sciences (UTIA).
  * FST4 is free software: you can redistribute it and/or modify it under
    the terms of the GNU Affero General Public License as published by the
    Free Software Foundation, version 3. The full text of the license is in
    the file LICENSE in the distribution root and at
    https://www.gnu.org/licenses/agpl-3.0.html.
  * FST4 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 Affero General Public
    License for more details.
  * Closed-source use without the AGPL's copyleft obligations requires a
    commercial license from UTIA; in case of interest write to
    utia@utia.cas.cz. LICENSING.md in the distribution root explains both
    options, the contributor license agreement that every contribution
    requires, and the contacts (technical questions and pull requests:
    somol@utia.cas.cz, with "FST" in the subject line).
========================================================================== */
 
#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <memory>
#include <vector>
#include <cassert>
#include "global.hpp"
 
namespace FST {

class ThreadPool
{
public:
        explicit ThreadPool(const unsigned int max_threads=2) : _max_threads(max_threads), threads(max_threads), _running_threads(0)
        {
                assert(max_threads>0);
                for(unsigned int i=0;i<_max_threads;i++) threads[i].tid=-1;
        }

        ~ThreadPool() {notify("ThreadPool destructor, idx=",tid_counter().load()); join_all();}
 
        unsigned int get_max_threads() const {return _max_threads;} 

        unsigned int go(const std::function<void()> &threadfunc)
        {
                std::unique_lock<std::mutex> lock_launcher(mutex_launcher);
                if(_running_threads==_max_threads)
                {
                        notify("Max allowed threads running, waiting for one to end...");
                        while (_running_threads==_max_threads) cond_launcher.wait(lock_launcher);
                }
                unsigned int free_tidx=0;
                while(free_tidx<_max_threads && threads[free_tidx].tid!=-1) free_tidx++;
                assert(free_tidx<_max_threads);
                {
                        notify("Launching thread, tid=",tid_counter().load(),"...");
                        {
                                _running_threads++;
                        }
                        if(threads[free_tidx].m_thread)
                        {
                                notify("Joining past ",free_tidx,"th thread before lanching new one.");
                                threads[free_tidx].m_thread->join();
                        }
                        {
                                threads[free_tidx].tid=tid_counter()++;
                                threads[free_tidx].m_thread.reset(new std::thread(std::bind(&ThreadPool::launch_thread,this,threadfunc,free_tidx)));
                                notify("Thread tid=",threads[free_tidx].tid," (",free_tidx,"th) launched...");
                        }
                }
                return free_tidx;
        }

        void join_all()
        {
                notify("Calling join_all...");
                for(unsigned int i=0; i<_max_threads; i++)
                {
                        // take ownership of the thread handle under the lock, but join outside it -
                        // the finishing worker needs mutex_launcher in launch_thread()
                        ThreadPtr t;
                        {
                                std::unique_lock<std::mutex> lock(mutex_launcher);
                                assert((threads[i].m_thread && threads[i].tid>=0) || (threads[i].tid==-1));
                                t.swap(threads[i].m_thread);
                        }
                        if(t)
                        {
                                notify("join_all: Joining ",i,"th thread.");
                                t->join();
                                std::unique_lock<std::mutex> lock(mutex_launcher);
                                threads[i].tid=-1;
                        }
                }
                {
                        std::unique_lock<std::mutex> lock(mutex_launcher);
                        _running_threads=0;
                }
                notify("join_all finished...");
        }
        
        unsigned int get_running_threads() const {return _running_threads;}
 
private:
        // the next procedure must be re-entrant
        void launch_thread(const std::function<void()> &threadfunc, unsigned int threadsid)
        {
                notify("Starting launch_thread(tid=",threads[threadsid].tid,", ",threadsid,"th) ...");
                threadfunc();
                notify("Finishing launch_thread(tid=",threads[threadsid].tid,", ",threadsid,"th).");
                {
                        std::unique_lock<std::mutex> lock(mutex_launcher);
                        _running_threads--;
                        threads[threadsid].tid=-1;
                }
                {
                        cond_launcher.notify_one();
                }
        }
private:
        typedef std::unique_ptr<std::thread> ThreadPtr; 
        typedef struct {
                ThreadPtr m_thread; 
                int tid;            
        } ThreadsArray;
        const unsigned int _max_threads; 
        std::vector<ThreadsArray> threads; 
 
private:
        static std::atomic<unsigned int>& tid_counter() {static std::atomic<unsigned int> c(0); return c;}
        unsigned int _running_threads; 
 
        std::mutex mutex_launcher; 
        std::condition_variable cond_launcher; 
};
 
} // namespace
#endif // FSTTHREADPOOL_H ///:~