Poll_port.C

//  Copyright (c) 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_port.C
//
// Poll provider using Solaris Ports
 
 
#include "Event_p.h"
 
#ifdef HAVE_PORT_CREATE
 
#ifdef HAVE_PORT_H
#include <port.h>
#endif
 
namespace Poll {
 
    typedef std::set<User *> polldatamap_t;
 
    class Poll_port : public Provider {
    private:
        static const int port_maxevents = 256;
 
        pthread_mutex_t       _mutex;        
 
        polldatamap_t            _internal;     
 
        int                      _portfd;       
        bool                     _inpoll;       
        bool                     _wakepending;  
        bool                     _terminating;  
        bool                     _queued;       
        ::Event::Manager        *_manager;      
        ProviderEvent            _myevent;      
        ProviderEvent            _termevent;    
 
        int PortControl (User *user) {
            int port_events = 0;
 
            if ( user->_pevents & Event_In )
                port_events |= POLLIN;
 
            if ( user->_pevents & Event_Out )
                port_events |= POLLOUT;
 
            // Only associate if events of interest
            if ( port_events &&
                 port_associate (_portfd, PORT_SOURCE_FD,
                                 user->_fd, user->_pevents,
                                 static_cast<void *>(user)) != 0 ) {
                MSG_IOevent_Pollerror_LOG ("port_associate");
                return NOTOK;
            }
 
            return OK;
        }
 
    public:
        Poll_port ();
 
        virtual ~Poll_port ();
 
        virtual int Register (User *user);
 
        virtual void Deregister (User *user);
 
        virtual void Control (User *user);
 
        virtual void Deliver (pollAction *msg);
    };
 
    Poll_port::Poll_port ()
    {
        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;
 
        _portfd = port_create ();
        if ( _portfd < 0 )
            MSG_IOevent_Pollerror_LOG ("port_create");
 
        _manager = ::Event::Manager::GetManager();
 
        _manager->QueueAtExit (&_termevent);
    }
 
    Poll_port::~Poll_port ()
    {
        pthread_mutex_destroy (&_mutex);
        if ( _portfd >= 0 ) close (_portfd);
    }
 
    void Poll_port::Deliver (pollAction *msg)
    {
        int rc;
        LOG_DEBUG (( "Poll_port::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);
 
            bool wake = false;
            if ( _inpoll && !_wakepending ) {
                 wake = true;
                _wakepending = true;
            }
 
            pthread_mutex_unlock(&_mutex);
 
            if ( wake && port_alert (_portfd, PORT_ALERT_SET, -1, 0) != 0 )
                MSG_IOevent_Pollerror_LOG ("port_alert");
 
            return;
        } // end of handling termination
 
        const timespec *wakeat = _manager->NextEventDue();
        timespec *timep = 0;
        timespec now;
 
        if ( wakeat ) {
            getNow(now);
            // timespec_diff returns difference in microseconds.
            // returns 0 if delay preceeds now.
            int timeout = timespec_diff (wakeat, &now);
            now.tv_sec = timeout/1000000;
            now.tv_nsec = (timeout%1000000)*1000;
            LOG_DEBUG ((" timeout=%d", timeout));
            timep = &now;
        }
 
        _inpoll = true; // about to be true
 
        pthread_mutex_unlock(&_mutex);
 
        // Tell event manager we might go away for a while
        _manager->Blocking();
 
        port_event_t events[port_maxevents];
 
        // Attempt wait for (at least) one event
        unsigned nfds = 1;
        LOG_DEBUG (("Poll_port::Wait: port_getn"));
        rc = port_getn (_portfd, events,port_maxevents, &nfds, timep);
        int eno = errno;
        LOG_DEBUG (("Poll_port::Wait port_getn()=%d (%d) nfds=%u",
                    rc, errno, nfds));
 
        // Tell event manager we are back
        _manager->Unblocked();
 
        // Timeout and EINTR are not treated as errors.
        if ( rc != 0 && eno != ETIME && eno != EINTR ) {
            if ( (rc = pthread_mutex_lock (&_mutex)) != 0 ) {
                MSG_IOevent_Pollthread_LOG (rc);
                _manager->QueueAt (&_myevent, 10000);
                _queued = true;
                _inpoll = false;
                return;
            }
 
            _inpoll = false;
 
            errno = eno;
            MSG_IOevent_Pollerror_LOG ("port_get");
            // 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].portev_source == PORT_SOURCE_ALERT ) {
                // Must be terminating, no point in carrying on
                break;
            } else if ( events[i].portev_source == PORT_SOURCE_FD ) {
                User *user = static_cast<User *>(events[i].portev_user);
 
                polldatamap_t::iterator entry =
                    _internal.find (user);
 
                // Check was not unregistered in the interim
                if ( entry == _internal.end() ) continue;
 
                // Translate events into external mask
                // If error, clear all events
                pollmsg these_events;
                if ( events[i].portev_events & POLLIN )
                    these_events.events |= Event_In;
 
                if ( events[i].portev_events & POLLOUT )
                    these_events.events |= Event_Out;
 
                if ( events[i].portev_events & POLLERR )
                    these_events.events |= Event_Err;
 
                if ( events[i].portev_events & POLLHUP )
                    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_port: send %d events %x",
                            user->_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) ) {
                    // No need to dissociate
                    _internal.erase (entry);
                }
            }
        }
 
        // Only queue event for processing with poll 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_port::Register (User *user)
    {
        int rc;
 
        LOG_DEBUG (("Poll_port: Register: %d", user->_fd));
 
        if ( (rc = pthread_mutex_lock (&_mutex)) != 0 )
            return rc;
 
        if ( _terminating ) {
            rc = 0;
        } else if ( _portfd < 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);
 
            } else {
                rc = -1;
            }
        }
 
        // Have an 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_port::Deregister (User *user)
    {
        LOG_DEBUG (("Poll_port: Deregister: %d", user->_fd));
 
        if ( (pthread_mutex_lock (&_mutex)) != 0 )
            return;
 
        polldatamap_t::iterator entry = _internal.find (user);
 
        if ( entry != _internal.end() ) {
 
            // Disassociate if currently associated
            // Note: ENOENT is not regarded as an error,
            // It means the FD was not actually associated
            if (user->_pevents && 
                port_dissociate (_portfd, PORT_SOURCE_FD, user->_fd) &&
                errno != ENOENT )
                MSG_IOevent_Pollerror_LOG ("port_dissociate");
 
            // Do NOT cancel the event, as the poll user should be the
            // only user of this.
 
            _internal.erase (entry);
        }
 
        pthread_mutex_unlock (&_mutex);
    }
 
    void Poll_port::Control (User *user)
    {
        int fd = user->_fd;
 
        LOG_DEBUG (("Poll_port: Control: %d %x", fd, user->_uevents));
 
        if ( _terminating ) return;
 
        user->_pevents = user->_uevents;
        (void) PortControl (user);
    }
 
    Provider *CreatePollportProvider ()
    {
        return new Poll_port();
    }
}
 
#else
 
namespace Poll {
 
    // Don't have Solaris ports therefore cannot have this kind of provider
    Provider *CreatePollportProvider ()
    {
        return 0;
    }
}
#endif