Drain io_context after shutdown of plugins

application_base.cpp

@@ -185,7 +185,7 @@ void application_base::startup(boost::asio::io_service& io_serv) {
 
    } catch( ... ) {
       clean_up_signal_thread();
-      shutdown();
+      shutdown_plugins();
       throw;
    }
 
@@ -441,7 +441,7 @@ void application_base::handle_exception(std::exception_ptr eptr, std::string_vie
    }
 }
 
-void application_base::shutdown() {
+void application_base::shutdown_plugins() {
    std::exception_ptr eptr = nullptr;
 
    for(auto ritr = running_plugins.rbegin();
@@ -454,8 +454,17 @@ void application_base::shutdown() {
          handle_exception(std::current_exception(), (*ritr)->name());
       }
    }
-   for(auto ritr = running_plugins.rbegin();
-       ritr != running_plugins.rend(); ++ritr) {
+
+   // if we caught an exception while shutting down a plugin, rethrow it so that main()
+   // can catch it and report the error
+   if (eptr)
+      std::rethrow_exception(eptr);
+}
+
+void application_base::destroy_plugins() {
+   std::exception_ptr eptr = nullptr;
+
+   for(auto ritr = running_plugins.rbegin(); ritr != running_plugins.rend(); ++ritr) {
       try {
          plugins.erase((*ritr)->name());
       } catch(...) {
@@ -474,7 +483,6 @@ void application_base::shutdown() {
          eptr = std::current_exception();
       handle_exception(std::current_exception(), "plugin cleanup");
    }
-   quit();
 
    // if we caught an exception while shutting down a plugin, rethrow it so that main()
    // can catch it and report the error

include/appbase/application_base.hpp

@@ -115,7 +115,6 @@ class application_base {
    }
 
    void startup(boost::asio::io_service& io_serv);
-   void shutdown();
 
    /**
     *  Wait until quit(), SIGINT or SIGTERM and then shutdown.
@@ -126,15 +125,15 @@ class application_base {
       std::exception_ptr eptr = nullptr;
       {
          auto& io_serv{exec.get_io_service()};
-         boost::asio::io_service::work work(io_serv);
+         boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work = boost::asio::make_work_guard(io_serv);
          (void)work;
          bool more = true;
 
          while (more || io_serv.run_one()) {
-            if (is_quiting())
-               break;
             try {
                io_serv.poll(); // queue up any ready; allowing high priority item to get into the queue
+               if (io_serv.stopped())
+                  break;
                // execute the highest priority item
                more = exec.execute_highest();
             } catch (...) {
@@ -145,9 +144,22 @@ class application_base {
             }
          }
 
+         try {
+            shutdown_plugins();   // may rethrow exceptions
+         } catch (...) {
+            if (!eptr)
+               eptr = std::current_exception();
+         }
+
+         work.reset();
+         io_serv.restart();
+         // plugins shutdown down, drain io_context of anything posted while shutting down before destroying plugins
+         while (io_serv.poll())
+            ;
+
          try {
             exec.clear(); // make sure the queue is empty
-            shutdown();   // may rethrow exceptions
+            destroy_plugins();
          } catch (...) {
             if (!eptr)
                eptr = std::current_exception();
@@ -290,6 +302,9 @@ class application_base {
    }
    ///@}
 
+   void shutdown_plugins();
+   void destroy_plugins();
+
    application_base(std::shared_ptr<void>&& e); ///< protected because application is a singleton that should be accessed via instance()
 
    /// !!! must be dtor'ed after plugins

tests/shutdown_test.cpp

@@ -0,0 +1,79 @@
+#include <appbase/application.hpp>
+#include <thread>
+
+#include <boost/test/unit_test.hpp>
+
+namespace bpo = boost::program_options;
+using bpo::options_description;
+using bpo::variables_map;
+
+static bool thing = true;
+struct thing_better_be_alive {
+   ~thing_better_be_alive() noexcept(false) {
+      if(!thing)
+         throw "BOOM";
+   }
+};
+
+class thready_plugin : public appbase::plugin<thready_plugin> {
+   public:
+
+     template <typename Lambda>
+     void plugin_requires(Lambda&& l) {}
+
+     void set_program_options( options_description& cli, options_description& cfg ) override {}
+
+     void thread_work() {
+        boost::asio::post(ctx, [&]() {
+           thing_better_be_alive better_be;
+           boost::asio::post(appbase::app().get_io_service(), [&,is_it=std::move(better_be)]() {
+	      thread_work();
+	   });
+	});
+     }
+
+     void plugin_initialize( const variables_map& options ) {}
+     void plugin_startup()  {
+        for(unsigned i = 0; i < 48*4; i++)
+           thread_work();
+
+        for(unsigned i = 0; i < 48; ++i)
+           threads.emplace_back([this]() {
+              ctx.run();
+           });
+     }
+     void plugin_shutdown() {
+        usleep(100000);  //oh gee it takes a while to stop
+        ctx.stop();
+        for(unsigned i = 0; i < 48; ++i)
+          threads[i].join();
+     }
+
+     ~thready_plugin() {
+        thing = false;
+     }
+
+     boost::asio::io_context ctx;
+     boost::asio::executor_work_guard<boost::asio::io_context::executor_type> wg = boost::asio::make_work_guard(ctx);
+
+   private:
+     std::vector<std::thread> threads;
+};
+
+
+BOOST_AUTO_TEST_CASE(test_shutdown)
+{
+   appbase::application::register_plugin<thready_plugin>();
+   appbase::scoped_app app;
+
+   const char* argv[] = { "nodoes" };
+   if( !app->initialize<thready_plugin>( 1, const_cast<char**>(argv) ) )
+      return;
+   app->startup();
+   boost::asio::post(appbase::app().get_io_service(), [&](){
+      std::this_thread::sleep_for(std::chrono::milliseconds(5));
+      app->quit();
+   });
+   app->exec();
+
+}