HubDeviceSelect.hxx

 
#ifndef _UTILS_HUBDEVICESELECT_HXX_
#define _UTILS_HUBDEVICESELECT_HXX_
 
#include "openmrn_features.h"
 
// Only compile HubDeviceSelect if the OS has implementation for
// Executor::select.
#ifdef OPENMRN_FEATURE_EXECUTOR_SELECT
 
#include <unistd.h>
#include <stdio.h>
#include <fcntl.h>
 
#include "executor/StateFlow.hxx"
#include "utils/Hub.hxx"
#include "utils/LimitedPool.hxx"
 
template <class BType> struct SelectBufferInfo
{
};
 
template <> struct SelectBufferInfo<HubFlow::buffer_type>
{
    static void resize_target(HubFlow::buffer_type *b)
    {
        b->data()->resize(64);
    }
    static void check_target_size(HubFlow::buffer_type *b, int remaining)
    {
        HASSERT(remaining >= 0);
        HASSERT(remaining <= 64);
        b->data()->resize(64 - remaining);
    }
    static bool needs_read_fully()
    {
        return false;
    }
 
    static bool limit_input()
    {
        return true;
    }
};
 
template <class T>
struct SelectBufferInfo<Buffer<HubContainer<StructContainer<T>>>>
{
    typedef Buffer<HubContainer<StructContainer<T>>> buffer_type;
 
    static void resize_target(buffer_type *b)
    {
    }
    static void check_target_size(buffer_type *b, int remaining)
    {
        HASSERT(remaining == 0);
    }
    static bool needs_read_fully()
    {
        return true;
    }
 
    static bool limit_input()
    {
        return true;
    }
};
 
template <>
struct SelectBufferInfo<Buffer<CanHubData>> {
    typedef Buffer<CanHubData> buffer_type;
    
    static void resize_target(buffer_type *b)
    {
    }
    static void check_target_size(buffer_type *b, int remaining)
    {
        HASSERT(remaining == 0);
    }
    static bool needs_read_fully()
    {
        return true;
    }
 
    static bool limit_input()
    {
        // We should never throttle a CAN-bus reader.
        return false;
    }
};
 
int hubdevice_incoming_packet_limit();
    
template <class HFlow> class HubDeviceSelectReadFlow : public StateFlowBase
{
public:
    typedef typename HFlow::buffer_type buffer_type;
 
    HubDeviceSelectReadFlow(FdHubPortService *device,
        typename HFlow::port_type *dst, typename HFlow::port_type *skip_member)
        : StateFlowBase(device)
        , shouldThrottle_(SelectBufferInfo<buffer_type>::limit_input())
        , b_(nullptr)
        , dst_(dst)
        , skipMember_(skip_member)
    {
        set_limit_input(shouldThrottle_);
        this->start_flow(STATE(allocate_buffer));
    }
 
    void set_limit_input(bool should_throttle)
    {
        shouldThrottle_ = should_throttle;
        int limit = hubdevice_incoming_packet_limit();
        if (shouldThrottle_ && limit < 1000000 && !inputPool_)
        {
            inputPool_.reset(new LimitedPool(sizeof(buffer_type), limit));
        }
    }
 
    void shutdown()
    {
        auto *e = this->service()->executor();
        if (e->is_selected(&selectHelper_))
        {
            e->unselect(&selectHelper_);
        }
        set_terminated();
        notify_barrier();
    }
 
    FdHubPortService *device()
    {
        return static_cast<FdHubPortService *>(this->service());
    }
 
    Action allocate_buffer()
    {
        // If inputPool is nullptr, this will use the destination flow's pool,
        // which is typically mainBufferPool.
        return this->allocate_and_call(dst_, STATE(try_read), inputPool_.get());
    }
 
    Action try_read()
    {
        b_ = this->get_allocation_result(dst_);
        b_->data()->skipMember_ = skipMember_;
        SelectBufferInfo<buffer_type>::resize_target(b_);
        if (SelectBufferInfo<buffer_type>::needs_read_fully())
        {
            return this->read_repeated(&selectHelper_, device()->fd(),
                (void *)b_->data()->data(), b_->data()->size(),
                STATE(read_done), 0);
        }
        else
        {
            return this->read_single(&selectHelper_, device()->fd(),
                (void *)b_->data()->data(), b_->data()->size(),
                STATE(read_done), 0);
        }
    }
 
    Action read_done()
    {
        if (selectHelper_.hasError_)
        {
            b_->unref();
            set_terminated();
            device()->report_read_error();
            notify_barrier();
            return exit();
        }
        SelectBufferInfo<buffer_type>::check_target_size(
            b_, selectHelper_.remaining_);
        dst_->send(b_, 0);
        b_ = nullptr;
        return this->call_immediately(STATE(allocate_buffer));
    }
 
private:
    void notify_barrier()
    {
        if (barrierOwned_)
        {
            barrierOwned_ = false;
            device()->barrier_.notify();
        }
    }
 
    bool barrierOwned_{true};
    bool shouldThrottle_;
    StateFlowSelectHelper selectHelper_{this};
    buffer_type *b_;
    std::unique_ptr<LimitedPool> inputPool_;
    typename HFlow::port_type *dst_;
    typename HFlow::port_type *skipMember_;
};
 
