Re: [PATCH 1/4] dibs: introduce ring-buffer send/receive API

[Niklas Schnelle <schnelle@linux.ibm.com>]

On Thu, 2026-10-08 at 10:08 +0200, Julian Ruess wrote:
> From: Tobias Schumacher <ts@linux.ibm.com>
> 
> Introduce a centralized API for sending and receiving data between dibs
> peers via ring buffers. With this API, drivers no longer need to
> interact directly with DMBs and manage read/write pointers.
> 
> A driver must first register its ring buffer using
> dibs_register_ring_buffer() and set the rdmb_tok via
> dibs_ring_set_rdmb_tok(). After that, data can be sent using
> dibs_ring_send().
> 
> Receiving messages is split into two steps: dibs_ring_recv() returns a
> pointer to the next DIBS message if one is available. After the message
> is processed, dibs_ring_ack() must be called to free up the message
> buffers and to advance the read pointer. If no message is available,
> dibs_ring_recv() returns NULL.

I think this commit message could provide a few more details on the
functioning of the ring, it's data layout and parts (e.g.
rd_dmb/wr_dmb/meta_dmb), constraints (like power of 2) and how the data
transfers work with respect to the head/tail pointers (i.e. refer to
the CIRC_… macros) and where what part resides. Quite a bit of this is
in the comments in the header but reading this patch from the start I
only noticed late and would really have liked more of it in the commit
message. Also I think there should be a section on how
recovery/reconnect works.

I'm still torn about whether it would make sense to put this in
Documentation/dibs/ where its existence is also easier to pick out from
the change stats. I believe in earlier internal discussions I was
favoring comments and commit messages but since this has grown in
complexity and I think a documentation file would be nice.

