ioasync_server.h

Go to the documentation of this file.

// Evo C++ Library
/* Copyright 2019 Justin Crowell
Distributed under the BSD 2-Clause License -- see included file LICENSE.txt for details.
*/

#pragma once
#ifndef INCL_evo_ioasync_server_h
#define INCL_evo_ioasync_server_h

#include "ioasync_base.h"

namespace evo {


template<class T=AsyncBuffers>
class AsyncServerReplyT {
public:
    typedef T OutBuffer;                
    typedef AsyncServerReplyT<T> This;  

    enum WriterFlags {
        wfNONE = 0,
        wfDEFERRED = 0x01,      
        wfDEFERRED_LAST = 0x03  
    };

    struct Writer : public OutBuffer::BulkWrite {
        using OutBuffer::BulkWrite::init;

        Writer(This& parent, ulong id, SizeT buf_size, WriterFlags flags=wfNONE) {
            if (int(flags) & wfDEFERRED) {
                // Deferred
                parent.send_end();
                if (id == parent.next_id_) {
                    // This is the current reply
                    init(parent.buf_, buf_size);
                    if (flags == wfDEFERRED_LAST)
                        ++parent.next_id_;
                } else {
                    // Either append to existing reply in queue, or insert new reply -- keep queue ordered by ID
                    ReplyItem* rsp;
                    parent.deferred_get_queue_item(rsp, id);
                    init(rsp->data, buf_size);
                }
            } else if (id == parent.prev_id_) {
                // Append to previous send
                if (parent.prev_ == NULL)
                    init(parent.buf_, buf_size);
                else
                    init(parent.prev_->data, buf_size);
            } else {
                // New send
                parent.send_end();
                if (id == parent.next_id_) {
                    init(parent.buf_, buf_size);
                    ++parent.next_id_;
                    parent.prev_ = NULL;
                } else {
                    parent.prev_ = parent.queue_.addnew().advLast();
                    parent.prev_->id = id;
                    init(parent.prev_->data, buf_size);
                }
                parent.prev_id_ = id;
            }
        }
    };

    AsyncServerReplyT(T& bufs) : buf_(bufs), deferred_count_(0), gen_id_(1), next_id_(1), prev_id_(0), prev_(NULL) {
    }

    ulong gen_id() {
        return gen_id_++;
    }

    ulong deferred_active() const {
        return deferred_count_;
    }

    template<class U>
    void deferred_start(U& context) {
        ++deferred_count_;
        context.addref();
    }

    template<class U>
    bool deferred_end(U& context) {
        --deferred_count_;
        send_end();
        return context.endref();
    }

    This& deferred_send(ulong id, String& data, bool last) {
        send_end();
        if (id == next_id_) {
            // This is the current reply
            buf_.write(data.data(), data.size());
            if (last)
                ++next_id_;
        } else {
            // Either append to existing reply in queue, or insert new reply -- keep queue ordered by ID
            ReplyItem* rsp;
            if (deferred_get_queue_item(rsp, id))
                rsp->data.add(data);
            else
                rsp->data = data;
        }
        return *this;
    }

    void send(ulong id, String& data) {
        if (id == prev_id_) {
            // Append to previous send
            if (prev_ == NULL)
                buf_.write(data.data(), data.size());
            else
                prev_->data.add(data);
        } else {
            // New send
            send_end();
            if (id == next_id_) {
                buf_.write(data.data(), data.size());
                ++next_id_;
            } else {
                prev_ = queue_.addnew().advLast();
                prev_->id = id;
                prev_->data = data;
            }
            prev_id_ = id;
        }
    }

    void send_end() {
        ReplyItem* rsp;
        while ((rsp = queue_.advFirst()) != NULL && rsp->id == next_id_) {
            buf_.write(rsp->data.data(), rsp->data.size());
            queue_.popq();
            ++next_id_;
        }
    }

    void nosend(ulong id) {
        if (id + 1 == gen_id_)
            --gen_id_;
    }

private:
    // Disable copying
    AsyncServerReplyT(const This&);
    This& operator=(const This&);

    // Reply item in queue
    struct ReplyItem {
        ulong id;
        String data;

        ReplyItem() : id(0) {
        }
        ReplyItem(const ReplyItem& src) : id(src.id), data(src.data) {
        }
        ReplyItem& operator=(const ReplyItem& src) {
            id = src.id;
            data = src.data;
            return *this;
        }
    };

