BLEHubPort.hxx

Go to the documentation of this file.

 
#ifndef _OPENLCB_BLEHUBPORT_HXX_
#define _OPENLCB_BLEHUBPORT_HXX_
 
#include <functional>
 
#include "executor/StateFlow.hxx"
#include "openlcb/BLEDefs.hxx"
#include "os/OS.hxx"
#include "utils/DirectHub.hxx"
 
extern DataBufferPool g_direct_hub_data_pool;
 
namespace openlcb
{
 
class BLEHubPort : public DirectHubPort<uint8_t[]>,
                   private StateFlowBase,
                   public BLEProtocolEngine
{
public:
    using SendFunction = std::function<void(const uint8_t *data, size_t len)>;
 
    static constexpr size_t MAX_BYTES_PER_WRITE = 220;
 
    BLEHubPort(DirectHubInterface<uint8_t[]> *hub,
        std::unique_ptr<MessageSegmenter> segmenter, Service *ble_write_service,
        SendFunction send_function, Notifiable *on_error = nullptr)
        : StateFlowBase(ble_write_service)
        , pendingShutdown_(false)
        , waitingForAck_(false)
        , sendFunction_(std::move(send_function))
        , onError_(on_error)
        , input_(this, hub, std::move(segmenter))
    {
        // Sets the initial state of the write flow to the stage where we read
        // the next entry from the queue.
        wait_and_call(STATE(read_queue));
        notRunning_ = true;
 
        hub->register_port(this);
    }
 
    void disconnect_and_delete() override
    {
        {
            AtomicHolder l(lock());
            HASSERT(!pendingShutdown_);
            input_.hub_->unregister_port(this);
            // After this, the next notify will eventually reach the shutdown
            // and delete state.
            pendingShutdown_ = true;
            if (notRunning_)
            {
                notRunning_ = 0;
                notify();
            }
        }
        // Synchronization point that ensures that there is no currently
        // running Executable on this executor. This ensures that is no
        // currently pending nor will there be any future invocations of
        // sendFunction_ after this function returns.
        service()->executor()->sync_run([]() {});
    }
 
    // ===== API from the BLE stack connection =====
 
    void ack()
    {
        ack_helper();
        LOG(ALWAYS, "BLE ack, pend %d", sendPending_);
    }
 
    void nack()
    {
        ack_helper();
        LOG(ALWAYS, "BLE nack, pend %d", sendPending_);
    }
 
    void input_data(const uint8_t *data, size_t len)
    {
        input_.input_data(data, len);
    }
 
    void send(MessageAccessor<uint8_t[]> *msg) override
    {
        if (pendingShutdown_)
        {
            // Port already closed. Ignore data to send.
            return;
        }
        {
            AtomicHolder h(lock());
            if (pendingTail_ && pendingTail_->buf_.try_append_from(msg->buf_))
            {
                totalPendingSize_ += msg->buf_.size();
                // Successfully enqueued the bytes into the tail of the queue.
                // Nothing else to do here.
                return;
            }
        }
 
        BufferType *b;
        mainBufferPool->alloc(&b);
        b->data()->buf_.reset(msg->buf_);
        if (msg->done_)
        {
            b->set_done(msg->done_->new_child());
        }
        // Checks if we need to wake up the flow.
        {
            AtomicHolder h(lock());
            if (pendingShutdown_)
            {
                // Catch race condition when port is already closed.
                b->unref();
                return;
            }
            pendingQueue_.insert_locked(b);
            totalPendingSize_ += msg->buf_.size();
            pendingTail_ = b->data();
            if (notRunning_)
            {
                notRunning_ = 0;
                // When we have exactly one buffer at hand, we release the done
                // notify of it to allow the upstream stack to generate more
                // data quicker.
                b->set_done(nullptr);
            }
            else
            {
                // flow already running. Skip notify.
                return;
            }
        }
        notify();
    }
 
    Action read_queue()
    {
        if (pendingShutdown_)
        {
            return call_immediately(STATE(shutdown_and_exit));
        }
        BufferType *head;
        {
            AtomicHolder h(lock());
            head = static_cast<BufferType *>(pendingQueue_.next_locked().item);
            if (!head)
            {
                notRunning_ = true;
                return wait();
            }
            HASSERT(head);
            if (head->data() == pendingTail_)
            {
                pendingTail_ = nullptr;
            }
            totalPendingSize_ -= head->data()->buf_.size();
        }
 
        currentHead_.reset(head);
        return call_immediately(STATE(do_write));
    }
 
    Action do_write()
    {
        if (pendingShutdown_)
        {
            return call_immediately(STATE(shutdown_and_exit));
        }
 
        const uint8_t *read_ptr;
        size_t num_bytes;
        auto &b = currentHead_->data()->buf_;
        read_ptr = b.data_read_pointer(&num_bytes);
        if (num_bytes > MAX_BYTES_PER_WRITE)
        {
            num_bytes = MAX_BYTES_PER_WRITE;
        }
        {
            AtomicHolder h(lock());
            ++sendPending_;
        }
        LOG(INFO, "BLE send %d bytes pendcount %d queuesize %d/%d", (int)num_bytes, sendPending_, (int)totalPendingSize_, pendingQueue_.pending());
        sendFunction_(read_ptr, num_bytes);
        b.data_read_advance(num_bytes);
        if (b.size())
        {
            return yield();
        }
        else
        {
            currentHead_.reset();
            AtomicHolder h(lock());
            if (pendingShutdown_)
            {
                return call_immediately(STATE(shutdown_and_exit));
            }
            if (sendPending_ > 1)
            {
                waitingForAck_ = 1;
                return wait_and_call(STATE(read_queue));
            }
            else if (pendingQueue_.empty())
            {
                // go back to sleep
                notRunning_ = 1;
                return wait_and_call(STATE(read_queue));
            }
            else
            {
                return yield_and_call(STATE(read_queue));
            }
        }
    }
 
    Action shutdown_and_exit()
    {
        // Synchronization with other threads that might have written
        // pendingShutdown == true.
        {
            AtomicHolder h(lock());
        }
 
        // Releases all buffers.
        currentHead_.reset();
        while (auto *h = static_cast<BufferType *>(pendingQueue_.next().item))
        {
            h->unref();
        }
        return delete_this();
    }
 
protected:
    void ack_helper()
    {
        {
            AtomicHolder h(lock());
            --sendPending_;
            if (waitingForAck_ && sendPending_ <= 1)
            {
                waitingForAck_ = false;
                notify();
            }
        }
    }
 
    class InputFlow : public StateFlowBase, private Atomic
    {
    public:
        InputFlow(BLEHubPort* parent, DirectHubInterface<uint8_t[]> *hub,
            std::unique_ptr<MessageSegmenter> segmenter)
            : StateFlowBase(hub->get_service())
            , parent_(parent)
            , segmenter_(std::move(segmenter))
            , hub_(hub)
        {
            segmenter_->clear();
            flowWaiting_ = true;
            wait_and_call(STATE(take_input));
        }
 
        void input_data(const uint8_t *data, size_t len)
        {
            while (len)
            {
                uint8_t *dst = nullptr;
                size_t free = 0;
                if (!appendBuf_.free())
                {
                    DataBuffer *p;
                    g_direct_hub_data_pool.alloc(&p);
                    appendBuf_.append_empty_buffer(p);
                }
                dst = appendBuf_.data_write_pointer();
                free = appendBuf_.free();
                if (len < free) {
                    free = len;
                }
                memcpy(dst, data, free);
                len -= free;
                data += free;
                appendBuf_.data_write_advance(free);
            }
            {
                AtomicHolder h(this);
                HASSERT(transferBuf_.try_append_from(appendBuf_, true));
                if (flowWaiting_) {
                    flowWaiting_ = false;
                    notify();
                }
            }
            appendBuf_.data_read_advance(appendBuf_.size());
        }
 
        Action take_input() {
            AtomicHolder h(this);
            HASSERT(segmentBuf_.try_append_from(transferBuf_, true));
            transferBuf_.data_read_advance(transferBuf_.size());
            return call_immediately(STATE(segment_head));
        }
 
        Action segment_head() {
            size_t len = 0;
            const uint8_t* ptr = segmentBuf_.data_read_pointer(&len);
            if (!len) {
                AtomicHolder h(this);
                if (transferBuf_.size()) {
                    return call_immediately(STATE(take_input));
                }
                flowWaiting_ = true;
                return wait_and_call(STATE(take_input));
            }
            ssize_t segment_size = segmenter_->segment_message(ptr, len);
            size_t xfer = len;
            if (segment_size) {
                xfer = segment_size - outputBuf_.size();
            }
            LinkedDataBufferPtr p;
            p.reset(segmentBuf_, xfer);
            segmentBuf_.data_read_advance(xfer);
            HASSERT(outputBuf_.try_append_from(p, true));
            if (segment_size) {
                segmenter_->clear();
                // completed data
                return call_immediately(STATE(send_output));
            } else {
                // Need to segment more data.
                return again();
            }
        }
 
        Action send_output() {
            // We expect either an inline call to our run() method or
            // later a callback on the executor. This sequence of calls
            // prepares for both of those options.
            wait_and_call(STATE(send_callback));
            inlineCall_ = 1;
            sendComplete_ = 0;
            hub_->enqueue_send(this); // causes the callback
            inlineCall_ = 0;
            if (sendComplete_)
            {
                return send_done();
            }
            return wait();
        }
 
        Action send_callback()
        {
            auto *m = hub_->mutable_message();
            //m->set_done(buf_.tail()->new_child());
            m->source_ = parent_;
            // This call transfers the chained head of the current buffers,
            // taking additional references where necessary or transferring the
            // existing reference. It adjusts the skip_ and size_ arguments in
            // buf_ to continue from where we left off.
            m->buf_ = std::move(outputBuf_);
            hub_->do_send();
            sendComplete_ = 1;
            if (inlineCall_)
            {
                // do not disturb current state.
                return wait();
            }
            else
            {
                // we were called queued; go back to running the flow on the
                // main executor.
                return yield_and_call(STATE(send_done));
            }
        }
 
        Action send_done()
        {
            // Goes back to looking at more data from the transfered buffers.
            return call_immediately(STATE(segment_head));
        }
        
    private:
        friend class BLEHubPort;
 
        BLEHubPort* parent_;
 
        bool flowWaiting_;
        uint8_t inlineCall_ : 1;
        uint8_t sendComplete_ : 1;
        
        LinkedDataBufferPtr appendBuf_;
        LinkedDataBufferPtr transferBuf_;
        LinkedDataBufferPtr segmentBuf_;
        LinkedDataBufferPtr outputBuf_;
        std::unique_ptr<MessageSegmenter> segmenter_;
        DirectHubInterface<uint8_t[]> *hub_;
        
    };
 
    struct OutputDataEntry
    {
        LinkedDataBufferPtr buf_;
    };
    typedef Buffer<OutputDataEntry> BufferType;
    typedef Q QueueType;
 
    Atomic *lock()
    {
        return pendingQueue_.lock();
    }
 
    bool pendingShutdown_ : 1;
    bool notRunning_ : 1;
    bool waitingForAck_ : 1;
 
    QueueType pendingQueue_;
    OutputDataEntry *pendingTail_ = nullptr;
    size_t totalPendingSize_ = 0;
 
    BufferPtr<OutputDataEntry> currentHead_;
 
    SendFunction sendFunction_;
    Notifiable *onError_ = nullptr;
 
    int sendPending_ {0};
 
    InputFlow input_;
 
}; // class BLEHubPort
 
} // namespace openlcb
 
#endif // _OPENLCB_BLEHUBPORT_HXX_