c++ - 我可以执行获取我的 `std::future` 并等待它吗？

Question

因此您可以创建一个 std::future，它在调用 .get() 之前不起作用:

auto f_deferred = std::async( std::launch::deferred, []{ std::cout << "I ran\n"; } );

您还可以编写可等待的 std::future，并且可以在任何线程中通过代码随时准备就绪:

std::packaged_task<void()> p( []( std::cout << "I also ran\n"; } );
auto f_waitable = p.get_future();

如果您调用 f_deferred.wait_for(1ms)，它不会费心等待。如果您调用 f_deferred.get()，您选择的 lambda(在本例中，将执行打印 "I ran\n" 的 lambda。

如果您调用 f_waitable.get()，则管理任务的代码无法知道有人正在等待 future 。但是，如果您调用 f_deferred.wait(1ms);，您只会立即获得 future_status::deferred。

有什么办法可以将这两者结合起来吗？

一个具体的用例是当人们排队任务时线程池返回 future 。如果未排队的 future 是 .get()，我想使用被阻塞的线程来执行任务而不是让它空闲。另一方面，我希望拥有返回 future 的人能够确定任务是否完成，甚至等待有限的时间让任务完成。 (在您等待的情况下，我可以接受您的线程在等待期间处于空闲状态)

如果做不到这一点，在即将到来的提案中是否有比让我的线程池返回一个具有所有限制的 future 更好地解决这个问题的解决方案？我听说 future 没有 future ， future 解决的问题存在更好的解决方案。

Answer 1

我不确定这是否正是您所需要的，但它可以说明我在评论中的建议。至少，如果它不能满足您的所有需求，我希望它能给您一些实现所需内容的想法。

免责声明:这是非常粗糙的。许多事情肯定可以更优雅、更高效地完成。

#include <iostream>
#include <thread>
#include <future>
#include <memory>
#include <functional>
#include <queue>
#include <random>
#include <chrono>
#include <mutex>

typedef std::packaged_task<void()> task;
typedef std::shared_ptr<task> task_ptr;
typedef std::lock_guard<std::mutex> glock;
typedef std::unique_lock<std::mutex> ulock;
typedef unsigned int uint;
typedef unsigned long ulong;

// For sync'd std::cout
std::mutex cout_mtx;

// For task scheduling
std::mutex task_mtx;
std::condition_variable task_cv;

// Prevents main() from exiting
// before the last worker exits
std::condition_variable kill_switch;

// RNG engine
std::mt19937_64 engine;

// Random sleep (in ms)
std::uniform_int_distribution<int> sleep(100, 10000);

// Task queue
std::queue<task_ptr> task_queue;

static uint tasks = 0;
static std::thread::id main_thread_id;
static uint workers = 0;

template<typename T>
class Task
{
    // Not sure if this needs
    // to be std::atomic.
    // A simple bool might suffice.
    std::atomic<bool> working;
    task_ptr tp;

public:

    Task(task_ptr _tp)
        :
          working(false),
          tp(_tp)
    {}

    inline T get()
    {
        working.store(true);
        (*tp)();
        return tp->get_future().get();
    }

    inline bool is_working()
    {
        return working.load();
    }
};

auto task_factory()
{
    return std::make_shared<task>([&]
    {
        uint task_id(0);
        {
            glock lk(cout_mtx);
            task_id = ++tasks;
            if (std::this_thread::get_id() == main_thread_id)
            {
                std::cout << "Executing task " << task_id << " in main thread.\n";
            }
            else
            {
                std::cout << "Executing task " << task_id << " in worker " << std::this_thread::get_id() << ".\n";
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(sleep(engine)));
        {
            glock lk(cout_mtx);
            std::cout << "\tTask " << task_id << " completed.\n";
        }
    });
}

auto func_factory()
{
    return [&]
    {

        while(true)
        {
            ulock lk(task_mtx);
            task_cv.wait(lk, [&]{ return !task_queue.empty(); });
            Task<void> task(task_queue.front());
            task_queue.pop();

            // Check if the task has been assigned
            if (!task.is_working())
            {
                // Sleep for a while and check again.
                // If it is still not assigned after 1 s,
                // start working on it.
                // You can also place these checks
                // directly in Task::get()
                {
                    glock lk(cout_mtx);
                    std::cout << "\tTask not started, waiting 1 s...\n";
                }
                lk.unlock();
                std::this_thread::sleep_for(std::chrono::milliseconds(1000));
                lk.lock();
                if (!task.is_working())
                {
                    {
                        glock lk(cout_mtx);
                        std::cout << "\tTask not started after 1 s, commencing work...\n";
                    }
                    lk.unlock();
                    task.get();
                    lk.lock();
                }

                if (task_queue.empty())
                {
                    break;
                }
            }
        }
    };
}

int main()
{
    engine.seed(std::chrono::high_resolution_clock::now().time_since_epoch().count());

    std::cout << "Main thread: " << std::this_thread::get_id() << "\n";
    main_thread_id = std::this_thread::get_id();

    for (int i = 0; i < 50; ++i)
    {
        task_queue.push(task_factory());
    }

    std::cout << "Tasks enqueued: " << task_queue.size() << "\n";

    // Spawn 5 workers
    for (int i = 0; i < 5; ++i)
    {
        std::thread([&]
        {
            {
                ulock lk(task_mtx);
                ++workers;
                task_cv.wait(lk);
                {
                    glock lk(cout_mtx);
                    std::cout << "\tWorker started\n";
                }
            }

            auto fn(func_factory());
            fn();

            ulock lk(task_mtx);
            --workers;
            if (workers == 0)
            {
                kill_switch.notify_all();
            }

        }).detach();
    }

    // Notify all workers to start processing the queue
    task_cv.notify_all();

    // This is the important bit:
    // Tasks can be executed by the main thread
    // as well as by the workers.
    // In fact, any thread can grab a task from the queue,
    // check if it is running and start working
    // on it if it is not.
    auto fn(func_factory());
    fn();

    ulock lk(task_mtx);
    if (workers > 0)
    {
        kill_switch.wait(lk);
    }

    return 0;
}

这是我的 CMakeLists.txt

cmake_minimum_required(VERSION 3.2)

project(tp_wait)

set(CMAKE_CXX_COMPILER "clang++")
set(CMAKE_CXX_STANDARD 14)
set(CMAKE_CXX_STANDARD_REQUIRED ON)

set(CMAKE_BUILD_TYPE "Debug" CACHE STRING "Build type" FORCE)

find_package(Threads REQUIRED)

add_executable(${PROJECT_NAME} "main.cpp")
target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT})