c++ - boost::asio bug? task_io_service before destruction of io_service -
i found weird bug inside code. here's self contained test case managed down to.
#include <memory> #include <thread> #include <stack> #include <system_error> #include <boost/asio.hpp> using boost::asio::io_service; using std::placeholders::_1; class async_service { public: async_service(); async_service(size_t number_threads); ~async_service(); async_service(const async_service&) = delete; void operator=(const async_service&) = delete; void spawn(); void shutdown(); io_service& get_service(); const io_service& get_service() const; private: io_service service_; io_service::work* work_; std::vector<std::thread> threads_; }; async_service::async_service() : work_(nullptr) { } async_service::async_service(size_t number_threads) : work_(nullptr) { (size_t = 0; < number_threads; ++i) spawn(); } async_service::~async_service() { std::cout << __pretty_function__ << std::endl; service_.stop(); (std::thread& t: threads_) t.join(); } void run_service(io_service* service) { service->run(); } void async_service::spawn() { if (!work_) work_ = new io_service::work(service_); threads_.push_back(std::thread(run_service, &service_)); } void async_service::shutdown() { delete work_; work_ = nullptr; } io_service& async_service::get_service() { return service_; } const io_service& async_service::get_service() const { return service_; } // -------------------------------------------------------------- template <typename... args> class subscriber : public std::enable_shared_from_this<subscriber<args...>> { public: typedef std::function<void (args...)> handler_type; typedef std::shared_ptr<subscriber<args...>> ptr; subscriber(async_service& service) : strand_(service.get_service()) { } void subscribe(handler_type handle) { strand_.dispatch( std::bind(&subscriber<args...>::do_subscribe, this->shared_from_this(), handle)); } void relay(args... params) { strand_.dispatch( std::bind(&subscriber<args...>::do_relay, this->shared_from_this(), std::forward<args>(params)...)); } private: typedef std::stack<handler_type> registry_stack; void do_subscribe(handler_type handle) { registry_.push(handle); } void do_relay(args... params) { registry_stack notify_copy = std::move(registry_); registry_ = registry_stack(); while (!notify_copy.empty()) { notify_copy.top()(params...); notify_copy.pop(); } assert(notify_copy.empty()); } io_service::strand strand_; registry_stack registry_; }; // -------------------------------------------------------- class lala_channel_proxy : public std::enable_shared_from_this<lala_channel_proxy> { public: typedef std::function<void (const std::error_code&)> receive_inventory_handler; lala_channel_proxy(async_service& service) : strand_(service.get_service()) { inventory_subscriber_ = std::make_shared<inventory_subscriber_type>(service); } void start() { read_header(); } void subscribe_inventory(receive_inventory_handler handle_receive) { inventory_subscriber_->subscribe(handle_receive); } typedef subscriber<const std::error_code&> inventory_subscriber_type; void read_header() { strand_.post( std::bind(&lala_channel_proxy::handle_read_header, shared_from_this(), boost::system::error_code(), 0)); } void handle_read_header(const boost::system::error_code& ec, size_t bytes_transferred) { std::cout << "inventory ----------" << std::endl; inventory_subscriber_->relay(std::error_code()); sleep(1.0); read_header(); } io_service::strand strand_; inventory_subscriber_type::ptr inventory_subscriber_; }; typedef std::shared_ptr<lala_channel_proxy> lala_channel_proxy_ptr; class lala_channel { public: lala_channel(async_service& service) { lala_channel_proxy_ptr proxy = std::make_shared<lala_channel_proxy>(service); proxy->start(); //weak_proxy_ = proxy; strong_proxy_ = proxy; } void subscribe_inventory( lala_channel_proxy::receive_inventory_handler handle_receive) { lala_channel_proxy_ptr proxy = strong_proxy_; proxy->subscribe_inventory(handle_receive); } lala_channel_proxy_ptr strong_proxy_; // has weak pointer channel pimpl allow // die, whether uses weak_ptr or shared_ptr makes // no difference. //std::weak_ptr<channel_proxy> weak_proxy_; }; typedef std::shared_ptr<lala_channel> lala_channel_ptr; //typedef lala_channel_proxy_ptr lala_channel_ptr; class session : public std::enable_shared_from_this<session> { public: typedef std::function<void (const std::error_code&)> completion_handler; session(async_service& service, async_service& mempool_service, async_service& disk_service) : strand_(service.get_service()), txpool_strand_(mempool_service.get_service()), chain_strand_(disk_service.get_service()), service_(service) { } void start() { auto this_ptr = shared_from_this(); lala_channel_ptr node = std::make_shared<lala_channel>(service_); node->subscribe_inventory( std::bind(&session::inventory, this_ptr, _1, node)); (size_t = 0; < 500; ++i) { chain_strand_.post( []() { std::cout << "here!" << std::endl; sleep(2); }); } } private: void inventory(const std::error_code& ec, lala_channel_ptr node) { if (ec) { std::cerr << ec.message() << std::endl; return; } auto this_ptr = shared_from_this(); txpool_strand_.post([]() {}); node->subscribe_inventory( std::bind(&session::inventory, this_ptr, _1, node)); } async_service& service_; io_service::strand txpool_strand_, strand_, chain_strand_; }; int main() { // first level { // bug happens ordering of async_service's // means triggered when destroyed in // reverse order. async_service network_service(1), disk_service(1), mempool_service(1); //async_service network_service(1), mempool_service(1), disk_service(1); //async_service disk_service(1), mempool_service(1), network_service(1); //async_service disk_service(1), network_service(1), mempool_service(1); //async_service mempool_service(1), disk_service(1), network_service(1); //async_service mempool_service(1), network_service(1), disk_service(1); // second level { // should kept alive io_service auto s = std::make_shared<session>(network_service, mempool_service, disk_service); s->start(); } //network_service.shutdown(); //disk_service.shutdown(); //mempool_service.shutdown(); sleep(3); // never gets past here } std::cout << "exiting..." << std::endl; return 0; } when run it, this:
$ g++ -std=c++0x /tmp/ideone_y6oli.cpp -lboost_system -pthread -ggdb $ gdb a.out gnu gdb (ubuntu/linaro 7.4-2012.04-0ubuntu2) 7.4-2012.04 copyright (c) 2012 free software foundation, inc. license gplv3+: gnu gpl version 3 or later <http://gnu.org/licenses/gpl.html> free software: free change , redistribute it. there no warranty, extent permitted law. type "show copying" , "show warranty" details. gdb configured "x86_64-linux-gnu". bug reporting instructions, please see: <http://bugs.launchpad.net/gdb-linaro/>... reading symbols /home/genjix/src/brokenlibbtc/a.out...done. (gdb) r starting program: /home/genjix/src/brokenlibbtc/a.out [thread debugging using libthread_db enabled] using host libthread_db library "/lib/x86_64-linux-gnu/libthread_db.so.1". [new thread 0x7ffff6deb700 (lwp 28098)] [new thread 0x7ffff65ea700 (lwp 28099)] [new thread 0x7ffff5de9700 (lwp 28100)] inventory ---------- here! inventory ---------- here! inventory ---------- async_service::~async_service() async_service::~async_service() [thread 0x7ffff5de9700 (lwp 28100) exited] program received signal sigsegv, segmentation fault. [switching thread 0x7ffff6deb700 (lwp 28098)] 0x0000000000405873 in boost::asio::detail::task_io_service::wake_one_idle_thread_and_unlock (this=0x6255e0, lock=...) @ /usr/include/boost/asio/detail/impl/task_io_service.ipp:461 461 first_idle_thread_ = idle_thread->next; same thing boost 1.48 , 1.49.
i wonder why happening. happens highly particular configuration. if change bug not occur.
async_service convenience wrapper around io_service. strangely if change io_service *io_service , not delete io_service error doesn't happen... surely should not matter?
if @ sourcecode in main(), there 3 async_service objects created. each 1 of them manages lifetime of single io_service.
// bug happens ordering of async_service's // means triggered when destroyed in // reverse order. async_service network_service(1), disk_service(1), mempool_service(1); //async_service network_service(1), mempool_service(1), disk_service(1); //async_service disk_service(1), mempool_service(1), network_service(1); //async_service disk_service(1), network_service(1), mempool_service(1); //async_service mempool_service(1), disk_service(1), network_service(1); //async_service mempool_service(1), network_service(1), disk_service(1); the subscriber class represents subscribe ... call thing particular event. session , channel thing adapted larger program might seem big tangled/confusing.
one problem session::inventory, when executed thread under first argument constructor (network_service in failing case), attempts access strand initialized using second argument (mempool_service) .
void inventory(const std::error_code& ec, lala_channel_ptr node) { if (ec) { std::cerr << ec.message() << std::endl; return; } auto this_ptr = shared_from_this(); txpool_strand_.post([]() {}); // <-- 1 problem here. node->subscribe_inventory( std::bind(&session::inventory, this_ptr, _1, node)); } given order of destruction, mempool_service has been destroyed, , access there fail somewhere during execution of post.
Comments
Post a Comment