3

所以你可以创建一个在被调用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"执行。

如果您调用f_waitable.get()，则管理任务的代码无法知道有人在等待未来。但如果你打电话f_deferred.wait(1ms);，你会future_status::deferred立即得到。

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

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

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

4

1 回答 1

0

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

免责声明：这是非常粗糙的。很多事情当然可以更优雅、更有效地完成。

#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})
于 2017-10-18T23:35:20.403 回答