dev_uring.c

// SPDX-License-Identifier: GPL-2.0
/*
 * FUSE: Filesystem in Userspace
 * Copyright (c) 2023-2024 DataDirect Networks.
 */

#include "dev.h"
#include "args.h"
#include "dev_uring_i.h"
#include "fuse_trace.h"

#include <linux/bitmap.h>
#include <linux/fs.h>
#include <linux/io_uring/cmd.h>

static bool __read_mostly enable_uring;
module_param(enable_uring, bool, 0644);
MODULE_PARM_DESC(enable_uring,
		 "Enable userspace communication through io-uring");

#define FUSE_URING_IOV_SEGS 2 /* header and payload */
#define FUSE_URING_IOV_HEADERS 0
#define FUSE_URING_IOV_PAYLOAD 1

#define FUSE_URING_ADD_QUEUE_FLAGS	(FUSE_URING_ZERO_COPY)

bool fuse_uring_enabled(void)
{
	return enable_uring;
}

struct fuse_uring_pdu {
	struct fuse_ring_ent *ent;
};

struct fuse_zero_copy_bvs {
	unsigned int nr_bvs;
	struct bio_vec bvs[];
};

static const struct fuse_iqueue_ops fuse_io_uring_ops;

enum fuse_uring_header_type {
	/* struct fuse_in_header / struct fuse_out_header */
	FUSE_URING_HEADER_IN_OUT,
	/* per op code header */
	FUSE_URING_HEADER_OP,
	/* struct fuse_uring_ent_in_out header */
	FUSE_URING_HEADER_RING_ENT,
};

static inline bool bufpool_enabled(struct fuse_ring_queue *queue)
{
	return queue->payload_mode == FUSE_PAYLOAD_BUFPOOL;
}

static inline bool bufpool_registered(struct fuse_ring_queue *queue)
{
	return queue->bufpool && queue->bufpool->registered;
}

/*
 * For a registered bufpool, every sqe that drives a payload import (REGISTER,
 * COMMIT_AND_FETCH) must carry the registered buffer index of the pool.
 * This also must be called from the command's issue handler, where cmd->sqe is
 * still valid
 */
static inline bool fuse_uring_cmd_index_ok(struct io_uring_cmd *cmd,
					   struct fuse_ring_queue *queue)
{
	if (!bufpool_registered(queue))
		return true;

	return (cmd->flags & IORING_URING_CMD_FIXED) &&
	       READ_ONCE(cmd->sqe->buf_index) == queue->bufpool->registered_index;
}

static void uring_cmd_set_ring_ent(struct io_uring_cmd *cmd,
				   struct fuse_ring_ent *ring_ent)
{
	struct fuse_uring_pdu *pdu =
		io_uring_cmd_to_pdu(cmd, struct fuse_uring_pdu);

	pdu->ent = ring_ent;
}

static struct fuse_ring_ent *uring_cmd_to_ring_ent(struct io_uring_cmd *cmd)
{
	struct fuse_uring_pdu *pdu =
		io_uring_cmd_to_pdu(cmd, struct fuse_uring_pdu);

	return pdu->ent;
}

static void fuse_uring_flush_bg(struct fuse_ring_queue *queue)
{
	struct fuse_ring *ring = queue->ring;
	struct fuse_chan *fch = ring->chan;

	lockdep_assert_held(&queue->lock);
	lockdep_assert_held(&fch->bg_lock);

	/*
	 * Allow one bg request per queue, ignoring global fc limits.
	 * This prevents a single queue from consuming all resources and
	 * eliminates the need for remote queue wake-ups when global
	 * limits are met but this queue has no more waiting requests.
	 */
	while ((fch->active_background < fch->max_background ||
		!queue->active_background) &&
	       (!list_empty(&queue->fuse_req_bg_queue))) {
		struct fuse_req *req;

		req = list_first_entry(&queue->fuse_req_bg_queue,
				       struct fuse_req, list);
		fch->active_background++;
		queue->active_background++;

		list_move_tail(&req->list, &queue->fuse_req_queue);
	}
}

