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.

8 files changed, 494 insertions, 273 deletions

diff --git a/client/x11/Makefile.am b/client/x11/Makefile.am
index 9d87127b..aaeb3313 100644
--- a/client/x11/Makefile.am
+++ b/client/x11/Makefile.am
@@ -70,6 +70,7 @@ RED_COMMON_SRCS = 					\
 	$(top_srcdir)/client/pixels_source.h		\
 	$(top_srcdir)/client/platform.h			\
 	$(top_srcdir)/client/playback_channel.cpp	\
+	$(top_srcdir)/client/process_loop.cpp		\
 	$(top_srcdir)/client/quic.cpp			\
 	$(top_srcdir)/client/record_channel.cpp		\
 	$(top_srcdir)/client/red_channel.cpp		\
@@ -106,6 +107,7 @@ spicec_SOURCES =			\
 	atomic_count.h			\
 	events_loop_p.cpp		\
 	events_loop_p.h			\
+	event_sources_p.cpp		\
 	main.cpp			\
 	named_pipe.h                    \
 	named_pipe.cpp                  \
diff --git a/client/x11/event_sources_p.cpp b/client/x11/event_sources_p.cpp
new file mode 100644
index 00000000..2e6af138
--- /dev/null
+++ b/client/x11/event_sources_p.cpp
@@ -0,0 +1,339 @@
+/*
+   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 <sys/epoll.h>
+#include <sys/fcntl.h>
+
+#include "event_sources.h"
+#include "debug.h"
+#include "utils.h"
+
+#ifdef USING_EVENT_FD
+#include <sys/eventfd.h>
+#endif
+
+#define NUM_EPOLL_EVENTS 10
+
+#ifdef USING_EVENT_FD
+#define WRITE_FD _event_fd
+#define EVENT_DATA_TYPE eventfd_t
+#else
+#define WRITE_FD _event_write_fd
+#define EVENT_DATA_TYPE uint8_t
+#endif
+
+class EventWrapper {
+public:
+    EventWrapper(EventSources& owner, EventSource& event)
+        : _owner (owner)
+        , _event (&event)
+        , _refs (1)
+    {
+    }
+
+    EventWrapper* ref()
+    {
+        _refs++;
+        return this;
+    }
+
+    void unref()
+    {
+        if (!--_refs) {
+            _owner.remove_wrapper(this);
+            delete this;
+        }
+    }
+
+    EventSource* get_event()
+    {
+        return _event;
+    }
+
+    void invalidate()
+    {
+        _event = NULL;
+    }
+
+private:
+    EventSources& _owner;
+    EventSource* _event;
+    int _refs;
+};
+
+EventSources::EventSources()
+{
+    _epoll = epoll_create(NUM_EPOLL_EVENTS);
+    if (_epoll == -1) {
+        THROW("create epool failed");
+    }
+}
+
+EventSources::~EventSources()
+{
+    Events::iterator iter = _events.begin();
+    for (; iter != _events.end(); iter++) {
+        delete *iter;
+    }
+    close(_epoll);
+}
+
+bool EventSources::wait_events(int timeout_ms)
+{
+    struct epoll_event events[NUM_EPOLL_EVENTS];
+    int num_events = epoll_wait(_epoll, events, NUM_EPOLL_EVENTS, timeout_ms);
+
+    if (num_events == -1) {
+        if (errno == EINTR) {
+            return false;
+        }
+        THROW("wait error eventfd failed");
+    }
+
+    for (int i = 0; i < num_events; i++) {
+        ((EventWrapper*)events[i].data.ptr)->ref();
+    }
+
+    for (int i = 0; i < num_events; i++) {
+        EventWrapper* wrapper;
+        EventSource* event;
+
+        wrapper = (EventWrapper *)events[i].data.ptr;
+        if ((event = wrapper->get_event())) {
+            event->action();
+        }
+        wrapper->unref();
+    }
+    return false;
+}
+
+void EventSources::add_trigger(Trigger& trigger)
+{
+    int fd = trigger.get_fd();
+    EventWrapper* wrapper = new EventWrapper(*this, trigger);
+    struct epoll_event event;
+    event.data.ptr = wrapper;
+    event.events = EPOLLIN;
+    if (epoll_ctl(_epoll, EPOLL_CTL_ADD, fd, &event) == -1) {
+        THROW("epoll add failed");
+    }
+    _events.push_back(wrapper);
+}
+
+void EventSources_p::remove_wrapper(EventWrapper* wrapper)
+{
+    Events::iterator iter = _events.begin();
+    for (;; iter++) {
+        if (iter == _events.end()) {
+            THROW("wrapper not found");
+        }
+        if ((*iter) == wrapper) {
+            _events.erase(iter);
+            return;
+        }
+    }
+}
+
+void EventSources::remove_trigger(Trigger& trigger)
+{
+    Events::iterator iter = _events.begin();
+    for (;; iter++) {
+        if (iter == _events.end()) {
+            THROW("trigger not found");
+        }
+        if ((*iter)->get_event() == &trigger) {
+            (*iter)->invalidate();
+            (*iter)->unref();
+            break;
+        }
+    }
+    int fd = trigger.get_fd();
+    if (epoll_ctl(_epoll, EPOLL_CTL_DEL, fd, NULL) == -1) {
+        THROW("epoll remove failed");
+    }
+}
+
+EventSources::Trigger::Trigger()
+{
+#ifdef USING_EVENT_FD
+    _event_fd = eventfd(0, 0);
+    if (_event_fd == -1) {
+        THROW("create eventfd failed");
+    }
+#else
+    int fd[2];
+    if (pipe(fd) == -1) {
+        THROW("create pipe failed");
+    }
+    _event_fd = fd[0];
+    _event_write_fd = fd[1];
+#endif
+    int flags;
+    if ((flags = fcntl(_event_fd, F_GETFL)) == -1) {
+        THROW("failed to set eventfd non block: %s", strerror(errno));
+    }
+
+    if (fcntl(_event_fd, F_SETFL, flags | O_NONBLOCK) == -1) {
+        THROW("failed to set eventfd non block: %s", strerror(errno));
+    }
+}
+
+EventSources::Trigger::~Trigger()
+{
+    close(_event_fd);
+#ifndef USING_EVENT_FD
+    close(_event_write_fd);
+#endif
+}
+
+void EventSources::Trigger::trigger()
+{
+    Lock lock(_lock);
+    if (_pending_int) {
+        return;
+    }
+    _pending_int = true;
+    static const EVENT_DATA_TYPE val = 1;
+    if (::write(WRITE_FD, &val, sizeof(val)) != sizeof(val)) {
+        THROW("write event failed");
+    }
+}
+
+bool Trigger_p::reset_event()
+{
+    Lock lock(_lock);
+    if (!_pending_int) {
+        return false;
+    }
+    EVENT_DATA_TYPE val;
+    if (read(_event_fd, &val, sizeof(val)) != sizeof(val)) {
+        THROW("event read error");
+    }
+    _pending_int = false;
+    return true;
+}
+
+void EventSources::Trigger::reset()
+{
+    reset_event();
+}
+
+void EventSources::Trigger::action()
+{
+    if (reset_event()) {
+        on_event();
+    }
+}
+
+static void set_non_blocking(int fd)
+{
+    int flags;
+    if ((flags = fcntl(fd, F_GETFL)) == -1) {
+        THROW("failed to set socket non block: %s", strerror(errno));
+    }
+
+    if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
+        THROW("failed to set socket non block: %s", strerror(errno));
+    }
+}
+
+static void set_blocking(int fd)
+{
+    int flags;
+    if ((flags = fcntl(fd, F_GETFL)) == -1) {
+        THROW("failed to clear socket non block: %s", strerror(errno));
+    }
+
+    if (fcntl(fd, F_SETFL, flags & ~O_NONBLOCK) == -1) {
+        THROW("failed to clear socket non block: %s", strerror(errno));
+    }
+}
+
+static void add_to_poll(int fd, int epoll, EventWrapper* wrapper)
+{
+    struct epoll_event event;
+    event.data.ptr = wrapper;
+    event.events = EPOLLIN | EPOLLOUT | EPOLLET;
+    if (epoll_ctl(epoll, EPOLL_CTL_ADD, fd, &event) == -1) {
+        THROW("epoll add failed");
+    }
+}
+
+void EventSources::add_socket(Socket& socket)
+{
+    int fd = socket.get_socket();
+    set_non_blocking(fd);
+    EventWrapper* wrapper = new EventWrapper(*this, socket);
+    add_to_poll(fd, _epoll, wrapper);
+    _events.push_back(wrapper);
+}
+
+static bool remove_event(EventSources_p::Events& events, EventSource& event)
+{
+    EventSources_p::Events::iterator iter = events.begin();
+    for (;; iter++) {
+        if (iter == events.end()) {
+            return false;
+        }
+        if ((*iter)->get_event() == &event) {
+            (*iter)->invalidate();
+            (*iter)->unref();
+            return true;
+        }
+    }
+}
+
+void EventSources::remove_socket(Socket& socket)
+{
+    if (!remove_event(_events, socket)) {
+        THROW("socket not found");
+    }
+    int fd = socket.get_socket();
+    if (epoll_ctl(_epoll, EPOLL_CTL_DEL, fd, NULL) == -1) {
+        THROW("epoll remove failed");
+    }
+    set_blocking(fd);
+}
+
+void EventSources::add_file(File& file)
+{
+    int fd = file.get_fd();
+    set_non_blocking(fd);
+    EventWrapper* wrapper = new EventWrapper(*this, file);
+    add_to_poll(fd, _epoll, wrapper);
+    _events.push_back(wrapper);
+}
+
+void EventSources::remove_file(File& file)
+{
+    if (!remove_event(_events, file)) {
+        THROW("file not found");
+    }
+    int fd = file.get_fd();
+    if (epoll_ctl(_epoll, EPOLL_CTL_DEL, fd, NULL) == -1) {
+        THROW("epoll remove failed");
+    }
+    set_blocking(fd);
+}
+
+void EventSources::add_handle(Handle& file)
+{
+}
+
+void EventSources::remove_handle(Handle& file)
+{
+}
diff --git a/client/x11/event_sources_p.h b/client/x11/event_sources_p.h
new file mode 100644
index 00000000..591a781f
--- /dev/null
+++ b/client/x11/event_sources_p.h
@@ -0,0 +1,63 @@
+/*
+   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/>.
+*/
+
+#ifndef _H_EVENT_SOURCES_P
+#define _H_EVENT_SOURCES_P
+
+#include "common.h"
+#include "threads.h"
+
+#if __GLIBC__ > 2 || (__GLIBC__ == 2 && __GLIBC_MINOR__ >= 8)
+#define USING_EVENT_FD
+#endif
+
+#define INFINITE -1
+
+class EventWrapper;
+
+class EventSources_p {
+public:
+    void remove_wrapper(EventWrapper*);
+
+public:
+    int _epoll;
+    typedef std::list<EventWrapper*> Events;
+    Events _events;
+
+    friend class EventWrapper;
+};
+
+class Trigger_p {
+public:
+    Trigger_p() : _pending_int (false) {}
+    int get_fd() { return _event_fd;}
+    bool reset_event();
+
+public:
+    int _event_fd;
+#ifndef USING_EVENT_FD
+    int _event_write_fd;
+#endif
+    bool _pending_int;
+    Mutex _lock;
+};
+
+class Handle_p {
+};
+
+#endif
+
diff --git a/client/x11/events_loop_p.cpp b/client/x11/events_loop_p.cpp
index 4db1469c..a1c450b5 100644
--- a/client/x11/events_loop_p.cpp
+++ b/client/x11/events_loop_p.cpp
@@ -38,7 +38,7 @@
 
 class EventWrapper {
 public:
-    EventWrapper(EventsLoop& owner, EventSource& event)
+    EventWrapper(EventsLoop& owner, EventSourceOld& event)
         : _owner (owner)
         , _event (&event)
         , _refs (1)
@@ -59,7 +59,7 @@ public:
         }
     }
 
