DirectHub.cxx

Go to the documentation of this file.

 
// #define LOGLEVEL VERBOSE
 
#include "utils/DirectHub.hxx"
 
#include <algorithm>
#include <fcntl.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <vector>
 
#include "executor/AsyncNotifiableBlock.hxx"
#include "executor/StateFlow.hxx"
#include "nmranet_config.h"
#include "utils/logging.h"
#include "utils/socket_listener.hxx"
 
DataBufferPool g_direct_hub_data_pool(
    config_directhub_port_incoming_buffer_size());
DataBufferPool g_direct_hub_kbyte_pool(1024);
 
class DirectHubService : public Service
{
public:
    typedef Q QueueType;
 
    DirectHubService(ExecutorBase *e)
        : Service(e)
        , busy_(0)
    {
    }
 
    Atomic *lock()
    {
        return pendingSend_.lock();
    }
 
    void enqueue_caller(Executable *caller)
    {
        {
            AtomicHolder h(lock());
            if (busy_)
            {
                pendingSend_.insert_locked(caller);
                return;
            }
            busy_ = 1;
        }
        caller->run();
    }
 
    void on_done()
    {
        // De-queues the next entry.
        Result deq;
        {
            AtomicHolder h(lock());
            if (pendingSend_.empty())
            {
                busy_ = 0;
                return;
            }
            deq = pendingSend_.next();
        }
        // Schedules it on the executor.
        executor()->add(static_cast<Executable *>(deq.item), deq.index);
    }
 
    unsigned busy_ : 1;
    QueueType pendingSend_;
};
 
template <class T>
class DirectHubImpl : public DirectHubInterface<T>,
                      protected StateFlowBase,
                      private Atomic
{
public:
    DirectHubImpl(DirectHubService *service)
        : StateFlowBase(service)
    {
    }
 
    ~DirectHubImpl()
    {
        delete service();
    }
 
    Service *get_service() override
    {
        return service();
    }
 
    void register_port(DirectHubPort<T> *port) override
    {
        AtomicHolder h(this);
        ports_.push_back(port);
    }
 
    void unregister_port(DirectHubPort<T> *port) override
    {
        SyncNotifiable n;
        unregister_port(port, &n);
        n.wait_for_notification();
    }
 
    void unregister_port(DirectHubPort<T> *port, Notifiable *done) override
    {
        // By enqueueing on the service we ensure that the state flow is not
        // processing any packets while the code below is running.
        service()->enqueue_caller(new CallbackExecutable([this, port, done]() {
            {
                AtomicHolder h(this);
                ports_.erase(std::remove(ports_.begin(), ports_.end(), port),
                    ports_.end());
            }
            done->notify();
            service()->on_done();
        }));
    }
 
    void enqueue_send(Executable *caller) override
    {
        service()->enqueue_caller(caller);
    }
 
    MessageAccessor<T> *mutable_message() override
    {
        return &msg_;
    }
 
    void do_send() override
    {
        unsigned next_port = 0;
        while (true)
        {
            DirectHubPort<T> *p;
            {
 
                AtomicHolder h(this);
                if (next_port >= ports_.size())
                {
                    break;
                }
                p = ports_[next_port];
                ++next_port;
            }
            if (should_send_to(p))
            {
                p->send(&msg_);
            }
        }
        msg_.clear();
        service()->on_done();
    }
 
    bool should_send_to(DirectHubPort<T> *p)
    {
        return static_cast<HubSource *>(p) != msg_.source_;
    }
 
private:
    DirectHubService *service()
    {
        return static_cast<DirectHubService *>(StateFlowBase::service());
    }
 
    std::vector<DirectHubPort<T> *> ports_;
 
    MessageAccessor<T> msg_;
}; // class DirectHubImpl
 
DirectHubInterface<uint8_t[]> *create_hub(ExecutorBase *e)
{
    auto *s = new DirectHubService(e);
    auto *dh = new DirectHubImpl<uint8_t[]>(s);
    return dh;
}
 
