make pa_mempool_stat thread-safe/lock-free

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

5 files changed, 76 insertions, 71 deletions

diff --git a/src/pulsecore/cli-command.c b/src/pulsecore/cli-command.c
index 8ea9262b..ae475c3a 100644
--- a/src/pulsecore/cli-command.c
+++ b/src/pulsecore/cli-command.c
@@ -259,20 +259,20 @@ static int pa_cli_command_stat(pa_core *c, pa_tokenizer *t, pa_strbuf *buf, PA_G
     stat = pa_mempool_get_stat(c->mempool);
     
     pa_strbuf_printf(buf, "Memory blocks currently allocated: %u, size: %s.\n",
-                     stat->n_allocated,
-                     pa_bytes_snprint(s, sizeof(s), stat->allocated_size));
+                     (unsigned) AO_load_acquire_read((AO_t*) &stat->n_allocated),
+                     pa_bytes_snprint(s, sizeof(s), (size_t) AO_load_acquire_read((AO_t*) &stat->allocated_size)));
 
     pa_strbuf_printf(buf, "Memory blocks allocated during the whole lifetime: %u, size: %s.\n",
-                     stat->n_accumulated,
-                     pa_bytes_snprint(s, sizeof(s), stat->accumulated_size));
+                     (unsigned) AO_load_acquire_read((AO_t*) &stat->n_accumulated),
+                     pa_bytes_snprint(s, sizeof(s), (size_t) AO_load_acquire_read((AO_t*) &stat->accumulated_size)));
 
     pa_strbuf_printf(buf, "Memory blocks imported from other processes: %u, size: %s.\n",
-                     stat->n_imported,
-                     pa_bytes_snprint(s, sizeof(s), stat->imported_size));
+                     (unsigned) AO_load_acquire_read((AO_t*) &stat->n_imported),
+                     pa_bytes_snprint(s, sizeof(s), (size_t) AO_load_acquire_read((AO_t*) &stat->imported_size)));
 
     pa_strbuf_printf(buf, "Memory blocks exported to other processes: %u, size: %s.\n",
-                     stat->n_exported,
-                     pa_bytes_snprint(s, sizeof(s), stat->exported_size));
+                     (unsigned) AO_load_acquire_read((AO_t*) &stat->n_exported),
+                     pa_bytes_snprint(s, sizeof(s), (size_t) AO_load_acquire_read((AO_t*) &stat->exported_size)));
 
     pa_strbuf_printf(buf, "Total sample cache size: %s.\n",
                      pa_bytes_snprint(s, sizeof(s), pa_scache_total_size(c)));
@@ -289,8 +289,8 @@ static int pa_cli_command_stat(pa_core *c, pa_tokenizer *t, pa_strbuf *buf, PA_G
         pa_strbuf_printf(buf,
                          "Memory blocks of type %s: %u allocated/%u accumulated.\n",
                          type_table[k],
-                         stat->n_allocated_by_type[k],
-                         stat->n_accumulated_by_type[k]);
+                         (unsigned) AO_load_acquire_read(&stat->n_allocated_by_type[k]),
+                         (unsigned) AO_load_acquire_read(&stat->n_accumulated_by_type[k]));
     
     return 0;
 }