> 
> Co-developed-by: Julian Ruess <julianr@linux.ibm.com>
> Signed-off-by: Julian Ruess <julianr@linux.ibm.com>
> Signed-off-by: Tobias Schumacher <ts@linux.ibm.com>
> ---
>  drivers/dibs/dibs_main.c | 493 ++++++++++++++++++++++++++++++++++++++++++++++-
>  include/linux/dibs.h     |  99 ++++++++++
>  2 files changed, 590 insertions(+), 2 deletions(-)
> 
> diff --git a/drivers/dibs/dibs_main.c b/drivers/dibs/dibs_main.c
> index 20c50997a7cf..e3a33bc42fe8 100644
> --- a/drivers/dibs/dibs_main.c
> +++ b/drivers/dibs/dibs_main.c
> @@ -13,6 +13,9 @@
>  #include <linux/slab.h>
>  #include <linux/err.h>
>  #include <linux/dibs.h>
> +#include <linux/circ_buf.h>
> +#include <linux/log2.h>
> +#include <asm/barrier.h>
>  
>  #include "dibs_loopback.h"
>  
> @@ -20,7 +23,7 @@ MODULE_DESCRIPTION("Direct Internal Buffer Sharing class");
>  MODULE_LICENSE("GPL");
>  
>  static const struct class dibs_class = {
> -	.name		= "dibs",
> +	.name = "dibs",
>  };
>  
>  /* use an array rather a list for fast mapping: */
> @@ -93,7 +96,8 @@ int dibs_unregister_client(struct dibs_client *client)
>  		max_dmbs = dibs->ops->max_dmbs();
>  		for (int i = 0; i < max_dmbs; ++i) {
>  			if (dibs->dmb_clientid_arr[i] == client->id) {
> -				WARN(1, "%s: attempt to unregister '%s' with registered dmb(s)\n",
> +				WARN(1,
> +				     "%s: attempt to unregister '%s' with registered dmb(s)\n",

Nit: This line as well as the change in dibs_class look like overly
eager auto format instead of an actual change. This is one of the
reasons I personally only do auto format over selected lines in my
editor and no auto format on save. Also I'd keep the above literal in
one line with the WARN().

>  				     __func__, client->name);
>  				rc = -EBUSY;
>  				goto err_reg_dmb;
> @@ -245,6 +249,491 @@ void dibs_dev_del(struct dibs_dev *dibs)
>  }
>  EXPORT_SYMBOL_GPL(dibs_dev_del);
>  
> +/**
> + * dibs_ring_peer_meta() - take a snapshot of the peer's ring metadata
> + * @rb: pointer to the ring buffer
> + * @peer: snapshot to fill in
> + *
> + * The peer writes its metadata dmb at any time, so read each field once and
> + * copy to the snapshot.
> + */
> +static void dibs_ring_peer_meta(struct dibs_ring_buffer *rb,
> +				struct dibs_ring_meta *peer)
> +{
> +	peer->meta_dmb_tok = READ_ONCE(rb->rd_ring_meta->meta_dmb_tok);
> +	peer->buff_dmb_tok = READ_ONCE(rb->rd_ring_meta->buff_dmb_tok);
> +	peer->size = READ_ONCE(rb->rd_ring_meta->size);
> +	peer->head = READ_ONCE(rb->rd_ring_meta->head);
> +	peer->tail = READ_ONCE(rb->rd_ring_meta->tail);
> +	peer->reserved = 0;
> +}
> +
> +static int dibs_ring_send_hdr(struct dibs_ring_buffer *rb, u64 meta_dmb_tok)
> +{
> +	int res;
> +
> +	if (!meta_dmb_tok)
> +		return 0;
> +
> +	res = rb->dibs->ops->move_data(rb->dibs, meta_dmb_tok, 0, true, 0,
> +				       &rb->wr_ring_meta,
> +				       sizeof(struct dibs_ring_meta));
> +

Nit: Stray empty line

> +	if (!res)
> +		rb->hdr_sent = true;
> +
> +	return res;
> +}
> +
> +/**
> + * dibs_ring_register() - register a DIBS ring buffer
> + * @name: name of the ring buffer. This will be used in debug output to
> + *        support identifying the ring buffer.
> + * @rb: pointer to the ring buffer struct. The fields in the struct don't have
> + *      to be initialized.
> + * @client: pointer to the DIBS client
> + * @dibs: pointer to the DIBS device
> + * @rgid: the remote GID that will be allowed to write into this ring buffer
> + * @size: size of the ring buffer in bytes
> + *
> + * Register and initialize a DIBS ring buffer. Allocates the metadata DMB and
> + * the payload DMB, initializes the fields in dibs_ring_buffer.
> + *
> + * Return: 0 in case of success, an error code in case of failure
> + */
> +int dibs_ring_register(char *name, struct dibs_ring_buffer *rb,
> +		       struct dibs_client *client, struct dibs_dev *dibs,
> +		       uuid_t rgid, u32 size)
> +{
> +	int ret;
> +
> +	if (!size || !is_power_of_2(size))
> +		return -EINVAL;
> +
> +	strscpy(rb->name, name, sizeof(rb->name));
> +	rb->dibs = dibs;
> +	rb->dmb.rgid = rgid;
> +
> +	rb->rd_meta_dmb.dmb_len = ALIGN(sizeof(struct dibs_ring_meta), 4096);
> +	rb->rd_meta_dmb.rgid = rgid;
> +	ret = dibs->ops->register_dmb(dibs, &rb->rd_meta_dmb, client);
> +	if (ret)
> +		return ret;
> +
> +	rb->dmb.dmb_len = ALIGN(size, 4096);

Nit: The two DMB registrations would be easier to read if the dmb.rgid
was set here just like it is done for the rd_meta_dmb.rgid above and
how the dmb.dmb_len is only set here.

> +	ret = dibs->ops->register_dmb(dibs, &rb->dmb, client);
> +	if (ret) {
> +		dibs->ops->unregister_dmb(rb->dibs, &rb->rd_meta_dmb);
> +		return ret;
> +	}
> +
> +	rb->dmb_registered = 1;
> +
> +	rb->wr_ring_meta.meta_dmb_tok = rb->rd_meta_dmb.dmb_tok;
> +	rb->wr_ring_meta.buff_dmb_tok = rb->dmb.dmb_tok;
> +	rb->wr_ring_meta.size = size;
> +	rb->wr_ring_meta.head = 0;
> +	rb->wr_ring_meta.tail = 0;
> +	rb->wr_ring_meta.reserved = 0;
> +
> +	rb->rd_ring_meta = rb->rd_meta_dmb.cpu_addr;
> +	rb->peer_buff_dmb_tok = 0;
> +	rb->msg_size = 0;
> +	rb->hdr_sent = false;
> +
> +	return ret;
> +}
> +EXPORT_SYMBOL_GPL(dibs_ring_register);
> +
> +/**
> + * dibs_ring_unregister() - unregister a DIBS ring buffer
> + * @rb: pointer to the ring buffer to unregister
> + *
> + * Free the resources allocated by dibs_ring_register().
> + *
> + * Return: 0 in case of success, an error code in case of failure
> + */
> +int dibs_ring_unregister(struct dibs_ring_buffer *rb)
> +{
> +	struct dibs_ring_meta peer;
> +	int ret;
> +
> +	rb->wr_ring_meta.meta_dmb_tok = 0;
> +	rb->wr_ring_meta.buff_dmb_tok = 0;
> +	rb->wr_ring_meta.size = 0;
> +	rb->wr_ring_meta.head = 0;
> +	rb->wr_ring_meta.tail = 0;
> +
> +	dibs_ring_peer_meta(rb, &peer);
> +	if (peer.meta_dmb_tok) {
> +		ret = dibs_ring_send_hdr(rb, peer.meta_dmb_tok);
> +		if (ret)
> +			pr_warn("%s: failed to send header update: %d\n",
> +				__func__, ret);
> +	}

I think the function doc should mention that this sends a final header
update and explain why this is and what it looks like,

> +
> +	ret = rb->dibs->ops->unregister_dmb(rb->dibs, &rb->dmb);
> +	if (ret)
> +		pr_warn("%s: failed to unregister payload dmb: %d\n", __func__,
> +			ret);
> +
> +	ret = rb->dibs->ops->unregister_dmb(rb->dibs, &rb->rd_meta_dmb);
> +	if (ret)
> +		pr_warn("%s: failed to unregister metadata dmb: %d\n", __func__,
> +			ret);
> +
> +	rb->dmb_registered = 0;
> +	return ret;
> +}
> +EXPORT_SYMBOL_GPL(dibs_ring_unregister);
> +
> +/**
> + * dibs_ring_peer_replaced() - check if the peer has replaced its ring buffer
> + * @rb: pointer to the ring buffer
> + * @peer: snapshot of the peer's metadata
> + *
> + * The fabric hands out a new dmb token for every registration, so a buffer
> + * token that differs from the one this side is synchronized to means the peer
> + * ring this side writes into is a different one.

This function doc should mention under what circumstances this may be
the case. I'm assuming this only happens during recovery after a peer
was restarted/reconnected?  For more details this should probably be
part of the documentation and then the comment can refer to that.

> + *
> + * Return: true if the peer ring has been replaced, false otherwise
> + */
> +static bool dibs_ring_peer_replaced(struct dibs_ring_buffer *rb,
> +				    const struct dibs_ring_meta *peer)
> +{
> +	return peer->buff_dmb_tok != rb->peer_buff_dmb_tok;
> +}
> +
> +/**
> + * dibs_ring_sync_peer() - synchronize with a replaced peer ring buffer
> + * @rb: pointer to the ring buffer
> + * @peer: snapshot of the peer's metadata
> + *
> + * Restart both pointers: head is this side's write position in the peer's
> + * buffer, which is a new and empty one, and tail is this side's read position
> + * in its own buffer, which the peer has started to write at offset 0 again.
> + * Sends the updated metadata header to the peer and only then remembers the
> + * new token, so that a failed send is retried on the next call.
> + *
> + * A peer that has torn its ring down is not synchronized to, it is only
> + * forgotten - this side cannot send anything until the peer registers again.

Same comment as above.

> + */
> +static void dibs_ring_sync_peer(struct dibs_ring_buffer *rb,
> +				const struct dibs_ring_meta *peer)
> +{
> +	if (!peer->buff_dmb_tok) {

Should the condition be a call to dibs_ring_peer_ready()?

> +		rb->wr_ring_meta.tail = 0;
> +		rb->peer_buff_dmb_tok = 0;
> +		rb->hdr_sent = false;
> +		return;
> +	}
> +
> +	rb->wr_ring_meta.head = 0;
> +	rb->wr_ring_meta.tail = 0;
> +
> +	if (dibs_ring_send_hdr(rb, peer->meta_dmb_tok)) {
> +		pr_warn("%s(%s): failed to send header after peer restart\n",
> +			__func__, rb->name);
> +		return;
> +	}
> +
> +	rb->peer_buff_dmb_tok = peer->buff_dmb_tok;
> +}
> +
--- snip ---
>  
> +/* Ring Buffer Communication
> + * -------------------------
> + * Based on the DMBs, DIBS provides a ring buffer mechanism for bi-directional
> + * communication between two peers. For each ring-buffer connection, two DMBs
> + * are allocated per peer:
> + * - one DMB holding metadata about the ring and
> + * - one DMB holding the actual data to be transferred.
> + *
> + * Metadata consists of parameters like the DMB tokens, size, head pointer and
> + * tail pointer. Note that the metadata stored in the locally allocated
> + * metadata DMB does not necessarily corelate to the locally allocated payload
> + * DMB. As illustrated below, each peer has access to one wr_ring_meta DMB
> + * (the one the other peer has allocated) and one rd_ring_meta DMB (the one
> + * it has allocated itself). This is because the locally allocated DMB can only
> + * be written by the other peer and vice versa.
> + *
> + * Peer A                   Peer B
> + * ---------------          ---------------
> + * wr_ring_meta ----------> rd_ring_meta
> + * rd_ring_meta <---------- wr_ring_meta
> + * send_buffer  ----------> recv_buffer
> + * recv_buffer  <---------- send_buffer
> + *
> + * Consider for example that peer A wants to send data to peer B. It first has
> + * to check how much data it previously has written (head pointer) and how far
> + * peer B has read (tail pointer) and compare this to the size. Since for send
> + * operations head is updated by peer A, it is located in the wr_ring_meta DMB.
> + * Tail is updated by the receiving side, peer B in this case, so peer A will
> + * find the current value in rd_ring_meta. The payload DMB was allocated by
> + * peer B, so peer B also updated the size - peer A therefore finds the correct
> + * value in rd_ring_meta.
> + *
> + * Now peer a can write data into the send_buffer and update the head pointer
> + * in wr_ring_meta. Peer B then reads this head pointer from its rd_ring_meta
> + * DMB, process the data and update tail in its wr_ring_meta.
> + */

I saw this way too late going through the patch from top to bottom.
Then remembered that I had read this in an internal version where I
think I also saw it late. I think Documentation changes always get
sorted to the top and stand out in the stats so that's an argument for
adding docs I guess ;)

> +struct dibs_ring_meta {
> +	/* meta_dmb_tok - Token for the metadata dmb. */
> +	u64 meta_dmb_tok;
> +	/* buff_dmb_tok - Token for the buffer dmb */
> +	u64 buff_dmb_tok;
> +	/* size - size of the buffer for receiving data in number of byte */
> +	u32 size;
> +	/* head - points to the head of the ring buffer, i.e., the element
> +	 * that will be written next
> +	 */
> +	u32 head;
> +	/* tail - points to the end of the ring-buffer, i.e., the element
> +	 * that will be read next
> +	 */
> +	u32 tail;
> +	/* reserved - keeps the struct at its natural size, must be zero */
> +	u32 reserved;
> +};
> +
--- snip ---
>  /* Functions to be called by dibs device drivers:
> @@ -461,4 +554,10 @@ int dibs_dev_add(struct dibs_dev *dibs);
>   */
>  void dibs_dev_del(struct dibs_dev *dibs);
>  
> +struct dibs_msg_hdr {
> +	u8 version;
> +	u8 type;
> +	u16 datalen;
> +} __packed;
> +
>  #endif	/* _DIBS_H */

I have so far only thoroughly read through a small part of the patch 
so regard this as a set of first comments. I feel like it is already
actionable though so wanted to reply quickly rather than wait. 

Overall I really like the abstraction and concept!

Thanks,
Niklas