Re: [RFC PATCH 03/35] libceph: Add a new data container type, ceph_databuf

Viacheslav Dubeyko
2025-03-14T20:06:20Z

On Thu, 2025-03-13 at 23:32 +0000, David Howells wrote:
> Add a new ceph data container type, ceph_databuf, that can carry a list of
> pages in a bvec and use an iov_iter to handle describe the data to the next
> layer down. The iterator can also be used to refer to other types, such as
> ITER_FOLIOQ.
>
> There are two ways of loading the bvec. One way is to allocate a buffer
> with space in it and then add data, expanding the space as needed; the
> other is to splice in pages, expanding the bvec[] as needed.
>
> This is intended to replace all other types.
>

We definitely need to think about unit-tests or self-tests here.

> Signed-off-by: David Howells <dhowells@redhat.com>
> cc: Viacheslav Dubeyko <slava@dubeyko.com>
> cc: Alex Markuze <amarkuze@redhat.com>
> cc: Ilya Dryomov <idryomov@gmail.com>
> cc: ceph-devel@vger.kernel.org
> cc: linux-fsdevel@vger.kernel.org
> ---
> include/linux/ceph/databuf.h | 131 +++++++++++++++++++++
> include/linux/ceph/messenger.h | 6 +-
> include/linux/ceph/osd_client.h | 3 +
> net/ceph/Makefile | 3 +-
> net/ceph/databuf.c | 200 ++++++++++++++++++++++++++++++++
> net/ceph/messenger.c | 20 +++-
> net/ceph/osd_client.c | 11 +-
> 7 files changed, 369 insertions(+), 5 deletions(-)
> create mode 100644 include/linux/ceph/databuf.h
> create mode 100644 net/ceph/databuf.c
>
> diff --git a/include/linux/ceph/databuf.h b/include/linux/ceph/databuf.h
> new file mode 100644
> index 000000000000..14c7a6449467
> --- /dev/null
> +++ b/include/linux/ceph/databuf.h
> @@ -0,0 +1,131 @@
> +/* SPDX-License-Identifier: GPL-2.0 */
> +#ifndef __FS_CEPH_DATABUF_H
> +#define __FS_CEPH_DATABUF_H
> +
> +#include <asm/byteorder.h>
> +#include <linux/refcount.h>
> +#include <linux/blk_types.h>
> +
> +struct ceph_databuf {
> + struct bio_vec *bvec; /* List of pages */

So, maybe we need to think about folios now?

> + struct bio_vec inline_bvec[1]; /* Inline bvecs for small buffers */
> + struct iov_iter iter; /* Iterator defining occupied data */
> + size_t limit; /* Maximum length before expansion required */
> + size_t nr_bvec; /* Number of bvec[] that have pages */

Folios? :)

> + size_t max_bvec; /* Size of bvec[] */
> + refcount_t refcnt;
> + bool put_pages; /* T if pages in bvec[] need to be put*/

Maybe folios? :)

> +};
> +
> +struct ceph_databuf *ceph_databuf_alloc(size_t min_bvec, size_t space,
> + unsigned int data_source, gfp_t gfp);
> +struct ceph_databuf *ceph_databuf_get(struct ceph_databuf *dbuf);
> +void ceph_databuf_release(struct ceph_databuf *dbuf);
> +int ceph_databuf_append(struct ceph_databuf *dbuf, const void *d, size_t l);

I think that declaration is important and argument names needs to be clear
enough. Short name is good but it could be confusing. Why not len instead of l?
And I am still guessing what d means. :)

