[pulseaudio-discuss] [PATCH 04/11] srchannel: Add the shared ringbuffer object

[Tanu Kaskinen]

On Tue, 2014-04-29 at 15:22 +0200, David Henningsson wrote:
> An shm ringbuffer that is used for low overhead server-client communication.
> Signalling is done through eventfd semaphores - it's based on pa_fdsem to avoid
> syscalls if nothing is waiting on the other side.
> 
> Signed-off-by: David Henningsson <david.henningsson at canonical.com>
> ---
>  src/Makefile.am           |   3 +-
>  src/pulsecore/srchannel.c | 297 ++++++++++++++++++++++++++++++++++++++++++++++
>  src/pulsecore/srchannel.h |  62 ++++++++++
>  3 files changed, 361 insertions(+), 1 deletion(-)
>  create mode 100644 src/pulsecore/srchannel.c
>  create mode 100644 src/pulsecore/srchannel.h
> 
> diff --git a/src/Makefile.am b/src/Makefile.am
> index 5c2d5bc..9e72982 100644
> --- a/src/Makefile.am
> +++ b/src/Makefile.am
> @@ -618,6 +618,7 @@ libpulsecommon_ at PA_MAJORMINOR@_la_SOURCES = \
>  		pulsecore/creds.h \
>  		pulsecore/dynarray.c pulsecore/dynarray.h \
>  		pulsecore/endianmacros.h \
> +		pulsecore/fdsem.c pulsecore/fdsem.h \
>  		pulsecore/flist.c pulsecore/flist.h \
>  		pulsecore/g711.c pulsecore/g711.h \
>  		pulsecore/hashmap.c pulsecore/hashmap.h \
> @@ -651,6 +652,7 @@ libpulsecommon_ at PA_MAJORMINOR@_la_SOURCES = \
>  		pulsecore/queue.c pulsecore/queue.h \
>  		pulsecore/random.c pulsecore/random.h \
>  		pulsecore/refcnt.h \
> +		pulsecore/srchannel.c pulsecore/srchannel.h \
>  		pulsecore/sample-util.c pulsecore/sample-util.h \
>  		pulsecore/shm.c pulsecore/shm.h \
>  		pulsecore/bitset.c pulsecore/bitset.h \
> @@ -880,7 +882,6 @@ libpulsecore_ at PA_MAJORMINOR@_la_SOURCES = \
>  		pulsecore/core-scache.c pulsecore/core-scache.h \
>  		pulsecore/core-subscribe.c pulsecore/core-subscribe.h \
>  		pulsecore/core.c pulsecore/core.h \
> -		pulsecore/fdsem.c pulsecore/fdsem.h \
>  		pulsecore/hook-list.c pulsecore/hook-list.h \
>  		pulsecore/ltdl-helper.c pulsecore/ltdl-helper.h \
>  		pulsecore/modargs.c pulsecore/modargs.h \
> diff --git a/src/pulsecore/srchannel.c b/src/pulsecore/srchannel.c
> new file mode 100644
> index 0000000..15485b2
> --- /dev/null
> +++ b/src/pulsecore/srchannel.c
> @@ -0,0 +1,297 @@
> +/***
> +  This file is part of PulseAudio.
> +
> +  Copyright 2014 David Henningsson, Canonical Ltd.
> +
> +  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 "srchannel.h"
> +
> +#include <pulsecore/atomic.h>
> +#include <pulse/xmalloc.h>
> +
> +/* #define DEBUG_SRCHANNEL */
> +
> +/* This ringbuffer might be useful in other contexts too, but
> +   right now it's only used inside the srchannel, so let's keep it here
> +   for the time being. */
> +typedef struct pa_ringbuffer pa_ringbuffer;
> +struct pa_ringbuffer {
> +    pa_atomic_t *count;
> +    int capacity;
> +    uint8_t *memory;
> +    int readindex, writeindex;
> +};
> +
> +static void *pa_ringbuffer_peek(pa_ringbuffer *r, int *count) {
> +    int c = pa_atomic_load(r->count);
> +    if (r->readindex + c > r->capacity)
> +        *count = r->capacity - r->readindex;
> +    else
> +        *count = c;
> +    return r->memory + r->readindex;
> +}
> +
> +/* Returns true only if the buffer was completely full before the drop. */
> +static bool pa_ringbuffer_drop(pa_ringbuffer *r, int count) {
> +    bool b = pa_atomic_sub(r->count, count) >= r->capacity;
> +    r->readindex += count;
> +    r->readindex %= r->capacity;
> +    return b;
> +}
> +
> +static void *pa_ringbuffer_begin_write(pa_ringbuffer *r, int *count) {
> +    int c = pa_atomic_load(r->count);
> +    *count = PA_MIN(r->capacity - r->writeindex, r->capacity - c);
> +    return r->memory + r->writeindex;
> +}
> +
> +static void pa_ringbuffer_end_write(pa_ringbuffer *r, int count) {
> +    pa_atomic_add(r->count, count);
> +    r->writeindex += count;
> +    r->writeindex %= r->capacity;
> +}
> +
> +struct pa_srchannel {
> +    pa_ringbuffer rb_read, rb_write;
> +    pa_fdsem *sem_read, *sem_write;
> +    pa_memblock *memblock;
> +    void *cb_userdata;
> +    pa_srchannel_cb_t callback;
> +    pa_io_event *read_event;
> +    pa_mainloop_api *mainloop;
> +};
> +
> +ssize_t pa_srchannel_write(pa_srchannel *sr, const void *data, size_t l) {
> +    ssize_t written = 0;

Why ssize_t instead of size_t?

> +    while (l > 0) {
> +        int towrite;
> +        void *ptr = pa_ringbuffer_begin_write(&sr->rb_write, &towrite);
> +        if ((size_t) towrite > l)

Why is towrite an int and not size_t? Probably because
pa_ringbuffer_begin_write() takes an int pointer, but why doesn't it
take a size_t pointer?

> +            towrite = l;
> +        if (towrite == 0) {
> +#ifdef DEBUG_SRCHANNEL
> +            pa_log("srchannel output buffer full");
> +#endif
> +            break;
> +        }
> +        memcpy(ptr, data, towrite);
> +        pa_ringbuffer_end_write(&sr->rb_write, towrite);
> +        written += towrite;
> +        data = (uint8_t*) data + towrite;
> +        l -= towrite;
> +    }
> +#ifdef DEBUG_SRCHANNEL
> +    pa_log("Wrote %d bytes to srchannel, signalling fdsem", (int) written);
> +#endif
> +    pa_fdsem_post(sr->sem_write);

This should be sem_read?

> +    return written;
> +}
> +
> +ssize_t pa_srchannel_read(pa_srchannel *sr, void *data, size_t l) {
> +    ssize_t isread = 0;
> +    while (l > 0) {
> +        int toread;
> +        void *ptr = pa_ringbuffer_peek(&sr->rb_read, &toread);
> +        if ((size_t) toread > l)

Same comments about the types as in pa_srchannel_write().

Minor thing: "isread" sounds like it's a boolean. I suggest
"read_so_far" as the variable name, but feel free to ignore this if you
prefer "isread".

> +void pa_srchannel_free(pa_srchannel *sr)
> +{
> +#ifdef DEBUG_SRCHANNEL
> +    pa_log("Freeing srchannel");
> +#endif
> +    if (!sr)
> +        return;

Please don't allow NULL pointers in freeing functions.

-- 
Tanu