perf(physics): performance.multithreading switches the work over many threads, which each job sizes by how much it has: the physics steps a thread per 128 vehicle steps, up to all logical cores

utilities/Globals.cpp

@@ -497,9 +497,9 @@ bool global_settings::ConfigParseSimulation(cParser& Parser, const std::string&
         return true;
     }
 
-    if (token == "performance.cpucores")
+    if (token == "performance.multithreading")
     {
-        ParseOne(Parser, performance_cpucores);
+        ParseOne(Parser, performance_multithreading);
         return true;
     }
 
@@ -1914,7 +1914,7 @@ global_settings::export_as_text( std::ostream &Output ) const {
     export_as_text( Output, "physics.sleepheld", physics_sleepheld );
     export_as_text( Output, "physics.roadlod", physics_roadlod );
     export_as_text( Output, "physics.pipewake", physics_pipewake );
-    export_as_text( Output, "performance.cpucores", performance_cpucores );
+    export_as_text( Output, "performance.multithreading", performance_multithreading );
     export_as_text( Output, "debuglog", iWriteLogEnabled );
     export_as_text( Output, "multiplelogs", MultipleLogs );
     export_as_text( Output, "showsystemconsole", ShowSystemConsole );

utilities/Globals.h

@@ -117,7 +117,7 @@ struct global_settings {
     bool physics_sleepheld{ true }; // vehicles standing still and held by their brakes let their physics sleep, see TMoverParameters::accelerating()
     bool physics_roadlod{ true }; // road vehicles far from the camera go by one step of the physics in a frame, see vehicle_table::update()
     float physics_pipewake{ 0.001f }; // bar: air flowing along a pipe wakes the physics of a vehicle when it changes its pressure by more than this. 0: any flow does
-    int performance_cpucores{ 0 }; // threads for the work done in parallel (worker_pool), the main one included. 0: all logical cores of the machine
+    bool performance_multithreading{ true }; // work split over the logical cores of the machine (worker_pool), as many as each job is worth
     bool bnewAirCouplers{ true };
     float fMoveLight{ 0.f }; // numer dnia w roku albo -1
     bool FakeLight{ false }; // toggle between fixed and dynamic daylight

utilities/Timer.h

@@ -204,7 +204,7 @@ struct subsystem_stopwatches {
         // the steps of the physics: threads, the groups of vehicles they were split in, the largest one, the groups done in the main
         // thread (vehicle_table::update_steps_parallel())
         unsigned int threads { 1 };
-        unsigned int pool { 1 }; // threads of worker_pool
+        unsigned int capacity { 1 }; // threads a job of worker_pool can get
         std::string singlethread; // what keeps the steps in one thread, if anything
         unsigned int groups { 0 };
         unsigned int largestgroup { 0 };

utilities/workerpool.cpp

@@ -15,63 +15,42 @@ http://mozilla.org/MPL/2.0/.
 
 #include <random>
 
-namespace
-{
-
-// threads by the setting: all the logical cores of the machine when it's 0 or less, otherwise as many as it says, up to a sane limit
-std::size_t wanted_threads()
-{
-	auto const cores{static_cast<int>(std::max(1u, std::thread::hardware_concurrency()))};
-	auto const setting{Global.performance_cpucores};
-	return static_cast<std::size_t>(setting > 0 ? std::min(setting, 64) : cores);
-}
-
-} // namespace
-
 worker_pool::~worker_pool()
 {
-	resize(0);
+	m_stop = true;
+	for (auto &data : m_threads)
+	{
+		data->start.release();
+	}
+	for (auto &data : m_threads)
+	{
+		data->thread.join();
+	}
 }
 
-std::size_t worker_pool::size()
+std::size_t worker_pool::capacity()
 {
-	auto const wanted{wanted_threads()};
-	if (wanted != m_threads.size() + 1)
+	if (false == Global.performance_multithreading)
 	{
-		resize(wanted - 1);
+		return 1;
 	}
-	return m_threads.size() + 1;
+	return std::max<std::size_t>(1, std::thread::hardware_concurrency());
 }
 
-void worker_pool::resize(std::size_t const Workers)
+void worker_pool::grow(std::size_t const Workers)
 {
-	{
-		std::scoped_lock lock(m_mutex);
-		m_stop = true;
-	}
-	m_start.notify_all();
-	for (auto &thread : m_threads)
-	{
-		thread.join();
-	}
-	m_threads.clear();
-	m_stop = false;
-
 	m_threads.reserve(Workers);
-	for (std::size_t index = 0; index < Workers; ++index)
+	while (m_threads.size() < Workers)
 	{
-		// the generation of the jobs is given from here: read by the thread, a job begun meanwhile would be taken for done
-		m_threads.emplace_back(&worker_pool::worker, this, m_generation);
+		auto &data{*m_threads.emplace_back(std::make_unique<thread_data>())};
+		data.thread = std::thread(&worker_pool::worker, this, std::ref(data));
 	}
 }
 
-void worker_pool::run(std::size_t const Count, std::function<void(std::size_t)> const &Work)
+void worker_pool::run(std::size_t const Count, std::size_t const Threads, std::function<void(std::size_t)> const &Work)
 {
-	if (Count == 0)
-	{
-		return;
-	}
-	if ((size() == 1) || (Count == 1))
+	auto const threads{std::min({Threads, capacity(), Count})};
+	if (threads <= 1)
 	{
 		for (std::size_t index = 0; index < Count; ++index)
 		{
@@ -79,17 +58,21 @@ void worker_pool::run(std::size_t const Count, std::function<void(std::size_t)>
 		}
 		return;
 	}
+	grow(threads - 1);
 
 	{
 		std::scoped_lock lock(m_mutex);
 		m_work = &Work;
 		m_count = Count;
 		m_next = 0;
-		m_busy = m_threads.size();
+		m_busy = threads - 1;
 		m_exception = nullptr;
-		++m_generation;
 	}
-	m_start.notify_all();
+	// the job is set before the threads are woken: the semaphores pass it on to them
+	for (std::size_t thread = 0; thread < threads - 1; ++thread)
+	{
+		m_threads[thread]->start.release();
+	}
 	// the calling thread works too
 	work(Work, Count);
 
@@ -126,25 +109,23 @@ void worker_pool::work(std::function<void(std::size_t)> const &Work, std::size_t
 	}
 }
 
-void worker_pool::worker(unsigned int Generation)
+void worker_pool::worker(thread_data &Data)
 {
 	// draws of Random() in this thread go to an engine of its own: the one of the simulation can't be shared between threads
 	std::mt19937 randomengine(static_cast<std::mt19937::result_type>(true_random_seed()));
 	set_thread_random_engine(&randomengine);
 
-	auto generation{Generation};
 	while (true)
 	{
+		Data.start.acquire();
+		if (m_stop)
+		{
+			break;
+		}
 		std::function<void(std::size_t)> const *job{nullptr};
 		std::size_t count{0};
 		{
-			std::unique_lock lock(m_mutex);
-			m_start.wait(lock, [&]() { return m_stop || (m_generation != generation); });
-			if (m_stop)
-			{
-				break;
-			}
-			generation = m_generation;
+			std::scoped_lock lock(m_mutex);
 			job = m_work;
 			count = m_count;
 		}
@@ -152,8 +133,9 @@ void worker_pool::worker(unsigned int Generation)
 		{
 			std::scoped_lock lock(m_mutex);
 			--m_busy;
+			// notified under the lock: the job and the pool stay as they are until the caller sees the count
+			m_done.notify_one();
 		}
-		m_done.notify_one();
 	}
 	set_thread_random_engine(nullptr);
 }

utilities/workerpool.h

@@ -14,13 +14,16 @@ http://mozilla.org/MPL/2.0/.
 #include <cstddef>
 #include <exception>
 #include <functional>
+#include <memory>
 #include <mutex>
+#include <semaphore>
 #include <thread>
 #include <vector>
 
-// threads for work split into independent parts, as many as Global.performance_cpucores asks for (0: all logical cores of the
-// machine). the thread which asks for the work takes part in it too, so the pool keeps one thread fewer. the threads wait between
-// the jobs, and the pool is made again when the setting changes
+// threads for work split into independent parts (Global.performance_multithreading). each job says how many threads its work is
+// worth, up to all the logical cores of the machine; the thread which asks for the work takes part in it too. the threads are
+// made when a job asks for more of them than there are, and wait between the jobs: a job wakes as many of them as it wants, each
+// by a semaphore of its own
 class worker_pool
 {
   public:
@@ -29,32 +32,41 @@ class worker_pool
 	worker_pool &operator=(worker_pool const &) = delete;
 	~worker_pool();
 
-	// threads which take part in run(), the calling one included, by the current setting
-	std::size_t size();
-	// calls Work(index) once for each index below Count, spread over the threads, and returns when all the calls are done. the
-	// calls of one index follow no order. not reentrant: Work can't call run(). an exception thrown by Work is passed on, once
-	// the other calls are done
-	void run(std::size_t const Count, std::function<void(std::size_t)> const &Work);
+	// threads a job can get, the calling one included: all the logical cores with Global.performance_multithreading, otherwise one
+	static std::size_t capacity();
+	// calls Work(index) once for each index below Count, on at most Threads threads (the calling one included, at most capacity()),
+	// and returns when all the calls are done. the calls follow no order. not reentrant: Work can't call run(). an exception
+	// thrown by Work is passed on, once the other calls are done
+	void run(std::size_t const Count, std::size_t const Threads, std::function<void(std::size_t)> const &Work);
+	// threads made so far, the calling one included
+	std::size_t size() const
+	{
+		return m_threads.size() + 1;
+	}
 
   private:
-	void resize(std::size_t const Workers);
-	// Generation: of the last job begun before the thread was made
-	void worker(unsigned int Generation);
+	struct thread_data
+	{
+		std::binary_semaphore start{0}; // released for each job the thread takes part in, and to stop it
+		std::thread thread;
+	};
+
+	// makes threads up to the given count
+	void grow(std::size_t const Workers);
+	void worker(thread_data &Data);
 	// takes the indices left of the current job
 	void work(std::function<void(std::size_t)> const &Work, std::size_t const Count);
 
-	std::vector<std::thread> m_threads;
+	std::vector<std::unique_ptr<thread_data>> m_threads;
 	std::mutex m_mutex;
-	std::condition_variable m_start;
 	std::condition_variable m_done;
-	// the current job
+	// the current job, set before the threads are woken for it
 	std::function<void(std::size_t)> const *m_work{nullptr};
 	std::size_t m_count{0};
 	std::atomic<std::size_t> m_next{0};
-	std::size_t m_busy{0}; // threads of the pool still in it
-	unsigned int m_generation{0}; // of the jobs, for the threads to tell a new one
+	std::size_t m_busy{0}; // threads of the pool woken for it which aren't done
 	std::exception_ptr m_exception;
-	bool m_stop{false};
+	std::atomic<bool> m_stop{false};
 };
 
 // the pool of the application