/builddir/build/BUILD/libutil++-1.4.0/src/sella/net/SocketMux.cpp

/*
** libutil++
** $Id: SocketMux.cpp 1197 2014-10-14 22:26:11Z sella $
** Copyright (c) 2011-2014 Digital Genesis, LLC. All Rights Reserved.
** Released under the LGPL Version 2.1 License.
** http://www.digitalgenesis.com
*/

static const char rcsid[] __attribute__((used)) = "$Id: SocketMux.cpp 1197 2014-10-14 22:26:11Z sella $";

#include "SocketMux.h"
#include "../util/CommonMacro.h"
#include "../util/Time.h"

#include <stdio.h>
#include <fcntl.h>
#include <netdb.h>
#include <errno.h>
#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/select.h>

#define EPOLL_SIZE 64 /* Value not used since Linux 2.6.8, but must be greater than 0. */

/* Global Variables */
extern int errno;

using namespace sella::net;

SocketMux::SocketMux(ssize_t epollMaxEvents, bool pthread_cancel) throw (SocketException) :
        e(-1),
        fdmax(-1),
        cancel(pthread_cancel),
        epollMaxEvents(epollMaxEvents),
        events(NULL)
{
        FD_ZERO(&rfdset);
        FD_ZERO(&wfdset);
        FD_ZERO(&efdset);

        events = (epoll_event*) calloc(epollMaxEvents, sizeof(struct epoll_event));

        if ((e = epoll_create(EPOLL_SIZE)) == -1) {
                THROW2(SocketException, "epoll_create1()");
        }
}

SocketMux::~SocketMux() {
        try {
                clear();
        } catch (SocketException) { }

        close(e);
        SAFE_FREE(events);
}

int SocketMux::epoll(struct timeval *tv) throw (SocketException) {
        int c, msec = (int) sella::util::Time::tvtomsec(*tv);
        bool event;

        clearLists();

        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
        c = ::epoll_wait(e, events, epollMaxEvents, msec);
        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);

        if (c < 0) {
                if (errno == EINTR) {
                        return 0;
                } else {
                        THROW2(SocketException, "epoll_wait()");
                }
        } else if (c == 0) {
                return c;
        }

        for (int i = 0; i < c; i++) {
                event = (bool) (events[i].events & EPOLLIN);
                if (event) {
                        auto it = rMap.find(events[i].data.fd);
                        if (it != rMap.end()) {
                                it->second->read = true;

                                rList.push_front(it->second);
                        }
                }

                event = (bool) (events[i].events & EPOLLOUT);
                if (event) {
                        auto it = wMap.find(events[i].data.fd);
                        if (it != wMap.end()) {
                                it->second->write = true;

                                wList.push_front(it->second);
                        }
                }

                event = (bool) ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP));
                if (event) {
                        auto it = eMap.find(events[i].data.fd);
                        if (it != eMap.end()) {
                                it->second->except = true;

                                eList.push_front(it->second);
                        }
                }
        }

        return c;
}

int SocketMux::select(struct timeval *tv) throw (SocketException) {
        int f = 0, c;
        fd_set rfds, wfds, efds;

        clearLists();
        FD_ZERO(&rfds);
        FD_COPY(&rfds, &rfdset);
        FD_ZERO(&wfds);
        FD_COPY(&wfds, &wfdset);
        FD_ZERO(&efds);
        FD_COPY(&efds, &efdset);

        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
        c = ::select(fdmax + 1, &rfds, &wfds, &efds, tv);
        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);

        if (c < 0) {
                if (errno == EINTR) {
                        return 0;
                } else {
                        THROW2(SocketException, "select() [%d]", errno);
                }
        } else if (c == 0) {
                return c;
        }

        for (auto it = rMap.cbegin(); it != rMap.cend(); ++it) {
                if (FD_ISSET(it->first, &rfds)) {
                        it->second->read = true;
                        rList.push_front(it->second);

                        if (++f >= c) return c;
                }
        }

        for (auto it = wMap.cbegin(); it != wMap.cend(); ++it) {
                if (FD_ISSET(it->first, &wfds)) {
                        it->second->write = true;
                        wList.push_front(it->second);

                        if (++f >= c) return c;
                }
        }

        for (auto it = eMap.cbegin(); it != eMap.cend(); ++it) {
                if (FD_ISSET(it->first, &efds)) {
                        it->second->except = true;
                        eList.push_front(it->second);

                        if (++f >= c) return c;
                }
        }

        return c;
}

