ioasync_base.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_base_h
#define INCL_evo_ioasync_base_h

#include "iosock.h"
#include "event.h"
#include "thread.h"
#include "string.h"
#include "substring.h"
#include "atomic_buffer_queue.h"
#include "logger.h"

// Requires libevent 2.0+
#if !defined(LIBEVENT_VERSION_NUMBER)
    #include <event2/event.h>
    #include <event2/buffer.h>
    #include <event2/bufferevent.h>
    #include <event2/thread.h>
#endif
#define EVO_LIBEVENT_MINVER 0x02000000
#if LIBEVENT_VERSION_NUMBER < EVO_LIBEVENT_MINVER
    #error Evo Async I/O requires libevent 2.0+
#endif
#if !defined(EVO_ASYNC_MULTI_THREAD)    

    #define EVO_ASYNC_MULTI_THREAD 0
#endif

#if defined(EVO_MSVC_YEAR)
    #pragma comment(lib, "libevent_core.lib")
    #pragma comment(lib, "advapi32.lib")
#endif

namespace evo {


class AsyncEventLoop {
public:
    typedef struct event_base* Handle;

    AsyncEventLoop() {
        static FirstInitHelper first_init_helper;
        evbase_ = ::event_base_new();
        if (evbase_ == NULL)
            abort(); // This shouldn't happen
    }

    ~AsyncEventLoop() {
        ::event_base_free(evbase_);
    }

    Handle handle()
        { return evbase_; }

    bool active() const {
        return !shutdown_.load(EVO_ATOMIC_ACQUIRE);
    }

    bool run1() {
        int result = ::event_base_loop(evbase_, EVLOOP_ONCE);
        if (result < 0)
            return false;
        else if (result > 0)
            shutdown_.store(1, EVO_ATOMIC_RELEASE);
        return true;
    }

    void shutdown() {
        if (shutdown_.compare_set(0, 1, EVO_ATOMIC_ACQ_REL, EVO_ATOMIC_ACQUIRE)) {
            if (::event_base_loopexit(evbase_, NULL) != 0)
                abort(); // This should never happen
        }
    }

private:
    struct FirstInitHelper {
        FirstInitHelper() {
            static Mutex mutex;
            Mutex::Lock lock(mutex);
            if (::event_get_version_number() < EVO_LIBEVENT_MINVER)
                abort(); // Abort if libevent is too old
        #if EVO_ASYNC_MULTI_THREAD
            int result =
            #if defined(_WIN32)
                ::evthread_use_windows_threads();
            #else
                ::evthread_use_pthreads();
            #endif
            assert( result == 0 );
            if (result != 0)
                abort(); // Abort if can't initialize libevent for multithreaded
        #endif
        }
    };

    Handle evbase_;
    AtomicInt shutdown_;
};


class AsyncBuffers {
public:
    class BulkWrite {
    public:
        BulkWrite() {
            parent_ = NULL;
            ptr_ = end_ = NULL;
        }

        BulkWrite(String& str, SizeT size) {
            init(str, size);
        }

        BulkWrite(AsyncBuffers& parent, size_t size) {
            init(parent, size);
        }

        ~BulkWrite() {
            assert( ptr_ == end_ );
        }

        BulkWrite& init(String& str, SizeT size) {
            const SizeT used = str.used();
            ptr_ = str.advBuffer(used + size) + used;
            end_ = ptr_ + size;
            str.advSize(size);
            return *this;
        }

        BulkWrite& init(AsyncBuffers& parent, size_t size) {
            if (parent.output_ != NULL) {
                int result = ::evbuffer_reserve_space(parent.output_, size, &data_, 1);
                if (result != 1)
                    abort(); // This should never happen
                assert( data_.iov_base != NULL );
                assert( data_.iov_len >= size );
                data_.iov_len = size;
                ptr_ = (char*)data_.iov_base;
                end_ = ptr_ + size;
                parent_ = &parent;
            } else {
                parent_ = NULL;
                ptr_ = end_ = NULL;
                abort(); // This should never happen
            }
            return *this;
        }

        bool error() const {
            return (ptr_ == NULL);
        }

        char* ptr() {
            return ptr_;
        }

        BulkWrite& addsize(size_t size) {
            ptr_ += size;
            assert( ptr_ <= end_ );
            if (parent_ != NULL && ptr_ >= end_) {
                int result = ::evbuffer_commit_space(parent_->output_, &data_, 1);
                if (result != 0)
                    abort(); // This should never happen
                ptr_ = end_ = NULL;
            }
            return *this;
        }

