diff --git a/.gitignore b/.gitignore index 3db6f52..05effbb 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,4 @@ MODULE.bazel.lock bazel-* /udf-runner-cpp/v2/coverage-html/ .serena +udf-runner-cpp/v2/.clwb diff --git a/udf-runner-cpp/v2/BUILD.bazel b/udf-runner-cpp/v2/BUILD.bazel index 0458242..ad5c410 100644 --- a/udf-runner-cpp/v2/BUILD.bazel +++ b/udf-runner-cpp/v2/BUILD.bazel @@ -273,6 +273,55 @@ cc_library( ], ) +cc_library( + name = "context", + srcs = [ + "context/context.cc", + "context/context_manager.cc", + "context/high_level_context.cc", + "context/background_io_thread.cc", + "context/protocol_codec.cc", + ], + hdrs = [ + "include/exasol/udf/v2/context/context_factory.hpp", + "include/exasol/udf/v2/context/high_level_context.hpp", + "include/exasol/udf/v2/context/background_io_thread.hpp", + "include/exasol/udf/v2/context/context_manager.hpp", + "include/exasol/udf/v2/context/low_level_context.hpp", + "include/exasol/udf/v2/context/protocol_codec.hpp", + ], + includes = ["include"], + copts = ["-std=c++23"], + deps = [ + ":interface", + ":event_fd_factory", + ":socket", + ], + target_compatible_with = ["@platforms//os:linux"], +) + +cc_test( + name = "high_level_context_test", + srcs = ["high_level_context_test.cc"], + copts = ["-std=c++23"], + deps = [ + ":context", + "//test/common:test_common", + "@googletest//:gtest_main", + ], +) + +cc_test( + name = "context_factory_test", + srcs = ["context_factory_test.cc"], + copts = ["-std=c++23"], + deps = [ + ":context", + "@googletest//:gtest_main", + ], + target_compatible_with = ["@platforms//os:linux"], +) + cc_test( name = "json_schema_validation_test", srcs = ["json_schema_validation_test.cc"], diff --git a/udf-runner-cpp/v2/context/background_io_thread.cc b/udf-runner-cpp/v2/context/background_io_thread.cc new file mode 100644 index 0000000..d91f05b --- /dev/null +++ b/udf-runner-cpp/v2/context/background_io_thread.cc @@ -0,0 +1,610 @@ +#include + +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace exasol::udf::v2::context +{ +namespace +{ + + constexpr std::size_t frame_prefix_size = sizeof(std::uint32_t); + constexpr std::size_t maximum_frame_size = 64U * 1024U * 1024U; + using DecodedFrame = std::shared_ptr; + + struct PendingFrame + { + DecodedFrame frame; + }; + + struct CallState + { + std::deque inbound; + bool accepted{false}; + bool closed{false}; + }; + + struct ContextState + { + mutable std::mutex mutex; + std::condition_variable changed; + std::unordered_map calls; + std::deque unaccepted_calls; + std::deque control; + interface::OperationStatus terminal{interface::OperationStatus::Ok}; + + [[nodiscard]] bool has_message() const + { + if (!control.empty() || !unaccepted_calls.empty()) + { + return true; + } + return std::any_of(calls.begin(), calls.end(), + [](const auto& entry) { return !entry.second.inbound.empty(); }); + } + }; + + [[nodiscard]] interface::OperationStatus terminal_status(const ContextState& state) + { + return state.terminal == interface::OperationStatus::Ok + ? interface::OperationStatus::PeerClosed + : state.terminal; + } + + [[nodiscard]] interface::MessageStatus message_status(const ContextState& state) + { + if (state.has_message()) + { + return interface::MessageStatus::MessageAvailable; + } + switch (state.terminal) + { + case interface::OperationStatus::Ok: + return interface::MessageStatus::NoMessage; + case interface::OperationStatus::TimedOut: + return interface::MessageStatus::TimedOut; + case interface::OperationStatus::Cancelled: + return interface::MessageStatus::Cancelled; + case interface::OperationStatus::PeerClosed: + return interface::MessageStatus::PeerClosed; + case interface::OperationStatus::ProtocolError: + case interface::OperationStatus::TransportError: + return interface::MessageStatus::ConnectionError; + } + return interface::MessageStatus::ConnectionError; + } + + [[nodiscard]] std::optional deadline( + interface::Timeout timeout) + { + return timeout ? std::optional(std::chrono::steady_clock::now() + *timeout) : std::nullopt; + } + + class Call final : public interface::Call + { + public: + Call(std::shared_ptr state, + BackgroundIoThreadInterface& backend, + const std::uint64_t stream_id) + : state_(std::move(state)), backend_(backend), stream_id_(stream_id) + { + } + + [[nodiscard]] interface::MessageStatus receive_status() const override + { + const std::lock_guard lock(state_->mutex); + const auto iterator = state_->calls.find(stream_id_); + if (iterator == state_->calls.end()) + { + return interface::MessageStatus::ConnectionError; + } + if (!iterator->second.inbound.empty()) + { + return interface::MessageStatus::MessageAvailable; + } + return state_->terminal == interface::OperationStatus::Ok + ? interface::MessageStatus::NoMessage + : message_status(*state_); + } + + [[nodiscard]] interface::Result receive( + interface::Timeout timeout) override + { + std::unique_lock lock(state_->mutex); + const auto end = deadline(timeout); + const auto ready = [this] { + const auto iterator = state_->calls.find(stream_id_); + return iterator == state_->calls.end() || !iterator->second.inbound.empty() || + state_->terminal != interface::OperationStatus::Ok; + }; + if (!ready() && end && !state_->changed.wait_until(lock, *end, ready)) + { + return interface::Result::failure( + interface::OperationStatus::TimedOut); + } + if (!ready()) + { + state_->changed.wait(lock, ready); + } + const auto iterator = state_->calls.find(stream_id_); + if (iterator == state_->calls.end()) + { + return interface::Result::failure( + interface::OperationStatus::ProtocolError); + } + if (iterator->second.inbound.empty()) + { + return interface::Result::failure( + terminal_status(*state_)); + } + auto frame = std::move(iterator->second.inbound.front().frame); + iterator->second.inbound.pop_front(); + return interface::Result::success( + interface::CallMessageView(std::move(frame))); + } + + [[nodiscard]] interface::Result send(interface::CallMessageBuilder message, + interface::Timeout timeout) override + { + return backend_.send_call(stream_id_, std::move(message), timeout); + } + + private: + std::shared_ptr state_; + BackgroundIoThreadInterface& backend_; + std::uint64_t stream_id_; + }; + + class ControlStream final : public interface::ControlStream + { + public: + ControlStream(std::shared_ptr state, BackgroundIoThreadInterface& backend) + : state_(std::move(state)), backend_(backend) + { + } + + [[nodiscard]] interface::MessageStatus receive_status() const override + { + const std::lock_guard lock(state_->mutex); + if (!state_->control.empty()) + { + return interface::MessageStatus::MessageAvailable; + } + return state_->terminal == interface::OperationStatus::Ok + ? interface::MessageStatus::NoMessage + : message_status(*state_); + } + + [[nodiscard]] interface::Result receive( + interface::Timeout timeout) override + { + std::unique_lock lock(state_->mutex); + const auto end = deadline(timeout); + const auto ready = [this] { + return !state_->control.empty() || + state_->terminal != interface::OperationStatus::Ok; + }; + if (!ready() && end && !state_->changed.wait_until(lock, *end, ready)) + { + return interface::Result::failure( + interface::OperationStatus::TimedOut); + } + if (!ready()) + { + state_->changed.wait(lock, ready); + } + if (state_->control.empty()) + { + return interface::Result::failure( + terminal_status(*state_)); + } + auto frame = std::move(state_->control.front().frame); + state_->control.pop_front(); + return interface::Result::success( + interface::ControlMessageView(std::move(frame))); + } + + [[nodiscard]] interface::Result send( + interface::ControlMessageBuilder message) override + { + return backend_.send_control(std::move(message)); + } + + private: + std::shared_ptr state_; + BackgroundIoThreadInterface& backend_; + }; + + class BackgroundIoThread final : public BackgroundIoThreadInterface + { + public: + explicit BackgroundIoThread(socket::OwnedFileDescriptor connection) + : socket_(socket::UnixSocket::adopt_native_handle(std::move(connection))), + state_(std::make_shared()), + control_stream_(state_, *this), + stop_fd_(make_linux_event_fd()) + { + const int flags = ::fcntl(socket_.native_handle(), F_GETFL, 0); + if (flags == -1 || ::fcntl(socket_.native_handle(), F_SETFL, flags | O_NONBLOCK) == -1) + { + throw std::system_error(errno, std::generic_category(), + "set context socket non-blocking"); + } + } + + ~BackgroundIoThread() override + { + request_stop(); + join(); + } + + void start() override + { + std::lock_guard lock(thread_mutex_); + if (!thread_.joinable()) + { + thread_ = std::thread([this] { run(); }); + } + } + + void request_stop() noexcept override + { + stop_requested_.store(true, std::memory_order_release); + try + { + stop_fd_->write_notification(); + } + catch (const std::system_error&) + { + } + state_->changed.notify_all(); + } + + void join() noexcept override + { + std::lock_guard lock(thread_mutex_); + if (thread_.joinable()) + { + thread_.join(); + } + } + + [[nodiscard]] interface::MessageStatus receive_status() const override + { + const std::lock_guard lock(state_->mutex); + return message_status(*state_); + } + + [[nodiscard]] bool is_new_inbound_call_available() const override + { + const std::lock_guard lock(state_->mutex); + return !state_->unaccepted_calls.empty(); + } + + [[nodiscard]] interface::MessageStatus wait_for_message(interface::Timeout timeout) override + { + std::unique_lock lock(state_->mutex); + const auto end = deadline(timeout); + const auto ready = [this] { + return state_->has_message() || state_->terminal != interface::OperationStatus::Ok; + }; + if (!ready() && end && !state_->changed.wait_until(lock, *end, ready)) + { + return interface::MessageStatus::TimedOut; + } + if (!ready()) + { + state_->changed.wait(lock, ready); + } + return message_status(*state_); + } + + [[nodiscard]] interface::Result> accept_call() override + { + std::unique_lock lock(state_->mutex); + state_->changed.wait(lock, [this] { + return !state_->unaccepted_calls.empty() || + state_->terminal != interface::OperationStatus::Ok; + }); + if (state_->unaccepted_calls.empty()) + { + return interface::Result>::failure( + terminal_status(*state_)); + } + const auto stream_id = state_->unaccepted_calls.front(); + state_->unaccepted_calls.pop_front(); + state_->calls.at(stream_id).accepted = true; + return interface::Result>::success( + std::make_unique(state_, *this, stream_id)); + } + + [[nodiscard]] interface::Result> open_call( + interface::CallMessageBuilder message) override + { + auto opening = std::move(message).take(); + if (!opening.open_call.has_value()) + { + return interface::Result>::failure( + interface::OperationStatus::ProtocolError, + interface::ErrorInfo{"missing_open_call", + "an opening call message is required"}); + } + const auto stream_id = next_stream_id_++; + auto encoded = context::protocol::encode_call(stream_id, std::move(opening)); + if (!encoded) + { + return interface::Result>::failure(encoded.status, + encoded.error); + } + enqueue(std::move(encoded.value->bytes)); + { + std::lock_guard lock(state_->mutex); + state_->calls.emplace(stream_id, CallState{}).first->second.accepted = true; + } + return interface::Result>::success( + std::make_unique(state_, *this, stream_id)); + } + + [[nodiscard]] interface::Result send_call(const std::uint64_t stream_id, + interface::CallMessageBuilder message, + interface::Timeout) override + { + auto value = std::move(message).take(); + if (value.open_call.has_value()) + { + return interface::Result::failure( + interface::OperationStatus::ProtocolError, + interface::ErrorInfo{"open_call_after_open", "call is already open"}); + } + auto encoded = context::protocol::encode_call(stream_id, std::move(value)); + if (!encoded) + { + return interface::Result::failure(encoded.status, encoded.error); + } + enqueue(std::move(encoded.value->bytes)); + return interface::Result::success(); + } + + [[nodiscard]] interface::Result send_control( + interface::ControlMessageBuilder message) override + { + auto encoded = context::protocol::encode_control(std::move(message).take()); + if (!encoded) + { + return interface::Result::failure(encoded.status, encoded.error); + } + enqueue(std::move(encoded.value->bytes)); + return interface::Result::success(); + } + + [[nodiscard]] interface::ControlStream& control_stream() override + { + return control_stream_; + } + + private: + void run() noexcept + { + try + { + while (!stop_requested_.load(std::memory_order_acquire)) + { + const bool has_outbound = has_outbound_frames(); + pollfd descriptors[] = { + {socket_.native_handle(), + static_cast(POLLIN | POLLERR | POLLHUP | + (has_outbound ? POLLOUT : 0)), + 0}, + {stop_fd_->native_handle(), POLLIN, 0}, + }; + const int result = ::poll(descriptors, 2, -1); + if (result == -1) + { + if (errno == EINTR) + { + continue; + } + fail(interface::OperationStatus::TransportError); + return; + } + if ((descriptors[1].revents & POLLIN) != 0) + { + stop_fd_->read_notification(); + if (stop_requested_.load(std::memory_order_acquire)) + { + return; + } + } + if ((descriptors[0].revents & (POLLERR | POLLHUP)) != 0) + { + fail(interface::OperationStatus::PeerClosed); + return; + } + if ((descriptors[0].revents & POLLIN) != 0) + { + read_available(); + } + if ((descriptors[0].revents & POLLOUT) != 0) + { + write_available(); + } + } + } + catch (const std::system_error&) + { + fail(interface::OperationStatus::TransportError); + } + socket_.close(); + } + + void read_available() + { + std::byte buffer[16U * 1024U]; + for (;;) + { + const auto count = socket_.read_some(std::span(buffer)); + if (count == 0) + { + fail(interface::OperationStatus::PeerClosed); + return; + } + receive_buffer_.insert(receive_buffer_.end(), buffer, buffer + count); + decode_available(); + if (count < std::size(buffer)) + { + return; + } + } + } + + void decode_available() + { + while (receive_buffer_.size() >= frame_prefix_size) + { + std::uint32_t payload_size = 0; + std::memcpy(&payload_size, receive_buffer_.data(), sizeof(payload_size)); + const std::size_t frame_size = frame_prefix_size + payload_size; + if (frame_size < frame_prefix_size || frame_size > maximum_frame_size) + { + fail(interface::OperationStatus::ProtocolError); + return; + } + if (receive_buffer_.size() < frame_size) + { + return; + } + const auto decoded = + exasol::udf::protocol::DecodedFrame::decode(std::span( + reinterpret_cast(receive_buffer_.data()), frame_size)); + if (!decoded) + { + fail(interface::OperationStatus::ProtocolError); + return; + } + dispatch(std::make_shared(*decoded)); + receive_buffer_.erase(receive_buffer_.begin(), + receive_buffer_.begin() + frame_size); + } + } + + void dispatch(DecodedFrame frame) + { + const auto stream_id = frame->frame().stream_id(); + std::lock_guard lock(state_->mutex); + if (stream_id == 0) + { + state_->control.push_back({std::move(frame)}); + } + else + { + auto& call = state_->calls[stream_id]; + const bool was_empty = call.inbound.empty(); + call.inbound.push_back({std::move(frame)}); + if (was_empty && !call.accepted) + { + state_->unaccepted_calls.push_back(stream_id); + } + } + state_->changed.notify_all(); + } + + void fail(const interface::OperationStatus status) + { + { + std::lock_guard lock(state_->mutex); + if (state_->terminal == interface::OperationStatus::Ok) + { + state_->terminal = status; + } + } + state_->changed.notify_all(); + } + + void enqueue(std::vector frame) + { + { + std::lock_guard lock(outbound_mutex_); + outbound_.push_back(std::move(frame)); + } + stop_fd_->write_notification(); + } + + [[nodiscard]] bool has_outbound_frames() const + { + const std::lock_guard lock(outbound_mutex_); + return !outbound_.empty(); + } + + void write_available() + { + std::lock_guard lock(outbound_mutex_); + while (!outbound_.empty()) + { + try + { + const auto written = socket_.write_some( + std::span(outbound_.front()).subspan(outbound_offset_)); + outbound_offset_ += written; + if (outbound_offset_ == outbound_.front().size()) + { + outbound_.pop_front(); + outbound_offset_ = 0; + } + } + catch (const std::system_error& error) + { + if (error.code() == std::errc::resource_unavailable_try_again) + { + return; + } + throw; + } + } + } + + socket::UnixSocket socket_; + std::shared_ptr state_; + ControlStream control_stream_; + std::unique_ptr stop_fd_; + std::vector receive_buffer_; + mutable std::mutex outbound_mutex_; + std::deque> outbound_; + std::size_t outbound_offset_{0}; + std::uint64_t next_stream_id_{1}; + std::atomic_bool stop_requested_{false}; + std::mutex thread_mutex_; + std::thread thread_; + }; + +} // namespace + +std::unique_ptr make_background_io_thread( + socket::OwnedFileDescriptor connection) +{ + return std::make_unique(std::move(connection)); +} + +} // namespace exasol::udf::v2::context diff --git a/udf-runner-cpp/v2/context/context.cc b/udf-runner-cpp/v2/context/context.cc new file mode 100644 index 0000000..19d3918 --- /dev/null +++ b/udf-runner-cpp/v2/context/context.cc @@ -0,0 +1,56 @@ +#include + +#include +#include + +namespace exasol::udf::v2::context +{ + +LowLevelContext::LowLevelContext(std::unique_ptr background_thread) + : background_thread_(std::move(background_thread)) +{ + if (background_thread_ == nullptr) + { + throw std::invalid_argument("background I/O thread must not be null"); + } + background_thread_->start(); +} + +LowLevelContext::~LowLevelContext() +{ + background_thread_->request_stop(); + background_thread_->join(); +} + +interface::MessageStatus LowLevelContext::receive_status() const +{ + return background_thread_->receive_status(); +} + +bool LowLevelContext::is_new_inbound_call_available() const +{ + return background_thread_->is_new_inbound_call_available(); +} + +interface::MessageStatus LowLevelContext::wait_for_message(interface::Timeout timeout) +{ + return background_thread_->wait_for_message(timeout); +} + +interface::Result> LowLevelContext::accept_call() +{ + return background_thread_->accept_call(); +} + +interface::Result> LowLevelContext::open_call( + interface::CallMessageBuilder opening_message) +{ + return background_thread_->open_call(std::move(opening_message)); +} + +interface::ControlStream& LowLevelContext::control_stream() +{ + return background_thread_->control_stream(); +} + +} // namespace exasol::udf::v2::context diff --git a/udf-runner-cpp/v2/context/context_manager.cc b/udf-runner-cpp/v2/context/context_manager.cc new file mode 100644 index 0000000..1a8c89c --- /dev/null +++ b/udf-runner-cpp/v2/context/context_manager.cc @@ -0,0 +1,63 @@ +#include +#include + +#include + +#include +#include + +namespace exasol::udf::v2::context +{ +namespace +{ + + class DefaultFactory final : public BackgroundIoThreadFactory + { + public: + [[nodiscard]] std::unique_ptr create( + socket::OwnedFileDescriptor connection) override + { + return make_background_io_thread(std::move(connection)); + } + }; + +} // namespace + +ContextManager::ContextManager() : ContextManager(std::make_unique()) +{ +} + +ContextManager::ContextManager(std::unique_ptr factory) + : factory_(std::move(factory)) +{ + if (factory_ == nullptr) + { + throw std::invalid_argument("background I/O thread factory must not be null"); + } +} + +std::unique_ptr ContextManager::create_context( + socket::OwnedFileDescriptor connection) +{ + return std::make_unique(factory_->create(std::move(connection))); +} + +} // namespace exasol::udf::v2::context + +namespace exasol::udf::v2 +{ + +std::unique_ptr create_context(socket::OwnedFileDescriptor connection) +{ + return context::ContextManager().create_context(std::move(connection)); +} + +std::unique_ptr create_context( + socket::OwnedFileDescriptor connection, + std::unique_ptr background_thread_factory) +{ + return context::ContextManager(std::move(background_thread_factory)) + .create_context(std::move(connection)); +} + +} // namespace exasol::udf::v2 diff --git a/udf-runner-cpp/v2/context/high_level_context.cc b/udf-runner-cpp/v2/context/high_level_context.cc new file mode 100644 index 0000000..44d7ada --- /dev/null +++ b/udf-runner-cpp/v2/context/high_level_context.cc @@ -0,0 +1,47 @@ +#include + +#include +#include + +namespace exasol::udf::v2::context +{ + +interface::Result HighLevelContext::initialize(const std::uint32_t workers) +{ + if (workers == 0) + { + return interface::Result::failure( + interface::OperationStatus::ProtocolError, + interface::ErrorInfo{"invalid_worker_count", + "number of supported workers must be positive"}); + } + + interface::ControlMessageBuilder message; + message.server_capabilities(2, 0, std::endian::native, workers) + .add_payload("high_level_name", protocol_name) + .add_payload("high_level_version", protocol_version); + return context_.control_stream().send(std::move(message)); +} + +interface::Result HighLevelContext::send_metadata(const std::string_view name, + const std::string_view value, + const bool binary) +{ + interface::ControlMessageBuilder message; + message.add_payload(name, value, binary); + return context_.control_stream().send(std::move(message)); +} + +interface::Result> HighLevelContext::open_call( + const std::string_view call_name, const interface::Payloads& payloads) +{ + interface::CallMessageBuilder message; + message.open_call(call_name); + for (const auto& payload : payloads) + { + message.add_payload(payload.name, payload.value, payload.binary); + } + return context_.open_call(std::move(message)); +} + +} // namespace exasol::udf::v2::context diff --git a/udf-runner-cpp/v2/context/protocol_codec.cc b/udf-runner-cpp/v2/context/protocol_codec.cc new file mode 100644 index 0000000..708952c --- /dev/null +++ b/udf-runner-cpp/v2/context/protocol_codec.cc @@ -0,0 +1,393 @@ +#include + +#include + +#include +#include +#include +#include +#include + +namespace exasol::udf::v2::context::protocol +{ + +using interface::CallMessage; +using interface::ControlMessage; +using interface::ErrorInfo; +using interface::Next; +using interface::OpenCall; +using interface::OperationStatus; +using interface::Payload; +using interface::Payloads; +using interface::Result; +using interface::ServerCapabilities; + +namespace wire = exasol::udf::protocol; +using Builder = exasol::udf::v2::third_party::flatbuffers::FlatBufferBuilder; +using Verifier = exasol::udf::v2::third_party::flatbuffers::Verifier; + +namespace +{ + + [[nodiscard]] Result failure(std::string code, std::string message) + { + return Result::failure(OperationStatus::ProtocolError, + ErrorInfo{std::move(code), std::move(message)}); + } + + [[nodiscard]] Result decode_failure(std::string message) + { + return Result::failure(OperationStatus::ProtocolError, + ErrorInfo{"invalid_frame", std::move(message)}); + } + + [[nodiscard]] bool has_unsupported_arrow(const CallMessage& message) + { + return message.data_schema.has_value() || message.record_batch.has_value(); + } + + [[nodiscard]] exasol::udf::v2::third_party::flatbuffers::Offset make_payloads( + Builder& builder, const Payloads& payloads) + { + std::vector> values; + values.reserve(payloads.size()); + for (const auto& payload : payloads) + { + const auto name = builder.CreateString(payload.name); + exasol::udf::v2::third_party::flatbuffers::Offset value; + wire::PayloadValue type; + if (payload.binary) + { + const auto bytes = builder.CreateVector( + reinterpret_cast(payload.value.data()), + payload.value.size()); + value = wire::CreateBinaryPayload(builder, bytes).Union(); + type = wire::PayloadValue_BinaryPayload; + } + else + { + value = + wire::CreateStringPayload(builder, builder.CreateString(payload.value)).Union(); + type = wire::PayloadValue_StringPayload; + } + values.emplace_back(wire::CreatePayload(builder, name, type, value)); + } + return wire::CreatePayloads(builder, builder.CreateVector(values)); + } + + [[nodiscard]] exasol::udf::v2::third_party::flatbuffers::Offset make_error( + Builder& builder, const ErrorInfo& error) + { + return wire::CreateError(builder, builder.CreateString(error.code()), + builder.CreateString(error.message())); + } + + [[nodiscard]] Result encode( + std::uint64_t stream_id, const std::variant& message) + { + Builder builder; + exasol::udf::v2::third_party::flatbuffers::Offset payloads; + exasol::udf::v2::third_party::flatbuffers::Offset next; + exasol::udf::v2::third_party::flatbuffers::Offset error; + exasol::udf::v2::third_party::flatbuffers::Offset open_call; + exasol::udf::v2::third_party::flatbuffers::Offset close; + wire::ControlMessageValue value_type = wire::ControlMessageValue_NONE; + exasol::udf::v2::third_party::flatbuffers::Offset value; + + if (const auto* control = std::get_if(&message)) + { + if (stream_id != 0) + { + return failure("control_stream_id", "control messages must use stream 0"); + } + if (control->payloads) + { + payloads = make_payloads(builder, *control->payloads); + } + if (control->error) + { + error = make_error(builder, *control->error); + } + if (control->server_capabilities) + { + const auto& capabilities = *control->server_capabilities; + const auto version = + wire::CreateVersion(builder, capabilities.major, capabilities.minor); + const auto endian = capabilities.endianness == std::endian::little + ? wire::Endianness_Little + : wire::Endianness_Big; + value_type = wire::ControlMessageValue_ServerCapabilities; + value = wire::CreateServerCapabilities(builder, version, endian, + capabilities.number_of_supported_workers) + .Union(); + } + else if (control->keep_alive) + { + value_type = wire::ControlMessageValue_KeepAlive; + value = wire::CreateKeepAlive(builder).Union(); + } + if (control->close_connection) + { + if (value_type != wire::ControlMessageValue_NONE) + { + return failure( + "multiple_wire_variants", + "close_connection cannot share a frame with another wire variant"); + } + close = wire::CreateCloseControlMessage(builder, 0, + wire::CreateCloseConnection(builder)); + value_type = wire::ControlMessageValue_CloseControlMessage; + value = close.Union(); + } + } + else + { + const auto* call = std::get_if(&message); + if (stream_id == 0) + { + return failure("call_stream_id", "call messages cannot use stream 0"); + } + if (has_unsupported_arrow(*call)) + { + return failure("arrow_transport_unimplemented", + "Arrow schema and record-batch transport is not implemented"); + } + if (call->payloads) + { + payloads = make_payloads(builder, *call->payloads); + } + if (call->outbound_next) + { + next = wire::CreateNext(builder, call->outbound_next->byte_budget, + call->outbound_next->reset, call->outbound_next->row_id); + } + if (call->error) + { + error = make_error(builder, *call->error); + } + if (call->open_call) + { + open_call = + wire::CreateOpenCall(builder, builder.CreateString(call->open_call->call_name)); + value_type = wire::ControlMessageValue_OpenCall; + value = open_call.Union(); + } + if (call->close_call) + { + if (value_type != wire::ControlMessageValue_NONE) + { + return failure("multiple_wire_variants", + "open_call and close_call cannot share a frame"); + } + close = wire::CreateCloseControlMessage(builder, wire::CreateCloseCall(builder), 0); + value_type = wire::ControlMessageValue_CloseControlMessage; + value = close.Union(); + } + } + + const auto control = + wire::CreateControlMessage(builder, payloads, next, 0, error, value_type, value); + const auto frame = wire::CreateFrame(builder, stream_id, control, 0); + wire::FinishSizePrefixedFrameBuffer(builder, frame); + const auto* data = reinterpret_cast(builder.GetBufferPointer()); + return Result::success( + EncodedFrame{std::vector(data, data + builder.GetSize())}); + } + + [[nodiscard]] Result decode_payloads(const wire::Payloads* values) + { + Payloads payloads; + if (values == nullptr || values->payloads() == nullptr) + { + return Result::success(std::move(payloads)); + } + payloads.reserve(values->payloads()->size()); + for (const auto* value : *values->payloads()) + { + if (value == nullptr || value->name() == nullptr || value->payload() == nullptr) + { + return Result::failure( + OperationStatus::ProtocolError, + ErrorInfo{"invalid_payload", "payload is incomplete"}); + } + Payload payload; + payload.name = value->name()->str(); + if (value->payload_type() == wire::PayloadValue_StringPayload) + { + const auto* string_value = value->payload_as_StringPayload(); + if (string_value == nullptr || string_value->value() == nullptr) + { + return Result::failure( + OperationStatus::ProtocolError, + ErrorInfo{"invalid_payload", "string payload is incomplete"}); + } + payload.value = string_value->value()->str(); + } + else if (value->payload_type() == wire::PayloadValue_BinaryPayload) + { + const auto* binary_value = value->payload_as_BinaryPayload(); + if (binary_value == nullptr || binary_value->value() == nullptr) + { + return Result::failure( + OperationStatus::ProtocolError, + ErrorInfo{"invalid_payload", "binary payload is incomplete"}); + } + payload.binary = true; + payload.value.assign(reinterpret_cast(binary_value->value()->Data()), + binary_value->value()->size()); + } + else + { + return Result::failure( + OperationStatus::ProtocolError, + ErrorInfo{"invalid_payload", "unknown payload type"}); + } + payloads.emplace_back(std::move(payload)); + } + return Result::success(std::move(payloads)); + } + + [[nodiscard]] Result decode_verified(const wire::Frame& frame) + { + const auto* wire_control = frame.control_message(); + if (wire_control == nullptr) + { + return decode_failure("frame has no control message"); + } + const bool is_control = frame.stream_id() == 0; + if (is_control) + { + ControlMessage message; + if (wire_control->payloads()) + { + auto payloads = decode_payloads(wire_control->payloads()); + if (!payloads) + { + return Result::failure(payloads.status, payloads.error); + } + message.payloads = std::move(*payloads.value); + } + if (wire_control->error()) + { + message.error.emplace(wire_control->error()->code()->str(), + wire_control->error()->message()->str()); + } + if (wire_control->value_type() == wire::ControlMessageValue_ServerCapabilities) + { + const auto* capabilities = wire_control->value_as_ServerCapabilities(); + if (capabilities == nullptr || capabilities->supported_version() == nullptr) + { + return decode_failure("server capabilities are incomplete"); + } + message.server_capabilities = ServerCapabilities{ + capabilities->supported_version()->major(), + capabilities->supported_version()->minor(), + capabilities->endianness() == wire::Endianness_Little ? std::endian::little + : std::endian::big, + capabilities->number_of_supported_workers()}; + } + else if (wire_control->value_type() == wire::ControlMessageValue_KeepAlive) + { + message.keep_alive = true; + } + else if (wire_control->value_type() == wire::ControlMessageValue_CloseControlMessage) + { + const auto* close = wire_control->value_as_CloseControlMessage(); + message.close_connection = close != nullptr && close->close_connection() != nullptr; + if (close != nullptr && close->close_call() != nullptr) + { + return decode_failure("close_call is not valid on stream 0"); + } + } + if (wire_control->next() != nullptr || wire_control->data_schema() != nullptr) + { + return decode_failure("call fields are not valid on stream 0"); + } + return Result::success(DecodedFrame{0, std::move(message)}); + } + + CallMessage message; + if (wire_control->payloads()) + { + auto payloads = decode_payloads(wire_control->payloads()); + if (!payloads) + { + return Result::failure(payloads.status, payloads.error); + } + message.payloads = std::move(*payloads.value); + } + if (wire_control->next()) + { + message.next = Next{wire_control->next()->reset(), wire_control->next()->row_id()}; + } + if (wire_control->error()) + { + message.error.emplace(wire_control->error()->code()->str(), + wire_control->error()->message()->str()); + } + switch (wire_control->value_type()) + { + case wire::ControlMessageValue_NONE: + break; + case wire::ControlMessageValue_OpenCall: + { + const auto* open = wire_control->value_as_OpenCall(); + if (open == nullptr || open->call_name() == nullptr) + { + return decode_failure("open_call is incomplete"); + } + message.open_call = OpenCall{open->call_name()->str()}; + break; + } + case wire::ControlMessageValue_CloseControlMessage: + { + const auto* close = wire_control->value_as_CloseControlMessage(); + if (close == nullptr || close->close_connection() != nullptr) + { + return decode_failure("close_connection is not valid on a call stream"); + } + message.close_call = close->close_call() != nullptr; + break; + } + default: + return decode_failure("control-only value on a call stream"); + } + if (wire_control->data_schema() != nullptr || frame.data_record_batch_metadata() != nullptr) + { + return decode_failure("Arrow schema and record-batch transport is not implemented"); + } + return Result::success(DecodedFrame{frame.stream_id(), std::move(message)}); + } + +} // namespace + +Result encode_control(ControlMessage message) +{ + return encode(0, std::move(message)); +} + +Result encode_call(const std::uint64_t stream_id, CallMessage message) +{ + return encode(stream_id, std::move(message)); +} + +Result decode(const std::span frame) +{ + if (frame.empty() || frame.size() > std::numeric_limits::max()) + { + return decode_failure("frame is empty or too large"); + } + Verifier verifier(reinterpret_cast(frame.data()), frame.size()); + if (!wire::VerifySizePrefixedFrameBuffer(verifier)) + { + return decode_failure("FlatBuffer verification failed"); + } + const auto* decoded = wire::GetSizePrefixedFrame(frame.data()); + if (decoded == nullptr) + { + return decode_failure("FlatBuffer root is missing"); + } + return decode_verified(*decoded); +} + +} // namespace exasol::udf::v2::context::protocol diff --git a/udf-runner-cpp/v2/context_factory_test.cc b/udf-runner-cpp/v2/context_factory_test.cc new file mode 100644 index 0000000..9733639 --- /dev/null +++ b/udf-runner-cpp/v2/context_factory_test.cc @@ -0,0 +1,229 @@ +#include +#include + +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace exasol::udf::v2 +{ +namespace +{ + + struct State + { + bool started{false}; + bool stop_requested{false}; + bool joined{false}; + bool received_connection{false}; + }; + + class ControlStreamStub final : public interface::ControlStream + { + public: + [[nodiscard]] interface::MessageStatus receive_status() const override + { + return interface::MessageStatus::NoMessage; + } + + [[nodiscard]] interface::Result receive( + interface::Timeout) override + { + return interface::Result::failure( + interface::OperationStatus::TimedOut); + } + + [[nodiscard]] interface::Result send(interface::ControlMessageBuilder) override + { + return interface::Result::success(); + } + }; + + class BackgroundThreadStub final : public context::BackgroundIoThreadInterface + { + public: + BackgroundThreadStub(std::shared_ptr state, socket::OwnedFileDescriptor connection) + : state_(std::move(state)), connection_(std::move(connection)) + { + state_->received_connection = connection_.is_open(); + } + + ~BackgroundThreadStub() override = default; + + void start() override + { + state_->started = true; + } + + void request_stop() noexcept override + { + state_->stop_requested = true; + } + + void join() noexcept override + { + state_->joined = true; + } + + [[nodiscard]] interface::MessageStatus receive_status() const override + { + return interface::MessageStatus::MessageAvailable; + } + + [[nodiscard]] bool is_new_inbound_call_available() const override + { + return true; + } + + [[nodiscard]] interface::MessageStatus wait_for_message(interface::Timeout) override + { + return interface::MessageStatus::TimedOut; + } + + [[nodiscard]] interface::Result> accept_call() override + { + return interface::Result>::failure( + interface::OperationStatus::Cancelled); + } + + [[nodiscard]] interface::Result> open_call( + interface::CallMessageBuilder) override + { + return interface::Result>::failure( + interface::OperationStatus::ProtocolError); + } + + [[nodiscard]] interface::ControlStream& control_stream() override + { + return control_stream_; + } + + [[nodiscard]] interface::Result send_call(std::uint64_t, + interface::CallMessageBuilder, + interface::Timeout) override + { + return interface::Result::success(); + } + + [[nodiscard]] interface::Result send_control( + interface::ControlMessageBuilder) override + { + return interface::Result::success(); + } + + private: + std::shared_ptr state_; + socket::OwnedFileDescriptor connection_; + ControlStreamStub control_stream_; + }; + + class BackgroundThreadFactoryStub final : public context::BackgroundIoThreadFactory + { + public: + explicit BackgroundThreadFactoryStub(std::shared_ptr state) + : state_(std::move(state)) + { + } + + [[nodiscard]] std::unique_ptr create( + socket::OwnedFileDescriptor connection) override + { + return std::make_unique(state_, std::move(connection)); + } + + private: + std::shared_ptr state_; + }; + + socket::OwnedFileDescriptor make_test_connection() + { + return socket::OwnedFileDescriptor::adopt_native_handle(::dup(STDIN_FILENO)); + } + + std::pair make_socket_pair() + { + int descriptors[2]{}; + if (::socketpair(AF_UNIX, SOCK_STREAM, 0, descriptors) == -1) + { + throw std::system_error(errno, std::generic_category(), "create test socket pair"); + } + return {socket::OwnedFileDescriptor::adopt_native_handle(descriptors[0]), + socket::OwnedFileDescriptor::adopt_native_handle(descriptors[1])}; + } + +} // namespace + +TEST(ContextFactoryTest, InjectsFactoryAndOwnsBackgroundThreadLifetime) +{ + const auto state = std::make_shared(); + auto context = create_context(make_test_connection(), + std::make_unique(state)); + + EXPECT_TRUE(state->received_connection); + EXPECT_TRUE(state->started); + EXPECT_EQ(context->receive_status(), interface::MessageStatus::MessageAvailable); + EXPECT_TRUE(context->is_new_inbound_call_available()); + EXPECT_EQ(context->wait_for_message(), interface::MessageStatus::TimedOut); + EXPECT_EQ(context->accept_call().status, interface::OperationStatus::Cancelled); + EXPECT_EQ(context->open_call(interface::CallMessageBuilder{}).status, + interface::OperationStatus::ProtocolError); + EXPECT_NE(&context->control_stream(), nullptr); + + context.reset(); + + EXPECT_TRUE(state->stop_requested); + EXPECT_TRUE(state->joined); +} + +TEST(ContextFactoryTest, RejectsNullFactory) +{ + EXPECT_THROW(static_cast(create_context(make_test_connection(), nullptr)), + std::invalid_argument); +} + +TEST(ContextManagerTest, CreatesLowLevelContextFromOwnedConnection) +{ + const auto state = std::make_shared(); + context::ContextManager manager(std::make_unique(state)); + auto context = manager.create_context(make_test_connection()); + + EXPECT_TRUE(state->received_connection); + EXPECT_TRUE(state->started); + context.reset(); + EXPECT_TRUE(state->stop_requested); + EXPECT_TRUE(state->joined); +} + +TEST(ContextFactoryTest, BackgroundThreadReportsPeerClose) +{ + auto [connection, peer] = make_socket_pair(); + auto context = create_context(std::move(connection)); + peer.close(); + + EXPECT_EQ(context->wait_for_message(std::chrono::seconds(1)), + interface::MessageStatus::PeerClosed); +} + +TEST(ContextFactoryTest, BackgroundThreadRejectsOversizedFrame) +{ + auto [connection, peer] = make_socket_pair(); + auto context = create_context(std::move(connection)); + constexpr std::uint32_t oversized_payload = 64U * 1024U * 1024U + 1U; + ASSERT_EQ(::send(peer.native_handle(), &oversized_payload, sizeof(oversized_payload), 0), + sizeof(oversized_payload)); + + EXPECT_EQ(context->wait_for_message(std::chrono::seconds(1)), + interface::MessageStatus::ConnectionError); +} + +} // namespace exasol::udf::v2 diff --git a/udf-runner-cpp/v2/high_level_context_test.cc b/udf-runner-cpp/v2/high_level_context_test.cc new file mode 100644 index 0000000..5599855 --- /dev/null +++ b/udf-runner-cpp/v2/high_level_context_test.cc @@ -0,0 +1,54 @@ +#include + +#include "test/common/test_doubles.hpp" + +#include + +namespace exasol::udf::v2::context +{ + +TEST(HighLevelContextTest, InitializesProtocolCapabilities) +{ + test::ContextDouble context; + HighLevelContext high_level(context); + + ASSERT_TRUE(high_level.initialize(4)); + const auto& message = context.last_control_message(); + ASSERT_TRUE(message.has_value()); + ASSERT_TRUE(message->server_capabilities.has_value()); + EXPECT_EQ(message->server_capabilities->major, 2U); + EXPECT_EQ(message->server_capabilities->number_of_supported_workers, 4U); + ASSERT_TRUE(message->payloads.has_value()); + ASSERT_EQ(message->payloads->size(), 2U); + EXPECT_EQ(message->payloads->at(0).name, "high_level_name"); + EXPECT_EQ(message->payloads->at(0).value, protocol_name); + EXPECT_EQ(message->payloads->at(1).name, "high_level_version"); + EXPECT_EQ(message->payloads->at(1).value, protocol_version); +} + +TEST(HighLevelContextTest, OpensNamedCallWithPayloads) +{ + test::ContextDouble context; + HighLevelContext high_level(context); + + const interface::Payloads payloads{{"script_name", "example", false}}; + ASSERT_TRUE(high_level.open_call("run", payloads)); + ASSERT_TRUE(context.last_opening.has_value()); + ASSERT_TRUE(context.last_opening->open_call.has_value()); + EXPECT_EQ(context.last_opening->open_call->call_name, "run"); + ASSERT_TRUE(context.last_opening->payloads.has_value()); + ASSERT_EQ(context.last_opening->payloads->size(), 1U); + EXPECT_EQ(context.last_opening->payloads->front().name, "script_name"); +} + +TEST(HighLevelContextTest, RejectsZeroWorkers) +{ + test::ContextDouble context; + HighLevelContext high_level(context); + + const auto result = high_level.initialize(0); + EXPECT_EQ(result.status, interface::OperationStatus::ProtocolError); + EXPECT_EQ(result.error->code(), "invalid_worker_count"); +} + +} // namespace exasol::udf::v2::context diff --git a/udf-runner-cpp/v2/include/exasol/udf/v2/context/background_io_thread.hpp b/udf-runner-cpp/v2/include/exasol/udf/v2/context/background_io_thread.hpp new file mode 100644 index 0000000..556091e --- /dev/null +++ b/udf-runner-cpp/v2/include/exasol/udf/v2/context/background_io_thread.hpp @@ -0,0 +1,69 @@ +#pragma once + +#include +#include + +#include +#include + +namespace exasol::udf::v2::context +{ + +/** + * Abstracts the background I/O owner used by a worker-facing context. + * + * The worker-facing context remains responsible for the public lifetime and + * operation contract. This interface is the construction seam for replacing + * the production thread with a deterministic test backend. + */ +class BackgroundIoThreadInterface +{ +public: + virtual ~BackgroundIoThreadInterface() = default; + + BackgroundIoThreadInterface(const BackgroundIoThreadInterface&) = delete; + BackgroundIoThreadInterface& operator=(const BackgroundIoThreadInterface&) = delete; + BackgroundIoThreadInterface(BackgroundIoThreadInterface&&) = delete; + BackgroundIoThreadInterface& operator=(BackgroundIoThreadInterface&&) = delete; + + virtual void start() = 0; + virtual void request_stop() noexcept = 0; + virtual void join() noexcept = 0; + + [[nodiscard]] virtual interface::MessageStatus receive_status() const = 0; + [[nodiscard]] virtual bool is_new_inbound_call_available() const = 0; + [[nodiscard]] virtual interface::MessageStatus wait_for_message(interface::Timeout timeout) = 0; + [[nodiscard]] virtual interface::Result> accept_call() = 0; + [[nodiscard]] virtual interface::Result> open_call( + interface::CallMessageBuilder opening_message) = 0; + [[nodiscard]] virtual interface::Result send_call(std::uint64_t stream_id, + interface::CallMessageBuilder message, + interface::Timeout timeout) = 0; + [[nodiscard]] virtual interface::Result send_control( + interface::ControlMessageBuilder message) = 0; + [[nodiscard]] virtual interface::ControlStream& control_stream() = 0; + +protected: + BackgroundIoThreadInterface() = default; +}; + +/** Creates one background I/O owner for an accepted connection. */ +class BackgroundIoThreadFactory +{ +public: + virtual ~BackgroundIoThreadFactory() = default; + + BackgroundIoThreadFactory() = default; + BackgroundIoThreadFactory(const BackgroundIoThreadFactory&) = delete; + BackgroundIoThreadFactory& operator=(const BackgroundIoThreadFactory&) = delete; + BackgroundIoThreadFactory(BackgroundIoThreadFactory&&) = delete; + BackgroundIoThreadFactory& operator=(BackgroundIoThreadFactory&&) = delete; + + [[nodiscard]] virtual std::unique_ptr create( + socket::OwnedFileDescriptor connection) = 0; +}; + +[[nodiscard]] std::unique_ptr make_background_io_thread( + socket::OwnedFileDescriptor connection); + +} // namespace exasol::udf::v2::context diff --git a/udf-runner-cpp/v2/include/exasol/udf/v2/context/context_factory.hpp b/udf-runner-cpp/v2/include/exasol/udf/v2/context/context_factory.hpp new file mode 100644 index 0000000..f926568 --- /dev/null +++ b/udf-runner-cpp/v2/include/exasol/udf/v2/context/context_factory.hpp @@ -0,0 +1,22 @@ +#pragma once + +#include +#include +#include +#include + +#include + +namespace exasol::udf::v2 +{ + +/** Creates a context backed by the production background I/O thread. */ +[[nodiscard]] std::unique_ptr create_context( + socket::OwnedFileDescriptor connection); + +/** Creates a context with an injected background-thread factory. */ +[[nodiscard]] std::unique_ptr create_context( + socket::OwnedFileDescriptor connection, + std::unique_ptr background_thread_factory); + +} // namespace exasol::udf::v2 diff --git a/udf-runner-cpp/v2/include/exasol/udf/v2/context/context_manager.hpp b/udf-runner-cpp/v2/include/exasol/udf/v2/context/context_manager.hpp new file mode 100644 index 0000000..49bda93 --- /dev/null +++ b/udf-runner-cpp/v2/include/exasol/udf/v2/context/context_manager.hpp @@ -0,0 +1,32 @@ +#pragma once + +#include +#include +#include + +#include + +namespace exasol::udf::v2::context +{ + +/** Constructs one low-level context for each owned connected connection. */ +class ContextManager final +{ +public: + ContextManager(); + explicit ContextManager(std::unique_ptr factory); + + ContextManager(const ContextManager&) = delete; + ContextManager& operator=(const ContextManager&) = delete; + ContextManager(ContextManager&&) = delete; + ContextManager& operator=(ContextManager&&) = delete; + + /** Transfers @p connection ownership into the new context. */ + [[nodiscard]] std::unique_ptr create_context( + socket::OwnedFileDescriptor connection); + +private: + std::unique_ptr factory_; +}; + +} // namespace exasol::udf::v2::context diff --git a/udf-runner-cpp/v2/include/exasol/udf/v2/context/high_level_context.hpp b/udf-runner-cpp/v2/include/exasol/udf/v2/context/high_level_context.hpp new file mode 100644 index 0000000..fad0ffc --- /dev/null +++ b/udf-runner-cpp/v2/include/exasol/udf/v2/context/high_level_context.hpp @@ -0,0 +1,39 @@ +#pragma once + +#include + +#include +#include +#include + +namespace exasol::udf::v2::context +{ + +inline constexpr std::string_view protocol_name = "exasol.udf"; +inline constexpr std::string_view protocol_version = "2.0-dev"; + +/** High-level protocol facade built on a worker-facing low-level context. */ +class HighLevelContext final +{ +public: + explicit HighLevelContext(interface::Context& context) noexcept : context_(context) + { + } + + /** Advertises the high-level protocol and low-level capabilities. */ + [[nodiscard]] interface::Result initialize(std::uint32_t number_of_supported_workers); + + /** Sends or replaces one connection-scoped metadata payload. */ + [[nodiscard]] interface::Result send_metadata(std::string_view name, + std::string_view value, + bool binary = false); + + /** Opens a named high-level call with its opening payloads. */ + [[nodiscard]] interface::Result> open_call( + std::string_view call_name, const interface::Payloads& payloads = {}); + +private: + interface::Context& context_; +}; + +} // namespace exasol::udf::v2::context diff --git a/udf-runner-cpp/v2/include/exasol/udf/v2/context/low_level_context.hpp b/udf-runner-cpp/v2/include/exasol/udf/v2/context/low_level_context.hpp new file mode 100644 index 0000000..d46741c --- /dev/null +++ b/udf-runner-cpp/v2/include/exasol/udf/v2/context/low_level_context.hpp @@ -0,0 +1,35 @@ +#pragma once + +#include +#include + +#include + +namespace exasol::udf::v2::context +{ + +/** Concrete worker-facing context for one low-level protocol connection. */ +class LowLevelContext final : public interface::Context +{ +public: + explicit LowLevelContext(std::unique_ptr background_thread); + ~LowLevelContext() override; + + LowLevelContext(const LowLevelContext&) = delete; + LowLevelContext& operator=(const LowLevelContext&) = delete; + LowLevelContext(LowLevelContext&&) = delete; + LowLevelContext& operator=(LowLevelContext&&) = delete; + + [[nodiscard]] interface::MessageStatus receive_status() const override; + [[nodiscard]] bool is_new_inbound_call_available() const override; + [[nodiscard]] interface::MessageStatus wait_for_message(interface::Timeout timeout) override; + [[nodiscard]] interface::Result> accept_call() override; + [[nodiscard]] interface::Result> open_call( + interface::CallMessageBuilder opening_message) override; + [[nodiscard]] interface::ControlStream& control_stream() override; + +private: + std::unique_ptr background_thread_; +}; + +} // namespace exasol::udf::v2::context diff --git a/udf-runner-cpp/v2/include/exasol/udf/v2/context/protocol_codec.hpp b/udf-runner-cpp/v2/include/exasol/udf/v2/context/protocol_codec.hpp new file mode 100644 index 0000000..a10b50a --- /dev/null +++ b/udf-runner-cpp/v2/include/exasol/udf/v2/context/protocol_codec.hpp @@ -0,0 +1,31 @@ +#pragma once + +#include +#include + +#include +#include +#include +#include +#include + +namespace exasol::udf::v2::context::protocol +{ + +struct EncodedFrame +{ + std::vector bytes; +}; + +struct DecodedFrame +{ + std::uint64_t stream_id{}; + std::variant message; +}; + +[[nodiscard]] interface::Result encode_control(interface::ControlMessage message); +[[nodiscard]] interface::Result encode_call(std::uint64_t stream_id, + interface::CallMessage message); +[[nodiscard]] interface::Result decode(std::span frame); + +} // namespace exasol::udf::v2::context::protocol diff --git a/udf-runner-cpp/v2/include/exasol/udf/v2/interface/content_types.hpp b/udf-runner-cpp/v2/include/exasol/udf/v2/interface/content_types.hpp index cdb09ab..b3fddd4 100644 --- a/udf-runner-cpp/v2/include/exasol/udf/v2/interface/content_types.hpp +++ b/udf-runner-cpp/v2/include/exasol/udf/v2/interface/content_types.hpp @@ -4,10 +4,12 @@ #include +#include #include #include #include #include +#include namespace exasol::udf::v2::interface { @@ -28,6 +30,27 @@ struct ArrowCDataDeleter using ArrowSchemaPtr = std::unique_ptr>; using ArrowArrayPtr = std::unique_ptr>; +/// One named high-level protocol payload. +struct Payload +{ + std::string name; // NOLINT(misc-non-private-member-variables-in-classes) + std::string value; // NOLINT(misc-non-private-member-variables-in-classes) + bool binary{false}; // NOLINT(misc-non-private-member-variables-in-classes) +}; + +using Payloads = std::vector; + +/// Capabilities advertised during low-level connection initialization. +struct ServerCapabilities +{ + std::uint32_t major{}; // NOLINT(misc-non-private-member-variables-in-classes) + std::uint32_t minor{}; // NOLINT(misc-non-private-member-variables-in-classes) + std::endian endianness{ + std::endian::native}; // NOLINT(misc-non-private-member-variables-in-classes) + std::uint32_t + number_of_supported_workers{}; // NOLINT(misc-non-private-member-variables-in-classes) +}; + /// Metadata identifying a newly opened call. struct OpenCall { @@ -75,6 +98,7 @@ struct CallMessage ~CallMessage() = default; std::optional open_call; // NOLINT(misc-non-private-member-variables-in-classes) + std::optional payloads; // NOLINT(misc-non-private-member-variables-in-classes) std::optional data_schema; // NOLINT(misc-non-private-member-variables-in-classes) std::optional next; // NOLINT(misc-non-private-member-variables-in-classes) std::optional @@ -94,9 +118,12 @@ struct ControlMessage ControlMessage& operator=(ControlMessage&&) noexcept = delete; ~ControlMessage() = default; - bool keep_alive{false}; // NOLINT(misc-non-private-member-variables-in-classes) - std::optional error; // NOLINT(misc-non-private-member-variables-in-classes) - bool close_connection{false}; // NOLINT(misc-non-private-member-variables-in-classes) + std::optional + server_capabilities; // NOLINT(misc-non-private-member-variables-in-classes) + bool keep_alive{false}; // NOLINT(misc-non-private-member-variables-in-classes) + std::optional payloads; // NOLINT(misc-non-private-member-variables-in-classes) + std::optional error; // NOLINT(misc-non-private-member-variables-in-classes) + bool close_connection{false}; // NOLINT(misc-non-private-member-variables-in-classes) }; } // namespace exasol::udf::v2::interface