Re: Solving thundering Herd with glibc...

Am 14.05.2025 um 17:05 schrieb Vir Campestris:

It took us 3 years to track down why our customers were reporting occasional faults :(

I guess that your code wasn't a 101 LOC test-code but with much
more code.
Here's the code where I stopped development:
#if defined(_WIN32)
#include <Windows.h>
#endif
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <semaphore>
#include <vector>
#include <string_view>
#if defined(__unix__)
#include <sys/resource.h>
#endif
using namespace std;
struct params
{
params( unsigned argc, char **argv );
bool outside, add, all;
};
int main( int argc, char **argv )
{
constexpr size_t N = 10'000;
cout << N << " rounds" << endl;
int hc = thread::hardware_concurrency(), nClients = hc - 1;
int64_t tLast = 0;
for( unsigned outside = 0; outside <= 1; ++outside )
{
cout << (outside ? "outside:" : "inside:") << endl;
for( unsigned all = 0; all <= 1; ++all )
{
cout << (all ? "\tall:" : "\tone:") << endl;
mutex mtx;
int signalled = 0;
condition_variable cv;
atomic_int ai( 0 );
binary_semaphore bs( false );
vector<jthread> threads;
atomic_int64_t nVoluntary( 0 );
atomic_bool stop( false );
for( int c = nClients; c; --c )
threads.emplace_back( [&]
{
for( size_t r = N; r; --r )
{
unique_lock lock( mtx );
cv.wait( lock, [&] { return (bool)signalled; } );
--signalled;
lock.unlock();
if( ai.fetch_sub( 1, memory_order_relaxed ) == 1 )
bs.release( 1 );
}
#if defined(__unix__)
rusage ru;
getrusage( RUSAGE_THREAD, &ru );
nVoluntary.fetch_add( ru.ru_nvcsw, memory_order_relaxed );
#endif
} );
for( size_t r = N; r; --r )
{
auto notify = [&]
{
if( all )
cv.notify_all();
else
for( int c = nClients; c; cv.notify_one(), --c );
};
unique_lock lock( mtx );
signalled = nClients;
if( !outside )
notify();
ai.store( nClients, memory_order_relaxed );
lock.unlock();
if( outside )
notify();
bs.acquire();
}
stop.store( true, memory_order_relaxed );
threads.resize( 0 );
#if defined(_WIN32)
FILETIME ftDummy, ftKernel, ftUser;
GetProcessTimes( GetCurrentProcess(), &ftDummy, &ftDummy, &ftKernel, &ftUser );
auto ftToU64 = []( FILETIME const &ft ) { return (uint64_t)ft.dwHighDateTime << 32 | ft.dwLowDateTime; };
int64_t t = ftToU64( ftKernel ) + ftToU64( ftUser );
cout << "\t\t" << (double)(t - tLast) / 1.0e7 << " seconds" << endl;
tLast = t;
#elif defined(__unix__)
rusage ru;
getrusage( RUSAGE_SELF, &ru );
auto tvToU64 = []( timeval const &tv ) { return (uint64_t)tv.tv_sec * 1'000'000u + tv.tv_usec; };
int64_t t = tvToU64( ru.ru_utime ) + tvToU64( ru.ru_stime );
cout << "\t\t" << (double)nVoluntary.load( memory_order_relaxed ) / nClients << " context switches per thread" << endl;
cout << "\t\t" << (double)(t - tLast) / 1.0e6 << " seconds" << endl;
tLast = t;
#endif
}
}
}