EventLoop.cc

Go to the documentation of this file.

/*
 * Copyright 2006-2009 Savarese Software Research Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.savarese.com/software/ApacheLicense-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <vector>

#include <boost/multi_index_container.hpp>
#include <boost/multi_index/mem_fun.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/hashed_index.hpp>

#include <ssrc/wisp/service/EventLoop.h>
#include <ssrc/wisp/service/EventPort.h>

__BEGIN_NS_SSRC_WISP_SERVICE

using namespace std::rel_ops;

enum { ByExpiration, ByIdentity };

typedef EventPort::event_type event_type;
typedef EventPort::events_type events_type;
typedef std::vector<event_type> io_event_container;

struct EventLoopState {
  typedef boost::multi_index_container<
    EventHandler*,
    boost::multi_index::indexed_by<
      boost::multi_index::ordered_non_unique<
        boost::multi_index::const_mem_fun<EventHandler, const TimeValue &,
                                          &EventHandler::expiration> >,
      boost::multi_index::hashed_unique<boost::multi_index::identity<EventHandler*> >
      > > event_handler_container;

  bool running;
  EventPort port;
  unsigned int io_handler_count;
  io_event_container io_events;
  event_handler_container event_handlers;

  EventLoopState() :
    running(false), port(),
    io_handler_count(0), io_events(), event_handlers()
  { }
};

typedef EventLoopState::event_handler_container event_handler_container;

const EventLoop::EventIO EventLoop::None   = EventPort::None;
const EventLoop::EventIO EventLoop::Read   = EventPort::Read;
const EventLoop::EventIO EventLoop::Write  = EventPort::Write;
const EventLoop::EventIO EventLoop::Error  = EventPort::Error;
const EventLoop::EventIO EventLoop::Hangup = EventPort::Hangup;
/*
const EventLoop::EventIO EventLoop::Signal  = EventPort::Signal;
const EventLoop::EventIO EventLoop::Timeout = EventPort::Timeout;
*/

EventLoop::EventLoop() SSRC_DECL_THROW(std::runtime_error) :
  _state(new EventLoopState())
{ }

EventLoop::~EventLoop() { delete _state; }

unsigned int EventLoop::count_handlers() {
  return _state->event_handlers.size();
}

unsigned int EventLoop::count_io_handlers() {
  return _state->io_handler_count;
}

bool EventLoop::running() {
  return _state->running;
}

void EventLoop::add_handler(EventHandler & handler,
                            const int events,
                            const TimeValue & timeout,
                            const bool once)
  SSRC_DECL_THROW(std::runtime_error)
{
  if(handler.event_descriptor() != EventHandler::NoDescriptor && events != None) {
    if(_state->port.add(handler.event_descriptor(), events, &handler, once)) {
      if(_state->io_events.size() == _state->io_handler_count)
        _state->io_events.push_back(event_type());
      ++_state->io_handler_count;
    }
  }

  handler._handled_timeout = false;
  handler._once    = once;
  handler._timeout = timeout;

#if defined(WISP_HAVE_NONPERSISTENT_EVENTS) || defined(WISP_HAVE_KQUEUE)
  handler._events = events;
#endif

  if(timeout == InfiniteTimeValue)
    handler._expiration = InfiniteTimeValue;
  else {
    TimeValue expiration;
    expiration.now_mono()+=timeout;
    handler._expiration = expiration;
  }

  _state->event_handlers.insert(&handler);

}

void EventLoop::remove_handler(EventHandler & handler) {
  const bool is_io_event =
    (handler.event_descriptor() != EventHandler::NoDescriptor);

  if(is_io_event) {
    // We do not check return value of remove because it returns false
    // only if the descriptor is not associated with the port.  When
    // that happens (should happen only on HAVE_NONPERSISTENT_EVENTS
    // platforms such as Solaris), we want to remove the handler anyway.
    // Other errors throw an exception.
#if defined(WISP_HAVE_KQUEUE)
    _state->port.remove(handler.event_descriptor(), handler.events());
#else
    _state->port.remove(handler.event_descriptor());
#endif
  }

  // CAREFUL!  The order of the tests matters as we erase as a side-effect
  // and don't want to decrement io_handler_count for timeout handlers.
  if(_state->event_handlers.get<ByIdentity>().erase(&handler) > 0 &&
     is_io_event)
  {
    --_state->io_handler_count;
    // We can't shrink the vector because it could mess up start loop.
    //_state->io_events.pop_back();
  }
}

void EventLoop::start() {
  EventInfo info(*this);
  TimeValue now;
  io_event_container & io_events = _state->io_events;
  event_handler_container & event_handlers = _state->event_handlers;
  EventPort & port = _state->port;

  _state->running = true;
  while(count_io_handlers() > 0 && _state->running) {
    int fd_count, timeout = -1;
    event_handler_container::iterator it = event_handlers.begin();
    EventHandler *handler = *it;

    if(io_events.size() != count_io_handlers())
      io_events.resize(count_io_handlers());

    if(handler->has_timeout()) {
      now.now_mono();

      if(now >= handler->expiration())
        timeout = 0;
      else
        timeout = (handler->expiration() - now).to_milliseconds();
    }

    fd_count = port.wait(&io_events[0], count_io_handlers(), timeout);

    if(fd_count < 0) {
      // TODO: log exit condition, including errno and strerror(errno).
      stop();
      break;
    }

    info._now = now.now_mono();

    for(int i = 0; i < fd_count; ++i) {
      handler = EventPort::get_handler(io_events[i]);
      info._io_events = EventPort::get_events(io_events[i]);
      info._timeout = (handler->expiration() <= now);

      /*
        if(info.signal_event())
        handler->handle_signal(info);
      */

      if(info.error_event())
        handler->handle_error(info);

      if(info.hangup_event())
        handler->handle_hangup(info);

      if(info.read_event())
        handler->handle_read(info);

      if(info.write_event())
        handler->handle_write(info);

      if(info.timeout_event()) {
        handler->handle_timeout(info);
        handler->_handled_timeout = true;
      } else if(handler->_once) {
        remove_handler(*handler);
      }

#if defined(WISP_HAVE_NONPERSISTENT_EVENTS)
      // Solaris ports require reassociation after every event.
      // TODO: log error and decrement io_handler_count if this fails?
      if(!handler->_once) {
        port.add(handler->event_descriptor(), handler->_events, handler,
                 handler->_once);
      }
#endif
    }

    info._timeout = true;
    info._io_events = None;

    it = event_handlers.begin();
    std::iterator_traits<event_handler_container::iterator>::difference_type
      count = std::distance(it, event_handlers.upper_bound(now));

    while(count-- > 0) {
      handler = *it;
      it = event_handlers.erase(it);

      // Insert beforehand to allow handler the option to remove itself.
      if(!handler->_once) {
        handler->_expiration = now + handler->timeout();
        event_handlers.insert(handler);
      } else
        remove_handler(*handler);

      if(!handler->_handled_timeout)
        handler->handle_timeout(info);
      else
        handler->_handled_timeout = false;
    }
  }
}

void EventLoop::stop() {
  // TODO: do something to wake up the poll.
  _state->running = false;
}


__END_NS_SSRC_WISP_SERVICE