个人中心
gpt4 book ai didi

c++ - 如何在 Boost Asio 上结合 strand wrapper 和 priority wrapper

转载 作者：搜寻专家 更新时间：2023-10-31 02:11:33 30 4
gpt4 key购买 nike

我想同时使用 Boost.Asio 的 strand 和 prioritized wrapper。

在我编写代码之前，我阅读了以下信息:

Boost asio priority and strand

boost::asio and Active Object

http://thread.gmane.org/gmane.comp.lib.boost.asio.user/3531

Why do I need strand per connection when using boost::asio?

我想使用包装方法，因为我想使用各种异步 API，例如 async_read、async_write 和 async_connect。根据http://thread.gmane.org/gmane.comp.lib.boost.asio.user/3531 , 貌似priority wrapper和strand wrapper可以结合起来。

所以我根据下面的例子写了代码:

http://www.boost.org/doc/libs/1_63_0/doc/html/boost_asio/example/cpp03/invocation/prioritised_handlers.cpp

这是我的代码:

#include <iostream>
#include <functional>
#include <queue>
#include <vector>
#include <thread>
#include <mutex>

#include <boost/asio.hpp>
#include <boost/optional.hpp>

#define ENABLE_STRAND 1
#define ENABLE_PRIORITY 1

class handler_priority_queue {
public:
    template <typename Handler>
    void add(int priority, Handler&& handler) {
        std::cout << "add(" << priority << ")" << std::endl;
        std::lock_guard<std::mutex> g(mtx_);
        handlers_.emplace(priority, std::forward<Handler>(handler));
    }

    void execute_all() {
        auto top = [&]() -> boost::optional<queued_handler> {
            std::lock_guard<std::mutex> g(mtx_);
            if (handlers_.empty()) return boost::none;
            boost::optional<queued_handler> opt = handlers_.top();
            handlers_.pop();
            return opt;
        };
        while (auto h_opt = top()) {
            h_opt.get().execute();
        }
    }

    template <typename Handler>
    class wrapped_handler {
    public:
        wrapped_handler(handler_priority_queue& q, int p, Handler h)
            : queue_(q), priority_(p), handler_(std::move(h))
        {
        }

        template <typename... Args>
        void operator()(Args&&... args) {
            std::cout << "operator() " << std::endl;
            handler_(std::forward<Args>(args)...);
        }

        //private:
        handler_priority_queue& queue_;
        int priority_;
        Handler handler_;
    };

    template <typename Handler>
    wrapped_handler<Handler> wrap(int priority, Handler&& handler) {
        return wrapped_handler<Handler>(*this, priority, std::forward<Handler>(handler));
    }

private:
    class queued_handler {
    public:
        template <typename Handler>
        queued_handler(int p, Handler&& handler)
            : priority_(p), function_(std::forward<Handler>(handler))
        {
            std::cout << "queued_handler()" << std::endl;
        }

        void execute() {
            std::cout << "execute(" << priority_ << ")" << std::endl;
            function_();
        }

        friend bool operator<(
            queued_handler const& lhs,
            queued_handler const & rhs) {
            return lhs.priority_ < rhs.priority_;
        }

    private:
        int priority_;
        std::function<void()> function_;
    };

    std::priority_queue<queued_handler> handlers_;
    std::mutex mtx_;
};

// Custom invocation hook for wrapped handlers.
template <typename Function, typename Handler>
void asio_handler_invoke(Function&& f,
                         handler_priority_queue::wrapped_handler<Handler>* h) {
    std::cout << "asio_handler_invoke " << std::endl;
    h->queue_.add(h->priority_, std::forward<Function>(f));
}

//----------------------------------------------------------------------

int main() {
    int const num_of_threads = 4;
    int const num_of_tasks = 5;

    boost::asio::io_service ios;
    boost::asio::strand strand(ios);


    handler_priority_queue pq;

    for (int i = 0; i != num_of_tasks; ++i) {
        ios.post(
#if ENABLE_STRAND
            strand.wrap(
#endif
#if ENABLE_PRIORITY
                pq.wrap(
                    i,
#endif
                    [=] {
                        std::cout << "[called] " << i << "," << std::this_thread::get_id() << std::endl;
                    }
#if ENABLE_PRIORITY
                )
#endif
#if ENABLE_STRAND
            )
#endif
        );
    }

    std::vector<std::thread> pool;
    for (int i = 0; i != num_of_threads; ++i) {
        pool.emplace_back([&]{
                std::cout << "before run_one()" << std::endl;
                while (ios.run_one()) {
                    std::cout << "before poll_one()" << std::endl;
                    while (ios.poll_one())
                        ;
                    std::cout << "before execute_all()" << std::endl;
                    pq.execute_all();
                }
            }
        );
    }
    for (auto& t : pool) t.join();
}

包装器由以下宏启用:

#define ENABLE_STRAND 1
#define ENABLE_PRIORITY 1

