ciosrv.h

/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

#ifndef CIOSRV_H
#define CIOSRV_H

#include <set>
#include <list>
#include <bitset>
#include <utility>
#include <map>
#include <algorithm>
#include <vector>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/select.h>
#include <errno.h>
#include <time.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <sys/stat.h>


#include "rofl/common/cmemory.h"
#include "rofl/common/logging.h"
#include "rofl/common/croflexception.h"
#include "rofl/common/thread_helper.h"
#include "rofl/common/cpipe.h"
#include "rofl/common/cevents.h"
#include "rofl/common/ctimers.h"
#include "rofl/common/ctimer.h"

namespace rofl {

class ciosrv;

class cioloop {
public:

        static cioloop&
        get_loop(pthread_t tid = 0) {
                if (0 == tid) {
                        tid = pthread_self();
                }
                if (cioloop::threads.find(tid) == cioloop::threads.end()) {
                        cioloop::threads[tid] = new cioloop(tid);
                }
                return *(cioloop::threads[tid]);
        };

        static void
        drop_loop(pthread_t tid) {
                if (cioloop::threads.find(tid) == cioloop::threads.end()) {
                        return;
                }
                delete cioloop::threads[tid];
                cioloop::threads.erase(tid);
        };

public:

        void
        run() {
                cioloop::get_loop().run_loop();
        };

        void
        stop() {
                if (cioloop::threads.find(tid) == cioloop::threads.end()) {
                        return;
                }
                cioloop::threads[tid]->keep_on_running = false;
                {
                        RwLock lock(rfds_rwlock, RwLock::RWLOCK_WRITE);
                        rfds.clear();
                }
                {
                        RwLock lock(wfds_rwlock, RwLock::RWLOCK_WRITE);
                        wfds.clear();
                }
                {
                        RwLock lock(events_rwlock, RwLock::RWLOCK_WRITE);
                        events.clear();
                }
                {
                        RwLock lock(timers_rwlock, RwLock::RWLOCK_WRITE);
                        timers.clear();
                }
                if (tid != pthread_self()) {
                        //pipe.writemsg('1'); // wakeup main loop, just in case
                }
        };

        void
        shutdown() {
                for (std::map<pthread_t, cioloop*>::iterator
                                it = cioloop::threads.begin(); it != cioloop::threads.end(); ++it) {
                        delete it->second;
                }
                cioloop::threads.clear();
        };

        pthread_t
        get_tid() const { return tid; };

protected:

        friend class ciosrv;

        void
        add_readfd(ciosrv* iosrv, int fd) {
                RwLock lock(rfds_rwlock, RwLock::RWLOCK_WRITE);
                rfds[fd] = iosrv;

                minrfd = (minrfd > (unsigned int)(fd+0)) ? (unsigned int)(fd+0) : minrfd;
                maxrfd = (maxrfd < (unsigned int)(fd+1)) ? (unsigned int)(fd+1) : maxrfd;

                if (tid != pthread_self()) {
                        pipe.writemsg('1'); // wakeup main loop, just in case
                }
        };

        void
        drop_readfd(ciosrv* iosrv, int fd) {
                RwLock lock(rfds_rwlock, RwLock::RWLOCK_WRITE);
                rfds[fd] = NULL;

                if (minrfd == (unsigned int)(fd+0)) {
                        minrfd = rfds.size();
                        for (unsigned int i = fd + 1; i < rfds.size(); i++) {
                                if (rfds[i] != NULL) {
                                        minrfd = i;
                                        break;
                                }
                        }
                }

                if (maxrfd == (unsigned int)(fd+1)) {
                        maxrfd = 0;
                        for (unsigned int i = fd; i > 0; i--) {
                                if (rfds[i] != NULL) {
                                        maxrfd = (i+1);
                                        break;
                                }
                        }
                }

                if (tid != pthread_self()) {
                        pipe.writemsg('1'); // wakeup main loop, just in case
                }
        };

        void
        add_writefd(ciosrv* iosrv, int fd) {
                RwLock lock(wfds_rwlock, RwLock::RWLOCK_WRITE);
                wfds[fd] = iosrv;

                minwfd = (minwfd > (unsigned int)(fd+0)) ? (unsigned int)(fd+0) : minwfd;
                maxwfd = (maxwfd < (unsigned int)(fd+1)) ? (unsigned int)(fd+1) : maxwfd;

                if (tid != pthread_self()) {
                        pipe.writemsg('1'); // wakeup main loop, just in case
                }
        };

        void
        drop_writefd(ciosrv* iosrv, int fd) {
                RwLock lock(wfds_rwlock, RwLock::RWLOCK_WRITE);
                wfds[fd] = NULL;

                if (minwfd == (unsigned int)(fd+0)) {
                        minwfd = wfds.size();
                        for (unsigned int i = fd + 1; i < wfds.size(); i++) {
                                if (wfds[i] != NULL) {
                                        minwfd = i;
                                        break;
                                }
                        }
                }

                if (maxwfd == (unsigned int)(fd+1)) {
                        maxwfd = 0;
                        for (unsigned int i = fd; i > 0; i--) {
                                if (wfds[i] != NULL) {
                                        maxwfd = (i+1);
                                        break;
                                }
                        }
                }

                if (tid != pthread_self()) {
                        pipe.writemsg('1'); // wakeup main loop, just in case
                }
        };