-    EventSource* get_event()
+    EventSourceOld* get_event()
     {
         return _event;
     }
@@ -71,7 +71,7 @@ public:
 
 private:
     EventsLoop& _owner;
-    EventSource* _event;
+    EventSourceOld* _event;
     int _refs;
 };
 
@@ -116,7 +116,7 @@ void EventsLoop::run_once(int timeout_milli)
     }
     for (int i = 0; i < num_events; i++) {
         EventWrapper* wrapper;
-        EventSource* event;
+        EventSourceOld* event;
 
         wrapper = (EventWrapper *)events[i].data.ptr;
         if ((event = wrapper->get_event())) {
@@ -218,7 +218,7 @@ void EventsLoop::Trigger::trigger()
     }
 }
 
-bool Trigger_p::reset_event()
+bool EventsLoop_p::Trigger_p::reset_event()
 {
     Lock lock(_lock);
     if (!_pending_int) {
@@ -287,7 +287,7 @@ void EventsLoop::add_socket(Socket& socket)
     _events.push_back(wrapper);
 }
 
-static bool remove_event(EventsLoop_p::Events& events, EventSource& event)
+static bool remove_event(EventsLoop_p::Events& events, EventSourceOld& event)
 {
     EventsLoop_p::Events::iterator iter = events.begin();
     for (;; iter++) {
diff --git a/client/x11/events_loop_p.h b/client/x11/events_loop_p.h
index 0a70d3ed..d339ca4b 100644
--- a/client/x11/events_loop_p.h
+++ b/client/x11/events_loop_p.h
@@ -31,6 +31,7 @@ class EventWrapper;
 
 class EventsLoop_p {
 public:
+    class Trigger_p;
     void remove_wrapper(EventWrapper*);
 
 public:
@@ -41,7 +42,7 @@ public:
     friend class EventWrapper;
 };
 
-class Trigger_p {
+class EventsLoop_p::Trigger_p {
 public:
     Trigger_p() : _pending_int (false) {}
     int get_fd() { return _event_fd;}
diff --git a/client/x11/named_pipe.cpp b/client/x11/named_pipe.cpp
index 73d103c0..18c390e3 100644
--- a/client/x11/named_pipe.cpp
+++ b/client/x11/named_pipe.cpp
@@ -23,7 +23,7 @@
 #include "utils.h"
 #include "debug.h"
 
-Session::Session(int fd, EventsLoop& events_loop)
+Session::Session(int fd, ProcessLoop& events_loop)
     : _fd_client(fd)
     , _events_loop(events_loop)
 {
@@ -121,7 +121,7 @@ int LinuxListener::create_socket(const char *socket_name)
 }
 
 LinuxListener::LinuxListener(const char *name, NamedPipe::ListenerInterface &listener_interface,
-                             EventsLoop &events_loop)
+                             ProcessLoop& events_loop)
     : _listener_interface (listener_interface)
     , _events_loop (events_loop)
 {
diff --git a/client/x11/named_pipe.h b/client/x11/named_pipe.h
index ab41f7f3..b32ede88 100644
--- a/client/x11/named_pipe.h
+++ b/client/x11/named_pipe.h
@@ -20,11 +20,11 @@
 
 #include "platform.h"
 #include "x_platform.h"
-#include "events_loop.h"
+#include "process_loop.h"
 
-class Session: public EventsLoop::Socket {
+class Session: public EventSources::Socket {
 public:
-    Session(int fd, EventsLoop& events_loop);
+    Session(int fd, ProcessLoop& events_loop);
     virtual ~Session();
     void bind(NamedPipe::ConnectionInterface* conn_interface);
 
@@ -37,13 +37,13 @@ public:
 private:
     NamedPipe::ConnectionInterface *_conn_interface;
     int _fd_client;
-    EventsLoop &_events_loop;
+    ProcessLoop &_events_loop;
 };
 
-class LinuxListener: public EventsLoop::Socket {
+class LinuxListener: public EventSources::Socket {
 public:
     LinuxListener(const char *name, NamedPipe::ListenerInterface &listener_interface,
-                  EventsLoop &events_loop);
+                  ProcessLoop& events_loop);
     virtual ~LinuxListener();
     void on_event();
     virtual int get_socket() {return _listen_socket;}
@@ -55,7 +55,7 @@ private:
     NamedPipe::ListenerInterface &_listener_interface;
     int _listen_socket;
     std::string _name;
-    EventsLoop &_events_loop;
+    ProcessLoop &_events_loop;
 };
 
 #endif
diff --git a/client/x11/platform.cpp b/client/x11/platform.cpp
index f0c68235..8f318f2d 100644
--- a/client/x11/platform.cpp
+++ b/client/x11/platform.cpp
@@ -50,7 +50,7 @@
 #include "resource.h"
 #include "res.h"
 #include "cursor.h"
-#include "events_loop.h"
+#include "process_loop.h"
 
 #define DWORD uint32_t
 #define BOOL bool
@@ -66,10 +66,8 @@ static Display* x_display = NULL;
 static XVisualInfo **vinfo = NULL;
 static GLXFBConfig **fb_config;
 
-static EventsLoop events_loop;
 static XContext win_proc_context;
-static bool quit_request = false;
-static pthread_t main_thread;
+static ProcessLoop* main_loop = NULL;
 static int focus_count = 0;
 
 static bool using_xrandr_1_0 = false;
@@ -108,16 +106,19 @@ static Platform::DisplayModeListner* display_mode_listener = &default_display_mo
 
 NamedPipe::ListenerRef NamedPipe::create(const char *name, ListenerInterface& listener_interface)
 {
-    return (ListenerRef)(new LinuxListener(name, listener_interface, events_loop));
+    ASSERT(main_loop && main_loop->is_same_thread(pthread_self()));
+    return (ListenerRef)(new LinuxListener(name, listener_interface, *main_loop));
 }
 
 void NamedPipe::destroy(ListenerRef listener_ref)
 {
+    ASSERT(main_loop && main_loop->is_same_thread(pthread_self()));
     delete (LinuxListener *)listener_ref;
 }
 
 void NamedPipe::destroy_connection(ConnectionRef conn_ref)
 {
+    ASSERT(main_loop && main_loop->is_same_thread(pthread_self()));
     delete (Session *)conn_ref;
 }
 
@@ -137,81 +138,45 @@ int32_t NamedPipe::write(ConnectionRef conn_ref, const uint8_t* buf, int32_t siz
     return -1;
 }
 
-class XEventHandler: public EventsLoop::File {
+class XEventHandler: public EventSources::File {
 public:
-    XEventHandler(Display& x_display);
-    virtual void on_event() {}
+    XEventHandler(Display& x_display, XContext& win_proc_context);
+    virtual void on_event();
     virtual int get_fd() {return _x_fd;}
 
 private:
+    Display& _x_display;
+    XContext& _win_proc_context;
     int _x_fd;
 };
 
-XEventHandler::XEventHandler(Display& x_display)
+XEventHandler::XEventHandler(Display& x_display, XContext& win_proc_context)
+    : _x_display (x_display)
+    , _win_proc_context (win_proc_context)
 {
     if ((_x_fd = ConnectionNumber(&x_display)) == -1) {
         THROW("get x fd failed");
     }
 }
 
-class WakeupEventHandler: public EventsLoop::Trigger {
-public:
-    virtual void on_event() {}
-};
-
-static WakeupEventHandler wakeup_handler;
-
-#define NSEC_PER_SEC (1000 * 1000 * 1000)
-
-class Timer {
-public:
-    Timer(timer_proc_t proc, void* opaque);
-    ~Timer();
-
-    void arm(uint32_t msec);
-    void disarm();
-    bool action(const timespec& now);
-
-    const timespec& get_experatoin() const { return _experatoin;}
-
-    static int get_timout();
-    static void timers_action();
-
-private:
-    void calc_next_experatoin_time();
-
-    static bool timespec_less(const timespec& time, const timespec& from);
-    static bool timespec_equal(const timespec& time, const timespec& from);
-    static bool timespec_less_or_equal(const timespec& time, const timespec& from);
-
-public:
-    static RecurciveMutex timers_lock;
+void XEventHandler::on_event()
+{
+    while (XPending(&_x_display)) {
+        XPointer proc_pointer;
+        XEvent event;
 
-private:
-    timer_proc_t _proc;
-    void* _opaque;
-    bool _armed;
-    timespec _interval;
-    timespec _experatoin;
-
-    class Compare {
-    public:
-        bool operator () (const Timer* timer1, const Timer* timer2) const
-        {
-            if (!Timer::timespec_less(timer1->get_experatoin(), timer2->get_experatoin())) {
-                return Timer::timespec_equal(timer1->get_experatoin(), timer2->get_experatoin()) ?
-                                                                            timer1 < timer2 : false;
-            }
-            return true;
+        XNextEvent(&_x_display, &event);
+        if (event.xany.window == None) {
+            LOG_WARN("invalid window");
+            continue;
         }
-    };
 
-    typedef std::set<Timer*, Compare> TimersSet;
-    static TimersSet armed_timers;
-};
-
-Timer::TimersSet Timer::armed_timers;
-RecurciveMutex Timer::timers_lock;
+        if (XFindContext(&_x_display, event.xany.window, _win_proc_context, &proc_pointer)) {
+            THROW("no window proc");
+        }
+        ((XPlatform::win_proc_t)proc_pointer)(event);
+    }
+}
 
 Display* XPlatform::get_display()
 {
@@ -242,44 +207,8 @@ void XPlatform::cleare_win_proc(Window win)
 
 void Platform::send_quit_request()
 {
-    quit_request = true;
-    wakeup();
-}
-
-void Platform::wait_events()
-{
-    ASSERT(pthread_self() == main_thread);
-    XFlush(x_display);
-    if (!XPending(x_display)) {
-        events_loop.run_once(Timer::get_timout());
-        Timer::timers_action();
-    }
-}
-
-void Platform::wakeup()
-{
-    wakeup_handler.trigger();
-}
-
-bool Platform::process_events()
-{
-    ASSERT(pthread_self() == main_thread);
-    while (XPending(x_display)) {
-        XPointer proc_pointer;
-        XEvent event;
-
-        XNextEvent(x_display, &event);
-        if (event.xany.window == None) {
-            LOG_WARN("invalid window");
-            continue;
-        }
-
-        if (XFindContext(x_display, event.xany.window, win_proc_context, &proc_pointer)) {
-            THROW("no window proc");
-        }
-        ((XPlatform::win_proc_t)proc_pointer)(event);
-    }
-    return quit_request;
+    ASSERT(main_loop);
+    main_loop->quit(0);
 }
 
 uint64_t Platform::get_monolithic_time()
@@ -341,155 +270,6 @@ void Platform::set_thread_priority(void* thread, Platform::ThreadPriority in_pri
     }
 }
 
-Timer::Timer(timer_proc_t proc, void* opaque)
-    : _proc (proc)
-    , _opaque (opaque)
-    , _armed (false)
-{
-}
-
-Timer::~Timer()
-{
-    disarm();
-}
-
-void Timer::arm(uint32_t msec)
-{
-    disarm();
-    _interval.tv_sec = msec / 1000;
-    _interval.tv_nsec = (msec % 1000) * 1000 * 1000;
-    if (clock_gettime(CLOCK_MONOTONIC, &_experatoin)) {
-        THROW("gettime failed %s", strerror(errno));
-    }
-    calc_next_experatoin_time();
-    _armed = true;
-    armed_timers.insert(this);
-}
-
-void Timer::disarm()
-{
-    if (!_armed) {
-        return;
-    }
-    armed_timers.erase(this);
-    _armed = false;
-}
-
-#define TINER_COMPENSATION
-
-bool Timer::action(const timespec& now)
-{
-    ASSERT(_armed);
-    ASSERT(now.tv_nsec < NSEC_PER_SEC);
-
-    if (timespec_less(now, _experatoin)) {
-        return false;
-    }
-    armed_timers.erase(this);
-#ifndef TINER_COMPENSATION
-    _experatoin = now;
-#endif
-    calc_next_experatoin_time();
-#ifdef TINER_COMPENSATION
-    if (timespec_less_or_equal(_experatoin, now)) {
-        _experatoin = now;
-        calc_next_experatoin_time();
-    }
-#endif
-    armed_timers.insert(this);
-    _proc(_opaque, (TimerID)this);
-    return true;
-}
-
-int Timer::get_timout()
-{
-    RecurciveLock lock(Timer::timers_lock);
-    TimersSet::iterator iter;
-    iter = armed_timers.begin();
-    if (iter == armed_timers.end()) {
-        return -1;
-    }
-
-    timespec now;
-    if (clock_gettime(CLOCK_MONOTONIC, &now)) {
-        THROW("gettime failed %s", strerror(errno));
-    }
-
-    const timespec& next_time = (*iter)->get_experatoin();
-
-    if (!timespec_less(now, next_time)) {
-        return 0;
-    }
-    return ((next_time.tv_nsec - now.tv_nsec) / 1000 / 1000) +
-           (next_time.tv_sec - now.tv_sec) * 1000;
-}
-
-void Timer::timers_action()
-{
-    RecurciveLock lock(timers_lock);
-    timespec now;
-    if (clock_gettime(CLOCK_MONOTONIC, &now)) {
-        THROW("gettime failed %s", strerror(errno));
-    }
-
-    TimersSet::iterator iter;
-    while ((iter = armed_timers.begin()) != armed_timers.end() && (*iter)->action(now));
-}
-
-void Timer::calc_next_experatoin_time()
-{
-    _experatoin.tv_nsec += _interval.tv_nsec;
-    _experatoin.tv_sec += (_experatoin.tv_nsec / NSEC_PER_SEC) + _interval.tv_sec;
-    _experatoin.tv_nsec %= NSEC_PER_SEC;
-}
-
-bool Timer::timespec_less(const timespec& time, const timespec& from)
-{
-    return time.tv_sec < from.tv_sec || (time.tv_sec == from.tv_sec && time.tv_nsec < from.tv_nsec);
-}
-
-bool Timer::timespec_equal(const timespec& time, const timespec& from)
-{
-    return time.tv_sec == from.tv_sec && time.tv_nsec <= from.tv_nsec;
-}
-
-bool Timer::timespec_less_or_equal(const timespec& time, const timespec& from)
-{
-    return time.tv_sec < from.tv_sec ||
-           (time.tv_sec == from.tv_sec && time.tv_nsec <= from.tv_nsec);
-}
-
-TimerID Platform::create_interval_timer(timer_proc_t proc, void* opaque)
-{
-    return (TimerID) new Timer(proc, opaque);
-}
-
-bool Platform::activate_interval_timer(TimerID timer, unsigned int millisec)
-{
-    RecurciveLock lock(Timer::timers_lock);
-    ((Timer*)timer)->arm(millisec);
-    if (pthread_self() != main_thread) {
-        wakeup();
-    }
-    return true;
-}
-
-bool Platform::deactivate_interval_timer(TimerID timer)
-{
-    RecurciveLock lock(Timer::timers_lock);
-    ((Timer*)timer)->disarm();
-    return true;
-}
-
-void Platform::destroy_interval_timer(TimerID timer)
-{
-    if (timer == INVALID_TIMER) {
-        return;
-    }
-    RecurciveLock lock(Timer::timers_lock);
-    delete (Timer*)timer;
-}
-
 void Platform::set_event_listener(EventListener* listener)
 {
     event_listener = listener ? listener : &default_event_listener;
@@ -2278,13 +2058,11 @@ void Platform::init()
 {
     int err, ev;
     int threads_enable;
-    XEventHandler *x_event_handler;
 
     DBG(0, "");
 
     threads_enable = XInitThreads();
 
-    main_thread = pthread_self();
 
     if (!(x_display = XOpenDisplay(NULL))) {
         THROW("open X display failed");
@@ -2341,10 +2119,6 @@ void Platform::init()
     init_xrandr();
     init_xrender();
 
-    x_event_handler = new XEventHandler(*x_display);
-    events_loop.add_file(*x_event_handler);
-    events_loop.add_trigger(wakeup_handler);
-
     struct sigaction act;
     memset(&act, 0, sizeof(act));
     sigfillset(&act.sa_mask);
@@ -2368,6 +2142,14 @@ void Platform::init()
     atexit(cleanup);
 }
 
+void Platform::set_process_loop(ProcessLoop& main_process_loop)
+{
+    main_loop = &main_process_loop;
+    XEventHandler *x_event_handler;
+    x_event_handler = new XEventHandler(*x_display, win_proc_context);
+    main_loop->add_file(*x_event_handler);
+}
+
 uint32_t Platform::get_keyboard_modifiers()
 {
     XKeyboardState keyboard_state;
@@ -2641,3 +2423,37 @@ LocalCursor* Platform::create_default_cursor()
     return new XDefaultCursor();
 }
 
+class PlatformTimer: public Timer {
+public:
+    PlatformTimer(timer_proc_t proc, void* opaque) : _proc(proc), _opaque(opaque) {}
+    void response(AbstractProcessLoop& events_loop) {_proc(_opaque, (TimerID)this);}
+
+private:
+    timer_proc_t _proc;
+    void* _opaque;
+};
+
+TimerID Platform::create_interval_timer(timer_proc_t proc, void* opaque)
+{
+    return (TimerID)(new PlatformTimer(proc, opaque));
+}
+
+bool Platform::activate_interval_timer(TimerID timer, unsigned int millisec)
+{
+    ASSERT(main_loop);
+    main_loop->activate_interval_timer((PlatformTimer*)timer, millisec);
+    return true;
+}
+
+bool Platform::deactivate_interval_timer(TimerID timer)
+{
+    ASSERT(main_loop);
+    main_loop->deactivate_interval_timer((PlatformTimer*)timer);
+    return true;
+}
+
+void Platform::destroy_interval_timer(TimerID timer)
+{
+    deactivate_interval_timer(timer);
+    ((PlatformTimer*)timer)->unref();
+}