template <class HFlow, class ReadFlow = HubDeviceSelectReadFlow<HFlow>>
class HubDeviceSelect : public FdHubPortService, private Atomic
{
public:
#ifndef __WINNT__
    HubDeviceSelect(
        HFlow *hub, const char *path, Notifiable *on_error = nullptr)
        : FdHubPortService(
              hub->service()->executor(), ::open(path, O_RDWR | O_NONBLOCK))
        , hub_(hub)
        , readFlow_(this, hub, &writeFlow_)
        , writeFlow_(this)
    {
        HASSERT(fd_ >= 0);
        barrier_.reset(
            on_error ? on_error : EmptyNotifiable::DefaultInstance());
        barrier_.new_child();
        hub_->register_port(write_port());
        isRegistered_ = true;
    }
#endif
 
    HubDeviceSelect(HFlow *hub, int fd, Notifiable *on_error = nullptr)
        : FdHubPortService(hub->service()->executor(), fd)
        , hub_(hub)
        , readFlow_(this, hub, &writeFlow_)
        , writeFlow_(this)
    {
        HASSERT(fd_ >= 0);
        barrier_.reset(
            on_error ? on_error : EmptyNotifiable::DefaultInstance());
        barrier_.new_child();
#ifdef __WINNT__
        unsigned long par = 1;
        ioctlsocket(fd_, FIONBIO, &par);
#else
        ::fcntl(fd, F_SETFL, O_RDWR | O_NONBLOCK);
#endif
        hub_->register_port(write_port());
        isRegistered_ = true;
    }
 
    virtual ~HubDeviceSelect()
    {
        if (fd_ >= 0) {
            unregister_write_port();
            close_fd();
            executor()->sync_run([this]() {
                readFlow_.shutdown();
                writeFlow_.shutdown();
            });
        }
        bool completed = false;
        while (!completed)
        {
            executor()->sync_run([this, &completed]() {
                if (barrier_.is_done())
                {
                    completed = true;
                }
            });
        }
    }
 
    HFlow *hub()
    {
        return hub_;
    }
 
    typename HFlow::port_type *write_port()
    {
        return &writeFlow_;
    }
 
    void unregister_write_port()
    {
        LOG(VERBOSE, "HubDeviceSelect::unregister write port %p %p",
            write_port(), &writeFlow_);
        {
            AtomicHolder h(this);
            if (!isRegistered_)
            {
                return;
            }
            isRegistered_ = false;
        }
        hub_->unregister_port(&writeFlow_);
        /* We put an empty message at the end of the queue. This will cause
         * wait until all pending messages are dealt with, and then ping the
         * barrier notifiable, commencing the shutdown. */
        auto *b = writeFlow_.alloc();
        b->set_done(&barrier_);
        writeFlow_.send(b);
    }
 
    bool write_done()
    {
        return writeFlow_.is_waiting();
    }
 
protected:
    typedef StateFlow<typename HFlow::buffer_type, QList<1>> WriteFlowBase;
    class WriteFlow : public WriteFlowBase
    {
    public:
        WriteFlow(HubDeviceSelect *dev)
            : WriteFlowBase(dev)
        {
        }
 
        ~WriteFlow()
        {
            HASSERT(this->is_waiting());
        }
 
        void shutdown()
        {
            // The fd must be set to negative already to ensure the shutdown
            // completes successfully.
            HASSERT(device()->fd() < 0);
            auto* e = this->service()->executor();
            if (!selectHelper_.is_empty() && e->is_selected(&selectHelper_)) {
                e->unselect(&selectHelper_);
                // will make the internal_try_write exit immediately
                selectHelper_.remaining_ = 0;
                // actually wake up the flow
                this->notify();
            }
        }
 
        HubDeviceSelect *device()
        {
            return static_cast<HubDeviceSelect *>(this->service());
        }
 
        StateFlowBase::Action entry() OVERRIDE
        {
            if (device()->fd() < 0) {
                return this->release_and_exit();
            }
            return this->write_repeated(&selectHelper_, device()->fd(),
                this->message()->data()->data(),
                this->message()->data()->size(), STATE(write_done),
                this->priority());
        }
 
        StateFlowBase::Action write_done()
        {
            if (selectHelper_.hasError_) {
                device()->report_write_error();
            }
            return this->release_and_exit();
        }
 
    private:
        StateFlowBase::StateFlowSelectHelper selectHelper_{this};
    };
 
protected:
 
    void report_write_error() override
    {
        readFlow_.shutdown();
        unregister_write_port();
        close_fd();
    }
 
    void report_read_error() override
    {
        unregister_write_port();
        close_fd();
    }
 
    void close_fd()
    {
        int fd = -1;
        {
            AtomicHolder h(this);
            fd = fd_;
            if (fd < 0)
            {
                return;
            }
            fd_ = -1;
        }
        // This is a workaround that sometimes my linux kernel gets stuck in
        // ::read when I closed the fd like this, even though the fd is
        // O_NONBLOCK.
        executor()->add(new CallbackExecutable([this, fd]() {
            ::close(fd);
            readFlow_.shutdown();
            writeFlow_.shutdown();
        }));
    }
 
    HFlow *hub_;
    ReadFlow readFlow_;
    WriteFlow writeFlow_;
    bool isRegistered_;
};
 
#endif // FEATURE_EXECUTOR_SELECT
 
 
#endif // _UTILS_HUBDEVICESELECT_HXX_