ASSA::Reactor Class Reference

#include <Reactor.h>

Public Member Functions
	Reactor ()
	Constructor. More...
	~Reactor ()
	Destructor. More...
TimerId	registerTimerHandler (EventHandler *eh_, const TimeVal &tv_, const std::string &name_="<unknown>")
	Register Timer Event handler with Reactor. More...
bool	registerIOHandler (EventHandler *eh_, handler_t fd_, EventType et_=RWE_EVENTS)
	Register I/O Event handler with Reactor. More...
bool	removeHandler (EventHandler *eh_, EventType et_=ALL_EVENTS)
	Remove Event handler from reactor for either all I/O events or timeout event or both. More...
bool	removeTimerHandler (TimerId id_)
	Remove Timer event from the queue. More...
bool	removeIOHandler (handler_t fd_)
	Remove IO Event handler from reactor. More...
void	waitForEvents (void)
	Main waiting loop that blocks indefinitely processing events. More...
void	waitForEvents (TimeVal *tv_)
	Wait for events for time specified. More...
void	stopReactor (void)
	Stop Reactor's activity. More...
void	deactivate (void)
	Deactivate Reactor. More...

Private Types
typedef std::map< u_int, EventHandler * >	Fd2Eh_Map_Type
	no cloning More...
typedef Fd2Eh_Map_Type::iterator	Fd2Eh_Map_Iter

Private Member Functions
	Reactor (const Reactor &)
Reactor &	operator= (const Reactor &)
	no cloning More...
void	adjust_maxfdp1 (handler_t fd_)
	Adjust maxfdp1 in a portable way (win32 ignores maxfd alltogether). More...
bool	handleError (void)
	Handle error in select(2) loop appropriately. More...
bool	dispatch (int minimum_)
	Notify all EventHandlers registered on respecful events occured. More...
int	isAnyReady (void)
	Return number of file descriptors ready accross all sets. More...
bool	checkFDs (void)
	Check mask for bad file descriptors. More...
void	dispatchHandler (FdSet &mask_, Fd2Eh_Map_Type &fdSet_, EH_IO_Callback callback_)
	Call handler's callback and, if callback returns negative value, remove it from the Reactor. More...
void	calculateTimeout (TimeVal *&howlong_, TimeVal *maxwait_)
	Calculate closest timeout. More...

Private Attributes
int	m_fd_setsize
	Max number of open files per process. More...
handler_t	m_maxfd_plus1
	Max file descriptor number (in all sets) plus 1. More...
bool	m_active
	Flag that indicates whether Reactor is active or had been stopped. More...
Fd2Eh_Map_Type	m_readSet
	Event handlers awaiting on READ_EVENT. More...
Fd2Eh_Map_Type	m_writeSet
	Event handlers awaiting on WRITE_EVENT. More...
Fd2Eh_Map_Type	m_exceptSet
	Event handlers awaiting on EXCEPT_EVENT. More...
MaskSet	m_waitSet
	Handlers to wait for event on. More...
MaskSet	m_readySet
	Handlers that are ready for processing. More...
TimerQueue	m_tqueue
	The queue of Timers. More...

Detailed Description

Definition at line 57 of file Reactor.h.

Member Typedef Documentation

Fd2Eh_Map_Iter

typedef Fd2Eh_Map_Type::iterator ASSA::Reactor::Fd2Eh_Map_Iter

private

Definition at line 155 of file Reactor.h.

Fd2Eh_Map_Type

typedef std::map<u_int, EventHandler*> ASSA::Reactor::Fd2Eh_Map_Type

private

no cloning

Definition at line 154 of file Reactor.h.

Constructor & Destructor Documentation

Reactor() [1/2]

Reactor::Reactor

(

)

Constructor.

Maximum number of sockets supported (per process) Win32 defines it to 64 in winsock2.h.

Initialize winsock2 library

Definition at line 24 of file Reactor.cpp.

References m_fd_setsize, ASSA::REACTTRACE, trace_with_mask, and ~Reactor().

           :
    m_fd_setsize  (1024),
    m_maxfd_plus1 (0),
    m_active      (true)
{
    trace_with_mask("Reactor::Reactor",REACTTRACE);

#if defined(WIN32)
    m_fd_setsize = FD_SETSIZE;

#else  // POSIX
    struct rlimit rlim;
    rlim.rlim_max = 0;

    if ( getrlimit (RLIMIT_NOFILE, &rlim) == 0 ) {
        m_fd_setsize = rlim.rlim_cur;
    }
#endif

#if defined (WIN32)
    WSADATA data;
    WSAStartup (MAKEWORD (2, 2), &data);
#endif
}

