Poll::Poll_select Class Reference

! Poll provider using poll() itself. More...

Inheritance diagram for Poll::Poll_select:

Public Member Functions
	Poll_select ()
	Tell users we are stopping.
virtual int	Register (User *user)
	Register an endpoint.
virtual void	Deregister (User *user)
	Deregister endpoint.
virtual void	Control (User *user)
	Set the events which are interesting for end point.
virtual void	Deliver (pollAction *msg)
	Interface to itself, via Event Manager.
Public Member Functions inherited from Poll::Provider
virtual	~Provider ()
	Virtual destructor.

Additional Inherited Members
Static Public Member Functions inherited from Poll::Provider
static Provider *	GetPollService ()
	Implement as a singleton.
static int	ProviderFactory (const char *type, MSGstruct *msg)
	Factory method.for stream provider.

Detailed Description

! Poll provider using poll() itself.

Definition at line 88 of file Poll_select.C.

Constructor & Destructor Documentation

Poll_select()

Poll::Poll_select::Poll_select

(

)

Tell users we are stopping.

Definition at line 127 of file Poll_select.C.

    {
        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::Poll_select::~Poll_select ( )

virtual

Definition at line 155 of file Poll_select.C.

    {
        pthread_mutex_destroy (&_mutex);
        if ( _wakefds[0] != INVALID_ENDPOINT) closesocket (_wakefds[0]);
        if ( _wakefds[1] != INVALID_ENDPOINT) closesocket (_wakefds[1]);
    }

Member Function Documentation

Register()

int Poll::Poll_select::Register ( User *  user )

virtual

Register an endpoint.

Returns

zero on success, -1 if already registered or other system errno value

Implements Poll::Provider.

Definition at line 492 of file Poll_select.C.

    {
        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;
    }

References Poll::User::_fd.

Deregister()

void Poll::Poll_select::Deregister ( User *  user )

virtual

Deregister endpoint.

Implements Poll::Provider.

Definition at line 524 of file Poll_select.C.

    {
        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);
    }

References Poll::User::_fd.

Control()

void Poll::Poll_select::Control ( User *  user )

virtual

Set the events which are interesting for end point.

Implements Poll::Provider.

Definition at line 550 of file Poll_select.C.

    {
        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");
        }
    }

References Poll::User::_fd, Poll::User::_pevents, and Poll::User::_uevents.

Deliver()

void Poll::Poll_select::Deliver ( pollAction *  msg )

virtual

Interface to itself, via Event Manager.

Implements Poll::Provider.

Definition at line 228 of file Poll_select.C.

    {
        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);
    }

References Poll::Event_Err, Poll::Event_Out, getNow(), and timespec_diff().

---

The documentation for this class was generated from the following file:

• Poll_select.C