perf(physics): the steps of the vehicle physics on many threads, in groups of vehicles which act on one another (performance.cpucores, 0: all cores)

utilities/Globals.cpp

@@ -497,6 +497,12 @@ bool global_settings::ConfigParseSimulation(cParser& Parser, const std::string&
         return true;
     }
 
+    if (token == "performance.cpucores")
+    {
+        ParseOne(Parser, performance_cpucores);
+        return true;
+    }
+
     if (token == "enabletraction")
     {
         ParseOne(Parser, bEnableTraction);
@@ -1908,6 +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, "debuglog", iWriteLogEnabled );
     export_as_text( Output, "multiplelogs", MultipleLogs );
     export_as_text( Output, "showsystemconsole", ShowSystemConsole );

utilities/Globals.h

@@ -117,6 +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 bnewAirCouplers{ true };
     float fMoveLight{ 0.f }; // numer dnia w roku albo -1
     bool FakeLight{ false }; // toggle between fixed and dynamic daylight

utilities/Timer.h

@@ -11,6 +11,7 @@ http://mozilla.org/MPL/2.0/.
 
 #include <array>
 #include <chrono>
+#include <mutex>
 #include <string>
 #include <unordered_map>
 
@@ -111,8 +112,10 @@ public:
         unsigned int busiestevents { 0 };
     };
     // Owner: the vehicle, Label: its type and name, read only when the second ends