static bool can_zero_copy_req(struct fuse_ring_ent *ent, struct fuse_req *req)
{
	struct fuse_args *args = req->args;

	if (!ent->queue->zero_copy || !args->zero_copy)
		return false;

	if (args->opcode != FUSE_READ && args->opcode != FUSE_WRITE)
		return false;

	return args->in_pages || args->out_pages;
}

static void zero_copy_unregister(struct io_uring_cmd *cmd,
				 struct fuse_ring_ent *ent,
				 unsigned int issue_flags)
{
	if (ent->zero_copied) {
		int err = io_buffer_unregister(cmd, ent->zero_copy_index,
					       issue_flags);

		if (err)
			pr_warn_ratelimited("qid=%d zero-copy unregister failed: %d\n",
					    ent->queue->qid, err);
		ent->zero_copied = false;
	}
}

static void fuse_uring_req_end(struct fuse_ring_ent *ent, struct fuse_req *req,
			       int error, unsigned int issue_flags)
{
	struct fuse_ring_queue *queue = ent->queue;
	struct fuse_ring *ring = queue->ring;
	struct fuse_chan *fch = ring->chan;

	lockdep_assert_not_held(&queue->lock);
	spin_lock(&queue->lock);
	ent->fuse_req = NULL;
	list_del_init(&req->list);
	if (test_bit(FR_BACKGROUND, &req->flags)) {
		queue->active_background--;
		spin_lock(&fch->bg_lock);
		fuse_request_bg_finish(fch, req);
		fuse_uring_flush_bg(queue);
		spin_unlock(&fch->bg_lock);
	}

	spin_unlock(&queue->lock);

	zero_copy_unregister(ent->cmd, ent, issue_flags);

	if (error)
		req->out.h.error = error;

	clear_bit(FR_SENT, &req->flags);
	fuse_request_end(req);
}

/* Abort all list queued request on the given ring queue */
static void fuse_uring_abort_end_queue_requests(struct fuse_ring_queue *queue)
{
	struct fuse_req *req;
	LIST_HEAD(req_list);

	spin_lock(&queue->lock);
	list_for_each_entry(req, &queue->fuse_req_queue, list)
		clear_bit(FR_PENDING, &req->flags);
	list_splice_init(&queue->fuse_req_queue, &req_list);
	spin_unlock(&queue->lock);

	/* must not hold queue lock to avoid order issues with fi->lock */
	fuse_dev_end_requests(&req_list);
}

void fuse_uring_abort_end_requests(struct fuse_ring *ring)
{
	int qid;
	struct fuse_ring_queue *queue;
	struct fuse_chan *fch = ring->chan;

	for (qid = 0; qid < ring->nr_queues; qid++) {
		queue = READ_ONCE(ring->queues[qid]);
		if (!queue)
			continue;

		WARN_ON_ONCE(fch->max_background != UINT_MAX);
		spin_lock(&queue->lock);
		queue->stopped = true;
		spin_lock(&fch->bg_lock);
		fuse_uring_flush_bg(queue);
		spin_unlock(&fch->bg_lock);
		spin_unlock(&queue->lock);
		fuse_uring_abort_end_queue_requests(queue);
	}
}

static bool ent_list_request_expired(struct fuse_chan *fch, struct list_head *list)
{
	struct fuse_ring_ent *ent;
	struct fuse_req *req;

	ent = list_first_entry_or_null(list, struct fuse_ring_ent, list);
	if (!ent)
		return false;

	req = ent->fuse_req;

	return time_is_before_jiffies(req->create_time +
				      fch->timeout.req_timeout);
}

bool fuse_uring_request_expired(struct fuse_chan *fch)
{
	struct fuse_ring *ring = fch->ring;
	struct fuse_ring_queue *queue;
	int qid;

	if (!ring)
		return false;

	for (qid = 0; qid < ring->nr_queues; qid++) {
		queue = READ_ONCE(ring->queues[qid]);
		if (!queue)
			continue;

		spin_lock(&queue->lock);
		if (fuse_request_expired(fch, &queue->fuse_req_queue) ||
		    fuse_request_expired(fch, &queue->fuse_req_bg_queue) ||
		    ent_list_request_expired(fch, &queue->ent_w_req_queue) ||
		    ent_list_request_expired(fch, &queue->ent_in_userspace)) {
			spin_unlock(&queue->lock);
			return true;
		}
		spin_unlock(&queue->lock);
	}

	return false;
}

