prepare move to rtpoll

10 files changed, 429 insertions, 63 deletions

diff --git a/src/pulsecore/core-util.c b/src/pulsecore/core-util.c
index 0d243ee6..ba3a9b21 100644
--- a/src/pulsecore/core-util.c
+++ b/src/pulsecore/core-util.c
@@ -2631,3 +2631,24 @@ char *pa_realpath(const char *path) {
 
     return t;
 }
+
+pa_bool_t pa_linux_newer_than(unsigned major, unsigned minor, unsigned micro) {
+
+#ifdef __linux__
+    unsigned _major, _minor, _micro;
+    struct utsname u;
+
+    pa_assert_se(uname(&u) == 0);
+
+    if (sscanf(u.release, "%u.%u.%u", &_major, &_minor, &_micro) != 3)
+        return FALSE;
+
+    return
+        (_major > major) ||
+        (_major == major && _minor > minor) ||
+        (_major == major && _minor == minor && _micro > micro);
+
+#endif
+
+    return FALSE;
+}
diff --git a/src/pulsecore/core-util.h b/src/pulsecore/core-util.h
index 0ba33f31..9ab84b4c 100644
--- a/src/pulsecore/core-util.h
+++ b/src/pulsecore/core-util.h
@@ -223,4 +223,6 @@ char *pa_unescape(char *p);
 
 char *pa_realpath(const char *path);
 
+pa_bool_t pa_linux_newer_than(unsigned major, unsigned minor, unsigned micro);
+
 #endif
diff --git a/src/pulsecore/macro.h b/src/pulsecore/macro.h
index 20015bf5..301d1d16 100644
--- a/src/pulsecore/macro.h
+++ b/src/pulsecore/macro.h
@@ -226,6 +226,8 @@ typedef int pa_bool_t;
 #define PA_DEBUG_TRAP raise(SIGTRAP)
 #endif
 
+typedef void (*pa_function_t) (...);
+
 /* We include this at the very last place */
 #include <pulsecore/log.h>
 
diff --git a/src/pulsecore/pstream.c b/src/pulsecore/pstream.c
index ef1105ba..c92b5baf 100644
--- a/src/pulsecore/pstream.c
+++ b/src/pulsecore/pstream.c
@@ -38,7 +38,6 @@
 #include <netinet/in.h>
 #endif
 
-
 #include <pulse/xmalloc.h>
 
 #include <pulsecore/winsock.h>
diff --git a/src/pulsecore/rtclock.c b/src/pulsecore/rtclock.c
index 56ab2ef0..11eb8eae 100644
--- a/src/pulsecore/rtclock.c
+++ b/src/pulsecore/rtclock.c
@@ -145,6 +145,15 @@ struct timeval* pa_rtclock_from_wallclock(struct timeval *tv) {
     return tv;
 }
 
