diff --git a/CMakeLists.txt b/CMakeLists.txt index 6bd71ff7..73c0566c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -94,6 +94,7 @@ list(APPEND CMAKE_MODULE_PATH "${VCPKG_INSTALLED_PATH}/share") set(SOURCES "model/Texture.cpp" "utilities/Timer.cpp" +"utilities/workerpool.cpp" "world/Track.cpp" "world/TrackSleepers.cpp" "world/TrackBlend.cpp" diff --git a/rendering/opengl46/opengl46renderer.cpp b/rendering/opengl46/opengl46renderer.cpp index 092fe894..e5e17b7a 100644 --- a/rendering/opengl46/opengl46renderer.cpp +++ b/rendering/opengl46/opengl46renderer.cpp @@ -861,6 +861,13 @@ void opengl46_renderer::SwapBuffers() auto text { std::to_string( Value ) }; return std::string( std::max( 0, 6 - static_cast( text.size() ) ), ' ' ) + text; }; auto const nearlimit { std::to_string( static_cast( Timer::subsystem.VEHICLE_NEAR ) ) + " m" }; + text += timeline( + 3, + ( counts.threads > 1 ? + "steps on " + std::to_string( counts.threads ) + " threads: " + std::to_string( counts.groups ) + " groups, largest " + + std::to_string( counts.largestgroup ) + ", " + std::to_string( counts.maingroups ) + " in the main thread (forces by kind not timed)" : + std::string{ "steps on 1 thread" } ), + "" ); char const *const kindnames[] = { "rail, within ", "rail, farther", "road, within ", "road, farther" }; text += timeline( 3, "by kind (near: " + nearlimit + ")", " count stand forces update" ); for( std::size_t kind = 0; kind < 4; ++kind ) { diff --git a/utilities/Globals.cpp b/utilities/Globals.cpp index ef73f0da..72a6db77 100644 --- a/utilities/Globals.cpp +++ b/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 ); diff --git a/utilities/Globals.h b/utilities/Globals.h index d8427b27..262ebb77 100644 --- a/utilities/Globals.h +++ b/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 diff --git a/utilities/Timer.h b/utilities/Timer.h index 07326a76..8a703d97 100644 --- a/utilities/Timer.h +++ b/utilities/Timer.h @@ -11,6 +11,7 @@ http://mozilla.org/MPL/2.0/. #include #include +#include #include #include @@ -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 m_events {}; std::unordered_map 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 kind {}; std::array 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 { diff --git a/utilities/utilities.cpp b/utilities/utilities.cpp index 0c18e002..d302768a 100644 --- a/utilities/utilities.cpp +++ b/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() diff --git a/utilities/utilities.h b/utilities/utilities.h index d98b7bc9..e0933612 100644 --- a/utilities/utilities.h +++ b/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. diff --git a/utilities/workerpool.cpp b/utilities/workerpool.cpp new file mode 100644 index 00000000..e38123a6 --- /dev/null +++ b/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 + +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(std::max(1u, std::thread::hardware_concurrency()))}; + auto const setting{Global.performance_cpucores}; + return static_cast(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 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 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(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(true_random_seed())); + set_thread_random_engine(&randomengine); + + auto generation{Generation}; + while (true) + { + std::function 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; +} diff --git a/utilities/workerpool.h b/utilities/workerpool.h new file mode 100644 index 00000000..762e1407 --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include +#include + +// 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 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 const &Work, std::size_t const Count); + + std::vector m_threads; + std::mutex m_mutex; + std::condition_variable m_start; + std::condition_variable m_done; + // the current job + std::function const *m_work{nullptr}; + std::size_t m_count{0}; + std::atomic 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(); diff --git a/vehicle/DynObj.cpp b/vehicle/DynObj.cpp index fb961262..abbabb38 100644 --- a/vehicle/DynObj.cpp +++ b/vehicle/DynObj.cpp @@ -22,6 +22,7 @@ http://mozilla.org/MPL/2.0/. #include "vehicle/Driver.h" #include "utilities/Globals.h" #include "utilities/Timer.h" +#include "utilities/workerpool.h" #include "utilities/Logs.h" #include "utilities/glmHelpers.h" #include "Console.h" @@ -8679,6 +8680,20 @@ vehicle_table::update( double Deltatime, int Iterationcount ) { } Timer::subsystem.vehicles_prepare.stop(); Timer::subsystem.vehicles_forces.start(); + auto const totaltime { Deltatime * Iterationcount }; // całkowity czas + // the steps of the physics on many threads (Global.performance_cpucores). not in a network session, which keeps the draws of + // Random() in step across the machines: the threads have engines of their own + auto &workers { Workers() }; + auto const parallel { + ( workers.size() > 1 ) + && ( Global.iMultiplayer == 0 ) + && Global.network_servers.empty() + && ( false == Global.network_client.has_value() ) }; + counts.threads = ( parallel ? static_cast( workers.size() ) : 1u ); + if( parallel ) { + update_steps_parallel( Deltatime, Iterationcount, coarse ); + } + else { auto mark { std::chrono::steady_clock::now() }; if( Iterationcount > 1 ) { // ABu: ponizsze wykonujemy tylko jesli wiecej niz jedna iteracja @@ -8699,7 +8714,6 @@ vehicle_table::update( double Deltatime, int Iterationcount ) { } } } - auto const totaltime { Deltatime * Iterationcount }; // całkowity czas { std::size_t index { 0 }; for( auto *vehicle : m_items ) { @@ -8709,12 +8723,13 @@ vehicle_table::update( double Deltatime, int Iterationcount ) { lap( index++, Timer::subsystem.vehicles_forces_kind, mark ); } } + } Timer::subsystem.vehicles_forces.stop(); Timer::subsystem.vehicles_update.start(); { std::size_t index { 0 }; - mark = std::chrono::steady_clock::now(); + auto mark { std::chrono::steady_clock::now() }; for( auto *vehicle : m_items ) { auto const coarsestep { ( index < coarse.size() ) && ( coarse[ index ] != 0 ) }; // Ra 2015-01: tylko tu przelicza sieć trakcyjną @@ -8728,6 +8743,108 @@ vehicle_table::update( double Deltatime, int Iterationcount ) { erase_disabled(); } +// the steps of the physics of vehicle_table::update() (UpdateForce() and FastUpdate() of each vehicle, Iterationcount times) on the +// threads of worker_pool. a step acts on the vehicle itself, on the ones coupled with it, and on the ones seen ahead or behind it +// (CollisionDetect() changes the speed of the other one); the movement along the tracks, the events and the sounds wait for +// Update(). so the vehicles are put in groups which act on nothing outside them: each group is done by one thread, its vehicles +// in their order, and the groups at once, which gives what doing them one by one gives. the groups of vehicles exchanging their +// load, which can load the models of it, are done in the main thread. Coarse: as in update() +void +vehicle_table::update_steps_parallel( double const Deltatime, int const Iterationcount, std::vector const &Coarse ) { + + auto const itemcount { m_items.size() }; + // the groups, by joining the vehicles with the ones they act on + m_groupparent.resize( itemcount ); + std::iota( std::begin( m_groupparent ), std::end( m_groupparent ), 0u ); + m_grouplookup.clear(); + m_grouplookup.reserve( 2 * itemcount ); + for( std::size_t index = 0; index < itemcount; ++index ) { + m_grouplookup.emplace( m_items[ index ], static_cast( index ) ); + m_grouplookup.emplace( m_items[ index ]->MoverParameters, static_cast( index ) ); + } + auto const root = [ this ]( std::uint32_t Index ) { + while( m_groupparent[ Index ] != Index ) { + m_groupparent[ Index ] = m_groupparent[ m_groupparent[ Index ] ]; + Index = m_groupparent[ Index ]; + } + return Index; }; + auto const join = [ & ]( std::uint32_t const Index, void const *Other ) { + if( Other == nullptr ) { return; } + auto const lookup { m_grouplookup.find( Other ) }; + if( lookup == m_grouplookup.end() ) { return; } + auto const left { root( Index ) }; + auto const right { root( lookup->second ) }; + if( left != right ) { + m_groupparent[ std::max( left, right ) ] = std::min( left, right ); + } }; + for( std::size_t index = 0; index < itemcount; ++index ) { + auto const *mover { m_items[ index ]->MoverParameters }; + for( auto const end : { end::front, end::rear } ) { + join( static_cast( index ), mover->Neighbours[ end ].vehicle ); + join( static_cast( index ), mover->Couplers[ end ].Connected ); + } + } + auto constexpr none { std::numeric_limits::max() }; + m_groupof.assign( itemcount, none ); + std::size_t groupcount { 0 }; + for( std::size_t index = 0; index < itemcount; ++index ) { + auto &group { m_groupof[ root( static_cast( index ) ) ] }; + if( group == none ) { + group = static_cast( groupcount++ ); + if( m_groups.size() < groupcount ) { m_groups.emplace_back(); } + m_groups[ group ].clear(); + } + m_groups[ group ].push_back( static_cast( index ) ); + } + // the groups done in the main thread + m_threadgroups.clear(); + m_maingroups.clear(); + std::size_t largest { 0 }; + for( std::size_t group = 0; group < groupcount; ++group ) { + largest = std::max( largest, m_groups[ group ].size() ); + auto const loading { std::any_of( + std::begin( m_groups[ group ] ), std::end( m_groups[ group ] ), + [ this ]( std::uint32_t const Index ) { + auto const *vehicle { m_items[ Index ] }; + return ( vehicle->MoverParameters->LoadStatus != 0 ) || ( vehicle->LoadExchangeTime() > 0.f ); } ) }; + ( loading ? m_maingroups : m_threadgroups ).push_back( static_cast( group ) ); + } + auto &counts { Timer::subsystem.vehicles }; + counts.groups = static_cast( groupcount ); + counts.largestgroup = static_cast( largest ); + counts.maingroups = static_cast( m_maingroups.size() ); + + auto &workers { Workers() }; + auto const pass = [ & ]( auto const &Step ) { + workers.run( + m_threadgroups.size(), + [ & ]( std::size_t const Group ) { + for( auto const index : m_groups[ m_threadgroups[ Group ] ] ) { + Step( index ); + } } ); + for( auto const group : m_maingroups ) { + for( auto const index : m_groups[ group ] ) { + Step( index ); + } + } }; + + auto const coarse = [ & ]( std::uint32_t const Index ) { + return ( Index < Coarse.size() ) && ( Coarse[ Index ] != 0 ); }; + if( Iterationcount > 1 ) { + // ABu: ponizsze wykonujemy tylko jesli wiecej niz jedna iteracja + for( int iteration = 0; iteration < Iterationcount - 1; ++iteration ) { + pass( [ & ]( std::uint32_t const Index ) { + if( false == coarse( Index ) ) { m_items[ Index ]->UpdateForce( Deltatime ); } } ); + pass( [ & ]( std::uint32_t const Index ) { + if( false == coarse( Index ) ) { m_items[ Index ]->FastUpdate( Deltatime ); } } ); + } + } + auto const totaltime { Deltatime * Iterationcount }; + pass( [ & ]( std::uint32_t const Index ) { + // the vehicles going by one step take it with the time of all of them + m_items[ Index ]->UpdateForce( coarse( Index ) ? totaltime : Deltatime ); } ); +} + // legacy method, checks for presence and height of traction wire for specified vehicle void vehicle_table::update_traction( TDynamicObject *Vehicle ) { diff --git a/vehicle/DynObj.h b/vehicle/DynObj.h index da7b347d..8b4c47c0 100644 --- a/vehicle/DynObj.h +++ b/vehicle/DynObj.h @@ -880,6 +880,9 @@ public: // legacy method, checks for presence and height of traction wire for specified vehicle void update_traction( TDynamicObject *Vehicle ); + // the steps of the physics of update() on many threads, see the definition + void + update_steps_parallel( double const Deltatime, int const Iterationcount, std::vector const &Coarse ); // legacy method, sends list of vehicles over network void DynamicList( bool const Onlycontrolled = false ) const; @@ -888,6 +891,14 @@ private: // maintenance; removes from tracks consists with vehicles marked as disabled bool erase_disabled(); + // working space of update_steps_parallel(), kept between the updates: the groups of the vehicles (joined by union find, by + // the vehicle and by its mover), the group of each root, the groups done on the threads and in the main thread + std::vector m_groupparent; + std::unordered_map m_grouplookup; + std::vector m_groupof; + std::vector> m_groups; + std::vector m_threadgroups; + std::vector m_maingroups; };