spice client: creating a general process loop.

The process loop is responsible for: 1) waiting for events 2) timers 3) events queue for
actions that should be performed in the context of the thread and are pushed from other threads.
The benefits:
1) remove duplicity: till now, there was one implementaion of events loop for the channels and
another one for the main thread.
2) timers can be executed on each thread and not only on the main thread.
3) events can be pushed to each thread and not only to the main thread.
In this commit, only the main thread was modified to use the new process loop.

1 files changed, 390 insertions, 0 deletions

diff --git a/client/process_loop.cpp b/client/process_loop.cpp
new file mode 100644
index 00000000..41aac43e
--- /dev/null
+++ b/client/process_loop.cpp
@@ -0,0 +1,390 @@
+/*
+   Copyright (C) 2009 Red Hat, Inc.
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License as
+   published by the Free Software Foundation; either version 2 of
+   the License, or (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "common.h"
+#include "process_loop.h"
+#include "debug.h"
+#include "platform.h"
+#include "utils.h"
+
+SyncEvent::SyncEvent()
+    : _err (false)
+    , _ready (false)
+{
+}
+
+SyncEvent::~SyncEvent()
+{
+}
+
+void SyncEvent::response(AbstractProcessLoop& events_loop)
+{
+    try {
+        do_response(events_loop);
+    } catch (Exception& e) {
+        LOG_WARN("unhandle exception: %s", e.what());
+        _err = true;
+    } catch (...) {
+        _err = true;
+    }
+    Lock lock(_mutex);
+    _ready = true;
+    _condition.notify_one();
+}
+
+void SyncEvent::wait()
+{
+#ifdef RED_DEBUG
+    ASSERT(!_process_loop || !_process_loop->is_same_thread(pthread_self()));
+#endif
+    Lock lock(_mutex);
+    while (!_ready) {
+        _condition.wait(lock);
+    }
+}
+
+class ProcessLoop::QuitEvent: public Event {
+public:
+    QuitEvent(int error_code) : _error_code(error_code) {}
+    virtual void response(AbstractProcessLoop& events_loop);
+private:
+    int _error_code;
+};
+
+void ProcessLoop::QuitEvent::response(AbstractProcessLoop& events_loop)
+{
+    events_loop.do_quit(_error_code);
+}
+
+/* EventsQueue */
+
+EventsQueue::EventsQueue(AbstractProcessLoop& owner)
+    : _events_gen (0)
+    , _owner (owner)
+{
+}
+
+EventsQueue::~EventsQueue()
+{
+    clear_queue();
+}
+
+void EventsQueue::clear_queue()
+{
+    Lock lock(_events_lock);
+    while (!_events.empty()) {
+        Event* event = _events.front();
+        _events.pop_front();
+        event->unref();
+    }
+}
+
+int EventsQueue::push_event(Event* event)
+{
+    Lock lock(_events_lock);
+    _events.push_back(event);
+    event->_generation = _events_gen;
+    event->ref();
+#ifdef RED_DEBUG
+    event->set_process_loop(&_owner);
+#endif
+    return _events.size();
+}
+
+void EventsQueue::process_events()
+{
+    _events_gen++;
+
+    for (;;) {
+        Event* event;
+        Lock lock(_events_lock);
+        if (_events.empty()) {
+            return;
+        }
+        event = _events.front();
+        if (event->_generation == _events_gen) {
+            return;
+        }
+        _events.pop_front();
+
+        lock.unlock();
+        event->response(_owner);
+#ifdef RED_DEBUG
+        event->set_process_loop(NULL);
+#endif
+        event->unref();
+    }
+}
+
+bool EventsQueue::is_empty()
+{
+    Lock lock(_events_lock);
+    return _events.empty();
+}
+
+/* Timers Queue */
+
+Timer::Timer()
+    : _is_armed (false)
+{
+}
+
+Timer::~Timer()
+{
+}
+
+void Timer::arm(uint32_t msec)
+{
+    _interval = msec;
+    _expiration = get_now();
+    calc_next_expiration_time();
+    _is_armed = true;
+}
+
+void Timer::disarm()
+{
+    _is_armed = false;
+}
+
+uint64_t Timer::get_now()
+{
+    return (Platform::get_monolithic_time() / 1000 / 1000);
+}
+
+TimersQueue::TimersQueue(AbstractProcessLoop& owner)
+    : _owner (owner)
+{
+}
+
+TimersQueue::~TimersQueue()
+{
+    clear_queue();
+}
+
+void TimersQueue::clear_queue()
+{
+    RecurciveLock lock(_timers_lock);
+    TimersSet::iterator iter;
+    for (iter = _armed_timers.begin(); iter != _armed_timers.end(); iter++) {
+        (*iter)->disarm();
+    }
+    _armed_timers.clear();
+}
+
+void TimersQueue::activate_interval_timer(Timer* timer, unsigned int millisec)
+{
+    RecurciveLock lock(_timers_lock);
+    timer->ref();
+    deactivate_interval_timer(timer);
+    timer->arm(millisec);
+    _armed_timers.insert(timer);
+}
+
+void TimersQueue::deactivate_interval_timer(Timer* timer)
+{
+    RecurciveLock lock(_timers_lock);
+    if (timer->is_armed()) {
+#ifdef  RED_DEBUG
+        int ret =
+#endif
+        _armed_timers.erase(timer);
+        ASSERT(ret);
+        timer->disarm();
+        timer->unref();
+    }
+}
+
+int TimersQueue::get_soonest_timeout()
+{
+    RecurciveLock lock(_timers_lock);
+    TimersSet::iterator iter;
+    iter = _armed_timers.begin();
+    if (iter == _armed_timers.end()) {
+        return INFINITE;
+    }
+
+    uint64_t now = Timer::get_now();
+    uint64_t next_time = (*iter)->get_expiration();
+
+    if (next_time <= now) {
+        return 0;
+    }
+    return (int)(next_time - now);
+}
+
+#define TIMER_COMPENSATION
+
+void TimersQueue::timers_action()
+{
+    RecurciveLock lock(_timers_lock);
+    uint64_t now = Timer::get_now();
+    TimersSet::iterator iter;
+
+    while (((iter = _armed_timers.begin()) != _armed_timers.end()) &&
+           ((*iter)->get_expiration() <= now)) {
+        Timer* timer = *iter;
+        _armed_timers.erase(iter);
+#ifndef TIMER_COMPENSATION
+        timer->_experatoin = now;
+#endif
+        timer->calc_next_expiration_time();
+#ifdef TIMER_COMPENSATION
+        if (timer->_expiration <= now) {
+            timer->_expiration = now;
+            timer->calc_next_expiration_time();
+        }
+#endif
+        _armed_timers.insert(timer);
+        timer->response(_owner);
+    }
+}
+
+ProcessLoop::ProcessLoop(void* owner)
+    : _events_queue (*this)
+    , _timers_queue (*this)
+    , _owner (owner)
+    , _quitting (false)
+    , _exit_code (0)
+    , _started (false)
+
+{
+    _event_sources.add_trigger(_wakeup_trigger);
+}
+
+ProcessLoop::~ProcessLoop()
+{
+    _event_sources.remove_trigger(_wakeup_trigger);
+}
+
+int ProcessLoop::run()
+{
+    _thread = pthread_self();
+    _started = true;
+    for (;;) {
+        if (_event_sources.wait_events(_timers_queue.get_soonest_timeout())) {
+            _quitting = true;
+            break;
+        }
+        _timers_queue.timers_action();
+        process_events_queue();
+        if (_quitting) {
+            break;
+        }
+    }
+
+    return _exit_code;
+}
+
+void ProcessLoop::do_quit(int error_code)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    if (_quitting) {
+        return;
+    }
+    _quitting = true;
+    _exit_code = error_code;
+}
+
+void ProcessLoop::quit(int error_code)
+{
+    AutoRef<QuitEvent> quit_event(new QuitEvent(error_code));
+    push_event(*quit_event);
+}
+
+void ProcessLoop::process_events_queue()
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _events_queue.process_events();
+    if (!_events_queue.is_empty()) {
+        wakeup();
+    }
+}
+
+void ProcessLoop::wakeup()
+{
+    _wakeup_trigger.trigger();
+}
+
+void ProcessLoop::add_trigger(EventSources::Trigger& trigger)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _event_sources.add_trigger(trigger);
+}
+
+void ProcessLoop::remove_trigger(EventSources::Trigger& trigger)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _event_sources.remove_trigger(trigger);
+}
+
+void ProcessLoop::add_socket(EventSources::Socket& socket)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _event_sources.add_socket(socket);
+}
+
+void ProcessLoop::remove_socket(EventSources::Socket& socket)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _event_sources.remove_socket(socket);
+}
+
+void ProcessLoop::add_file(EventSources::File& file)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _event_sources.add_file(file);
+}
+
+void ProcessLoop::remove_file(EventSources::File& file)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _event_sources.remove_file(file);
+}
+
+void ProcessLoop::add_handle(EventSources::Handle& handle)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _event_sources.add_handle(handle);
+}
+
+void ProcessLoop::remove_handle(EventSources::Handle& handle)
+{
+    ASSERT(!_started || pthread_equal(pthread_self(), _thread));
+    _event_sources.remove_handle(handle);
+}
+
+void ProcessLoop::push_event(Event* event)
+{
+    int queue_size = _events_queue.push_event(event);
+    if (queue_size == 1) { // queue was empty before the push
+        wakeup();
+    }
+}
+
+void ProcessLoop::activate_interval_timer(Timer* timer, unsigned int millisec)
+{
+    _timers_queue.activate_interval_timer(timer, millisec);
+
+    if (_started && !pthread_equal(pthread_self(), _thread)) {
+        wakeup();
+    }
+}
+
+void ProcessLoop::deactivate_interval_timer(Timer* timer)
+{
+    _timers_queue.deactivate_interval_timer(timer);
+}