void fuse_uring_destruct(struct fuse_chan *fch)
{
	struct fuse_ring *ring = fch->ring;
	int qid;

	if (!ring)
		return;

	for (qid = 0; qid < ring->nr_queues; qid++) {
		struct fuse_ring_queue *queue = READ_ONCE(ring->queues[qid]);
		struct fuse_ring_ent *ent, *next;

		if (!queue)
			continue;

		WARN_ON(!list_empty(&queue->ent_avail_queue));
		WARN_ON(!list_empty(&queue->ent_w_req_queue));
		WARN_ON(!list_empty(&queue->ent_commit_queue));
		WARN_ON(!list_empty(&queue->ent_in_userspace));

		list_for_each_entry_safe(ent, next, &queue->ent_released,
					 list) {
			list_del_init(&ent->list);
			kfree(ent);
		}

		kfree(queue->fpq.processing);
		kfree(queue->bufpool);
		kfree(queue);
		WRITE_ONCE(ring->queues[qid], NULL);
	}

	kfree(ring->queues);
	kfree(ring);
	fch->ring = NULL;
}

/*
 * Basic ring setup for this connection based on the provided configuration
 */
static struct fuse_ring *fuse_uring_create(struct fuse_chan *fch)
{
	struct fuse_ring *ring;
	size_t nr_queues = num_possible_cpus();
	size_t max_payload_size;

	ring = kzalloc_obj(*ring, GFP_KERNEL_ACCOUNT);
	if (!ring)
		return NULL;

	ring->queues = kzalloc_objs(struct fuse_ring_queue *, nr_queues,
				    GFP_KERNEL_ACCOUNT);
	if (!ring->queues)
		goto out_err;

	max_payload_size = max(FUSE_MIN_READ_BUFFER, fch->max_write);
	max_payload_size = max(max_payload_size, fch->max_pages * PAGE_SIZE);

	spin_lock(&fch->lock);
	if (!fch->connected) {
		spin_unlock(&fch->lock);
		goto out_err;
	}

	init_waitqueue_head(&ring->stop_waitq);

	ring->nr_queues = nr_queues;
	ring->chan = fch;
	ring->max_payload_sz = max_payload_size;
	smp_store_release(&fch->ring, ring);

	spin_unlock(&fch->lock);
	return ring;

out_err:
	kfree(ring->queues);
	kfree(ring);
	return NULL;
}

void fuse_uring_conn_init(struct fuse_chan *fch)
{
	if (fuse_uring_create(fch))
		fch->io_uring = 1;
}

static struct fuse_ring_queue *fuse_uring_create_queue(struct fuse_ring *ring,
						       int qid, bool zero_copy,
						       bool fail_if_exists)
{
	struct fuse_chan *fch = ring->chan;
	struct fuse_ring_queue *queue;
	struct list_head *pq;

	queue = kzalloc_obj(*queue, GFP_KERNEL_ACCOUNT);
	if (!queue)
		return ERR_PTR(-ENOMEM);
	pq = fuse_pqueue_alloc();
	if (!pq) {
		kfree(queue);
		return ERR_PTR(-ENOMEM);
	}

	queue->qid = qid;
	queue->ring = ring;
	spin_lock_init(&queue->lock);
	queue->zero_copy = zero_copy;

	INIT_LIST_HEAD(&queue->ent_avail_queue);
	INIT_LIST_HEAD(&queue->ent_commit_queue);
	INIT_LIST_HEAD(&queue->ent_w_req_queue);
	INIT_LIST_HEAD(&queue->ent_in_userspace);
	INIT_LIST_HEAD(&queue->fuse_req_queue);
	INIT_LIST_HEAD(&queue->fuse_req_bg_queue);
	INIT_LIST_HEAD(&queue->ent_released);

	fuse_pqueue_init(&queue->fpq);
	queue->fpq.processing = pq;

	spin_lock(&fch->lock);
	if (ring->queues[qid]) {
		spin_unlock(&fch->lock);
		kfree(queue->fpq.processing);
		kfree(queue->bufpool);
		kfree(queue);
		return fail_if_exists ? ERR_PTR(-EEXIST) : ring->queues[qid];
	}

	/*
	 * fch->lock serializes concurrent creators for this qid.
	 * smp_store_release() are for the lockless readers who must see a
	 * fully initialized queue after &ring->queues[qid] is set
	 */
	smp_store_release(&ring->queues[qid], queue);
	spin_unlock(&fch->lock);

	return queue;
}

