Poll_poll.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_poll.C
//
// Poll provider using poll
//
// @VERSION@
 
#include <errno.h>
#include <map>
 
#include "Event_p.h"
 
#ifdef HAVE_POLL
 
namespace Poll {
 
    typedef std::map<int, User *> polldatamap_t;
 
    class Poll_poll : public Provider {
    private:
        pthread_mutex_t       _mutex;        
 
        polldatamap_t            _internal;     
        struct pollfd           *_ufds;         
        int                      _nfds;         
        int                      _sfds;         
 
        int                      _wakefds[2];   
        bool                     _inpoll;       
        bool                     _wakepending;  
        bool                     _terminating;  
        bool                     _queued;       
        ::Event::Manager        *_manager;      
        ProviderEvent            _myevent;      
        ProviderEvent            _termevent;    
 
    public:
        Poll_poll ();
 
        virtual ~Poll_poll ();
 
        virtual int Register (User *user);
 
        virtual void Deregister (User *user);
 
        virtual void Control (User *user);
 
        virtual void Deliver (pollAction *msg);
    };
 
    Poll_poll::Poll_poll ()
    {
        pthread_mutex_init (&_mutex, NULL);
 
        pollAction action;
 
        action.terminate = false;
        _myevent.SetUserData (this, action);
 
        action.terminate = true;
        _termevent.SetUserData (this, action);
 
        _ufds = 0;
        _nfds = 1;  // Count pipe, although not yet open
        _sfds = 0;
 
        _inpoll = false;
        _wakepending = false;
 
        _terminating = false;
        _queued      = false;
 
        // Leave creating pipe until we can report error
        _wakefds[0] = _wakefds[1] = -1;
 
        _manager = ::Event::Manager::GetManager();
 
        _manager->QueueAtExit (&_termevent);
    }
 
    Poll_poll::~Poll_poll ()
    {
        pthread_mutex_destroy (&_mutex);
        if ( _ufds ) delete[] _ufds;
        if ( _wakefds[0] >= 0 ) close (_wakefds[0]);
        if ( _wakefds[1] >= 0 ) close (_wakefds[1]);
    }
 
    void Poll_poll::Deliver (pollAction *msg)
    {
        int rc;
        LOG_DEBUG (( "Poll_poll::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->second->_event.SendAtExit (cancelmsg);
 
            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");
            }
            return;
        } // end of handling termination
 
        if ( _wakefds[0] == -1 && pipe (_wakefds) != 0 ) {
            MSG_IOevent_Pollerror_LOG ("pipe");
            _manager->QueueAt (&_myevent, 1000);
            _queued = true;
            pthread_mutex_unlock(&_mutex);
            return;
        }
 
        if ( _ufds == 0 ) {
            _sfds = _nfds > 1024 ? _nfds : 1024;
 
            _ufds = new pollfd[_sfds];
        } else if ( _nfds > _sfds ) {
 
            delete[] _ufds;
 
            while ( _sfds < _nfds ) _sfds += 1024;
 
            _ufds = new pollfd[_sfds];
        }
 
        // First entry is always for wake-up FD
        _ufds[0].fd      = _wakefds[0];
        _ufds[0].events  = POLLIN;
        _ufds[0].revents = 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 < 0 || scan->second->_pevents == 0 ) continue;
 
            _ufds[nfds].fd     = scan->first;
            _ufds[nfds].events = scan->second->_pevents;
            _ufds[nfds].revents = 0;
 
            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();
 
        LOG_DEBUG (("Poll_poll::Wait: poll nfds=%d", nfds));
        rc = poll (_ufds, nfds, timeout);
        LOG_DEBUG (("Poll_poll::Wait poll()=%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;
                
            if ( eno == EINTR ) {
                LOG_DEBUG (( "Poll_poll::Wait: poll EINTR"));
                // Queue event to self again.
                _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO);
 
            } else {
                MSG_IOevent_Pollerror_LOG ("poll");
                // Try again in one second
                _manager->QueueAt (&_myevent, 1000);
            }
 
            _queued = true;
            pthread_mutex_unlock(&_mutex);
            return;
        }
 
        // Soak up wakeup bytes outside critical section
        if ( _ufds[0].revents & POLLIN ) {
            char buffer[128];
            if ( read (_wakefds[0], buffer, sizeof buffer) < 0 )
                MSG_IOevent_Pollerror_LOG ("read");
            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;
 
        for ( int i = 1; i < nfds; i++ )
            if ( _ufds[i].revents ) {
                polldatamap_t::iterator entry = _internal.find (_ufds[i].fd);
 
                // Check was not unregistered in the interim
                if ( entry == _internal.end() ) continue;
 
                // Clear requested events
                // Do this in case the event reaches the user after
                // the poller event gets reexcuted.
                entry->second->_pevents = 0;
 
                // Translate events into external mask
                pollmsg these_events;
                these_events.events = _ufds[i].revents;
 
                // 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_poll: send %d events %x",
                            entry->first, these_events.events));
 
                entry->second->_event.Send(these_events);
 
                // If the events include an error then the endpoint
                // is automatically deregistered
                if ( these_events.events & (POLLERR|POLLHUP|POLLNVAL) ) {
                    _internal.erase (entry);
                    _nfds--;
                }
            }
 
        // Count FDs which remain
        // Need to find if still waiting FDs from the real data
        // in case this was updated while in poll
 
        unsigned waitingfds = 0;
 
        for ( scan = _internal.begin(); scan != _internal.end(); ++scan )
            if ( scan->first >= 0 && scan->second->_pevents != 0 )
                waitingfds++;
 
        // Only queue event for processing with poll if items remain
        LOG_DEBUG (("Poll_poll: waitingfds=%d", waitingfds));
        if ( waitingfds ) {
            _queued = true;
            _manager->Queue (&_myevent, ::Event::Manager::MAXPRIO, true);
        } else {
            _queued = false;
        }
 
        pthread_mutex_unlock(&_mutex);
    }
 
    int Poll_poll::Register (User *user)
    {
        int rc;
        int fd = user->_fd;
 
        LOG_DEBUG (("Poll_poll: Register: %d", 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_poll::Deregister (User *user)
    {
        int fd = user->_fd;
 
        LOG_DEBUG (("Poll_poll: Deregister: %d", 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_poll::Control (User *user)
    {
        int fd = user->_fd;
 
        LOG_DEBUG (("Poll_poll: Control: %d %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 ( write (_wakefds[1], bytev, 1) < 0 )
                MSG_IOevent_Pollerror_LOG ("write");
        }
    }
 
    Provider *CreatePollpollProvider ()
    {
        return new Poll_poll();
    }
}
 
#else
 
namespace Poll {
 
    // Don't have poll() therefore cannot have this kind of provider
    Provider *CreatePollpollProvider ()
    {
        return 0;
    }
}
#endif