        void
        has_timer(ciosrv* iosrv) {
                RwLock lock(timers_rwlock, RwLock::RWLOCK_WRITE);
                timers[iosrv] = true;
                if (tid != pthread_self()) {
                        pipe.writemsg('1'); // wakeup main loop, just in case
                }
        };

        void
        has_no_timer(ciosrv *iosrv) {
                RwLock lock(timers_rwlock, RwLock::RWLOCK_WRITE);
                timers[iosrv] = false;
        };

        void
        has_event(ciosrv* iosrv) {
                RwLock lock(events_rwlock, RwLock::RWLOCK_WRITE);
                events[iosrv] = true;
                if (tid != pthread_self()) {
                        pipe.writemsg('1'); // wakeup main loop, just in case
                }
        };

        void
        has_no_event(ciosrv* iosrv) {
                RwLock lock(events_rwlock, RwLock::RWLOCK_WRITE);
                events[iosrv] = false;
        };

private:

        cioloop(pthread_t tid = 0) :
                tid(tid),
                keep_on_running(false) {

                if (0 == tid) {
                        this->tid = pthread_self();
                }

                struct rlimit rlim;
                if (getrlimit(RLIMIT_NOFILE, &rlim) < 0) {
                        throw eSysCall("getrlimit()");
                }
                RwLock rlock(rfds_rwlock, RwLock::RWLOCK_WRITE);
                RwLock wlock(wfds_rwlock, RwLock::RWLOCK_WRITE);
                for (unsigned int i = 0; i < rlim.rlim_cur; i++) {
                        rfds.push_back(NULL);
                        wfds.push_back(NULL);
                }
                minrfd = rfds.size();
                maxrfd = 0;
                minwfd = wfds.size();
                maxwfd = 0;
        };

        virtual
        ~cioloop() {};

        cioloop(cioloop const& t) {
                *this = t;
        };

        cioloop&
        operator= (cioloop const& t) {
                if (this == &t)
                        return *this;
                return *this;
        };

        void
        run_loop();


        static void
        child_sig_handler (int x);



public:

        friend std::ostream&
        operator<< (std::ostream& os, cioloop const& ioloop) {
                os << indent(0) << "<cioloop tid:0x"
                                << std::hex << ioloop.get_tid() << std::dec << ">" << std::endl;

                {
                        RwLock lock(ioloop.rfds_rwlock, RwLock::RWLOCK_READ);
                        os << indent(2) << "<read-fds: ";
                        for (unsigned int i = 0; i < ioloop.rfds.size(); ++i) {
                                if (NULL != ioloop.rfds.at(i)) {
                                        os << i << " ";
                                }
                        }
                        os << ">" << std::endl;
                }

                {
                        RwLock lock(ioloop.wfds_rwlock, RwLock::RWLOCK_READ);
                        os << indent(2) << "<write-fds: ";
                        for (unsigned int i = 0; i < ioloop.wfds.size(); ++i) {
                                if (NULL != ioloop.wfds.at(i)) {
                                        os << i << " ";
                                }
                        }
                        os << ">" << std::endl;
                }

                {
                        RwLock lock(ioloop.timers_rwlock, RwLock::RWLOCK_READ);
                        os << indent(2) << "<instances with timer needs: ";
                        // locking?
                        for (std::map<ciosrv*, bool>::const_iterator
                                        it = ioloop.timers.begin(); it != ioloop.timers.end(); ++it) {
                                os << it->first << ":" << it->second << " ";
                        }
                        os << ">" << std::endl;
                }

                {
                        RwLock lock(ioloop.events_rwlock, RwLock::RWLOCK_READ);
                        os << indent(2) << "<instances with event needs: ";
                        // locking?
                        for (std::map<ciosrv*, bool>::const_iterator
                                        it = ioloop.events.begin(); it != ioloop.events.end(); ++it) {
                                os << it->first << ":" << it->second << " ";
                        }
                        os << ">" << std::endl;
                }

                return os;
        };

private:

        static PthreadRwLock                                    threads_rwlock;
        static std::map<pthread_t, cioloop*>    threads;

        std::vector<ciosrv*>                                    rfds;
        mutable PthreadRwLock                                   rfds_rwlock;
        std::vector<ciosrv*>                                    wfds;
        mutable PthreadRwLock                                   wfds_rwlock;
        std::map<ciosrv*, bool>                                 timers;
        mutable PthreadRwLock                                   timers_rwlock;
        std::map<ciosrv*, bool>                                 events;
        mutable PthreadRwLock                                   events_rwlock;

        cpipe                                                                   pipe;
        pthread_t                                               tid;
        bool                                                                    keep_on_running;