static void fuse_uring_stop_fuse_req_end(struct fuse_req *req)
{
	clear_bit(FR_SENT, &req->flags);
	req->out.h.error = -ECONNABORTED;
	fuse_request_end(req);
}

/*
 * Release a request/entry on connection tear down
 */
static void fuse_uring_entry_teardown(struct fuse_ring_ent *ent)
{
	struct fuse_req *req;
	struct io_uring_cmd *cmd;

	struct fuse_ring_queue *queue = ent->queue;

	spin_lock(&queue->lock);
	cmd = ent->cmd;
	ent->cmd = NULL;
	req = ent->fuse_req;
	ent->fuse_req = NULL;
	if (req) {
		/* remove entry from queue->fpq->processing */
		list_del_init(&req->list);
	}

	/*
	 * The entry must not be freed immediately, due to access of direct
	 * pointer access of entries through IO_URING_F_CANCEL - there is a risk
	 * of race between daemon termination (which triggers IO_URING_F_CANCEL
	 * and accesses entries without checking the list state first
	 */
	list_move(&ent->list, &queue->ent_released);
	ent->state = FRRS_RELEASED;
	spin_unlock(&queue->lock);

	if (cmd)
		io_uring_cmd_done(cmd, -ENOTCONN, IO_URING_F_UNLOCKED);

	if (req)
		fuse_uring_stop_fuse_req_end(req);
}

static void fuse_uring_stop_list_entries(struct list_head *head,
					 struct fuse_ring_queue *queue,
					 enum fuse_ring_req_state exp_state)
{
	struct fuse_ring *ring = queue->ring;
	struct fuse_ring_ent *ent, *next;
	ssize_t queue_refs = SSIZE_MAX;
	LIST_HEAD(to_teardown);

	spin_lock(&queue->lock);
	list_for_each_entry_safe(ent, next, head, list) {
		if (ent->state != exp_state) {
			pr_warn("entry teardown qid=%d state=%d expected=%d",
				queue->qid, ent->state, exp_state);
			continue;
		}

		ent->state = FRRS_TEARDOWN;
		list_move(&ent->list, &to_teardown);
	}
	spin_unlock(&queue->lock);

	/* no queue lock to avoid lock order issues */
	list_for_each_entry_safe(ent, next, &to_teardown, list) {
		fuse_uring_entry_teardown(ent);
		queue_refs = atomic_dec_return(&ring->queue_refs);
		WARN_ON_ONCE(queue_refs < 0);
	}
}

static void fuse_uring_teardown_entries(struct fuse_ring_queue *queue)
{
	fuse_uring_stop_list_entries(&queue->ent_in_userspace, queue,
				     FRRS_USERSPACE);
	fuse_uring_stop_list_entries(&queue->ent_avail_queue, queue,
				     FRRS_AVAILABLE);
}

static void fuse_uring_teardown_all_queues(struct fuse_ring *ring)
{
	int qid;

	for (qid = 0; qid < ring->nr_queues; qid++) {
		struct fuse_ring_queue *queue = READ_ONCE(ring->queues[qid]);

		if (!queue)
			continue;

		fuse_uring_teardown_entries(queue);
	}
}

