IfTcpImpl.hxx

Go to the documentation of this file.

 
#ifndef _OPENLCB_IFTCPIMPL_HXX_
#define _OPENLCB_IFTCPIMPL_HXX_
 
#include "openlcb/If.hxx"
#include "utils/Hub.hxx"
#include "utils/HubDeviceSelect.hxx"
 
namespace openlcb
{
 
class TcpDefs
{
public:
    static void render_tcp_message(const GenMessage &msg,
        NodeID gateway_node_id, long long sequence, string *tgt)
    {
        bool has_dst = Defs::get_mti_address(msg.mti);
        HASSERT(tgt);
        string &target = *tgt;
        target.assign(HDR_LEN + msg.payload.size() +
                (has_dst ? MSG_ADR_PAYLOAD_OFS : MSG_GLOBAL_PAYLOAD_OFS),
            '\0');
        uint16_t flags = FLAGS_OPENLCB_MSG;
        error_to_data(flags, &target[HDR_FLAG_OFS]);
        unsigned sz = target.size() - HDR_SIZE_END;
        target[HDR_SIZE_OFS] = (sz >> 16) & 0xff;
        target[HDR_SIZE_OFS + 1] = (sz >> 8) & 0xff;
        target[HDR_SIZE_OFS + 2] = sz & 0xff;
        node_id_to_data(gateway_node_id, &target[HDR_GATEWAY_OFS]);
        node_id_to_data(sequence, &target[HDR_TIMESTAMP_OFS]);
        error_to_data(msg.mti, &target[HDR_LEN + MSG_MTI_OFS]);
        node_id_to_data(msg.src.id, &target[HDR_LEN + MSG_SRC_OFS]);
        if (has_dst)
        {
            node_id_to_data(msg.dst.id, &target[HDR_LEN + MSG_DST_OFS]);
            memcpy(&target[HDR_LEN + MSG_ADR_PAYLOAD_OFS], msg.payload.data(),
                msg.payload.size());
        }
        else
        {
            memcpy(&target[HDR_LEN + MSG_GLOBAL_PAYLOAD_OFS],
                msg.payload.data(), msg.payload.size());
        }
    }
 
    static int get_tcp_message_len(const void *data, size_t len)
    {
        if (len < HDR_SIZE_END)
        {
            return -1;
        }
        const uint8_t *src = static_cast<const uint8_t *>(data);
        uint32_t sz = src[HDR_SIZE_OFS];
        sz <<= 8;
        sz |= src[HDR_SIZE_OFS + 1];
        sz <<= 8;
        sz |= src[HDR_SIZE_OFS + 2];
        return sz + HDR_SIZE_END;
    }
 
    static bool parse_tcp_message(const string &src, GenMessage *tgt)
    {
        int expected_size = get_tcp_message_len(src.data(), src.size());
        if (expected_size > (int)src.size() || expected_size < MIN_MESSAGE_SIZE)
        {
            LOG(WARNING, "Incomplete or incorrectly formatted TCP message.");
            return false;
        }
        uint16_t flags = data_to_error(&src[0]);
        if ((flags & FLAGS_OPENLCB_MSG) == 0)
        {
            return false;
        }
        tgt->clear();
        HASSERT((flags & FLAGS_CHAINING) == 0);
        tgt->flagsDst = 0;
        tgt->flagsSrc = 0;
        if (flags & FLAGS_FRAGMENT_NOT_FIRST)
        {
            tgt->set_flag_dst(GenMessage::DSTFLAG_NOT_FIRST_MESSAGE);
        }
        if (flags & FLAGS_FRAGMENT_NOT_LAST)
        {
            tgt->set_flag_dst(GenMessage::DSTFLAG_NOT_LAST_MESSAGE);
        }
        const char *msg = &src[HDR_LEN];
        // We have already checked the size to be long enough for a global
        // message with 0 bytes payload. So source address and MTI is ok to
        // parse now.
        tgt->mti = (Defs::MTI)data_to_error(msg + MSG_MTI_OFS);
        tgt->src.clear();
        tgt->dst.clear();
        tgt->src.id = data_to_node_id(msg + MSG_SRC_OFS);
        // now contains the length of the message after the header.
        int payload_bytes = expected_size - HDR_LEN;
        int payload_ofs;
        if (Defs::get_mti_address(tgt->mti))
        {
            payload_ofs = MSG_ADR_PAYLOAD_OFS;
            tgt->dst.id = data_to_node_id(msg + MSG_DST_OFS);
        }
        else
        {
            payload_ofs = MSG_GLOBAL_PAYLOAD_OFS;
        }
        payload_bytes -= payload_ofs;
        if (payload_bytes < 0)
        {
            LOG(WARNING, "TCP message to short.");
            return false;
        }
        tgt->payload.assign(msg + payload_ofs, payload_bytes);
        return true;
    }
 
