async.hpp

Go to the documentation of this file.

#ifndef CORAL_ASYNC_HPP
#define CORAL_ASYNC_HPP

#include <cassert>
#include <exception>
#include <functional>
#include <future>
#include <memory>
#include <thread>
#include <utility>

#include "zmq.hpp"

#include "coral/config.h"
#include "coral/error.hpp"
#include "coral/log.hpp"
#include "coral/net/reactor.hpp"
#include "coral/net/zmqx.hpp"
#include "coral/util.hpp"


namespace coral
{
namespace async
{


template<typename StackData, typename Result>
struct CommThreadTask
{
    using Type = std::function<void(
        coral::net::Reactor&,
        StackData&,
        std::promise<Result>)>;
};

// Specialisation of the above for `StackData = void`.
template<typename Result>
struct CommThreadTask<void, Result>
{
    using Type = std::function<void(
        coral::net::Reactor&,
        std::promise<Result>)>;
};


namespace detail
{
    template<typename StackData>
    struct CommThreadAnyTask
    {
        using Type = std::function<void(coral::net::Reactor&, StackData&)>;
        using SharedPtr = std::shared_ptr<Type>;
        using WeakPtr = std::weak_ptr<Type>;
    };

    template<>
    struct CommThreadAnyTask<void>
    {
        using Type = std::function<void(coral::net::Reactor&)>;
        using SharedPtr = std::shared_ptr<Type>;
        using WeakPtr = std::weak_ptr<Type>;
    };
} // namespace detail


template<typename StackData>
class CommThread
{
public:
    explicit CommThread();

    ~CommThread() CORAL_NOEXCEPT;

    // Copying is disabled
    CommThread(const CommThread&) = delete;
    CommThread& operator=(const CommThread&) = delete;

    CommThread(CommThread&& other) CORAL_NOEXCEPT;

    CommThread& operator=(CommThread&& other) CORAL_NOEXCEPT;

    template<typename Result>
    std::future<Result> Execute(
        typename CommThreadTask<StackData, Result>::Type task);

    void Shutdown();

    bool Active() const CORAL_NOEXCEPT;

private:
    // Waits for the background thread to terminate and performs cleanup
    void WaitForThreadTermination();
    void DestroySilently() CORAL_NOEXCEPT;

    bool m_active;
    zmq::socket_t m_socket;
    std::future<void> m_threadStatus;
    typename detail::CommThreadAnyTask<StackData>::WeakPtr m_nextTask;
};


class CommThreadDead : public std::exception
{
public:
    CommThreadDead(std::exception_ptr originalException) CORAL_NOEXCEPT;

    std::exception_ptr OriginalException() const CORAL_NOEXCEPT;

    const char* what() const CORAL_NOEXCEPT override;

private:
    std::exception_ptr m_originalException;
};


// =============================================================================
// Templates
// =============================================================================

namespace detail
{
    template<typename StackData>
    void CommThreadMessagingLoop(
        zmq::socket_t& bgSocket,
        typename CommThreadAnyTask<StackData>::SharedPtr nextTask)
    {
        coral::net::Reactor reactor;
        StackData data;
        reactor.AddSocket(
            bgSocket,
            [nextTask, &data] (coral::net::Reactor& r, zmq::socket_t& s) {
                char dummy;
                s.recv(&dummy, 1);

                // Now, the foreground thread is blocked waiting for our
                // reply, so we can freely access `nextTask`.
                typename CommThreadAnyTask<StackData>::Type myTask;
                swap(*nextTask, myTask);

                // Unblock foreground thread again and then run the task.
                s.send("", 0);
                myTask(r, data);
            });
        reactor.Run();
    }

    template<>
    inline void CommThreadMessagingLoop<void>(
        zmq::socket_t& bgSocket,
        typename CommThreadAnyTask<void>::SharedPtr nextTask)
    {
        coral::net::Reactor reactor;
        reactor.AddSocket(
            bgSocket,
            [nextTask] (coral::net::Reactor& r, zmq::socket_t& s) {
                char dummy;
                s.recv(&dummy, 1);

                // Now, the foreground thread is blocked waiting for our
                // reply, so we can freely access `nextTask`.
                CommThreadAnyTask<void>::Type myTask;
                swap(*nextTask, myTask);

                // Unblock foreground thread again and then run the task.
                s.send("", 0);
                myTask(r);
            });
        reactor.Run();
    }