/*
 * Log state debug info
 */
static void fuse_uring_log_ent_state(struct fuse_ring *ring)
{
	int qid;
	struct fuse_ring_ent *ent;

	for (qid = 0; qid < ring->nr_queues; qid++) {
		struct fuse_ring_queue *queue = READ_ONCE(ring->queues[qid]);

		if (!queue)
			continue;

		spin_lock(&queue->lock);
		/*
		 * Log entries from the intermediate queue, the other queues
		 * should be empty
		 */
		list_for_each_entry(ent, &queue->ent_w_req_queue, list) {
			pr_info(" ent-req-queue ring=%p qid=%d ent=%p state=%d\n",
				ring, qid, ent, ent->state);
		}
		list_for_each_entry(ent, &queue->ent_commit_queue, list) {
			pr_info(" ent-commit-queue ring=%p qid=%d ent=%p state=%d\n",
				ring, qid, ent, ent->state);
		}
		spin_unlock(&queue->lock);
	}
	ring->stop_debug_log = 1;
}

static void fuse_uring_async_stop_queues(struct work_struct *work)
{
	struct fuse_ring *ring =
		container_of(work, struct fuse_ring, async_teardown_work.work);

	fuse_uring_teardown_all_queues(ring);

	/*
	 * Some ring entries might be in the middle of IO operations,
	 * i.e. in process to get handled by file_operations::uring_cmd
	 * or on the way to userspace - we could handle that with conditions in
	 * run time code, but easier/cleaner to have an async tear down handler
	 * If there are still queue references left
	 */
	if (atomic_read(&ring->queue_refs) > 0) {
		if (time_after(jiffies,
			       ring->teardown_time + FUSE_URING_TEARDOWN_TIMEOUT))
			fuse_uring_log_ent_state(ring);

		schedule_delayed_work(&ring->async_teardown_work,
				      FUSE_URING_TEARDOWN_INTERVAL);
	} else {
		wake_up_all(&ring->stop_waitq);
		fuse_conn_put(ring->chan->conn);
	}
}

/*
 * Stop the ring queues
 */
void fuse_uring_stop_queues(struct fuse_ring *ring)
{
	fuse_uring_teardown_all_queues(ring);

	if (atomic_read(&ring->queue_refs) > 0) {
		fuse_conn_get(ring->chan->conn);
		ring->teardown_time = jiffies;
		INIT_DELAYED_WORK(&ring->async_teardown_work,
				  fuse_uring_async_stop_queues);
		schedule_delayed_work(&ring->async_teardown_work,
				      FUSE_URING_TEARDOWN_INTERVAL);
	} else {
		wake_up_all(&ring->stop_waitq);
	}
}

/*
 * Handle IO_URING_F_CANCEL, typically should come on daemon termination.
 *
 * Releasing the last entry should trigger fuse_dev_release() if
 * the daemon was terminated
 */
static void fuse_uring_cancel(struct io_uring_cmd *cmd,
			      unsigned int issue_flags)
{
	struct fuse_ring_ent *ent = uring_cmd_to_ring_ent(cmd);
	struct fuse_ring_queue *queue;
	bool need_cmd_done = false;

	/*
	 * direct access on ent - it must not be destructed as long as
	 * IO_URING_F_CANCEL might come up
	 */
	queue = ent->queue;
	spin_lock(&queue->lock);
	if (ent->state == FRRS_AVAILABLE) {
		list_del_init(&ent->list);
		need_cmd_done = true;
		ent->cmd = NULL;
	}
	spin_unlock(&queue->lock);

	if (need_cmd_done) {
		/* no queue lock to avoid lock order issues */
		io_uring_cmd_done(cmd, -ENOTCONN, issue_flags);
		kfree(ent);
		if (atomic_dec_and_test(&queue->ring->queue_refs))
			wake_up_all(&queue->ring->stop_waitq);
	}
}

static void fuse_uring_prepare_cancel(struct io_uring_cmd *cmd, int issue_flags,
				      struct fuse_ring_ent *ring_ent)
{
	uring_cmd_set_ring_ent(cmd, ring_ent);
	io_uring_cmd_mark_cancelable(cmd, issue_flags);
}

