MemoryConfigStream.hxx

Go to the documentation of this file.

 
#ifndef _OPENLCB_MEMORYCONFIGSTREAM_HXX_
#define _OPENLCB_MEMORYCONFIGSTREAM_HXX_
 
#include "openlcb/If.hxx"
#include "openlcb/MemoryConfig.hxx"
#include "openlcb/StreamSender.hxx"
#include "openlcb/StreamTransport.hxx"
 
namespace openlcb
{
 
class MemorySpaceStreamReadFlow : public StateFlowBase
{
public:
    using CallbackFn = std::function<void(uint16_t)>;
 
    MemorySpaceStreamReadFlow(Node *node, MemorySpace *space, NodeHandle dst,
        uint8_t dst_stream_id, uint32_t ofs, uint32_t len,
        CallbackFn started_cb)
        : StateFlowBase(node->iface())
        , dst_(dst)
        , startedCb_(std::move(started_cb))
        , space_(space)
        , dstStreamId_(dst_stream_id)
        , node_(node)
        , ofs_(ofs)
        , len_(len)
    {
        LOG(INFO, "starting streamed read, dst %02x", dstStreamId_);
        start_flow(STATE(alloc_stream));
    }
 
    uint8_t get_src_stream_id()
    {
        return srcStreamId_;
    }
 
    uint8_t get_dst_stream_id()
    {
        return dstStreamId_;
    }
 
private:
    Action alloc_stream()
    {
        return allocate_and_call(
            STATE(got_sender), stream_transport()->sender_allocator());
    }
 
    Action got_sender()
    {
        LOG(INFO, "got sender");
        sender_ =
            full_allocation_result(stream_transport()->sender_allocator());
        senderCan_ = static_cast<StreamSenderCan *>(sender_);
        HASSERT(senderCan_);
        return call_immediately(STATE(initiate_stream));
    }
 
    Action initiate_stream()
    {
        LOG(INFO, "initiate");
        srcStreamId_ = stream_transport()->get_send_stream_id();
        senderCan_->start_stream(node_, dst_, srcStreamId_, dstStreamId_);
        return call_immediately(STATE(wait_for_started));
    }
 
    Action wait_for_started()
    {
        auto state = senderCan_->get_state();
        if (state == StreamSender::RUNNING)
        {
            dstStreamId_ = senderCan_->get_dst_stream_id();
            startedCb_(0);
            return call_immediately(STATE(alloc_buffer));
        }
        if (state == StreamSender::STATE_ERROR)
        {
            auto err = senderCan_->get_error();
            LOG(INFO, "failed to start stream: 0x%04x", err);
            startedCb_(err);
            return call_immediately(STATE(done_stream));
        }
        return sleep_and_call(
            &timer_, MSEC_TO_NSEC(3), STATE(wait_for_started));
    }
 
    Action alloc_buffer()
    {
        return allocate_and_call<RawData>(
            nullptr, STATE(have_raw_buffer), &sendBufferPool_);
    }
 
    Action have_raw_buffer()
    {
        LOG(INFO, "have raw buf len %u", (unsigned)len_);
 
        RawBufferPtr raw_buffer(get_allocation_result<RawData>(nullptr));
        sendBuffer_ = get_buffer_deleter(sender_->alloc());
        sendBuffer_->data()->set_from(std::move(raw_buffer), 0);
        return call_immediately(STATE(try_read));
    }
 
    Action try_read()
    {
        if (!len_)
        {
            sender_->send(sendBuffer_.release());
            senderCan_->close_stream();
            return call_immediately(STATE(wait_for_close));
        }
        size_t free = sendBuffer_->data()->free_space();
        if (!free)
        {
            sender_->send(sendBuffer_.release());
            return call_immediately(STATE(alloc_buffer));
        }
        size_t cnt = len_;
        if (free < cnt)
        {
            cnt = free;
        }
        uint8_t *ptr = sendBuffer_->data()->append_ptr();
        MemorySpace::errorcode_t err;
        size_t copied = space_->read(ofs_, ptr, cnt, &err, this);
        sendBuffer_->data()->append_complete(copied);
        ofs_ += copied;
        len_ -= copied;
        if (err == MemorySpace::ERROR_AGAIN)
        {
            return wait();
        }
        if (err == MemoryConfigDefs::ERROR_OUT_OF_BOUNDS)
        {
            sender_->send(sendBuffer_.release());
            senderCan_->close_stream();
            return call_immediately(STATE(wait_for_close));
        }
        if (!err)
        {
            return again();
        }
        else
        {
            LOG(INFO, "error reading input stream: %04x", err);
            sender_->send(sendBuffer_.release());
            senderCan_->close_stream(err);
            return call_immediately(STATE(wait_for_close));
        }
    }
 
    Action wait_for_close()
    {
        auto state = senderCan_->get_state();
        if (state == StreamSender::CLOSING && senderCan_->is_waiting())
        {
            // Sender is done and empty.
            return call_immediately(STATE(done_stream));
        }
        if (state == StreamSender::STATE_ERROR && senderCan_->is_waiting())
        {
            // Sender has errored and consumed / thrown away all data.
            // There is no place really to show the error.
            LOG(INFO, "Stream sender error: 0x%04x", senderCan_->get_error());
            return call_immediately(STATE(done_stream));
        }
        return sleep_and_call(&timer_, MSEC_TO_NSEC(3), STATE(wait_for_close));
    }
 
    Action done_stream()
    {
        senderCan_->clear();
        stream_transport()->sender_allocator()->typed_insert(sender_);
        sender_ = nullptr;
        return delete_this();
    }
 
    StreamTransport *stream_transport()
    {
        return node_->iface()->stream_transport();
    }
 
