projects/linux·blob master · refs

1905 lines · 47.2 KiB

1// SPDX-License-Identifier: GPL-2.0
2/*
3 * FUSE: Filesystem in Userspace
4 * Copyright (c) 2023-2024 DataDirect Networks.
5 */
6
7#include "dev.h"
8#include "args.h"
9#include "dev_uring_i.h"
10#include "fuse_trace.h"
11
12#include <linux/bitmap.h>
13#include <linux/fs.h>
14#include <linux/io_uring/cmd.h>
15
16static bool __read_mostly enable_uring;
17module_param(enable_uring, bool, 0644);
18MODULE_PARM_DESC(enable_uring,
19 "Enable userspace communication through io-uring");
20
21#define FUSE_URING_IOV_SEGS 2 /* header and payload */
22#define FUSE_URING_IOV_HEADERS 0
23#define FUSE_URING_IOV_PAYLOAD 1
24
25#define FUSE_URING_ADD_QUEUE_FLAGS (FUSE_URING_ZERO_COPY)
26
27bool fuse_uring_enabled(void)
28{
29 return enable_uring;
30}
31
32struct fuse_uring_pdu {
33 struct fuse_ring_ent *ent;
34};
35
36struct fuse_zero_copy_bvs {
37 unsigned int nr_bvs;
38 struct bio_vec bvs[];
39};
40
41static const struct fuse_iqueue_ops fuse_io_uring_ops;
42
43enum fuse_uring_header_type {
44 /* struct fuse_in_header / struct fuse_out_header */
45 FUSE_URING_HEADER_IN_OUT,
46 /* per op code header */
47 FUSE_URING_HEADER_OP,
48 /* struct fuse_uring_ent_in_out header */
49 FUSE_URING_HEADER_RING_ENT,
50};
51
52static inline bool bufpool_enabled(struct fuse_ring_queue *queue)
53{
54 return queue->payload_mode == FUSE_PAYLOAD_BUFPOOL;
55}
56
57static inline bool bufpool_registered(struct fuse_ring_queue *queue)
58{
59 return queue->bufpool && queue->bufpool->registered;
60}
61
62/*
63 * For a registered bufpool, every sqe that drives a payload import (REGISTER,
64 * COMMIT_AND_FETCH) must carry the registered buffer index of the pool.
65 * This also must be called from the command's issue handler, where cmd->sqe is
66 * still valid
67 */
68static inline bool fuse_uring_cmd_index_ok(struct io_uring_cmd *cmd,
69 struct fuse_ring_queue *queue)
70{
71 if (!bufpool_registered(queue))
72 return true;
73
74 return (cmd->flags & IORING_URING_CMD_FIXED) &&
75 READ_ONCE(cmd->sqe->buf_index) == queue->bufpool->registered_index;
76}
77
78static void uring_cmd_set_ring_ent(struct io_uring_cmd *cmd,
79 struct fuse_ring_ent *ring_ent)
80{
81 struct fuse_uring_pdu *pdu =
82 io_uring_cmd_to_pdu(cmd, struct fuse_uring_pdu);
83
84 pdu->ent = ring_ent;
85}
86
87static struct fuse_ring_ent *uring_cmd_to_ring_ent(struct io_uring_cmd *cmd)
88{
89 struct fuse_uring_pdu *pdu =
90 io_uring_cmd_to_pdu(cmd, struct fuse_uring_pdu);
91
92 return pdu->ent;
93}
94
95static void fuse_uring_flush_bg(struct fuse_ring_queue *queue)
96{
97 struct fuse_ring *ring = queue->ring;
98 struct fuse_chan *fch = ring->chan;
99
100 lockdep_assert_held(&queue->lock);
101 lockdep_assert_held(&fch->bg_lock);
102
103 /*
104 * Allow one bg request per queue, ignoring global fc limits.
105 * This prevents a single queue from consuming all resources and
106 * eliminates the need for remote queue wake-ups when global
107 * limits are met but this queue has no more waiting requests.
108 */
109 while ((fch->active_background < fch->max_background ||
110 !queue->active_background) &&
111 (!list_empty(&queue->fuse_req_bg_queue))) {
112 struct fuse_req *req;
113
114 req = list_first_entry(&queue->fuse_req_bg_queue,
115 struct fuse_req, list);
116 fch->active_background++;
117 queue->active_background++;
118
119 list_move_tail(&req->list, &queue->fuse_req_queue);
120 }
121}
122
123static bool can_zero_copy_req(struct fuse_ring_ent *ent, struct fuse_req *req)
124{
125 struct fuse_args *args = req->args;
126
127 if (!ent->queue->zero_copy || !args->zero_copy)
128 return false;
129
130 if (args->opcode != FUSE_READ && args->opcode != FUSE_WRITE)
131 return false;
132
133 return args->in_pages || args->out_pages;
134}
135
136static void zero_copy_unregister(struct io_uring_cmd *cmd,
137 struct fuse_ring_ent *ent,
138 unsigned int issue_flags)
139{
140 if (ent->zero_copied) {
141 int err = io_buffer_unregister(cmd, ent->zero_copy_index,
142 issue_flags);
143
144 if (err)
145 pr_warn_ratelimited("qid=%d zero-copy unregister failed: %d\n",
146 ent->queue->qid, err);
147 ent->zero_copied = false;
148 }
149}
150
151static void fuse_uring_req_end(struct fuse_ring_ent *ent, struct fuse_req *req,
152 int error, unsigned int issue_flags)
153{
154 struct fuse_ring_queue *queue = ent->queue;
155 struct fuse_ring *ring = queue->ring;
156 struct fuse_chan *fch = ring->chan;
157
158 lockdep_assert_not_held(&queue->lock);
159 spin_lock(&queue->lock);
160 ent->fuse_req = NULL;
161 list_del_init(&req->list);
162 if (test_bit(FR_BACKGROUND, &req->flags)) {
163 queue->active_background--;
164 spin_lock(&fch->bg_lock);
165 fuse_request_bg_finish(fch, req);
166 fuse_uring_flush_bg(queue);
167 spin_unlock(&fch->bg_lock);
168 }
169
170 spin_unlock(&queue->lock);
171
172 zero_copy_unregister(ent->cmd, ent, issue_flags);
173
174 if (error)
175 req->out.h.error = error;
176
177 clear_bit(FR_SENT, &req->flags);
178 fuse_request_end(req);
179}
180
181/* Abort all list queued request on the given ring queue */
182static void fuse_uring_abort_end_queue_requests(struct fuse_ring_queue *queue)
183{
184 struct fuse_req *req;
185 LIST_HEAD(req_list);
186
187 spin_lock(&queue->lock);
188 list_for_each_entry(req, &queue->fuse_req_queue, list)
189 clear_bit(FR_PENDING, &req->flags);
190 list_splice_init(&queue->fuse_req_queue, &req_list);
191 spin_unlock(&queue->lock);
192
193 /* must not hold queue lock to avoid order issues with fi->lock */
194 fuse_dev_end_requests(&req_list);
195}
196
197void fuse_uring_abort_end_requests(struct fuse_ring *ring)
198{
199 int qid;
200 struct fuse_ring_queue *queue;
201 struct fuse_chan *fch = ring->chan;
202
203 for (qid = 0; qid < ring->nr_queues; qid++) {
204 queue = READ_ONCE(ring->queues[qid]);
205 if (!queue)
206 continue;
207
208 WARN_ON_ONCE(fch->max_background != UINT_MAX);
209 spin_lock(&queue->lock);
210 queue->stopped = true;
211 spin_lock(&fch->bg_lock);
212 fuse_uring_flush_bg(queue);
213 spin_unlock(&fch->bg_lock);
214 spin_unlock(&queue->lock);
215 fuse_uring_abort_end_queue_requests(queue);
216 }
217}
218
219static bool ent_list_request_expired(struct fuse_chan *fch, struct list_head *list)
220{
221 struct fuse_ring_ent *ent;
222 struct fuse_req *req;
223
224 ent = list_first_entry_or_null(list, struct fuse_ring_ent, list);
225 if (!ent)
226 return false;
227
228 req = ent->fuse_req;
229
230 return time_is_before_jiffies(req->create_time +
231 fch->timeout.req_timeout);
232}
233
234bool fuse_uring_request_expired(struct fuse_chan *fch)
235{
236 struct fuse_ring *ring = fch->ring;
237 struct fuse_ring_queue *queue;
238 int qid;
239
240 if (!ring)
241 return false;
242
243 for (qid = 0; qid < ring->nr_queues; qid++) {
244 queue = READ_ONCE(ring->queues[qid]);
245 if (!queue)
246 continue;
247
248 spin_lock(&queue->lock);
249 if (fuse_request_expired(fch, &queue->fuse_req_queue) ||
250 fuse_request_expired(fch, &queue->fuse_req_bg_queue) ||
251 ent_list_request_expired(fch, &queue->ent_w_req_queue) ||
252 ent_list_request_expired(fch, &queue->ent_in_userspace)) {
253 spin_unlock(&queue->lock);
254 return true;
255 }
256 spin_unlock(&queue->lock);
257 }
258
259 return false;
260}
261
262void fuse_uring_destruct(struct fuse_chan *fch)
263{
264 struct fuse_ring *ring = fch->ring;
265 int qid;
266
267 if (!ring)
268 return;
269
270 for (qid = 0; qid < ring->nr_queues; qid++) {
271 struct fuse_ring_queue *queue = READ_ONCE(ring->queues[qid]);
272 struct fuse_ring_ent *ent, *next;
273
274 if (!queue)
275 continue;
276
277 WARN_ON(!list_empty(&queue->ent_avail_queue));
278 WARN_ON(!list_empty(&queue->ent_w_req_queue));
279 WARN_ON(!list_empty(&queue->ent_commit_queue));
280 WARN_ON(!list_empty(&queue->ent_in_userspace));
281
282 list_for_each_entry_safe(ent, next, &queue->ent_released,
283 list) {
284 list_del_init(&ent->list);
285 kfree(ent);
286 }
287
288 kfree(queue->fpq.processing);
289 kfree(queue->bufpool);
290 kfree(queue);
291 WRITE_ONCE(ring->queues[qid], NULL);
292 }
293
294 kfree(ring->queues);
295 kfree(ring);
296 fch->ring = NULL;
297}
298
299/*
300 * Basic ring setup for this connection based on the provided configuration
301 */
302static struct fuse_ring *fuse_uring_create(struct fuse_chan *fch)
303{
304 struct fuse_ring *ring;
305 size_t nr_queues = num_possible_cpus();
306 size_t max_payload_size;
307
308 ring = kzalloc_obj(*ring, GFP_KERNEL_ACCOUNT);
309 if (!ring)
310 return NULL;
311
312 ring->queues = kzalloc_objs(struct fuse_ring_queue *, nr_queues,
313 GFP_KERNEL_ACCOUNT);
314 if (!ring->queues)
315 goto out_err;
316
317 max_payload_size = max(FUSE_MIN_READ_BUFFER, fch->max_write);
318 max_payload_size = max(max_payload_size, fch->max_pages * PAGE_SIZE);
319
320 spin_lock(&fch->lock);
321 if (!fch->connected) {
322 spin_unlock(&fch->lock);
323 goto out_err;
324 }
325
326 init_waitqueue_head(&ring->stop_waitq);
327
328 ring->nr_queues = nr_queues;
329 ring->chan = fch;
330 ring->max_payload_sz = max_payload_size;
331 smp_store_release(&fch->ring, ring);
332
333 spin_unlock(&fch->lock);
334 return ring;
335
336out_err:
337 kfree(ring->queues);
338 kfree(ring);
339 return NULL;
340}
341
342void fuse_uring_conn_init(struct fuse_chan *fch)
343{
344 if (fuse_uring_create(fch))
345 fch->io_uring = 1;
346}
347
348static struct fuse_ring_queue *fuse_uring_create_queue(struct fuse_ring *ring,
349 int qid, bool zero_copy,
350 bool fail_if_exists)
351{
352 struct fuse_chan *fch = ring->chan;
353 struct fuse_ring_queue *queue;
354 struct list_head *pq;
355
356 queue = kzalloc_obj(*queue, GFP_KERNEL_ACCOUNT);
357 if (!queue)
358 return ERR_PTR(-ENOMEM);
359 pq = fuse_pqueue_alloc();
360 if (!pq) {
361 kfree(queue);
362 return ERR_PTR(-ENOMEM);
363 }
364
365 queue->qid = qid;
366 queue->ring = ring;
367 spin_lock_init(&queue->lock);
368 queue->zero_copy = zero_copy;
369
370 INIT_LIST_HEAD(&queue->ent_avail_queue);
371 INIT_LIST_HEAD(&queue->ent_commit_queue);
372 INIT_LIST_HEAD(&queue->ent_w_req_queue);
373 INIT_LIST_HEAD(&queue->ent_in_userspace);
374 INIT_LIST_HEAD(&queue->fuse_req_queue);
375 INIT_LIST_HEAD(&queue->fuse_req_bg_queue);
376 INIT_LIST_HEAD(&queue->ent_released);
377
378 fuse_pqueue_init(&queue->fpq);
379 queue->fpq.processing = pq;
380
381 spin_lock(&fch->lock);
382 if (ring->queues[qid]) {
383 spin_unlock(&fch->lock);
384 kfree(queue->fpq.processing);
385 kfree(queue->bufpool);
386 kfree(queue);
387 return fail_if_exists ? ERR_PTR(-EEXIST) : ring->queues[qid];
388 }
389
390 /*
391 * fch->lock serializes concurrent creators for this qid.
392 * smp_store_release() are for the lockless readers who must see a
393 * fully initialized queue after &ring->queues[qid] is set
394 */
395 smp_store_release(&ring->queues[qid], queue);
396 spin_unlock(&fch->lock);
397
398 return queue;
399}
400
401static void fuse_uring_stop_fuse_req_end(struct fuse_req *req)
402{
403 clear_bit(FR_SENT, &req->flags);
404 req->out.h.error = -ECONNABORTED;
405 fuse_request_end(req);
406}
407
408/*
409 * Release a request/entry on connection tear down
410 */
411static void fuse_uring_entry_teardown(struct fuse_ring_ent *ent)
412{
413 struct fuse_req *req;
414 struct io_uring_cmd *cmd;
415
416 struct fuse_ring_queue *queue = ent->queue;
417
418 spin_lock(&queue->lock);
419 cmd = ent->cmd;
420 ent->cmd = NULL;
421 req = ent->fuse_req;
422 ent->fuse_req = NULL;
423 if (req) {
424 /* remove entry from queue->fpq->processing */
425 list_del_init(&req->list);
426 }
427
428 /*
429 * The entry must not be freed immediately, due to access of direct
430 * pointer access of entries through IO_URING_F_CANCEL - there is a risk
431 * of race between daemon termination (which triggers IO_URING_F_CANCEL
432 * and accesses entries without checking the list state first
433 */
434 list_move(&ent->list, &queue->ent_released);
435 ent->state = FRRS_RELEASED;
436 spin_unlock(&queue->lock);
437
438 if (cmd)
439 io_uring_cmd_done(cmd, -ENOTCONN, IO_URING_F_UNLOCKED);
440
441 if (req)
442 fuse_uring_stop_fuse_req_end(req);
443}
444
445static void fuse_uring_stop_list_entries(struct list_head *head,
446 struct fuse_ring_queue *queue,
447 enum fuse_ring_req_state exp_state)
448{
449 struct fuse_ring *ring = queue->ring;
450 struct fuse_ring_ent *ent, *next;
451 ssize_t queue_refs = SSIZE_MAX;
452 LIST_HEAD(to_teardown);
453
454 spin_lock(&queue->lock);
455 list_for_each_entry_safe(ent, next, head, list) {
456 if (ent->state != exp_state) {
457 pr_warn("entry teardown qid=%d state=%d expected=%d",
458 queue->qid, ent->state, exp_state);
459 continue;
460 }
461
462 ent->state = FRRS_TEARDOWN;
463 list_move(&ent->list, &to_teardown);
464 }
465 spin_unlock(&queue->lock);
466
467 /* no queue lock to avoid lock order issues */
468 list_for_each_entry_safe(ent, next, &to_teardown, list) {
469 fuse_uring_entry_teardown(ent);
470 queue_refs = atomic_dec_return(&ring->queue_refs);
471 WARN_ON_ONCE(queue_refs < 0);
472 }
473}
474
475static void fuse_uring_teardown_entries(struct fuse_ring_queue *queue)
476{
477 fuse_uring_stop_list_entries(&queue->ent_in_userspace, queue,
478 FRRS_USERSPACE);
479 fuse_uring_stop_list_entries(&queue->ent_avail_queue, queue,
480 FRRS_AVAILABLE);
481}
482
483static void fuse_uring_teardown_all_queues(struct fuse_ring *ring)
484{
485 int qid;
486
487 for (qid = 0; qid < ring->nr_queues; qid++) {
488 struct fuse_ring_queue *queue = READ_ONCE(ring->queues[qid]);
489
490 if (!queue)
491 continue;
492
493 fuse_uring_teardown_entries(queue);
494 }
495}
496
497/*
498 * Log state debug info
499 */
500static void fuse_uring_log_ent_state(struct fuse_ring *ring)
501{
502 int qid;
503 struct fuse_ring_ent *ent;
504
505 for (qid = 0; qid < ring->nr_queues; qid++) {
506 struct fuse_ring_queue *queue = READ_ONCE(ring->queues[qid]);
507
508 if (!queue)
509 continue;
510
511 spin_lock(&queue->lock);
512 /*
513 * Log entries from the intermediate queue, the other queues
514 * should be empty
515 */
516 list_for_each_entry(ent, &queue->ent_w_req_queue, list) {
517 pr_info(" ent-req-queue ring=%p qid=%d ent=%p state=%d\n",
518 ring, qid, ent, ent->state);
519 }
520 list_for_each_entry(ent, &queue->ent_commit_queue, list) {
521 pr_info(" ent-commit-queue ring=%p qid=%d ent=%p state=%d\n",
522 ring, qid, ent, ent->state);
523 }
524 spin_unlock(&queue->lock);
525 }
526 ring->stop_debug_log = 1;
527}
528
529static void fuse_uring_async_stop_queues(struct work_struct *work)
530{
531 struct fuse_ring *ring =
532 container_of(work, struct fuse_ring, async_teardown_work.work);
533
534 fuse_uring_teardown_all_queues(ring);
535
536 /*
537 * Some ring entries might be in the middle of IO operations,
538 * i.e. in process to get handled by file_operations::uring_cmd
539 * or on the way to userspace - we could handle that with conditions in
540 * run time code, but easier/cleaner to have an async tear down handler
541 * If there are still queue references left
542 */
543 if (atomic_read(&ring->queue_refs) > 0) {
544 if (time_after(jiffies,
545 ring->teardown_time + FUSE_URING_TEARDOWN_TIMEOUT))
546 fuse_uring_log_ent_state(ring);
547
548 schedule_delayed_work(&ring->async_teardown_work,
549 FUSE_URING_TEARDOWN_INTERVAL);
550 } else {
551 wake_up_all(&ring->stop_waitq);
552 fuse_conn_put(ring->chan->conn);
553 }
554}
555
556/*
557 * Stop the ring queues
558 */
559void fuse_uring_stop_queues(struct fuse_ring *ring)
560{
561 fuse_uring_teardown_all_queues(ring);
562
563 if (atomic_read(&ring->queue_refs) > 0) {
564 fuse_conn_get(ring->chan->conn);
565 ring->teardown_time = jiffies;
566 INIT_DELAYED_WORK(&ring->async_teardown_work,
567 fuse_uring_async_stop_queues);
568 schedule_delayed_work(&ring->async_teardown_work,
569 FUSE_URING_TEARDOWN_INTERVAL);
570 } else {
571 wake_up_all(&ring->stop_waitq);
572 }
573}
574
575/*
576 * Handle IO_URING_F_CANCEL, typically should come on daemon termination.
577 *
578 * Releasing the last entry should trigger fuse_dev_release() if
579 * the daemon was terminated
580 */
581static void fuse_uring_cancel(struct io_uring_cmd *cmd,
582 unsigned int issue_flags)
583{
584 struct fuse_ring_ent *ent = uring_cmd_to_ring_ent(cmd);
585 struct fuse_ring_queue *queue;
586 bool need_cmd_done = false;
587
588 /*
589 * direct access on ent - it must not be destructed as long as
590 * IO_URING_F_CANCEL might come up
591 */
592 queue = ent->queue;
593 spin_lock(&queue->lock);
594 if (ent->state == FRRS_AVAILABLE) {
595 list_del_init(&ent->list);
596 need_cmd_done = true;
597 ent->cmd = NULL;
598 }
599 spin_unlock(&queue->lock);
600
601 if (need_cmd_done) {
602 /* no queue lock to avoid lock order issues */
603 io_uring_cmd_done(cmd, -ENOTCONN, issue_flags);
604 kfree(ent);
605 if (atomic_dec_and_test(&queue->ring->queue_refs))
606 wake_up_all(&queue->ring->stop_waitq);
607 }
608}
609
610static void fuse_uring_prepare_cancel(struct io_uring_cmd *cmd, int issue_flags,
611 struct fuse_ring_ent *ring_ent)
612{
613 uring_cmd_set_ring_ent(cmd, ring_ent);
614 io_uring_cmd_mark_cancelable(cmd, issue_flags);
615}
616
617/*
618 * Checks for errors and stores it into the request
619 */
620static int fuse_uring_out_header_has_err(struct fuse_out_header *oh,
621 struct fuse_req *req)
622{
623 int err;
624
625 err = -EINVAL;
626 if (oh->unique == 0) {
627 /* Not supported through io-uring yet */
628 pr_warn_once("notify through fuse-io-uring not supported\n");
629 goto err;
630 }
631
632 if (oh->error <= -ERESTARTSYS || oh->error > 0)
633 goto err;
634
635 if (oh->error) {
636 err = oh->error;
637 goto err;
638 }
639
640 err = -ENOENT;
641 if ((oh->unique & ~FUSE_INT_REQ_BIT) != req->in.h.unique) {
642 pr_warn_ratelimited("unique mismatch, expected: %llu got %llu\n",
643 req->in.h.unique,
644 oh->unique & ~FUSE_INT_REQ_BIT);
645 goto err;
646 }
647
648 /*
649 * Is it an interrupt reply ID?
650 * XXX: Not supported through fuse-io-uring yet, it should not even
651 * find the request - should not happen.
652 */
653 WARN_ON_ONCE(oh->unique & FUSE_INT_REQ_BIT);
654
655 err = 0;
656err:
657 return err;
658}
659
660static int ring_header_type_offset(enum fuse_uring_header_type type)
661{
662 switch (type) {
663 case FUSE_URING_HEADER_IN_OUT:
664 return 0;
665 case FUSE_URING_HEADER_OP:
666 return offsetof(struct fuse_uring_req_header, op_in);
667 case FUSE_URING_HEADER_RING_ENT:
668 return offsetof(struct fuse_uring_req_header, ring_ent_in_out);
669 default:
670 WARN_ONCE(1, "Invalid header type: %d\n", type);
671 return -EINVAL;
672 }
673}
674
675static int copy_header_to_ring(struct fuse_ring_ent *ent,
676 enum fuse_uring_header_type type,
677 const void *header, size_t header_size)
678{
679 int offset = ring_header_type_offset(type);
680 void __user *ring;
681
682 if (offset < 0)
683 return offset;
684
685 ring = (void __user *)ent->headers + offset;
686
687 if (copy_to_user(ring, header, header_size)) {
688 pr_info_ratelimited("Copying header to ring failed.\n");
689 return -EFAULT;
690 }
691
692 return 0;
693}
694
695static int copy_header_from_ring(struct fuse_ring_ent *ent,
696 enum fuse_uring_header_type type, void *header,
697 size_t header_size)
698{
699 int offset = ring_header_type_offset(type);
700 const void __user *ring;
701
702 if (offset < 0)
703 return offset;
704
705 ring = (void __user *)ent->headers + offset;
706
707 if (copy_from_user(header, ring, header_size)) {
708 pr_info_ratelimited("Copying header from ring failed.\n");
709 return -EFAULT;
710 }
711
712 return 0;
713}
714
715static int fuse_uring_import_payload(struct fuse_ring_ent *ent, int dir,
716 struct iov_iter *iter,
717 unsigned int issue_flags)
718{
719 void __user *base = ent->payload.iov_base;
720 size_t len = ent->payload.iov_len;
721 int err = 0;
722
723 if (!base) {
724 memset(iter, 0, sizeof(*iter));
725 return 0;
726 }
727
728 if (bufpool_registered(ent->queue))
729 err = io_uring_cmd_import_fixed((u64)(uintptr_t)base, len, dir,
730 iter, ent->cmd, issue_flags);
731 else
732 err = import_ubuf(dir, base, len, iter);
733
734 if (err)
735 pr_info_ratelimited("fuse: Import of user buffer failed\n");
736
737 return err;
738}
739
740static int setup_fuse_copy_state(struct fuse_copy_state *cs,
741 struct fuse_req *req,
742 struct fuse_ring_ent *ent, int dir,
743 struct iov_iter *iter,
744 unsigned int issue_flags)
745{
746 int err;
747
748 err = fuse_uring_import_payload(ent, dir, iter, issue_flags);
749 if (err)
750 return err;
751
752 fuse_copy_init(cs, dir == ITER_DEST, iter);
753
754 if (ent->zero_copied)
755 cs->skip_folio_copy = true;
756
757 cs->is_uring = true;
758 cs->req = req;
759
760 return 0;
761}
762
763static int fuse_uring_copy_from_ring(struct fuse_req *req,
764 struct fuse_ring_ent *ent,
765 unsigned int issue_flags)
766{
767 struct fuse_copy_state cs;
768 struct fuse_args *args = req->args;
769 struct iov_iter iter;
770 int err;
771 struct fuse_uring_ent_in_out ring_in_out;
772
773 err = copy_header_from_ring(ent, FUSE_URING_HEADER_RING_ENT,
774 &ring_in_out, sizeof(ring_in_out));
775 if (err)
776 return err;
777
778 err = setup_fuse_copy_state(&cs, req, ent, ITER_SOURCE, &iter,
779 issue_flags);
780 if (err)
781 return err;
782
783 err = fuse_copy_out_args(&cs, args, ring_in_out.payload_sz);
784 fuse_copy_finish(&cs);
785 return err;
786}
787
788static void fuse_zero_copy_release(void *priv)
789{
790 struct fuse_zero_copy_bvs *zc_bvs = priv;
791 unsigned int i;
792
793 for (i = 0; i < zc_bvs->nr_bvs; i++)
794 folio_put(page_folio(zc_bvs->bvs[i].bv_page));
795
796 kvfree(zc_bvs);
797}
798
799static int fuse_uring_set_up_zero_copy(struct fuse_ring_ent *ent,
800 struct fuse_req *req,
801 unsigned int issue_flags)
802{
803 struct fuse_args_pages *ap;
804 int err, i, ddir = 0;
805 struct fuse_zero_copy_bvs *zc_bvs;
806 struct bio_vec *bvs;
807
808 /* out_pages indicates a read, in_pages indicates a write */
809 if (req->args->out_pages)
810 ddir |= IO_BUF_DEST;
811 if (req->args->in_pages)
812 ddir |= IO_BUF_SOURCE;
813
814 ap = container_of(req->args, typeof(*ap), args);
815
816 zc_bvs = kvmalloc_flex(*zc_bvs, bvs, ap->num_folios,
817 GFP_KERNEL_ACCOUNT);
818 if (!zc_bvs)
819 return -ENOMEM;
820
821 zc_bvs->nr_bvs = ap->num_folios;
822 bvs = zc_bvs->bvs;
823 for (i = 0; i < ap->num_folios; i++) {
824 bvs[i].bv_page = folio_page(ap->folios[i], 0);
825 bvs[i].bv_offset = ap->descs[i].offset;
826 bvs[i].bv_len = ap->descs[i].length;
827 folio_get(ap->folios[i]);
828 }
829
830 err = io_buffer_register_bvec(ent->cmd, bvs, ap->num_folios,
831 fuse_zero_copy_release, zc_bvs,
832 ddir, ent->zero_copy_index,
833 issue_flags);
834 if (err) {
835 fuse_zero_copy_release(zc_bvs);
836 return err;
837 }
838
839 ent->zero_copied = true;
840
841 return 0;
842}
843
844/*
845 * Copy data from the req to the ring buffer
846 */
847static int fuse_uring_args_to_ring(struct fuse_req *req,
848 struct fuse_ring_ent *ent,
849 unsigned int issue_flags)
850{
851 struct fuse_copy_state cs;
852 struct fuse_args *args = req->args;
853 struct fuse_in_arg *in_args = args->in_args;
854 int num_args = args->in_numargs;
855 int err;
856 struct iov_iter iter;
857 struct fuse_uring_ent_in_out ent_in_out = {
858 .flags = 0,
859 .commit_id = req->in.h.unique,
860 };
861
862 if (can_zero_copy_req(ent, req)) {
863 ent_in_out.flags |= FUSE_URING_ENT_ZERO_COPY;
864 err = fuse_uring_set_up_zero_copy(ent, req, issue_flags);
865 if (err)
866 return err;
867 }
868
869 err = setup_fuse_copy_state(&cs, req, ent, ITER_DEST, &iter,
870 issue_flags);
871 if (err)
872 return err;
873
874 if (num_args > 0) {
875 /*
876 * Expectation is that the first argument is the per op header.
877 * Some op code have that as zero size.
878 */
879 if (args->in_args[0].size > 0) {
880 err = copy_header_to_ring(ent, FUSE_URING_HEADER_OP,
881 in_args->value,
882 in_args->size);
883 if (err)
884 return err;
885 }
886 in_args++;
887 num_args--;
888 }
889
890 /* copy the payload */
891 err = fuse_copy_args(&cs, num_args, args->in_pages,
892 (struct fuse_arg *)in_args, 0);
893 fuse_copy_finish(&cs);
894 if (err) {
895 pr_info_ratelimited("%s fuse_copy_args failed\n", __func__);
896 return err;
897 }
898
899 ent_in_out.payload_sz = cs.ring.copied_sz;
900 /*
901 * on a zero-copied write the pages are registered for the server to
902 * read via a fixed-buffer op rather than copied into the payload
903 * buffer, so copied_sz does not account for it. The server still needs
904 * the total inbound size to know how many bytes to read from the
905 * registered buffer, so add the page arg (always the last in-arg) back
906 * in
907 */
908 if (cs.skip_folio_copy && args->in_pages)
909 ent_in_out.payload_sz +=
910 args->in_args[args->in_numargs - 1].size;
911
912 if (bufpool_enabled(ent->queue) && ent->payload.iov_base)
913 ent_in_out.offset =
914 (uintptr_t)ent->payload.iov_base - ent->queue->bufpool->base_uaddr;
915
916 return copy_header_to_ring(ent, FUSE_URING_HEADER_RING_ENT,
917 &ent_in_out, sizeof(ent_in_out));
918}
919
920static int fuse_uring_copy_to_ring(struct fuse_ring_ent *ent,
921 struct fuse_req *req,
922 unsigned int issue_flags)
923{
924 struct fuse_ring_queue *queue = ent->queue;
925 struct fuse_in_header in_header;
926 int err;
927
928 err = -EIO;
929 if (WARN_ON(ent->state != FRRS_FUSE_REQ)) {
930 pr_err("qid=%d ring-req=%p invalid state %d on send\n",
931 queue->qid, ent, ent->state);
932 return err;
933 }
934
935 err = -EINVAL;
936 if (WARN_ON(req->in.h.unique == 0))
937 return err;
938
939 /* copy the request */
940 err = fuse_uring_args_to_ring(req, ent, issue_flags);
941 if (unlikely(err)) {
942 pr_info_ratelimited("Copy to ring failed: %d\n", err);
943 return err;
944 }
945
946 /* copy fuse_in_header */
947 in_header = req->in.h;
948 return copy_header_to_ring(ent, FUSE_URING_HEADER_IN_OUT, &in_header,
949 sizeof(in_header));
950}
951
952static bool fuse_uring_req_has_copyable_payload(struct fuse_ring_ent *ent,
953 struct fuse_req *req)
954{
955 struct fuse_args *args = req->args;
956
957 if (!can_zero_copy_req(ent, req))
958 return args->in_numargs > 1 || args->out_numargs;
959
960 /*
961 * the asymmetry between in_numargs > 2 and out_numargs > 1 is because
962 * the per-op header is extracted before fuse_copy_args() for inargs but
963 * not for outargs
964 */
965 if ((args->in_numargs > 1) && (!args->in_pages || args->in_numargs > 2))
966 return true;
967 if (args->out_numargs && (!args->out_pages || args->out_numargs > 1))
968 return true;
969
970 return false;
971}
972
973static int fuse_uring_select_buffer(struct fuse_ring_ent *ent)
974{
975 struct fuse_ring_queue *queue = ent->queue;
976 struct fuse_bufpool *pool = queue->bufpool;
977 unsigned int id;
978
979 lockdep_assert_held(&queue->lock);
980
981 id = find_first_bit(pool->free_map, pool->nr_bufs);
982 if (id >= pool->nr_bufs)
983 return -ENOBUFS;
984
985 WARN_ON_ONCE(ent->payload.iov_base);
986 __clear_bit(id, pool->free_map);
987
988 ent->buf_id = id;
989 ent->payload.iov_base =
990 (void __user *)(pool->base_uaddr + id * pool->buf_size);
991 ent->payload.iov_len = pool->buf_size;
992
993 return 0;
994}
995
996static void fuse_uring_recycle_buffer(struct fuse_ring_ent *ent)
997{
998 struct iovec *ent_payload = &ent->payload;
999 struct fuse_ring_queue *queue = ent->queue;
1000 struct fuse_bufpool *pool;
1001
1002 lockdep_assert_held(&queue->lock);
1003
1004 if (!bufpool_enabled(queue) || !ent_payload->iov_base)
1005 return;
1006
1007 pool = queue->bufpool;
1008
1009 /* a buffer should never be recycled twice */
1010 WARN_ON_ONCE(test_bit(ent->buf_id, pool->free_map));
1011 __set_bit(ent->buf_id, pool->free_map);
1012
1013 memset(ent_payload, 0, sizeof(*ent_payload));
1014 ent->buf_id = 0;
1015}
1016
1017static int fuse_uring_next_req_update_buffer(struct fuse_ring_ent *ent,
1018 struct fuse_req *req)
1019{
1020 bool buffer_selected;
1021 bool has_payload;
1022
1023 if (!bufpool_enabled(ent->queue))
1024 return 0;
1025
1026 buffer_selected = !!ent->payload.iov_base;
1027 has_payload = fuse_uring_req_has_copyable_payload(ent, req);
1028
1029 if (has_payload && !buffer_selected)
1030 return fuse_uring_select_buffer(ent);
1031
1032 if (!has_payload && buffer_selected)
1033 fuse_uring_recycle_buffer(ent);
1034
1035 return 0;
1036}
1037
1038static int fuse_uring_prep_buffer(struct fuse_ring_ent *ent,
1039 struct fuse_req *req)
1040{
1041 if (!bufpool_enabled(ent->queue))
1042 return 0;
1043
1044 /* no payload to copy, can skip selecting a buffer */
1045 if (!fuse_uring_req_has_copyable_payload(ent, req))
1046 return 0;
1047
1048 return fuse_uring_select_buffer(ent);
1049}
1050
1051static int fuse_uring_prepare_send(struct fuse_ring_ent *ent,
1052 struct fuse_req *req,
1053 unsigned int issue_flags)
1054{
1055 int err;
1056
1057 err = fuse_uring_copy_to_ring(ent, req, issue_flags);
1058 if (!err) {
1059 set_bit(FR_SENT, &req->flags);
1060 trace_fuse_request_sent(req);
1061 } else {
1062 /*
1063 * Copying the request failed. Remove the entry from the
1064 * ent_w_req_queue list and terminate the request
1065 */
1066 spin_lock(&ent->queue->lock);
1067 list_del_init(&ent->list);
1068 ent->state = FRRS_INVALID;
1069 spin_unlock(&ent->queue->lock);
1070
1071 fuse_uring_req_end(ent, req, err, issue_flags);
1072 }
1073
1074 return err;
1075}
1076
1077/* Used to find the request on SQE commit */
1078static void fuse_uring_add_to_pq(struct fuse_ring_ent *ent)
1079{
1080 struct fuse_ring_queue *queue = ent->queue;
1081 struct fuse_pqueue *fpq = &queue->fpq;
1082 unsigned int hash;
1083 struct fuse_req *req = ent->fuse_req;
1084
1085 req->ring_entry = ent;
1086 hash = fuse_req_hash(req->in.h.unique);
1087 list_move_tail(&req->list, &fpq->processing[hash]);
1088}
1089
1090/*
1091 * Make a ring entry available for fuse_req assignment
1092 */
1093static void fuse_uring_ent_avail(struct fuse_ring_ent *ent,
1094 struct fuse_ring_queue *queue)
1095{
1096 WARN_ON_ONCE(!ent->cmd);
1097 list_move(&ent->list, &queue->ent_avail_queue);
1098 ent->state = FRRS_AVAILABLE;
1099}
1100
1101/*
1102 * Assign a fuse queue entry to the given entry
1103 */
1104static void fuse_uring_add_req_to_ring_ent(struct fuse_ring_ent *ent,
1105 struct fuse_req *req)
1106{
1107 struct fuse_ring_queue *queue = ent->queue;
1108
1109 lockdep_assert_held(&queue->lock);
1110
1111 if (WARN_ON_ONCE(ent->state != FRRS_AVAILABLE &&
1112 ent->state != FRRS_COMMIT)) {
1113 pr_warn("%s qid=%d state=%d\n", __func__, ent->queue->qid,
1114 ent->state);
1115 }
1116
1117 clear_bit(FR_PENDING, &req->flags);
1118
1119 /* Until fuse_uring_add_to_pq() the req is not attached to any list */
1120 list_del_init(&req->list);
1121
1122 ent->fuse_req = req;
1123 ent->state = FRRS_FUSE_REQ;
1124 list_move_tail(&ent->list, &queue->ent_w_req_queue);
1125}
1126
1127/* Fetch the next fuse request if available */
1128static struct fuse_req *fuse_uring_ent_assign_req(struct fuse_ring_ent *ent)
1129 __must_hold(&queue->lock)
1130{
1131 struct fuse_req *req;
1132 struct fuse_ring_queue *queue = ent->queue;
1133 struct list_head *req_queue = &queue->fuse_req_queue;
1134
1135 lockdep_assert_held(&queue->lock);
1136
1137 /* get and assign the next entry while it is still holding the lock */
1138 req = list_first_entry_or_null(req_queue, struct fuse_req, list);
1139 if (!req || fuse_uring_next_req_update_buffer(ent, req)) {
1140 fuse_uring_recycle_buffer(ent);
1141 return NULL;
1142 }
1143
1144 fuse_uring_add_req_to_ring_ent(ent, req);
1145 return req;
1146}
1147
1148/*
1149 * Read data from the ring buffer, which user space has written to
1150 * This is comparible with handling of classical write(/dev/fuse).
1151 * Also make the ring request available again for new fuse requests.
1152 */
1153static void fuse_uring_commit(struct fuse_ring_ent *ent, struct fuse_req *req,
1154 unsigned int issue_flags)
1155{
1156 struct fuse_out_header out_header;
1157 ssize_t err = -EFAULT;
1158
1159 if (copy_header_from_ring(ent, FUSE_URING_HEADER_IN_OUT, &out_header,
1160 sizeof(out_header)))
1161 goto out;
1162 req->out.h = out_header;
1163
1164 err = fuse_uring_out_header_has_err(&req->out.h, req);
1165 if (err) {
1166 /* req->out.h.error already set */
1167 goto out;
1168 }
1169
1170 err = fuse_uring_copy_from_ring(req, ent, issue_flags);
1171out:
1172 fuse_uring_req_end(ent, req, err, issue_flags);
1173}
1174
1175/*
1176 * Get the next fuse req.
1177 *
1178 * Returns true if the next fuse request has been assigned to the ent.
1179 * Else, there is no next fuse request and this returns false.
1180 */
1181static bool fuse_uring_get_next_fuse_req(struct fuse_ring_ent *ent,
1182 struct fuse_ring_queue *queue,
1183 unsigned int issue_flags)
1184{
1185 int err;
1186 struct fuse_req *req;
1187
1188retry:
1189 spin_lock(&queue->lock);
1190 fuse_uring_ent_avail(ent, queue);
1191 req = fuse_uring_ent_assign_req(ent);
1192 spin_unlock(&queue->lock);
1193
1194 if (req) {
1195 err = fuse_uring_prepare_send(ent, req, issue_flags);
1196 if (err)
1197 goto retry;
1198 }
1199
1200 return req != NULL;
1201}
1202
1203static int fuse_ring_ent_set_commit(struct fuse_ring_ent *ent)
1204{
1205 struct fuse_ring_queue *queue = ent->queue;
1206
1207 lockdep_assert_held(&queue->lock);
1208
1209 if (WARN_ON_ONCE(ent->state != FRRS_USERSPACE))
1210 return -EIO;
1211
1212 ent->state = FRRS_COMMIT;
1213 list_move(&ent->list, &queue->ent_commit_queue);
1214
1215 return 0;
1216}
1217
1218static void fuse_uring_send(struct fuse_ring_ent *ent, struct io_uring_cmd *cmd,
1219 ssize_t ret, unsigned int issue_flags)
1220{
1221 struct fuse_ring_queue *queue = ent->queue;
1222
1223 spin_lock(&queue->lock);
1224 ent->state = FRRS_USERSPACE;
1225 list_move_tail(&ent->list, &queue->ent_in_userspace);
1226 ent->cmd = NULL;
1227 fuse_uring_add_to_pq(ent);
1228 spin_unlock(&queue->lock);
1229
1230 io_uring_cmd_done(cmd, ret, issue_flags);
1231}
1232
1233/* FUSE_URING_CMD_COMMIT_AND_FETCH handler */
1234static int fuse_uring_commit_fetch(struct io_uring_cmd *cmd, int issue_flags,
1235 struct fuse_chan *fch)
1236{
1237 const struct fuse_uring_cmd_req *cmd_req = io_uring_sqe128_cmd(cmd->sqe,
1238 struct fuse_uring_cmd_req);
1239 struct fuse_ring_ent *ent;
1240 int err;
1241 struct fuse_ring *ring = fch->ring;
1242 struct fuse_ring_queue *queue;
1243 uint64_t commit_id = READ_ONCE(cmd_req->commit_id);
1244 unsigned int qid = READ_ONCE(cmd_req->qid);
1245 struct fuse_pqueue *fpq;
1246 struct fuse_req *req;
1247
1248 err = -ENOTCONN;
1249 if (!ring)
1250 return err;
1251
1252 if (qid >= ring->nr_queues)
1253 return -EINVAL;
1254
1255 queue = READ_ONCE(ring->queues[qid]);
1256 if (!queue)
1257 return err;
1258 fpq = &queue->fpq;
1259
1260 if (!READ_ONCE(fch->connected))
1261 return err;
1262
1263 spin_lock(&queue->lock);
1264 if (unlikely(queue->stopped)) {
1265 spin_unlock(&queue->lock);
1266 return err;
1267 }
1268
1269 if (!fuse_uring_cmd_index_ok(cmd, queue)) {
1270 spin_unlock(&queue->lock);
1271 return -EINVAL;
1272 }
1273
1274 /* Find a request based on the unique ID of the fuse request
1275 * This should get revised, as it needs a hash calculation and list
1276 * search. And full struct fuse_pqueue is needed (memory overhead).
1277 * As well as the link from req to ring_ent.
1278 */
1279 req = fuse_request_find(fpq, commit_id);
1280 err = -ENOENT;
1281 if (!req) {
1282 pr_info("qid=%d commit_id %llu not found\n", queue->qid,
1283 commit_id);
1284 spin_unlock(&queue->lock);
1285 return err;
1286 }
1287 list_del_init(&req->list);
1288 ent = req->ring_entry;
1289 req->ring_entry = NULL;
1290
1291 err = fuse_ring_ent_set_commit(ent);
1292 if (err != 0) {
1293 pr_info_ratelimited("qid=%d commit_id %llu state %d",
1294 queue->qid, commit_id, ent->state);
1295 fuse_uring_recycle_buffer(ent);
1296 spin_unlock(&queue->lock);
1297 /*
1298 * Unregister any zero copyable pages since ent->cmd is null
1299 * when it hits fuse_uring_req_end() in this path
1300 */
1301 zero_copy_unregister(cmd, ent, issue_flags);
1302 fuse_uring_req_end(ent, req, err, issue_flags);
1303 return err;
1304 }
1305
1306 ent->cmd = cmd;
1307 spin_unlock(&queue->lock);
1308
1309 /* without the queue lock, as other locks are taken */
1310 fuse_uring_prepare_cancel(cmd, issue_flags, ent);
1311 fuse_uring_commit(ent, req, issue_flags);
1312
1313 /*
1314 * Fetching the next request is absolutely required as queued
1315 * fuse requests would otherwise not get processed - committing
1316 * and fetching is done in one step vs legacy fuse, which has separated
1317 * read (fetch request) and write (commit result).
1318 *
1319 * If there is no next request or if all buffers are busy (if using a
1320 * bufpool), the cmd is not returned to userspace. The entry is left
1321 * available and the cmd only returns to userspace when there's a
1322 * next request and an available buffer.
1323 */
1324 if (fuse_uring_get_next_fuse_req(ent, queue, issue_flags))
1325 fuse_uring_send(ent, cmd, 0, issue_flags);
1326 return 0;
1327}
1328
1329static bool is_ring_ready(struct fuse_ring *ring, int current_qid)
1330{
1331 int qid;
1332 struct fuse_ring_queue *queue;
1333 bool ready = true;
1334
1335 for (qid = 0; qid < ring->nr_queues && ready; qid++) {
1336 if (current_qid == qid)
1337 continue;
1338
1339 queue = READ_ONCE(ring->queues[qid]);
1340 if (!queue) {
1341 ready = false;
1342 break;
1343 }
1344
1345 spin_lock(&queue->lock);
1346 if (list_empty(&queue->ent_avail_queue))
1347 ready = false;
1348 spin_unlock(&queue->lock);
1349 }
1350
1351 return ready;
1352}
1353
1354/*
1355 * fuse_uring_req_fetch command handling
1356 */
1357static int fuse_uring_do_register(struct fuse_ring_ent *ent,
1358 struct io_uring_cmd *cmd,
1359 unsigned int issue_flags)
1360{
1361 struct fuse_ring_queue *queue = ent->queue;
1362 struct fuse_ring *ring = queue->ring;
1363 struct fuse_chan *fch = ring->chan;
1364 struct fuse_iqueue *fiq = &fch->iq;
1365
1366 spin_lock(&fch->lock);
1367 /* abort teardown path is running or has run */
1368 if (!fch->connected) {
1369 spin_unlock(&fch->lock);
1370 if (atomic_dec_and_test(&ring->queue_refs))
1371 wake_up_all(&ring->stop_waitq);
1372 kfree(ent);
1373 return -ECONNABORTED;
1374 }
1375 spin_unlock(&fch->lock);
1376
1377 fuse_uring_prepare_cancel(cmd, issue_flags, ent);
1378
1379 spin_lock(&queue->lock);
1380 ent->cmd = cmd;
1381 fuse_uring_ent_avail(ent, queue);
1382 spin_unlock(&queue->lock);
1383
1384 if (!READ_ONCE(ring->ready)) {
1385 bool ready = is_ring_ready(ring, queue->qid);
1386
1387 if (ready) {
1388 WRITE_ONCE(fiq->ops, &fuse_io_uring_ops);
1389 smp_store_release(&ring->ready, true);
1390 wake_up_all(&fch->blocked_waitq);
1391 }
1392 }
1393 return 0;
1394}
1395
1396/*
1397 * sqe->addr is a ptr to an iovec array, iov[FUSE_URING_IOV_HEADERS] has the
1398 * headers, iov[FUSE_URING_IOV_PAYLOAD] the payload
1399 */
1400static int fuse_uring_get_iovec_from_sqe(const struct io_uring_sqe *sqe,
1401 struct iovec iov[FUSE_URING_IOV_SEGS])
1402{
1403 struct iovec __user *uiov = u64_to_user_ptr(READ_ONCE(sqe->addr));
1404 struct iov_iter iter;
1405 ssize_t ret;
1406
1407 if (sqe->len != FUSE_URING_IOV_SEGS)
1408 return -EINVAL;
1409
1410 /*
1411 * Direction for buffer access will actually be READ and WRITE,
1412 * using write for the import should include READ access as well.
1413 */
1414 ret = import_iovec(WRITE, uiov, FUSE_URING_IOV_SEGS,
1415 FUSE_URING_IOV_SEGS, &iov, &iter);
1416 if (ret < 0)
1417 return ret;
1418
1419 return 0;
1420}
1421
1422static struct fuse_ring_ent *
1423fuse_uring_create_ring_ent(struct io_uring_cmd *cmd,
1424 struct fuse_ring_queue *queue)
1425{
1426 const struct fuse_uring_cmd_req *cmd_req =
1427 io_uring_sqe128_cmd(cmd->sqe, struct fuse_uring_cmd_req);
1428 struct fuse_ring *ring = queue->ring;
1429 struct fuse_ring_ent *ent;
1430 struct iovec iov[FUSE_URING_IOV_SEGS];
1431 struct iovec *headers, *payload;
1432 unsigned int zero_copy_index;
1433
1434 int err;
1435
1436 err = fuse_uring_get_iovec_from_sqe(cmd->sqe, iov);
1437 if (err) {
1438 pr_info_ratelimited("Failed to get iovec from sqe, err=%d\n",
1439 err);
1440 return ERR_PTR(err);
1441 }
1442
1443 zero_copy_index = READ_ONCE(cmd_req->ent_zero_copy_buf_index);
1444 if (zero_copy_index && !queue->zero_copy)
1445 return ERR_PTR(-EINVAL);
1446
1447 err = -EINVAL;
1448 headers = &iov[FUSE_URING_IOV_HEADERS];
1449 if (headers->iov_len < sizeof(struct fuse_uring_req_header)) {
1450 pr_info_ratelimited("Invalid header len %zu\n", headers->iov_len);
1451 return ERR_PTR(err);
1452 }
1453
1454 payload = &iov[FUSE_URING_IOV_PAYLOAD];
1455
1456 spin_lock(&queue->lock);
1457 if (bufpool_enabled(queue)) {
1458 if (payload->iov_base || payload->iov_len ||
1459 !fuse_uring_cmd_index_ok(cmd, queue)) {
1460 spin_unlock(&queue->lock);
1461 return ERR_PTR(err);
1462 }
1463 } else {
1464 if (payload->iov_len < ring->max_payload_sz) {
1465 spin_unlock(&queue->lock);
1466 pr_info_ratelimited("Invalid req payload len %zu\n",
1467 payload->iov_len);
1468 return ERR_PTR(err);
1469 }
1470 if (queue->zero_copy) {
1471 spin_unlock(&queue->lock);
1472 pr_info_ratelimited("Can only use zero copy with bufpools\n");
1473 return ERR_PTR(err);
1474 }
1475 queue->payload_mode = FUSE_PAYLOAD_PER_ENT;
1476 }
1477 spin_unlock(&queue->lock);
1478
1479 err = -ENOMEM;
1480 ent = kzalloc_obj(*ent, GFP_KERNEL_ACCOUNT);
1481 if (!ent)
1482 return ERR_PTR(err);
1483
1484 INIT_LIST_HEAD(&ent->list);
1485
1486 ent->queue = queue;
1487 ent->headers = headers->iov_base;
1488 if (queue->payload_mode == FUSE_PAYLOAD_PER_ENT)
1489 ent->payload = *payload;
1490 ent->zero_copy_index = zero_copy_index;
1491
1492 atomic_inc(&ring->queue_refs);
1493 return ent;
1494}
1495
1496/*
1497 * Register header and payload buffer with the kernel and puts the
1498 * entry as "ready to get fuse requests" on the queue
1499 */
1500static int fuse_uring_register(struct io_uring_cmd *cmd,
1501 unsigned int issue_flags, struct fuse_chan *fch)
1502{
1503 const struct fuse_uring_cmd_req *cmd_req = io_uring_sqe128_cmd(cmd->sqe,
1504 struct fuse_uring_cmd_req);
1505 struct fuse_ring *ring = smp_load_acquire(&fch->ring);
1506 struct fuse_ring_queue *queue;
1507 struct fuse_ring_ent *ent;
1508 unsigned int qid = READ_ONCE(cmd_req->qid);
1509
1510 if (!ring)
1511 return -EINVAL;
1512
1513 if (qid >= ring->nr_queues) {
1514 pr_info_ratelimited("fuse: Invalid ring qid %u\n", qid);
1515 return -EINVAL;
1516 }
1517
1518 queue = READ_ONCE(ring->queues[qid]);
1519 if (!queue) {
1520 queue = fuse_uring_create_queue(ring, qid, false, false);
1521 if (IS_ERR(queue))
1522 return PTR_ERR(queue);
1523 }
1524
1525 /*
1526 * The created queue above does not need to be destructed in
1527 * case of entry errors below, will be done at ring destruction time.
1528 */
1529
1530 ent = fuse_uring_create_ring_ent(cmd, queue);
1531 if (IS_ERR(ent))
1532 return PTR_ERR(ent);
1533
1534 return fuse_uring_do_register(ent, cmd, issue_flags);
1535}
1536
1537static int fuse_uring_add_queue(struct io_uring_cmd *cmd, struct fuse_chan *fch)
1538{
1539 const struct fuse_uring_cmd_req *cmd_req =
1540 io_uring_sqe128_cmd(cmd->sqe, struct fuse_uring_cmd_req);
1541 struct fuse_ring *ring = smp_load_acquire(&fch->ring);
1542 unsigned int qid = READ_ONCE(cmd_req->qid);
1543 uint64_t flags = READ_ONCE(cmd_req->flags);
1544 struct fuse_ring_queue *queue;
1545 bool zero_copy = flags & FUSE_URING_ZERO_COPY;
1546
1547 if (!ring)
1548 return -EINVAL;
1549
1550 if (qid >= ring->nr_queues) {
1551 pr_info_ratelimited("fuse: Invalid ring qid %u\n", qid);
1552 return -EINVAL;
1553 }
1554
1555 if (flags & ~FUSE_URING_ADD_QUEUE_FLAGS)
1556 return -EINVAL;
1557
1558 if (zero_copy && !capable(CAP_SYS_ADMIN))
1559 return -EPERM;
1560
1561 queue = fuse_uring_create_queue(ring, qid, zero_copy, true);
1562 if (IS_ERR(queue))
1563 return PTR_ERR(queue);
1564
1565 return 0;
1566}
1567
1568static int fuse_uring_add_bufpool(struct io_uring_cmd *cmd,
1569 struct fuse_chan *fch)
1570{
1571 const struct fuse_uring_cmd_req *cmd_req =
1572 io_uring_sqe128_cmd(cmd->sqe, struct fuse_uring_cmd_req);
1573 unsigned int qid = READ_ONCE(cmd_req->qid);
1574 uint64_t flags = READ_ONCE(cmd_req->flags);
1575 /* paired with the smp_store_release() in fuse_uring_create */
1576 struct fuse_ring *ring = smp_load_acquire(&fch->ring);
1577 struct fuse_ring_queue *queue;
1578 struct fuse_bufpool *pool;
1579 uintptr_t pool_uaddr;
1580 unsigned int pool_len, nr_bufs;
1581 size_t pool_size, buf_size;
1582 bool registered = cmd->flags & IORING_URING_CMD_FIXED;
1583
1584 if (!ring || qid >= ring->nr_queues || flags)
1585 return -EINVAL;
1586
1587 /* reserved for future use, must be zero */
1588 if (READ_ONCE(cmd_req->bufpool.reserved))
1589 return -EINVAL;
1590
1591 /* Pairs with smp_store_release() in fuse_uring_create_queue() */
1592 queue = smp_load_acquire(&ring->queues[qid]);
1593 if (!queue)
1594 return -EINVAL;
1595
1596 pool_uaddr = READ_ONCE(cmd_req->bufpool.uaddr);
1597 pool_len = READ_ONCE(cmd_req->bufpool.len);
1598
1599 /* each buffer holds the max payload size */
1600 buf_size = queue->ring->max_payload_sz;
1601
1602 nr_bufs = pool_len / buf_size;
1603 if (!nr_bufs)
1604 return -EINVAL;
1605
1606 pool_size = struct_size(pool, free_map, BITS_TO_LONGS(nr_bufs));
1607 pool = kzalloc(pool_size, GFP_KERNEL_ACCOUNT);
1608 if (!pool)
1609 return -ENOMEM;
1610
1611 pool->base_uaddr = pool_uaddr;
1612 pool->buf_size = buf_size;
1613 pool->nr_bufs = nr_bufs;
1614 /* all buffers are free */
1615 bitmap_set(pool->free_map, 0, nr_bufs);
1616
1617 /*
1618 * A registered bufpool is reached through an io_uring fixed buffer, so
1619 * the pool is registered iff this command was submitted with
1620 * IORING_URING_CMD_FIXED. The registered buffer index is taken from
1621 * sqe->buf_index.
1622 */
1623 if (registered) {
1624 pool->registered = true;
1625 pool->registered_index = READ_ONCE(cmd->sqe->buf_index);
1626 }
1627
1628 spin_lock(&queue->lock);
1629 if (queue->payload_mode != FUSE_PAYLOAD_UNSET) {
1630 spin_unlock(&queue->lock);
1631 kfree(pool);
1632 return -EINVAL;
1633 }
1634 queue->bufpool = pool;
1635 queue->payload_mode = FUSE_PAYLOAD_BUFPOOL;
1636 spin_unlock(&queue->lock);
1637
1638 return 0;
1639}
1640
1641/*
1642 * Entry function from io_uring to handle the given passthrough command
1643 * (op code IORING_OP_URING_CMD)
1644 */
1645int fuse_uring_cmd(struct io_uring_cmd *cmd, unsigned int issue_flags)
1646{
1647 struct fuse_dev *fud;
1648 struct fuse_chan *fch;
1649 u32 cmd_op = cmd->cmd_op;
1650 int err;
1651
1652 if ((unlikely(issue_flags & IO_URING_F_CANCEL))) {
1653 fuse_uring_cancel(cmd, issue_flags);
1654 return 0;
1655 }
1656
1657 /* This extra SQE size holds struct fuse_uring_cmd_req */
1658 if (!(issue_flags & IO_URING_F_SQE128))
1659 return -EINVAL;
1660
1661 fud = fuse_get_dev(cmd->file);
1662 if (IS_ERR(fud)) {
1663 pr_info_ratelimited("No fuse device found\n");
1664 return PTR_ERR(fud);
1665 }
1666 fch = fud->chan;
1667
1668 /*
1669 * The ring is sized from values negotiated by FUSE_INIT
1670 *
1671 * Pairs with smp_store_release() in fuse_chan_set_initialized()
1672 */
1673 if (!smp_load_acquire(&fch->initialized))
1674 return -EAGAIN;
1675
1676 if (fch->abort_with_err)
1677 return -ECONNABORTED;
1678 if (!fch->connected)
1679 return -ENOTCONN;
1680
1681 /* Once a connection has io-uring enabled on it, it can't be disabled */
1682 if (!enable_uring && !fch->io_uring) {
1683 pr_info_ratelimited("fuse-io-uring is disabled by module parameter\n");
1684 return -EOPNOTSUPP;
1685 }
1686
1687 if (!fch->io_uring) {
1688 pr_info_ratelimited(
1689 "fuse-io-uring not enabled on this connection\n");
1690 return -EOPNOTSUPP;
1691 }
1692
1693 switch (cmd_op) {
1694 case FUSE_IO_URING_CMD_REGISTER:
1695 err = fuse_uring_register(cmd, issue_flags, fch);
1696 if (err) {
1697 pr_info_once("FUSE_IO_URING_CMD_REGISTER failed err=%d\n",
1698 err);
1699 fch->io_uring = 0;
1700 wake_up_all(&fch->blocked_waitq);
1701 return err;
1702 }
1703 break;
1704 case FUSE_IO_URING_CMD_COMMIT_AND_FETCH:
1705 err = fuse_uring_commit_fetch(cmd, issue_flags, fch);
1706 if (err) {
1707 pr_info_once("FUSE_IO_URING_COMMIT_AND_FETCH failed err=%d\n",
1708 err);
1709 return err;
1710 }
1711 break;
1712 case FUSE_IO_URING_CMD_ADD_QUEUE:
1713 err = fuse_uring_add_queue(cmd, fch);
1714 if (err)
1715 pr_info_once("FUSE_IO_URING_CMD_ADD_QUEUE failed err=%d\n",
1716 err);
1717 return err;
1718 case FUSE_IO_URING_CMD_ADD_BUFPOOL:
1719 err = fuse_uring_add_bufpool(cmd, fch);
1720 if (err)
1721 pr_info_once("FUSE_IO_URING_ADD_BUFPOOL failed err=%d\n",
1722 err);
1723 return err;
1724 default:
1725 return -EINVAL;
1726 }
1727
1728 return -EIOCBQUEUED;
1729}
1730
1731/*
1732 * This prepares and sends the ring request in fuse-uring task context.
1733 * User buffers are not mapped yet - the application does not have permission
1734 * to write to it - this has to be executed in ring task context.
1735 */
1736static void fuse_uring_send_in_task(struct io_tw_req tw_req, io_tw_token_t tw)
1737{
1738 unsigned int issue_flags = IO_URING_CMD_TASK_WORK_ISSUE_FLAGS;
1739 struct io_uring_cmd *cmd = io_uring_cmd_from_tw(tw_req);
1740 struct fuse_ring_ent *ent = uring_cmd_to_ring_ent(cmd);
1741 struct fuse_ring_queue *queue = ent->queue;
1742 int err;
1743
1744 if (!tw.cancel) {
1745 err = fuse_uring_prepare_send(ent, ent->fuse_req, issue_flags);
1746 if (err) {
1747 if (!fuse_uring_get_next_fuse_req(ent, queue,
1748 issue_flags))
1749 return;
1750 err = 0;
1751 }
1752 fuse_uring_send(ent, cmd, err, issue_flags);
1753 } else {
1754 err = -ECANCELED;
1755
1756 spin_lock(&queue->lock);
1757 list_del_init(&ent->list);
1758 fuse_uring_recycle_buffer(ent);
1759 spin_unlock(&queue->lock);
1760
1761 io_uring_cmd_done(cmd, err, issue_flags);
1762
1763 fuse_uring_req_end(ent, ent->fuse_req, err, issue_flags);
1764 kfree(ent);
1765 if (atomic_dec_and_test(&queue->ring->queue_refs))
1766 wake_up_all(&queue->ring->stop_waitq);
1767 }
1768}
1769
1770static struct fuse_ring_queue *fuse_uring_task_to_queue(struct fuse_ring *ring)
1771{
1772 unsigned int qid;
1773 struct fuse_ring_queue *queue;
1774
1775 qid = task_cpu(current);
1776
1777 if (WARN_ONCE(qid >= ring->nr_queues,
1778 "Core number (%u) exceeds nr queues (%zu)\n", qid,
1779 ring->nr_queues))
1780 qid = 0;
1781
1782 queue = READ_ONCE(ring->queues[qid]);
1783 WARN_ONCE(!queue, "Missing queue for qid %d\n", qid);
1784
1785 return queue;
1786}
1787
1788static void fuse_uring_dispatch_ent(struct fuse_ring_ent *ent)
1789{
1790 struct io_uring_cmd *cmd = ent->cmd;
1791
1792 uring_cmd_set_ring_ent(cmd, ent);
1793 io_uring_cmd_complete_in_task(cmd, fuse_uring_send_in_task);
1794}
1795
1796/* queue a fuse request and send it if a ring entry is available */
1797void fuse_uring_queue_fuse_req(struct fuse_iqueue *fiq, struct fuse_req *req)
1798{
1799 struct fuse_ring *ring = req->chan->ring;
1800 struct fuse_ring_queue *queue;
1801 struct fuse_ring_ent *ent = NULL;
1802 int err;
1803
1804 err = -EINVAL;
1805 queue = fuse_uring_task_to_queue(ring);
1806 if (!queue)
1807 goto err;
1808
1809 fuse_request_assign_unique(fiq, req);
1810
1811 spin_lock(&queue->lock);
1812 err = -ENOTCONN;
1813 if (unlikely(queue->stopped))
1814 goto err_unlock;
1815
1816 set_bit(FR_URING, &req->flags);
1817 req->ring_queue = queue;
1818 ent = list_first_entry_or_null(&queue->ent_avail_queue,
1819 struct fuse_ring_ent, list);
1820
1821 if (!ent || fuse_uring_prep_buffer(ent, req)) {
1822 list_add_tail(&req->list, &queue->fuse_req_queue);
1823 spin_unlock(&queue->lock);
1824 return;
1825 }
1826
1827 fuse_uring_add_req_to_ring_ent(ent, req);
1828 spin_unlock(&queue->lock);
1829 fuse_uring_dispatch_ent(ent);
1830 return;
1831
1832err_unlock:
1833 spin_unlock(&queue->lock);
1834err:
1835 req->out.h.error = err;
1836 clear_bit(FR_PENDING, &req->flags);
1837 fuse_request_end(req);
1838}
1839
1840bool fuse_uring_queue_bq_req(struct fuse_req *req)
1841{
1842 struct fuse_chan *fch = req->chan;
1843 struct fuse_ring *ring = fch->ring;
1844 struct fuse_ring_queue *queue;
1845 struct fuse_ring_ent *ent = NULL;
1846
1847 queue = fuse_uring_task_to_queue(ring);
1848 if (!queue)
1849 return false;
1850
1851 spin_lock(&queue->lock);
1852 if (unlikely(queue->stopped)) {
1853 spin_unlock(&queue->lock);
1854 return false;
1855 }
1856
1857 set_bit(FR_URING, &req->flags);
1858 req->ring_queue = queue;
1859 list_add_tail(&req->list, &queue->fuse_req_bg_queue);
1860
1861 ent = list_first_entry_or_null(&queue->ent_avail_queue,
1862 struct fuse_ring_ent, list);
1863 spin_lock(&fch->bg_lock);
1864 fch->num_background++;
1865 if (fch->num_background == fch->max_background)
1866 fch->blocked = 1;
1867 fuse_uring_flush_bg(queue);
1868 spin_unlock(&fch->bg_lock);
1869
1870 /*
1871 * Due to bg_queue flush limits there might be other bg requests
1872 * in the queue that need to be handled first. Or no further req
1873 * might be available.
1874 */
1875 req = list_first_entry_or_null(&queue->fuse_req_queue, struct fuse_req,
1876 list);
1877 if (ent && req && !fuse_uring_prep_buffer(ent, req)) {
1878 fuse_uring_add_req_to_ring_ent(ent, req);
1879 spin_unlock(&queue->lock);
1880 fuse_uring_dispatch_ent(ent);
1881 } else {
1882 spin_unlock(&queue->lock);
1883 }
1884
1885 return true;
1886}
1887
1888bool fuse_uring_remove_pending_req(struct fuse_req *req)
1889{
1890 struct fuse_ring_queue *queue = req->ring_queue;
1891
1892 return fuse_remove_pending_req(req, &queue->lock);
1893}
1894
1895static const struct fuse_iqueue_ops fuse_io_uring_ops = {
1896 /* should be send over io-uring as enhancement */
1897 .send_forget = fuse_dev_queue_forget,
1898
1899 /*
1900 * could be send over io-uring, but interrupts should be rare,
1901 * no need to make the code complex
1902 */
1903 .send_interrupt = fuse_dev_queue_interrupt,
1904 .send_req = fuse_uring_queue_fuse_req,
1905};