Poll_select.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_select.C
//
// Poll provider using select
//
// @VERSION@
 
// Define this here before the inclusion of the Winsock header files
// but only redefine on Windows
#include <isode/config.h>
#ifdef IC_WIN32
#define FD_SETSIZE 4096
#endif
 
#include <errno.h>
#include <map>
 
#include "Event_p.h"
 
#include <isode/ll/internet.h>
 
#ifdef HAVE_SYS_SELECT_H
#include <sys/select.h>
#endif
 
namespace Poll {
 
    // Wrap up an FD set implementation, so that we can have an extensible
    // set on Windows.
 
#ifdef IC_WIN32
    class isode_fd_set {
    private:
        fd_set         _ifds;
        fd_set        *_fdsp;
 
    public:
        isode_fd_set (int nfds) { 
            if ( nfds > FD_SETSIZE ) {
                _fdsp = (fd_set *)
                    smalloc (sizeof (*_fdsp)+(nfds-FD_SETSIZE)*sizeof(endpoint_t));
                _fdsp->fd_count = 0;
 
            } else {
                _fdsp = &_ifds;
                FD_ZERO (_fdsp);
            }
        }
 
        ~isode_fd_set () { if ( _fdsp != &_ifds ) free (_fdsp); }
 
        void set (endpoint_t fd) { FD_SET (fd, _fdsp); }
 
        bool is_set (endpoint_t fd) { return 0 != FD_ISSET(fd, _fdsp); }
 
        fd_set *fds () { return _fdsp; }
    };
#else
    class isode_fd_set {
    private:
        fd_set         _fds;
 
    public:
        isode_fd_set (int nfds ARGNOTUSED) { FD_ZERO (&_fds); }
 
        void set (endpoint_t fd) { FD_SET (fd, &_fds); }
 
        bool is_set (endpoint_t fd) { return FD_ISSET (fd, &_fds); }
 
        fd_set *fds () { return &_fds; }
    };
#endif
 
    typedef std::map<int, User *> polldatamap_t;
 
    class Poll_select : public Provider {
    private:
        pthread_mutex_t       _mutex;        
 
        polldatamap_t            _internal;     
        int                      _nfds;         
 
        struct sockaddr_in       _addr;         
        endpoint_t               _wakefds[2];   
        bool                     _inpoll;       
        bool                     _wakepending;  
        bool                     _terminating;  
        bool                     _queued;       
        ::Event::Manager        *_manager;      
        ProviderEvent            _myevent;      
        ProviderEvent            _termevent;    
 
        int setup_wakeup ();                    
        void tellTerminating();                 
 
    public:
        Poll_select ();
 
        virtual ~Poll_select ();
 
        virtual int Register (User *user);
 
        virtual void Deregister (User *user);
 
        virtual void Control (User *user);
 
        virtual void Deliver (pollAction *msg);
    };
 
    Poll_select::Poll_select ()
    {
        pthread_mutex_init (&_mutex, NULL);
 
        pollAction action;
 
        action.terminate = false;
        _myevent.SetUserData (this, action);
 
        action.terminate = true;
        _termevent.SetUserData (this, action);
 
        _nfds = 1;  // Count pipe, although not yet open
 
        _inpoll = false;
        _wakepending = false;
 
        _terminating = false;
        _queued      = false;
 
        // Leave creating sockets until we can report error
        _wakefds[0] = _wakefds[1] = INVALID_ENDPOINT;
 
        _manager = ::Event::Manager::GetManager();
 
        _manager->QueueAtExit (&_termevent);
    }
 
    Poll_select::~Poll_select ()
    {
        pthread_mutex_destroy (&_mutex);
        if ( _wakefds[0] != INVALID_ENDPOINT) closesocket (_wakefds[0]);
        if ( _wakefds[1] != INVALID_ENDPOINT) closesocket (_wakefds[1]);
    }
 
