HubDevice.hxx

Go to the documentation of this file.

 
#ifndef _UTILS_HUBDEVICE_HXX_
#define _UTILS_HUBDEVICE_HXX_
 
#include <unistd.h>
 
#include "openmrn_features.h"
#include "utils/Hub.hxx"
#include "executor/SemaphoreNotifiableBlock.hxx"
 
template <class Data> class FdHubWriteFlow;
 
class FdHubPortBase : public FdHubPortInterface, private Atomic
{
public:
#ifdef ESP_PLATFORM
    // TODO: shrink these if possible
    static const int kWriteThreadStackSize = 2048;
    static const int kReadThreadStackSize = 2048;
#else
    static const int kWriteThreadStackSize = 1000;
    static const int kReadThreadStackSize = 1000;
#endif // ESP_PLATFORM
 
    FdHubPortBase(int fd, Notifiable *done)
        : FdHubPortInterface(fd)
        , writeThread_(fill_thread_name('W', fd), 3, kWriteThreadStackSize)
        , writeService_(&writeThread_)
        , barrier_(done)
        , hasError_(0)
        , writeExitEnqueued_(0)
    {
        barrier_.new_child();
    }
 
    virtual ~FdHubPortBase()
    {
    }
 
    static void fill_thread_name(char *buf, char mode, int fd);
 
protected:
    const char *fill_thread_name(char mode, int fd) {
        fill_thread_name(threadName_, mode, fd);
        return threadName_;
    }
 
    virtual void unregister_write_port() = 0;
 
    void report_error()
    {
        {
            AtomicHolder h(this);
            if (hasError_)
            {
                return;
            }
            else
            {
                hasError_ = 1;
            }
        }
        ::close(fd_);
        unregister_write_port();
    }
 
    class ReadThreadBase : public OSThread
    {
    public:
        ReadThreadBase(FdHubPortBase *port) : port_(port)
        {
        }
 
        virtual int unit() = 0;
        virtual int buf_size() = 0;
        virtual void send_message(const void *buf, int size) = 0;
        void *entry() OVERRIDE
        {
            AutoNotify bn(&port_->barrier_);
            uint8_t buf[buf_size()];
            while (true)
            {
                int done = 0;
                while (done < unit())
                {
                    {
                        AtomicHolder h(port_);
                        if (port_->hasError_)
                        {
                            return NULL;
                        }
                    }
                    ssize_t ret =
                        ::read(port_->fd_, &buf[done], buf_size() - done);
                    if (ret > 0)
                    {
                        done += ret;
                        continue;
                    }
                    if ((ret < 0) && (errno == EINTR || errno == EAGAIN))
                    {
                        continue;
                    }
// Now: we have an error.
#if OPENMRN_FEATURE_BSD_SOCKETS_REPORT_EOF_ERROR
                    if (!ret)
                    {
                        LOG_ERROR("EOF reading fd %d", port_->fd_);
                    }
                    else
                    {
                        LOG_ERROR("Error reading fd %d: (%d) %s", port_->fd_,
                            errno, strerror(errno));
                    }
#endif // OPENMRN_FEATURE_BSD_SOCKETS_REPORT_EOF_ERROR
                    port_->report_error();
                    return NULL;
                }
                send_message(buf, done);
            }
        }
 
    protected:
        FdHubPortBase *port_;
    };
 
protected:
    template <class Data> friend class FdHubWriteFlow;
 
    char threadName_[30];
    Executor<1> writeThread_;
    Service writeService_;
    BarrierNotifiable barrier_;
    unsigned hasError_ : 1;
    unsigned writeExitEnqueued_ : 1;
};
 
template <class Data>
class FdHubWriteFlow : public StateFlow<Buffer<Data>, QList<1>>
{
public:
    FdHubWriteFlow(FdHubPortBase *parent)
        : StateFlow<Buffer<Data>, QList<1>>(&parent->writeService_)
        , port_(parent)
    {
    }
 
    StateFlowBase::Action entry() OVERRIDE
    {
        const uint8_t *buf =
            reinterpret_cast<const uint8_t *>(this->message()->data()->data());
        size_t size = this->message()->data()->size();
        while (size > 0)
        {
            {
                AtomicHolder h(port_);
                if (port_->hasError_)
                {
                    if (port_->writeExitEnqueued_ && this->queue_empty())
                    {
                        StateFlowBase::Action a = this->set_terminated();
                        this->release();
                        return a;
                    }
                    return this->release_and_exit();
                }
            }
            ssize_t ret = ::write(port_->fd_, buf, size);
            if (ret > 0)
            {
                size -= ret;
                buf += ret;
                continue;
            }
// now: we have an error.
#if OPENMRN_FEATURE_BSD_SOCKETS_REPORT_EOF_ERROR
            if (!ret)
            {
                LOG_ERROR("EOF writing fd %d", port_->fd_);
            }
            else
            {
                LOG_ERROR("Error writing fd %d: (%d) %s", port_->fd_, errno,
                    strerror(errno));
            }
#endif // OPENMRN_FEATURE_BSD_SOCKETS_REPORT_EOF_ERROR
            port_->report_error();
            break;
        }
        return this->release_and_exit();
    }
 
    FdHubPortBase *port_;
};
 
template <class HFlow> class FdHubPort : public FdHubPortBase
{
public:
    FdHubPort(HFlow *hub, int fd, Notifiable *done)
        : FdHubPortBase(fd, done)
        , hub_(hub)
        , writeFlow_(this)
        , readThread_(this)
    {
        hub_->register_port(&writeFlow_);
    }
 
    ~FdHubPort() OVERRIDE
    {
        writeThread_.shutdown();
    }
 
    void unregister_write_port() OVERRIDE
    {
        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);
    }
 
    class ReadThread : public ReadThreadBase
    {
    public:
        ReadThread(FdHubPort<HFlow> *port) : ReadThreadBase(port)
        {
            init();
            start(port->fill_thread_name('R', port->fd_), 0,
                  port->kReadThreadStackSize);
        }
        
        ~ReadThread() {
            delete semaphores_;
        }
 
        FdHubPort<HFlow> *port()
        {
            return static_cast<FdHubPort<HFlow> *>(port_);
        }
 
        int unit() OVERRIDE
        {
            return kUnit;
        }
        int buf_size() OVERRIDE
        {
            return kBufSize;
        }
        void send_message(const void *buf, int size) OVERRIDE;
 
    private:
        void init();
        SemaphoreNotifiableBlock* semaphores_{nullptr};
 
        static const int kUnit;
        static const int kBufSize;
    };
 
private:
    friend class ReadThread;
 
    HFlow *hub_;
    FdHubWriteFlow<typename HFlow::value_type> writeFlow_;
    ReadThread readThread_;
};
 
#endif // _UTILS_HUBDEVICE_HXX_