diff --git a/src/pulsecore/memblock.c b/src/pulsecore/memblock.c
index 516ade13..70bcf677 100644
--- a/src/pulsecore/memblock.c
+++ b/src/pulsecore/memblock.c
@@ -112,39 +112,40 @@ static void stat_add(pa_memblock*b) {
     assert(b);
     assert(b->pool);
 
-    b->pool->stat.n_allocated ++;
-    b->pool->stat.n_accumulated ++;
-    b->pool->stat.allocated_size += b->length;
-    b->pool->stat.accumulated_size += b->length;
+    AO_fetch_and_add1_release_write(&b->pool->stat.n_allocated);
+    AO_fetch_and_add_release_write(&b->pool->stat.allocated_size, (AO_t) b->length);
+
+    AO_fetch_and_add1_release_write(&b->pool->stat.n_accumulated);
+    AO_fetch_and_add_release_write(&b->pool->stat.accumulated_size, (AO_t) b->length);
 
     if (b->type == PA_MEMBLOCK_IMPORTED) {
-        b->pool->stat.n_imported++;
-        b->pool->stat.imported_size += b->length;
+        AO_fetch_and_add1_release_write(&b->pool->stat.n_imported);
+        AO_fetch_and_add_release_write(&b->pool->stat.imported_size, (AO_t) b->length);
     }
 
-    b->pool->stat.n_allocated_by_type[b->type]++;
-    b->pool->stat.n_accumulated_by_type[b->type]++;
+    AO_fetch_and_add1_release_write(&b->pool->stat.n_allocated_by_type[b->type]);
+    AO_fetch_and_add1_release_write(&b->pool->stat.n_accumulated_by_type[b->type]);
 }
 
 static void stat_remove(pa_memblock *b) {
     assert(b);
     assert(b->pool);
 
-    assert(b->pool->stat.n_allocated > 0);
-    assert(b->pool->stat.allocated_size >= b->length);
+    assert(AO_load_acquire_read(&b->pool->stat.n_allocated) > 0);
+    assert(AO_load_acquire_read(&b->pool->stat.allocated_size) >= (AO_t) b->length);
            
-    b->pool->stat.n_allocated --;
-    b->pool->stat.allocated_size -= b->length;
+    AO_fetch_and_sub1_release_write(&b->pool->stat.n_allocated);
+    AO_fetch_and_add_release_write(&b->pool->stat.allocated_size,  (AO_t) (-b->length));
 
     if (b->type == PA_MEMBLOCK_IMPORTED) {
-        assert(b->pool->stat.n_imported > 0);
-        assert(b->pool->stat.imported_size >= b->length);
+        assert(AO_load_acquire_read(&b->pool->stat.n_imported) > 0);
+        assert(AO_load_acquire_read(&b->pool->stat.imported_size) >= (AO_t) b->length);
         
-        b->pool->stat.n_imported --;
-        b->pool->stat.imported_size -= b->length;
+        AO_fetch_and_sub1_release_write(&b->pool->stat.n_imported);
+        AO_fetch_and_add_release_write(&b->pool->stat.imported_size, (AO_t)  (-b->length));
     }
 
-    b->pool->stat.n_allocated_by_type[b->type]--;
+    AO_fetch_and_sub1_release_write(&b->pool->stat.n_allocated_by_type[b->type]);
 }
 
 static pa_memblock *memblock_new_appended(pa_mempool *p, size_t length);
