Poll_epoll.C

//  Copyright (c) 2005-2010, Isode Limited, London, England.
//  All rights reserved.
//                                                                       
//  Acquisition and use of this software and related materials for any      
//  purpose requires a written licence agreement from Isode Limited,
//  or a written licence from an organisation licenced by Isode Limited
//  to grant such a licence.
//
 
 
// 
//
// Poll_epoll.C
//
// Poll provider using epoll
//
// @VERSION@
 
#include <errno.h>
#include <set>
 
#include "Event_p.h"
 
#ifdef HAVE_SYS_EPOLL_H
#include <sys/epoll.h>
#endif
 
#ifdef HAVE_EPOLL_WAIT
 
namespace Poll {
 
    typedef std::set<User *> polldatamap_t;
 
    class Poll_epoll : public Provider {
    private:
        static const int epoll_maxevents = 256;
 
        pthread_mutex_t       _mutex;        
 
        polldatamap_t            _internal;     
 
        int                      _epfd;         
        int                      _wakefds[2];   
        bool                     _inpoll;       
        bool                     _wakepending;  
        bool                     _terminating;  
        bool                     _queued;       
        ::Event::Manager        *_manager;      
        ProviderEvent            _myevent;      
        ProviderEvent            _termevent;    
 
        int EpollControl (User *user) {
            epoll_event event;
 
            event.events = 0;
            
            if ( user->_pevents & Event_In )
                event.events |= (EPOLLIN | EPOLLPRI | EPOLLERR | EPOLLHUP);
 
            if ( user->_pevents & Event_Out )
                event.events |= (EPOLLOUT | EPOLLERR | EPOLLHUP);
 
            event.events |= EPOLLONESHOT;
 
            event.data.ptr = static_cast<void *>(user);
 
            if ( epoll_ctl (_epfd, EPOLL_CTL_MOD, user->_fd, &event) != 0 ) {
                MSG_IOevent_Pollerror_LOG ("epoll_ctl mod");
                return NOTOK;
            } else {
                return OK;
            }
        }
 
    public:
        Poll_epoll ();
 
        virtual ~Poll_epoll ();
 
        virtual int Register (User *user);
 
        virtual void Deregister (User *user);
 
        virtual void Control (User *user);
 
        virtual void Deliver (pollAction *msg);
    };
 
#if defined(__clang__)
    const int Poll_epoll::epoll_maxevents;
#endif
 
    Poll_epoll::Poll_epoll ()
    {
        pthread_mutex_init (&_mutex, NULL);
 
        pollAction action;
 
        action.terminate = false;
        _myevent.SetUserData (this, action);
 
        action.terminate = true;
        _termevent.SetUserData (this, action);
 
        _inpoll = false;
        _wakepending = false;
 
        _terminating = false;
        _queued      = false;
 
        _epfd = epoll_create (100);
        if ( _epfd < 0 )
            MSG_IOevent_Pollerror_LOG ("epoll_create");
 
        // Leave creating pipe etc. until we can report error
        _wakefds[0] = _wakefds[1] = -1;
 
        _manager = ::Event::Manager::GetManager();
 
        _manager->QueueAtExit (&_termevent);
    }
 
    Poll_epoll::~Poll_epoll ()
    {
        pthread_mutex_destroy (&_mutex);
        if ( _wakefds[0] >= 0 ) close (_wakefds[0]);
        if ( _wakefds[1] >= 0 ) close (_wakefds[1]);
        if ( _epfd >= 0 ) close (_epfd);
    }
 