    // Note: Some of the parameters are only shared_ptr because VS2013 has
    // a bug that prevents the use of move-only objects as arguments to the
    // thread function.
    template<typename StackData>
    void CommThreadBackground(
        std::shared_ptr<zmq::socket_t> bgSocket,
        std::shared_ptr<std::promise<void>> statusNotifier,
        typename CommThreadAnyTask<StackData>::SharedPtr nextTask)
        CORAL_NOEXCEPT
    {
        try {
            CommThreadMessagingLoop<StackData>(
                *bgSocket,
                std::move(nextTask));

            // We should possibly use set_value_at_thread_exit() and
            // set_exception_at_thread_exit() in the following, but those are
            // not supported in GCC 4.9.
            statusNotifier->set_value();
        } catch (...) {
            statusNotifier->set_exception(
                std::current_exception());
        }

        // This is to avoid the potential race condition where the background
        // thread dies after the foreground thread has sent a task notification
        // and is waiting to receive an acknowledgement.
        bgSocket->send("", 0);
    }
} // namespace detail


template<typename StackData>
CommThread<StackData>::CommThread()
    : m_active{true}
    , m_socket{coral::net::zmqx::GlobalContext(), ZMQ_PAIR}
    , m_threadStatus{}
    , m_nextTask{}
{
    auto bgSocket =
        std::make_shared<zmq::socket_t>(coral::net::zmqx::GlobalContext(), ZMQ_PAIR);
    bgSocket->setsockopt(ZMQ_LINGER, 0);
    m_socket.setsockopt(ZMQ_LINGER, 0);
    const auto endpoint = "inproc://" + coral::util::RandomUUID();
    bgSocket->bind(endpoint);
    m_socket.connect(endpoint);

    auto statusNotifier = std::make_shared<std::promise<void>>();
    m_threadStatus = statusNotifier->get_future();

    auto sharedTask =
        std::make_shared<typename detail::CommThreadAnyTask<StackData>::Type>(/*empty*/);
    m_nextTask = sharedTask;

    std::thread{&detail::CommThreadBackground<StackData>,
        bgSocket, statusNotifier, sharedTask}.detach();
}


template<typename StackData>
CommThread<StackData>::~CommThread() CORAL_NOEXCEPT
{
    DestroySilently();
}


template<typename StackData>
CommThread<StackData>::CommThread(CommThread&& other) CORAL_NOEXCEPT
    : m_active{other.m_active}
    , m_socket{std::move(other.m_socket)}
    , m_threadStatus{std::move(other.m_threadStatus)}
    , m_nextTask{std::move(other.m_nextTask)}
{
    other.m_active = false;
}


template<typename StackData>
CommThread<StackData>& CommThread<StackData>::operator=(CommThread&& other)
    CORAL_NOEXCEPT
{
    DestroySilently();
    m_active = other.m_active;
    m_socket = std::move(other.m_socket);
    m_threadStatus = std::move(other.m_threadStatus);
    m_nextTask = std::move(other.m_nextTask);
    other.m_active = false;
    return *this;
}


namespace detail
{
    template<typename StackData, typename Result>
    struct CommThreadFunctions
    {
        static typename CommThreadAnyTask<StackData>::Type WrapTask(
            typename CommThreadTask<StackData, Result>::Type task,
            std::promise<Result> promise)
        {
            const auto sharedPromise =
                std::make_shared<typename std::promise<Result>>(std::move(promise));
            return [task, sharedPromise]
                (coral::net::Reactor& reactor, StackData& data)
            {
                task(reactor, data, std::move(*sharedPromise));
            };
        }
    };


    template<typename Result>
    struct CommThreadFunctions<void, Result>
    {
        static typename CommThreadAnyTask<void>::Type WrapTask(
            typename CommThreadTask<void, Result>::Type task,
            std::promise<Result> promise)
        {
            const auto sharedPromise =
                std::make_shared<typename std::promise<Result>>(std::move(promise));
            return [task, sharedPromise]
                (coral::net::Reactor& reactor)
            {
                task(reactor, std::move(*sharedPromise));
            };
        }
    };
} // namespace detail


template<typename StackData>
template<typename Result>
std::future<Result> CommThread<StackData>::Execute(
    typename CommThreadTask<StackData, Result>::Type task)
{
    CORAL_PRECONDITION_CHECK(Active());
    CORAL_INPUT_CHECK(task);

    auto promise = std::promise<Result>{};
    auto future = promise.get_future();

    if (auto sharedTask = m_nextTask.lock()) {
        assert(!(*sharedTask));
        *sharedTask = detail::CommThreadFunctions<StackData, Result>::WrapTask(
            std::move(task),
            std::move(promise));

        // Notify background thread that a task is ready and wait for it
        // to acknowledge it before returning.
        m_socket.send("", 0);
        char dummy;
        m_socket.recv(&dummy, 1);
        return future;
    } else {
        // The weak_ptr has expired, meaning that the thread (which holds
        // a shared_ptr to the same object) must be dead.
        WaitForThreadTermination();

        // The above function should have thrown.  If it didn't, it probably
        // means that calling code did something stupid.
        coral::log::Log(
            coral::log::error,
            "CommThread background thread has terminated silently and "
            "unexpectedly.  Perhaps Reactor::Stop() was called?");
        std::terminate();
    }
}


namespace detail
{
    inline void CommThreadShutdown(
        coral::net::Reactor& reactor,
        std::promise<void> promise)
    {
        reactor.Stop();
        promise.set_value();
    }

