store_call.h

Go to the documentation of this file.

 
#pragma once
 
#include <neograph/api.h>
#include <neograph/program/blocking.h>
 
#include <asio/awaitable.hpp>
#include <asio/co_spawn.hpp>
#include <asio/io_context.hpp>
#include <asio/use_awaitable.hpp>
 
#include <exception>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <type_traits>
#include <utility>
 
namespace neograph::async {
NEOGRAPH_API bool in_host_turn() noexcept;
}  // namespace neograph::async
 
namespace neograph::program {
 
struct BlockingStorePolicy {
    std::function<std::string()> refuse;
 
    bool await_synchronous_stores = false;
};
 
namespace detail {
 
class NEOGRAPH_PROGRAM_API InlineStoreCalls {
public:
    InlineStoreCalls() noexcept;
    ~InlineStoreCalls();
    InlineStoreCalls(const InlineStoreCalls&)            = delete;
    InlineStoreCalls& operator=(const InlineStoreCalls&) = delete;
    static bool active() noexcept;
};
 
NEOGRAPH_PROGRAM_API bool offload_store_call(bool offloads) noexcept;
 
class NEOGRAPH_PROGRAM_API BlockingIoContext {
public:
    BlockingIoContext();
    ~BlockingIoContext();
    BlockingIoContext(const BlockingIoContext&)            = delete;
    BlockingIoContext& operator=(const BlockingIoContext&) = delete;
    asio::io_context&  get() noexcept { return *io_; }
    void discard() noexcept { reusable_ = false; }
 
private:
    std::unique_ptr<asio::io_context> io_;
    bool                              reusable_ = true;
};
 
NEOGRAPH_PROGRAM_API asio::awaitable<void> offload_to_blocking_pool(std::function<void()> work);
 
template <typename R, typename Fn>
asio::awaitable<R> offload_sync_call(Fn fn) {
    auto owned = std::make_shared<Fn>(std::move(fn));
    if constexpr (std::is_void_v<R>) {
        co_await offload_to_blocking_pool([owned] { (*owned)(); });
        co_return;
    } else {
        auto result = std::make_shared<std::optional<R>>();
        co_await offload_to_blocking_pool([owned, result] { result->emplace((*owned)()); });
        co_return std::move(**result);
    }
}
 
template <typename R>
asio::awaitable<R> failed_store_call(std::exception_ptr error) {
    std::rethrow_exception(std::move(error));
    if constexpr (!std::is_void_v<R>) co_return R{};  // never reached; this is what makes the function a coroutine
}
 
template <typename R>
asio::awaitable<std::optional<R>> box_store_call(asio::awaitable<R> call) {
    co_return std::optional<R>(co_await std::move(call));
}
 
[[noreturn]] inline void refuse_blocking_store_call(std::string_view operation, std::string reason) {
    throw ProgramBlockingCallRefused(std::string(operation), {}, std::move(reason));
}
 
inline void check_blocking_store_call(const BlockingStorePolicy& policy, std::string_view operation) {
    if (policy.refuse)
        if (auto reason = policy.refuse(); !reason.empty()) refuse_blocking_store_call(operation, std::move(reason));
    // A view that has no policy (the catalog's, the module resolver's, the grant resolver's) still must not block the
    // thread that turns a host's event loop.
    if (async::in_host_turn())
        refuse_blocking_store_call(operation,
                                   "it is called inside a turn of a host's event loop, which a blocked thread cannot turn");
}
 
inline void run_blocking_wait(BlockingIoContext& wait) {
    try {
        wait.get().run();
    } catch (...) {
        wait.discard();
        throw;
    }
}
 
template <typename R>
R block_on_store(asio::awaitable<R> call, const BlockingStorePolicy& policy, std::string_view operation) {
    check_blocking_store_call(policy, operation);
    BlockingIoContext  wait;
    std::optional<R>   value;
    std::exception_ptr failure;
    bool               finished = false;
    asio::co_spawn(wait.get(), box_store_call<R>(std::move(call)),
                   [&](std::exception_ptr error, std::optional<R> result) {
                       failure  = std::move(error);
                       value    = std::move(result);
                       finished = true;
                   });
    run_blocking_wait(wait);
    // run() returns only when the call is done (the context stays busy for as long as the coroutine exists), so this is
    // a defence: an unfinished call is never reported as a result.
    if (!finished) {
        wait.discard();
        refuse_blocking_store_call(operation, "the store call did not finish on this thread");
    }
    if (failure) std::rethrow_exception(failure);
    return std::move(*value);
}
 
inline void block_on_store(asio::awaitable<void> call, const BlockingStorePolicy& policy, std::string_view operation) {
    check_blocking_store_call(policy, operation);
    BlockingIoContext  wait;
    std::exception_ptr failure;
    bool               finished = false;
    asio::co_spawn(wait.get(), std::move(call), [&](std::exception_ptr error) {
        failure  = std::move(error);
        finished = true;
    });
    run_blocking_wait(wait);
    // run() returns only when the call is done (the context stays busy for as long as the coroutine exists), so this is
    // a defence: an unfinished call is never reported as a result.
    if (!finished) {
        wait.discard();
        refuse_blocking_store_call(operation, "the store call did not finish on this thread");
    }
    if (failure) std::rethrow_exception(failure);
}
 
template <typename R>
R run_to_completion(asio::awaitable<R> call, std::string_view operation) {
    InlineStoreCalls   inline_calls;
    BlockingIoContext  wait;
    std::optional<R>   value;
    std::exception_ptr failure;
    bool               finished = false;
    asio::co_spawn(wait.get(), box_store_call<R>(std::move(call)),
                   [&](std::exception_ptr error, std::optional<R> result) {
                       failure  = std::move(error);
                       value    = std::move(result);
                       finished = true;
                   });
    run_blocking_wait(wait);
    if (!finished) {
        wait.discard();
        refuse_blocking_store_call(operation, "the call did not finish on this thread");
    }
    if (failure) std::rethrow_exception(failure);
    return std::move(*value);
}
 
inline void run_to_completion(asio::awaitable<void> call, std::string_view operation) {
    InlineStoreCalls   inline_calls;
    BlockingIoContext  wait;
    std::exception_ptr failure;
    bool               finished = false;
    asio::co_spawn(wait.get(), std::move(call), [&](std::exception_ptr error) {
        failure  = std::move(error);
        finished = true;
    });
    run_blocking_wait(wait);
    if (!finished) {
        wait.discard();
        refuse_blocking_store_call(operation, "the call did not finish on this thread");
    }
    if (failure) std::rethrow_exception(failure);
}
 
}  // namespace detail
}  // namespace neograph::program