2
\$\begingroup\$

I have addressed the critique for this post and resubmitted it for iterative review; C++20 multithread pool class.

This is an iteration from a prior post; C++ thread pool class.

Class for creating thread pools with upcoming C++ standard C++2a (-std=gnu++2a -fconcepts).

  • It does not accept args for functions unless they are bound (std::bind).
  • Multiple functions may be enqueued via variadic template or stl vector.
  • The number of threads created in the pool is relative to the hardware concurrency by means of a user multiplier. If the hardware concurrency cannot be determined it will default to four and the number of threads created will be two, four, or eight depending on the multiplier used.

Please review code correctness, best practices, design and implementation.

Please assume the namespace Mercer is to be used as a cross-platform library.

This code was also available on GitHub, but now contains the current iteration.

mercer.h

#ifndef MERCER_H_0000
#define MERCER_H_0000

namespace Mercer
{
    enum struct execution: bool {failure, success};
}

#endif //MERCER_H_0000

multithread.h

#ifndef MULTITHREAD_H_0000
#define MULTITHREAD_H_0000

// GCC Bug 84330 - [6/7 Regression] [concepts] ICE with broken constraint
// https://gcc.gnu.org/bugzilla/show_bug.cgi?id=84330
#ifndef __cpp_concepts
static_assert(false, "Concepts TS NOT found");
#endif

#include <deque>
#include <queue>
#include <mutex>
#include <vector>
#include <memory>
#include <thread>
#include <atomic>
#include <functional>
#include <condition_variable>

#include "mercer.h"

namespace Mercer
{
//Multithread class
//if !joinable no new routines may be enqueued
class multithread
{
    class implementation;
    std::unique_ptr<implementation> data;

public:

    enum struct concurrency: int  {half, full, twofold};

    multithread();//concurrency::full
    multithread(concurrency quantity);
    ~multithread();

    execution enqueue(std::function<void()>&&) noexcept;
    execution enqueue(const std::function<void()>&);

    execution enqueue(std::vector<std::function<void()>>&&);
    execution enqueue(const std::vector<std::function<void()>>&);

    template<typename ... dataType>
    execution enqueue(std::function<void()>&& proximate   ,
                      std::function<void()>&& penproximate,
                      dataType&& ...          parameters  )
    {
        if(enqueue(std::move<std::function<void()>&>(proximate))
                == execution::failure)
            return execution::failure;
        else
            return enqueue(std::move<std::function<void()>&>(penproximate),
                           std::forward<dataType            >(parameters  )...);
    }

    execution join();
    execution detach();

    bool thrown() const noexcept;
    std::exception_ptr getNextException() const;
    //If thrown()==true, will never throw
    //If get final exception, thrown() will reset to false
};

}//namespace Mercer
#endif //MULTITHREAD_H_0000

debug.cpp

#ifndef DEBUG_H_0000
#define DEBUG_H_0000

#include <cstdio>

struct debug
{
    template< typename ... params>
    void operator()(const char* file, int line, const char* func, 
                    const char* note, int value )
    {
        fprintf(stdout, "%16s %4d %-16s %-16s %4d %c",       \
                        file, line, func, note, value, '\n');
    }
};

#ifdef NDEBUG
#define REPORT(...)
#else
debug REPORT;
#define REPORT(...) REPORT( __FILE__, __LINE__, __func__, __VA_ARGS__);
#endif

#endif //DEBUG_H

multithread.cpp

#include "multithread.h"
#include "debug.cpp"
using Mercer::multithread;
using Mercer::execution;
using function = std::function<void()>;

template <typename dataType>
    struct is_std_vector: std::false_type {};
template <typename dataType>
    struct is_std_vector< std::vector<dataType> >: std::true_type {};

template <typename dataType>
    concept bool is_vector = is_std_vector<dataType>::value;

template <typename dataType>
    concept bool is_invocable = std::is_invocable<dataType>::value;

