HubDeviceNonBlock.hxx

 
#ifndef _UTILS_HUBDEVICENONBLOCK_HXX_
#define _UTILS_HUBDEVICENONBLOCK_HXX_
 
#include "openmrn_features.h"
#ifdef OPENMRN_FEATURE_FD_CAN_DEVICE
 
#include <unistd.h>
#include <stdio.h>
#include <fcntl.h>
 
#include "executor/StateFlow.hxx"
#ifdef __FreeRTOS__
#include "freertos/can_ioctl.h"
#else
#include "can_ioctl.h"
#endif
#include "utils/Hub.hxx"
 
#ifdef __FreeRTOS__
extern int ioctl(int fd, unsigned long int key, ...);
#endif // __FreeRTOS__
 
template <class HFlow> class HubDeviceNonBlock : public Destructable, private Atomic, public Service
{
public:
    HubDeviceNonBlock(HFlow *hub, const char *path)
        : Service(hub->service()->executor())
        , fd_(::open(path, O_RDWR | O_NONBLOCK))
        , hub_(hub)
        , readFlow_(this)
        , writeFlow_(this)
    {
        HASSERT(fd_ >= 0);
        hub_->register_port(write_port());
    }
 
    virtual ~HubDeviceNonBlock()
    {
        hub_->unregister_port(write_port());
    }
 
    HFlow *hub()
    {
        return hub_;
    }
 
    typename HFlow::port_type *write_port()
    {
        return &writeFlow_;
    }
 
    int fd()
    {
        return fd_;
    }
 
protected:
    class ReadFlow : public StateFlowBase
    {
    public:
        ReadFlow(HubDeviceNonBlock *device)
            : StateFlowBase(device)
            , b_(nullptr)
        {
            this->start_flow(STATE(allocate_buffer));
        }
 
        HubDeviceNonBlock *device()
        {
            return static_cast<HubDeviceNonBlock *>(this->service());
        }
 
        void notify() OVERRIDE
        {
            service()->executor()->add(this, 0);
        }
 
        void notify_from_isr() OVERRIDE
        {
            // We override to priority zero.
            service()->executor()->add_from_isr(this, 0);
        }
 
        Action allocate_buffer()
        {
            return this->allocate_and_call(device()->hub(), STATE(try_read));
        }
 
        Action try_read()
        {
            b_ = this->get_allocation_result(device()->hub());
            b_->data()->skipMember_ = device()->write_port();
            return this->call_immediately(STATE(retry_read));
        }
 
        Action retry_read()
        {
            HASSERT(b_);
            int ret =
                ::read(device()->fd(), b_->data()->data(), b_->data()->size());
            if (ret <= 0)
            {
                // We are blocked. There is no race condition here, because the
                // contract of the ioctl is that if there is any data in the
                // queuy, they will immediately call us.
                HASSERT(::ioctl(device()->fd(), CAN_IOC_READ_ACTIVE, this) == 0);
                return this->wait();
            }
            else
            {
                HASSERT((unsigned)ret == b_->data()->size());
                device()->hub()->send(b_, 0);
                b_ = nullptr;
                return this->call_immediately(STATE(allocate_buffer));
            }
        }
 
    private:
        typename HFlow::buffer_type *b_;
    };
 
    typedef StateFlow<typename HFlow::buffer_type, QList<1>> WriteFlowBase;
    class WriteFlow : public WriteFlowBase
    {
    public:
        WriteFlow(HubDeviceNonBlock *dev)
            : WriteFlowBase(dev)
        {
        }
 
        HubDeviceNonBlock *device()
        {
            return static_cast<HubDeviceNonBlock *>(this->service());
        }
 
        void notify() OVERRIDE
        {
            this->service()->executor()->add(this, 0);
        }
 
        void notify_from_isr() OVERRIDE
        {
            this->service()->executor()->add_from_isr(this, 0);
        }
 
        StateFlowBase::Action entry() OVERRIDE
        {
            bufferPos_ = static_cast<uint8_t*>(this->message()->data()->data());
            len_ = this->message()->data()->size();
            return this->call_immediately(STATE(try_write));
        }
 
        StateFlowBase::Action try_write()
        {
            ssize_t ret = ::write(device()->fd(), bufferPos_, len_);
            if (!ret)
            {
                // We are blocked. There is no race condition here, because the
                // contract of the ioctl is that if there is any data in the
                // queuy, they will immediately call us.
                HASSERT(::ioctl(device()->fd(), CAN_IOC_WRITE_ACTIVE, this) == 0);
                return this->wait();
            }
            else if (ret > 0)
            {
                len_ -= ret;
                bufferPos_ += ret;
                if (!len_)
                {
                    return this->release_and_exit();
                }
                else
                {
                    return this->again();
                }
            }
            else
            {
                // error!
                DIE("Error writing.");
            }
        }
 
    private:
        uint8_t *bufferPos_;
        ssize_t len_;
    };
 
protected:
    int fd_;
    HFlow *hub_;
    ReadFlow readFlow_;
    WriteFlow writeFlow_;
};
 
#endif // OPENMRN_FEATURE_FD_CAN_DEVICE
#endif // _UTILS_HUBDEVICENONBLOCK_HXX_