~Reactor()

Reactor::~Reactor

(

)

Destructor.

Definition at line 55 of file Reactor.cpp.

References deactivate(), m_exceptSet, m_readSet, m_writeSet, ASSA::REACTTRACE, registerTimerHandler(), and trace_with_mask.

Referenced by Reactor().

{
    trace_with_mask("Reactor::~Reactor",REACTTRACE);

    m_readSet.clear   ();
    m_writeSet.clear  ();
    m_exceptSet.clear ();
    deactivate ();
}

Reactor() [2/2]

ASSA::Reactor::Reactor ( const Reactor &  )

private

Member Function Documentation

adjust_maxfdp1()

void Reactor::adjust_maxfdp1 ( handler_t  fd_ )

private

Adjust maxfdp1 in a portable way (win32 ignores maxfd alltogether).

If the socket descriptor that has just been eliminated was the maxfd+1, we readjust to the next highest.

Win32 implementation of select() ignores this value altogether.

Definition at line 701 of file Reactor.cpp.

References DL, m_maxfd_plus1, m_waitSet, ASSA::MaskSet::max_fd(), ASSA::REACT, ASSA::REACTTRACE, and trace_with_mask.

Referenced by removeHandler(), removeIOHandler(), and stopReactor().

{
#if !defined (WIN32)  /* POSIX */

    trace_with_mask("Reactor::adjust_maxfdp1", REACTTRACE);

    if (m_maxfd_plus1 == fd_ + 1)
    {
        m_maxfd_plus1 = m_waitSet.max_fd () + 1;
        DL((REACT,"maxfd+1 adjusted to %d\n", m_maxfd_plus1));
    }
#endif
}

calculateTimeout()

void Reactor::calculateTimeout ( TimeVal *&  howlong_, TimeVal *  maxwait_  )

private

Calculate closest timeout.

If TimerQueue is not empty, then return smallest of maxtimeout and first in the queue. Otherwise, return maxtimeout.

Parameters

maxwait_	(in) how long we are expected to wait for event(s).
howlong_	(out) how long we are going to wait.

Definition at line 421 of file Reactor.cpp.

References DL, ASSA::TimerQueue::dump(), ASSA::TimeVal::gettimeofday(), ASSA::TimerQueue::isEmpty(), m_tqueue, ASSA::REACT, ASSA::REACTTRACE, ASSA::TimerQueue::top(), trace_with_mask, waitForEvents(), and ASSA::TimeVal::zeroTime().

Referenced by isAnyReady(), and waitForEvents().

{
    trace_with_mask("Reactor::calculateTimeout",REACTTRACE);

    TimeVal now;
    TimeVal tv;

    if (m_tqueue.isEmpty () ) {
        howlong_ = maxwait_;
        goto done;
    }
    now = TimeVal::gettimeofday ();
    tv = m_tqueue.top ();

    if (tv < now) {
        /*---
          It took too long to get here (fraction of a millisecond),
          and top timer had already expired. In this case,
          perform non-blocking select in order to drain the timer queue.
          ---*/
        *howlong_ = 0;
    }
    else {
        DL((REACT,"--------- Timer Queue ----------\n"));
        m_tqueue.dump();
        DL((REACT,"--------------------------------\n"));

        if (maxwait_ == NULL || *maxwait_ == TimeVal::zeroTime ()) {
            *howlong_ = tv - now;
        }
        else {
            *howlong_ = (*maxwait_+now) < tv ? *maxwait_ : tv-now;
        }
    }

 done:
    if (howlong_ != NULL) {
        DL((REACT,"delay (%f)\n", double (*howlong_) ));
    }
    else {
        DL((REACT,"delay (forever)\n"));
    }
}

checkFDs()

bool Reactor::checkFDs ( void  )

private

Check mask for bad file descriptors.

Returns

true if any fd(s) were found and removed; false otherwise

Definition at line 317 of file Reactor.cpp.

