transport.h

Go to the documentation of this file.

// Copyright (c) 2026 Tinverse LLC. All rights reserved.
// SPDX-License-Identifier: LicenseRef-Tinverse-Commercial
 
 
#pragma once
 
#include <array>
#include <cstddef>
#include <cstdint>
#include <utility>
 
#include "tsm/runtime/transport.h"
#include "tsm/ticks.h"
 
namespace tsm::transport {
 
namespace detail {
 
template<typename Serializer>
concept serializer_has_schema_version = requires
{
    Serializer::schema_version;
};
 
template<typename Serializer>
concept serializer_has_event_id = requires
{
    Serializer::event_id;
};
 
} // namespace detail
 
enum class status : std::uint8_t
{
    accepted,
    empty,
    full,
    malformed,
    rejected,
    serialization_failed,
    deserialization_failed,
    stale_slot,
    version_mismatch
};
 
template<std::size_t MaxBytes>
struct payload_buffer
{
    std::array<std::byte, MaxBytes> bytes{};
    std::size_t size{};
 
    [[nodiscard]] bool assign(std::byte const* data, std::size_t length)
    {
        if (length > MaxBytes) {
            return false;
        }
        for (std::size_t i = 0; i < length; ++i) {
            bytes[i] = data[i];
        }
        size = length;
        return true;
    }
 
    void clear()
    {
        size = 0U;
    }
};
 
struct serializer_header
{
    std::uint32_t schema_version{ 1U };
    std::uint16_t event_id{};
    std::uint16_t flags{};
};
 
template<std::size_t MaxBytes>
struct serialized_event
{
    serializer_header header{};
    std::uint32_t sequence{};
    payload_buffer<MaxBytes> payload{};
};
 
template<typename Serializer>
struct serializer_adapter
{
    template<std::size_t MaxBytes>
    [[nodiscard]] static status encode(
      typename Serializer::event_type const& event,
      serialized_event<MaxBytes>& record)
    {
        return Serializer::encode(event, record) ? status::accepted
                                                 : status::serialization_failed;
    }
 
    template<std::size_t MaxBytes>
    [[nodiscard]] static status decode(serialized_event<MaxBytes> const& record,
                                       typename Serializer::event_type& event)
    {
        if constexpr (detail::serializer_has_schema_version<Serializer>) {
            if (record.header.schema_version != Serializer::schema_version) {
                return status::version_mismatch;
            }
        }
        if constexpr (detail::serializer_has_event_id<Serializer>) {
            if (record.header.event_id != Serializer::event_id) {
                return status::malformed;
            }
        }
        return Serializer::decode(record, event)
                 ? status::accepted
                 : status::deserialization_failed;
    }
};
 
template<std::size_t MaxBytes>
struct m2m_frame
{
    std::uint32_t source_id{};
    std::uint32_t destination_id{};
    std::uint32_t sequence{};
    std::uint16_t kind{};
    bool crc_ok{ true };
    payload_buffer<MaxBytes> payload{};
};
 
template<std::size_t MaxBytes>
struct dds_sample
{
    std::uint32_t topic_id{};
    std::uint32_t writer_id{};
    std::uint32_t sequence{};
    bool valid{ true };
    payload_buffer<MaxBytes> payload{};
};
 
template<std::size_t MaxBytes>
struct rpc_request
{
    std::uint32_t client_id{};
    std::uint32_t correlation_id{};
    std::uint16_t method{};
    payload_buffer<MaxBytes> payload{};
};
 
struct rpc_reply
{
    std::uint32_t correlation_id{};
    status result{ status::rejected };
    bool event_handled{};
};
 
template<std::size_t MaxBytes>
struct replay_record
{
    tsm::tick_duration tick{};
    std::uint16_t event_id{};
    payload_buffer<MaxBytes> payload{};
};
 
template<std::size_t MaxBytes>
struct event_io_message
{
    std::uint32_t channel_id{};
    std::uint32_t source_id{};
    std::uint32_t sequence{};
    std::uint16_t event_id{};
    bool valid{ true };
    payload_buffer<MaxBytes> payload{};
};
 
template<typename Codec>
struct m2m_adapter
{
    template<typename Runtime, std::size_t MaxBytes>
    [[nodiscard]] static status receive(Runtime& runtime,
                                        m2m_frame<MaxBytes> const& frame)
    {
        if (!frame.crc_ok) {
            return status::malformed;
        }
 
        typename Codec::event_type event{};
        if (!Codec::decode(frame, event)) {
            return status::malformed;
        }
        return tsm::runtime::transport::local_transport::send(runtime,
                                                              std::move(event))
                 ? status::accepted
                 : status::rejected;
    }
};
 
template<typename Codec>
struct dds_adapter
{
    template<typename Runtime, std::size_t MaxBytes>
    [[nodiscard]] static status receive(Runtime& runtime,
                                        dds_sample<MaxBytes> const& sample)
    {
        if (!sample.valid) {
            return status::malformed;
        }
 
        typename Codec::event_type event{};
        if (!Codec::decode(sample, event)) {
            return status::malformed;
        }
        return tsm::runtime::transport::local_transport::send(runtime,
                                                              std::move(event))
                 ? status::accepted
                 : status::rejected;
    }
};
 
template<typename Codec>
struct rpc_adapter
{
    template<typename Runtime, std::size_t MaxBytes>
    [[nodiscard]] static rpc_reply request(Runtime& runtime,
                                           rpc_request<MaxBytes> const& request)
    {
        typename Codec::event_type event{};
        if (!Codec::decode(request, event)) {
            return { request.correlation_id, status::malformed, false };
        }
 
        const bool handled = tsm::runtime::transport::local_transport::send(
          runtime, std::move(event));
        return { request.correlation_id,
                 handled ? status::accepted : status::rejected,
                 handled };
    }
};
 
template<typename Codec>
struct replay_adapter
{
    template<typename Runtime, std::size_t MaxBytes>
    [[nodiscard]] static status apply(Runtime& runtime,
                                      replay_record<MaxBytes> const& record)
    {
        typename Codec::event_type event{};
        if (!Codec::decode(record, event)) {
            return status::malformed;
        }
        return tsm::runtime::transport::local_transport::send(runtime,
                                                              std::move(event))
                 ? status::accepted
                 : status::rejected;
    }
};
 
template<typename Codec>
struct event_io_adapter
{
    template<typename Runtime, std::size_t MaxBytes>
    [[nodiscard]] static status receive(
      Runtime& runtime,
      event_io_message<MaxBytes> const& message)
    {
        if (!message.valid) {
            return status::malformed;
        }
 
        typename Codec::event_type event{};
        if (!Codec::decode(message, event)) {
            return status::malformed;
        }
        return tsm::runtime::transport::local_transport::send(runtime,
                                                              std::move(event))
                 ? status::accepted
                 : status::rejected;
    }
};
 
template<typename Slot, std::size_t Capacity>
class shared_memory_channel
{
    static_assert(Capacity > 1U,
                  "tsm: shared_memory_channel requires at least two slots");
 
