3
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ Did valgrind report any more details that you didn't include in the question? Can you explain in a sentence or two why you think "no race condition should be possible"? \$\endgroup\$ – GraniteRobert May 30 '14 at 19:57
  • \$\begingroup\$ Is all of that code necessary to answer your question? This is Code Review, but it is still a lot of code. \$\endgroup\$ – Paul Draper Jun 1 '14 at 2:14
1
\$\begingroup\$

I haven't taken time to understand all the code so I won't offer a firm answer to your problem. But I'll suggest where to start thinking about the problem.

Consider these code fragments:

state_write = set_pipeline_state_write()       // fragment 1, thread 1

if(state_write) {...}                          // fragment 2, thread 1

state_read = writer.set_pipeline_state_read()  // fragment 3, thread 2

The structure of the code suggests that the stuff in {...} in fragment 2 needs the pipeline to be in the "write" state. But I see nothing that guarantees that condition.

In particular, what happens if the threads overlap execution as follows?

  • Thread 1 executes fragment 1, so the local variable state_write is true.
  • Thread 2 executes fragment 3, so now the pipeline is in the "read" state.
  • Thread 1 executes fragment 2, tests local variable state_write and executes {...}, believing that the pipeline is in the "write" state when it is really in the "read" state.

This is just one of many possible ways the execution of two threads could overlap. On most systems, a thread can be preempted after any arbitrary CPU instruction. In some cases it can be interrupted in the middle of an instruction. Certainly a typical C++ statement can be interrupted several times.

While a thread is inactive, any global state can change unless the program does something to prevent it. In this case, it looks like state_write can be obsolete by the time the if statement consumes the value.

If the code overlaps as I outlined above, I think it will behave badly. Either

  1. The code is wrong because it thinks the pipeline is "write" when it is actually "read", or
  2. The if(state_write) test is irrelevant, in which case the intent of the code is completely misleading.
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.