1

I wrote a sample code to run parallel instances of for_each I am unable to join the threads, in the below code. I am little early to concurrent programming so im not sure if i have done everything right.

template <typename Iterator, typename F>
class for_each_block
{
public :
        void operator()(Iterator start, Iterator end, F f) {
            cout << this_thread::get_id << endl;
            this_thread::sleep_for(chrono::seconds(5));
            for_each(start, end, [&](auto& x) { f(x); });
    }
};

typedef unsigned const long int ucli;

template <typename Iterator, typename F>
void for_each_par(Iterator first, Iterator last, F f)
{
    ucli size = distance(first, last);
    if (!size)
        return;
    ucli min_per_thread = 4;
    ucli max_threads = (size + min_per_thread - 1) / min_per_thread;
    ucli hardware_threads = thread::hardware_concurrency();

    ucli no_of_threads = min(max_threads, hardware_threads != 0 ? hardware_threads : 4);

    ucli block_size = size / no_of_threads;

    vector<thread> vf(no_of_threads);
    Iterator block_start = first;
    for (int i = 0; i < (no_of_threads - 1); i++)
    {
        Iterator end = first;
        advance(end, block_size);
        vf.push_back(std::move(thread(for_each_block<Iterator, F>(),first,end,f)));
        first = end;
    }
    vf.push_back(std::move(thread(for_each_block<Iterator, F>(), first, last, f)));
    cout << endl;
    cout << vf.size() << endl;
    for(auto& x: vf)
    {
        if (x.joinable())
            x.join();
        else
            cout << "threads not joinable " << endl;
    }

    this_thread::sleep_for(chrono::seconds(100));
}

int main()
{
    vector<int> v1 = { 1,8,12,5,4,9,20,30,40,50,10,21,34,33 };
    for_each_par(v1.begin(), v1.end(), print_type<int>);
return 0;
}

In the above code i am getting threads not joinable. I have also tried with async futures still i get the same. Am i missing something here?

Any help is greatly appreciated , Thank you in advance ..

2 Answers 2

4
vector<thread> vf(no_of_threads);

This creates a vector with no_of_threads default-initialized threads. Since they're default initialized, none of them will be joinable. You probably meant to do:

vector<thread> vf;
vf.reserve(no_of_threads);

P.S.: std::move on a temporary is redundant :); consider changing this:

vf.push_back(std::move(thread(for_each_block<Iterator, F>(), first, last, f)));

to this:

vf.emplace_back(for_each_block<Iterator, F>(), first, last, f);
3
  • Thanks a lot, it works now. But i have one problem. All threads print same id. Can you also please suggest more(im sure i might have done somethings wrong way). Also i always had confusion for push_back and emplace_back thanks for clarifying :) Jul 3, 2016 at 8:53
  • 1
    @KartikV this_thread::get_id is a function, and it should be called. Right now you are printing function pointer value. Jul 3, 2016 at 8:58
  • @Ocelot so true, i feel so stupid. both of you have opened my eyes. I should look close and clear to avoid silly mistakes. Thank you guys. Jul 3, 2016 at 9:05
1

This may or may not be interesting. I had a go at refactoring the code to use what I think is a more idiomatic approach. I'm not saying that your approach is wrong, but since you're learning thread management I thought you may be interested in what else is possible.

Feel free to flame/question as appropriate. Comments inline:

#include <vector>
#include <chrono>
#include <thread>
#include <mutex>
#include <iomanip>
#include <future>

using namespace std;

//
// provide a means of serialising writing to a stream.
//
struct locker
{
    locker() : _lock(mutex()) {}

    static std::mutex& mutex() { static std::mutex m; return m; }
    std::unique_lock<std::mutex> _lock;
};
std::ostream& operator<<(std::ostream& os, const locker& l) {
    return os;
}

//
// fill in the missing work function
//
template<class T>
void print_type(const T& t) {
    std::cout << locker() << hex << std::this_thread::get_id() << " : " << dec << t << std::endl;
}

// put this in your personable library.
// the standards committee really should have given us ranges by now...
template<class I1, class I2>
struct range_impl
{
    range_impl(I1 i1, I2 i2) : _begin(i1), _end(i2) {};

