StreamSender.hxx

Go to the documentation of this file.

 
#ifndef _OPENLCB_STREAMSENDER_HXX_
#define _OPENLCB_STREAMSENDER_HXX_
 
#include "executor/StateFlow.hxx"
#include "openlcb/CanDefs.hxx"
#include "openlcb/DatagramDefs.hxx"
#include "openlcb/IfCan.hxx"
#include "openlcb/StreamDefs.hxx"
#include "utils/ByteBuffer.hxx"
#include "utils/LimitedPool.hxx"
#include "utils/format_utils.hxx"
 
namespace openlcb
{
 
class StreamSender : public StateFlow<ByteBuffer, QList<1>>
{
public:
    StreamSender(Service *s)
        : StateFlow<ByteBuffer, QList<1>>(s)
    {
    }
 
    enum StreamSenderState : uint8_t
    {
        IDLE,
        STARTED,
        INITIATING,
        RUNNING,
        FULL,
        CLOSING,
        STATE_ERROR
    };
};
 
class StreamSenderCan : public StreamSender
{
public:
    StreamSenderCan(Service *service, IfCan *iface)
        : StreamSender(service)
        , ifCan_(iface)
        , sleeping_(false)
        , requestClose_(false)
        , requestInit_(false)
    {
    }
 
    StreamSenderCan &start_stream(Node *src, NodeHandle dst,
        uint8_t source_stream_id,
        uint8_t dst_stream_id = StreamDefs::INVALID_STREAM_ID)
    {
        DASSERT(state_ == IDLE);
        state_ = STARTED;
        node_ = src;
        dst_ = dst;
        totalByteCount_ = 0;
        localStreamId_ = source_stream_id;
        dstStreamId_ = dst_stream_id;
        HASSERT(sleeping_ == false);
        HASSERT(requestClose_ == 0);
        requestInit_ = true;
        trigger();
        streamFlags_ = 0;
        streamAdditionalFlags_ = 0;
        streamWindowSize_ = StreamDefs::MAX_PAYLOAD;
        streamWindowRemaining_ = 0;
        errorCode_ = 0;
        return *this;
    }
 
    void close_stream(uint16_t error_code = 0)
    {
        requestClose_ = true;
        trigger();
    }
 
    StreamSenderCan &set_proposed_window_size(uint16_t window_size)
    {
        HASSERT(state_ == STARTED);
        streamWindowSize_ = window_size;
        return *this;
    }
 
    StreamSenderCan &set_stream_uid(NodeID stream_uid)
    {
        HASSERT(state_ == STARTED);
        return *this;
    }
 
    void clear()
    {
        if (state_ == STATE_ERROR || state_ == CLOSING)
        {
            state_ = IDLE;
        }
    }
 
#ifdef GTEST
    bool shutdown()
    {
        if (sleeping_)
        {
            timer_.trigger();
            sleeping_ = false;
            return true;
        }
        return false;
    }
#endif
 
    StreamSenderState get_state()
    {
        return state_;
    }
 
    uint16_t get_error()
    {
        return errorCode_;
    }
 
    uint8_t get_src_stream_id()
    {
        return localStreamId_;
    }
 
    uint8_t get_dst_stream_id()
    {
        return dstStreamId_;
    }
 
    Action entry() override
    {
        if (requestInit_)
        {
            requestInit_ = 0;
            return call_immediately(STATE(initiate_stream));
        }
        if (state_ == STATE_ERROR || state_ == CLOSING)
        {
            return release_and_exit();
        }
        DASSERT(state_ == RUNNING);
        if (!streamWindowRemaining_)
        {
            // We ran out of the current stream window size.
            return call_immediately(STATE(wait_for_stream_proceed));
        }
        if (!remaining())
        {
            // We ran out of the current chunk of stream payload from the
            // source.
            if (requestClose_ && queue_empty())
            {
                requestClose_ = false;
                return call_immediately(STATE(do_close_stream));
            }
            return release_and_exit();
        }
        return call_immediately(STATE(allocate_can_buffer));
    }
 
private:
    void trigger()
    {
        auto *b = alloc();
        this->send(b);
    }
 
    Action initiate_stream()
    {
        // Grabs alias / node ID from the cache.
        node_->iface()->canonicalize_handle(&dst_);
        return allocate_and_call(node_->iface()->addressed_message_write_flow(),
            STATE(send_init_stream));
    }
 