    void Poll_epoll::Deliver (pollAction *msg)
    {
        int rc;
        LOG_DEBUG (( "Poll_epoll::Deliver ()"));
 
        if ( (rc = pthread_mutex_lock (&_mutex)) != 0 ) {
            MSG_IOevent_Pollthread_LOG (rc);
            _manager->QueueAt (&_myevent, 10000);
            _queued = true;
            return;
        }
 
        if ( _terminating ) {
            pthread_mutex_unlock(&_mutex);
            return;
        }
 
        if ( msg->terminate ) {
            // System is shutting down
 
            // Prevents queuing
            _queued = true;
 
            _terminating = true;
 
            // Tell users that we are terminating
            // There is an implied removal of registrations
            // The PollUser MUST NOT refer to the EventSource again
            // Note: this is in critical section, so would get deadlock
            // if the PollUser called back to Deregister.
 
            ::Poll::pollmsg cancelmsg;
            cancelmsg.events = Poll::Event_Terminate;
 
            polldatamap_t::iterator scan;
            for ( scan = _internal.begin(); scan != _internal.end(); ++scan )
                (*scan)->_event.SendAtExit (cancelmsg);
 
            // (It is a pity we need to use the wakup mechanism, as it is
            //  only needed for termination)
            bool wake = false;
            if ( _inpoll && !_wakepending ) {
                 wake = true;
                _wakepending = true;
            }
 
            pthread_mutex_unlock(&_mutex);
 
            if ( wake ) {
                char bytev[1];
                bytev[0] = 2;
                if ( write (_wakefds[1], bytev, 1) < 1 )
                    MSG_IOevent_Pollerror_LOG ("write");
            }
 
            close (_epfd);
            return;
        } // end of handling termination
 
        if ( _wakefds[0] == -1 ) {
            if ( pipe (_wakefds) != 0 ) {
                MSG_IOevent_Pollerror_LOG ("pipe");
                _manager->QueueAt (&_myevent, 1000);
                _queued = true;
                pthread_mutex_unlock(&_mutex);
                return;
            }
 
            epoll_event wevent;
 
            wevent.events = EPOLLIN;
            wevent.data.ptr = 0;
 
            if ( epoll_ctl (_epfd, EPOLL_CTL_ADD, _wakefds[0], &wevent) != 0 ){
                MSG_IOevent_Pollerror_LOG ("epoll_ctl");
                _manager->QueueAt (&_myevent, 1000);
                _queued = true;
                pthread_mutex_unlock(&_mutex);
                return;
            }
        }
 
        int timeout = -1;
 
        const timespec *wakeat = _manager->NextEventDue ();
 
        if ( wakeat ) {
            struct timespec now;
            getNow(now);
            // timespec_diff returns difference in microseconds.
            // returns 0 if delay preceeds now.
            timeout = timespec_diff (wakeat, &now)/1000;
            LOG_DEBUG (("epoll timeout=%d", timeout));
        }
 
        _inpoll = true; // about to be true
 
        pthread_mutex_unlock(&_mutex);
 
        // Tell event manager we might go away for a while
        _manager->Blocking();
 
        epoll_event events[epoll_maxevents];
 
        LOG_DEBUG (("Poll_epoll::Wait: epoll_wait"));
        int nfds = epoll_wait (_epfd, events, epoll_maxevents, timeout);
        LOG_DEBUG (("Poll_epoll::Wait poll()=%d", nfds));
        int eno = errno; // Preserve errno
        
        // Tell event manager we are back
        _manager->Unblocked();
 
        if ( nfds < 0 ) {
            if ( (rc = pthread_mutex_lock (&_mutex)) != 0 ) {
                MSG_IOevent_Pollthread_LOG (rc);
                _manager->QueueAt (&_myevent, 10000);
                _queued = true;
                _inpoll = false;
                return;
            }
 
            _inpoll = false;
                
            if ( eno == EINTR ) {
                LOG_DEBUG (( "Poll_epoll::Wait: epoll_wait EINTR"));
                // Queue event to self again.
                _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO);
 
            } else {
                MSG_IOevent_Pollerror_LOG ("epoll_wait");
                // Try again in one second
                _manager->QueueAt (&_myevent, 1000);
            }
 
            _queued = true;
            pthread_mutex_unlock(&_mutex);
            return;
        }
 
        if ( (rc = pthread_mutex_lock (&_mutex)) != 0 ) {
            MSG_IOevent_Pollthread_LOG (rc);
            _manager->QueueAt (&_myevent, 10000);
            _queued = true;
            _inpoll = false;
            _wakepending = false;
            return;
        }
 
        _inpoll = false;
        _wakepending = false;
 
        if ( _terminating )
            return;
 
        pollmsg    doevents;
 
