【问题标题】:Boost asio thread_pool join does not wait for tasks to be finishedBoost asio thread_pool join 不等待任务完成
【发布时间】:2020-08-03 07:23:21
【问题描述】:

考虑函数

#include <iostream>
#include <boost/bind.hpp>
#include <boost/asio.hpp>

void foo(const uint64_t begin, uint64_t *result)
{
    uint64_t prev[] = {begin, 0};
    for (uint64_t i = 0; i < 1000000000; ++i)
    {
        const auto tmp = (prev[0] + prev[1]) % 1000;
        prev[1] = prev[0];
        prev[0] = tmp;
    }
    *result = prev[0];
}

void batch(boost::asio::thread_pool &pool, const uint64_t a[])
{
    uint64_t r[] = {0, 0};
    boost::asio::post(pool, boost::bind(foo, a[0], &r[0]));
    boost::asio::post(pool, boost::bind(foo, a[1], &r[1]));

    pool.join();
    std::cerr << "foo(" << a[0] << "): " << r[0] << " foo(" << a[1] << "): " << r[1] << std::endl;
}

其中foo 是一个简单的“纯”函数,它对begin 执行计算并将结果写入指针*result。 使用来自batch 的不同输入调用此函数。在这里将每个调用分派到另一个 CPU 内核可能是有益的。

现在假设批处理函数被调用了几万次。因此,一个线程池会很好,它在所有顺序批处理调用之间共享。

试试这个(为了简单起见,只有 3 个调用)

int main(int argn, char **)
{
    boost::asio::thread_pool pool(2);

    const uint64_t a[] = {2, 4};
    batch(pool, a);

    const uint64_t b[] = {3, 5};
    batch(pool, b);

    const uint64_t c[] = {7, 9};
    batch(pool, c);
}

导致结果

foo(2): 2 foo(4): 4
富(3):0富(5):0
富(7):0富(9):0

所有三行同时出现,而 foo 的计算需要大约 3 秒。 我假设只有第一个 join 真正等待池完成所有工作。 其他的结果无效。 (未初始化的值) 这里重用线程池的最佳做法是什么?