int SocketMux::rselect(struct timeval *tv) throw (SocketException) {
        int f = 0, c;
        fd_set rfds;

        rList.clear();
        FD_ZERO(&rfds);
        FD_COPY(&rfds, &rfdset);

        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
        c = ::select(fdmax + 1, &rfds, NULL, NULL, tv);
        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);

        if (c < 0) {
                if (errno == EINTR) {
                        return 0;
                } else {
                        THROW2(SocketException, "select() [%d]", errno);
                }
        } else if (c == 0) {
                return c;
        }

        for (auto it = rMap.cbegin(); it != rMap.cend(); ++it) {
                if (FD_ISSET(it->first, &rfds)) {
                        it->second->read = true;
                        rList.push_front(it->second);

                        if (++f >= c) return c;
                }
        }

        return c;
}

int SocketMux::wselect(struct timeval *tv) throw (SocketException) {
        int f = 0, c;
        fd_set wfds;

        wList.clear();
        FD_ZERO(&wfds);
        FD_COPY(&wfds, &wfdset);

        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
        c = ::select(fdmax + 1, NULL, &wfds, NULL, tv);
        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);

        if (c < 0) {
                if (errno == EINTR) {
                        return 0;
                } else {
                        THROW2(SocketException, "select() [%d]", errno);
                }
        } else if (c == 0) {
                return c;
        }

        for (auto it = wMap.cbegin(); it != wMap.cend(); ++it) {
                if (FD_ISSET(it->first, &wfds)) {
                        it->second->write = true;
                        wList.push_front(it->second);

                        if (++f >= c) return c;
                }
        }

        return c;
}

int SocketMux::eselect(struct timeval *tv) throw (SocketException) {
        int f = 0, c;
        fd_set efds;

        eList.clear();
        FD_ZERO(&efds);
        FD_COPY(&efds, &efdset);

        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
        c = ::select(fdmax + 1, NULL, NULL, &efds, tv);
        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);

        if (c < 0) {
                if (errno == EINTR) {
                        return 0;
                } else {
                        THROW2(SocketException, "select() [%d]", errno);
                }
        } else if (c == 0) {
                return c;
        }

        for (auto it = eMap.cbegin(); it != eMap.cend(); ++it) {
                if (FD_ISSET(it->first, &efds)) {
                        it->second->except = true;
                        eList.push_front(it->second);

                        if (++f >= c) return c;
                }
        }

        return c;
}

int SocketMux::rwselect(struct timeval *tv) throw (SocketException) {
        int f = 0, c;
        fd_set rfds, wfds;

        rList.clear();
        wList.clear();
        FD_ZERO(&rfds);
        FD_COPY(&rfds, &rfdset);
        FD_ZERO(&wfds);
        FD_COPY(&wfds, &wfdset);

        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
        c = ::select(fdmax + 1, &rfds, &wfds, NULL, tv);
        if (cancel) pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);

        if (c < 0) {
                if (errno == EINTR) {
                        return 0;
                } else {
                        THROW2(SocketException, "select() [%d]", errno);
                }
        } else if (c == 0) {
                return c;
        }

        for (auto it = rMap.cbegin(); it != rMap.cend(); ++it) {
                if (FD_ISSET(it->first, &rfds)) {
                        it->second->read = true;
                        rList.push_front(it->second);

                        if (++f >= c) return c;
                }
        }

        for (auto it = wMap.cbegin(); it != wMap.cend(); ++it) {
                if (FD_ISSET(it->first, &wfds)) {
                        it->second->write = true;
                        wList.push_front(it->second);

                        if (++f >= c) return c;
                }
        }

        return c;
}

void SocketMux::clear(void) throw (SocketException) {
        fdmax = -1;

        clearLists();
        rMap.clear();
        wMap.clear();
        eMap.clear();

        FD_ZERO(&rfdset);
        FD_ZERO(&wfdset);
        FD_ZERO(&efdset);

        close(e);

        if ((e = epoll_create(EPOLL_SIZE)) == -1) {
                THROW2(SocketException, "epoll_create1()");
        }
}

