diff --git a/rendering/opengl46/opengl46renderer.cpp b/rendering/opengl46/opengl46renderer.cpp index cceed1a4..bd1ad340 100644 --- a/rendering/opengl46/opengl46renderer.cpp +++ b/rendering/opengl46/opengl46renderer.cpp @@ -864,9 +864,9 @@ void opengl46_renderer::SwapBuffers() text += timeline( 3, ( counts.threads > 1 ? - "steps on " + std::to_string( counts.threads ) + " threads: " + std::to_string( counts.groups ) + " groups, largest " + "steps on " + std::to_string( counts.threads ) + " of " + std::to_string( counts.capacity ) + " 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)" : - "steps on 1 thread (pool of " + std::to_string( counts.pool ) + ( counts.singlethread.empty() ? std::string{ ")" } : ", off by " + counts.singlethread + ")" ) ), + "steps on 1 thread of " + std::to_string( counts.capacity ) + ( counts.singlethread.empty() ? std::string{} : ": " + counts.singlethread ) ), "" ); char const *const kindnames[] = { "rail, within ", "rail, farther", "road, within ", "road, farther" }; text += timeline( 3, "by kind (near: " + nearlimit + ")", " count stand forces update" ); diff --git a/utilities/Globals.cpp b/utilities/Globals.cpp index 72a6db77..bf62948e 100644 --- a/utilities/Globals.cpp +++ b/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 ); diff --git a/utilities/Globals.h b/utilities/Globals.h index 262ebb77..cbe79954 100644 --- a/utilities/Globals.h +++ b/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 diff --git a/utilities/Timer.h b/utilities/Timer.h index 189a97be..4197079e 100644 --- a/utilities/Timer.h +++ b/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 }; diff --git a/utilities/workerpool.cpp b/utilities/workerpool.cpp index e38123a6..74149224 100644 --- a/utilities/workerpool.cpp +++ b/utilities/workerpool.cpp @@ -15,63 +15,42 @@ http://mozilla.org/MPL/2.0/. #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); + 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(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())}; + data.thread = std::thread(&worker_pool::worker, this, std::ref(data)); } } -void worker_pool::run(std::size_t const Count, std::function const &Work) +void worker_pool::run(std::size_t const Count, std::size_t const Threads, std::function 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 } 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 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(true_random_seed())); set_thread_random_engine(&randomengine); - auto generation{Generation}; while (true) { + Data.start.acquire(); + if (m_stop) + { + break; + } 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; + 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); } diff --git a/utilities/workerpool.h b/utilities/workerpool.h index 762e1407..9771638f 100644 --- a/utilities/workerpool.h +++ b/utilities/workerpool.h @@ -14,13 +14,16 @@ http://mozilla.org/MPL/2.0/. #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 +// 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 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 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 const &Work, std::size_t const Count); - std::vector m_threads; + std::vector> 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 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::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 m_stop{false}; }; // the pool of the application diff --git a/vehicle/DynObj.cpp b/vehicle/DynObj.cpp index c8268db5..af171d51 100644 --- a/vehicle/DynObj.cpp +++ b/vehicle/DynObj.cpp @@ -8681,26 +8681,39 @@ 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 (network.server, .client), - // which keeps the draws of Random() in step across the machines: the threads have engines of their own. the messages to - // external programs (Global.iMultiplayer) don't mind - auto &workers { Workers() }; - auto const parallel { - ( workers.size() > 1 ) - && Global.network_servers.empty() - && ( false == Global.network_client.has_value() ) }; - counts.threads = ( parallel ? static_cast( workers.size() ) : 1u ); - // why not, for the debug panel - counts.pool = static_cast( workers.size() ); - counts.singlethread = - ( workers.size() <= 1 ? "performance.cpucores " + std::to_string( Global.performance_cpucores ) : - false == Global.network_servers.empty() ? std::string{ "network.server" } : - Global.network_client.has_value() ? std::string{ "network.client" } : - std::string{} ); + // the steps of the physics on many threads (Global.performance_multithreading), as many as the work is worth: a thread for each + // STEPS_PER_THREAD steps of vehicles with the physics on, up to all the cores. not in a network session (network.server, + // .client), which keeps the draws of Random() in step across the machines: the threads have engines of their own. the messages + // to external programs (Global.iMultiplayer) don't mind + auto constexpr STEPS_PER_THREAD { 128 }; + std::size_t work { 0 }; + for( std::size_t index = 0; index < m_items.size(); ++index ) { + auto const *vehicle { m_items[ index ] }; + if( vehicle->MoverParameters->PhysicActivation || vehicle->MechInside ) { + work += ( coarse[ index ] != 0 ? 1 : static_cast( std::max( Iterationcount, 1 ) ) ); + } + } + auto const network { ( false == Global.network_servers.empty() ) || Global.network_client.has_value() }; + auto const threads { ( + network ? + 1 : + std::min( worker_pool::capacity(), std::max( 1, ( work + STEPS_PER_THREAD / 2 ) / STEPS_PER_THREAD ) ) ) }; + // what keeps them in one thread, for the debug panel + counts.capacity = static_cast( worker_pool::capacity() ); + counts.singlethread = ( + false == Global.performance_multithreading ? std::string{ "performance.multithreading off" } : + network ? std::string{ "network session" } : + threads <= 1 ? std::string{ "too little work" } : + std::string{} ); + counts.threads = 1; + auto const parallel { ( threads > 1 ) && update_steps_parallel( Deltatime, Iterationcount, coarse, threads ) }; if( parallel ) { - update_steps_parallel( Deltatime, Iterationcount, coarse ); + counts.threads = static_cast( threads ); } else { + if( threads > 1 ) { + counts.singlethread = "a vehicle coupled with one it doesn't see"; + } auto mark { std::chrono::steady_clock::now() }; if( Iterationcount > 1 ) { // ABu: ponizsze wykonujemy tylko jesli wiecej niz jedna iteracja @@ -8750,45 +8763,48 @@ 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 ) { +// the steps of the physics of vehicle_table::update() (UpdateForce() and FastUpdate() of each vehicle, Iterationcount times) on up +// to Threads 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, and each group goes through all its steps in +// one thread, its vehicles in their order, the groups at once: 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(). returns false, having done +// nothing, when a vehicle is coupled with one it doesn't see next to it, which the groups don't account for +bool +vehicle_table::update_steps_parallel( double const Deltatime, int const Iterationcount, std::vector const &Coarse, std::size_t const Threads ) { auto const itemcount { m_items.size() }; - // the groups, by joining the vehicles with the ones they act on + for( std::size_t index = 0; index < itemcount; ++index ) { + m_items[ index ]->m_tableindex = static_cast( index ); + } + // the groups, by joining the vehicles with the ones they act on. the coupled ones are those next to them 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 const *neighbour { mover->Neighbours[ end ].vehicle }; + if( ( mover->Couplers[ end ].Connected != nullptr ) + && ( ( neighbour == nullptr ) || ( neighbour->MoverParameters != mover->Couplers[ end ].Connected ) ) ) { + return false; + } + if( neighbour == nullptr ) { continue; } + auto const other { neighbour->m_tableindex }; + if( ( other >= itemcount ) || ( m_items[ other ] != neighbour ) ) { + // not in the table, shouldn't happen + return false; + } + auto const left { root( static_cast( index ) ) }; + auto const right { root( other ) }; + if( left != right ) { + m_groupparent[ std::max( left, right ) ] = std::min( left, right ); + } } } auto constexpr none { std::numeric_limits::max() }; @@ -8803,7 +8819,7 @@ vehicle_table::update_steps_parallel( double const Deltatime, int const Iteratio } m_groups[ group ].push_back( static_cast( index ) ); } - // the groups done in the main thread + // the groups done in the main thread, and the others largest first, for the threads to end close to one another m_threadgroups.clear(); m_maingroups.clear(); std::size_t largest { 0 }; @@ -8816,40 +8832,45 @@ vehicle_table::update_steps_parallel( double const Deltatime, int const Iteratio return ( vehicle->MoverParameters->LoadStatus != 0 ) || ( vehicle->LoadExchangeTime() > 0.f ); } ) }; ( loading ? m_maingroups : m_threadgroups ).push_back( static_cast( group ) ); } + std::sort( + std::begin( m_threadgroups ), std::end( m_threadgroups ), + [ this ]( std::uint32_t const Left, std::uint32_t const Right ) { + return m_groups[ Left ].size() > m_groups[ Right ].size(); } ); 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 ); - } - } }; - + // all the steps of a group, in the order update() makes them + auto const totaltime { Deltatime * Iterationcount }; 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 steps = [ & ]( std::uint32_t const Group ) { + auto const &members { m_groups[ Group ] }; + if( Iterationcount > 1 ) { + // ABu: ponizsze wykonujemy tylko jesli wiecej niz jedna iteracja + for( int iteration = 0; iteration < Iterationcount - 1; ++iteration ) { + for( auto const index : members ) { + if( false == coarse( index ) ) { m_items[ index ]->UpdateForce( Deltatime ); } + } + for( auto const index : members ) { + if( false == coarse( index ) ) { m_items[ index ]->FastUpdate( Deltatime ); } + } + } } + for( auto const index : members ) { + // the vehicles going by one step take it with the time of all of them + m_items[ index ]->UpdateForce( coarse( index ) ? totaltime : Deltatime ); + } }; + + Workers().run( + m_threadgroups.size(), Threads, + [ & ]( std::size_t const Group ) { + steps( m_threadgroups[ Group ] ); } ); + for( auto const group : m_maingroups ) { + steps( group ); } - 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 ); } ); + return true; } // legacy method, checks for presence and height of traction wire for specified vehicle diff --git a/vehicle/DynObj.h b/vehicle/DynObj.h index 8b4c47c0..9bbb8f97 100644 --- a/vehicle/DynObj.h +++ b/vehicle/DynObj.h @@ -655,6 +655,8 @@ private: TDynamicObject * GetFirstDynamic(int cpl_type, int cf = 1); void ABuSetModelShake(glm::dvec3 mShake); + // place in vehicle_table::m_items, set by its update for the groups of the physics steps on many threads + std::uint32_t m_tableindex { 0 }; // McZapkie-010302 TController *Mechanik; TController *ctOwner; // wskażnik na obiekt zarządzający składem @@ -880,9 +882,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 ); + // the steps of the physics of update() on up to Threads threads, see the definition. false: they can't be split, nothing done + bool + update_steps_parallel( double const Deltatime, int const Iterationcount, std::vector const &Coarse, std::size_t const Threads ); // legacy method, sends list of vehicles over network void DynamicList( bool const Onlycontrolled = false ) const; @@ -891,10 +893,9 @@ 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 + // working space of update_steps_parallel(), kept between the updates: the groups of the vehicles (joined by union find), 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;