Fixing shutdown when parcels are still in flight

hpx/plugins/parcelport/tcp/receiver.hpp

@@ -58,6 +58,7 @@ namespace hpx { namespace parcelset { namespace policies { namespace tcp
           , parcelport_(parcelport)
           , timer_()
           , mtx_()
+          , operation_in_flight_(0)
         {}
 
         ~receiver()
@@ -128,6 +129,13 @@ namespace hpx { namespace parcelset { namespace policies { namespace tcp
                 socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
                 socket_.close(ec);    // close the socket to give it back to the OS
             }
+
+            while(operation_in_flight_ != 0)
+            {
+                if(threads::get_self_ptr())
+                    hpx::this_thread::suspend(hpx::threads::pending,
+                        "tcp::reveiver::shutdown");
+            }
         }
 
     private:
@@ -137,13 +145,15 @@ namespace hpx { namespace parcelset { namespace policies { namespace tcp
         void handle_read_header(boost::system::error_code const& e,
             std::size_t bytes_transferred, Handler handler)
         {
+            HPX_ASSERT(operation_in_flight_ == 0);
             if (e) {
                 handler(e);
 
                 // Issue a read operation to read the next parcel.
 //                 async_read(handler);
             }
             else {
+                ++operation_in_flight_;
                 // Determine the length of the serialized data.
                 std::uint64_t inbound_size = buffer_.size_;
 
@@ -231,6 +241,7 @@ namespace hpx { namespace parcelset { namespace policies { namespace tcp
         {
             if (e) {
                 handler(e);
+                --operation_in_flight_;
 
                 // Issue a read operation to read the next parcel.
 //                 async_read(handler);
@@ -288,6 +299,7 @@ namespace hpx { namespace parcelset { namespace policies { namespace tcp
         {
             if (e) {
                 handler(e);
+                --operation_in_flight_;
 
                 // Issue a read operation to read the next parcel.
 //                 async_read(handler);
@@ -330,8 +342,10 @@ namespace hpx { namespace parcelset { namespace policies { namespace tcp
         void handle_write_ack(boost::system::error_code const& e,
             Handler handler)
         {
+            HPX_ASSERT(operation_in_flight_ != 0);
             // Inform caller that data has been received ok.
             handler(e);
+            --operation_in_flight_;
 
             // Issue a read operation to read the next parcel.
             if (!e)
@@ -355,6 +369,7 @@ namespace hpx { namespace parcelset { namespace policies { namespace tcp
         util::high_resolution_timer timer_;
 
         mutex_type mtx_;
+        hpx::util::atomic_count operation_in_flight_;
     };
 }}}}
 

hpx/runtime/parcelset/parcelhandler.hpp

@@ -108,6 +108,8 @@ namespace hpx { namespace parcelset
 
         void initialize(naming::resolver_client &resolver, applier::applier *applier);
 
+        void flush_parcels();
+
         /// \brief Stop all parcel ports associated with this parcelhandler
         void stop(bool blocking = true);
 

hpx/runtime/parcelset/parcelport.hpp

@@ -93,6 +93,8 @@ namespace hpx { namespace parcelset
         ///                 the routine returns immediately.
         virtual bool run(bool blocking = true) = 0;
 
+        virtual void flush_parcels() = 0;
+
         /// Stop the parcelport I/O thread pool.
         ///
         /// \param blocking [in] If blocking is set to \a false the routine will

hpx/runtime/parcelset/parcelport_impl.hpp

@@ -19,6 +19,7 @@
 #include <hpx/runtime/parcelset/parcelport.hpp>
 #include <hpx/runtime/threads/thread.hpp>
 #include <hpx/throw_exception.hpp>
+#include <hpx/util/atomic_count.hpp>
 #include <hpx/util/bind.hpp>
 #include <hpx/util/connection_cache.hpp>
 #include <hpx/util/io_service_pool.hpp>
@@ -173,18 +174,26 @@ namespace hpx { namespace parcelset
             return success;
         }
 
-        void stop(bool blocking = true)
+        void flush_parcels()
         {
             do_background_work(0);
 
             // make sure no more work is pending, wait for service pool to get
             // empty
-            while(operations_in_flight_ != 0)
+
+            while(operations_in_flight_ != 0 || get_pending_parcels_count(false) != 0)
             {
                 if(threads::get_self_ptr())
+                {
                     hpx::this_thread::suspend(hpx::threads::pending,
                         "parcelport_impl::stop");
+                }
             }
+        }
+
+        void stop(bool blocking = true)
+        {
+            flush_parcels();
 
             io_service_pool_.stop();
             if (blocking) {
@@ -942,7 +951,7 @@ namespace hpx { namespace parcelset
         typedef hpx::lcos::local::spinlock mutex_type;
 
         int archive_flags_;
-        boost::atomic<std::size_t> operations_in_flight_;
+        hpx::util::atomic_count operations_in_flight_;
 
         boost::atomic<std::size_t> num_thread_;
         std::size_t const max_background_thread_;

src/runtime/components/server/runtime_support_server.cpp

@@ -716,8 +716,10 @@ namespace hpx { namespace components { namespace server
     {
         applier::applier& appl = hpx::applier::get_applier();
         naming::resolver_client& agas_client = appl.get_agas_client();
+        parcelset::parcelhandler& ph = appl.get_parcel_handler();
 
         agas_client.start_shutdown();
+        ph.flush_parcels();
 
         std::uint32_t locality_id = get_locality_id();
 
@@ -755,7 +757,9 @@ namespace hpx { namespace components { namespace server
             threads::threadmanager_base& tm = appl.get_thread_manager();
 
             while (tm.get_thread_count() > 1)
+            {
                 this_thread::yield();
+            }
 
             return 0;
         }

src/runtime/parcelset/parcelhandler.cpp

@@ -277,6 +277,18 @@ namespace hpx { namespace parcelset
         return did_some_work;
     }
 
+    void parcelhandler::flush_parcels()
+    {
+        // now flush all parcel ports to be shut down
+        for (pports_type::value_type& pp : pports_)
+        {
+            if(pp.first > 0)
+            {
+                pp.second->flush_parcels();
+            }
+        }
+    }
+
     void parcelhandler::stop(bool blocking)
     {
         // now stop all parcel ports

src/runtime/parcelset/parcelport.cpp

@@ -207,7 +207,22 @@ namespace hpx { namespace parcelset
     std::int64_t parcelport::get_pending_parcels_count(bool /*reset*/)
     {
         std::lock_guard<lcos::local::spinlock> l(mtx_);
-        return pending_parcels_.size();
+        std::int64_t count = 0;
+        for (auto && p : pending_parcels_)
+        {
+#if defined(HPX_PARCELSET_PENDING_PARCELS_WORKAROUND)
+            count += hpx::util::get<0>(p.second)->size();
+            HPX_ASSERT(
+                hpx::util::get<0>(p.second)->size() ==
+                hpx::util::get<1>(p.second).size());
+#else
+            count += hpx::util::get<0>(p.second).size();
+            HPX_ASSERT(
+                hpx::util::get<0>(p.second).size() ==
+                hpx::util::get<1>(p.second).size());
+#endif
+        }
+        return count;
     }
 
     ///////////////////////////////////////////////////////////////////////////