> +int ceph_databuf_reserve(struct ceph_databuf *dbuf, size_t space, gfp_t gfp);
> +int ceph_databuf_insert_frag(struct ceph_databuf *dbuf, unsigned int ix,
> + size_t len, gfp_t gfp);
> +
> +static inline
> +struct ceph_databuf *ceph_databuf_req_alloc(size_t min_bvec, size_t space, gfp_t gfp)
> +{
> + return ceph_databuf_alloc(min_bvec, space, ITER_SOURCE, gfp);
> +}
> +
> +static inline
> +struct ceph_databuf *ceph_databuf_reply_alloc(size_t min_bvec, size_t space, gfp_t gfp)
> +{
> + struct ceph_databuf *dbuf;
> +
> + dbuf = ceph_databuf_alloc(min_bvec, space, ITER_DEST, gfp);
> + if (dbuf)
> + iov_iter_reexpand(&dbuf->iter, space);
> + return dbuf;
> +}
> +
> +static inline struct page *ceph_databuf_page(struct ceph_databuf *dbuf,
> + unsigned int ix)
> +{
> + return dbuf->bvec[ix].bv_page;
> +}
> +
> +#define kmap_ceph_databuf_page(dbuf, ix) \
> + kmap_local_page(ceph_databuf_page(dbuf, ix));
> +

I am still thinking that we need to base the new code on folio.

> +static inline int ceph_databuf_encode_64(struct ceph_databuf *dbuf, u64 v)
> +{
> + __le64 ev = cpu_to_le64(v);
> + return ceph_databuf_append(dbuf, &ev, sizeof(ev));
> +}
> +static inline int ceph_databuf_encode_32(struct ceph_databuf *dbuf, u32 v)
> +{
> + __le32 ev = cpu_to_le32(v);
> + return ceph_databuf_append(dbuf, &ev, sizeof(ev));
> +}
> +static inline int ceph_databuf_encode_16(struct ceph_databuf *dbuf, u16 v)
> +{
> + __le16 ev = cpu_to_le16(v);
> + return ceph_databuf_append(dbuf, &ev, sizeof(ev));
> +}
> +static inline int ceph_databuf_encode_8(struct ceph_databuf *dbuf, u8 v)
> +{
> + return ceph_databuf_append(dbuf, &v, 1);
> +}

Maybe, encode_8, encode_16, encode_32, encode_64? I mean reverse sequence here.

> +static inline int ceph_databuf_encode_string(struct ceph_databuf *dbuf,
> + const char *s, u32 len)
> +{
> + int ret = ceph_databuf_encode_32(dbuf, len);
> + if (ret)
> + return ret;
> + if (len)
> + return ceph_databuf_append(dbuf, s, len);
> + return 0;
> +}
> +
> +static inline size_t ceph_databuf_len(struct ceph_databuf *dbuf)
> +{
> + return dbuf->iter.count;
> +}
> +
> +static inline void ceph_databuf_added_data(struct ceph_databuf *dbuf,
> + size_t len)
> +{
> + dbuf->iter.count += len;
> +}
> +
> +static inline void ceph_databuf_reply_ready(struct ceph_databuf *reply,
> + size_t len)
> +{
> + reply->iter.data_source = ITER_SOURCE;
> + iov_iter_truncate(&reply->iter, len);
> +}
> +
> +static inline void ceph_databuf_reset_reply(struct ceph_databuf *reply)
> +{
> + iov_iter_bvec(&reply->iter, ITER_DEST,
> + reply->bvec, reply->nr_bvec, reply->limit);
> +}
> +
> +static inline void ceph_databuf_append_page(struct ceph_databuf *dbuf,
> + struct page *page,
> + unsigned int offset,
> + unsigned int len)
> +{
> + BUG_ON(dbuf->nr_bvec >= dbuf->max_bvec);
> + bvec_set_page(&dbuf->bvec[dbuf->nr_bvec++], page, len, offset);
> + dbuf->iter.count += len;
> + dbuf->iter.nr_segs++;

Why do we assign len to dbuf->iter.count but only increment dbuf->iter.nr_segs?

> +}
> +
> +static inline void *ceph_databuf_enc_start(struct ceph_databuf *dbuf)
> +{
> + return page_address(ceph_databuf_page(dbuf, 0)) + dbuf->iter.count;
> +}
> +
> +static inline void ceph_databuf_enc_stop(struct ceph_databuf *dbuf, void *p)
> +{
> + dbuf->iter.count = p - page_address(ceph_databuf_page(dbuf, 0));
> + BUG_ON(dbuf->iter.count > dbuf->limit);
> +}

The same about page...

