rework subscription code: try to drop redundant queued events

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1211 fefdeb5f-60dc-0310-8127-8f9354f1896f

src/pulsecore/core-subscribe.c

@@ -43,75 +43,81 @@
 struct pa_subscription {
     pa_core *core;
     int dead;
-    void (*callback)(pa_core *c, pa_subscription_event_type_t t, uint32_t index, void *userdata);
+
+    pa_subscription_cb_t callback;
     void *userdata;
     pa_subscription_mask_t mask;
 
-    pa_subscription *prev, *next;
+    PA_LLIST_FIELDS(pa_subscription);
 };
 
 struct pa_subscription_event {
+    pa_core *core;
+
     pa_subscription_event_type_t type;
     uint32_t index;
+
+    PA_LLIST_FIELDS(pa_subscription_event);
 };
 
 static void sched_event(pa_core *c);
 
 /* Allocate a new subscription object for the given subscription mask. Use the specified callback function and user data */
-pa_subscription* pa_subscription_new(pa_core *c, pa_subscription_mask_t m, void (*callback)(pa_core *c, pa_subscription_event_type_t t, uint32_t index, void *userdata), void *userdata) {
+pa_subscription* pa_subscription_new(pa_core *c, pa_subscription_mask_t m, pa_subscription_cb_t callback, void *userdata) {
     pa_subscription *s;
-    assert(c);
     
-    s = pa_xmalloc(sizeof(pa_subscription));
+    assert(c);
+    assert(m);
+    assert(callback);
+
+    s = pa_xnew(pa_subscription, 1);
     s->core = c;
     s->dead = 0;
     s->callback = callback;
     s->userdata = userdata;
     s->mask = m;
 
-    if ((s->next = c->subscriptions))
+    PA_LLIST_PREPEND(pa_subscription, c->subscriptions, s);
-        s->next->prev = s;
-    s->prev = NULL;
-    c->subscriptions = s;
     return s;
 }
 
 /* Free a subscription object, effectively marking it for deletion */
 void pa_subscription_free(pa_subscription*s) {
-    assert(s && !s->dead);
+    assert(s);
+    assert(!s->dead);
+    
     s->dead = 1;
     sched_event(s->core);
 }
 
-static void free_item(pa_subscription *s) {
+static void free_subscription(pa_subscription *s) {
-    assert(s && s->core);
+    assert(s);
+    assert(s->core);
 
-    if (s->prev)
+    PA_LLIST_REMOVE(pa_subscription, s->core->subscriptions, s);
-        s->prev->next = s->next;
+    pa_xfree(s);
-    else
+}
-        s->core->subscriptions = s->next;
 
-    if (s->next)
+static void free_event(pa_subscription_event *s) {
-        s->next->prev = s->prev;
+    assert(s);
+    assert(s->core);
 
+    if (!s->next)
+        s->core->subscription_event_last = s->prev;
+    
+    PA_LLIST_REMOVE(pa_subscription_event, s->core->subscription_event_queue, s);
     pa_xfree(s);
 }
 
 /* Free all subscription objects */
 void pa_subscription_free_all(pa_core *c) {
-    pa_subscription_event *e;
     assert(c);
     
     while (c->subscriptions)
-        free_item(c->subscriptions);
+        free_subscription(c->subscriptions);
 
-    if (c->subscription_event_queue) {
+    while (c->subscription_event_queue)
-        while ((e = pa_queue_pop(c->subscription_event_queue)))
+        free_event(c->subscription_event_queue);
-            pa_xfree(e);
-        
-        pa_queue_free(c->subscription_event_queue, NULL, NULL);
-        c->subscription_event_queue = NULL;
-    }
 
     if (c->subscription_defer_event) {
         c->mainloop->defer_free(c->subscription_defer_event);
@@ -119,48 +125,31 @@ void pa_subscription_free_all(pa_core *c) {
     }
 }
 
-#if 0
+#ifdef DEBUG
-static void dump_event(pa_subscription_event*e) {
+static void dump_event(const char * prefix, pa_subscription_event*e) {
-    switch (e->type & PA_SUBSCRIPTION_EVENT_FACILITY_MASK) {
+    const char * const fac_table[] = {
-        case PA_SUBSCRIPTION_EVENT_SINK:
+        [PA_SUBSCRIPTION_EVENT_SINK] = "SINK",
-            pa_log(__FILE__": SINK_EVENT");
+        [PA_SUBSCRIPTION_EVENT_SOURCE] = "SOURCE",
-            break;
+        [PA_SUBSCRIPTION_EVENT_SINK_INPUT] = "SINK_INPUT",
-        case PA_SUBSCRIPTION_EVENT_SOURCE:
+        [PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT] = "SOURCE_OUTPUT",
-            pa_log(__FILE__": SOURCE_EVENT");
+        [PA_SUBSCRIPTION_EVENT_MODULE] = "MODULE",
-            break;
+        [PA_SUBSCRIPTION_EVENT_CLIENT] = "CLIENT",
-        case PA_SUBSCRIPTION_EVENT_SINK_INPUT:
+        [PA_SUBSCRIPTION_EVENT_SAMPLE_CACHE] = "SAMPLE_CACHE",
-            pa_log(__FILE__": SINK_INPUT_EVENT");
+        [PA_SUBSCRIPTION_EVENT_SERVER] = "SERVER",
-            break;
+        [PA_SUBSCRIPTION_EVENT_AUTOLOAD] = "AUTOLOAD"
-        case PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT:
+    };
-            pa_log(__FILE__": SOURCE_OUTPUT_EVENT");
-            break;
-        case PA_SUBSCRIPTION_EVENT_MODULE:
-            pa_log(__FILE__": MODULE_EVENT");
-            break;
-        case PA_SUBSCRIPTION_EVENT_CLIENT:
-            pa_log(__FILE__": CLIENT_EVENT");
-            break;
-        default:
-            pa_log(__FILE__": OTHER");
-            break;
-    }
 
-    switch (e->type & PA_SUBSCRIPTION_EVENT_TYPE_MASK) {
+    const char * const type_table[] = {
-        case PA_SUBSCRIPTION_EVENT_NEW:
+        [PA_SUBSCRIPTION_EVENT_NEW] = "NEW",
-            pa_log(__FILE__":  NEW");
+        [PA_SUBSCRIPTION_EVENT_CHANGE] = "CHANGE",
-            break;
+        [PA_SUBSCRIPTION_EVENT_REMOVE] = "REMOVE"
-        case PA_SUBSCRIPTION_EVENT_CHANGE:
+    };
-            pa_log(__FILE__":  CHANGE");
-            break;
-        case PA_SUBSCRIPTION_EVENT_REMOVE:
-            pa_log(__FILE__":  REMOVE");
-            break;
-        default:
-            pa_log(__FILE__":  OTHER");
-            break;
-    }
 
-    pa_log(__FILE__":  %u", e->index);
+    pa_log(__FILE__": %s event (%s|%s|%u)",
+           prefix,
+           fac_table[e->type & PA_SUBSCRIPTION_EVENT_FACILITY_MASK],
+           type_table[e->type & PA_SUBSCRIPTION_EVENT_TYPE_MASK],
+           e->index);
 }
 #endif
 
@@ -168,16 +157,17 @@ static void dump_event(pa_subscription_event*e) {
 static void defer_cb(pa_mainloop_api *m, pa_defer_event *de, void *userdata) {
     pa_core *c = userdata;
     pa_subscription *s;
-    assert(c && c->subscription_defer_event == de && c->mainloop == m);
+    
+    assert(c->mainloop == m);
+    assert(c);
+    assert(c->subscription_defer_event == de);
 
     c->mainloop->defer_enable(c->subscription_defer_event, 0);
 
     /* Dispatch queued events */
 
-    if (c->subscription_event_queue) {
+    while (c->subscription_event_queue) {
-        pa_subscription_event *e;
+        pa_subscription_event *e = c->subscription_event_queue;
-        
-        while ((e = pa_queue_pop(c->subscription_event_queue))) {
 
         for (s = c->subscriptions; s; s = s->next) {
             
@@ -185,8 +175,10 @@ static void defer_cb(pa_mainloop_api *m, pa_defer_event *de, void *userdata) {
                 s->callback(c, e->type, e->index, s->userdata);
         }
 
-            pa_xfree(e);
+#ifdef DEBUG
-        }
+        dump_event("Dispatched", e);
+#endif
+        free_event(e);
     }
 
     /* Remove dead subscriptions */
@@ -195,7 +187,7 @@ static void defer_cb(pa_mainloop_api *m, pa_defer_event *de, void *userdata) {
     while (s) {
         pa_subscription *n = s->next;
         if (s->dead)
-            free_item(s);
+            free_subscription(s);
         s = n;
     }
 }
@@ -217,17 +209,52 @@ void pa_subscription_post(pa_core *c, pa_subscription_event_type_t t, uint32_t i
     pa_subscription_event *e;
     assert(c);
 
-    e = pa_xmalloc(sizeof(pa_subscription_event));
+    if ((t & PA_SUBSCRIPTION_EVENT_TYPE_MASK) != PA_SUBSCRIPTION_EVENT_NEW) {
+        pa_subscription_event *i, *n;
+        
+        /* Check for duplicates */
+        for (i = c->subscription_event_last; i; i = n) {
+            n = i->prev;
+                
+            /* not the same object type */
+            if (((t ^ i->type) & PA_SUBSCRIPTION_EVENT_FACILITY_MASK))
+                continue;
+            
+            /* not the same object */
+            if (i->index != index)
+                continue;
+
+            if ((t & PA_SUBSCRIPTION_EVENT_TYPE_MASK) == PA_SUBSCRIPTION_EVENT_REMOVE) {
+                /* This object is being removed, hence there is no
+                 * point in keeping the old events regarding this
+                 * entry in the queue. */
+
+                free_event(i);
+                pa_log_debug(__FILE__": dropped redundant event.");
+                continue;
+            }
+
+            if ((t & PA_SUBSCRIPTION_EVENT_TYPE_MASK) == PA_SUBSCRIPTION_EVENT_CHANGE) {
+                /* This object has changed. If a "new" or "change" event for
+                 * this object is still in the queue we can exit. */
+
+                pa_log_debug(__FILE__": dropped redundant event.");
+                return;
+            }
+        }
+    }
+
+    e = pa_xnew(pa_subscription_event, 1);
+    e->core = c;
     e->type = t;
     e->index = index;
 
-    if (!c->subscription_event_queue) {
+    PA_LLIST_INSERT_AFTER(pa_subscription_event, c->subscription_event_queue, c->subscription_event_last, e); 
-        c->subscription_event_queue = pa_queue_new();
+    c->subscription_event_last = e;
-        assert(c->subscription_event_queue);
+
-    }
+#ifdef DEBUG
+    dump_event("Queued", e);
+#endif
 
-    pa_queue_push(c->subscription_event_queue, e);
     sched_event(c);
 }
-
-

src/pulsecore/core-subscribe.h

@@ -28,7 +28,9 @@ typedef struct pa_subscription_event pa_subscription_event;
 #include <pulsecore/core.h>
 #include <pulsecore/native-common.h>
 
-pa_subscription* pa_subscription_new(pa_core *c, pa_subscription_mask_t m,  void (*callback)(pa_core *c, pa_subscription_event_type_t t, uint32_t index, void *userdata), void *userdata);
+typedef void (*pa_subscription_cb_t)(pa_core *c, pa_subscription_event_type_t t, uint32_t idx, void *userdata);
+
+pa_subscription* pa_subscription_new(pa_core *c, pa_subscription_mask_t m,  pa_subscription_cb_t cb, void *userdata);
 void pa_subscription_free(pa_subscription*s);
 void pa_subscription_free_all(pa_core *c);
 

src/pulsecore/core.c

@@ -74,8 +74,9 @@ pa_core* pa_core_new(pa_mainloop_api *m) {
     c->scache_auto_unload_event = NULL;
 
     c->subscription_defer_event = NULL;
-    c->subscription_event_queue = NULL;
+    PA_LLIST_HEAD_INIT(pa_subscription, c->subscriptions);
-    c->subscriptions = NULL;
+    PA_LLIST_HEAD_INIT(pa_subscription_event, c->subscription_event_queue);
+    c->subscription_event_last = NULL;
 
     c->memblock_stat = pa_memblock_stat_new();
 

src/pulsecore/core.h

@@ -24,14 +24,15 @@
 
 typedef struct pa_core pa_core;
 
-#include <pulsecore/idxset.h>
-#include <pulsecore/hashmap.h>
 #include <pulse/mainloop-api.h>
 #include <pulse/sample.h>
+#include <pulsecore/idxset.h>
+#include <pulsecore/hashmap.h>
 #include <pulsecore/memblock.h>
 #include <pulsecore/resampler.h>
 #include <pulsecore/queue.h>
 #include <pulsecore/core-subscribe.h>
+#include <pulsecore/llist.h>
 
 /* The core structure of PulseAudio. Every PulseAudio daemon contains
  * exactly one of these. It is used for storing kind of global
@@ -58,8 +59,9 @@ struct pa_core {
     pa_defer_event *module_defer_unload_event;
 
     pa_defer_event *subscription_defer_event;
-    pa_queue *subscription_event_queue;
+    PA_LLIST_HEAD(pa_subscription, subscriptions);
-    pa_subscription *subscriptions;
+    PA_LLIST_HEAD(pa_subscription_event, subscription_event_queue);
+    pa_subscription_event *subscription_event_last;
 
     pa_memblock_stat *memblock_stat;