Merge pull request #3046 from msimberg/suspend-thread

Add OS thread suspension

examples/resource_partitioner/shared_priority_scheduler.hpp

@@ -565,7 +565,7 @@ namespace threads {
             }
 
             ///////////////////////////////////////////////////////////////////////
-            bool cleanup_terminated(bool delete_all = false)
+            bool cleanup_terminated(bool delete_all)
             {
                 bool empty = true;
                 for (std::size_t i = 0; i != queues_.size(); ++i)
@@ -593,6 +593,11 @@ namespace threads {
                 return empty;
             }
 
+            bool cleanup_terminated(std::size_t num_threads, bool delete_all)
+            {
+                return cleanup_terminated(delete_all);
+            }
+
             ///////////////////////////////////////////////////////////////////////
             // create a new thread and schedule it if the initial state is equal
             // to pending

hpx/runtime/threads/detail/scheduled_thread_pool.hpp

@@ -103,6 +103,11 @@ namespace hpx { namespace threads { namespace detail
                 state, priority, num, reset);
         }
 
+        std::int64_t get_background_thread_count()
+        {
+            return sched_->Scheduler::get_background_thread_count();
+        }
+
         bool enumerate_threads(
             util::function_nonser<bool(thread_id_type)> const& f,
             thread_state_enum state) const
@@ -117,6 +122,7 @@ namespace hpx { namespace threads { namespace detail
 
         void set_scheduler_mode(threads::policies::scheduler_mode mode)
         {
+            mode_ = mode;
             return sched_->Scheduler::set_scheduler_mode(mode);
         }
 
@@ -127,6 +133,8 @@ namespace hpx { namespace threads { namespace detail
         void stop_locked(Lock& l, bool blocking = true);
         void stop (std::unique_lock<compat::mutex>& l, bool blocking = true);
 
+        void resume(error_code& ec = throws);
+
         ///////////////////////////////////////////////////////////////////
         compat::thread& get_os_thread_handle(std::size_t global_thread_num)
         {
@@ -144,6 +152,22 @@ namespace hpx { namespace threads { namespace detail
             return threads_.size();
         }
 
+        std::size_t get_active_os_thread_count() const
+        {
+            std::size_t active_os_thread_count = 0;
+            for (std::size_t thread_num = 0; thread_num < threads_.size();
+                ++thread_num)
+            {
+                if (sched_->Scheduler::get_state(thread_num).load() ==
+                    state_running)
+                {
+                    ++active_os_thread_count;
+                }
+            }
+
+            return active_os_thread_count;
+        }
+
 #ifdef HPX_HAVE_THREAD_STEALING_COUNTS
         std::int64_t get_num_pending_misses(std::size_t num, bool reset)
         {
@@ -242,6 +266,12 @@ namespace hpx { namespace threads { namespace detail
         void remove_processing_unit(
             std::size_t virt_core, error_code& = hpx::throws);
 
+        // Suspend the given processing unit on the scheduler.
+        void suspend_processing_unit(std::size_t virt_core, error_code&);
+
+        // Resume the given processing unit on the scheduler.
+        void resume_processing_unit(std::size_t virt_core, error_code&);
+
     protected:
         friend struct init_tss_helper<Scheduler>;
 
@@ -251,6 +281,11 @@ namespace hpx { namespace threads { namespace detail
             std::size_t thread_num, std::shared_ptr<compat::barrier> startup,
             error_code& ec = hpx::throws);
 
+        void suspend_processing_unit_internal(std::size_t virt_core,
+            error_code& = hpx::throws);
+        void resume_processing_unit_internal(std::size_t virt_core,
+            error_code& = hpx::throws);
+
     private:
         std::vector<compat::thread> threads_;           // vector of OS-threads