        BulkWrite& add(const char* data, size_t size) {
            assert( ptr_ != NULL );
            memcpy(ptr_, data, size);
            addsize(size);
            return *this;
        }

        BulkWrite& add(char ch) {
            assert( ptr_ != NULL );
            *ptr_ = ch;
            addsize(1);
            return *this;
        }

    private:
        BulkWrite(const BulkWrite&);
        BulkWrite& operator=(const BulkWrite&);

        AsyncBuffers*         parent_;
        struct evbuffer_iovec data_;
        char*                 ptr_;
        char*                 end_;
    };

    AsyncBuffers() {
        bev_    = NULL;
        input_  = NULL;
        output_ = NULL;
        read_offset_ = 0;
    }

    void reset() {
        bev_    = NULL;
        input_  = NULL;
        output_ = NULL;
        read_offset_ = 0;
    }

    void attach(struct bufferevent* bev) {
        bev_    = bev;
        input_  = ::bufferevent_get_input(bev);
        output_ = ::bufferevent_get_output(bev);
    }

    void attach_write(struct bufferevent* bev) {
        bev_    = bev;
        output_ = ::bufferevent_get_output(bev);
    }

    void attach_read() {
        input_ = ::bufferevent_get_input(bev_);
    }

    size_t read_size()
        { return ::evbuffer_get_length(input_); }

    bool read_fixed(SubString& data, SizeT size, SizeT max_size=0) {
        assert( max_size == 0 || max_size >= size );
        if (::evbuffer_get_length(input_) < size) {
            ::bufferevent_setwatermark(bev_, EV_READ, size, max_size);
            return false;
        }
        read_offset_ = size;
        data.set((char*)::evbuffer_pullup(input_, size), size);
        return true;
    }

    template<class T>
    bool read_fixed_helper(T& parent, SizeT& fixed_size, SizeT size, SizeT max_size=0, void* context=NULL) {
        for (;;) {
            SubString data;
            if (!read_fixed(data, size, max_size)) {
                fixed_size = size;
                return true; // wait for more data
            }
            fixed_size = 0;
            if (!parent.on_read_fixed(fixed_size, data, context))
                return false;
            read_flush();
            if (fixed_size <= 0)
                break;
        }
        return true;
    }

    void read_reset(size_t max_size, size_t min_size=0) {
        ::bufferevent_setwatermark(bev_, EV_READ, min_size, max_size);
    }

    bool read_line(SubString& data) {
        size_t block_len = ::evbuffer_get_contiguous_space(input_);
        if (block_len > 0) {
            size_t tmp_len;
            for (;;) {
                char* ptr = (char*)::evbuffer_pullup(input_, block_len);
                while (read_offset_ < block_len) {
                    if (ptr[read_offset_] == '\n') {
                        assert( read_offset_ <= IntegerT<SizeT>::MAX );
                        if (read_offset_ > 0 && ptr[read_offset_-1] == '\r')
                            data.set(ptr, (SizeT)(read_offset_-1));
                        else
                            data.set(ptr, (SizeT)read_offset_);
                        ++read_offset_;
                        return true;
                    }
                    ++read_offset_;
                }
                if (block_len < (tmp_len=::evbuffer_get_length(input_))) {
                    block_len = tmp_len;
                } else
                    break;
            }
        }
        return false;
    }

    void read_flush() {
        if (read_offset_ > 0) {
            if (::evbuffer_drain(input_, read_offset_) != 0)
                abort(); // This should never happen
            read_offset_ = 0;
        }
    }

    size_t write_size() const
        { return (output_ == NULL ? 0 : ::evbuffer_get_length(output_)); }

    void write_clear() {
        if (output_ != NULL)
            ::evbuffer_drain(output_, ::evbuffer_get_length(output_));
    }

    void write_reserve(size_t size)
        { ::evbuffer_expand(output_, size); }

    void write(const char* data, size_t size)
        { ::evbuffer_add(output_, data, size); }

private:
    struct bufferevent* bev_;
    struct evbuffer* input_;
    struct evbuffer* output_;
    size_t read_offset_;
};


enum AsyncReadResult {
    arrERROR = 0,   
    arrDONE,        
    arrMORE         
};

enum AsyncError {
    aeNONE = 0,     
    aeCONNECT,      
    aeCLOSED,       
    aeIO,           
    aeIO_READ,      
    aeIO_WRITE,     
    aeTIMEOUT,      
    aeCLIENT        
};

inline const char* async_error_msg(AsyncError err) {
    if (err <= aeNONE || err > aeCLIENT)
        return "Unknown error";
    const char* MSG[] = {
        "Connection refused",
        "Socket closed by other side",
        "Unrecoverable I/O error",
        "Read error",
        "Write error",
        "Timed out",
        "Client protocol error"
    };
    return MSG[(int)err - 1];
}


class AsyncBase {
public:
    LoggerPtr<> logger; 