struct multithread::implementation
{
    enum struct close: bool {detach, join};
    enum struct init : bool {move  , copy};

    std::atomic <bool>               open      ;
    std::deque  <function>           line      ;
    std::mutex                       door      ;
    std::condition_variable          guard     ;
    std::vector <std::thread>        pool      ;
    std::queue  <std::exception_ptr> exceptions;

    void worker()
    {
        function next;
        bool perpetual = true;
        while(perpetual)
        {
            std::unique_lock lock(door);
            guard.wait(lock,
                [&] { return !line.empty() || !open; } );
            if(!line.empty())
            {
                next = std::move<function&>(line.front());
                line.pop_front();
                if(!open && line.empty())
                    perpetual = false;
                lock.unlock();
                guard.notify_one();
                try
                {
                    next();
                }
                catch(...)
                {
                    exceptions.emplace(
                        std::current_exception() );
                }
            }
            else if(!open)
                perpetual = false;
        }
    }

    std::vector<std::thread> openPool(concurrency quantity)
    {
        std::vector<std::thread> temp;
        unsigned threads = std::thread::hardware_concurrency();
        if(threads==0)
            threads=4;
        switch(quantity)
        {
            case concurrency::half   : threads /= 2; break;
            case concurrency::full   :               break;
            case concurrency::twofold: threads *= 2; break;
        }
        temp.reserve(threads);
        for(auto i=threads; i>0; i--)
            temp.emplace_back( [&](){ worker(); } );
        return temp;
    }

    implementation(concurrency quantity) :
        open(true), line(), door(), guard(),
        pool( openPool(quantity) )
        {}

    template<close closeType>
    execution close()
    {
        if (open==true)
        {
            open = false;
            guard.notify_all();
            for (auto&& thread : pool)
                if (thread.joinable())
                    switch(closeType)
                    {
                        case close::join  : thread.join()  ; break;
                        case close::detach: thread.detach(); break;
                    }
            pool.clear();   
            pool.shrink_to_fit();
        }
        else
            return execution::failure;
        return execution::success;
    }

    template<init initType, is_invocable function>
    void enqueueSafe(function&& item) noexcept
    {
        REPORT("move",0);
        line.emplace_back(std::move<function&>(item));
    }

    template<init initType, is_invocable function>
    void enqueueSafe(const function& item)
    {
        REPORT("copy",0);
        line.push_back(item);
    }

    template<init initType, is_vector vector>
    void enqueueSafe(vector&& item)
    {
        REPORT("move",0);
        line.insert(line.end(),
            make_move_iterator(item.begin()) ,
            make_move_iterator(item.end  ()));
    }

    template<init initType, is_vector vector>
    void enqueueSafe(const vector& item)
    {
        REPORT("copy",0);
        line.insert(line.end(), item.begin(), item.end());
    }

    template<init initType, typename function>
    execution enqueue(function&& item)
    {
        if (open==true)
        { 
            std::scoped_lock lock(door);
            enqueueSafe<initType>(item);
            guard.notify_all();
            return execution::success;
        }
        else
            return execution::failure;
    }
};

multithread::multithread(concurrency quantity):
    data(std::make_unique<implementation>(quantity))
{}

multithread::multithread():
    data(std::make_unique<implementation>(concurrency::full))
{}

execution multithread::join()
{
    return data->close<implementation::close::join>();
}

execution multithread::detach()
{
    return data->close<implementation::close::detach>();
}

multithread::~multithread()
{
    join();
}

execution multithread::enqueue(function&& item) noexcept
{
    REPORT("function",0);
    return data->enqueue<implementation::init::move>(item);
}

execution multithread::enqueue(const function& item)
{
    REPORT("const function",0);
    return data->enqueue<implementation::init::copy>(item);
}

execution multithread::enqueue(std::vector<function>&& adjunct)
{
    REPORT("vector",0);
    return data->enqueue<implementation::init::move>(adjunct);
}