    static unsigned guess_priority(const string &tcp_payload)
    {
        if (tcp_payload.size() < ABS_MTI_OFS + 2)
        {
            return UINT_MAX;
        }
        auto mti = (Defs::MTI)data_to_error(tcp_payload.data() + ABS_MTI_OFS);
        return Defs::mti_priority(mti);
    }
 
    enum
    {
        FLAGS_OPENLCB_MSG = 0x8000,
        FLAGS_CHAINING = 0x4000,
        FLAGS_RESVD1_ZERO_CHECK = 0x3000,
        FLAGS_FRAGMENT_NOT_FIRST = 0x0800,
        FLAGS_FRAGMENT_NOT_LAST = 0x0400,
        FLAGS_RESVD2_IGNORED = 0x03FF,
 
        HDR_FLAG_OFS = 0,
        HDR_SIZE_OFS = 2,
        HDR_GATEWAY_OFS = 2 + 3,
        HDR_SIZE_END = HDR_GATEWAY_OFS,
        HDR_TIMESTAMP_OFS = 2 + 3 + 6,
        HDR_LEN = 2 + 3 + 6 + 6,
 
        MSG_MTI_OFS = 0,
        MSG_SRC_OFS = 2,
        MSG_DST_OFS = 2 + 6,
        MSG_ADR_PAYLOAD_OFS = 2 + 6 + 6,
 
        MSG_GLOBAL_PAYLOAD_OFS = MSG_DST_OFS,
 
        MIN_MESSAGE_SIZE = MSG_GLOBAL_PAYLOAD_OFS + HDR_LEN,
        MIN_ADR_MESSAGE_SIZE = MSG_ADR_PAYLOAD_OFS + HDR_LEN,
 
        ABS_MTI_OFS = HDR_LEN + MSG_MTI_OFS,
    };
 
private:
    TcpDefs();
};
 
class SequenceNumberGenerator : public Destructable
{
public:
    virtual long long get_sequence_number() = 0;
};
 
class ClockBaseSequenceNumberGenerator : public SequenceNumberGenerator
{
public:
    long long get_sequence_number() override
    {
        long long current_time_ms = os_get_time_monotonic() / 1000000;
        if (current_time_ms > sequence_)
        {
            sequence_ = current_time_ms;
        }
        else
        {
            ++sequence_;
        }
        return sequence_;
    }
 
private:
    long long sequence_ = 0;
};
 
class TcpSendFlow : public MessageStateFlowBase
{
public:
    TcpSendFlow(If *service, NodeID gateway_node_id,
        HubPortInterface *send_target, HubPortInterface *skip_member,
        SequenceNumberGenerator *sequence)
        : MessageStateFlowBase(service)
        , sendTarget_(send_target)
        , skipMember_(skip_member)
        , gatewayId_(gateway_node_id)
        , sequenceNumberGenerator_(sequence)
    {
    }
 
    NodeID get_gateway_node_id()
    {
        return gatewayId_;
    }
 
private:
    Action entry() override
    {
        auto id = nmsg()->dst.id;
        if (id)
        {
            auto *n = iface()->lookup_local_node(id);
            if (n)
            {
                // Addressed message loopback.
                auto *b = transfer_message();
                b->data()->dstNode = n;
                iface()->dispatcher()->send(b, priority());
                return release_and_exit();
            }
        }
        return allocate_and_call(sendTarget_, STATE(render_src_message));
    }
 
    Action render_src_message()
    {
        auto b = get_buffer_deleter(get_allocation_result(sendTarget_));
        TcpDefs::render_tcp_message(*nmsg(), gatewayId_,
            sequenceNumberGenerator_->get_sequence_number(), b->data());
        b->data()->skipMember_ = skipMember_;
        sendTarget_->send(b.release(), nmsg()->priority());
 
        // Checks and performs global loopback.
        if (!nmsg()->dst.id)
        {
            iface()->dispatcher()->send(transfer_message(), priority());
        }
        return release_and_exit();
    }
 
    GenMessage *nmsg()
    {
        return message()->data();
    }
 
    If *iface()
    {
        return static_cast<If *>(service());
    }
 
    HubPortInterface *sendTarget_;
    HubPortInterface *skipMember_;
    NodeID gatewayId_;
    SequenceNumberGenerator *sequenceNumberGenerator_;
};
 
class FdToTcpParser : public StateFlowBase
{
public:
    FdToTcpParser(FdHubPortService *s, HubPortInterface *dst,
        HubPortInterface *skipMember)
        : StateFlowBase(s)
        , dst_(dst)
        , skipMember_(skipMember)
    {
        HASSERT(s->fd() >= 0);
        bufEnd_ = 0;
        bufOfs_ = 0;
        start_flow(STATE(start_msg));
    }
 