/*
 * Checks for errors and stores it into the request
 */
static int fuse_uring_out_header_has_err(struct fuse_out_header *oh,
					 struct fuse_req *req)
{
	int err;

	err = -EINVAL;
	if (oh->unique == 0) {
		/* Not supported through io-uring yet */
		pr_warn_once("notify through fuse-io-uring not supported\n");
		goto err;
	}

	if (oh->error <= -ERESTARTSYS || oh->error > 0)
		goto err;

	if (oh->error) {
		err = oh->error;
		goto err;
	}

	err = -ENOENT;
	if ((oh->unique & ~FUSE_INT_REQ_BIT) != req->in.h.unique) {
		pr_warn_ratelimited("unique mismatch, expected: %llu got %llu\n",
				    req->in.h.unique,
				    oh->unique & ~FUSE_INT_REQ_BIT);
		goto err;
	}

	/*
	 * Is it an interrupt reply ID?
	 * XXX: Not supported through fuse-io-uring yet, it should not even
	 *      find the request - should not happen.
	 */
	WARN_ON_ONCE(oh->unique & FUSE_INT_REQ_BIT);

	err = 0;
err:
	return err;
}

static int ring_header_type_offset(enum fuse_uring_header_type type)
{
	switch (type) {
	case FUSE_URING_HEADER_IN_OUT:
		return 0;
	case FUSE_URING_HEADER_OP:
		return offsetof(struct fuse_uring_req_header, op_in);
	case FUSE_URING_HEADER_RING_ENT:
		return offsetof(struct fuse_uring_req_header, ring_ent_in_out);
	default:
		WARN_ONCE(1, "Invalid header type: %d\n", type);
		return -EINVAL;
	}
}

static int copy_header_to_ring(struct fuse_ring_ent *ent,
			       enum fuse_uring_header_type type,
			       const void *header, size_t header_size)
{
	int offset = ring_header_type_offset(type);
	void __user *ring;

	if (offset < 0)
		return offset;

	ring = (void __user *)ent->headers + offset;

	if (copy_to_user(ring, header, header_size)) {
		pr_info_ratelimited("Copying header to ring failed.\n");
		return -EFAULT;
	}

	return 0;
}

static int copy_header_from_ring(struct fuse_ring_ent *ent,
				 enum fuse_uring_header_type type, void *header,
				 size_t header_size)
{
	int offset = ring_header_type_offset(type);
	const void __user *ring;

	if (offset < 0)
		return offset;

	ring = (void __user *)ent->headers + offset;

	if (copy_from_user(header, ring, header_size)) {
		pr_info_ratelimited("Copying header from ring failed.\n");
		return -EFAULT;
	}

	return 0;
}

static int fuse_uring_import_payload(struct fuse_ring_ent *ent, int dir,
				     struct iov_iter *iter,
				     unsigned int issue_flags)
{
	void __user *base = ent->payload.iov_base;
	size_t len = ent->payload.iov_len;
	int err = 0;

	if (!base) {
		memset(iter, 0, sizeof(*iter));
		return 0;
	}

	if (bufpool_registered(ent->queue))
		err = io_uring_cmd_import_fixed((u64)(uintptr_t)base, len, dir,
						iter, ent->cmd, issue_flags);
	else
		err = import_ubuf(dir, base, len, iter);

	if (err)
		pr_info_ratelimited("fuse: Import of user buffer failed\n");

	return err;
}

static int setup_fuse_copy_state(struct fuse_copy_state *cs,
				 struct fuse_req *req,
				 struct fuse_ring_ent *ent, int dir,
				 struct iov_iter *iter,
				 unsigned int issue_flags)
{
	int err;

	err = fuse_uring_import_payload(ent, dir, iter, issue_flags);
	if (err)
		return err;

	fuse_copy_init(cs, dir == ITER_DEST, iter);

	if (ent->zero_copied)
		cs->skip_folio_copy = true;

	cs->is_uring = true;
	cs->req = req;

	return 0;
}