    int Poll_select::setup_wakeup () {
 
        if ( (_wakefds[1] = socket (AF_INET, SOCK_DGRAM, 0)) == INVALID_ENDPOINT ) {
            MSG_IOevent_Pollerror_LOG ("socket");
            return NOTOK;
        }
 
        _addr.sin_family = AF_INET;
 
        isode_inet_aton ("127.0.0.1", &_addr.sin_addr);
 
        u_short port;
        for ( port = 0xFFFF; port > 0; port-- ) {
            _addr.sin_port = htons (port);
 
            if ( (_wakefds[0] = socket (AF_INET, SOCK_DGRAM, 0)) == INVALID_ENDPOINT ) {
                MSG_IOevent_Pollerror_LOG ("socket");
                closesocket (_wakefds[1]);
                _wakefds[1] = INVALID_ENDPOINT;
                return NOTOK;
            }
 
            if ( bind (_wakefds[0], reinterpret_cast<sockaddr *>(&_addr),
                       sizeof (_addr)) == 0 )
                break;
 
            isode_set_errno ();
            
            // Always need to close port, even if we are trying again
            closesocket (_wakefds[0]);
            _wakefds[0] = INVALID_ENDPOINT;
 
            if ( errno == EADDRINUSE ) {
                LOG_DEBUG (("Failed to bind to port %d - in use", port));
            } else {
                MSG_IOevent_Pollerror_LOG ("bind");
                closesocket (_wakefds[1]);
                _wakefds[1] = INVALID_ENDPOINT;
                return NOTOK;
            }
        }
 
        if ( port == 0 ) {
            MSG_IOevent_Pollerror_LOG ("bind");
            return NOTOK;
        }
 
        return OK;
    }
 
    void Poll_select::tellTerminating ()
    {
        // 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->second->_event.SendAtExit (cancelmsg);
    }
 