启用两个宏后，我得到以下结果:

before run_one()
asio_handler_invoke
add(0)
queued_handler()
before poll_one()
asio_handler_invoke
add(1)
queued_handler()
asio_handler_invoke
add(2)
queued_handler()
asio_handler_invoke
add(3)
queued_handler()
asio_handler_invoke
add(4)
queued_handler()
before execute_all()
execute(4)
execute(3)
execute(2)
execute(1)
execute(0)
before run_one()
before run_one()
before run_one()

我希望我得到

[called] priority,thread_id

输出为

[called] 1,140512649541376

但我没有得到它。

似乎在函数execute()中，调用了function_()，但没有调用wrapped_handler::operator()。 (函数 execute() 在我的代码中从 pq.execute_all(); 调用。)

void execute() {
    std::cout << "execute(" << priority_ << ")" << std::endl;
    function_(); // It is called.
}

template <typename Handler>
class wrapped_handler {
public:

    template <typename... Args>
    void operator()(Args&&... args) { // It is NOT called
        std::cout << "operator() " << std::endl;
        handler_(std::forward<Args>(args)...);
    }

我在 function_() 被调用后跟踪了序列。

调用了以下函数:

https://github.com/boostorg/asio/blob/boost-1.63.0/include/boost/asio/detail/wrapped_handler.hpp#L191 https://github.com/boostorg/asio/blob/boost-1.63.0/include/boost/asio/detail/wrapped_handler.hpp#L76 https://github.com/boostorg/asio/blob/boost-1.63.0/include/boost/asio/strand.hpp#L158 https://github.com/boostorg/asio/blob/boost-1.63.0/include/boost/asio/detail/impl/strand_service.hpp#L55 https://github.com/boostorg/asio/blob/boost-1.63.0/include/boost/asio/detail/impl/strand_service.ipp#L94

然后在函数 bool strand_service::do_dispatch(implementation_type& impl, operation* op) 中，操作 op 没有被调用，而是被插入了下一行的队列中:

https://github.com/boostorg/asio/blob/boost-1.63.0/include/boost/asio/detail/impl/strand_service.ipp#L111

我不确定为什么将 function_() 分派(dispatch)给 strand_service。我认为 strand wrapper 已经在我的代码中的以下位置展开:

template <typename Function, typename Handler>
void asio_handler_invoke(Function&& f,
                         handler_priority_queue::wrapped_handler<Handler>* h) {
    std::cout << "asio_handler_invoke " << std::endl;
    h->queue_.add(h->priority_, std::forward<Function>(f));
}

如果我只启用优先包装器，我会得到以下结果。看起来像我预期的那样工作。

before run_one()
asio_handler_invoke
add(0)
queued_handler()
before poll_one()
asio_handler_invoke
add(1)
queued_handler()
asio_handler_invoke
add(2)
queued_handler()
asio_handler_invoke
add(3)
queued_handler()
asio_handler_invoke
add(4)
queued_handler()
before execute_all()
execute(4)
operator()
[called] 4,140512649541376
execute(3)
operator()
[called] 3,140512649541376
execute(2)
operator()
[called] 2,140512649541376
execute(1)
operator()
[called] 1,140512649541376
execute(0)
operator()
[called] 0,140512649541376
before run_one()
before run_one()
before run_one()

如果我只启用 strand wrapper，我会得到以下结果。似乎也按我的预期工作。

before run_one()
[called] 0,140127385941760
before poll_one()
[called] 1,140127385941760
[called] 2,140127385941760
[called] 3,140127385941760
[called] 4,140127385941760
before execute_all()
before run_one()
before run_one()
before run_one()

有什么想法吗？

最佳答案

我解决了这个问题。

I'm not sure why the function_() is dispatched to strand_service. I think that strand wrapper has already been unwraped at the following point in my code:

template <typename Function, typename Handler>
void asio_handler_invoke(Function&& f,
                         handler_priority_queue::wrapped_handler<Handler>* h) {
    std::cout << "asio_handler_invoke " << std::endl;
    h->queue_.add(h->priority_, std::forward<Function>(f));
}

参数f是原始处理程序。这意味着优先队列包装和链包装处理程序。绞线包装在外面。所以当调用f时，它被分派(dispatch)给strand_service。此过程发生在同一个 strand_service 中，因此不会调用处理程序。

为了解决这个问题，将h->handler_加入优先级队列而不是f如下:

// Custom invocation hook for wrapped handlers.
template <typename Function, typename Handler>
void asio_handler_invoke(Function&& f,
                         handler_priority_queue::wrapped_handler<Handler>* h) {
    std::cout << "asio_handler_invoke " << std::endl;
    h->queue_.add(h->priority_, h->handler_);
}

handler_是类模板wrapped_handler的成员变量。它包含未包装的处理程序。

完整代码如下:

#include <iostream>
#include <functional>
#include <queue>
#include <vector>
#include <thread>
#include <mutex>

#include <boost/asio.hpp>
#include <boost/optional.hpp>

#define ENABLE_STRAND 1
#define ENABLE_PRIORITY 1

class handler_priority_queue {
public:
    template <typename Handler>
    void add(int priority, Handler&& handler) {
        std::cout << "add(" << priority << ")" << std::endl;
        std::lock_guard<std::mutex> g(mtx_);
        handlers_.emplace(priority, std::forward<Handler>(handler));
    }