  public:
    struct writable_slot
    {
        Slot* slot{};
        std::uint32_t sequence{};
    };
 
    struct readable_slot
    {
        Slot const* slot{};
        std::uint32_t sequence{};
    };
 
    [[nodiscard]] status try_acquire_write(writable_slot& out)
    {
        if (full()) {
            return status::full;
        }
        auto& slot = slots_[write_index_];
        out = { &slot.value, slot.sequence };
        return status::accepted;
    }
 
    [[nodiscard]] bool acquire_write(writable_slot& out)
    {
        return try_acquire_write(out) == status::accepted;
    }
 
    [[nodiscard]] status try_commit_write(writable_slot const& written)
    {
        auto& slot = slots_[write_index_];
        if (written.slot != &slot.value || slot.ready ||
            written.sequence != slot.sequence) {
            return status::stale_slot;
        }
        slot.sequence = written.sequence + 1U;
        slot.ready = true;
        write_index_ = next(write_index_);
        ++count_;
        return status::accepted;
    }
 
    [[nodiscard]] bool commit_write(writable_slot const& written)
    {
        return try_commit_write(written) == status::accepted;
    }
 
    [[nodiscard]] status try_acquire_read(readable_slot& out) const
    {
        if (empty()) {
            return status::empty;
        }
        auto const& slot = slots_[read_index_];
        out = { &slot.value, slot.sequence };
        return status::accepted;
    }
 
    [[nodiscard]] bool acquire_read(readable_slot& out) const
    {
        return try_acquire_read(out) == status::accepted;
    }
 
    [[nodiscard]] status try_release_read(readable_slot const& read)
    {
        auto& slot = slots_[read_index_];
        if (read.slot != &slot.value || !slot.ready ||
            read.sequence != slot.sequence) {
            return status::stale_slot;
        }
        slot.ready = false;
        read_index_ = next(read_index_);
        --count_;
        return status::accepted;
    }
 
    [[nodiscard]] bool release_read(readable_slot const& read)
    {
        return try_release_read(read) == status::accepted;
    }
 
    [[nodiscard]] bool empty() const
    {
        return count_ == 0U;
    }
    [[nodiscard]] bool full() const
    {
        return count_ == Capacity;
    }
    [[nodiscard]] std::size_t size() const
    {
        return count_;
    }
 
  private:
    struct storage_slot
    {
        Slot value{};
        std::uint32_t sequence{};
        bool ready{};
    };
 
    static constexpr std::size_t next(std::size_t index)
    {
        return (index + 1U) % Capacity;
    }
 
    std::array<storage_slot, Capacity> slots_{};
    std::size_t write_index_{};
    std::size_t read_index_{};
    std::size_t count_{};
};
 
template<typename Serializer>
struct shared_memory_adapter
{
    template<typename Runtime, std::size_t MaxBytes, std::size_t Capacity>
    [[nodiscard]] static status receive_one(
      Runtime& runtime,
      shared_memory_channel<serialized_event<MaxBytes>, Capacity>& channel)
    {
        typename shared_memory_channel<serialized_event<MaxBytes>,
                                       Capacity>::readable_slot readable{};
        const status acquired = channel.try_acquire_read(readable);
        if (acquired != status::accepted) {
            return acquired;
        }
 
        typename Serializer::event_type event{};
        const status decoded =
          serializer_adapter<Serializer>::decode(*readable.slot, event);
        if (decoded != status::accepted) {
            static_cast<void>(channel.release_read(readable));
            return decoded;
        }
 
        const bool handled = tsm::runtime::transport::local_transport::send(
          runtime, std::move(event));
        static_cast<void>(channel.release_read(readable));
        return handled ? status::accepted : status::rejected;
    }
};
 
} // namespace tsm::transport