@@ -190,7 +191,7 @@ static struct mempool_slot* mempool_allocate_slot(pa_mempool *p) {
         slot = (struct mempool_slot*) ((uint8_t*) p->memory.ptr + (p->block_size * p->n_init++));
     else {
         pa_log_debug("Pool full");
-        p->stat.n_pool_full++;
+        AO_fetch_and_add1_release_write(&p->stat.n_pool_full);
         return NULL;
     }
 
@@ -247,7 +248,7 @@ pa_memblock *pa_memblock_new_pool(pa_mempool *p, size_t length) {
         b->data = mempool_slot_data(slot);
     } else {
         pa_log_debug("Memory block too large for pool: %u > %u", length, p->block_size - sizeof(struct mempool_slot));
-        p->stat.n_too_large_for_pool++;
+        AO_fetch_and_add1_release_write(&p->stat.n_too_large_for_pool);
         return NULL;
     }
 
@@ -371,7 +372,7 @@ void pa_memblock_unref(pa_memblock*b) {
 static void memblock_make_local(pa_memblock *b) {
     assert(b);
 
-    b->pool->stat.n_allocated_by_type[b->type]--;
+    AO_fetch_and_sub1_release_write(&b->pool->stat.n_allocated_by_type[b->type]);
 
     if (b->length <= b->pool->block_size - sizeof(struct mempool_slot)) {
         struct mempool_slot *slot;
@@ -397,8 +398,8 @@ static void memblock_make_local(pa_memblock *b) {
     b->data = pa_xmemdup(b->data, b->length);
 
 finish:
-    b->pool->stat.n_allocated_by_type[b->type]++;
-    b->pool->stat.n_accumulated_by_type[b->type]++;
+    AO_fetch_and_add1_release_write(&b->pool->stat.n_allocated_by_type[b->type]);
+    AO_fetch_and_add1_release_write(&b->pool->stat.n_accumulated_by_type[b->type]);
 }
 
 void pa_memblock_unref_fixed(pa_memblock *b) {
@@ -418,10 +419,10 @@ static void memblock_replace_import(pa_memblock *b) {
     assert(b);
     assert(b->type == PA_MEMBLOCK_IMPORTED);
 
-    assert(b->pool->stat.n_imported > 0);
-    assert(b->pool->stat.imported_size >= b->length);
-    b->pool->stat.n_imported --;
-    b->pool->stat.imported_size -= b->length;
+    assert(AO_load_acquire_read(&b->pool->stat.n_imported) > 0);
+    assert(AO_load_acquire_read(&b->pool->stat.imported_size) >= (AO_t) b->length);
+    AO_fetch_and_sub1_release_write(&b->pool->stat.n_imported);
+    AO_fetch_and_add_release_write(&b->pool->stat.imported_size, (AO_t) - b->length);
 
     seg = b->per_type.imported.segment;
     assert(seg);
@@ -486,7 +487,7 @@ void pa_mempool_free(pa_mempool *p) {
     while (p->exports)
         pa_memexport_free(p->exports);
 
-    if (p->stat.n_allocated > 0)
+    if (AO_load_acquire_read(&p->stat.n_allocated) > 0)
         pa_log_warn("WARNING! Memory pool destroyed but not all memory blocks freed!");
     
     pa_shm_free(&p->memory);
@@ -685,11 +686,11 @@ int pa_memexport_process_release(pa_memexport *e, uint32_t id) {
 
 /*     pa_log("Processing release for %u", id); */
 
-    assert(e->pool->stat.n_exported > 0);
-    assert(e->pool->stat.exported_size >= e->slots[id].block->length);
+    assert(AO_load_acquire_read(&e->pool->stat.n_exported) > 0);
+    assert(AO_load_acquire_read(&e->pool->stat.exported_size) >= (AO_t) e->slots[id].block->length);
     
-    e->pool->stat.n_exported --;
-    e->pool->stat.exported_size -= e->slots[id].block->length;
+    AO_fetch_and_sub1_release_write(&e->pool->stat.n_exported);
+    AO_fetch_and_add_release_write(&e->pool->stat.exported_size, (AO_t) -e->slots[id].block->length);
     
     pa_memblock_unref(e->slots[id].block);
     e->slots[id].block = NULL;
@@ -786,8 +787,8 @@ int pa_memexport_put(pa_memexport *e, pa_memblock *b, uint32_t *block_id, uint32
     *offset = (uint8_t*) b->data - (uint8_t*) memory->ptr;
     *size = b->length;
 
-    e->pool->stat.n_exported ++;
-    e->pool->stat.exported_size += b->length;
+    AO_fetch_and_add1_release_write(&e->pool->stat.n_exported);
+    AO_fetch_and_add_release_write(&e->pool->stat.exported_size, (AO_t) b->length);
 
     return 0;
 }
diff --git a/src/pulsecore/memblock.h b/src/pulsecore/memblock.h
index 60a0c900..d4f2b7aa 100644
--- a/src/pulsecore/memblock.h
+++ b/src/pulsecore/memblock.h
@@ -74,21 +74,25 @@ struct pa_memblock {
     } per_type;
 };
 
+/* Please note that updates to this structure are not locked,
+ * i.e. n_allocated might be updated at a point in time where
+ * n_accumulated is not yet. Take these values with a grain of salt,
+ * threy are here for purely statistical reasons.*/
 struct pa_mempool_stat {
-    unsigned n_allocated;
-    unsigned n_accumulated;
-    unsigned n_imported;
-    unsigned n_exported;
-    size_t allocated_size;
-    size_t accumulated_size;
-    size_t imported_size;
-    size_t exported_size;
-
-    unsigned n_too_large_for_pool;
-    unsigned n_pool_full;
-
-    unsigned n_allocated_by_type[PA_MEMBLOCK_TYPE_MAX];
-    unsigned n_accumulated_by_type[PA_MEMBLOCK_TYPE_MAX];
+    AO_t n_allocated;
+    AO_t n_accumulated;
+    AO_t n_imported;
+    AO_t n_exported;
+    AO_t allocated_size;
+    AO_t accumulated_size;
+    AO_t imported_size;
+    AO_t exported_size;
+
+    AO_t n_too_large_for_pool;
+    AO_t n_pool_full;
+
+    AO_t n_allocated_by_type[PA_MEMBLOCK_TYPE_MAX];
+    AO_t n_accumulated_by_type[PA_MEMBLOCK_TYPE_MAX];
 };
 
 /* Allocate a new memory block of type PA_MEMBLOCK_MEMPOOL or PA_MEMBLOCK_APPENDED, depending on the size */
diff --git a/src/pulsecore/protocol-native.c b/src/pulsecore/protocol-native.c
index 0f015071..38c024b7 100644
--- a/src/pulsecore/protocol-native.c
+++ b/src/pulsecore/protocol-native.c
@@ -1112,10 +1112,10 @@ static void command_stat(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t
     stat = pa_mempool_get_stat(c->protocol->core->mempool);
     
     reply = reply_new(tag);
-    pa_tagstruct_putu32(reply, stat->n_allocated);
-    pa_tagstruct_putu32(reply, stat->allocated_size);
-    pa_tagstruct_putu32(reply, stat->n_accumulated);
-    pa_tagstruct_putu32(reply, stat->accumulated_size);
+    pa_tagstruct_putu32(reply, (uint32_t) AO_load_acquire_read((AO_t*) &stat->n_allocated));
+    pa_tagstruct_putu32(reply, (uint32_t) AO_load_acquire_read((AO_t*) &stat->allocated_size));
+    pa_tagstruct_putu32(reply, (uint32_t) AO_load_acquire_read((AO_t*) &stat->n_accumulated));
+    pa_tagstruct_putu32(reply, (uint32_t) AO_load_acquire_read((AO_t*) &stat->accumulated_size));
     pa_tagstruct_putu32(reply, pa_scache_total_size(c->protocol->core));
     pa_pstream_send_tagstruct(c->pstream, reply);
 }
diff --git a/src/tests/memblock-test.c b/src/tests/memblock-test.c
index 11198ae6..ef2e0ad7 100644
--- a/src/tests/memblock-test.c
+++ b/src/tests/memblock-test.c
@@ -54,16 +54,16 @@ static void print_stats(pa_mempool *p, const char *text) {
            "n_pool_full = %u\n"
            "}\n",
            text,
-           s->n_allocated,
-           s->n_accumulated,
-           s->n_imported,
-           s->n_exported,
-           (unsigned long) s->allocated_size,
-           (unsigned long) s->accumulated_size,
-           (unsigned long) s->imported_size,
-           (unsigned long) s->exported_size,
-           s->n_too_large_for_pool,
-           s->n_pool_full);
+           (unsigned) AO_load_acquire_read((AO_t*) &s->n_allocated),
+           (unsigned) AO_load_acquire_read((AO_t*) &s->n_accumulated),
+           (unsigned) AO_load_acquire_read((AO_t*) &s->n_imported),
+           (unsigned) AO_load_acquire_read((AO_t*) &s->n_exported),
+           (unsigned long) AO_load_acquire_read((AO_t*) &s->allocated_size),
+           (unsigned long) AO_load_acquire_read((AO_t*) &s->accumulated_size),
+           (unsigned long) AO_load_acquire_read((AO_t*) &s->imported_size),
+           (unsigned long) AO_load_acquire_read((AO_t*) &s->exported_size),
+           (unsigned) AO_load_acquire_read((AO_t*) &s->n_too_large_for_pool),
+           (unsigned) AO_load_acquire_read((AO_t*) &s->n_pool_full));
 }
 
 int main(int argc, char *argv[]) {