[PATCH] include: Add single-producer single-consumer ringbuffer implementation.

[Davide Beatrici]

Please note that I wrote a test for the ringbuffer, but I'm not sure 
where I should put it in the repository.

[Davide Beatrici]

I apologize, the patch was not attached to the initial message.

From: Davide Beatrici 
Date: Tue, 15 Feb 2022 01:06:13 +0100
Subject: [PATCH] include: Add single-producer single-consumer ringbuffer
  implementation.

The ringbuffer was written for the PipeWire driver I'm working on.

I'm committing it separately because its implementation is generic,
allowing it to be used pretty much universally.

Signed-off-by: Davide Beatrici 
---
  include/wine/ringbuffer.h | 156 ++++++++++++++++++++++++++++++++++++++
  1 file changed, 156 insertions(+)
  create mode 100644 include/wine/ringbuffer.h

diff --git a/include/wine/ringbuffer.h b/include/wine/ringbuffer.h
new file mode 100644
index 00000000000..50c4651082d
--- /dev/null
+++ b/include/wine/ringbuffer.h
@@ -0,0 +1,156 @@
+/*
+ * Single-producer single-consumer ringbuffer implementation
+ *
+ * Copyright (C) 2022 Davide Beatrici
+ *
+ * this library 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.
+ *
+ * this library 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 this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 
02110-1301, USA
+ */
+
+#ifndef __WINE_SERVER_RINGBUFFER_H
+#define __WINE_SERVER_RINGBUFFER_H
+
+#define WIN32_NO_STATUS
+#include "windef.h"
+
+#include <stdlib.h>
+#include <string.h>
+
+struct ringbuffer
+{
+    BYTE *buf;
+    UINT32 head;
+    UINT32 tail;
+    UINT32 size;
+    volatile UINT32 pending;
+};
+
+static inline struct ringbuffer *ringbuffer_new(const UINT32 size)
+{
+    struct ringbuffer *ringbuffer = malloc(sizeof(*ringbuffer));
+    if (!ringbuffer)
+        return NULL;
+
+    memset(ringbuffer, 0, sizeof(*ringbuffer));
+
+    ringbuffer->buf = malloc(size);
+    if (!ringbuffer->buf) {
+        free(ringbuffer);
+        return NULL;
+    }
+
+    ringbuffer->size = size;
+
+    return ringbuffer;
+}
+
+static inline void ringbuffer_free(struct ringbuffer *ringbuffer)
+{
+    free(ringbuffer->buf);
+    free(ringbuffer);
+}
+
+static inline void ringbuffer_reset(struct ringbuffer *ringbuffer)
+{
+    ringbuffer->head = 0;
+    ringbuffer->tail = 0;
+    ringbuffer->pending = 0;
+}
+
+static inline UINT32 ringbuffer_size(const struct ringbuffer 
*ringbuffer)
+{
+    return ringbuffer->size;
+}
+
+static inline UINT32 ringbuffer_readable(const struct ringbuffer 
*ringbuffer)
+{
+    return ringbuffer->pending;
+}
+
+static inline UINT32 ringbuffer_writable(const struct ringbuffer 
*ringbuffer)
+{
+    return ringbuffer->size - ringbuffer->pending;
+}
+
+static inline UINT32 ringbuffer_read(struct ringbuffer *ringbuffer, 
void *dst, UINT32 size)
+{
+    BYTE *buf = dst;
+    const UINT32 avail = ringbuffer_readable(ringbuffer);
+
+    if (size > avail)
+        size = avail;
+
+    if (!size)
+        return 0;
+
+    if (ringbuffer->head + size <= ringbuffer->size) {
+        memcpy(buf, &ringbuffer->buf[ringbuffer->head], size);
+
+        ringbuffer->head += size;
+    } else {
+        UINT32 size_part = ringbuffer->size - ringbuffer->head;
+
+        memcpy(buf, &ringbuffer->buf[ringbuffer->head], size_part);
+
+        buf += size_part;
+        size_part = size - size_part;
+        memcpy(buf, ringbuffer->buf, size_part);
+
+        ringbuffer->head = size_part;
+    }
+
+    if (ringbuffer->head == ringbuffer->size)
+        ringbuffer->head = 0;
+
+    InterlockedExchangeAdd((volatile LONG *)&ringbuffer->pending, 
-size);
+
+    return size;
+}
+
+static inline UINT32 ringbuffer_write(struct ringbuffer *ringbuffer, 
const void *src, UINT32 size)
+{
+    const BYTE *buf = src;
+    const UINT32 avail = ringbuffer_writable(ringbuffer);
+
+    if (size > avail)
+        size = avail;
+
+    if (!size)
+        return size;
+
+    if (ringbuffer->tail + size <= ringbuffer->size) {
+        memcpy(&ringbuffer->buf[ringbuffer->tail], buf, size);
+
+        ringbuffer->tail += size;
+    } else {
+        UINT32 size_part = ringbuffer->size - ringbuffer->tail;
+
+        memcpy(&ringbuffer->buf[ringbuffer->tail], buf, size_part);
+
+        buf += size_part;
+        size_part = size - size_part;
+        memcpy(ringbuffer->buf, buf, size_part);
+
+        ringbuffer->tail = size_part;
+    }
+
+    if (ringbuffer->tail == ringbuffer->size)
+        ringbuffer->tail = 0;
+
+    InterlockedExchangeAdd((volatile LONG *)&ringbuffer->pending, 
size);
+
+    return size;
+}
+
+#endif  /* __WINE_SERVER_RINGBUFFER_H */

