ThreadPool.cpp

Go to the documentation of this file.

/*
 * This file is part of ArmarX.
 *
 * Copyright (C) 2011-2017, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
 *
 * ArmarX is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation.
 *
 * ArmarX 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, see <http://www.gnu.org/licenses/>.
 *
 * @package    ArmarX
 * @author     Mirko Waechter( mirko.waechter at kit dot edu)
 * @date       2018
 * @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
 *             GNU General Public License
 */
 
#include "ThreadPool.h"
 
#include <ArmarXCore/core/exceptions/Exception.h>
#include <ArmarXCore/core/exceptions/local/ExpressionException.h>
 
// This was transitively included from some other STL header earlier,
// but not anymore since GCC 12.  This change will make building boost
// asio fail now.  See also https://github.com/doxygen/doxygen/issues/8909
// for a similar issue in Doxygen.
#include <utility>
 
#include <boost/asio.hpp>
#include <boost/thread.hpp>
 
namespace armarx
{
    struct ThreadPool::Impl
    {
        Impl(std::size_t pool_size, bool queueTasks) :
            work(io_service), available(pool_size), queueTasks(queueTasks)
        {
        }
 
        boost::asio::io_service io_service;
        boost::asio::io_service::work work;
        boost::thread_group threads;
        std::size_t available;
        std::mutex mutex;
        bool queueTasks;
    };
 
    ThreadPool::ThreadPool(std::size_t pool_size, bool queueTasks) :
        impl(new Impl(pool_size, queueTasks))
    {
        for (std::size_t i = 0; i < pool_size; ++i)
        {
            impl->threads.create_thread(
                boost::bind(&boost::asio::io_service::run, &impl->io_service));
        }
    }
 
    ThreadPool::~ThreadPool()
    {
        // Force all threads to return from io_service::run().
        impl->io_service.stop();
 
        // Suppress all exceptions.
        try
        {
            impl->threads.join_all();
        }
        catch (const std::exception&)
        {
        }
    }
 
    ThreadPool::Handle
    ThreadPool::runTask(std::function<void()> task)
    {
        std::unique_lock lock(impl->mutex);
 
        // If no threads are available, then return.
        if (!impl->queueTasks && impl->available == 0)
        {
            return {};
        }
 
        // Decrement count, indicating thread is no longer available.
        if (!impl->queueTasks)
        {
            --impl->available;
        }
        auto promise = std::make_shared<std::promise<void>>();
        // Post a wrapped task into the queue.
        impl->io_service.post(
            [this, task, promise]
            {
                wrap_task(task);
                promise->set_value();
            });
        return Handle{promise->get_future()};
    }
 
    int
    ThreadPool::getAvailableTaskCount() const
    {
        std::unique_lock lock(impl->mutex);
        if (!impl->queueTasks)
        {
            return impl->available;
        }
        else
        {
            return -1;
        }
    }
 
    void
    ThreadPool::wrap_task(std::function<void()> task)
    {
        // Run the user supplied task.
        try
        {
            task();
        }
        // Suppress all exceptions.
        catch (const std::exception&)
        {
            std::unique_lock lock(impl->mutex);
            // Task has ended, so increment count of available threads.
            if (!impl->queueTasks)
            {
                ++impl->available;
            }
            handleExceptions();
        }
 
        // Task has finished, so increment count of available threads.
        std::unique_lock lock(impl->mutex);
        if (!impl->queueTasks)
        {
            ++impl->available;
        }
    }
 
    ThreadPool::Handle::Handle() : mutex(new std::mutex)
    {
    }
 
    ThreadPool::Handle::Handle(std::shared_future<void> functionFinished) :
        mutex(new std::mutex), functionFinished(functionFinished)
    {
    }
 
    ThreadPool::Handle::~Handle() noexcept(false)
    {
        if (!joined && isValid() && !detached)
        {
            throw LocalException()
                << "You did not join the thread pool handle before the handle was deleted!";
        }
    }
 
    bool
    ThreadPool::Handle::isValid() const
    {
        return functionFinished.valid();
    }
 
    void
    ThreadPool::Handle::join()
    {
        ARMARX_CHECK_EXPRESSION(mutex);
        std::lock_guard<decltype(*mutex)> lock(*mutex);
 
        if (!isValid())
        {
            return;
        }
        if (detached)
        {
            throw LocalException() << "You cannot join a detached thread!";
        }
        functionFinished.get();
        joined = true;
    }
 
    void
    ThreadPool::Handle::detach()
    {
        ARMARX_CHECK_EXPRESSION(mutex);
        std::lock_guard<decltype(*mutex)> lock(*mutex);
 
        if (!isValid())
        {
            return;
        }
        if (joined)
        {
            throw LocalException() << "You cannot detach a joined thread!";
        }
        detached = true;
    }
 
    const std::shared_future<void>&
    ThreadPool::Handle::getFuture() const
    {
        return functionFinished;
    }
 
    bool
    ThreadPool::Handle::isJoined() const
    {
        return joined;
    }
 
    bool
    ThreadPool::Handle::isDetached() const
    {
        return detached;
    }
 
} // namespace armarx