// SPDX-License-Identifier: GPL-2.0/* * FUSE: Filesystem in Userspace * Copyright (c) 2023-2024 DataDirect Networks. */ #include "fuse_i.h"#include "dev_uring_i.h"#include "fuse_dev_i.h"#include "fuse_trace.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 */  bool fuse_uring_enabled(void){	return enable_uring;} struct fuse_uring_pdu {	struct fuse_ring_ent *ent;}; static const struct fuse_iqueue_ops fuse_io_uring_ops; 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_conn *fc = ring->fc; 	lockdep_assert_held(&queue->lock);	lockdep_assert_held(&fc->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 ((fc->active_background < fc->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);		fc->active_background++;		queue->active_background++; 		list_move_tail(&req->list, &queue->fuse_req_queue);	}} static void fuse_uring_req_end(struct fuse_ring_ent *ent, struct fuse_req *req,			       int error){	struct fuse_ring_queue *queue = ent->queue;	struct fuse_ring *ring = queue->ring;	struct fuse_conn *fc = ring->fc; 	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(&fc->bg_lock);		fuse_uring_flush_bg(queue);		spin_unlock(&fc->bg_lock);	} 	spin_unlock(&queue->lock); 	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_conn *fc = ring->fc; 	for (qid = 0; qid < ring->nr_queues; qid++) {		queue = READ_ONCE(ring->queues[qid]);		if (!queue)			continue; 		queue->stopped = true; 		WARN_ON_ONCE(ring->fc->max_background != UINT_MAX);		spin_lock(&queue->lock);		spin_lock(&fc->bg_lock);		fuse_uring_flush_bg(queue);		spin_unlock(&fc->bg_lock);		spin_unlock(&queue->lock);		fuse_uring_abort_end_queue_requests(queue);	}} static bool ent_list_request_expired(struct fuse_conn *fc, 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 +				      fc->timeout.req_timeout);} bool fuse_uring_request_expired(struct fuse_conn *fc){	struct fuse_ring *ring = fc->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(fc, &queue->fuse_req_queue) ||		    fuse_request_expired(fc, &queue->fuse_req_bg_queue) ||		    ent_list_request_expired(fc, &queue->ent_w_req_queue) ||		    ent_list_request_expired(fc, &queue->ent_in_userspace)) {			spin_unlock(&queue->lock);			return true;		}		spin_unlock(&queue->lock);	} 	return false;} void fuse_uring_destruct(struct fuse_conn *fc){	struct fuse_ring *ring = fc->ring;	int qid; 	if (!ring)		return; 	for (qid = 0; qid < ring->nr_queues; qid++) {		struct fuse_ring_queue *queue = 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);		ring->queues[qid] = NULL;	} 	kfree(ring->queues);	kfree(ring);	fc->ring = NULL;} /* * Basic ring setup for this connection based on the provided configuration */static struct fuse_ring *fuse_uring_create(struct fuse_conn *fc){	struct fuse_ring *ring;	size_t nr_queues = num_possible_cpus();	struct fuse_ring *res = NULL;	size_t max_payload_size; 	ring = kzalloc_obj(*fc->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, fc->max_write);	max_payload_size = max(max_payload_size, fc->max_pages * PAGE_SIZE); 	spin_lock(&fc->lock);	if (fc->ring) {		/* race, another thread created the ring in the meantime */		spin_unlock(&fc->lock);		res = fc->ring;		goto out_err;	} 	init_waitqueue_head(&ring->stop_waitq); 	ring->nr_queues = nr_queues;	ring->fc = fc;	ring->max_payload_sz = max_payload_size;	smp_store_release(&fc->ring, ring); 	spin_unlock(&fc->lock);	return ring; out_err:	kfree(ring->queues);	kfree(ring);	return res;} static struct fuse_ring_queue *fuse_uring_create_queue(struct fuse_ring *ring,						       int qid){	struct fuse_conn *fc = ring->fc;	struct fuse_ring_queue *queue;	struct list_head *pq; 	queue = kzalloc_obj(*queue, GFP_KERNEL_ACCOUNT);	if (!queue)		return NULL;	pq = kzalloc_objs(struct list_head, FUSE_PQ_HASH_SIZE);	if (!pq) {		kfree(queue);		return NULL;	} 	queue->qid = qid;	queue->ring = ring;	spin_lock_init(&queue->lock); 	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); 	queue->fpq.processing = pq;	fuse_pqueue_init(&queue->fpq); 	spin_lock(&fc->lock);	if (ring->queues[qid]) {		spin_unlock(&fc->lock);		kfree(queue->fpq.processing);		kfree(queue);		return ring->queues[qid];	} 	/*	 * write_once and lock as the caller mostly doesn't take the lock at all	 */	WRITE_ONCE(ring->queues[qid], queue);	spin_unlock(&fc->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);} /* * 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 = 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){	int qid;	struct fuse_ring *ring =		container_of(work, struct fuse_ring, async_teardown_work.work); 	/* XXX code dup */	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);	} 	/*	 * 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);	}} /* * Stop the ring queues */void fuse_uring_stop_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);	} 	if (atomic_read(&ring->queue_refs) > 0) {		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,

Lines 1–500 of 1375. Load more