    void execute_all() {
        auto top = [&]() -> boost::optional<queued_handler> {
            std::lock_guard<std::mutex> g(mtx_);
            if (handlers_.empty()) return boost::none;
            boost::optional<queued_handler> opt = handlers_.top();
            handlers_.pop();
            return opt;
        };
        while (auto h_opt = top()) {
            h_opt.get().execute();
        }
    }

    template <typename Handler>
    class wrapped_handler {
    public:
        template <typename HandlerArg>
        wrapped_handler(handler_priority_queue& q, int p, HandlerArg&& h)
            : queue_(q), priority_(p), handler_(std::forward<HandlerArg>(h))
        {
        }

        template <typename... Args>
        void operator()(Args&&... args) {
            std::cout << "operator() " << std::endl;
            handler_(std::forward<Args>(args)...);
        }

        //private:
        handler_priority_queue& queue_;
        int priority_;
        Handler handler_;
    };

    template <typename Handler>
    wrapped_handler<Handler> wrap(int priority, Handler&& handler) {
        return wrapped_handler<Handler>(*this, priority, std::forward<Handler>(handler));
    }

private:
    class queued_handler {
    public:
        template <typename Handler>
        queued_handler(int p, Handler&& handler)
            : priority_(p), function_(std::forward<Handler>(handler))
        {
            std::cout << "queued_handler()" << std::endl;
        }

        void execute() {
            std::cout << "execute(" << priority_ << ")" << std::endl;
            function_();
        }

        friend bool operator<(
            queued_handler const& lhs,
            queued_handler const & rhs) {
            return lhs.priority_ < rhs.priority_;
        }

    private:
        int priority_;
        std::function<void()> function_;
    };

    std::priority_queue<queued_handler> handlers_;
    std::mutex mtx_;
};

// Custom invocation hook for wrapped handlers.
template <typename Function, typename Handler>
void asio_handler_invoke(Function&& f,
                         handler_priority_queue::wrapped_handler<Handler>* h) {
    std::cout << "asio_handler_invoke " << std::endl;
    h->queue_.add(h->priority_, h->handler_);
}

//----------------------------------------------------------------------

int main() {
    int const num_of_threads = 4;
    int const num_of_tasks = 5;

    boost::asio::io_service ios;
    boost::asio::strand strand(ios);


    handler_priority_queue pq;

    for (int i = 0; i != num_of_tasks; ++i) {
        ios.post(
#if ENABLE_STRAND
            strand.wrap(
#endif
#if ENABLE_PRIORITY
                pq.wrap(
                    i,
#endif
                    [=] {
                        std::cout << "[called] " << i << "," << std::this_thread::get_id() << std::endl;
                    }
#if ENABLE_STRAND
                )
#endif
#if ENABLE_PRIORITY
            )
#endif
        );
    }

    std::vector<std::thread> pool;
    for (int i = 0; i != num_of_threads; ++i) {
        pool.emplace_back([&]{
                std::cout << "before run_one()" << std::endl;
                while (ios.run_one()) {
                    std::cout << "before poll_one()" << std::endl;
                    while (ios.poll_one())
                        ;
                    std::cout << "before execute_all()" << std::endl;
                    pq.execute_all();
                }
            }
        );
    }
    for (auto& t : pool) t.join();
}

这是一个输出:

before run_one()
asio_handler_invoke
add(0)
queued_handler()
before poll_one()
asio_handler_invoke
add(1)
queued_handler()
asio_handler_invoke
add(2)
queued_handler()
asio_handler_invoke
add(3)
queued_handler()
asio_handler_invoke
add(4)
queued_handler()
before execute_all()
execute(4)
[called] 4,139903315736320
execute(3)
[called] 3,139903315736320
execute(2)
[called] 2,139903315736320
execute(1)
[called] 1,139903315736320
execute(0)
[called] 0,139903315736320
before run_one()
before run_one()
before run_one()

关于c++ - 如何在 Boost Asio 上结合 strand wrapper 和 priority wrapper，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43385897/

30 4 0
文章推荐： c++ - 取消上传任务
文章推荐： c++ - 使用 cpprest 发送 POST
文章推荐： c++ - 如果一个静态成员没有被初始化并且成员类型是类本身怎么办？
文章推荐： c++ - 包装函数参数？
搜寻专家
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
全站热门文章
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com