+struct timespec *pa_timespec_store(struct timespec *ts, pa_usec_t v)  {
+    pa_assert(ts);
+
+    ts->tv_sec = (time_t) (v / PA_USEC_PER_SEC);
+    ts->tv_nsec = (long) ((v % PA_USEC_PER_SEC) * PA_NSEC_PER_USEC);
+
+    return ts;
+}
+
 pa_usec_t pa_timespec_load(const struct timespec *ts) {
     pa_assert(ts);
 
diff --git a/src/pulsecore/rtclock.h b/src/pulsecore/rtclock.h
index 03cc1c72..fc925a48 100644
--- a/src/pulsecore/rtclock.h
+++ b/src/pulsecore/rtclock.h
@@ -42,6 +42,7 @@ void pa_rtclock_hrtimer_enable(void);
 
 struct timeval* pa_rtclock_from_wallclock(struct timeval *tv);
 
+struct timespec *pa_timespec_store(struct timespec *ts, pa_usec_t v);
 pa_usec_t pa_timespec_load(const struct timespec *ts);
 
 #endif
diff --git a/src/pulsecore/rtpoll-api.c b/src/pulsecore/rtpoll-api.c
new file mode 100644
index 00000000..78ef5531
--- /dev/null
+++ b/src/pulsecore/rtpoll-api.c
@@ -0,0 +1,157 @@
+/***
+  This file is part of PulseAudio.
+
+  Copyright 2009 Lennart Poettering
+
+  PulseAudio is free software; you can redistribute it and/or modify
+  it under the terms of the GNU Lesser General Public License as
+  published by the Free Software Foundation; either version 2.1 of the
+  License, or (at your option) any later version.
+
+  PulseAudio 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
+  Lesser General Public License for more details.
+
+  You should have received a copy of the GNU Lesser General Public
+  License along with PulseAudio; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+  USA.
+***/
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include "rtpoll-api.h"
+
+struct event {
+    pa_function_t callback;
+    void *userdata;
+    pa_function_t destroy;
+};
+
+PA_STATIC_FLIST_DECLARE(events, 0, pa_xfree);
+
+static short map_flags_to_libc(pa_io_event_flags_t flags) {
+    return (short)
+        ((flags & PA_IO_EVENT_INPUT ? POLLIN : 0) |
+         (flags & PA_IO_EVENT_OUTPUT ? POLLOUT : 0) |
+         (flags & PA_IO_EVENT_ERROR ? POLLERR : 0) |
+         (flags & PA_IO_EVENT_HANGUP ? POLLHUP : 0));
+}
+
+static pa_io_event_flags_t map_flags_from_libc(short flags) {
+    return
+        (flags & POLLIN ? PA_IO_EVENT_INPUT : 0) |
+        (flags & POLLOUT ? PA_IO_EVENT_OUTPUT : 0) |
+        (flags & POLLERR ? PA_IO_EVENT_ERROR : 0) |
+        (flags & POLLHUP ? PA_IO_EVENT_HANGUP : 0);
+}
+
+static pa_io_event* rtpoll_io_new(
+        pa_mainloop_api*a,
+        int fd,
+        pa_io_event_flags_t f,
+        pa_io_event_cb_t callback,
+        void *userdata) {
+
+    pa_rtpoll_item *i;
+    struct pollfd *pollfd;
+    struct event *d;
+
+    i = pa_rtpoll_item_new(a->userdata, PA_RTPOLL_LATE, 1);
+
+    pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
+    pollfd->fd = fd;
+    pollfd->events = map_flags_to_libc(f);
+
+    if (!(d = pa_flist_pop(PA_STATIC_FLIST_GET(events))))
+        d = pa_xnew(struct event, 1);
+
+    d->callback = (pa_function_t) callback;
+    d->userdata = userdata;
+    d->destroy = NULL;
+
+    pa_rtpoll_item_set_userdata(i, d);
+
+    pa_rtpoll_item_set_work_callback(i, work_callback);
+
+    return (pa_io_event*) i;
+}
+
+static void rtpoll_io_enable(pa_io_event *e, pa_io_event_flags_t f) {
+    struct pollfd *pollfd;
+    pa_rtpoll_item *i = (pa_rtpoll_item*) e;
+
+    pa_assert(i);
+
+    pollfd = pa_rtpoll_item_get_pollfd(i, NULL);
+    pollfd->events = map_flags_to_libc(f);
+}
+
+static void rtpoll_io_free(pa_io_event *e) {
+    pa_rtpoll_item *i = (pa_rtpoll_item*) e;
+    struct event *d;
+
+    pa_assert(i);
+
+    d = pa_rtpoll_item_get_userdata(i);
+
+    if (d->destroy_callback)
+        d->destroy_callback(pa_rtpoll_get_userdata(pa_rtpoll_item_rtpoll(i)), i, d->userdata);
+
+    if (pa_flist_push(PA_STATIC_FLIST_GET(events), d) < 0)
+        pa_xfree(d);
+
+    pa_rtpoll_item_free(i);
+}
+
+static void rtpoll_io_set_destroy(pa_io_event *e, pa_io_event_destroy_cb_t cb) {
+    pa_rtpoll_item *i = (pa_rtpoll_item*) e;
+    struct event *d;
+
+    pa_assert(e);
+
+    d = pa_rtpoll_item_get_userdata(i);
+
+    d->destroy = (pa_function_t) cb);
+}
+
+static const pa_mainloop_api vtable = {
+    .userdata = NULL,
+
+    .io_new = rtpoll_io_new,
+    .io_enable = rtpoll_io_enable,
+    .io_free = rtpoll_io_free,
+    .io_set_destroy= rtpoll_io_set_destroy,
+
+    .time_new = rtpoll_time_new,
+    .time_restart = rtpoll_time_restart,
+    .time_free = rtpoll_time_free,
+    .time_set_destroy = rtpoll_time_set_destroy,
+
+    .defer_new = rtpoll_defer_new,
+    .defer_enable = rtpoll_defer_enable,
+    .defer_free = rtpoll_defer_free,
+    .defer_set_destroy = rtpoll_defer_set_destroy,
+
+    .quit = rtpoll_quit,
+};
+
+pa_mainloop_api* pa_rtpoll_api_new(pa_rtpoll *p) {
+    pa_mainloop_api *api;
+
+    pa_assert(p);
+
+    api = pa_memdup(pa_mainloop_api, vtable, 1);
+    api->userdata = p;
+
+    return r;
+}
+
+void pa_rtpoll_api_free(pa_mainloop_api *api) {
+    pa_assert(p);
+
+    pa_xfree(p);
+}
diff --git a/src/pulsecore/rtpoll-api.h b/src/pulsecore/rtpoll-api.h
new file mode 100644
index 00000000..d5f7fa60
--- /dev/null
+++ b/src/pulsecore/rtpoll-api.h
@@ -0,0 +1,32 @@
+#ifndef foopulsertpollapihfoo
+#define foopulsertpollapihfoo
+
+/***
+  This file is part of PulseAudio.
+
+  Copyright 2009 Lennart Poettering
+
+  PulseAudio is free software; you can redistribute it and/or modify
+  it under the terms of the GNU Lesser General Public License as
+  published by the Free Software Foundation; either version 2.1 of the
+  License, or (at your option) any later version.
+
+  PulseAudio 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
+  Lesser General Public License for more details.
+
+  You should have received a copy of the GNU Lesser General Public
+  License along with PulseAudio; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+  USA.
+***/
+
+/* A pa_mainloop_api implementation on top of rtpoll */
+
+#include <pulse/mainloop-api.h>
+#include <pulsecore/rtpoll.h>
+
+pa_mainloop_api* pa_rtpoll_get_api(pa_rtpoll *p);
+
+#endif
diff --git a/src/pulsecore/rtpoll.c b/src/pulsecore/rtpoll.c
index aa8ca321..dd3e240a 100644
--- a/src/pulsecore/rtpoll.c
+++ b/src/pulsecore/rtpoll.c
@@ -52,6 +52,7 @@
 #include <pulsecore/core-util.h>
 #include <pulsecore/winsock.h>
 #include <pulsecore/ratelimit.h>