static int fuse_uring_copy_from_ring(struct fuse_req *req,
				     struct fuse_ring_ent *ent,
				     unsigned int issue_flags)
{
	struct fuse_copy_state cs;
	struct fuse_args *args = req->args;
	struct iov_iter iter;
	int err;
	struct fuse_uring_ent_in_out ring_in_out;

	err = copy_header_from_ring(ent, FUSE_URING_HEADER_RING_ENT,
				    &ring_in_out, sizeof(ring_in_out));
	if (err)
		return err;

	err = setup_fuse_copy_state(&cs, req, ent, ITER_SOURCE, &iter,
				    issue_flags);
	if (err)
		return err;

	err = fuse_copy_out_args(&cs, args, ring_in_out.payload_sz);
	fuse_copy_finish(&cs);
	return err;
}

static void fuse_zero_copy_release(void *priv)
{
	struct fuse_zero_copy_bvs *zc_bvs = priv;
	unsigned int i;

	for (i = 0; i < zc_bvs->nr_bvs; i++)
		folio_put(page_folio(zc_bvs->bvs[i].bv_page));

	kvfree(zc_bvs);
}

static int fuse_uring_set_up_zero_copy(struct fuse_ring_ent *ent,
				       struct fuse_req *req,
				       unsigned int issue_flags)
{
	struct fuse_args_pages *ap;
	int err, i, ddir = 0;
	struct fuse_zero_copy_bvs *zc_bvs;
	struct bio_vec *bvs;

	/* out_pages indicates a read, in_pages indicates a write */
	if (req->args->out_pages)
		ddir |= IO_BUF_DEST;
	if (req->args->in_pages)
		ddir |= IO_BUF_SOURCE;

	ap = container_of(req->args, typeof(*ap), args);

	zc_bvs = kvmalloc_flex(*zc_bvs, bvs, ap->num_folios,
			       GFP_KERNEL_ACCOUNT);
	if (!zc_bvs)
		return -ENOMEM;

	zc_bvs->nr_bvs = ap->num_folios;
	bvs = zc_bvs->bvs;
	for (i = 0; i < ap->num_folios; i++) {
		bvs[i].bv_page = folio_page(ap->folios[i], 0);
		bvs[i].bv_offset = ap->descs[i].offset;
		bvs[i].bv_len = ap->descs[i].length;
		folio_get(ap->folios[i]);
	}

	err = io_buffer_register_bvec(ent->cmd, bvs, ap->num_folios,
				      fuse_zero_copy_release, zc_bvs,
				      ddir, ent->zero_copy_index,
				      issue_flags);
	if (err) {
		fuse_zero_copy_release(zc_bvs);
		return err;
	}

	ent->zero_copied = true;

	return 0;
}

/*
 * Copy data from the req to the ring buffer
 */