        for ( int i = 0; i < nfds; i++ ) {
            if ( events[i].data.ptr == 0 ) {
                char buffer[128];
                if ( read (_wakefds[0], buffer, sizeof buffer) < 0 )
                    MSG_IOevent_Pollerror_LOG ("read");
 
            } else if ( events[i].events != 0 ) {
                User *user = static_cast<User *>(events[i].data.ptr);
 
                polldatamap_t::iterator entry =
                    _internal.find (user);
 
                // Check was not unregistered in the interim
                if ( entry == _internal.end() ) continue;
 
                int fd = user->_fd;
 
                // Translate events into external mask
                // If error, clear all events
                pollmsg these_events;
                if ( events[i].events & EPOLLIN )
                    these_events.events |= Event_In;
 
                if ( events[i].events & EPOLLOUT )
                    these_events.events |= Event_Out;
 
                if ( events[i].events & EPOLLERR )
                    these_events.events |= Event_Err;
 
                if ( events[i].events & EPOLLHUP )
                    these_events.events |= Event_Hup;
 
                // Under normal circumstances expect a PollAsyncEvent to
                // be queued to the event manager
                // But if it returns a non-zero result, then
                // we deregister the PollUser
 
                LOG_DEBUG (("Poll_epoll: send %d events %x",
                            fd, these_events.events));
 
                user->_event.Send(these_events);
 
                // If the events include an error then the endpoint
                // is automatically deregistered
                if ( these_events.events & (Event_Err|Event_Hup|Event_Nval) ){ 
                    // This is needed to keep valgrind quiet
                    struct epoll_event dummy;
                    memset (&dummy,0,sizeof dummy);
 
                    if ( epoll_ctl (_epfd, EPOLL_CTL_DEL, fd, &dummy)
                         != 0 )
                        MSG_IOevent_Pollerror_LOG ("epoll_ctl del");
                    _internal.erase (entry);
                }
            }
        }
 
        // Only queue event for processing with epoll if items remain
        if ( !_internal.empty() ) {
            _queued = true;
            _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO, true);
        } else {
            _queued = false;
        }
 
        LOG_DEBUG(("Ending processing, queued=%d", _queued));
        pthread_mutex_unlock(&_mutex);
    }
 
    int Poll_epoll::Register (User *user)
    {
        int rc;
 
        LOG_DEBUG (("Poll_epoll: Register: %d", user->_fd));
 
        if ( (rc = pthread_mutex_lock (&_mutex)) != 0 ) 
            return rc;
 
        if ( _terminating ) {
            rc = 0;
        } else if ( _epfd < 0 ) {
            rc = -1;
        } else {
 
            // Check to see if fd already registered
            polldatamap_t::iterator entry = _internal.find (user);
 
            if ( entry == _internal.end() ) {
                _internal.insert (user);
 
                epoll_event event;
 
                event.events = 0;
                event.data.ptr = static_cast<void *>(user);
 
                rc = epoll_ctl (_epfd, EPOLL_CTL_ADD, user->_fd, &event);
 
                if ( rc < 0 ) {
                    MSG_IOevent_Pollerror_LOG ("epoll_ctl add");
 
                    entry = _internal.find (user);
                    if ( entry != _internal.end() )
                        _internal.erase (entry);
                }
 
            } else {
                rc = -1;
            }
        }
 
        // Have a poll event if there are any registered FDs
        if ( !_inpoll && !_queued ) {
            LOG_DEBUG (("Control: adding Poll event"));
            _queued = true;
            _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO);
        }
 
        pthread_mutex_unlock (&_mutex);
        return rc;
    }
 
    void Poll_epoll::Deregister (User *user)
    {
        LOG_DEBUG (("Poll_epoll: Deregister: %d", user->_fd));
 
        if ( (pthread_mutex_lock (&_mutex)) != 0 ) 
            return;
 
        polldatamap_t::iterator entry = _internal.find (user);
 
        if ( entry != _internal.end() ) {
 
            // Do NOT cancel the event, as the poll user should be the
            // only user of this.
 
            _internal.erase (entry);
 
            // This is needed to keep valgrind quiet
            struct epoll_event dummy;
            memset (&dummy,0,sizeof dummy);
 
            // Don't log the EBADF case as the fd may have already been closed
            if ( epoll_ctl (_epfd, EPOLL_CTL_DEL, user->_fd, &dummy) != 0 &&
                 errno != EBADF )
                MSG_IOevent_Pollerror_LOG ("epoll_ctl del");
                
            // Do not wake poll(). If the end point were in use, any events
            // will be cancelled.
        }
 
        pthread_mutex_unlock (&_mutex);
    }
 
    void Poll_epoll::Control (User *user)
    {
        int fd = user->_fd;
 
        LOG_DEBUG (("Poll_epoll: Control: %d %x", fd, user->_uevents));
 
        if ( _terminating )
            return;
 
        user->_pevents = user->_uevents;
        (void) EpollControl (user);
    }
 
    Provider *CreatePollepollProvider ()
    {
        // Can be in glibc but not in the kernel
        int fd = epoll_create (1);
 
        if ( fd < 0 ) {
            LOG_DEBUG (("epoll_create() failed"));
            return 0;
        }
 
        close (fd);
 
        return new Poll_epoll();
    }
}
 
#else
 
namespace Poll {
 
    // Don't have epoll() therefore cannot have this kind of provider
    Provider *CreatePollepollProvider ()
    {
        return 0;
    }
}
#endif