2
\$\begingroup\$

I have implemented a class which encapsulates a thread of execution and provides an interface to send messages to it. It is based largely on an article written by Herb Sutter in 2010. It also uses the pImpl idiom as described by Scott Meyers in Effective Modern C++.

A comment on Stack Overflow from a guy with a high reputation recently stated that storing an std::thread in a unique_ptr was utter nonsense.

Now I'm wondering: Is there anything wrong with my implementation? Are there ways in which it could be improved?

thread.h

/**
 * \brief   Encapsulates a thread of execution and provides an interface to send messages to it.
 *
 * \details The object owns its own private thread and runs all of its work on that private thread. 
 *
 *          Calls on an active object should always be non-blocking asynchronous messages, so when a caller invokes
 *          the operation to send a message, the call just queues the message and returns immediately to the caller.
 *
 *          The thread mainline is an event-driven message pump that dequeues and executes the messages one at a 
 *          time on the private thread itself. Because the messages are processed sequentially, they are atomic with
 *          respect to each other. And because the object's private data is only accessed from the thread, there is
 *          no need to take a mutex lock or perform other synchronization on the private shared state.
 *
 *          The lifetime of the thread is tied directly to the lifetime of the object itself.
 */
class Thread
{
public:
    /**
     * \brief Starts the thread and message pump.
     *
     * \param name The name of the thread. Required only to aid debugging.
     */
    Thread(const std::string& name);

    /**
     * \brief   Stops the thread and message pump.
     *
     * \details A sentinel message is added behind any messages already waiting in the queue to signal the thread
     *          to finish and then it joins with the thread to wait for it to drain the queue and end.
     *
     *          Note that this is the only blocking call from the viewpoint of the caller.
     */
    virtual ~Thread();

    /**
     * \brief Queues the given message.
     *
     * \param message The message to queue.
     *
     * \return None
     */
    void send(std::function<void()> message);

    /**
     * \brief Queues a mesage to stop the message pump. effectively causing the thread to run to completion.
     *
     * \return None
     */
    void stop();

    /**
     * \brief Returns whether or not the thread has terminated.
     *
     * \return True if the thread has terminated, false otherwise.
     */
    bool hasTerminated() const;

private:
    Thread();
    Thread(const Thread&);
    const Thread& operator=(const Thread&);

    /** Private implementation details. */
    struct Impl;
    const std::unique_ptr<Impl> m_impl;
};

thread.cpp

/** Encapsulates the private implementation details of Thread. */
struct Thread::Impl
{
public:
    Impl(const std::string& name) :
        m_name(name),
        m_isDone(false),
        m_hasTerminated(false),
        m_queue(),
        m_thread()
    {
    }

    virtual ~Impl()
    {
    }

private:
    Impl();
    Impl(const Impl&);
    const Impl& operator=(const Impl&);

    /**
     * \brief The dispatch loop. Takes messages from the queue one at a time and processes them until it receives
     *        the sentinel message to terminate. If the queue is empty, it waits until a job is available.
     *
     * \return None
     */
    void run()
    {
        try
        {
            while (!m_isDone)
            {
                // wait until a job is available, then retrieve it and execute it on this thread
                std::shared_ptr<std::function<void()>> func = m_queue.pop();

                // execute the message
                (*func.get())();
            }
        }

        catch (const std::exception& ex)
        {
            // log error
        }

        catch (...)
        {
            // log error
        }

        m_hasTerminated = true;
    }

    /** The name of the thread. This is to aid debugging. */
    const std::string m_name;

    /** Flag to terminate the message pump. */
    bool m_isDone;

    /** Flag indicating the thread has terminated - i.e. run to completion. */
    std::atomic<bool> m_hasTerminated;

    /** The message queue. */
    SharedQueue<std::function<void()>> m_queue;

    /** The thread of execution that dequeues and executes the messages one at a time. */
    std::unique_ptr<std::thread> m_thread;

    friend class Thread;
};

Thread::Thread(const std::string& name) : m_impl(new Impl(name))
{
    // start the thread
    m_impl->m_thread = std::unique_ptr<std::thread>(new std::thread( [=]{ m_impl->run(); } ));
}

Thread::~Thread()
{
    // tell thread to exit if it hasn't already been told

    if (!m_impl->m_isDone)
    {
        send( [=]{ m_impl->m_isDone = true; } );
    }

    if (m_impl->m_thread->joinable())
    {
        // wait for the thread to terminate (i.e. run() to return)
        m_impl->m_thread->join();
    }
}

void Thread::send(std::function<void()> message)
{
    m_impl->m_queue.push(message);
}

void Thread::stop()
{
    // tell thread to exit
    send( [=]{ m_impl->m_isDone = true; } );
}

bool Thread::hasTerminated() const
{
    return m_impl->m_hasTerminated;
}
\$\endgroup\$

Your Answer

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

Browse other questions tagged or ask your own question.