ThreadPool.cc

Veure la documentació d'aquest fitxer.

/*
 * Copyright (c) 2012 Toni Corvera
 *
 * This file is part of TFCWeb.
 *
 * TFCWeb 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 3 of the License, or
 * (at your option) any later version.
 *
 * TFCWeb 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 TFCWeb. If not, see <http://www.gnu.org/licenses/>.
 */

#include "Log.h"
#include "ThreadPool.h"

#include <iostream>
#include <boost/lexical_cast.hpp>

using namespace std;
using boost::lexical_cast;

namespace {

using namespace tfc;

enum {
    INTERVAL_POLLING = 55,
    INTERVAL_TIMEOUT = 55
};

class WorkerThread {
public:
    // Nota: En la construcció encara estem en el thread principal!
    WorkerThread(ThreadPool::tipus_cua_tasques & q,
                 volatile bool * marca_de_final)
        : tasques(q), terminat_(marca_de_final)
    {
        assert( 0 != terminat_ );
    }

    WorkerThread(const WorkerThread & wt)
        : tasques(wt.tasques), terminat_(wt.terminat_)
    {
        assert( 0 != terminat_ );
    }

    void operator()() {
        using boost::posix_time::milliseconds;
        //cerr << "Executant thread " << boost::this_thread::get_id() << endl;
        assert( 0 != terminat_ );
        while (! *terminat_) {
            ThreadPool::tipus_tasca tasca;
            // INVARIANT:
            Es_Pot_Cridar_Sense_Arguments<ThreadPool::tipus_tasca>();
            const bool ok = tasques.pop(tasca, milliseconds(INTERVAL_TIMEOUT));
            if (ok) {
                tasca();
            }
        }
    }

private:
    ThreadPool::tipus_cua_tasques & tasques;
    volatile bool * terminat_;
};

} // ns anònim

namespace tfc {

ThreadPool::ThreadPool(const size_t num_threads)
    : n_threads_(num_threads), tancada_(false), marca_terminacio_(false)
{
    inicia();
}

ThreadPool::~ThreadPool() {
    termina();
    join(); // Cal esperar que finalitzin per no tenir efectes extranys
    for (size_t i=0; i<n_threads_; ++i) {
        delete threads_[i];
    }
}

void ThreadPool::programa(tipus_tasca tasca) throw (ErrorThreadPoolTancada) {
    // INVARIANT:
    Es_Pot_Cridar_Sense_Arguments<tipus_tasca>();
    if (tancada_) {
        throw ErrorThreadPoolTancada();
    }
    tasques_.push(tasca);
}

void ThreadPool::inicia() {
    threads_.reserve(n_threads_);
    for (size_t i=0; i<n_threads_; ++i) {
        threads_.push_back( new boost::thread(WorkerThread(tasques_, &marca_terminacio_)) );
    }
}

void ThreadPool::tanca() {
    tancada_ = true;
}

void ThreadPool::join() {
    tanca(); // No més tasques
    // TODO: Millor mètode?
    // Esperem que es finalitzin les tasques (o que s'indiqui finalització immediata)...
    while (!tasques_.empty() && !marca_terminacio_) {
        boost::this_thread::sleep(boost::posix_time::milliseconds(INTERVAL_POLLING));
    }
    // ... i terminem els workers
    termina();
    // join sobre cada thread
    for_each(threads_.begin(), threads_.end(), std::mem_fun<void>(&boost::thread::join));
}

void ThreadPool::termina() {
    tanca(); // No més tasques
    marca_terminacio_ = true;
}

} // ns tfc

// vim:set ts=4 et ai: //