EventLoop.cc

Go to the documentation of this file.

  /* Copyright 2006-2009, 2016 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(std::make_unique<EventLoopState>())
  { }
  
  EventLoop::~EventLoop() = default;
  
  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 and with automatically removed events
   // such as EV_ONESHOT on FreeBSD), we want to remove the handler anyway.
   // Other errors throw an exception.
  #if defined(WISP_HAVE_KQUEUE)
   if(!handler._once) {
   _state->port.remove(handler.event_descriptor(), handler.events());
   }
  #elif defined(WISP_HAVE_NONPERSISTENT_EVENTS)
   if(!handler._once) {
   _state->port.remove(handler.event_descriptor());
   }
  #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