    void Poll_select::Deliver (pollAction *msg)
    {
        int rc;
        LOG_DEBUG (( "Poll_select::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;
 
            tellTerminating();
 
            bool wake = false;
            if ( _inpoll && !_wakepending ) {
                wake = true;
                _wakepending = true;
            }
 
            pthread_mutex_unlock(&_mutex);
 
            if ( wake ) {
                char bytev[1];
                bytev[0] = 2;
                if ( sendto (_wakefds[1], bytev, 1, 0,
                             reinterpret_cast<sockaddr *>(&_addr), sizeof _addr) < 1 )
                    MSG_IOevent_Pollerror_LOG ("write");
            }
            return;
        } // end of handling termination
 
        if ( _wakefds[0] == INVALID_ENDPOINT && setup_wakeup() != OK ) {
            _manager->QueueAt (&_myevent, 1000);
            _queued = true;
            pthread_mutex_unlock(&_mutex);
            return;
        }
 
        isode_fd_set rfds (_nfds);
        isode_fd_set wfds (_nfds);
        isode_fd_set efds (_nfds);
 
        // First entry is always for wake-up FD
        endpoint_t maxfd = _wakefds[0];
 
        rfds.set (_wakefds[0]);
        efds.set (_wakefds[0]);
 
        // scan registered FDs for those which are needing polling
        int nfds = 1;
        polldatamap_t::const_iterator scan;
        for ( scan = _internal.begin(); scan != _internal.end(); ++scan ) {
            if ( scan->first == INVALID_ENDPOINT || scan->second->_pevents == 0 ) continue;
 
            if ( scan->second->_pevents & POLLIN )
                rfds.set (scan->first);
 
            if ( scan->second->_pevents & POLLOUT )
                wfds.set (scan->first);
 
            efds.set (scan->first);
 
            if ( scan->first > maxfd )
                maxfd = scan->first;
 
            nfds++;
        }
 
        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;
        }
 
        _inpoll = true; // about to be true
 
        pthread_mutex_unlock(&_mutex);
 
        // Tell event manager we might go away for a while
        _manager->Blocking();
 
        struct timeval timev, *tp;
 
        if ( timeout >= 0 ) {
            // timeout is in millisecs
            tp = &timev;
            tp->tv_sec  = timeout / 1000;
            tp->tv_usec = (timeout % 1000) * 1000; 
        } else {
            tp = 0;
        }
 
        LOG_DEBUG (("Poll_select::Wait: select nfds=%d", nfds));
        rc = select (maxfd+1, rfds.fds(), wfds.fds(), efds.fds(), tp);
        LOG_DEBUG (("Poll_select::Wait select()=%d", rc));
        int eno = errno; // Preserve errno
        
        // Tell event manager we are back
        _manager->Unblocked();
 
        if ( rc < 0 ) {
            if ( (rc = pthread_mutex_lock (&_mutex)) != 0 ) {
                MSG_IOevent_Pollthread_LOG (rc);
                _manager->QueueAt (&_myevent, 10000);
                _queued = true;
                _inpoll = false;
                return;
            }
 
            _inpoll = false;
                
#ifdef IC_WIN32
            int lasterr = WSAGetLastError();
            LOG_DEBUG (("Poll_select::Wait: select error %d", lasterr));
            switch ( lasterr ) {
            case WSANOTINITIALISED:
            case WSAENETDOWN:
                // No point in carrying on, terminate
                MSG_IOevent_Pollerror_LOG ("select");
                _queued = true;
                _terminating = true;
                tellTerminating();
                return;
 
            case WSAEINTR:
                LOG_DEBUG (( "Poll_select::Wait: select EINTR"));
                // Queue event to self again.
                _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO);
                break;
 
            default:
                MSG_IOevent_Pollerror_LOG ("select");
                // Try again in one second
                _manager->QueueAt (&_myevent, 1000);
                break;
            }
#else
            if ( eno == EINTR ) {
                LOG_DEBUG (( "Poll_select::Wait: select EINTR"));
                // Queue event to self again.
                _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO);
 
            } else {
                MSG_IOevent_Pollerror_LOG ("select");
                // Try again in one second
                _manager->QueueAt (&_myevent, 1000);
            }
#endif
 