    OutBuffer& buf_;        // write buffer
    ulong deferred_count_;  // deferred replies in progress, 0 for none
    ulong gen_id_;          // next ID to generate
    ulong next_id_;         // next ID to send
    List<ReplyItem> queue_; // reply queue, only used while replies are out of order
    ulong prev_id_;         // previously queued/sent request ID -- not used by deferred_send()
    ReplyItem* prev_;       // previously queued item, NULL if none -- not used by deferred_send()

    bool deferred_get_queue_item(ReplyItem*& rsp, ulong id) {
        SizeT i = 0, sz = queue_.size();
        if (sz > 0 && id > queue_[sz-1].id) {
            i = sz; // ID not in queue, insert at end
        } else {
            for (; i < sz; ++i) {
                rsp = &queue_.advItem(i);
                if (id < rsp->id)
                    break; // ID not in queue, insert here
                if (id == rsp->id) {
                    // Use existing reply in queue
                    return true;
                }
            }
        }

        // Insert new reply in queue
        rsp = &queue_.advItem(queue_.insertnew(i));
        rsp->id = id;
        return false;
    }
};

typedef AsyncServerReplyT<> AsyncServerReply;   


struct AsyncServerHandler {
    template<class T>
    struct DeferredContextT {
        typedef DeferredContextT<T> Context;    

        struct ReplyBase {
            Context& context;   
            ulong id;           

            ReplyBase(Context& context, ulong id) : context(context), id(id), finished(false) {
                context.handler->reply.deferred_start(context);
            }

            ~ReplyBase() {
                if (!finished && context.handler != NULL)
                    context.handler->logger.log(LOG_LEVEL_ERROR, "AsyncServer DeferredReply left unfinished");
            }

        protected:
            bool finished;      
        };

        typedef T Handler;  
        Handler* handler;   

        DeferredContextT(Handler& handler) : handler(&handler), refcount(1) {
        }

        ulong count() const {
            return refcount;
        }

        void addref() {
            ++refcount;
        }

        bool endref() {
            if (--refcount == 0) {
                assert( handler == NULL ); // failure here indicates deferred start/end call mismatch (shouldn't happen)
                delete this;
                return true;
            }
            return false;
        }

        bool detach() {
            handler = NULL;
            return endref();
        }

    private:
        ulong refcount;
    };

    enum ResponseType {
        rtNORMAL,       
        rtDEFERRED,     
        rtHANDLED,      
        rtCLOSE         
    };

    template<class T>
    struct ResponseResult {
        typedef T ResultType;   

        ResponseType type;  
        ResultType result;  

        ResponseResult() : type(rtCLOSE), result((ResultType)0) {
        }

        ResponseResult(const ResponseResult& src) : type(src.type), result(src.result) {
        }

        ResponseResult(ResponseType type) : type(type), result((ResultType)0) {
        }

        ResponseResult(ResultType result) : type(rtNORMAL), result(result) {
        }

        ResponseResult& operator=(ResponseType newtype) {
            type = newtype;
            result = (T)0;
            return *this;
        }

        ResponseResult& operator=(ResultType newresult) {
            type = rtNORMAL;
            result = newresult;
            return *this;
        }

        ResponseResult& operator=(const ResponseResult& src) {
            type = src.type;
            result = src.result;
            return * this;
        }
    };

    static const size_t MAX_INITIAL_READ = 8192;   // 8 KB

    struct Global {
    };

    template<class T=Global>
    struct SimpleSharedBase {
        typedef T Global;   

        bool on_init(AsyncBase& server, Global& global) {
            EVO_PARAM_UNUSED(server);
            EVO_PARAM_UNUSED(global);
            return true;
        }

        void on_uninit() {
        }
    };

    struct Shared : SimpleSharedBase<> {
    };

    AsyncBuffers buffers;   
    AsyncServerReply reply; 
    ulong id;               

    AsyncServerHandler() : reply(buffers), id(0) {
    }

    void set_id() {
        id = reply.gen_id();
    }
};


template<class T>
class AsyncServer : public AsyncBase {
public:
    typedef T ProtocolServer;
    typedef AsyncServer<ProtocolServer> This;
    typedef typename ProtocolServer::Handler::DeferredContext DeferredContext;
    typedef typename ProtocolServer::Handler::Global Global;
    typedef typename ProtocolServer::Handler::Shared Shared;

