12 ManualResetEvent
Implementation of the equivalent to a manual reset Windows event object in terms of C++11.
ManualResetEvent.h
// ManualResetEvent.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2013
#pragma once
#ifndef Ceda_cxThread_ManualResetEvent_H
#define Ceda_cxThread_ManualResetEvent_H
#include "cxThread.h"
#include "Ceda/cxUtils/CedaAssert.h"
#include <mutex>
#include <condition_variable>
namespace ceda
{
class ManualResetEvent
{
cxNotCloneable(ManualResetEvent)
public:
explicit ManualResetEvent(bool signaled = false) : signaled_(signaled) {}
void Signal()
{
std::lock_guard<std::mutex> lock(mutex_);
signaled_ = true;
cv_.notify_all();
}
void Reset()
{
std::lock_guard<std::mutex> lock(mutex_);
signaled_ = false;
}
void Wait()
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock,[this]{return signaled_;});
}
// Wait with a timeout.
// Returns true if the wait was completed successfully because the event had been signalled
bool Wait(uint32 msec)
{
std::unique_lock<std::mutex> lock(mutex_);
return cv_.wait_for(lock,std::chrono::milliseconds(msec),[this]{return signaled_;});
}
private:
std::mutex mutex_;
std::condition_variable cv_;
bool signaled_;
};
} // namespace ceda
#endif // include guardPerformance measurement
The performance of ManualResetEvent is measured in ThreadTests.cpp in txThread.
The following results were obtained with the windows-x64 build running under Windows 10 on a desktop with an AMD Ryzen 9 3900X 12-Core Processor:
Thread timing tests
std::thread::hardware_concurrency() = 24
Increment std::atomic<long> with multiple threads (avg of 54) 100000/0.00115s = 86.87 MHz
Increment std::atomic<long> (avg of 155) 100000/0.0004s = 250.1 MHz
Construct/Destruct std::mutex (avg of 74) 100000/0.000841s = 118.9 MHz
Lock std::mutex (avg of 35) 100000/0.00177s = 56.44 MHz
Construct/Destruct ManualResetEvent (avg of 43) 100000/0.00147s = 68.12 MHz
Signal(),Wait(),Reset() on ManualResetEvent (avg of 101) 10000/0.000612s = 16.33 MHz
Read and write thread_local int* (avg of 130) 1e+06/0.000476s = 2.099 GHz
ThreadTests.cpp
// ThreadTests.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2006
#include "Ceda/cxThread/ManualResetEvent.h"
#include "Ceda/cxUtils/TestTimer.h"
#include "Ceda/cxUtils/Tracer.h"
#include <thread>
#include <mutex>
#include <atomic>
#include <functional>
void execute_task()
{
//std::cout << "Hello, world" << std::endl;
}
void test1()
{
std::thread t(execute_task);
t.join(); // wait until call to execute_task() completes
}
void test2()
{
std::thread t[16];
for (int i=0 ; i < 16 ; ++i)
t[i] = std::thread(execute_task);
for (int i=0 ; i < 16 ; ++i)
t[i].join();
}
template <int N>
class Threads
{
public:
Threads() {}
Threads(std::function<void()> f)
{
for (int i=0 ; i < N ; ++i)
threads_[i] = std::thread(f);
}
void join()
{
for (int i=0 ; i < N ; ++i)
threads_[i].join();
}
private:
std::thread threads_[N];
};
void test3()
{
Threads<16> t(execute_task);
t.join();
}
class ObjectThatExecutesATask
{
public:
ObjectThatExecutesATask() : running_(false) {}
void ExecuteTask()
{
}
void Start()
{
running_ = true;
thread_ = std::thread
(
[this]
{
ExecuteTask();
std::lock_guard<std::mutex> lock(mutex_);
running_ = false;
cv_.notify_one();
}
);
}
void Stop()
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock,[this]{return !running_;});
}
private:
std::thread thread_;
std::mutex mutex_;
std::condition_variable cv_;
bool running_;
};
void test4()
{
ObjectThatExecutesATask x;
x.Start();
x.Stop();
}
///////////////////////////////////////////////////////////////////////////////////////////////////
/*
Q9400 machine Release x64
std::thread::hardware_concurrency() = 4
Increment std::atomic<long> with multiple threads 1.73 s 23.08 MHz
Increment std::atomic<long> 0.761 s 131.4 MHz
std::mutex 0.869 s 11.51 MHz
Read and write TLS 0.676 s 147.9 MHz
*/
void ThreadTimingTests(double timeForTestInSecs)
{
Tracer() << "Thread timing tests\n";
ceda::TraceIndenter indent(4);
double timePerTest = timeForTestInSecs / 18;
// Note: std::thread::hardware_concurrency() is only a hint, and the function might return 0
// if this information is not available
Tracer() << "std::thread::hardware_concurrency() = " << std::thread::hardware_concurrency() << '\n';
ceda::TimeCode("Increment std::atomic<long> with multiple threads", timePerTest, 10000000, 100000000,
[](int n)
{
const int NUMTHREADS = 4;
std::thread threads[NUMTHREADS];
std::atomic<long> v(0);
for (auto& t : threads)
{
t = std::thread
(
[&]
{
for (int i=0 ; i < n/NUMTHREADS ; ++i)
{
++v;
}
}
);
}
for (auto& t : threads)
{
t.join();
}
//Tracer() << "v = " << v << '\n';
cxAssert(v == n);
});
ceda::TimeCode("Increment std::atomic<long>", timePerTest, 10000000, 100000000,
[](int n)
{
std::atomic<long> v;
for (int i=0 ; i < n ; ++i)
{
++v;
}
});
ceda::TimeCode("Construct/Destruct std::mutex", timePerTest, 10000000, 10000000,
[](int n)
{
for (int i=0 ; i < n ; ++i)
{
std::mutex m;
}
});
ceda::TimeCode("Lock std::mutex", timePerTest, 10000000, 10000000,
[](int n)
{
std::mutex m;
for (int i=0 ; i < n ; ++i)
{
std::lock_guard<std::mutex> lock(m);
}
});
ceda::TimeCode("Construct/Destruct ManualResetEvent", timePerTest, 5000000, 10000000,
[](int n)
{
for (int i=0 ; i < n ; ++i)
{
ceda::ManualResetEvent e;
}
});
ceda::TimeCode("Signal(),Wait(),Reset() on ManualResetEvent", timePerTest, 1000000, 10000000,
[](int n)
{
ceda::ManualResetEvent e;
for (int i=0 ; i < n ; ++i)
{
e.Signal();
e.Wait();
e.Reset();
}
});
ceda::TimeCode("Read and write thread_local int*", timePerTest, 100000000, 1000000000,
[](int n)
{
static thread_local int* tls = nullptr;
[[maybe_unused]] volatile int k;
tls = new int;
for (int i=0 ; i < n ; ++i)
{
k = *tls; // read tls
*tls = i; // write tls
}
});
}