References ASSA::FdSet::clear(), DL, handleError(), m_fd_setsize, m_readSet, ASSA::REACT, ASSA::REACTTRACE, removeIOHandler(), ASSA::FdSet::setFd(), and trace_with_mask.

Referenced by handleError(), and removeIOHandler().

{
    trace_with_mask("Reactor::checkFDs",REACTTRACE);

    bool num_removed = false;
    FdSet mask;
    timeval poll = { 0, 0 };

    for (handler_t fd = 0; fd < m_fd_setsize; fd++) {
        if ( m_readSet[fd] != NULL ) {
            mask.setFd (fd);
            if ( ::select (fd+1, &mask, NULL, NULL, &poll) < 0 ) {
                removeIOHandler (fd);
                num_removed = true;
                DL((REACT,"Detected BAD FD: %d\n", fd ));
            }
            mask.clear (fd);
        }
    }
    return (num_removed);
}

deactivate()

void ASSA::Reactor::deactivate ( void  )

inline

Deactivate Reactor.

This function sets internal flag which notifies Reactor's internal event handling loop to abort its activity. It is mostly used when a slow system call is interrupted by the signal handler. The system call will be restarted by OS after control returns from the signal handler. Signal handler (GenServer::handle_signal()) should call this method to delay Reactor's deactivation.

Definition at line 234 of file Reactor.h.

References m_active.

Referenced by ASSA::GenServer::handle_signal(), ASSA::GenServer::stop_service(), and ~Reactor().

{  m_active = false; }

dispatch()

bool Reactor::dispatch ( int  minimum_ )

private

Notify all EventHandlers registered on respecful events occured.

Many UNIX systems will count a particular file descriptor in the ready_ only ONCE, even if it was flagged by select(2) in, say, both read and write masks.

Parameters

minimum_

number of file descriptors ready.

Definition at line 626 of file Reactor.cpp.

References ASSA::ASSAERR, dispatchHandler(), DL, ASSA::MaskSet::dump(), EL, ASSA::TimerQueue::expire(), ASSA::TimeVal::gettimeofday(), ASSA::EventHandler::handle_except(), ASSA::EventHandler::handle_read(), ASSA::EventHandler::handle_write(), ASSA::MaskSet::m_eset, m_exceptSet, m_readSet, m_readySet, ASSA::MaskSet::m_rset, m_tqueue, m_writeSet, ASSA::MaskSet::m_wset, ASSA::REACT, ASSA::REACTTRACE, stopReactor(), and trace_with_mask.

Referenced by dispatchHandler(), and waitForEvents().

{
    trace_with_mask("Reactor::dispatch", REACTTRACE);

    m_tqueue.expire (TimeVal::gettimeofday ());

    if ( ready_ < 0 )
    {
#if !defined (WIN32)
        EL((ASSAERR,"::select(3) error\n"));
#endif
        return (false);
    }
    if ( ready_ == 0 ) {
        return (true);
    }

    DL((REACT,"Dispatching %d FDs.\n",ready_));
    DL((REACT,"m_readySet:\n"));
    m_readySet.dump ();

    /*--- Writes first ---*/
    dispatchHandler (m_readySet.m_wset,
                     m_writeSet,
                     &EventHandler::handle_write);

    /*--- Exceptions next ---*/
    dispatchHandler (m_readySet.m_eset,
                     m_exceptSet,
                     &EventHandler::handle_except);

    /*--- Finally, the Reads ---*/
    dispatchHandler (m_readySet.m_rset,
                     m_readSet,
                     &EventHandler::handle_read);

    return (true);
}

dispatchHandler()

void Reactor::dispatchHandler ( FdSet &  mask_, Fd2Eh_Map_Type &  fdSet_, EH_IO_Callback  callback_  )

private

Call handler's callback and, if callback returns negative value, remove it from the Reactor.

This spot needs re-thinking.

When you have several high data-rate connections sending data at the same time, the one that had connected first would get lower FD number and would get data transfer preference over everybody else who has connected later on.