class DirectHubPortSelect : public DirectHubPort<uint8_t[]>,
                            private StateFlowBase
{
private:
    class DirectHubReadFlow : public StateFlowBase
    {
    public:
        DirectHubReadFlow(DirectHubPortSelect *parent,
            std::unique_ptr<MessageSegmenter> segmenter)
            : StateFlowBase(parent->service())
            , parent_(parent)
            , segmenter_(std::move(segmenter))
        {
            segmenter_->clear();
        }
 
        void start()
        {
            start_flow(STATE(alloc_for_read));
        }
 
        void read_shutdown()
        {
            auto *e = this->service()->executor();
            if (e->is_selected(&helper_))
            {
                // We're waiting in select on reads, we can cancel right now.
                e->unselect(&helper_);
                set_terminated();
                buf_.reset();
                parent_->read_flow_exit();
            }
            // Else we're waiting for the regular progress to wake up the
            // flow. It will check fd_ < 0 to exit.
        }
 
    private:
        Action alloc_for_read()
        {
            QMember *bn = pendingLimiterPool_.next().item;
            if (bn)
            {
                bufferNotifiable_ = pendingLimiterPool_.initialize(bn);
                return get_read_buffer();
            }
            else
            {
                pendingLimiterPool_.next_async(this);
                return wait_and_call(STATE(barrier_allocated));
            }
        }
 
        Action barrier_allocated()
        {
            QMember *bn;
            cast_allocation_result(&bn);
            HASSERT(bn);
            bufferNotifiable_ = pendingLimiterPool_.initialize(bn);
            return get_read_buffer();
        }
 
        Action get_read_buffer()
        {
            DataBuffer *p;
            LOG(VERBOSE, "read flow %p (fd %d): notif %p alloc() %u", this,
                parent_->fd_, (BarrierNotifiable *)bufferNotifiable_,
                (unsigned)mainBufferPool->total_size());
            // Since there is a limit on how many bufferNotifiable_'s can be,
            // and they are uniquely assigned to the buffers, we know that this
            // synchronous allocation can only happen for a few buffers
            // only. The buffers will get recycled through the main buffer pool
            // exactly at the time when the bufferNotifiable_ comes back to the
            // pendingLimiterPool_.
            g_direct_hub_data_pool.alloc(&p);
            if (buf_.head())
            {
                buf_.append_empty_buffer(p);
            }
            else
            {
                buf_.reset(p);
            }
            p->set_done(bufferNotifiable_);
            bufferNotifiable_ = nullptr;
            return do_some_read();
        }
 
        Action do_some_read()
        {
            if (parent_->fd_ < 0)
            {
                // Socket closed, terminate and exit.
                set_terminated();
                buf_.reset();
                parent_->read_flow_exit();
                return wait();
            }
            return read_single(&helper_, parent_->fd_,
                buf_.data_write_pointer(), buf_.free(), STATE(read_done));
        }
 
        Action read_done()
        {
            if (helper_.hasError_)
            {
                LOG(INFO, "%p: Error reading from fd %d: (%d) %s", parent_,
                    parent_->fd_, errno, strerror(errno));
                set_terminated();
                buf_.reset();
                parent_->report_read_error();
                return wait();
            }
            size_t bytes_arrived = buf_.free() - helper_.remaining_;
            segmentSize_ = segmenter_->segment_message(
                buf_.data_write_pointer(), bytes_arrived);
            buf_.data_write_advance(bytes_arrived);
            return eval_segment();
        }
 
        Action eval_segment()
        {
            if (segmentSize_ > 0)
            {
                // Complete message.
                segmenter_->clear();
                return call_immediately(STATE(send_prefix));
            }
            else
            {
                return incomplete_message();
            }
        }
 
        Action call_head_segmenter()
        {
            uint8_t *ptr;
            unsigned available;
            auto *n =
                buf_.head()->get_read_pointer(buf_.skip(), &ptr, &available);
            HASSERT(!n); // We must be at the tail.
            segmentSize_ = segmenter_->segment_message(ptr, available);
            return eval_segment();
        }
 
        Action incomplete_message()
        {
            if (!buf_.free())
            {
                return call_immediately(STATE(alloc_for_read));
            }
            return call_immediately(STATE(do_some_read));
        }
 
        Action send_prefix()
        {
            // 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;
            parent_->hub_->enqueue_send(this); // causes the callback
            inlineCall_ = 0;
            if (sendComplete_)
            {
                return send_done();
            }
            return wait();
        }
 
        Action send_callback()
        {
            auto *m = parent_->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_ = buf_.transfer_head(segmentSize_);
            parent_->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()
        {
            if (buf_.size())
            {
                // We still have unused data in the current buffer. We have to
                // segment that and send it to the hub.
                return call_head_segmenter();
            }
            if (buf_.free())
            {
                // We still have space in the current buffer. We can read more
                // data into that space.
                return do_some_read();
            }
            else
            {
                buf_.reset();
                return alloc_for_read();
            }
        }
 
        LinkedDataBufferPtr buf_;
        BarrierNotifiable *bufferNotifiable_;
        ssize_t segmentSize_;
        uint16_t inlineCall_ : 1;
        uint16_t sendComplete_ : 1;
        AsyncNotifiableBlock pendingLimiterPool_ {
            (unsigned)config_directhub_port_max_incoming_packets()};
        StateFlowSelectHelper helper_ {this};
        DirectHubPortSelect *parent_;
        std::unique_ptr<MessageSegmenter> segmenter_;
    } readFlow_;
 
    friend class DirectHubReadFlow;
 
public:
    DirectHubPortSelect(DirectHubInterface<uint8_t[]> *hub, int fd,
        std::unique_ptr<MessageSegmenter> segmenter,
        Notifiable *on_error = nullptr)
        : StateFlowBase(hub->get_service())
        , readFlow_(this, std::move(segmenter))
        , readFlowPending_(1)
        , writeFlowPending_(1)
        , hub_(hub)
        , fd_(fd)
        , onError_(on_error)
    {
#ifdef __WINNT__
        unsigned long par = 1;
        ioctlsocket(fd_, FIONBIO, &par);
#else
        ::fcntl(fd, F_SETFL, O_RDWR | O_NONBLOCK);
#endif
 
        // 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_ = 1;
 
        hub_->register_port(this);
        readFlow_.start();
        LOG(VERBOSE, "%p create fd %d", this, fd_);
    }
 
    ~DirectHubPortSelect()
    {
    }
 
    void send(MessageAccessor<uint8_t[]> *msg) override
    {
        if (fd_ < 0)
        {
            // Port already closed. Ignore data to send.
            return;
        }
        {
            AtomicHolder h(lock());
            if (pendingTail_ && pendingTail_->buf_.try_append_from(msg->buf_))
            {
                // 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 (fd_ < 0)
            {
                // 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;
            }
            else
            {
                // flow already running. Skip notify.
                return;
            }
        }
        notify();
    }
 
private:
    void shutdown()
    {
        HASSERT(fd_ < 0);
        {
            AtomicHolder h(lock());
            if (notRunning_)
            {
                // Queue is empty, waiting for new entries. There will be no new
                // entries because fd_ < 0.
                hub_->unregister_port(this, this);
                wait_and_call(STATE(report_and_exit));
            }
            // Else eventually we will get to check_for_new_message() which will
            // flush the queue, unregister the port and exit.
        }
    }
 
    Action read_queue()
    {
        BufferType *head;
        {
            AtomicHolder h(lock());
            head = static_cast<BufferType *>(pendingQueue_.next_locked().item);
            HASSERT(head);
            if (head->data() == pendingTail_)
            {
                pendingTail_ = nullptr;
            }
        }
        currentHead_.reset(head);
        nextToWrite_ = currentHead_->data()->buf_.head();
        nextToSkip_ = currentHead_->data()->buf_.skip();
        nextToSize_ = currentHead_->data()->buf_.size();
        return do_write();
    }
 
    Action do_write()
    {
        if (fd_ < 0)
        {
            // fd closed. Drop data to the floor.
            totalPendingSize_ -= nextToSize_;
            return check_for_new_message();
        }
        uint8_t *data;
        unsigned len;
        nextToWrite_ = nextToWrite_->get_read_pointer(nextToSkip_, &data, &len);
        if (len > nextToSize_)
        {
            len = nextToSize_;
        }
        nextToSkip_ = 0;
        nextToSize_ -= len;
        totalPendingSize_ -= len;
        totalWritten_ += len;
        LOG(VERBOSE, "write %u total %zu", (unsigned)len, totalWritten_);
        return write_repeated(
            &selectHelper_, fd_, data, len, STATE(write_done));
    }
 
    Action write_done()
    {
        if (selectHelper_.hasError_)
        {
            LOG(INFO, "%p: Error writing to fd %d: (%d) %s", this, fd_, errno,
                strerror(errno));
            // will close fd and notify the reader flow to exit.
            report_write_error();
            // Flushes the queue of messages. fd_ == -1 now so no write will be
            // attempted.
            return check_for_new_message();
        }
        if (nextToSize_)
        {
            return do_write();
        }
        return check_for_new_message();
    }
 
    Action check_for_new_message()
    {
        currentHead_.reset();
        AtomicHolder h(lock());
        if (pendingQueue_.empty())
        {
            if (fd_ < 0)
            {
                // unregisters the port. All the queue has been flushed now.
                hub_->unregister_port(this, this);
                return wait_and_call(STATE(report_and_exit));
            }
            notRunning_ = 1;
            return wait_and_call(STATE(read_queue));
        }
        else
        {
            return call_immediately(STATE(read_queue));
        }
    }
 
    Action report_and_exit()
    {
        set_terminated();
        currentHead_.reset();
        write_flow_exit();
        return wait();
    }
 
    void report_write_error()
    {
        int close_fd = -1;
        {
            AtomicHolder h(lock());
            if (fd_ >= 0)
            {
                std::swap(fd_, close_fd);
            }
        }
        if (close_fd >= 0)
        {
            ::close(close_fd);
        }
        readFlow_.read_shutdown();
    }
 
    void report_read_error()
    {
        int close_fd = -1;
        {
            AtomicHolder h(lock());
            if (fd_ >= 0)
            {
                std::swap(fd_, close_fd);
            }
        }
        if (close_fd >= 0)
        {
            ::close(close_fd);
        }
        // take read barrier
        read_flow_exit();
        // kill write flow
        shutdown();
    }
 
    void read_flow_exit()
    {
        LOG(VERBOSE, "%p exit read", this);
        flow_exit(true);
    }
 
    void write_flow_exit()
    {
        LOG(VERBOSE, "%p exit write", this);
        flow_exit(false);
    }
 
    void flow_exit(bool read)
    {
        bool del = false;
        {
            AtomicHolder h(lock());
            if (read)
            {
                readFlowPending_ = 0;
            }
            else
            {
                writeFlowPending_ = 0;
            }
            if (writeFlowPending_ == 0 && readFlowPending_ == 0)
            {
                del = true;
            }
        }
        if (del)
        {
            if (onError_)
            {
                onError_->notify();
            }
            delete this;
        }
    }
 
    Atomic *lock()
    {
        return pendingQueue_.lock();
    }
 
    struct OutputDataEntry
    {
        LinkedDataBufferPtr buf_;
    };
 
    friend class DirectHubReadFlow;
 
    typedef Buffer<OutputDataEntry> BufferType;
    typedef Q QueueType;
 
    size_t totalWritten_ {0};
 
    BufferPtr<OutputDataEntry> currentHead_;
    DataBuffer *nextToWrite_;
    unsigned nextToSkip_;
    unsigned nextToSize_;
    StateFlowSelectHelper selectHelper_ {this};
    // long long lastWriteTimeNsec_ = 0;
 
    QueueType pendingQueue_;
    OutputDataEntry *pendingTail_ = nullptr;
    size_t totalPendingSize_ = 0;
    uint8_t notRunning_ : 1;
    uint8_t readFlowPending_;
    uint8_t writeFlowPending_;
    DirectHubInterface<uint8_t[]> *hub_;
    int fd_;
    Notifiable *onError_ = nullptr;
};
 
extern DirectHubPortSelect *g_last_direct_hub_port;
DirectHubPortSelect *g_last_direct_hub_port = nullptr;
 
void create_port_for_fd(DirectHubInterface<uint8_t[]> *hub, int fd,
    std::unique_ptr<MessageSegmenter> segmenter, Notifiable *on_error)
{
    g_last_direct_hub_port =
        new DirectHubPortSelect(hub, fd, std::move(segmenter), on_error);
}
 
class DirectGcTcpHub
{
public:
    DirectGcTcpHub(DirectHubInterface<uint8_t[]> *gc_hub, int port);
    ~DirectGcTcpHub();
 
    bool is_started()
    {
        return tcpListener_.is_started();
    }
 
private:
    void OnNewConnection(int fd);
 
    DirectHubInterface<uint8_t[]> *gcHub_;
    SocketListener tcpListener_;
};
 
void DirectGcTcpHub::OnNewConnection(int fd)
{
#if 0    
    uint32_t rcvbuf;
    socklen_t len = sizeof(rcvbuf);
    int ret = getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len);
    if (ret >= 0)
    {
        LOG(ALWAYS, "Socket rcvbuf %u", (unsigned)rcvbuf);
    }
#endif    
    create_port_for_fd(gcHub_, fd,
        std::unique_ptr<MessageSegmenter>(create_gc_message_segmenter()));
}
 
DirectGcTcpHub::DirectGcTcpHub(DirectHubInterface<uint8_t[]> *gc_hub, int port)
    : gcHub_(gc_hub)
    , tcpListener_(port,
          std::bind(
              &DirectGcTcpHub::OnNewConnection, this, std::placeholders::_1))
{
}
 
DirectGcTcpHub::~DirectGcTcpHub()
{
    tcpListener_.shutdown();
}
 
void create_direct_gc_tcp_hub(DirectHubInterface<uint8_t[]> *hub, int port)
{
    new DirectGcTcpHub(hub, port);
}