+#include <pulsecore/prioq.h>
 
 #include "rtpoll.h"
 
@@ -61,7 +62,9 @@ struct pa_rtpoll {
     struct pollfd *pollfd, *pollfd2;
     unsigned n_pollfd_alloc, n_pollfd_used;
 
-    struct timeval next_elapse;
+    pa_prioq *prioq;
+
+    pa_usec_t elapse;
     pa_bool_t timer_enabled:1;
 
     pa_bool_t scan_for_dead:1;
@@ -70,16 +73,18 @@ struct pa_rtpoll {
     pa_bool_t rebuild_needed:1;
     pa_bool_t quit:1;
 
-#ifdef HAVE_PPOLL
+#if defined(HAVE_PPOLL) && defined(__linux__)
+    pa_bool_t use_ppoll:1;
+    pa_bool_t use_signals:1;
+
     pa_bool_t timer_armed:1;
-#ifdef __linux__
-    pa_bool_t dont_use_ppoll:1;
-#endif
     int rtsig;
     sigset_t sigset_unblocked;
     timer_t timer;
 #endif
 
+    void *userdata;
+
 #ifdef DEBUG_TIMING
     pa_usec_t timestamp;
     pa_usec_t slept, awake;
@@ -90,10 +95,15 @@ struct pa_rtpoll {
 
 struct pa_rtpoll_item {
     pa_rtpoll *rtpoll;
-    pa_bool_t dead;
-
     pa_rtpoll_priority_t priority;
 
+    pa_bool_t dead:1;
+
+    pa_bool_t timer_enabled:1;
+    pa_usec_t elapse;
+
+    pa_prioq_item *prioq_item;
+
     struct pollfd *pollfd;
     unsigned n_pollfd;
 
@@ -109,38 +119,40 @@ PA_STATIC_FLIST_DECLARE(items, 0, pa_xfree);
 
 static void signal_handler_noop(int s) { /* write(2, "signal\n", 7); */ }
 
-pa_rtpoll *pa_rtpoll_new(void) {
-    pa_rtpoll *p;
+static int item_compare(const void *_a, const void *_b) {
+    const pa_rtpoll_item *a = _a, *b = _b;
 
-    p = pa_xnew(pa_rtpoll, 1);
+    pa_assert(a->timer_enabled);
+    pa_assert(b->timer_enabled);
 
-#ifdef HAVE_PPOLL
+    if (a->elapse < b->elapse)
+        return -1;
+    if (a->elapse > b->elapse)
+        return 1;
 
-#ifdef __linux__
-    /* ppoll is broken on Linux < 2.6.16 */
-    p->dont_use_ppoll = FALSE;
+    return 0;
+}
 
-    {
-        struct utsname u;
-        unsigned major, minor, micro;
+pa_rtpoll *pa_rtpoll_new(void) {
+    pa_rtpoll *p;
 
-        pa_assert_se(uname(&u) == 0);
+    p = pa_xnew(pa_rtpoll, 1);
 
-        if (sscanf(u.release, "%u.%u.%u", &major, &minor, &micro) != 3 ||
-            (major < 2) ||
-            (major == 2 && minor < 6) ||
-            (major == 2 && minor == 6 && micro < 16))
+    p->userdata = NULL;
 
-            p->dont_use_ppoll = TRUE;
-    }
+#if defined(HAVE_PPOLL) && defined(__linux__)
+    /* ppoll() is broken on Linux < 2.6.16. Don't use it. */
+    p->use_ppoll = pa_linux_newer_than(2, 6, 16);
 
-#endif
+    /* Starting with Linux 2.6.28 ppoll() does no longer round up
+     * timeouts to multiple of HZ, hence using signal based timers is
+     * no longer necessary. */
+    p->use_signals = p->use_ppoll && !pa_linux_newer_than(2, 6, 28);
 
     p->rtsig = -1;
     sigemptyset(&p->sigset_unblocked);
     p->timer = (timer_t) -1;
     p->timer_armed = FALSE;
-
 #endif
 
     p->n_pollfd_alloc = 32;
@@ -148,7 +160,9 @@ pa_rtpoll *pa_rtpoll_new(void) {
     p->pollfd2 = pa_xnew(struct pollfd, p->n_pollfd_alloc);
     p->n_pollfd_used = 0;
 
-    memset(&p->next_elapse, 0, sizeof(p->next_elapse));
+    p->prioq = pa_prioq_new(item_compare);
+
+    p->elapse = 0;
     p->timer_enabled = FALSE;
 
     p->running = FALSE;
@@ -173,11 +187,10 @@ void pa_rtpoll_install(pa_rtpoll *p) {
 
     p->installed = TRUE;
 
-#ifdef HAVE_PPOLL
-# ifdef __linux__
-    if (p->dont_use_ppoll)
+#if defined(HAVE_PPOLL) && defined(__linux__)
+
+    if (!p->use_signals)
         return;
-# endif
 
     if ((p->rtsig = pa_rtsig_get_for_thread()) < 0) {
         pa_log_warn("Failed to reserve POSIX realtime signal.");
@@ -250,7 +263,6 @@ static void rtpoll_rebuild(pa_rtpoll *p) {
 
     if (ra)
         p->pollfd2 = pa_xrealloc(p->pollfd2, p->n_pollfd_alloc * sizeof(struct pollfd));
-
 }
 
 static void rtpoll_item_destroy(pa_rtpoll_item *i) {
@@ -262,11 +274,15 @@ static void rtpoll_item_destroy(pa_rtpoll_item *i) {
 
     PA_LLIST_REMOVE(pa_rtpoll_item, p->items, i);
 
+    pa_assert(p->n_pollfd_used >= i->n_pollfd);
     p->n_pollfd_used -= i->n_pollfd;
 
     if (pa_flist_push(PA_STATIC_FLIST_GET(items), i) < 0)
         pa_xfree(i);
 
+    if (i->prioq_item)
+        pa_prioq_remove(p->prioq, i->prioq_item);
+
     p->rebuild_needed = TRUE;
 }
 
@@ -279,11 +295,14 @@ void pa_rtpoll_free(pa_rtpoll *p) {
     pa_xfree(p->pollfd);
     pa_xfree(p->pollfd2);
 
-#ifdef HAVE_PPOLL
+#if defined(HAVE_PPOLL) && defined(__linux__)
     if (p->timer != (timer_t) -1)
         timer_delete(p->timer);
 #endif
 
+    if (p->prioq)
+        pa_prioq_free(p->prioq, NULL, NULL);
+
     pa_xfree(p);
 }
 
@@ -314,10 +333,36 @@ static void reset_all_revents(pa_rtpoll *p) {
     }
 }
 
+static pa_bool_t next_elapse(pa_rtpoll *p, pa_usec_t *usec) {
+    pa_rtpoll_item *i;
+
+    pa_assert(p);
+    pa_assert(usec);
+
+    i = pa_prioq_peek(p->prioq);
+
+    if (p->timer_enabled) {
+
+        if (i && i->timer_enabled)
+            *usec = PA_MIN(i->elapse, p->elapse);
+        else
+            *usec = p->elapse;
+
+        return TRUE;
+
+    } else if (i && i->timer_enabled) {
+        *usec = i->elapse;
+        return TRUE;
+    }
+
+    return FALSE;
+}
+
 int pa_rtpoll_run(pa_rtpoll *p, pa_bool_t wait) {
     pa_rtpoll_item *i;
     int r = 0;
-    struct timeval timeout;
+    pa_usec_t timeout;
+    pa_bool_t timeout_valid;
 
     pa_assert(p);
     pa_assert(!p->running);
@@ -381,15 +426,20 @@ int pa_rtpoll_run(pa_rtpoll *p, pa_bool_t wait) {
     if (p->rebuild_needed)
         rtpoll_rebuild(p);
 
-    memset(&timeout, 0, sizeof(timeout));
+    timeout = 0;
+    timeout_valid = FALSE;
 
     /* Calculate timeout */
-    if (wait && !p->quit && p->timer_enabled) {
-        struct timeval now;
-        pa_rtclock_get(&now);
+    if (wait && !p->quit) {
+        pa_usec_t elapse;
 
-        if (pa_timeval_cmp(&p->next_elapse, &now) > 0)
-            pa_timeval_add(&timeout, pa_timeval_diff(&p->next_elapse, &now));
+        if (next_elapse(p, &elapse)) {
+            pa_usec_t now;
+
+            now = pa_rtclock_usec();
+            timeout = now >= elapse ? 0 : elapse - now;
+            timeout_valid = TRUE;
+        }
     }
 
 #ifdef DEBUG_TIMING
@@ -404,20 +454,22 @@ int pa_rtpoll_run(pa_rtpoll *p, pa_bool_t wait) {
 #ifdef HAVE_PPOLL
 
 #ifdef __linux__
-    if (!p->dont_use_ppoll)
+    if (p->use_ppoll)
 #endif
     {
         struct timespec ts;
-        ts.tv_sec = timeout.tv_sec;
-        ts.tv_nsec = timeout.tv_usec * 1000;
-        r = ppoll(p->pollfd, p->n_pollfd_used, (!wait || p->quit || p->timer_enabled) ? &ts : NULL, p->rtsig < 0 ? NULL : &p->sigset_unblocked);
+        pa_timespec_store(&ts, timeout);
+        r = ppoll(p->pollfd, p->n_pollfd_used,
+                  (!wait || p->quit || timeout_valid) ? &ts : NULL,
+                  p->rtsig < 0 ? NULL : &p->sigset_unblocked);
     }
 #ifdef __linux__
     else
 #endif
 
 #endif
-        r = poll(p->pollfd, p->n_pollfd_used, (!wait || p->quit || p->timer_enabled) ? (int) ((timeout.tv_sec*1000) + (timeout.tv_usec / 1000)) : -1);
+        r = poll(p->pollfd, p->n_pollfd_used,
+                 (!wait || p->quit || timeout_valid) ? (int) (timeout / PA_USEC_PER_MSEC) : -1);
 
 #ifdef DEBUG_TIMING
     {
@@ -475,12 +527,10 @@ finish:
 static void update_timer(pa_rtpoll *p) {
     pa_assert(p);
 
-#ifdef HAVE_PPOLL
+#if defined(HAVE_PPOLL) && defined(__linux__)
 
-#ifdef __linux__
-    if (p->dont_use_ppoll)
+    if (!p->use_signals)
         return;
-#endif
 
     if (p->timer == (timer_t) -1) {
         struct sigevent se;
@@ -500,6 +550,7 @@ static void update_timer(pa_rtpoll *p) {
         struct itimerspec its;
         struct timespec ts = { .tv_sec = 0, .tv_nsec = 0 };
         sigset_t ss;
+        pa_usec_t elapse;
 
         if (p->timer_armed) {
             /* First disarm timer */
@@ -513,16 +564,16 @@ static void update_timer(pa_rtpoll *p) {
         }
 
         /* And install the new timer */
-        if (p->timer_enabled) {
+        if (next_elapse(p, &elapse)) {
             memset(&its, 0, sizeof(its));
 
-            its.it_value.tv_sec = p->next_elapse.tv_sec;
-            its.it_value.tv_nsec = p->next_elapse.tv_usec*1000;
+            pa_timespec_store(&its.it_value, elapse);
 
             /* Make sure that 0,0 is not understood as
              * "disarming" */
             if (its.it_value.tv_sec == 0 && its.it_value.tv_nsec == 0)
                 its.it_value.tv_nsec = 1;
+
             pa_assert_se(timer_settime(p->timer, TIMER_ABSTIME, &its, NULL) == 0);
         }
 
@@ -535,7 +586,10 @@ static void update_timer(pa_rtpoll *p) {
 void pa_rtpoll_set_timer_absolute(pa_rtpoll *p, pa_usec_t usec) {
     pa_assert(p);
 
-    pa_timeval_store(&p->next_elapse, usec);
+    if (p->timer_enabled && p->elapse == usec)
+        return
+
+    p->elapse = usec;
     p->timer_enabled = TRUE;
 
     update_timer(p);
@@ -547,22 +601,34 @@ void pa_rtpoll_set_timer_relative(pa_rtpoll *p, pa_usec_t usec) {
     /* Scheduling a timeout for more than an hour is very very suspicious */
     pa_assert(usec <= PA_USEC_PER_SEC*60ULL*60ULL);
 
-    pa_rtclock_get(&p->next_elapse);
-    pa_timeval_add(&p->next_elapse, usec);
-    p->timer_enabled = TRUE;
-
-    update_timer(p);
+    pa_rtpoll_set_timer_absolute(p, pa_rtclock_usec() + usec);
 }
 
-void pa_rtpoll_set_timer_disabled(pa_rtpoll *p) {
+void pa_rtpoll_disable_timer(pa_rtpoll *p) {
     pa_assert(p);
 
-    memset(&p->next_elapse, 0, sizeof(p->next_elapse));
+    if (!p->timer_enabled)
+        return;
+
+    p->elapse = 0;
     p->timer_enabled = FALSE;
 
     update_timer(p);
 }
 
+
+void pa_rtpoll_set_userdata(pa_rtpoll *p, void *userdata) {
+    pa_assert(p);
+
+    p->userdata = userdata;
+}
+
+void* pa_rtpoll_get_userdata(pa_rtpoll *p) {
+    pa_assert(p);
+
+    return p->userdata;
+}
+
 pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsigned n_fds) {
     pa_rtpoll_item *i, *j, *l = NULL;
 
@@ -576,6 +642,9 @@ pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsi
     i->n_pollfd = n_fds;
     i->pollfd = NULL;
     i->priority = prio;
+    i->timer_enabled = FALSE;
+    i->elapse = 0;
+    i->prioq_item = NULL;
 
     i->userdata = NULL;
     i->before_cb = NULL;
@@ -592,7 +661,7 @@ pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsi
     PA_LLIST_INSERT_AFTER(pa_rtpoll_item, p->items, j ? j->prev : l, i);
 
     if (n_fds > 0) {
-        p->rebuild_needed = 1;
+        p->rebuild_needed = TRUE;
         p->n_pollfd_used += n_fds;
     }
 
@@ -624,6 +693,63 @@ struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds) {
     return i->pollfd;
 }
 
+void pa_rtpoll_item_set_n_fds(pa_rtpoll_item *i, unsigned n_fds) {
+    pa_assert(i);
+
+    if (i->n_pollfd == n_fds)
+        return;
+
+    pa_assert(i->rtpoll->n_pollfd_used >= i->n_pollfd);
+    i->rtpoll->n_pollfd_used = i->rtpoll->n_pollfd_used - i->n_pollfd + n_fds;
+    i->n_pollfd = n_fds;
+
+    i->pollfd = NULL;
+    i->rtpoll->rebuild_needed = TRUE;
+}
+
+void pa_rtpoll_item_set_timer_absolute(pa_rtpoll_item *i, pa_usec_t usec){
+    pa_assert(i);
+
+    if (i->timer_enabled && i->elapse == usec)
+        return;
+
+    i->timer_enabled = TRUE;
+    i->elapse = usec;
+
+    if (i->prioq_item)
+        pa_prioq_reshuffle(i->rtpoll->prioq, i->prioq_item);
+    else
+        i->prioq_item = pa_prioq_put(i->rtpoll->prioq, i);
+
+    update_timer(i->rtpoll);
+}
+
+void pa_rtpoll_item_set_timer_relative(pa_rtpoll_item *i, pa_usec_t usec) {
+    pa_assert(i);
+
+    /* Scheduling a timeout for more than an hour is very very suspicious */
+    pa_assert(usec <= PA_USEC_PER_SEC*60ULL*60ULL);
+
+    pa_rtpoll_item_set_timer_absolute(i, pa_rtclock_usec() + usec);
+}
+
+void pa_rtpoll_item_disable_timer(pa_rtpoll_item *i) {
+    pa_assert(i);
+
+    if (!i->timer_enabled)
+        return;
+
+    i->timer_enabled = FALSE;
+    i->elapse = 0;
+
+    if (i->prioq_item) {
+        pa_prioq_remove(i->rtpoll->prioq, i->prioq_item);
+        i->prioq_item = NULL;
+    }
+
+    update_timer(i->rtpoll);
+}
+
 void pa_rtpoll_item_set_before_callback(pa_rtpoll_item *i, int (*before_cb)(pa_rtpoll_item *i)) {
     pa_assert(i);
     pa_assert(i->priority < PA_RTPOLL_NEVER);
@@ -657,6 +783,12 @@ void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i) {
     return i->userdata;
 }
 
+pa_rtpoll *pa_rtpoll_item_rtpoll(pa_rtpoll_item *i) {
+    pa_assert(i);
+
+    return i->rtpoll;
+}
+
 static int fdsem_before(pa_rtpoll_item *i) {
 
     if (pa_fdsem_before_poll(i->userdata) < 0)
diff --git a/src/pulsecore/rtpoll.h b/src/pulsecore/rtpoll.h
index 08776ef0..8c3cc70d 100644
--- a/src/pulsecore/rtpoll.h
+++ b/src/pulsecore/rtpoll.h
@@ -74,7 +74,10 @@ int pa_rtpoll_run(pa_rtpoll *f, pa_bool_t wait);
 
 void pa_rtpoll_set_timer_absolute(pa_rtpoll *p, pa_usec_t usec);
 void pa_rtpoll_set_timer_relative(pa_rtpoll *p, pa_usec_t usec);
-void pa_rtpoll_set_timer_disabled(pa_rtpoll *p);
+void pa_rtpoll_disable_timer(pa_rtpoll *p);
+
+void pa_rtpoll_set_userdata(pa_rtpoll *i, void *userdata);
+void* pa_rtpoll_get_userdata(pa_rtpoll *i);
 
 /* A new fd wakeup item for pa_rtpoll */
 pa_rtpoll_item *pa_rtpoll_item_new(pa_rtpoll *p, pa_rtpoll_priority_t prio, unsigned n_fds);
@@ -85,6 +88,8 @@ void pa_rtpoll_item_free(pa_rtpoll_item *i);
  * using the pointer and don't save the result anywhere */
 struct pollfd *pa_rtpoll_item_get_pollfd(pa_rtpoll_item *i, unsigned *n_fds);
 
+void pa_rtpoll_item_set_n_fds(pa_rtpoll_item *i, unsigned n_fds);
+
 /* Set the callback that shall be called when there's time to do some work: If the
  * callback returns a value > 0, the poll is skipped and the next
  * iteraton of the loop will start immediately. */
@@ -103,6 +108,12 @@ void pa_rtpoll_item_set_after_callback(pa_rtpoll_item *i, void (*after_cb)(pa_rt
 void pa_rtpoll_item_set_userdata(pa_rtpoll_item *i, void *userdata);
 void* pa_rtpoll_item_get_userdata(pa_rtpoll_item *i);
 
+void pa_rtpoll_item_set_timer_absolute(pa_rtpoll_item *i, pa_usec_t usec);
+void pa_rtpoll_item_set_timer_relative(pa_rtpoll_item *i, pa_usec_t usec);
+void pa_rtpoll_item_disable_timer(pa_rtpoll_item *i);
+
+pa_rtpoll *pa_rtpoll_item_rtpoll(pa_rtpoll_item *i);
+
 pa_rtpoll_item *pa_rtpoll_item_new_fdsem(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_fdsem *s);
 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq_read(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q);
 pa_rtpoll_item *pa_rtpoll_item_new_asyncmsgq_write(pa_rtpoll *p, pa_rtpoll_priority_t prio, pa_asyncmsgq *q);