+    // can be called by many threads at once (the physics of the vehicles in parallel)
     void
         note( source const Source, void const *Owner, std::string const *Type, std::string const *Name ) {
+            std::scoped_lock lock( m_mutex );
             ++m_events[ Source ];
             auto &owner { m_owners[ Owner ] };
             ++owner.events;
@@ -123,6 +126,7 @@ public:
         update() {
             auto const now { std::chrono::steady_clock::now() };
             if( now - m_start < std::chrono::seconds( 1 ) ) { return; }
+            std::scoped_lock lock( m_mutex );
             m_last = {};
             m_last.events = m_events;
             for( auto const &owner : m_owners ) {
@@ -147,6 +151,7 @@ private:
     std::array<unsigned int, source::count> m_events {};
     std::unordered_map<void const *, owner_data> m_owners;
     result m_last;
+    std::mutex m_mutex;
 };
 
 struct subsystem_stopwatches {
@@ -196,6 +201,12 @@ struct subsystem_stopwatches {
         // enabled vehicles of each kind (see vehicles_forces_kind), and of them the ones standing still
         std::array<unsigned int, 4> kind {};
         std::array<unsigned int, 4> standing {};
+        // 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 groups { 0 };
+        unsigned int largestgroup { 0 };
+        unsigned int maingroups { 0 };
         // rail vehicles standing still (below 0.01 km/h) with the physics on, and what keeps it on by the test of
         // TMoverParameters::ComputeTotalForce(); a vehicle can have several of these
         struct awake_reasons {

utilities/utilities.cpp

@@ -125,18 +125,33 @@ bool ClearFlag(int &Flag, int const Value)
 	}
 }
 
+namespace
+{
+thread_local std::mt19937 *t_random_engine{nullptr};
+
+std::mt19937 &random_engine()
+{
+	return t_random_engine != nullptr ? *t_random_engine : Global.random_engine;
+}
+} // namespace
+
+void set_thread_random_engine(std::mt19937 *Engine)
+{
+	t_random_engine = Engine;
+}
+
 double Random(double min, double max)
 {
 	if (max < min) { std::swap(min, max); } // std::uniform_real_distribution requires min <= max (inverted bounds are UB)
 	std::uniform_real_distribution dist(min, max);
-	return dist(Global.random_engine);
+	return dist(random_engine());
 }
 
 int Random(int min, int max)
 {
 	if (max < min) { std::swap(min, max); } // std::uniform_int_distribution requires min <= max (inverted bounds are UB)
 	std::uniform_int_distribution dist(min, max);
-	return dist(Global.random_engine);
+	return dist(random_engine());
 }
 
 std::uint64_t true_random_seed()

utilities/utilities.h

@@ -62,6 +62,9 @@ inline long Round(double const f)
 
 double Random(double a, double b);
 int Random(int min, int max);
+// the engine Random() draws from in the calling thread, in place of Global.random_engine, or none for that one again. the threads
+// of worker_pool have engines of their own: the one of the simulation can't be shared between threads
+void set_thread_random_engine(std::mt19937 *Engine);
 
 /// @author lxvia
 /// @brief Returns a random seed from the system's entropy source, if available. Otherwise, falls back to a clock-based seed.

utilities/workerpool.cpp

@@ -0,0 +1,165 @@
+/*
+This Source Code Form is subject to the
+terms of the Mozilla Public License, v.
+2.0. If a copy of the MPL was not
+distributed with this file, You can
+obtain one at
+http://mozilla.org/MPL/2.0/.
+*/
+
+#include "stdafx.h"
+#include "utilities/workerpool.h"
+
+#include "utilities/Globals.h"
+#include "utilities/utilities.h"
+
+#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);
+}
+
+std::size_t worker_pool::size()
+{
+	auto const wanted{wanted_threads()};
+	if (wanted != m_threads.size() + 1)
+	{
+		resize(wanted - 1);
+	}
+	return m_threads.size() + 1;
+}
+
+void worker_pool::resize(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)
+	{
+		// 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);
+	}
+}
+
+void worker_pool::run(std::size_t const Count, std::function<void(std::size_t)> const &Work)
+{
+	if (Count == 0)
+	{
+		return;
+	}
+	if ((size() == 1) || (Count == 1))
+	{
+		for (std::size_t index = 0; index < Count; ++index)
+		{
+			Work(index);
+		}
+		return;
+	}
+
+	{
+		std::scoped_lock lock(m_mutex);
+		m_work = &Work;
+		m_count = Count;
+		m_next = 0;
+		m_busy = m_threads.size();
+		m_exception = nullptr;
+		++m_generation;
+	}
+	m_start.notify_all();
+	// the calling thread works too
+	work(Work, Count);
+
+	std::unique_lock lock(m_mutex);
+	m_done.wait(lock, [this]() { return m_busy == 0; });
+	m_work = nullptr;
+	if (m_exception)
+	{
+		auto const exception{m_exception};
+		m_exception = nullptr;
+		lock.unlock();
+		std::rethrow_exception(exception);
+	}
+}
+
+void worker_pool::work(std::function<void(std::size_t)> const &Work, std::size_t const Count)
+{
+	try
+	{
+		for (auto index{m_next++}; index < Count; index = m_next++)
+		{
+			Work(index);
+		}
+	}
+	catch (...)
+	{
+		std::scoped_lock lock(m_mutex);
+		if (false == static_cast<bool>(m_exception))
+		{
+			m_exception = std::current_exception();
+		}
+		// the indices left are taken by nobody
+		m_next = Count;
+	}
+}
+
+void worker_pool::worker(unsigned int Generation)
+{
+	// 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)
+	{
+		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;
+			job = m_work;
+			count = m_count;
+		}
+		work(*job, count);
+		{
+			std::scoped_lock lock(m_mutex);
+			--m_busy;
+		}
+		m_done.notify_one();
+	}
+	set_thread_random_engine(nullptr);
+}
+
+worker_pool &Workers()
+{
+	static worker_pool pool;
+	return pool;
+}

utilities/workerpool.h

@@ -0,0 +1,61 @@
+/*
+This Source Code Form is subject to the
+terms of the Mozilla Public License, v.
+2.0. If a copy of the MPL was not
+distributed with this file, You can
+obtain one at
+http://mozilla.org/MPL/2.0/.
+*/
+
+#pragma once
+
+#include <atomic>
+#include <condition_variable>
+#include <cstddef>
+#include <exception>
+#include <functional>
+#include <mutex>
+#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
+class worker_pool
+{
+  public:
+	worker_pool() = default;
+	worker_pool(worker_pool const &) = delete;
+	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);
+
+  private:
+	void resize(std::size_t const Workers);
+	// Generation: of the last job begun before the thread was made
+	void worker(unsigned int Generation);
+	// 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::mutex m_mutex;
+	std::condition_variable m_start;
+	std::condition_variable m_done;
+	// the current job
+	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::exception_ptr m_exception;
+	bool m_stop{false};
+};
+
+// the pool of the application
+worker_pool &Workers();