    void shutdown()
    {
        auto *e = this->service()->executor();
        if (e->is_selected(&helper_))
        {
            e->unselect(&helper_);
            helper_.remaining_ = 0;
        }
        set_terminated();
        notify_barrier();
    }
 
private:
    FdHubPortService *device()
    {
        return static_cast<FdHubPortService *>(service());
    }
 
    Action start_msg()
    {
        msg_.clear();
        expectedLen_ = -1;
        return parse_bytes();
    }
 
    Action parse_bytes()
    {
        if (bufEnd_ <= bufOfs_)
        {
            return call_immediately(STATE(read_more_bytes));
        }
        int available = bufEnd_ - bufOfs_;
        if (expectedLen_ < 0)
        {
            // We have bytes in the read buffer but don't know the expected
            // size yet, so none in the assembly buffer. Let's guess at the
            // desired size of the assembly buffer.
            expectedLen_ =
                TcpDefs::get_tcp_message_len(buffer_ + bufOfs_, available);
            if (expectedLen_ < 0)
            {
                // failed. Not enough bytes. Maybe move existing bytes to the
                // beginning of the buffer and try to read more bytes.
                if (bufOfs_ > (READ_BUFFER_SIZE / 2))
                {
                    memmove(buffer_, buffer_ + bufOfs_, available);
                    bufEnd_ -= bufOfs_;
                    bufOfs_ = 0;
                }
                return call_immediately(STATE(read_more_bytes));
            }
        }
        // now: we have an expected length.
        DASSERT(expectedLen_ > 0);
        if (msg_.empty())
        {
            msg_.reserve(expectedLen_);
        }
        // Copy some bytes to the assembly buffer.
        int needed = expectedLen_ - msg_.size();
        if (needed > available)
        {
            needed = available;
        }
        msg_.append((const char *)(buffer_ + bufOfs_), needed);
        bufOfs_ += needed;
        if (msg_.size() >= (unsigned)expectedLen_)
        {
            // we're done.
            return send_entry();
        }
        else
        {
            HASSERT(bufOfs_ >= bufEnd_);
            return call_immediately(STATE(read_more_bytes));
        }
    }
 
    Action read_more_bytes()
    {
        if (bufEnd_ <= bufOfs_)
        {
            bufOfs_ = 0;
            bufEnd_ = 0;
        }
        return read_single(&helper_, device()->fd(), buffer_ + bufEnd_,
            READ_BUFFER_SIZE - bufEnd_, STATE(read_done), READ_PRIO);
    }
 
    Action read_done()
    {
        if (helper_.hasError_)
        {
            notify_barrier();
            set_terminated();
            device()->report_read_error();
            return exit();
        }
        bufEnd_ = READ_BUFFER_SIZE - helper_.remaining_;
        return parse_bytes();
    }
 
    Action send_entry()
    {
        return allocate_and_call(dst_, STATE(have_alloc_msg));
    }
 
    Action have_alloc_msg()
    {
        auto *b = get_allocation_result(dst_);
        auto prio = TcpDefs::guess_priority(msg_);
        b->data()->assign(std::move(msg_));
        b->data()->skipMember_ = skipMember_;
        dst_->send(b, prio);
        return call_immediately(STATE(start_msg));
    }
 
    void notify_barrier()
    {
        if (barrierOwned_)
        {
            barrierOwned_ = false;
            device()->barrier_.notify();
        }
    }
 
    static constexpr unsigned READ_BUFFER_SIZE = 300;
    static constexpr unsigned DEFAULT_PACKET_SIZE =
        TcpDefs::MIN_ADR_MESSAGE_SIZE + 16;
    static constexpr unsigned MIN_SIZE_GUESS = TcpDefs::HDR_SIZE_END;
    static constexpr unsigned READ_PRIO = Selectable::MAX_PRIO;
 
    uint8_t buffer_[READ_BUFFER_SIZE];
    uint8_t barrierOwned_{1};
    uint16_t bufEnd_;
    uint16_t bufOfs_;
    int expectedLen_;
    string msg_;
    HubPortInterface *dst_;
    HubPortInterface *skipMember_;
    StateFlowSelectHelper helper_{this};
};
 
using TcpHubDeviceSelect = HubDeviceSelect<HubFlow, FdToTcpParser>;
 
class TcpRecvFlow : public HubPortInterface, public Destructable
{
public:
    TcpRecvFlow(MessageHandler *target)
        : target_(target)
    {
    }
 
    void send(Buffer<HubData> *data, unsigned prio) override
    {
        auto src = get_buffer_deleter(data);
        auto dst = get_buffer_deleter(target_->alloc());
        dst->set_done(data->new_child());
        if (TcpDefs::parse_tcp_message(*src->data(), dst->data()))
        {
            target_->send(dst.release(), prio);
        }
    }
 
private:
    MessageHandler *target_;
};
 
} // namespace openlcb
 
#endif // _OPENLCB_IFTCPIMPL_HXX_