[Stefan Dösinger]

Hi,

Am Dienstag, 15. Februar 2022, 09:45:11 EAT schrieb Davide Beatrici:
> The ringbuffer was written for the PipeWire driver I'm working on.
>
> I'm committing it separately because its implementation is generic,
> allowing it to be used pretty much universally.

Having written the wined3d command stream ringbuffer (and now banging my head 
against making it faster and multiple-producer single-consumer I think there 
are a few improvements here.
> +struct ringbuffer
> +{
> - BYTE *buf;
> - UINT32 head;
> - UINT32 tail;
> - UINT32 size;
> - volatile UINT32 pending;
>
> +};

You can avoid the separate allocation by making buf a flexible array at the end 
of the structure:

struct ringbuffer
{
    UINT32 head, tail, size;
    BYTE buf[1];
};

And then malloc this with sizeof(ringbuffer) - sizeof(ringbuffer->buf) + 
requested size). Or even better, offsetof: malloc(offsetof(struct ringbuffer, 
buf[size]))

The pending is redundant. You can extrapolate the amount of bytes in the buffer 
from head and tail, although you have to be careful re wraparound. There might 
be an advantage because only one field is InterlockedExchange'd (pending) 
instead of two (head, tail), but I'd need some benchmarking to be convinced 
that it is worth it. I think by having a redundant pending member you'll never 
be able to make this multiple-writer capable.
> +static inline UINT32 ringbuffer_read(struct ringbuffer *ringbuffer,
> void *dst, UINT32 size)
> ...
> +static inline UINT32 ringbuffer_write(struct ringbuffer *ringbuffer,
> const void *src, UINT32 size)

The API of passing a destination / source pointer and memcpy'ing is 
unfortunate. For any user that writes a large amount of data the memcpy will 
ruin performance.

Consider a function that returns a pointer into the buffer data where the 
caller can write plus a submission function that can be called once the data 
is ready to be consumed. Likewise, the _read function could return a pointer 
to the data in the buffer rather than copy it.

The behavior of no-write or partial writes if the buffer is full is troublesome 
too. If the caller has packets that only make sense as a whole it'd force the 
caller to add logic for short writes/reads. It would be good to have a 
blocking version of write and read that waits until space/data is available.