    struct Stats {
        ulong active_connections;
        ulong accept_ok;
        ulong accept_err;
        ulong event_err;
        ulong reads;

        Stats() {
            active_connections = 0;
            accept_ok  = 0;
            accept_err = 0;
            event_err  = 0;
            reads      = 0;
        }
    };

    AsyncServer() : last_id_(0) {
        init();
    }

    Global& get_global() {
        return global_;
    }

    bool run(IoSocket::Handle listener) {
        {
            IoSocket listener_socket(listener);
            Error err = listener_socket.set_nonblock();
            if (err != ENone) {
                if (logger.check(LOG_LEVEL_ALERT)) {
                    String msg("AsyncServer listener error setting as non-blocking: ");
                    IoSocket::errormsg_out(msg, err);
                    logger.log_direct(LOG_LEVEL_ALERT, msg);
                }
                return false;
            }
            listener_socket.detach();
        }

        struct event* ev = ::event_new(evloop_->handle(), listener, EV_READ | EV_PERSIST, on_listener_ready, this);
        if (ev == NULL) {
            logger.log(LOG_LEVEL_ALERT, "AsyncServer libevent event_new() failed on listener -- this shouldn't happen");
            return false;
        }

        if (::event_add(ev, NULL) != 0) {
            logger.log(LOG_LEVEL_ALERT, "AsyncServer libevent event_add() failed on listener -- this shouldn't happen");
            return false;
        }

        if (!shared_.on_init(*this, global_)) {
            logger.log(LOG_LEVEL_ALERT, "AsyncServer Shared on_init() returned an error, indicating a bad configuration");
            return false;
        } else
            logger.log(LOG_LEVEL_DEBUG, "AsyncServer ready");

        run_eventloop();
        ::event_free(ev);
        return true;
    }

    bool run(Socket& listener) {
        return run(listener.device().handle);
    }

    void shutdown() {
        evloop_->shutdown();
    }

private:
    Global global_;
    Shared shared_;
    Stats stats_;
    ulong last_id_;

    using AsyncBase::runlocal;

    struct Connection {
        This& server;                       
        DeferredContext* deferred_context;  
        ProtocolServer   protocol_server;   
        struct bufferevent* bev;            
        SizeT  read_fixed_size_;            
        ulong id;                           

        Connection(This& async_server, struct bufferevent* bev, ulong id) :
                server(async_server), protocol_server(async_server.global_, async_server.shared_, async_server.logger.ptr), bev(bev), read_fixed_size_(0), id(id) {
            deferred_context = new DeferredContext(protocol_server.handler);

            ::bufferevent_setcb(bev, on_read, NULL, on_error, this);
            ::bufferevent_setwatermark(bev, EV_READ, T::MIN_INITIAL_READ, T::Handler::MAX_INITIAL_READ);
            if (server.read_timeout_ms_ > 0 || server.write_timeout_ms_ > 0) {
                struct timeval read_timeout, write_timeout;
                const int result = ::bufferevent_set_timeouts(bev,
                    get_timeout_ptr(read_timeout, server.read_timeout_ms_),
                    get_timeout_ptr(write_timeout, server.write_timeout_ms_)
                );
                if (result != 0)
                    server.logger.log(LOG_LEVEL_ERROR, "AsyncServer libevent bufferevent_set_timeouts() returned an error -- this shouldn't happen");
            }
        }

        ~Connection() {
            ::bufferevent_free(bev);
            --server.stats_.active_connections;
            if (!deferred_context->detach())
                server.logger.log(LOG_LEVEL_DEBUG_LOW, "AsyncServer cleanup, deferred pending");
        }

        bool enable() {
            if (::bufferevent_enable(bev, EV_READ | EV_WRITE) != 0) {
                server.logger.log(LOG_LEVEL_ALERT, "AsyncServer libevent bufferevent_enable() returned an error -- this shouldn't happen");
                return false;
            }
            return true;
        }
    };