WIN32 HACK: Without having restarted scan from the beginning, this causes crash due to the fact that firing a callback of EventHandler might have invalidated the iterator (happens with Connector's in a sync mode).

Definition at line 568 of file Reactor.cpp.

References ASSA::FdSet::clear(), dispatch(), DL, ASSA::EventHandler::get_id(), ASSA::FdSet::isSet(), ASSA::REACT, ASSA::REACTTRACE, removeIOHandler(), and trace_with_mask.

Referenced by dispatch(), and waitForEvents().

{
    trace_with_mask("Reactor::dispatchHandler",REACTTRACE);

    int ret = 0;
    handler_t fd;
    EventHandler* ehp = NULL;
    std::string eh_id;

    Fd2Eh_Map_Iter iter = fdSet_.begin ();

    while (iter != fdSet_.end ())
    {
        fd  = (*iter).first;
        ehp = (*iter).second;

        if (mask_.isSet (fd) && ehp != NULL)
        {
            eh_id = ehp->get_id ();
            DL((REACT,"Data detected from \"%s\"(fd=%d)\n",
                eh_id.c_str (), fd));

            ret = (ehp->*callback_) (fd); /* Fire up a callback */

            if (ret == -1) {
                removeIOHandler (fd);
            }
            else if (ret > 0) {
                DL((REACT,"%d bytes pending on fd=%d \"%s\"\n",
                    ret, fd, eh_id.c_str ()));
                //return;   <-- would starve other connections
            }
            else {
                DL((REACT,"All data from \"%s\"(fd=%d) are consumed\n",
                    eh_id.c_str (), fd));
                mask_.clear (fd);
            }
            iter = fdSet_.begin ();
        }
        else {
            iter++;
        }
    }
}

handleError()

bool Reactor::handleError ( void  )

private

Handle error in select(2) loop appropriately.

If commanded to stop, do so

Definition at line 341 of file Reactor.cpp.

References ASSA::ASSAERR, checkFDs(), DL, EL, isAnyReady(), m_active, ASSA::REACT, ASSA::REACTTRACE, and trace_with_mask.

Referenced by checkFDs(), and waitForEvents().

{
    trace_with_mask("Reactor::handleError",REACTTRACE);

    if ( !m_active ) {
        DL((REACT,"Received cmd to stop Reactor\n"));
        return (false);
    }

    /*---
      TODO: If select(2) returns before time expires, with
      a descriptor ready or with EINTR, timeval is not
      going to be updated with number of seconds remaining.
      This is true for all systems except Linux, which will
      do so. Therefore, to restart correctly in case of
      EINTR, we ought to take time measurement before and
      after select, and try to select() for remaining time.

      For now, we restart with the initial timing value.
      ---*/
    /*---
      BSD kernel never restarts select(2). SVR4 will restart if
      the SA_RESTART flag is specified when the signal handler
      for the signal delivered is installed. This means taht for
      portability, we must handle signal interrupts.
      ---*/

    if ( errno == EINTR ) {
        EL((REACT,"EINTR: interrupted select(2)\n"));
        /*
          If I was sitting in select(2) and received SIGTERM,
          the signal handler would have set m_active to 'false',
          and this function would have returned 'false' as above.
          For any other non-critical signals (USR1,...),
          we retry select.
        */
        return (true);
    }
    /*
      EBADF - bad file number. One of the file descriptors does
      not reference an open file to open(), close(), ioctl().
      This can happen if user closed fd and forgot to remove
      handler from Reactor.
    */
    if ( errno == EBADF ) {
        DL((REACT,"EBADF: bad file descriptor\n"));
        return (checkFDs ());
    }
    /*
      Any other error from select
    */
#if defined (WIN32)
    DL ((REACT,"select(3) error = %d\n", WSAGetLastError()));
#else
    EL((ASSAERR,"select(3) error\n"));
#endif
    return (false);
}

isAnyReady()

int Reactor::isAnyReady ( void  )

private

Return number of file descriptors ready accross all sets.

Definition at line 404 of file Reactor.cpp.

References calculateTimeout(), DL, ASSA::MaskSet::dump(), ASSA::MaskSet::m_eset, m_readySet, ASSA::MaskSet::m_rset, ASSA::MaskSet::m_wset, ASSA::FdSet::numSet(), ASSA::REACT, ASSA::REACTTRACE, and trace_with_mask.

Referenced by handleError(), and waitForEvents().

{
    trace_with_mask("Reactor::isAnyReady",REACTTRACE);

    int n = m_readySet.m_rset.numSet () +
        m_readySet.m_wset.numSet () +
        m_readySet.m_eset.numSet ();

    if ( n > 0 ) {
        DL((REACT,"m_readySet: %d FDs are ready for