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

[David Henningsson]

On 2014-05-02 17:00, Tanu Kaskinen wrote:
> 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?

I wanted it to look like pa_iochannel_write.

I prefer using int out of simplicity, as size_t always gives these 
"compared signed and unsigned" warnings.

Maybe we can just convert all size_t and ssize_t's to int instead?

>> +    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?

Same answer as above.

>> +            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?

Eh, no? Not sure why you think that?

There is a gotcha that I should probably comment though, but not sure if 
that's what confuses you.
There is only one semaphore that we listen on. The other one is the one 
we always write to.
That means that we use the same semaphore for two different meanings:
"I've written something in my outgoing buffer to you" and
"I've read something in my incoming buffer, and my incoming buffer was 
full, but it is no longer full".

I e, when the signalled process is woken up it should check its incoming 
buffer to see if there's anything to read, and if it's waiting to write 
more, it should check if it's okay to write more as well.

>> +    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".

Ok.


-- 
David Henningsson, Canonical Ltd.
https://launchpad.net/~diwic