3
\$\begingroup\$

Here the short story. I have a listener on a queue which sends me several jobs. Each job is launched in a thread and takes different times to finish. I have two main constraints :

  • I want to retrieve the result of the job when it finishes (hence the use of std::async rather than std::thread)
  • I want to have a collection of threads in the same kind of way as boost:thread_group but I cannot use a join_all/get_all method since I do not know when the job finishes and I do not want to be blocked.

So the idea chosen was to use the clean_ready_futures method which goes through the vector periodically and remove the ready futures such as proposed on this SO thread.

async_group.hpp

#pragma once

#include <vector>
#include <future>
#include <chrono>
#include <algorithm>
#include <mutex>

template<typename R>
bool is_future_ready(std::future<R> const& f) { 
    return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready; 
}

template <typename T>
struct async_group {
    std::vector<std::future<T>> futures;

    async_group()                               = default;
    async_group(const async_group&)             = delete;
    async_group& operator=(const async_group&)  = delete;
    async_group(async_group &&)                 = delete;

    // not working yet
    // template<class Function, class... Args>
    // void create_future(Function&& f, Args&&... args)  { 
    //     futures.emplace_back(std::async(
    //         std::launch::async,
    //         std::forward<Function>(f),
    //         std::forward<Args>(args)...
    //     ));
    // }
    void add_future(std::future<T>&& ret) { 
        std::lock_guard<std::mutex> lock{mutex}; 
        futures.emplace_back(std::move(ret)); 
    }

    std::size_t size() const { return futures.size(); }

    // Use this to have a similar behaviour as join_all()
    std::vector<T> get_all() {
        auto result = std::vector<T>{};
        result.reserve(size());
        for (auto & f : futures) {
            result.push_back(f.get());
        }
        std::lock_guard<std::mutex> lock{mutex};
        futures.clear();
        return result;
    }

    // Use this periodicall if you don't want to call get() and "detach" the asyncs
    std::vector<T> clean_ready_futures() {
        std::lock_guard<std::mutex> lock{mutex};
        // erase from futures vector and store ready futures in finished results
        auto finished_results = std::vector<T>{};
        futures.erase(
            std::remove_if(
                futures.begin(),
                futures.end(),
                [&finished_results](auto& f) -> bool {
                    if (is_future_ready(f)) { // future is ready then
                        finished_results.emplace_back(f.get()); // store the ready future in results
                        return true; // erase it from pending futures
                    }
                    return false;
                }
            ),
            futures.end()
        );

        return finished_results;
    }

private:
    std::mutex mutex;
};

illustative usage

#include <iostream>
#include "async_group.hpp"

std::string foo() {
    std::this_thread::sleep_for(std::chrono::milliseconds{200});
    return "foo";
}

std::string bar() {
    return "bar";
}

async_group<std::string>  pending_futures;

void launch_jobs() {
    pending_futures.add_future(std::async(std::launch::async, foo));
    pending_futures.add_future(std::async(std::launch::async, bar));
}

void clean_ready_jobs() {
    auto results = pending_futures.clean_ready_futures();
    std::cout << "Clean ready jobs: ";
    for (auto const& r : results) {
        std::cout << r << ' ';
    }
    std::cout << std::endl;
}

void get_all_jobs() {
    auto results = pending_futures.get_all();
    std::cout << "Clean ready jobs: ";
    for (auto const& r : results) {
        std::cout << r << ' ';
    }
    std::cout << std::endl;
}

int main() {
    launch_jobs();
    get_all_jobs(); // both foo and bar should be done
    launch_jobs();
    std::this_thread::sleep_for(std::chrono::milliseconds{10});
    clean_ready_jobs(); // bar should be done
    std::this_thread::sleep_for(std::chrono::milliseconds{200});
    clean_ready_jobs(); // foo should be done
}

Demo on Wandbox

Have I missed something ? How can this be improved ? I am looking forward to your insights.

\$\endgroup\$
  • \$\begingroup\$ I'm using Xcode 8.2 and the line if (is_future_ready(f)) { // future is ready then doesn't compile, while this may be a problem in my compiler, auto isn't guaranteed to work in lambda expressions arguments until C++14 according to everything I can find on the web. Example stackoverflow.com/questions/7709894/…. \$\endgroup\$ – pacmaninbw Jun 2 '17 at 14:19
  • 1
    \$\begingroup\$ Yes I will precise the C++14 tag. If you put [&finished_results](std::future<T>& f) instead of [&finished_results](auto& f) it will work \$\endgroup\$ – coincoin Jun 2 '17 at 14:19
1
\$\begingroup\$

We are assuming this is a multi threaded environemnt.

Thus all methods need to guard against access by a different thread and lock access to members.

// There is no lock here:
std::size_t size() const { return futures.size(); }

// There is no lock here:
std::vector<T> get_all() {
\$\endgroup\$
  • \$\begingroup\$ So adding std::lock_guard<std::mutex> lock{mutex};at the top scope of the method definition will do the work right ? \$\endgroup\$ – coincoin Jun 2 '17 at 20:03
  • \$\begingroup\$ @coincoin: Yes. \$\endgroup\$ – Martin York Jun 2 '17 at 21:51

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.