            _queued = true;
            pthread_mutex_unlock(&_mutex);
            return;
        }
 
        // Soak up wakeup bytes outside critical section
        if ( rfds.is_set (_wakefds[0]) ) {
            char buffer[128];
            struct sockaddr_in addr;
#ifdef RECVFROM_NEEDS_INTP
            int alen;
#else
            socklen_t alen;
#endif
            alen = sizeof addr;
            if ( recvfrom (_wakefds[0], buffer, sizeof buffer, 0,
                           reinterpret_cast<sockaddr *>(&addr), &alen) < 0 )
                MSG_IOevent_Pollerror_LOG ("recvfrom");
            rc--;
        }
 
        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;
 
        // Count FDs which remain
        int waitingfds = 0;
 
        // With select, we scan all the endpoints again.        
        polldatamap_t::iterator it1, it2;
        for ( it1 = _internal.begin(); it1 != _internal.end(); ++it1 ) {
            if ( it1->first == INVALID_ENDPOINT || it1->second->_pevents == 0 ) continue;
 
            pollmsg these_events;
 
            these_events.events = 0;
 
            if ( efds.is_set (it1->first) )
                these_events.events |= Event_Err;
 
            if ( rfds.is_set (it1->first) )
                these_events.events |= Event_In;
 
            if ( wfds.is_set (it1->first) )
                these_events.events |= Event_Out;
 
            if ( these_events.events != 0 ) {
                it1->second->_pevents = 0;
                it1->second->_event.Send(these_events);
            }
 
            // If events remain, then still need to wait
            if ( it1->second->_pevents != 0 )
                waitingfds++;
        }
 
        // Need a second scan, as we need to deregister all entries
        // which had errors
 
        for ( it1 = _internal.begin() ; it1 != _internal.end() ; ) {
            it2 = it1;
            it1++;
 
            if ( efds.is_set (it2->first) ) {
                // I hope this does not invalidate it1
                _internal.erase (it2);
                _nfds--;
            }
        }
 
        // Only queue event for processing with poll if items remain
        LOG_DEBUG (("Poll_select: waitingfds=%d", waitingfds));
        if ( waitingfds ) {
            _queued = true;
            _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO, true);
        } else {
            _queued = false;
        }
 
        pthread_mutex_unlock(&_mutex);
    }
 
    int Poll_select::Register (User *user)
    {
        int rc;
        endpoint_t fd = user->_fd;
 
        LOG_DEBUG (("Poll_select: Register: " ENDPOINT_FMT, fd));
 
        if ( (rc = pthread_mutex_lock (&_mutex)) != 0 ) 
            return rc;
 
        if ( _terminating ) {
            rc = 0;
        } else {
 
            // Check to see if fd already registered
            polldatamap_t::iterator entry = _internal.find (fd);
 
            if ( entry == _internal.end() ) {
                _internal[fd] = user;
                _nfds++;
 
                rc = 0;
 
            } else {
                rc = -1;
            }
        }
 
        pthread_mutex_unlock (&_mutex);
        return rc;
    }
 
    void Poll_select::Deregister (User *user)
    {
        endpoint_t fd = user->_fd;
 
        LOG_DEBUG (("Poll_select: Deregister: " ENDPOINT_FMT, fd));
 
        if ( (pthread_mutex_lock (&_mutex)) != 0 ) 
            return;
 
        polldatamap_t::iterator entry = _internal.find (fd);
 
        if ( entry != _internal.end() ) {
 
            // Do NOT cancel the event, as the poll user should be the
            // only user of this.
 
            _internal.erase (entry);
            _nfds--;
 
            // Do not wake poll(). If the end point were in use, any events
            // will be cancelled.
        }
 
        pthread_mutex_unlock (&_mutex);
    }
 
    void Poll_select::Control (User *user)
    {
        endpoint_t fd = user->_fd;
 
        LOG_DEBUG (("Poll_poll: Control: " ENDPOINT_FMT " %x", fd, user->_uevents));
 
        if ( (pthread_mutex_lock (&_mutex)) != 0 ) 
            return;
 
        if ( _terminating ) {
            pthread_mutex_unlock (&_mutex);
            return;
        }
 
        bool wake = false;
        unsigned old_pevents = user->_pevents;
 
        LOG_DEBUG (("Control: queued=%d inpoll=%d wakepending=%d uevents=%x pevents=%x",
                    _queued, _inpoll, _wakepending,
                    user->_uevents, old_pevents));
 
        user->_pevents = user->_uevents;
 
        // Check for events set in new mask which are NOT in old
        // If set need to wake poll unless wake pending or not in poll
        if ( _inpoll && !_wakepending &&
             (user->_uevents & (~old_pevents)) != 0 ) {
            wake = true;
 
            // Set here, in critical section, even though it will
            // not actually be true until later
            _wakepending = true;
 
            LOG_DEBUG (("Control: waking polling thread"));
 
        } else if ( !_inpoll && !_queued && user->_uevents != 0 ) {
            // If there is no event queued on event manager, and there
            // are events here to be waited for, then need to queue
 
            LOG_DEBUG (("Control: adding Poll event"));
            _queued = true;
            _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO);
        }
 
        pthread_mutex_unlock (&_mutex);
 
        if ( wake ) {
            // Then wake by writing byte
            char bytev[1];
            bytev[0] = 1;
            if ( sendto (_wakefds[1], bytev, 1, 0,
                         reinterpret_cast<sockaddr *>(&_addr), sizeof _addr) < 1 )
                MSG_IOevent_Pollerror_LOG ("write");
        }
    }
 
    Provider *CreatePollselectProvider ()
    {
        return new Poll_select();
    }
}