    LimitedPool sendBufferPool_ {sizeof(RawBuffer), 2, rawBufferPool};
    ByteBufferPtr sendBuffer_;
    StateFlowTimer timer_ {this};
    NodeHandle dst_;
    CallbackFn startedCb_;
    MemorySpace *space_;
    uint8_t srcStreamId_ {StreamDefs::INVALID_STREAM_ID};
    uint8_t dstStreamId_;
    Node *node_;
    uint32_t ofs_;
    uint32_t len_;
    StreamSenderCan *senderCan_;
    StreamSender *sender_;
};
 
class MemoryConfigStreamHandler : public MemoryConfigHandlerBase
{
public:
    MemoryConfigStreamHandler(MemoryConfigHandler *parent)
        : MemoryConfigHandlerBase(parent->dg_service())
        , parent_(parent)
    {
        parent_->set_stream_handler(this);
    }
 
    Action entry() override
    {
        LOG(INFO, "stream req");
        // The verification of the incoming data is already done by the calling
        // MemoryConfigHandler.
        response_.clear();
        const uint8_t *bytes = in_bytes();
        uint8_t cmd = bytes[1];
 
        switch (cmd & MemoryConfigDefs::COMMAND_MASK)
        {
            case MemoryConfigDefs::COMMAND_READ_STREAM:
            {
                return call_immediately(STATE(handle_read_stream));
            }
        }
        return respond_reject(Defs::ERROR_UNIMPLEMENTED_SUBCMD);
    }
 
private:
    Action handle_read_stream()
    {
        size_t len = message()->data()->payload.size();
        const uint8_t *bytes = in_bytes();
 
        if (len < 8)
        {
            return respond_reject(Defs::ERROR_INVALID_ARGS);
        }
        MemorySpace *space = get_space();
        if (!space)
        {
            return respond_reject(MemoryConfigDefs::ERROR_SPACE_NOT_KNOWN);
        }
 
        size_t stream_data_offset = 6;
        if (has_custom_space())
        {
            ++stream_data_offset;
        }
        if (len < stream_data_offset + 2)
        {
            return respond_reject(Defs::ERROR_INVALID_ARGS);
        }
 
        uint8_t dst_stream_id = bytes[stream_data_offset + 1];
        uint32_t num_bytes_to_read = 0xFFFFFFFFu;
        LOG(INFO, "dst stream id %02x", dst_stream_id);
        if (len >= stream_data_offset + 6)
        {
            memcpy(&num_bytes_to_read, bytes + stream_data_offset + 2, 4);
            num_bytes_to_read = be32toh(num_bytes_to_read);
        }
        streamErrorCode_ = Defs::ERROR_TEMPORARY;
        // This object is self-owned, so it will run `delete this`.
        readFlow_ = new MemorySpaceStreamReadFlow(message()->data()->dst,
            get_space(), message()->data()->src, dst_stream_id, get_address(),
            num_bytes_to_read,
            std::bind(&MemoryConfigStreamHandler::stream_start_cb, this,
                std::placeholders::_1));
        return wait_and_call(STATE(read_started));
    }
 
    void stream_start_cb(uint16_t error)
    {
        streamErrorCode_ = error;
        notify();
    }
 
    Action read_started()
    {
        uint16_t error = streamErrorCode_;
        size_t response_data_offset = 6;
        if (has_custom_space())
        {
            ++response_data_offset;
        }
        response_.reserve(response_data_offset + 6);
        response_.resize(response_data_offset + 2);
        uint8_t *response_bytes = out_bytes();
        response_bytes[0] = DATAGRAM_ID;
        response_bytes[1] = error ? MemoryConfigDefs::COMMAND_READ_STREAM_FAILED
                                  : MemoryConfigDefs::COMMAND_READ_STREAM_REPLY;
        set_address_and_space();
        if (error)
        {
            response_.resize(response_data_offset + 2);
            response_bytes[response_data_offset] = error >> 8;
            response_bytes[response_data_offset + 1] = error & 0xff;
        }
        else
        {
            response_.resize(response_data_offset + 2);
            response_bytes[response_data_offset] =
                readFlow_->get_src_stream_id();
            response_bytes[response_data_offset + 1] =
                readFlow_->get_dst_stream_id();
            if (message()->data()->payload.size() >= response_data_offset + 6)
            {
                response_.resize(response_data_offset + 6);
                memcpy(response_bytes + response_data_offset + 2,
                    in_bytes() + response_data_offset + 2, 4);
            }
        }
        return respond_ok(DatagramClient::REPLY_PENDING);
    }
 
    MemorySpace *get_space()
    {
        int space_number = get_space_number();
        if (space_number < 0)
            return nullptr;
        MemorySpace *space =
            registry()->lookup(message()->data()->dst, space_number);
        if (!space)
        {
            LOG(WARNING,
                "MemoryConfig: asked node 0x%012" PRIx64 " for unknown space "
                "%d. Source {0x%012" PRIx64 ", %03x}",
                message()->data()->dst->node_id(), space_number,
                message()->data()->src.id, message()->data()->src.alias);
            return nullptr;
        }
        if (!space->set_node(message()->data()->dst))
        {
            LOG(WARNING, "MemoryConfig: Global space %d rejected node.",
                space_number);
            return nullptr;
        }
        return space;
    }
 
    Registry *registry()
    {
        return parent_->registry();
        ;
    }
 
    MemoryConfigHandler *parent_;
 
    uint16_t streamErrorCode_;
 
    union
    {
        MemorySpaceStreamReadFlow *readFlow_;
    };
}; // class MemoryConfigStreamHandler
 
} // namespace openlcb
 
#endif // _OPENLCB_MEMORYCONFIGSTREAM_HXX_