    Action send_init_stream()
    {
        auto *b = get_allocation_result(
            node_->iface()->addressed_message_write_flow());
        b->data()->reset(Defs::MTI_STREAM_INITIATE_REQUEST, node_->node_id(),
            dst_,
            StreamDefs::create_initiate_request(
                streamWindowSize_, false, localStreamId_, dstStreamId_));
 
        node_->iface()->dispatcher()->register_handler(
            &streamInitiateReplyHandler_, Defs::MTI_STREAM_INITIATE_REPLY,
            Defs::MTI_EXACT);
 
        node_->iface()->addressed_message_write_flow()->send(b);
        sleeping_ = true;
        state_ = INITIATING;
        LOG(VERBOSE, "wait for stream init reply");
        return sleep_and_call(&timer_, SEC_TO_NSEC(STREAM_INIT_TIMEOUT_SEC),
            STATE(received_init_stream));
    }
 
    void stream_initiate_replied(Buffer<GenMessage> *message)
    {
        LOG(VERBOSE, "stream init reply: %s",
            string_to_hex(message->data()->payload).c_str());
        auto rb = get_buffer_deleter(message);
        if (message->data()->dstNode != node_ ||
            !node_->iface()->matching_node(dst_, message->data()->src))
        {
            LOG(INFO, "stream reply not for me");
            // Not for me.
            return;
        }
        const auto &payload = message->data()->payload;
        if (payload.size() < 6 || (uint8_t)payload[4] != localStreamId_)
        {
            LOG(INFO, "wrong stream ID %x %x", payload[4], localStreamId_);
            // Talking about another stream or incorrect data.
            return;
        }
        dstStreamId_ = payload[5];
        streamFlags_ = payload[2];
        streamAdditionalFlags_ = payload[3];
        streamWindowSize_ = (payload[0] << 8) | payload[1];
        // Grabs alias / node ID from the cache.
        node_->iface()->canonicalize_handle(&dst_);
 
        // We save the remote alias here if we haven't got any yet.
        if (message->data()->src.alias)
        {
            dst_.alias = message->data()->src.alias;
        }
 
        isLoopbackStream_ =
            (node_->iface()->lookup_local_node_handle(dst_) != nullptr);
 
        sleeping_ = false;
        timer_.trigger();
    }
 
    Action received_init_stream()
    {
        LOG(VERBOSE, "stream init reply wait done");
        node_->iface()->dispatcher()->unregister_handler(
            &streamInitiateReplyHandler_, Defs::MTI_STREAM_INITIATE_REPLY,
            Defs::MTI_EXACT);
        if (!(streamFlags_ & StreamDefs::FLAG_ACCEPT))
        {
            if (streamFlags_ & StreamDefs::FLAG_PERMANENT_ERROR)
            {
                return return_error(DatagramDefs::PERMANENT_ERROR |
                        (streamFlags_ << 8) | streamAdditionalFlags_,
                    "Stream initiate request was denied (permanent error).");
            }
            else
            {
                return return_error(Defs::ERROR_TEMPORARY |
                        (streamFlags_ << 8) | streamAdditionalFlags_,
                    "Stream initiate request was denied (temporary error).");
            }
        }
        if (!streamWindowSize_)
        {
            return return_error(DatagramDefs::PERMANENT_ERROR,
                "Inconsistency: zero buffer length but "
                "accepted stream request.");
        }
        streamWindowRemaining_ = streamWindowSize_;
        node_->iface()->dispatcher()->register_handler(
            &streamProceedHandler_, Defs::MTI_STREAM_PROCEED, Defs::MTI_EXACT);
        state_ = RUNNING;
        return entry();
    }
 
    Action do_close_stream()
    {
        node_->iface()->dispatcher()->unregister_handler(
            &streamProceedHandler_, Defs::MTI_STREAM_PROCEED, Defs::MTI_EXACT);
        return allocate_and_call(node_->iface()->addressed_message_write_flow(),
            STATE(send_close_stream));
    }
 
    Action send_close_stream()
    {
        auto *b = get_allocation_result(
            node_->iface()->addressed_message_write_flow());
        b->data()->reset(Defs::MTI_STREAM_COMPLETE, node_->node_id(), dst_,
            StreamDefs::create_close_request(
                localStreamId_, dstStreamId_, totalByteCount_));
 
        node_->iface()->addressed_message_write_flow()->send(b);
        state_ = CLOSING;
        return entry();
    }
 
    Action allocate_can_buffer()
    {
        return allocate_and_call(
            ifCan_->frame_write_flow(), STATE(got_frame), &canFramePool_);
    }
 
    Action got_frame()
    {
        auto *b = get_allocation_result(ifCan_->frame_write_flow());
 
        uint32_t can_id;
        NodeAlias local_alias =
            ifCan_->local_aliases()->lookup(node_->node_id());
        NodeAlias remote_alias = dst_.alias;
        CanDefs::set_datagram_fields(
            &can_id, local_alias, remote_alias, CanDefs::STREAM_DATA);
        auto *frame = b->data()->mutable_frame();
        SET_CAN_FRAME_ID_EFF(*frame, can_id);
 
        size_t len = compute_next_can_length();
 
        frame->can_dlc = len + 1;
        frame->data[0] = dstStreamId_;
        memcpy(&frame->data[1], payload(), len);
        advance(len);
 
        if (!isLoopbackStream_)
        {
            ifCan_->frame_write_flow()->send(b);
        }
        else
        {
            ifCan_->loopback_frame_write_flow()->send(b);
        }
        return entry();
    }
 
