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.

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

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

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

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?

Thanks,
Slava.