execution multithread::enqueue(const std::vector<function>& adjunct)
{
    REPORT("const vector",0);
    return data->enqueue<implementation::init::copy>(adjunct);
}

bool multithread::thrown() const noexcept
{
    return data->exceptions.empty() ? false : true;
}

std::exception_ptr multithread::getNextException() const
{
    if(thrown())
    {
        auto temp = std::move<std::exception_ptr&>(data->exceptions.front());
        data->exceptions.pop();
        return temp;
    }
    else
        throw std::out_of_range("Thrown() is false, no exception to get");
}
\$\endgroup\$
2
\$\begingroup\$

There's some weird stuff in there; I recommend trying to remove as much weird stuff as possible and then making a new post. Here's some examples:


template<init initType, is_invocable function>
void enqueueSafe(function&& item) noexcept
{
    REPORT("move",0);
    line.emplace_back(std::move<function&>(item));
}
  • Here init is an enum type and is_invocable is a concept, yet they have the same capitalization style. Prefer CamelCase for concepts, I'd say.

  • Similarly, function would normally be spelled with a short capitalized name, e.g. F. And why is it initType but is_invocable? Are we doing lowerCamelCase or snake_case?

  • You say noexcept, but this code executes a user-provided callback which is not necessarily noexcept. This is at least a yellow flag.

  • Now for the non-nitpicks. Template parameter initType is never used, and therefore at least should be unnamed. In fact, you should refactor this code to eliminate the unused parameter.

  • function&& is a forwarding reference, but you are calling std::move on it; that's a red flag. Use std::move with rvalue references, and std::forward with forwarding references, and never the twain shall meet.

  • You are passing an explicit template parameter to std::move; this is certainly wrong.

  • I'm fairly confident that in this case, it's okay to constrain function to be invocable, even though we can't tell if it'll be deduced as a reference type or not. In general, though, you should be very very paranoid about mixing Concepts with value categories. For example, void foo(Integral&& i) would do the wrong thing.

Putting all of this together, we get:

template<class F> requires Invocable<F>
void enqueueSafe(F&& item)
{
    REPORT("move", 0);
    line.emplace_back(std::forward<F>(item));
}

(I've switched to C++2a Working Draft syntax for illustrative purposes. You can switch it back to GCC terse syntax, of course.)


Again with the unused template parameters:

template< typename ... params>
void operator()(const char* file, int line, const char* func, 
                const char* note, int value )

params is unused, and therefore shouldn't exist.


template<init initType, is_vector vector>
void enqueueSafe(const vector& item)

template<close closeType>
execution close()

#define REPORT(...) REPORT( __FILE__, __LINE__, __func__, __VA_ARGS__);

You have a bad habit of reusing the exact same identifier for many different purposes: close here means two different things; vector is not std::vector (nor is it even the name of a class template); your REPORT macro looks like a bad joke. Try to find meaningful names for things that don't collide.


Finally, I'll point to one specific red flag:

    if (open==true)
    {
        open = false;
        ...

(Here open is the name of a std::atomic<bool> data member — naturally, not to be confused with close which is an enumeration type and/or member function!)

Anyone glancing at this code would say it looks like you ought to be doing

bool was_open = open.exchange(false);
if (was_open) {
    ...

Maybe you're confident that the test-and-check in your code is sufficient, and can't possibly introduce any bug or race condition. If so, you should either

  • add a code comment explaining exactly why your code is sufficient, or

  • just change it to use exchange anyway.

The benefit of the second alternative is that you don't have to write any comments to explain yourself — the code just looks right, so nobody will ask awkward questions. :)


I'm sure there are actual meaty things to review about the code itself, but I think all the weird stuff (std::move<foo&>(foo) and the like) is just too distracting. I'd say fix it up and re-post.

\$\endgroup\$
  • 1
    \$\begingroup\$ Thanks @Quuxplusone, I have addressed your critique and resubmitted for review. \$\endgroup\$ – Mercer Sep 11 '18 at 1:45

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.