【问题讨论】:

    标签: c++11 threadpool boost-asio


    【解决方案1】:

    最佳做法是不要重复使用池(如果您不断创建新池,池的用途是什么?)。

    如果您想确保将批次“计时”在一起,我建议在期货上使用 when_all

    Live On Coliru

    #define BOOST_THREAD_PROVIDES_FUTURE_WHEN_ALL_WHEN_ANY
    #include <iostream>
    #include <boost/bind.hpp>
    #include <boost/asio.hpp>
    #include <boost/thread.hpp>
    
    uint64_t foo(uint64_t begin) {
        uint64_t prev[] = {begin, 0};
        for (uint64_t i = 0; i < 1000000000; ++i) {
            const auto tmp = (prev[0] + prev[1]) % 1000;
            prev[1] = prev[0];
            prev[0] = tmp;
        }
        return prev[0];
    }
    
    void batch(boost::asio::thread_pool &pool, const uint64_t a[2])
    {
        using T = boost::packaged_task<uint64_t>;
    
        T tasks[] {
            T(boost::bind(foo, a[0])),
            T(boost::bind(foo, a[1])),
        };
    
        auto all = boost::when_all(
            tasks[0].get_future(),
            tasks[1].get_future());
    
        for (auto& t : tasks)
            post(pool, std::move(t));
    
        auto [r0, r1] = all.get();
        std::cerr << "foo(" << a[0] << "): " << r0.get() << " foo(" << a[1] << "): " << r1.get() << std::endl;
    }
    
    int main() {
        boost::asio::thread_pool pool(2);
    
        const uint64_t a[] = {2, 4};
        batch(pool, a);
    
        const uint64_t b[] = {3, 5};
        batch(pool, b);
    
        const uint64_t c[] = {7, 9};
        batch(pool, c);
    }
    

    打印

    foo(2): 2 foo(4): 4
    foo(3): 503 foo(5): 505
    foo(7): 507 foo(9): 509
    

    我会考虑

    • 概括
    • 消息队列

    广义的

    通过不对批量大小进行硬编码,使其更加灵活。毕竟,池大小已经固定,我们不需要“确保批次适合”之类的:

    Live On Coliru

    #define BOOST_THREAD_PROVIDES_FUTURE_WHEN_ALL_WHEN_ANY
    #include <iostream>
    #include <boost/bind.hpp>
    #include <boost/asio.hpp>
    #include <boost/thread.hpp>
    #include <boost/thread/future.hpp>
    
    struct Result { uint64_t begin, result; };
    
    Result foo(uint64_t begin) {
        uint64_t prev[] = {begin, 0};
        for (uint64_t i = 0; i < 1000000000; ++i) {
            const auto tmp = (prev[0] + prev[1]) % 1000;
            prev[1] = prev[0];
            prev[0] = tmp;
        }
        return { begin, prev[0] };
    }
    
    void batch(boost::asio::thread_pool &pool, std::vector<uint64_t> const a)
    {
        using T = boost::packaged_task<Result>;
        std::vector<T> tasks;
        tasks.reserve(a.size());
    
        for(auto begin : a)
            tasks.emplace_back(boost::bind(foo, begin));
    
        std::vector<boost::unique_future<T::result_type> > futures;
        for (auto& t : tasks) {
            futures.push_back(t.get_future());
            post(pool, std::move(t));
        }
    
        for (auto& fut : boost::when_all(futures.begin(), futures.end()).get()) {
            auto r = fut.get();
            std::cerr << "foo(" << r.begin << "): " << r.result << " ";
        }
        std::cout << std::endl;
    }
    
    int main() {
        boost::asio::thread_pool pool(2);
    
        batch(pool, {2});
        batch(pool, {4, 3, 5});
        batch(pool, {7, 9});
    }
    

    打印

    foo(2): 2 
    foo(4): 4 foo(3): 503 foo(5): 505 
    foo(7): 507 foo(9): 509 
    

    Generalized2:可变参数简化

    与普遍认为的相反(老实说,通常发生的情况),这次我们可以利用可变参数来摆脱所有中间向量(每一个):

    Live On Coliru

    void batch(boost::asio::thread_pool &pool, T... a)
    {
        auto launch = [&pool](uint64_t begin) {
            boost::packaged_task<Result> pt(boost::bind(foo, begin));
            auto fut = pt.get_future();
            post(pool, std::move(pt));
            return fut;
        };
    
        for (auto& r : {launch(a).get()...}) {
            std::cerr << "foo(" << r.begin << "): " << r.result << " ";
        }
    
        std::cout << std::endl;
    }
    

    如果你坚持及时输出结果,你仍然可以在组合中添加when_all(需要更多的英雄来解包元组):

    Live On Coliru

    template <typename...T>
    void batch(boost::asio::thread_pool &pool, T... a)
    {
        auto launch = [&pool](uint64_t begin) {
            boost::packaged_task<Result> pt(boost::bind(foo, begin));
            auto fut = pt.get_future();
            post(pool, std::move(pt));
            return fut;
        };
    
        std::apply([](auto&&... rfut) {
            Result results[] {rfut.get()...};
            for (auto& r : results) {
                std::cerr << "foo(" << r.begin << "): " << r.result << " ";
            }
        }, boost::when_all(launch(a)...).get());
    
        std::cout << std::endl;
    }
    

    两者仍然打印相同的结果

    消息队列

    这是非常自然的提升,并且可以跳过大多数复杂性。如果您还想按批次报告,则必须协调:

    Live On Coliru

    #include <iostream>
    #include <boost/asio.hpp>
    #include <memory>
    
    struct Result { uint64_t begin, result; };
    
    Result foo(uint64_t begin) {
        uint64_t prev[] = {begin, 0};
        for (uint64_t i = 0; i < 1000000000; ++i) {
            const auto tmp = (prev[0] + prev[1]) % 1000;
            prev[1] = prev[0];
            prev[0] = tmp;
        }
        return { begin, prev[0] };
    }
    
    using Group = std::shared_ptr<size_t>;
    void batch(boost::asio::thread_pool &pool, std::vector<uint64_t> begins) {
        auto group = std::make_shared<std::vector<Result> >(begins.size());
    
        for (size_t i=0; i < begins.size(); ++i) {
            post(pool, [i,begin=begins.at(i),group] {
                  (*group)[i] = foo(begin);
                  if (group.unique()) {
                      for (auto& r : *group) {
                          std::cout << "foo(" << r.begin << "): " << r.result << " ";
                          std::cout << std::endl;
                      }
                  }
              });
        }
    }
    
    int main() {
        boost::asio::thread_pool pool(2);
    
        batch(pool, {2});
        batch(pool, {4, 3, 5});
        batch(pool, {7, 9});
        pool.join();
    }
    

    请注意,这是对group 的并发访问,由于元素访问的限制,这是安全的。

    打印:

    foo(2): 2 
    foo(4): 4 foo(3): 503 foo(5): 505 
    foo(7): 507 foo(9): 509 
    

    【讨论】:

    • 添加了一个“分组消息队列”方法,只有 Asio,没有未来,更不用说 Boost 扩展了。
    • 为了矫枉过正,非分组消息队列(这次序列化访问控制台):coliru.stacked-crooked.com/a/0dac9dd96da9c1f2
    • 我刚刚意识到 Asio 附带了一个 fork_executor example,它正是这样做的:您可以“分组”任务并加入 执行程序(代表该组)而不是游泳池。由于 HTML 文档中没有列出任何执行程序示例,我已经错过了很长时间
    【解决方案2】:

    我刚刚遇到了这个隐藏在文档中的高级执行器示例:

    我刚刚意识到 Asio 附带了一个 fork_executor example,它正是这样做的:您可以“分组”任务并加入 执行程序(代表该组)而不是池。自从 HTML 文档中没有列出任何执行程序示例以来,我已经错过了很长时间 - sehe 21 mins ago

    所以事不宜迟,以下是适用于您问题的示例:

    Live On Coliru

    #define BOOST_BIND_NO_PLACEHOLDERS
    #include <boost/asio/thread_pool.hpp>
    #include <boost/asio/ts/executor.hpp>
    #include <condition_variable>
    #include <memory>
    #include <mutex>
    #include <queue>
    #include <thread>
    
    // A fixed-size thread pool used to implement fork/join semantics. Functions
    // are scheduled using a simple FIFO queue. Implementing work stealing, or
    // using a queue based on atomic operations, are left as tasks for the reader.
    class fork_join_pool : public boost::asio::execution_context {
      public:
        // The constructor starts a thread pool with the specified number of
        // threads. Note that the thread_count is not a fixed limit on the pool's
        // concurrency. Additional threads may temporarily be added to the pool if
        // they join a fork_executor.
        explicit fork_join_pool(std::size_t thread_count = std::thread::hardware_concurrency()*2)
                : use_count_(1), threads_(thread_count)
        {
            try {
                // Ask each thread in the pool to dequeue and execute functions
                // until it is time to shut down, i.e. the use count is zero.
                for (thread_count_ = 0; thread_count_ < thread_count; ++thread_count_) {
                    boost::asio::dispatch(threads_, [&] {
                        std::unique_lock<std::mutex> lock(mutex_);
                        while (use_count_ > 0)
                            if (!execute_next(lock))
                                condition_.wait(lock);
                    });
                }
            } catch (...) {
                stop_threads();
                threads_.join();
                throw;
            }
        }
    
        // The destructor waits for the pool to finish executing functions.
        ~fork_join_pool() {
            stop_threads();
            threads_.join();
        }
    
      private:
        friend class fork_executor;
    
        // The base for all functions that are queued in the pool.
        struct function_base {
            std::shared_ptr<std::size_t> work_count_;
            void (*execute_)(std::shared_ptr<function_base>& p);
        };
    
        // Execute the next function from the queue, if any. Returns true if a
        // function was executed, and false if the queue was empty.
        bool execute_next(std::unique_lock<std::mutex>& lock) {
            if (queue_.empty())
                return false;
            auto p(queue_.front());
            queue_.pop();
            lock.unlock();
            execute(lock, p);
            return true;
        }
    
        // Execute a function and decrement the outstanding work.
        void execute(std::unique_lock<std::mutex>& lock,
                     std::shared_ptr<function_base>& p) {
            std::shared_ptr<std::size_t> work_count(std::move(p->work_count_));
            try {
                p->execute_(p);
                lock.lock();
                do_work_finished(work_count);
            } catch (...) {
                lock.lock();
                do_work_finished(work_count);
                throw;
            }
        }
    
        // Increment outstanding work.
        void
        do_work_started(const std::shared_ptr<std::size_t>& work_count) noexcept {
            if (++(*work_count) == 1)
                ++use_count_;
        }
    
        // Decrement outstanding work. Notify waiting threads if we run out.
        void
        do_work_finished(const std::shared_ptr<std::size_t>& work_count) noexcept {
            if (--(*work_count) == 0) {
                --use_count_;
                condition_.notify_all();
            }
        }
    
        // Dispatch a function, executing it immediately if the queue is already
        // loaded. Otherwise adds the function to the queue and wakes a thread.
        void do_dispatch(std::shared_ptr<function_base> p,
                         const std::shared_ptr<std::size_t>& work_count) {
            std::unique_lock<std::mutex> lock(mutex_);
            if (queue_.size() > thread_count_ * 16) {
                do_work_started(work_count);
                lock.unlock();
                execute(lock, p);
            } else {
                queue_.push(p);
                do_work_started(work_count);
                condition_.notify_one();
            }
        }
    
        // Add a function to the queue and wake a thread.
        void do_post(std::shared_ptr<function_base> p,
                     const std::shared_ptr<std::size_t>& work_count) {
            std::lock_guard<std::mutex> lock(mutex_);
            queue_.push(p);
            do_work_started(work_count);
            condition_.notify_one();
        }
    
        // Ask all threads to shut down.
        void stop_threads() {
            std::lock_guard<std::mutex> lock(mutex_);
            --use_count_;
            condition_.notify_all();
        }
    
        std::mutex mutex_;
        std::condition_variable condition_;
        std::queue<std::shared_ptr<function_base>> queue_;
        std::size_t use_count_;
        std::size_t thread_count_;
        boost::asio::thread_pool threads_;
    };
    
    // A class that satisfies the Executor requirements. Every function or piece of
    // work associated with a fork_executor is part of a single, joinable group.
    class fork_executor {
      public:
        fork_executor(fork_join_pool& ctx)
                : context_(ctx), work_count_(std::make_shared<std::size_t>(0)) {}
    
        fork_join_pool& context() const noexcept { return context_; }
    
        void on_work_started() const noexcept {
            std::lock_guard<std::mutex> lock(context_.mutex_);
            context_.do_work_started(work_count_);
        }
    
        void on_work_finished() const noexcept {
            std::lock_guard<std::mutex> lock(context_.mutex_);
            context_.do_work_finished(work_count_);
        }
    
        template <class Func, class Alloc>
        void dispatch(Func&& f, const Alloc& a) const {
            auto p(std::allocate_shared<exFun<Func>>(
                typename std::allocator_traits<Alloc>::template rebind_alloc<char>(a),
                std::move(f), work_count_));
            context_.do_dispatch(p, work_count_);
        }
    
        template <class Func, class Alloc> void post(Func f, const Alloc& a) const {
            auto p(std::allocate_shared<exFun<Func>>(
                typename std::allocator_traits<Alloc>::template rebind_alloc<char>(a),
                std::move(f), work_count_));
            context_.do_post(p, work_count_);
        }
    
        template <class Func, class Alloc>
        void defer(Func&& f, const Alloc& a) const {
            post(std::forward<Func>(f), a);
        }
    
        friend bool operator==(const fork_executor& a, const fork_executor& b) noexcept {
            return a.work_count_ == b.work_count_;
        }
    
        friend bool operator!=(const fork_executor& a, const fork_executor& b) noexcept {
            return a.work_count_ != b.work_count_;
        }
    
        // Block until all work associated with the executor is complete. While it
        // is waiting, the thread may be borrowed to execute functions from the
        // queue.
        void join() const {
            std::unique_lock<std::mutex> lock(context_.mutex_);
            while (*work_count_ > 0)
                if (!context_.execute_next(lock))
                    context_.condition_.wait(lock);
        }
    
      private:
        template <class Func> struct exFun : fork_join_pool::function_base {
            explicit exFun(Func f, const std::shared_ptr<std::size_t>& w)
                    : function_(std::move(f)) {
                work_count_ = w;
                execute_ = [](std::shared_ptr<fork_join_pool::function_base>& p) {
                    Func tmp(std::move(static_cast<exFun*>(p.get())->function_));
                    p.reset();
                    tmp();
                };
            }
    
            Func function_;
        };
    
        fork_join_pool& context_;
        std::shared_ptr<std::size_t> work_count_;
    };
    
    // Helper class to automatically join a fork_executor when exiting a scope.
    class join_guard {
      public:
        explicit join_guard(const fork_executor& ex) : ex_(ex) {}
        join_guard(const join_guard&) = delete;
        join_guard(join_guard&&) = delete;
        ~join_guard() { ex_.join(); }
    
      private:
        fork_executor ex_;
    };
    
    //------------------------------------------------------------------------------
    
    #include <algorithm>
    #include <iostream>
    #include <random>
    #include <vector>
    #include <boost/bind.hpp>
    
    static void foo(const uint64_t begin, uint64_t *result)
    {
        uint64_t prev[] = {begin, 0};
        for (uint64_t i = 0; i < 1000000000; ++i) {
            const auto tmp = (prev[0] + prev[1]) % 1000;
            prev[1] = prev[0];
            prev[0] = tmp;
        }
        *result = prev[0];
    }
    
    void batch(fork_join_pool &pool, const uint64_t (&a)[2])
    {
        uint64_t r[] = {0, 0};
        {
            fork_executor fork(pool);
            join_guard join(fork);
            boost::asio::post(fork, boost::bind(foo, a[0], &r[0]));
            boost::asio::post(fork, boost::bind(foo, a[1], &r[1]));
            // fork.join(); // or let join_guard destructor run
        }
        std::cerr << "foo(" << a[0] << "): " << r[0] << " foo(" << a[1] << "): " << r[1] << std::endl;
    }
    
    int main() {
        fork_join_pool pool;
    
        batch(pool, {2, 4});
        batch(pool, {3, 5});
        batch(pool, {7, 9});
    }
    

    打印:

    foo(2): 2 foo(4): 4
    foo(3): 503 foo(5): 505
    foo(7): 507 foo(9): 509
    

    注意事项:

    • 执行器可以重叠/嵌套:您可以在单个 fork_join_pool 上使用多个可连接的 fork_executor,它们将为每个执行器加入不同的任务组

    查看库示例(执行递归分治合并排序)时,您会很容易理解。

    【讨论】:

    • 您说:“最好的做法是不要重复使用池”。这适用于您广泛使用线程池的用例吗?会不会每次vrs创建新线程时都创建一个新线程池在执行时间上不一样?
    • @user2987773 当然,这就是为什么您总是创建池或线程。相反,您将游泳池设计为耐用的。自从我回答以来,我确实学习了一个新功能:boost.org/doc/libs/1_78_0/doc/html/boost_asio/reference/…这至少可以扩大池。
    • 我从来没有问过为什么复用池不好。
    • @user2987773 很好,这就是为什么回答你在评论中提出的问题¯\_(ツ)_/¯ 也许我误解了你的意思
    • 除非我不太明白,因为我认为不重用池意味着您必须在每次使用后重新创建池,这会破坏目的。如前所述,您如何将游泳池设计为耐用的。 boost::wait_for_all(..) 也将修复这篇文章中的问题:“等待任务完成”或者它有一些缺点。
    猜你喜欢
    • 2020-05-22
    • 2021-12-22
    • 2021-08-05
    • 2015-11-16
    • 2023-03-10
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多