beacon/event_loop.cpp

   1 /**
   2  * beacon
   3  * Author: Lukas Krejci <krejci.l@centrum.cz>, (C) 2008
   4  * Copyright: See COPYING file that comes with this distribution
   5  */
   6
   7 #include <beacon/event_loop.hpp>
   8 #include <algorithm>
   9
  10 namespace beacon {
  11
  12
  13 //used by dequeue method to decide whether to remove an invokable from the queue
  14 bool token_compare(void const * token, invokable * inv) {
  15     return token == inv->token();
  16 }
  17
  18 void invoke_and_delete(invokable * inv) {
  19     try {
  20         inv->invoke();
  21     } catch (...) {
  22         //TODO figure out what to do here
  23     }
  24     delete inv;
  25 }
  26
  27 void delete_invokable(invokable * inv) {
  28     delete inv;
  29 }
  30
  31 void event_loop::enqueue(ev_type inv) {
  32     if (_running) {
  33         //enter the critical section
  34         detail::wait(_queue_guard);
  35
  36         _events.push_back(inv);
  37
  38         //leave the critical section
  39         detail::notify(_queue_guard);
  40     }
  41 }
  42
  43 unsigned event_loop::dequeue(void const * token) {
  44     //enter the critical section
  45     detail::wait(_queue_guard);
  46
  47     std::deque<ev_type>::iterator const & first_bad =
  48             remove_if(_events.begin(), _events.end(),
  49                       boost::bind(&token_compare, token, _1));
  50
  51     _dequeued.assign(first_bad, _events.end());
  52
  53     _events.erase(first_bad, _events.end());
  54
  55     unsigned size = _dequeued.size();
  56
  57     //leave the critical section
  58     detail::notify(_queue_guard);
  59
  60     return size;
  61 }
  62
  63 void event_loop::join() {
  64     if (_thread != 0) {
  65         _running = false;
  66
  67         while(!_finished) boost::thread::yield();
  68
  69         _thread->join();
  70     }
  71 }
  72
  73 void event_loop::run() {
  74     std::deque<ev_type> evs_cpy;
  75     std::deque<ev_type> deq_cpy;
  76
  77     while(_running) {
  78         detail::wait(_queue_guard);
  79
  80         unsigned evs = _events.size();
  81         unsigned des = _dequeued.size();
  82
  83         //check if there's anything to do at all...
  84         if ((evs | des) == 0) {
  85             detail::notify(_queue_guard);
  86             boost::thread::yield();
  87             continue;
  88         }
  89
  90         //copy the current events... this is to block the enquing as little
  91         //as possible.
  92         if (evs > 0) {
  93             evs_cpy = _events;
  94             _events.clear();
  95         }
  96
  97         if (des > 0) {
  98             deq_cpy = _dequeued;
  99             _dequeued.clear();
 100         }
 101
 102         detail::notify(_queue_guard);
 103
 104         //a little bit weird situation might occur here.
 105         //when an event is added to the queue while the processing
 106         //thread is in this code segment, it won't be processed
 107         //until the processing thread is woken up again when another
 108         //event is enqued or it is dying.
 109
 110         if (evs > 0) {
 111             std::for_each(evs_cpy.begin(), evs_cpy.end(), &invoke_and_delete);
 112         }
 113
 114         if (des > 0) {
 115             std::for_each(deq_cpy.begin(), deq_cpy.end(), &delete_invokable);
 116         }
 117
 118         boost::thread::yield();
 119     }
 120
 121     //flush the rest of the events before we die...
 122
 123     detail::wait(_queue_guard);
 124
 125     std::for_each(_events.begin(), _events.end(), &invoke_and_delete);
 126     _events.clear();
 127
 128     std::for_each(_dequeued.begin(), _dequeued.end(), &delete_invokable);
 129     _dequeued.clear();
 130
 131     _finished = true;
 132
 133     detail::notify(_queue_guard);
 134 }
 135
 136 } //namespace