static int fuse_uring_args_to_ring(struct fuse_req *req,
				   struct fuse_ring_ent *ent,
				   unsigned int issue_flags)
{
	struct fuse_copy_state cs;
	struct fuse_args *args = req->args;
	struct fuse_in_arg *in_args = args->in_args;
	int num_args = args->in_numargs;
	int err;
	struct iov_iter iter;
	struct fuse_uring_ent_in_out ent_in_out = {
		.flags = 0,
		.commit_id = req->in.h.unique,
	};

	if (can_zero_copy_req(ent, req)) {
		ent_in_out.flags |= FUSE_URING_ENT_ZERO_COPY;
		err = fuse_uring_set_up_zero_copy(ent, req, issue_flags);
		if (err)
			return err;
	}

	err = setup_fuse_copy_state(&cs, req, ent, ITER_DEST, &iter,
				    issue_flags);
	if (err)
		return err;

	if (num_args > 0) {
		/*
		 * Expectation is that the first argument is the per op header.
		 * Some op code have that as zero size.
		 */
		if (args->in_args[0].size > 0) {
			err = copy_header_to_ring(ent, FUSE_URING_HEADER_OP,
						  in_args->value,
						  in_args->size);
			if (err)
				return err;
		}
		in_args++;
		num_args--;
	}

	/* copy the payload */
	err = fuse_copy_args(&cs, num_args, args->in_pages,
			     (struct fuse_arg *)in_args, 0);
	fuse_copy_finish(&cs);
	if (err) {
		pr_info_ratelimited("%s fuse_copy_args failed\n", __func__);
		return err;
	}

	ent_in_out.payload_sz = cs.ring.copied_sz;
	/*
	 * on a zero-copied write the pages are registered for the server to
	 * read via a fixed-buffer op rather than copied into the payload
	 * buffer, so copied_sz does not account for it. The server still needs
	 * the total inbound size to know how many bytes to read from the
	 * registered buffer, so add the page arg (always the last in-arg) back
	 * in
	 */
	if (cs.skip_folio_copy && args->in_pages)
		ent_in_out.payload_sz +=
			args->in_args[args->in_numargs - 1].size;

	if (bufpool_enabled(ent->queue) && ent->payload.iov_base)
		ent_in_out.offset =
			(uintptr_t)ent->payload.iov_base - ent->queue->bufpool->base_uaddr;

	return copy_header_to_ring(ent, FUSE_URING_HEADER_RING_ENT,
				   &ent_in_out, sizeof(ent_in_out));
}

static int fuse_uring_copy_to_ring(struct fuse_ring_ent *ent,
				   struct fuse_req *req,
				   unsigned int issue_flags)
{
	struct fuse_ring_queue *queue = ent->queue;
	struct fuse_in_header in_header;
	int err;

	err = -EIO;
	if (WARN_ON(ent->state != FRRS_FUSE_REQ)) {
		pr_err("qid=%d ring-req=%p invalid state %d on send\n",
		       queue->qid, ent, ent->state);
		return err;
	}

	err = -EINVAL;
	if (WARN_ON(req->in.h.unique == 0))
		return err;

	/* copy the request */
	err = fuse_uring_args_to_ring(req, ent, issue_flags);
	if (unlikely(err)) {
		pr_info_ratelimited("Copy to ring failed: %d\n", err);
		return err;
	}

	/* copy fuse_in_header */
	in_header = req->in.h;
	return copy_header_to_ring(ent, FUSE_URING_HEADER_IN_OUT, &in_header,
				   sizeof(in_header));
}

static bool fuse_uring_req_has_copyable_payload(struct fuse_ring_ent *ent,
						struct fuse_req *req)
{
	struct fuse_args *args = req->args;

	if (!can_zero_copy_req(ent, req))
		return args->in_numargs > 1 || args->out_numargs;

	/*
	 * the asymmetry between in_numargs > 2 and out_numargs > 1 is because
	 * the per-op header is extracted before fuse_copy_args() for inargs but
	 * not for outargs
	 */
	if ((args->in_numargs > 1) && (!args->in_pages || args->in_numargs > 2))
		return true;
	if (args->out_numargs && (!args->out_pages || args->out_numargs > 1))
		return true;

	return false;
}

static int fuse_uring_select_buffer(struct fuse_ring_ent *ent)
{
	struct fuse_ring_queue *queue = ent->queue;
	struct fuse_bufpool *pool = queue->bufpool;
	unsigned int id;

	lockdep_assert_held(&queue->lock);

	id = find_first_bit(pool->free_map, pool->nr_bufs);
	if (id >= pool->nr_bufs)
		return -ENOBUFS;

	WARN_ON_ONCE(ent->payload.iov_base);
	__clear_bit(id, pool->free_map);

	ent->buf_id = id;
	ent->payload.iov_base =
		(void __user *)(pool->base_uaddr + id * pool->buf_size);
	ent->payload.iov_len = pool->buf_size;

	return 0;
}

static void fuse_uring_recycle_buffer(struct fuse_ring_ent *ent)
{
	struct iovec *ent_payload = &ent->payload;
	struct fuse_ring_queue *queue = ent->queue;
	struct fuse_bufpool *pool;