当 std::future 准备好被检索时向主线程发出信号

Question

我正在尝试了解std::async、std::future 系统。我不太明白的是你如何处理运行多个异步“任务”，然后根据返回的第一个、第二个等，运行一些额外的代码。

示例：假设您的主线程处于一个简单的循环中。现在，根据用户输入，您通过std::async 运行多个函数，并将期货保存在std::list 中。

我的问题是，我如何从std::async 函数传回可以指定哪个未来已完成的信息？

我的主线程基本上处于消息循环中，我需要做的是让std::async 运行的函数能够将一条消息排队，该消息以某种方式指定哪个未来已完成。问题是该函数无法访问未来。

我只是错过了什么吗？

这是我想要完成的一些伪代码；如果有办法也有办法使用取消令牌调用“取消”请求，则可以加分。

class RequestA
{
public:
    int input1;

    int output1;
};

main()
{
    while(1)
    {
       //check for completion
       // i.e. pop next "message"
       if(auto *completed_task = get_next_completed_task())
       {
          completed_task->run_continuation();
       }

       // other code to handle user input
       if(userSaidRunA())
       {
          // note that I don't want to use a raw pointer but
          // am not sure how to use future for this
          RequestA *a = new RequestA();
          run(a, OnRequestTypeAComplete);
       }

    }
}

void OnRequestTypeAComplete(RequestA &req)
{
    // Do stuff with req, want access to inputs and output
}

Answer 1

不幸的是，C++11 std::future 不提供延续和取消。您只能从std::future 检索结果一次。此外，future 从其析构函数中的std::async 块返回。有一个由 Adob​​e 的 Sean Parent 领导的小组。他们按照应有的方式实现了future、async、task。还具有像when_all、when_any 这样的延续功能。可能是你正在寻找的东西。无论如何看看this project。代码质量好，易于阅读。

如果平台相关的解决方案也适合您，您可以检查它们。对于 Windows，我知道 PPL 库。它还具有取消和继续的原语。

Answer 2

您可以创建一个包含标志的struct，并将对该标志的引用传递给您的线程函数。

有点像这样：

int stuff(std::atomic_bool& complete, std::size_t id)
{
    std::cout << "starting: " << id << '\n';

    // do stuff
    std::this_thread::sleep_for(std::chrono::milliseconds(hol::random_number(3000)));

    // generate value
    int value = hol::random_number(30);

    // signal end
    complete = true;
    std::cout << "ended: " << id << " -> " << value << '\n';

    return value;
}

struct task
{
    std::future<int> fut;
    std::atomic_bool complete;

    task() = default;
    task(task&& t): fut(std::move(t.fut)), complete(t.complete.load()) {}
};

int main()
{
    // list of tasks
    std::vector<task> tasks;

    // reserve enough spaces so that nothing gets reallocated
    // as that would invalidate the references to the atomic_bools
    // needed to signal the end of a thread
    tasks.reserve(3);

    // create a new task
    tasks.emplace_back();

    // start it running
    tasks.back().fut = std::async(std::launch::async, stuff, std::ref(tasks.back().complete), tasks.size());

    tasks.emplace_back();
    tasks.back().fut = std::async(std::launch::async, stuff, std::ref(tasks.back().complete), tasks.size());

    tasks.emplace_back();
    tasks.back().fut = std::async(std::launch::async, stuff, std::ref(tasks.back().complete), tasks.size());

    // Keep going as long as any of the tasks is incomplete
    while(std::any_of(std::begin(tasks), std::end(tasks),
        [](auto& t){ return !t.complete.load(); }))
    {

        // do some parallel stuff
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }

    // process the results

    int sum = 0;
    for(auto&& t: tasks)
        sum += t.fut.get();

    std::cout << "sum: " << sum << '\n';
}

Answer 3

这里使用std::unordered_map 而不是std::list 的解决方案，您无需修改​​callables。取而代之的是，您使用一个辅助函数，该函数为每个 task 分配一个 id 并在它们完成时通知：

class Tasks {
public:
    /*
     *  Helper to create the tasks in a safe way.
     *  lockTaskCreation is needed to guarantee newTask is (temporarilly)
     *  assigned before it is moved to the list of tasks
     */
    template <class R, class ...Args>
    void createNewTask(const std::function<R(Args...)>& f, Args... args) {
        std::unique_lock<std::mutex> lock(mutex);
        std::lock_guard<std::mutex> lockTaskCreation(mutexTaskCreation);
        newTask = std::async(std::launch::async, executeAndNotify<R, Args...>,
            std::move(lock), f, std::forward<Args>(args)...);
    }

private:
    /*
     *  Assign an id to the task, execute it, and notify when finishes
     */
    template <class R, class ...Args>
    static R executeAndNotify(std::unique_lock<std::mutex> lock,
        const std::function<R(Args...)>& f, Args... args)
    {
        {
            std::lock_guard<std::mutex> lockTaskCreation(mutexTaskCreation);
            tasks[std::this_thread::get_id()] = std::move(newTask);
        }
        lock.unlock();
        Notifier notifier;
        return f(std::forward<Args>(args)...);
    }

    /*
     *  Class to notify when a task is completed (follows RAII)
     */
    class Notifier {
    public:
        ~Notifier() {
            std::lock_guard<std::mutex> lock(mutex);
            finishedTasks.push(std::this_thread::get_id());
            cv.notify_one();
        }
    };

    /*
     *  Wait for a finished task.
     *  This function needs to be called in an infinite loop
     */
    static void waitForFinishedTask() {
        std::unique_lock<std::mutex> lock(mutex);
        cv.wait(lock, [] { return finishedTasks.size() || finish; });
        if (finishedTasks.size()) {
            auto threadId = finishedTasks.front();
            finishedTasks.pop();
            auto result = tasks.at(threadId).get();
            tasks.erase(threadId);
            std::cout << "task " << threadId
                << " returned: " << result << std::endl;
        }
    }

    static std::unordered_map<std::thread::id, std::future<int>> tasks;
    static std::mutex mutex;
    static std::mutex mutexTaskCreation;
    static std::queue<std::thread::id> finishedTasks;
    static std::condition_variable cv;
    static std::future<int> newTask;

    ...
};

...

那么，你可以这样调用异步任务：

int doSomething(int i) {
    std::this_thread::sleep_for(std::chrono::milliseconds(1));
    return i;
}

int main() {
    Tasks tasks;
    tasks.createNewTask(std::function<decltype(doSomething)>(doSomething), 10);
    return 0;
}

---

See a complete implementation run on Coliru