> +
> +#endif /* __FS_CEPH_DATABUF_H */
> diff --git a/include/linux/ceph/messenger.h b/include/linux/ceph/messenger.h
> index db2aba32b8a0..864aad369c91 100644
> --- a/include/linux/ceph/messenger.h
> +++ b/include/linux/ceph/messenger.h
> @@ -117,6 +117,7 @@ struct ceph_messenger {
>
> enum ceph_msg_data_type {
> CEPH_MSG_DATA_NONE, /* message contains no data payload */
> + CEPH_MSG_DATA_DATABUF, /* data source/destination is a data buffer */
> CEPH_MSG_DATA_PAGES, /* data source/destination is a page array */
> CEPH_MSG_DATA_PAGELIST, /* data source/destination is a pagelist */

So, the final replacement on databuf will be in the future?

> #ifdef CONFIG_BLOCK
> @@ -210,7 +211,10 @@ struct ceph_bvec_iter {
>
> struct ceph_msg_data {
> enum ceph_msg_data_type type;
> + struct iov_iter iter;
> + bool release_dbuf;
> union {
> + struct ceph_databuf *dbuf;
> #ifdef CONFIG_BLOCK
> struct {
> struct ceph_bio_iter bio_pos;
> @@ -225,7 +229,6 @@ struct ceph_msg_data {
> bool own_pages;
> };
> struct ceph_pagelist *pagelist;
> - struct iov_iter iter;
> };
> };
>
> @@ -601,6 +604,7 @@ extern void ceph_con_keepalive(struct ceph_connection *con);
> extern bool ceph_con_keepalive_expired(struct ceph_connection *con,
> unsigned long interval);
>
> +void ceph_msg_data_add_databuf(struct ceph_msg *msg, struct ceph_databuf *dbuf);
> void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
> size_t length, size_t offset, bool own_pages);
> extern void ceph_msg_data_add_pagelist(struct ceph_msg *msg,
> diff --git a/include/linux/ceph/osd_client.h b/include/linux/ceph/osd_client.h
> index 8fc84f389aad..b8fb5a71dd57 100644
> --- a/include/linux/ceph/osd_client.h
> +++ b/include/linux/ceph/osd_client.h
> @@ -16,6 +16,7 @@
> #include <linux/ceph/msgpool.h>
> #include <linux/ceph/auth.h>
> #include <linux/ceph/pagelist.h>
> +#include <linux/ceph/databuf.h>
>
> struct ceph_msg;
> struct ceph_snap_context;
> @@ -103,6 +104,7 @@ struct ceph_osd {
>
> enum ceph_osd_data_type {
> CEPH_OSD_DATA_TYPE_NONE = 0,
> + CEPH_OSD_DATA_TYPE_DATABUF,
> CEPH_OSD_DATA_TYPE_PAGES,
> CEPH_OSD_DATA_TYPE_PAGELIST,

The same question about replacement on databuf here? Is it future work?

> #ifdef CONFIG_BLOCK
> @@ -115,6 +117,7 @@ enum ceph_osd_data_type {
> struct ceph_osd_data {
> enum ceph_osd_data_type type;
> union {
> + struct ceph_databuf *dbuf;
> struct {
> struct page **pages;
> u64 length;
> diff --git a/net/ceph/Makefile b/net/ceph/Makefile
> index 8802a0c0155d..4b2e0b654e45 100644
> --- a/net/ceph/Makefile
> +++ b/net/ceph/Makefile
> @@ -15,4 +15,5 @@ libceph-y := ceph_common.o messenger.o msgpool.o buffer.o pagelist.o \
> auth_x.o \
> ceph_strings.o ceph_hash.o \
> pagevec.o snapshot.o string_table.o \
> - messenger_v1.o messenger_v2.o
> + messenger_v1.o messenger_v2.o \
> + databuf.o
> diff --git a/net/ceph/databuf.c b/net/ceph/databuf.c
> new file mode 100644
> index 000000000000..9d108fff5a4f
> --- /dev/null
> +++ b/net/ceph/databuf.c
> @@ -0,0 +1,200 @@
> +// SPDX-License-Identifier: GPL-2.0
> +/* Data container
> + *
> + * Copyright (C) 2023 Red Hat, Inc. All Rights Reserved.
> + * Written by David Howells (dhowells@redhat.com)
> + */
> +
> +#include <linux/export.h>
> +#include <linux/gfp.h>
> +#include <linux/slab.h>
> +#include <linux/uio.h>
> +#include <linux/pagemap.h>
> +#include <linux/highmem.h>
> +#include <linux/ceph/databuf.h>
> +
> +struct ceph_databuf *ceph_databuf_alloc(size_t min_bvec, size_t space,
> + unsigned int data_source, gfp_t gfp)
> +{
> + struct ceph_databuf *dbuf;
> + size_t inl = ARRAY_SIZE(dbuf->inline_bvec);
> +
> + dbuf = kzalloc(sizeof(*dbuf), gfp);
> + if (!dbuf)
> + return NULL;

I am guessing... Should we return error code here?

> +
> + refcount_set(&dbuf->refcnt, 1);
> +
> + if (min_bvec == 0 && space == 0) {
> + /* Do nothing */
> + } else if (min_bvec <= inl && space <= inl * PAGE_SIZE) {
> + dbuf->bvec = dbuf->inline_bvec;
> + dbuf->max_bvec = inl;
> + dbuf->limit = space;
> + } else if (min_bvec) {
> + min_bvec = umax(min_bvec, 16);

Why 16 here? Maybe, do we need to introduce some well explained constant?

> +
> + dbuf->bvec = kcalloc(min_bvec, sizeof(struct bio_vec), gfp);
> + if (!dbuf->bvec) {
> + kfree(dbuf);
> + return NULL;

Ditto. Should we return error code here?

> + }
> +
> + dbuf->max_bvec = min_bvec;

Why do we assign min_bvec to max_bvec? I am simply slightly confused why
argument of function is named as min_bvec, but finally we are saving min_bvec
value into max_bvec.

> + }
> +
> + iov_iter_bvec(&dbuf->iter, data_source, dbuf->bvec, 0, 0);
> +
> + if (space) {
> + if (ceph_databuf_reserve(dbuf, space, gfp) < 0) {
> + ceph_databuf_release(dbuf);
> + return NULL;

Ditto. Should we return error code here?

> + }
> + }
> + return dbuf;
> +}
> +EXPORT_SYMBOL(ceph_databuf_alloc);
> +
> +struct ceph_databuf *ceph_databuf_get(struct ceph_databuf *dbuf)

I see the point here. But do we really need to return pointer? Why not simply:

void ceph_databuf_get(struct ceph_databuf *dbuf)

> +{
> + if (!dbuf)
> + return NULL;
> + refcount_inc(&dbuf->refcnt);
> + return dbuf;
> +}
> +EXPORT_SYMBOL(ceph_databuf_get);
> +
> +void ceph_databuf_release(struct ceph_databuf *dbuf)
> +{
> + size_t i;
> +
> + if (!dbuf || !refcount_dec_and_test(&dbuf->refcnt))
> + return;
> +
> + if (dbuf->put_pages)
> + for (i = 0; i < dbuf->nr_bvec; i++)
> + put_page(dbuf->bvec[i].bv_page);
> + if (dbuf->bvec != dbuf->inline_bvec)
> + kfree(dbuf->bvec);
> + kfree(dbuf);
> +}
> +EXPORT_SYMBOL(ceph_databuf_release);
> +
> +/*
> + * Expand the bvec[] in the dbuf.
> + */
> +static int ceph_databuf_expand(struct ceph_databuf *dbuf, size_t req_bvec,
> + gfp_t gfp)
> +{
> + struct bio_vec *bvec = dbuf->bvec, *old = bvec;

I think that assigning (*old = bvec) looks confusing if we keep it on the same
line as bvec declaration and initialization. Why do not declare and not
initialize it on the next line?

> + size_t size, max_bvec, off = dbuf->iter.bvec - old;

I think it's too much declarations on the same line. Why not:

size_t size, max_bvec;
size_t off = dbuf->iter.bvec - old;

> + size_t inl = ARRAY_SIZE(dbuf->inline_bvec);
> +
> + if (req_bvec <= inl) {
> + dbuf->bvec = dbuf->inline_bvec;
> + dbuf->max_bvec = inl;
> + dbuf->iter.bvec = dbuf->inline_bvec + off;
> + return 0;
> + }
> +
> + max_bvec = roundup_pow_of_two(req_bvec);
> + size = array_size(max_bvec, sizeof(struct bio_vec));
> +
> + if (old == dbuf->inline_bvec) {
> + bvec = kmalloc_array(max_bvec, sizeof(struct bio_vec), gfp);
> + if (!bvec)
> + return -ENOMEM;
> + memcpy(bvec, old, inl);
> + } else {
> + bvec = krealloc(old, size, gfp);
> + if (!bvec)
> + return -ENOMEM;
> + }
> + dbuf->bvec = bvec;
> + dbuf->max_bvec = max_bvec;
> + dbuf->iter.bvec = bvec + off;
> + return 0;
> +}
> +
> +/* Allocate enough pages for a dbuf to append the given amount
> + * of dbuf without allocating.
> + * Returns: 0 on success, -ENOMEM on error.
> + */
> +int ceph_databuf_reserve(struct ceph_databuf *dbuf, size_t add_space,
> + gfp_t gfp)
> +{
> + struct bio_vec *bvec;
> + size_t i, req_bvec = DIV_ROUND_UP(dbuf->iter.count + add_space, PAGE_SIZE);

Why not:

size_t req_bvec = DIV_ROUND_UP(dbuf->iter.count + add_space, PAGE_SIZE);
size_t i;

> + int ret;
> +
> + dbuf->put_pages = true;
> + if (req_bvec > dbuf->max_bvec) {
> + ret = ceph_databuf_expand(dbuf, req_bvec, gfp);
> + if (ret < 0)
> + return ret;
> + }
> +
> + bvec = dbuf->bvec;
> + while (dbuf->nr_bvec < req_bvec) {
> + struct page *pages[16];

Why do we hardcoded 16 here but using some well defined constant?

And, again, why not folio?

> + size_t want = min(req_bvec, ARRAY_SIZE(pages)), got;
> +
> + memset(pages, 0, sizeof(pages));
> + got = alloc_pages_bulk(gfp, want, pages);
> + if (!got)
> + return -ENOMEM;
> + for (i = 0; i < got; i++)

Why do we use size_t for i and got? Why not int, for example?

> + bvec_set_page(&bvec[dbuf->nr_bvec + i], pages[i],
> + PAGE_SIZE, 0);
> + dbuf->iter.nr_segs += got;
> + dbuf->nr_bvec += got;

If I understood correctly, the ceph_databuf_append_page() uses slightly
different logic.

+ dbuf->iter.count += len;
+ dbuf->iter.nr_segs++;

But here we assign number of allocated pages to nr_segs. It is slightly
confusing. I think I am missing something here.

> + dbuf->limit = dbuf->nr_bvec * PAGE_SIZE;
> + }
> +
> + return 0;
> +}
> +EXPORT_SYMBOL(ceph_databuf_reserve);
> +
> +int ceph_databuf_append(struct ceph_databuf *dbuf, const void *buf, size_t len)
> +{
> + struct iov_iter temp_iter;
> +
> + if (!len)
> + return 0;
> + if (dbuf->limit - dbuf->iter.count > len &&
> + ceph_databuf_reserve(dbuf, len, GFP_NOIO) < 0)
> + return -ENOMEM;
> +
> + iov_iter_bvec(&temp_iter, ITER_DEST,
> + dbuf->bvec, dbuf->nr_bvec, dbuf->limit);
> + iov_iter_advance(&temp_iter, dbuf->iter.count);
> +
> + if (copy_to_iter(buf, len, &temp_iter) != len)
> + return -EFAULT;
> + dbuf->iter.count += len;
> + return 0;
> +}
> +EXPORT_SYMBOL(ceph_databuf_append);
> +
> +/*
> + * Allocate a fragment and insert it into the buffer at the specified index.
> + */
> +int ceph_databuf_insert_frag(struct ceph_databuf *dbuf, unsigned int ix,
> + size_t len, gfp_t gfp)
> +{
> + struct page *page;
> +

Why not folio?

> + page = alloc_page(gfp);
> + if (!page)
> + return -ENOMEM;
> +
> + bvec_set_page(&dbuf->bvec[ix], page, len, 0);
> +
> + if (dbuf->nr_bvec == ix) {
> + dbuf->iter.nr_segs = ix + 1;
> + dbuf->nr_bvec = ix + 1;
> + dbuf->iter.count += len;
> + }
> + return 0;
> +}
> +EXPORT_SYMBOL(ceph_databuf_insert_frag);
> diff --git a/net/ceph/messenger.c b/net/ceph/messenger.c
> index 1df4291cc80b..802f0b222131 100644
> --- a/net/ceph/messenger.c
> +++ b/net/ceph/messenger.c
> @@ -1872,7 +1872,9 @@ static struct ceph_msg_data *ceph_msg_data_add(struct ceph_msg *msg)
>
> static void ceph_msg_data_destroy(struct ceph_msg_data *data)
> {
> - if (data->type == CEPH_MSG_DATA_PAGES && data->own_pages) {
> + if (data->type == CEPH_MSG_DATA_DATABUF) {
> + ceph_databuf_release(data->dbuf);
> + } else if (data->type == CEPH_MSG_DATA_PAGES && data->own_pages) {
> int num_pages = calc_pages_for(data->offset, data->length);
> ceph_release_page_vector(data->pages, num_pages);
> } else if (data->type == CEPH_MSG_DATA_PAGELIST) {
> @@ -1880,6 +1882,22 @@ static void ceph_msg_data_destroy(struct ceph_msg_data *data)
> }
> }
>
> +void ceph_msg_data_add_databuf(struct ceph_msg *msg, struct ceph_databuf *dbuf)
> +{
> + struct ceph_msg_data *data;
> +
> + BUG_ON(!dbuf);
> + BUG_ON(!ceph_databuf_len(dbuf));
> +
> + data = ceph_msg_data_add(msg);
> + data->type = CEPH_MSG_DATA_DATABUF;
> + data->dbuf = ceph_databuf_get(dbuf);
> + data->iter = dbuf->iter;
> +
> + msg->data_length += ceph_databuf_len(dbuf);
> +}
> +EXPORT_SYMBOL(ceph_msg_data_add_databuf);
> +
> void ceph_msg_data_add_pages(struct ceph_msg *msg, struct page **pages,
> size_t length, size_t offset, bool own_pages)
> {
> diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
> index e359e70ad47e..c84634264377 100644
> --- a/net/ceph/osd_client.c
> +++ b/net/ceph/osd_client.c
> @@ -359,6 +359,8 @@ static u64 ceph_osd_data_length(struct ceph_osd_data *osd_data)
> switch (osd_data->type) {
> case CEPH_OSD_DATA_TYPE_NONE:
> return 0;
> + case CEPH_OSD_DATA_TYPE_DATABUF:
> + return ceph_databuf_len(osd_data->dbuf);
> case CEPH_OSD_DATA_TYPE_PAGES:
> return osd_data->length;
> case CEPH_OSD_DATA_TYPE_PAGELIST:
> @@ -379,7 +381,9 @@ static u64 ceph_osd_data_length(struct ceph_osd_data *osd_data)
>
> static void ceph_osd_data_release(struct ceph_osd_data *osd_data)
> {
> - if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES && osd_data->own_pages) {
> + if (osd_data->type == CEPH_OSD_DATA_TYPE_DATABUF) {
> + ceph_databuf_release(osd_data->dbuf);
> + } else if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES && osd_data->own_pages) {
> int num_pages;
>
> num_pages = calc_pages_for((u64)osd_data->offset,
> @@ -965,7 +969,10 @@ static void ceph_osdc_msg_data_add(struct ceph_msg *msg,
> {
> u64 length = ceph_osd_data_length(osd_data);
>
> - if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
> + if (osd_data->type == CEPH_OSD_DATA_TYPE_DATABUF) {
> + BUG_ON(!length);
> + ceph_msg_data_add_databuf(msg, osd_data->dbuf);
> + } else if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) {
> BUG_ON(length > (u64) SIZE_MAX);
> if (length)
> ceph_msg_data_add_pages(msg, osd_data->pages,
>
>

Thanks,
Slava.