I have a threaded pipeline with single producers and (possible) multiple consumers. A producer might wait or not, depending on the policy provided. A problem here is valgrind (helgrind) is reporting a race condition where no race condition should be possible (If so, my code is wrong).
#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <vector>
#include <thread>
namespace Thread {
enum class PipelinePolicy {
Wait,
Overwrite
};
namespace Detail {
template <typename Input> class PipelineReader;
template <typename Output> class PipelineWriter;
class PipelineThread
{
template <typename Input> friend class PipelineReader;
template <typename Output> friend class PipelineWriter;
// Types
// =====
public:
typedef std::chrono::steady_clock::time_point time_type;
typedef std::chrono::milliseconds period_type;
typedef unsigned size_type;
// Implementation
// ==============
protected:
class Implementation
{
friend class PipelineThread;
// Types
// =====
private:
typedef unsigned thread_state_type;
// Construction
// ============
protected:
Implementation()
: m_thread_state(ThreadStateIdle), m_wait_state(false)
{}
Implementation(const Implementation&) = delete;
Implementation& operator = (const Implementation&) = delete;
public:
/// Stop the thread.
virtual ~Implementation() {
stop();
}
// Element Access
// ==============
public:
std::thread::id id() const { return m_thread.get_id(); }
// State
// =====
private:
enum ThreadState {
ThreadStateStop,
ThreadStateIdle,
ThreadStateStart,
ThreadStateActive
};
public:
/// True, if the thread should start.
bool get_thread_state_start() const { return get_thread_state() == ThreadStateStart; }
/// True, if the thread should stop .
bool get_thread_state_active() const { return get_thread_state() == ThreadStateActive; }
/// True, if the thread should stop .
bool get_thread_state_stop() const { return get_thread_state() == ThreadStateStop; }
/// True, if the thread is joined.
bool get_thread_state_idle() const { return get_thread_state() == ThreadStateIdle; }
protected:
thread_state_type get_thread_state() const { return m_thread_state.load(std::memory_order_relaxed); }
private:
bool set_thread_state_start() {
thread_state_type test = ThreadStateIdle;
return m_thread_state.compare_exchange_strong(test, ThreadStateStart, std::memory_order_relaxed);
}
protected:
bool set_thread_state_active() {
thread_state_type test = ThreadStateStart;
return m_thread_state.compare_exchange_strong(test, ThreadStateActive, std::memory_order_relaxed);
}
private:
bool set_thread_state_stop() {
thread_state_type test = ThreadStateActive;
return m_thread_state.compare_exchange_strong(test, ThreadStateStop, std::memory_order_relaxed);
}
void set_thread_state_idle() {
m_thread_state.exchange(ThreadStateIdle, std::memory_order_relaxed);
}
// Start/Stop
// ==========
public:
virtual bool start() = 0;
protected:
template <typename Callable>
bool start_thread(Callable&& callable) {
if(set_thread_state_start()) {
m_thread = std::thread(std::move(callable));
return true;
}
return false;
}
public:
bool stop() {
bool result = set_thread_state_stop();
if(result) {
// Notify the possible waiting thread
resume();
}
if(get_thread_state_stop() && std::this_thread::get_id() != m_thread.get_id()) {
if(m_thread.joinable()) {
m_thread.join();
set_thread_state_idle();
}
}
return result;
}
// Wait/Resume
// ===========
private:
bool waiting() const { return m_wait_state.load(std::memory_order_relaxed); }
public:
bool wait() {
if(get_thread_state_stop()) return false;
else {
bool test = false;
if(m_wait_state.compare_exchange_strong(test, true)) {
std::unique_lock<std::mutex> lock(m_wait_mutex);
while(waiting() == true) {
m_wait_condition.wait(lock);
}
}
return ! get_thread_state_stop();
}
}
bool resume() {
bool test = true;
if(m_wait_state.compare_exchange_strong(test, false)) {
std::unique_lock<std::mutex> lock(m_wait_mutex);
if( ! waiting()) {
m_wait_condition.notify_one();
return true;
}
}
return false;
}
// Exception
// =========
protected:
/// Stop the pipeline and set the exception to the current active exception.
virtual void set_exception() {
set_thread_state_stop();
m_exception = std::current_exception();
// Notify the possible waiting thread
resume();
}
public:
/// The last exception in the pipeline.
std::exception_ptr exception() const { return m_exception; }
/// Throw the last exception in the pipeline, if availabe.
void rethrow_exception() const {
if(bool(m_exception))
std::rethrow_exception(m_exception);
}
// Data
// ====
private:
std::thread m_thread;
std::atomic<thread_state_type> m_thread_state;
std::mutex m_wait_mutex;
std::condition_variable m_wait_condition;
std::atomic<bool> m_wait_state;
std::exception_ptr m_exception;
};
protected:
Implementation& self() const { return *m_self; }
// Construction
// ============
protected:
void init(std::shared_ptr<Implementation>&& self) {
m_self = self;
}
// Element Access
// ==============
public:
std::thread::id id() const { return self().id(); }
// Exception
// =========
public:
/// The last exception in the pipeline.
std::exception_ptr exception() const { return self().exception(); }
/// Throw the last exception in the pipeline, if availabe.
void rethrow_exception() const {
self().rethrow_exception();
}
private:
std::shared_ptr<Implementation> m_self;
friend bool operator == (const PipelineThread&, const PipelineThread&) noexcept;
friend bool operator < (const PipelineThread&, const PipelineThread&) noexcept;
};
inline bool operator == (const PipelineThread& a, const PipelineThread& b) noexcept {
return a.m_self == b.m_self;
}
inline bool operator != (const PipelineThread& a, const PipelineThread& b) noexcept {
return ! (a == b);
}
inline bool operator < (const PipelineThread& a, const PipelineThread& b) noexcept {
return a.m_self < b.m_self;
}
inline bool operator <= (const PipelineThread& a, const PipelineThread& b) noexcept {
return ! (b < a);
}
inline bool operator > (const PipelineThread& a, const PipelineThread& b) noexcept {
return b < a;
}
inline bool operator >= (const PipelineThread& a, const PipelineThread& b) noexcept {
return ! (a < b);
}
template <typename Input>
class PipelineReader;
/// Generate output data.
template <typename Output>
class PipelineWriter : virtual public PipelineThread
{
friend class PipelineReader<Output>;
// Types
// =====
typedef PipelineThread Thread;
public:
typedef Output output_type;
typedef output_type signature();
// Implementation
// ==============
public:
class Implementation : virtual public Thread::Implementation
{
template <typename U> friend class PipelineReader<U>::Implementation;
// Types
// =====
private:
typedef unsigned pipeline_state_type;
// Construction
// ============
protected:
template <typename Period, typename PeriodRatio>
Implementation(
const std::chrono::duration<Period, PeriodRatio>& period, PipelinePolicy policy)
: m_period(std::chrono::duration_cast<period_type>(period)),
m_policy(policy),
m_pipeline_state(PipelineStateWriteReady)
{}
// Connections
// ===========
public:
/// Connect a reader.
/// \PRECONDITION The writer thread is idle.
/// \EXCEPTION std::logic_error, if the precondition is not met.
template <typename Input>
void connect(PipelineReader<Input>& reader) {
if( ! get_thread_state_idle()) throw std::logic_error("Pipeline: Connect");
if(std::find(m_readers.begin(), m_readers.end(), reader) == m_readers.end())
m_readers.push_back(reader);
}
// State
// =====
private:
enum PipelineState {
PipelineStateWriteReady,
PipelineStateWrite,
PipelineStateReadReady,
PipelineStateRead
};
pipeline_state_type get_pipeline_state() const {
return m_pipeline_state.load(std::memory_order_relaxed);
}
void set_pipeline_state_write_ready() {
m_pipeline_state.exchange(PipelineStateWriteReady, std::memory_order_relaxed);
}
bool set_pipeline_state_write() {
pipeline_state_type test = PipelineStateWriteReady;
if(m_pipeline_state.compare_exchange_strong(test, PipelineStateWrite, std::memory_order_relaxed)) return true;
else if(m_policy != PipelinePolicy::Wait && ! get_thread_state_stop()) {
test = PipelineStateReadReady;
return m_pipeline_state.compare_exchange_strong(test, PipelineStateWrite, std::memory_order_relaxed);
}
return false;
}
void set_pipeline_state_read_ready() {
m_pipeline_state.exchange(PipelineStateReadReady, std::memory_order_relaxed);
}
bool set_pipeline_state_read() {
pipeline_state_type test = PipelineStateReadReady;
return m_pipeline_state.compare_exchange_strong(test, PipelineStateRead, std::memory_order_relaxed);
}
// Start/Stop
// ==========
protected:
template <typename Callable>
bool start_thread(Callable&& callable) noexcept {
size_type result = ! m_readers.empty();
if(result) {
auto pos = m_readers.begin();
// Start readers
for( ; pos != m_readers.end(); ++pos) {
auto& reader = pos->self();
result = reader.start();
if(result) {
while( ! reader.get_thread_state_active()) std::this_thread::yield();
}
else break;
}
// Start the Writer
if(result) {
result = Thread::Implementation::start_thread(std::move(callable));
}
// Stop the readers
if( ! result) {
while(pos != m_readers.begin()) {
auto& reader = (--pos)->self();
reader.stop();
}
}
}
return result;
}
// Write
// =====
protected:
template <typename Callable>
bool write(Callable& write_function, time_type time) {
// Wait
bool state_write = false;
while( ! get_thread_state_stop()
&& ! (state_write = set_pipeline_state_write())
&& wait());
// Write, notify a reader and sleep (optionally)
if(state_write) {
// FIXME valgrind(helgrind) reports a race condition:
m_output = write_function();
set_pipeline_state_read_ready();
for(auto& reader : m_readers) {
if(reader.self().resume())
break;
}
if( ! this->get_thread_state_stop() && m_period != period_type::zero()) {
auto elapsed = std::chrono::steady_clock::now() - time;
auto sleep_period = m_period - (elapsed % m_period);
std::this_thread::sleep_for(sleep_period);
}
return true;
}
// Stop
return false;
}
template <typename Callable>
void write_loop(Callable& write_function, time_type time) {
try {
while(write(write_function, time));
// Pending data
if(get_pipeline_state() == PipelineStateReadReady) {
wait();
}
for(auto& reader : m_readers) {
reader.self().stop();
}
}
catch(...) {
set_exception();
}
}
// Data
// ====
private:
period_type m_period;
PipelinePolicy m_policy;
std::vector<PipelineThread> m_readers;
std::atomic<pipeline_state_type> m_pipeline_state;
output_type m_output;
};
protected:
Implementation& self() const {
return dynamic_cast<Implementation&>(Thread::self());
}
// Element Access
// ==============
public:
const period_type& period() const { return self().m_period; }
PipelinePolicy policy() const { return self().m_policy; }
};
// PipelineReader
// ==============
/// Process the output data of a writer.
template <typename Input>
class PipelineReader : virtual public PipelineThread
{
friend class PipelineWriter<Input>;
// Types
// =====
typedef PipelineThread Thread;
public:
typedef Input input_type;
typedef void signature(input_type);
// Implementation
// ==============
public:
class Implementation : virtual public Thread::Implementation
{
template <typename U> friend class PipelineWriter<U>::Implementation;
// Construction
// ============
public:
Implementation(PipelineWriter<Input> writer)
: m_writer(writer)
{}
void connect(PipelineReader& reader) { m_writer.self().connect(reader); }
// Element Access
// ==============
protected:
static typename Detail::PipelineWriter<Input>::Implementation&
get_writer(Implementation& pipeline) {
return pipeline.m_writer.self();
}
input_type& input() { return m_input; }
// Read
// ====
protected:
template <typename PipelineWriterImplementation>
bool read(PipelineWriterImplementation& writer)
{
// Wait
bool state_read = false;
while( ! get_thread_state_stop()
&& ! (state_read = writer.set_pipeline_state_read())
&& wait());
// Read and notify the writer
if(state_read) {
// FIXME valgrind(helgrind) reports a race condition:
m_input = std::move(writer.m_output);
writer.set_pipeline_state_write_ready();
writer.resume();
}
// Stop
return ! this->get_thread_state_stop();
}
template <typename PipelineWriterImplementation, typename Callable>
void read_loop(PipelineWriterImplementation& writer, Callable& read_function)
{
try {
while(read(writer)) {
read_function(m_input);
}
}
catch(...) {
set_exception();
}
}
// Exception
// =========
protected:
/// Stop the pipeline and set the exception of the reader and writer
/// to the current active exception.
virtual void set_exception() {
Thread::Implementation::set_exception();
m_writer.self().set_exception();
}
// Data
// ====
private:
PipelineWriter<Input> m_writer;
input_type m_input;
};
protected:
Implementation& self() const {
return dynamic_cast<Implementation&>(Thread::self());
}
void connect() { self().connect(*this); }
};
} // namespace Detail
/// The start point of a pipeline generating data.
template <typename Output>
class PipelineWriter : public Detail::PipelineWriter<Output>
{
// Types
// =====
private:
typedef Detail::PipelineThread Thread;
typedef Detail::PipelineWriter<Output> Writer;
public:
typedef typename Thread::time_type time_type;
typedef typename Thread::period_type period_type;
typedef typename Thread::size_type size_type;
typedef typename Writer::output_type output_type;
typedef output_type signature();
// Implementation
// ==============
public:
class Implementation : public Writer::Implementation
{
friend class PipelineWriter;
// Construction
// ============
public:
template <
typename Period, typename PeriodRatio,
typename Callable, typename... Arguments>
Implementation(
const std::chrono::duration<Period, PeriodRatio>& period,
PipelinePolicy policy,
Callable&& callable,
Arguments&&... arguments)
: Writer::Implementation(period, policy),
m_function(std::bind(
std::move(callable),
std::forward<Arguments>(arguments)...))
{}
// Start
// =====
private:
struct Invoke {
Implementation& pipeline;
Invoke(Implementation& pipeline) : pipeline(pipeline) {}
void operator () () const {
pipeline.set_thread_state_active();
time_type time = std::chrono::steady_clock::now();
pipeline.write_loop(pipeline.m_function, time);
}
};
// Start/Stop
// ==========
public:
bool start() {
return Writer::Implementation::start_thread(Invoke(*this));
}
private:
bool stop() {
bool result = Writer::Implementation::stop();
if(bool(this->exception())) {
while( ! this->get_thread_state_idle()) std::this_thread::yield();
this->rethrow_exception();
}
return result;
};
private:
std::function<signature> m_function;
};
protected:
Implementation& self() const {
return dynamic_cast<Implementation&>(Thread::self());
}
// Construction
// ============
public:
template <typename Callable, typename... Arguments>
PipelineWriter(
PipelinePolicy policy,
Callable&& callable, Arguments&&... arguments)
{
this->init(std::make_shared<Implementation>(
period_type::zero(),
policy,
std::move(callable),
std::forward<Arguments>(arguments)...));
}
template <
typename Period, typename PeriodRatio,
typename Callable, typename... Arguments>
PipelineWriter(
const std::chrono::duration<Period, PeriodRatio>& period,
PipelinePolicy policy,
Callable&& callable,
Arguments&&... arguments)
{
this->init(std::make_shared<Implementation>(
period,
policy,
std::move(callable),
std::forward<Arguments>(arguments)...));
}
// Start/Stop
// ==========
public:
/// Start the entire pipeline.
bool start() { return self().start(); }
/// Stop the entire pipeline.
bool stop() { return self().stop(); }
};
/// An intermediate element in a pipeline consuming and processing data.
template <typename Input, typename Output>
class Pipeline
: public Detail::PipelineReader<Input>,
public Detail::PipelineWriter<Output>
{
// Types
// =====
private:
typedef Detail::PipelineThread Thread;
typedef Detail::PipelineReader<Input> Reader;
typedef Detail::PipelineWriter<Output> Writer;
public:
typedef typename Thread::time_type time_type;
typedef typename Thread::period_type period_type;
typedef typename Thread::size_type size_type;
typedef typename Reader::input_type input_type;
typedef typename Writer::output_type output_type;
typedef output_type signature(input_type);
// Implementation
// ==============
public:
class Implementation
: public Reader::Implementation,
public Writer::Implementation
{
// Construction
// ============
public:
template <
typename Period, typename PeriodRatio,
typename Callable, typename... Arguments>
Implementation(
const std::chrono::duration<Period, PeriodRatio>& period,
PipelinePolicy policy,
Detail::PipelineWriter<Input> writer,
Callable&& callable,
Arguments&&... arguments)
: Reader::Implementation(writer),
Writer::Implementation(period, policy),
m_function(std::bind(
std::move(callable),
std::placeholders::_1,
std::forward<Arguments>(arguments)...))
{}
// Invoke
// ======
private:
struct Write {
Implementation& pipeline;
Write(Implementation& pipeline) : pipeline(pipeline) {}
output_type operator () () const {
auto& writer = Reader::Implementation::get_writer(pipeline);
pipeline.read(writer);
return pipeline.m_function(pipeline.input());
}
};
struct Invoke {
Write write;
Invoke(Implementation& pipeline) : write(pipeline) {}
void operator () () const {
write.pipeline.set_thread_state_active();
time_type time = std::chrono::steady_clock::now();
write.pipeline.write_loop(write, time);
}
};
// Start/Stop
// ==========
public:
bool start() {
return Writer::Implementation::start_thread(Invoke(*this));
}
private:
std::function<signature> m_function;
};
protected:
Implementation& self() const {
return dynamic_cast<Implementation&>(Thread::self());
}
// Construction
// ============
public:
template <typename Callable, typename... Arguments>
Pipeline(
PipelinePolicy policy,
Detail::PipelineWriter<Input> writer,
Callable&& callable, Arguments&&... arguments)
{
this->init(std::make_shared<Implementation>(
period_type::zero(),
policy,
writer,
std::move(callable),
std::forward<Arguments>(arguments)...));
this->connect();
}
template <
typename Period, typename PeriodRatio,
typename Callable, typename... Arguments>
Pipeline(
const std::chrono::duration<Period, PeriodRatio>& period,
PipelinePolicy policy,
Detail::PipelineWriter<Input> writer,
Callable&& callable,
Arguments&&... arguments)
{
this->init(std::make_shared<Implementation>(
period,
policy,
writer,
std::move(callable),
std::forward<Arguments>(arguments)...));
this->connect();
}
};
/// An end point of a pipeline consuming data.
template <typename Input>
class PipelineReader : public Detail::PipelineReader<Input>
{
// Types
// =====
private:
typedef Detail::PipelineThread Thread;
typedef Detail::PipelineReader<Input> Reader;
public:
typedef typename Thread::time_type time_type;
typedef typename Thread::period_type period_type;
typedef typename Thread::size_type size_type;
typedef Input input_type;
typedef void signature(input_type);
// Implementation
// ==============
public:
class Implementation : public Reader::Implementation
{
// Construction
// ============
public:
template <typename Callable, typename... Arguments>
Implementation(
Detail::PipelineWriter<Input> writer,
Callable&& callable,
Arguments&&... arguments)
: Reader::Implementation(writer),
m_function(std::bind(
std::move(callable),
std::placeholders::_1,
std::forward<Arguments>(arguments)...))
{}
// Invoke
// ======
public:
struct Invoke {
Implementation& pipeline;
Invoke(Implementation& pipeline) : pipeline(pipeline) {}
void operator () () const {
pipeline.set_thread_state_active();
auto& writer = Reader::Implementation::get_writer(pipeline);
pipeline.read_loop(writer, pipeline.m_function);
}
};
// Start/Stop
// ==========
public:
bool start() {
return Reader::Implementation::start_thread(Invoke(*this));
}
private:
std::function<signature> m_function;
};
protected:
Implementation& self() const {
return dynamic_cast<Implementation&>(Thread::self());
}
// Construction
// ============
public:
template <typename Callable, typename... Arguments>
PipelineReader(
Detail::PipelineWriter<Input> writer,
Callable&& callable, Arguments&&... arguments)
{
this->init(std::make_shared<Implementation>(
writer,
std::move(callable),
std::forward<Arguments>(arguments)...));
this->connect();
}
};
} // namespace Thread;
// Test
// ====
#include<iostream>
using Thread::PipelinePolicy;
using Thread::PipelineWriter;
using Thread::Pipeline;
using Thread::PipelineReader;
int produce() {
static int counter = 1;
return counter++;
}
double forward(int data) {
std::this_thread::sleep_for(std::chrono::milliseconds(3));
return data / 10.0;
}
static const unsigned ConsumerCount = 3;
std::vector<double> consumer_data[ConsumerCount];
bool consume(double data, unsigned which) {
if(2 <= data)
throw std::runtime_error("Exception");
std::this_thread::sleep_for(std::chrono::milliseconds(10));
consumer_data[which].push_back(data);
return true;
}
int main() {
try {
const auto policy = PipelinePolicy::Overwrite;
PipelineWriter<int> producer(
std::chrono::milliseconds(3),
policy,
produce);
Pipeline<int, double> forwarder(
std::chrono::milliseconds(2),
policy,
producer,
forward);
for(unsigned i = 0; i < ConsumerCount; ++i) {
PipelineReader<double>(
forwarder,
consume,
i);
}
try {
producer.start();
std::this_thread::sleep_for(std::chrono::milliseconds(100));
producer.stop();
}
catch(const std::exception& e) {
std::cerr << "Expected " << e.what() << '\n';
}
for(unsigned i = 0; i < ConsumerCount; ++i) {
std::cout << "Thread " << i << ":\n";
for(const auto& d : consumer_data[i]) {
std::cout << d << '\n';
}
}
}
catch(const std::exception& e) {
std::cerr << "Unexpected " << e.what() << '\n';
}
}
Note: The lines in question have a comment starting with // FIXME
Note: I eliminated some comments to make it fit into the size, allowed.