size_t SocketMux::size(void) const {
        return (rMap.size() + wMap.size() + eMap.size());
}

const SocketMux::shared_flist& SocketMux::getReadList(void) {
        return rList;
}

const SocketMux::shared_flist& SocketMux::getWriteList(void) {
        return wList;
}

const SocketMux::shared_flist& SocketMux::getExceptList(void) {
        return eList;
}

bool SocketMux::insertRead(Socket::shared socket) throw (SocketException) {
        fd_set *set = &rfdset;
        int_shared_map *map = &rMap;

        return insert(socket, set, map, EPOLLIN | EPOLLET);
}

bool SocketMux::removeRead(Socket::shared socket) throw (SocketException) {
        fd_set *set = &rfdset;
        int_shared_map *map = &rMap;

        return remove(socket, set, map, EPOLLIN | EPOLLET);
}

bool SocketMux::insertWrite(Socket::shared socket) throw (SocketException) {
        fd_set *set = &wfdset;
        int_shared_map *map = &wMap;

        return insert(socket, set, map, EPOLLOUT | EPOLLET);
}

bool SocketMux::removeWrite(Socket::shared socket) throw (SocketException) {
        fd_set *set = &wfdset;
        int_shared_map *map = &wMap;

        return remove(socket, set, map, EPOLLOUT | EPOLLET);
}

bool SocketMux::insertExcept(Socket::shared socket) throw (SocketException) {
        fd_set *set = &efdset;
        int_shared_map *map = &eMap;

        return insert(socket, set, map, EPOLLERR | EPOLLET);
}

bool SocketMux::removeExcept(Socket::shared socket) throw (SocketException) {
        fd_set *set = &efdset;
        int_shared_map *map = &eMap;

        return remove(socket, set, map, EPOLLERR | EPOLLET);
}

const SocketMux::shared SocketMux::shared_ptr(ssize_t epollMaxEvents, bool pthread_cancel) throw (SocketException) {
        return std::make_shared<SocketMux>(epollMaxEvents, pthread_cancel);
}

bool SocketMux::insert(Socket::shared socket, fd_set *set, int_shared_map *map, uint32_t events) throw (SocketException) {
        assert(set);
        assert(map);


        if (!socket) {
                THROW(SocketException, "Passed NULL socket");
        }

        int s = socket->getFD();

        if (s < 0 || s > FD_SETSIZE) {
                THROW(SocketException, "Socket fd(%d) less than 0 or greater than FD_SETSIZE(%u)", s, FD_SETSIZE);
        }

        if (!FD_ISSET(s, set)) {
                struct epoll_event event = {};
                event.data.fd = s;
                event.events = events;
                if (epoll_ctl(e, EPOLL_CTL_ADD, s, &event) != 0) {
                        THROW2(SocketException, "epoll_ctl()");
                }

                FD_SET(s, set);
                fdmax = (s > fdmax) ? s : fdmax;
                (*map)[s] = socket;
                socket->mux = true;

                return true;
        }

        return false;
}

bool SocketMux::remove(Socket::shared socket, fd_set *set, int_shared_map *map, uint32_t events) throw (SocketException) {
        assert(set);
        assert(map);

        if (!socket) {
                THROW(SocketException, "Passed NULL socket");
        }

        int s = socket->getFD();

        if (s < 0 || s > FD_SETSIZE) {
                THROW(SocketException, "Socket fd(%d) less than 0 or greater than FD_SETSIZE(%u)", s, FD_SETSIZE);
        }

        if (FD_ISSET(s, set)) {
                struct epoll_event event = {};
                event.data.fd = s;
                event.events = events;
                if (epoll_ctl(e, EPOLL_CTL_DEL, s, &event) != 0) {
                        THROW2(SocketException, "epoll_ctl()");
                }

                FD_CLR(s, set);

                if (s >= fdmax) { /* Find new fdmax */
                        for (; fdmax >= 0; fdmax--) {
                                if (FD_ISSET(fdmax, set)) {
                                        break;
                                }
                        }
                }

                map->erase(s);
                socket->mux = false;

                return true;
        }

        return false;
}

void SocketMux::clearLists(void) {
        rList.clear();
        wList.clear();
        eList.clear();
}

/*
** vim: noet ts=3 sw=3
*/

---