    Action wait_for_stream_proceed()
    {
        if (streamWindowRemaining_)
        {
            // received early stream_proceed response
            return call_immediately(STATE(stream_proceed_timeout));
        }
        sleeping_ = true;
        state_ = FULL;
        return sleep_and_call(&timer_, SEC_TO_NSEC(STREAM_PROCEED_TIMEOUT_SEC),
            STATE(stream_proceed_timeout));
    }
 
    void stream_proceed_received(Buffer<GenMessage> *message)
    {
        auto rb = get_buffer_deleter(message);
        if (message->data()->dstNode != node_ ||
            !node_->iface()->matching_node(dst_, message->data()->src))
        {
            // Not for me.
            return;
        }
 
        const auto &payload = message->data()->payload;
        if (payload.size() < 2 || (uint8_t)payload[0] != localStreamId_)
        {
            // Talking about another stream or incorrect data.
            return;
        }
 
 
        streamWindowRemaining_ += streamWindowSize_;
        if (sleeping_)
        {
            sleeping_ = false;
            timer_.trigger();
        }
    }
 
    Action stream_proceed_timeout()
    {
        if (!streamWindowRemaining_) // no proceed arrived
        {
            return return_error(Defs::ERROR_TEMPORARY,
                "Timed out waiting for stream proceed message.");
            // return call_immediately(STATE(close_stream));
        }
        state_ = RUNNING;
        return entry();
    }
 
private:
    size_t compute_next_can_length()
    {
        size_t ret = remaining();
        // Cannot exceed CAN frame max payload.
        if (ret > MAX_BYTES_PAYLOAD_PER_CAN_FRAME)
        {
            ret = MAX_BYTES_PAYLOAD_PER_CAN_FRAME;
        }
        // Cannot exceed remaining bytes in stream window.
        if (ret > streamWindowRemaining_)
        {
            ret = streamWindowRemaining_;
        }
        return ret;
    }
 
    size_t remaining()
    {
        return message()->data()->size_;
    }
 
    uint8_t *payload()
    {
        return message()->data()->data_;
    }
 
    void advance(size_t num_bytes)
    {
        message()->data()->advance(num_bytes);
        totalByteCount_ += num_bytes;
        streamWindowRemaining_ -= num_bytes;
    }
 
    Action return_error(uint32_t code, string message)
    {
        LOG(INFO, "error %x: %s", (unsigned)code, message.c_str());
        errorCode_ = code;
        state_ = STATE_ERROR;
        return release_and_exit();
    }
 
    static constexpr size_t STREAM_PROCEED_TIMEOUT_SEC = 20;
 
    static constexpr size_t STREAM_INIT_TIMEOUT_SEC = 20;
 
    static constexpr size_t MAX_BYTES_PAYLOAD_PER_CAN_FRAME = 7;
 
    static constexpr size_t MAX_FRAMES_IN_FLIGHT = 4;
 
    static constexpr size_t CAN_FRAME_ALLOC_SIZE =
        sizeof(CanFrameWriteFlow::message_type);
 
    MessageHandler::GenericHandler streamProceedHandler_ {
        this, &StreamSenderCan::stream_proceed_received};
    MessageHandler::GenericHandler streamInitiateReplyHandler_ {
        this, &StreamSenderCan::stream_initiate_replied};
 
    IfCan *ifCan_;
    Node *node_ {nullptr};
    NodeHandle dst_;
    size_t totalByteCount_ {0};
    StreamSenderState state_ {IDLE};
    uint8_t localStreamId_ {StreamDefs::INVALID_STREAM_ID};
    uint8_t dstStreamId_ {StreamDefs::INVALID_STREAM_ID};
    uint8_t isLoopbackStream_ : 1;
    uint8_t sleeping_ : 1;
    uint8_t requestClose_ : 1;
    uint8_t requestInit_ : 1;
    uint8_t streamFlags_ {0};
    uint8_t streamAdditionalFlags_ {0};
    uint16_t streamWindowSize_ {StreamDefs::MAX_PAYLOAD};
    uint16_t streamWindowRemaining_ {0};
    uint32_t errorCode_ {0};
    LimitedPool canFramePool_ {CAN_FRAME_ALLOC_SIZE, MAX_FRAMES_IN_FLIGHT};
    StateFlowTimer timer_ {this};
};
 
class StreamRendererCan : public StateFlow<ByteBuffer, QList<1>>
{ };
 
} // namespace openlcb
 
#endif // _OPENLCB_STREAMSENDER_HXX_