    struct OnTimer {
        struct event* timer_handle;     
        ulong timer_msec;               

        OnTimer() : timer_handle(NULL) {
        }

        virtual ~OnTimer() {
            ::event_free(timer_handle);
        }

        virtual void on_timer() = 0;

        bool timer_reset(ulong msec) {
            struct timeval tv;
            if (::event_add(timer_handle, get_timeout_ptr(tv, msec)) != 0) {
                ::event_free(timer_handle);
                return false;
            }
            timer_msec = msec;
            return true;
        }
    };

    AsyncBase() : parent_base_(NULL), child_base_(NULL), local_(true), evloop_(NULL), read_timeout_ms_(0), write_timeout_ms_(0) {
    }

    virtual ~AsyncBase() {
        if (parent_base_ == NULL) {
            if (child_base_ != NULL) {
                logger.log(LOG_LEVEL_ERROR, "AsyncBase internal cleanup error, parent destroyed before child");
                assert( false ); // this shouldn't happen
            }
            delete evloop_;
        } else {
            parent_base_->child_base_ = child_base_;
            if (child_base_ != NULL)
                child_base_->parent_base_ = parent_base_;
        }
    }

    virtual void set_logger(LoggerBase* newlogger) {
        logger.ptr = newlogger;
    }

    void set_timeout(ulong read_timeout_ms=0, ulong write_timeout_ms=0) {
        read_timeout_ms_  = read_timeout_ms;
        write_timeout_ms_ = write_timeout_ms;
    }

    bool set_timer(OnTimer& on_timer, ulong msec) {
        if (on_timer.timer_handle != NULL)
            ::event_free(on_timer.timer_handle);
        on_timer.timer_handle = ::event_new(evloop_->handle(), -1, 0, on_timer_event, &on_timer);
        if (on_timer.timer_handle == NULL)
            return false;
        return on_timer.timer_reset(msec);
    }

    bool runlocal() {
        if (parent_base_ != NULL || evloop_ == NULL)
            return false;
        if (evloop_->active() && !run_eventloop_once())
            return false;
        for (AsyncBase* p = this; p != NULL && evloop_->active(); p = p->child_base_)
            while (p->check_client_active())
                if (!run_eventloop_once())
                    return false;
        return true;
    }

protected:
    AsyncBase* parent_base_;    
    AsyncBase* child_base_;     
    bool local_;                

    AsyncEventLoop* evloop_;    
    ulong read_timeout_ms_;     
    ulong write_timeout_ms_;    

    void init() {
        if (evloop_ == NULL)
            evloop_ = new AsyncEventLoop;

    }

    void init_attach(AsyncBase& parent) {
        if (evloop_ == NULL && parent_base_ == NULL) {
            // Always attach at end of chain
            AsyncBase* p = &parent;
            while (p->child_base_ != NULL) {
                p = p->child_base_;
                assert( p->parent_base_ != NULL );
            }

            if (p->evloop_ == NULL) {
                // Init parent since this hasn't been done yet
                assert( p->parent_base_ == NULL );
                p->init();
            }

            parent_base_ = p;
            p->child_base_ = this;
            evloop_ = p->evloop_;
        }
    }

    virtual bool check_client_active() {
        return false;
    }

    bool run_eventloop_once() {
        if (parent_base_ != NULL || !evloop_->run1())
            return false;
        return true;
    }

    bool run_eventloop() {
        while (evloop_->active()) {
            if (!run_eventloop_once())
                return false;
        }
        return true;
    }

    static struct timeval* get_timeout_ptr(struct timeval& out, ulong ms) {
        if (ms > 0) {
            #if defined(_WIN32)
                SysWindows::set_timeval_ms(out, ms);
            #else
                SysLinux::set_timeval_ms(out, ms);
            #endif
            return &out;
        }
        return NULL;
    }

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

    static void on_timer_event(evutil_socket_t, short, void * arg) {
        ((OnTimer*)arg)->on_timer();
    }
};


}
#endif