    template<typename StackData>
    typename CommThreadTask<StackData, void>::Type CommThreadShutdownTask()
    {
        return [] (coral::net::Reactor& r, StackData&, std::promise<void> p)
        {
            CommThreadShutdown(r, std::move(p));
        };
    }

    template<>
    inline typename CommThreadTask<void, void>::Type CommThreadShutdownTask<void>()
    {
        return [] (coral::net::Reactor& r, std::promise<void> p)
        {
            CommThreadShutdown(r, std::move(p));
        };
    }
} // namespace detail


template<typename StackData>
void CommThread<StackData>::Shutdown()
{
    Execute<void>(detail::CommThreadShutdownTask<StackData>());
    WaitForThreadTermination();
}


template<typename StackData>
bool CommThread<StackData>::Active() const CORAL_NOEXCEPT
{
    return m_active;
}


template<typename StackData>
void CommThread<StackData>::WaitForThreadTermination()
    {
        assert(m_active);
        assert(m_threadStatus.valid());
        try {
            const auto cleanup = coral::util::OnScopeExit([this] ()
            {
                m_active = false;
                m_socket.close();
#ifdef _MSC_VER
                // Visual Studio does not "reset" the future after get().
                // See:  http://stackoverflow.com/q/33899615
                m_threadStatus = std::future<void>{};
#endif
            });
            m_threadStatus.get();
        }
#ifdef _MSC_VER
        // Visual Studio (versions up to and including 2015, at least)
        // has a bug where std::current_exception() returns a null
        // exception pointer if it is called during stack unwinding.
        // We have to do the best we can in this case, and special-case
        // some well-known exceptions.
        // See: http://stackoverflow.com/q/29652438
        catch (const zmq::error_t& e) {
            if (std::current_exception()) {
                throw CommThreadDead(std::current_exception());
            } else {
                throw CommThreadDead(std::make_exception_ptr(e));
            }
        }
        catch (const std::runtime_error& e) {
            if (std::current_exception()) {
                throw CommThreadDead(std::current_exception());
            } else {
                throw CommThreadDead(std::make_exception_ptr(e));
            }
        }
        catch (const std::logic_error& e) {
            if (std::current_exception()) {
                throw CommThreadDead(std::current_exception());
            } else {
                throw CommThreadDead(std::make_exception_ptr(e));
            }
        }
        catch (const std::exception& e) {
            if (std::current_exception()) {
                throw CommThreadDead(std::current_exception());
            } else {
                throw CommThreadDead(std::make_exception_ptr(
                    std::runtime_error(e.what())));
            }
        }
#else
        catch (...) {
            throw CommThreadDead(std::current_exception());
        }
#endif
    }


#ifdef _MSC_VER
#   pragma warning(push)
#   pragma warning(disable: 4101) // Unreferenced local variable 'e'
#endif

template<typename StackData>
void CommThread<StackData>::DestroySilently() CORAL_NOEXCEPT
{
    if (Active()) {
        try {
            Shutdown();
        } catch (const CommThreadDead& e) {
            try {
                std::rethrow_exception(e.OriginalException());
            } catch (const std::exception& e) {
                CORAL_LOG_DEBUG(
                    boost::format("Unexpected exception thrown in CommThread destructor: %s")
                    % e.what());
            }
        } catch (const std::exception& e) {
            CORAL_LOG_DEBUG(
                boost::format("Unexpected exception thrown in CommThread destructor: %s")
                % e.what());
        }
    }
    assert(!Active());
    assert(!static_cast<void*>(m_socket));
    assert(!m_threadStatus.valid());
}

#ifdef _MSC_VER
#   pragma warning(pop)
#endif


}} // namespace
#endif // header guard