        unsigned int                                                    minrfd; // lowest set readfd
        unsigned int                                                    maxrfd; // highest set readfd
        unsigned int                                                    minwfd; // lowest set writefd
        unsigned int                                                    maxwfd; // highest set writefd
};




class ciosrv : public ctimer_env {
public:

        ciosrv(pthread_t tid = 0);

        virtual
        ~ciosrv();

        ciosrv(const ciosrv& iosrv);

        ciosrv&
        operator= (const ciosrv& iosrv);

public:

        void
        notify(const cevent& event) {
                events.add_event(event);
                cioloop::get_loop(get_thread_id()).has_event(this);
        };

        pthread_t
        get_thread_id() const
        { return tid; };

protected:

        virtual void
        handle_event(
                        const rofl::cevent& event)
        {};

        virtual void
        handle_revent(
                        int fd)
        {};

        virtual void
        handle_wevent(int fd)
        {};

        virtual void
        handle_xevent(int fd)
        {};

        virtual void
        handle_timeout(int opaque, void *data = (void*)0)
        {};

protected:

        void
        register_filedesc_r(int fd) {
                rfds.insert(fd);
                cioloop::get_loop(get_thread_id()).add_readfd(this, fd);
        };

        void
        deregister_filedesc_r(int fd) {
                rfds.erase(fd);
                cioloop::get_loop(get_thread_id()).drop_readfd(this, fd);
        };

        void
        register_filedesc_w(int fd) {
                wfds.insert(fd);
                cioloop::get_loop(get_thread_id()).add_writefd(this, fd);
        };

        void
        deregister_filedesc_w(int fd) {
                wfds.erase(fd);
                cioloop::get_loop(get_thread_id()).drop_writefd(this, fd);
        };

protected:

        ctimer
        get_next_timer() {
                return timers.get_next_timer();
        };

        const rofl::ctimerid&
        register_timer(int opaque, const rofl::ctimespec& timespec) {
                if (timers.empty() || (get_thread_id() != pthread_self()))
                        cioloop::get_loop().has_timer(this);
                return timers.add_timer(ctimer(this, opaque, timespec));
        };

        const rofl::ctimerid&
        reset_timer(const rofl::ctimerid& timer_id, const rofl::ctimespec& timespec) {
                if (timers.empty() || (get_thread_id() != pthread_self()))
                        cioloop::get_loop().has_timer(this);
                return timers.reset(timer_id, timespec);
        };

        ctimerid&
        restart_timer(rofl::ctimerid& timer_id, int opaque, const rofl::ctimespec& timespec) {
                if (pending_timer(timer_id)) {
                        timer_id = reset_timer(timer_id, timespec);
                } else {
                        timer_id = register_timer(opaque, timespec);
                }
                return timer_id;
        };

        bool
        pending_timer(const rofl::ctimerid& timer_id) {
                return timers.pending(timer_id);
        };

        void
        cancel_timer(const rofl::ctimerid& timer_id) {
                timers.cancel(timer_id);
                if (timers.empty())
                        cioloop::get_loop().has_no_timer(this);
        };

        void
        cancel_all_timers() {
                timers.clear();
                cioloop::get_loop().has_no_timer(this);
        };

        void
        cancel_all_events() {
                events.clear();
                cioloop::get_loop().has_no_event(this);
        };

private:

        friend class cioloop;

        void
        __handle_event();

        void
        __handle_revent(int fd) {
                try {
                        handle_revent(fd);
                } catch (eEventsNotFound& e) {/*do nothing*/}
        };

        void
        __handle_timeout() {
                try {
                        ctimer timer = timers.get_expired_timer();
                        logging::trace << "[rofl-common][ciosrv][handle-timeout] timer: " << std::endl << timer;
                        handle_timeout(timer.get_opaque());
                } catch (eTimersNotFound& e) {/*do nothing*/}
        };

public:

        friend std::ostream&
        operator<< (std::ostream& os, const ciosrv& iosvc) {
                os << indent(0) << "<ciosrv >" << std::endl;
                        os << indent(2) << "<rfds: ";
                        for (std::set<int>::const_iterator it = iosvc.rfds.begin(); it != iosvc.rfds.end(); ++it) {
                                os << (*it) << " ";
                        }
                        os << ">" << std::endl;
                        os << indent(2) << "<wfds: ";
                        for (std::set<int>::const_iterator it = iosvc.wfds.begin(); it != iosvc.wfds.end(); ++it) {
                                os << (*it) << " ";
                        }
                        os << ">" << std::endl;

                        { indent i(2); os << iosvc.timers; }
                        { indent i(2); os << iosvc.events; }
                return os;
        };

private:

        static PthreadRwLock                    ciolist_rwlock;
        static std::set<ciosrv*>                ciolist;

        pthread_t                                               tid;
        std::set<int>                                   rfds;
        std::set<int>                                   wfds;
        ctimers                                                 timers;
        cevents                                                 events;
};





}; // end of namespace

#endif