    auto begin() const { return _begin; }
    auto end() const { return _end; }

    I1 _begin;
    I2 _end;
};

// distinct types because sometimes dissimilar iterators are comparable
template<class I1, class I2>
auto range(I1 i1, I2 i2) {
    return range_impl<I1, I2>(i1, i2);
}

//
// lets make a helper function so we can auto-deduce template args
//
template<class Iterator, typename F>
auto make_for_each_block(Iterator start, Iterator end, F&& f)
{
    // a lambda gives all the advantages of a function object with none
    // of the boilerplate.
    return [start, end, f = std::move(f)] {
        cout << locker() << this_thread::get_id() << endl;
        this_thread::sleep_for(chrono::seconds(1));

        // let's keep loops simple. for_each is a bit old-skool.
        for (auto& x : range(start, end)) {
            f(x);
        }
    };
}


template <typename Iterator, typename F>
void for_each_par(Iterator first, Iterator last, F f)
{
    if(auto size = distance(first, last))
    {
        std::size_t min_per_thread = 4;
        std::size_t max_threads = (size + min_per_thread - 1) / min_per_thread;
        std::size_t hardware_threads = thread::hardware_concurrency();

        auto no_of_threads = min(max_threads, hardware_threads != 0 ? hardware_threads : 4);

        auto block_size = size / no_of_threads;

        // futures give us two benefits:
        // 1. they automatically transmit exceptions
        // 2. no need for if(joinable) join. get is sufficient
        //
        vector<future<void>> vf;
        vf.reserve(no_of_threads - 1);
        for (auto count = no_of_threads ; --count ; )
        {
            //
            // I was thinking of refactoring this into std::generate_n but actually
            // it was less readable.
            //
            auto end = std::next(first, block_size);
            vf.push_back(async(launch::async, make_for_each_block(first, end, f)));
            first = end;
        }
        cout << locker() << endl << "threads: " << vf.size()  << " (+ main thread)" << endl;

        //
        // why spawn a thread for the remaining block? we may as well use this thread
        //
        /* auto partial_sum = */ make_for_each_block(first, last, f)();

        // join the threads
        // note that if the blocks returned a partial aggregate, we could combine them
        // here by using the values in the futures.
        for (auto& f : vf) f.get();
    }
}

int main()
{
    vector<int> v1 = { 1,8,12,5,4,9,20,30,40,50,10,21,34,33 };
    for_each_par(v1.begin(), v1.end(), print_type<int>);
    return 0;
}

sample output:

0x700000081000
0x700000104000

threads: 3 (+ main thread)
0x700000187000
0x100086000
0x700000081000 : 1
0x700000104000 : 5
0x700000187000 : 20
0x100086000 : 50
0x700000081000 : 8
0x700000104000 : 4
0x700000187000 : 30
0x100086000 : 10
0x700000081000 : 12
0x700000104000 : 9
0x700000187000 : 40
0x100086000 : 21
0x100086000 : 34
0x100086000 : 33
Program ended with exit code: 0

please explain std::move here: [start, end, f = std::move(f)] {...};

This is a welcome language feature that was made available in c++14. f = std::move(f) inside the capture block is equivalent to: decltype(f) new_f = std::move(f) except that the new variable is called f and not new_f. It allows us to std::move objects into lambdas rather than copy them.

For most function objects it won't matter - but some can large and this gives the compiler the opportunity to use a move rather than a copy if available.

3
  • It is very neat, locker, range are really cool. Can you please explain the move semantics you used for the function argument? Jul 3, 2016 at 17:05
  • Thanks a lot. Your helper functions are really useful, can u point me to any blog with more of such?(or may be you can write one if you get time ) Jul 3, 2016 at 19:23
  • @KartikV hmm. I tend not to follow c++ blogs. Most of them seem to be written by people who don't know the language very well. Ping me on [email protected] with questions, or you can have a look at some of my other answers here. There are some serious c++ experts on stack overflow. if I come across something I'll let you know. I have been thinking about running some courses when my next piece of work is finished (hopefully within he month). Jul 3, 2016 at 19:29

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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