AsioAsyncOp.h

/********************************************************************************
 *    Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH    *
 *                                                                              *
 *              This software is distributed under the terms of the             *
 *              GNU Lesser General Public Licence (LGPL) version 3,             *
 *                  copied verbatim in the file "LICENSE"                       *
 ********************************************************************************/

#ifndef FAIR_MQ_SDK_ASIOASYNCOP_H
#define FAIR_MQ_SDK_ASIOASYNCOP_H

#include <asio/associated_allocator.hpp>
#include <asio/associated_executor.hpp>
#include <asio/executor_work_guard.hpp>
#include <asio/system_executor.hpp>
#include <chrono>
#include <exception>
#include <fairmq/sdk/Error.h>
#include <fairmq/sdk/Traits.h>
#include <functional>
#include <memory>
#include <system_error>
#include <type_traits>
#include <utility>

#include <fairlogger/Logger.h>
#ifndef FAIR_LOG
#define FAIR_LOG LOG
#endif /* ifndef FAIR_LOG */

namespace fair {
namespace mq {
namespace sdk {

template<typename... SignatureArgTypes>
struct AsioAsyncOpImplBase
{
    virtual auto Complete(std::error_code, SignatureArgTypes...) -> void = 0;
    virtual auto IsCompleted() const -> bool = 0;
};

template<typename Executor1, typename Allocator1, typename Handler, typename... SignatureArgTypes>
struct AsioAsyncOpImpl : AsioAsyncOpImplBase<SignatureArgTypes...>
{
    using Allocator2 = typename asio::associated_allocator<Handler, Allocator1>::type;

    using Executor2 = typename asio::associated_executor<Handler, Executor1>::type;

    AsioAsyncOpImpl(const Executor1& ex1, Allocator1 alloc1, Handler&& handler)
        : fWork1(ex1)
        , fWork2(asio::get_associated_executor(handler, ex1))
        , fHandler(std::move(handler))
        , fAlloc1(std::move(alloc1))
    {}

    auto GetAlloc2() const -> Allocator2 { return asio::get_associated_allocator(fHandler, fAlloc1); }
    auto GetEx2() const -> Executor2 { return asio::get_associated_executor(fWork2); }

    auto Complete(std::error_code ec, SignatureArgTypes... args) -> void override
    {
        if (IsCompleted()) {
            throw RuntimeError("Async operation already completed");
        }

        GetEx2().dispatch(
            [=, handler = std::move(fHandler)]() mutable {
                try {
                    handler(ec, args...);
                } catch (const std::exception& e) {
                    FAIR_LOG(error) << "Uncaught exception in AsioAsyncOp completion handler: " << e.what();
                } catch (...) {
                    FAIR_LOG(error) << "Unknown uncaught exception in AsioAsyncOp completion handler.";
                }
            },
            GetAlloc2());

        fWork1.reset();
        fWork2.reset();
    }

    auto IsCompleted() const -> bool override
    {
        return !fWork1.owns_work() && !fWork2.owns_work();
    }

  private:
    asio::executor_work_guard<Executor1> fWork1;
    asio::executor_work_guard<Executor2> fWork2;
    Handler fHandler;
    Allocator1 fAlloc1;
};

template<typename Executor, typename Allocator, typename CompletionSignature>
struct AsioAsyncOp
{
};

template<typename Executor,
         typename Allocator,
         typename SignatureReturnType,
         typename SignatureFirstArgType,
         typename... SignatureArgTypes>
struct AsioAsyncOp<Executor,
                   Allocator,
                   SignatureReturnType(SignatureFirstArgType, SignatureArgTypes...)>
{
    static_assert(std::is_void<SignatureReturnType>::value,
                  "return value of CompletionSignature must be void");
    static_assert(std::is_same<SignatureFirstArgType, std::error_code>::value,
                  "first argument of CompletionSignature must be std::error_code");
    using Duration = std::chrono::milliseconds;

  private:
    using Impl = AsioAsyncOpImplBase<SignatureArgTypes...>;
    using ImplPtr = std::unique_ptr<Impl, std::function<void(Impl*)>>;
    ImplPtr fImpl;

  public:
    AsioAsyncOp()
        : fImpl(nullptr)
    {}

    template<typename Handler>
    AsioAsyncOp(Executor ex1, Allocator alloc1, Handler&& handler)
        : AsioAsyncOp()
    {
        // Async operation type to be allocated and constructed
        using Op = AsioAsyncOpImpl<Executor, Allocator, Handler, SignatureArgTypes...>;

        // Create allocator for concrete op type
        // Allocator2, see https://www.boost.org/doc/libs/1_70_0/doc/html/boost_asio/reference/asynchronous_operations.html#boost_asio.reference.asynchronous_operations.allocation_of_intermediate_storage
        using OpAllocator =
            typename std::allocator_traits<typename Op::Allocator2>::template rebind_alloc<Op>;
        OpAllocator opAlloc;

        // Allocate memory
        auto mem(std::allocator_traits<OpAllocator>::allocate(opAlloc, 1));

        // Construct object
        auto ptr(new (mem) Op(std::move(ex1),
                              std::move(alloc1),
                              std::forward<Handler>(handler)));

        // Assign ownership to this object
        fImpl = ImplPtr(ptr, [opAlloc](Impl* p) mutable {
            std::allocator_traits<OpAllocator>::deallocate(opAlloc, static_cast<Op*>(p), 1);
        });
    }

    template<typename Handler>
    AsioAsyncOp(Executor ex1, Handler&& handler)
        : AsioAsyncOp(std::move(ex1), Allocator(), std::forward<Handler>(handler))
    {}

    template<typename Handler>
    explicit AsioAsyncOp(Handler&& handler)
        : AsioAsyncOp(asio::system_executor(), std::forward<Handler>(handler))
    {}

    auto IsCompleted() -> bool { return (fImpl == nullptr) || fImpl->IsCompleted(); }

    auto Complete(std::error_code ec, SignatureArgTypes... args) -> void
    {
        if(IsCompleted()) {
            throw RuntimeError("Async operation already completed");
        }

        fImpl->Complete(ec, args...);
        fImpl.reset(nullptr);
    }

    auto Complete(SignatureArgTypes... args) -> void
    {
        Complete(std::error_code(), args...);
    }

    auto Cancel(SignatureArgTypes... args) -> void
    {
        Complete(MakeErrorCode(ErrorCode::OperationCanceled), args...);
    }

    auto Timeout(SignatureArgTypes... args) -> void
    {
        Complete(MakeErrorCode(ErrorCode::OperationTimeout), args...);
    }
};

} /* namespace sdk */
} /* namespace mq */
} /* namespace fair */

#endif /* FAIR_MQ_SDK_ASIOASYNCOP_H */