    static void on_listener_ready(evutil_socket_t listener, short event, void* self_ptr) {
        EVO_PARAM_UNUSED(event);
        This& self = *(This*)self_ptr;
        Error err;
        IoSocket listener_socket(listener), client_socket;
        if (!listener_socket.accept_nonblock(err, client_socket)) {
            if (self.logger.check(LOG_LEVEL_ALERT)) {
                String msg;
                msg = "AsyncServer socket accept failed: ";
                IoSocket::errormsg_out(msg, err);
                self.logger.log_direct(LOG_LEVEL_ALERT, msg);
            }
            ++self.stats_.accept_err;
            return;
        }
        listener_socket.detach();

        struct bufferevent* bev = ::bufferevent_socket_new(self.evloop_->handle(), client_socket.detach(), BEV_OPT_CLOSE_ON_FREE);
        if (bev == NULL) {
            self.logger.log(LOG_LEVEL_ALERT, "AsyncServer libevent bufferevent_socket_new() returned an error -- this shouldn't happen");
            ++self.stats_.accept_err;
            return;
        }

        // Connection takes ownership of bev, and is freed by on_read() or on_error() (when connection is closed)
        Connection* conn = new Connection(self, bev, ++self.last_id_);
        ++self.stats_.active_connections;
        if (!conn->enable()) {
            delete conn; // failed to enable connection event handling
            ++self.stats_.accept_err;
        } else
            ++self.stats_.accept_ok;
    }

    static void on_read(struct bufferevent* bev, void* conn_ptr) {
        Connection* conn = (Connection*)conn_ptr;
        LoggerPtr<>& logger = conn->server.logger;
        ++conn->server.stats_.reads;
        conn->protocol_server.handler.buffers.attach(bev);

        String logstr;
        AsyncBuffers* bufs = &conn->protocol_server.handler.buffers;
        if (conn->read_fixed_size_ > 0) {
            if (logger.check(LOG_LEVEL_DEBUG_LOW))
                logger.log_direct(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(72) << "AsyncServer connection " << conn->id << " fixed read: " << conn->read_fixed_size_);
            for (;;) {
                SubString data;
                if (!bufs->read_fixed(data, conn->read_fixed_size_))
                    return; // wait for more data
                conn->read_fixed_size_ = 0;
                if (!conn->protocol_server.on_read_fixed(conn->read_fixed_size_, data, conn->deferred_context)) {
                    delete conn;
                    if (logger.check(LOG_LEVEL_DEBUG_LOW))
                        logger.log(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(72) << "AsyncServer connection " << conn->id << " on_read_fixed() returned false to close");
                    return;
                }
                bufs->read_flush();
                if (conn->read_fixed_size_ <= 0)
                    break;
            }
            bufs->read_reset(ProtocolServer::MIN_INITIAL_READ, ProtocolServer::Handler::MAX_INITIAL_READ);
            if (bufs->read_size() == 0)
                return;
        }

        if (logger.check(LOG_LEVEL_DEBUG_LOW))
            logger.log_direct(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(72) << "AsyncServer connection " << conn->id << " read: " << bufs->read_size());
        if (!conn->protocol_server.on_read(conn->read_fixed_size_, *bufs, conn->deferred_context)) {
            delete conn;
            if (logger.check(LOG_LEVEL_DEBUG_LOW))
                logger.log_direct(LOG_LEVEL_DEBUG_LOW, logstr.set().reserve(72) << "AsyncServer connection " << conn->id << " on_read() returned false to close");
        }
    }

    static void on_error(struct bufferevent* bev, short error, void* conn_ptr) {
        EVO_PARAM_UNUSED(bev);
        Connection* conn = (Connection*)conn_ptr;
        AsyncError err;
        if (error & BEV_EVENT_EOF)
            err = aeCLOSED;
        else if (error & BEV_EVENT_TIMEOUT)
            err = aeTIMEOUT;
        else if (error & BEV_EVENT_READING)
            err = aeIO_READ;
        else if (error & BEV_EVENT_WRITING)
            err = aeIO_WRITE;
        else
            err = aeIO;
        ++conn->server.stats_.event_err;
        conn->protocol_server.on_error(err);

        const ulong conn_id = conn->id;
        LoggerPtr<>& logger = conn->server.logger;
        delete conn;

        if (logger.check(LOG_LEVEL_DEBUG_LOW)) {
            const SubString errmsg(async_error_msg(err));
            logger.log_direct(LOG_LEVEL_DEBUG, String().reserve(48 + errmsg.size()) << "AsyncServer connection " << conn_id << " error: " << errmsg);
        }
    }
};


}
#endif