// 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)

Lines 1–500 of 1905. Load more