Something went wrong. Try again.
Linux kernel mirror (for testing) git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457// 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/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
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;
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 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 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_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);
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 = 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); 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(); struct fuse_ring *res = NULL; 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; } if (fch->ring) { /* race, another thread created the ring in the meantime */ spin_unlock(&fch->lock); res = fch->ring; 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 res;}
static struct fuse_ring_queue *fuse_uring_create_queue(struct fuse_ring *ring, int qid){ 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 NULL; pq = fuse_pqueue_alloc(); 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);
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); 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(&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 = 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 setup_fuse_copy_state(struct fuse_copy_state *cs, struct fuse_ring *ring, struct fuse_req *req, struct fuse_ring_ent *ent, int dir, struct iov_iter *iter){ int err;
err = import_ubuf(dir, ent->payload, ring->max_payload_sz, iter); if (err) { pr_info_ratelimited("fuse: Import of user buffer failed\n"); return err; }
fuse_copy_init(cs, dir == ITER_DEST, iter);
cs->is_uring = true; cs->req = req;
return 0;}
static int fuse_uring_copy_from_ring(struct fuse_ring *ring, struct fuse_req *req, struct fuse_ring_ent *ent){ 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, ring, req, ent, ITER_SOURCE, &iter); if (err) return err;
err = fuse_copy_out_args(&cs, args, ring_in_out.payload_sz); fuse_copy_finish(&cs); return err;}
/* * Copy data from the req to the ring buffer */static int fuse_uring_args_to_ring(struct fuse_ring *ring, struct fuse_req *req, struct fuse_ring_ent *ent){ 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, };
err = setup_fuse_copy_state(&cs, ring, req, ent, ITER_DEST, &iter); 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; 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){ struct fuse_ring_queue *queue = ent->queue; struct fuse_ring *ring = queue->ring; 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(ring, req, ent); if (unlikely(err)) { pr_info_ratelimited("Copy to ring failed: %d\n", err); return err; }
/* copy fuse_in_header */ return copy_header_to_ring(ent, FUSE_URING_HEADER_IN_OUT, &req->in.h, sizeof(req->in.h));}
static int fuse_uring_prepare_send(struct fuse_ring_ent *ent, struct fuse_req *req){ int err;
err = fuse_uring_copy_to_ring(ent, req); if (!err) { set_bit(FR_SENT, &req->flags); trace_fuse_request_sent(req); } else { /* * Copying the request failed. Remove the entry from the * ent_w_req_queue list and terminate the request */ spin_lock(&ent->queue->lock); list_del_init(&ent->list); ent->state = FRRS_INVALID; spin_unlock(&ent->queue->lock);
fuse_uring_req_end(ent, req, err); }
return err;}
/* Used to find the request on SQE commit */static void fuse_uring_add_to_pq(struct fuse_ring_ent *ent){ struct fuse_ring_queue *queue = ent->queue; struct fuse_pqueue *fpq = &queue->fpq; unsigned int hash; struct fuse_req *req = ent->fuse_req;
req->ring_entry = ent; hash = fuse_req_hash(req->in.h.unique); list_move_tail(&req->list, &fpq->processing[hash]);}
/* * Make a ring entry available for fuse_req assignment */static void fuse_uring_ent_avail(struct fuse_ring_ent *ent, struct fuse_ring_queue *queue){ WARN_ON_ONCE(!ent->cmd); list_move(&ent->list, &queue->ent_avail_queue); ent->state = FRRS_AVAILABLE;}
/* * Assign a fuse queue entry to the given entry */static void fuse_uring_add_req_to_ring_ent(struct fuse_ring_ent *ent, struct fuse_req *req){ struct fuse_ring_queue *queue = ent->queue;
lockdep_assert_held(&queue->lock);
if (WARN_ON_ONCE(ent->state != FRRS_AVAILABLE && ent->state != FRRS_COMMIT)) { pr_warn("%s qid=%d state=%d\n", __func__, ent->queue->qid, ent->state); }
clear_bit(FR_PENDING, &req->flags);
/* Until fuse_uring_add_to_pq() the req is not attached to any list */ list_del_init(&req->list);
ent->fuse_req = req; ent->state = FRRS_FUSE_REQ; list_move_tail(&ent->list, &queue->ent_w_req_queue);}
/* Fetch the next fuse request if available */static struct fuse_req *fuse_uring_ent_assign_req(struct fuse_ring_ent *ent) __must_hold(&queue->lock){ struct fuse_req *req; struct fuse_ring_queue *queue = ent->queue; struct list_head *req_queue = &queue->fuse_req_queue;
lockdep_assert_held(&queue->lock);
/* get and assign the next entry while it is still holding the lock */ req = list_first_entry_or_null(req_queue, struct fuse_req, list); if (req) fuse_uring_add_req_to_ring_ent(ent, req);
return req;}
/* * Read data from the ring buffer, which user space has written to * This is comparible with handling of classical write(/dev/fuse). * Also make the ring request available again for new fuse requests. */static void fuse_uring_commit(struct fuse_ring_ent *ent, struct fuse_req *req, unsigned int issue_flags){ struct fuse_ring *ring = ent->queue->ring; ssize_t err = -EFAULT;
if (copy_header_from_ring(ent, FUSE_URING_HEADER_IN_OUT, &req->out.h, sizeof(req->out.h))) goto out;
err = fuse_uring_out_header_has_err(&req->out.h, req); if (err) { /* req->out.h.error already set */ goto out; }
err = fuse_uring_copy_from_ring(ring, req, ent);out: fuse_uring_req_end(ent, req, err);}
/* * Get the next fuse req. * * Returns true if the next fuse request has been assigned to the ent. * Else, there is no next fuse request and this returns false. */static bool fuse_uring_get_next_fuse_req(struct fuse_ring_ent *ent, struct fuse_ring_queue *queue){ int err; struct fuse_req *req;
retry: spin_lock(&queue->lock); fuse_uring_ent_avail(ent, queue); req = fuse_uring_ent_assign_req(ent); spin_unlock(&queue->lock);
if (req) { err = fuse_uring_prepare_send(ent, req); if (err) goto retry; }
return req != NULL;}
static int fuse_ring_ent_set_commit(struct fuse_ring_ent *ent){ struct fuse_ring_queue *queue = ent->queue;
lockdep_assert_held(&queue->lock);
if (WARN_ON_ONCE(ent->state != FRRS_USERSPACE)) return -EIO;
ent->state = FRRS_COMMIT; list_move(&ent->list, &queue->ent_commit_queue);
return 0;}
static void fuse_uring_send(struct fuse_ring_ent *ent, struct io_uring_cmd *cmd, ssize_t ret, unsigned int issue_flags){ struct fuse_ring_queue *queue = ent->queue;
spin_lock(&queue->lock); ent->state = FRRS_USERSPACE; list_move_tail(&ent->list, &queue->ent_in_userspace); ent->cmd = NULL; fuse_uring_add_to_pq(ent); spin_unlock(&queue->lock);
io_uring_cmd_done(cmd, ret, issue_flags);}
/* FUSE_URING_CMD_COMMIT_AND_FETCH handler */static int fuse_uring_commit_fetch(struct io_uring_cmd *cmd, int issue_flags, struct fuse_chan *fch){ const struct fuse_uring_cmd_req *cmd_req = io_uring_sqe128_cmd(cmd->sqe, struct fuse_uring_cmd_req); struct fuse_ring_ent *ent; int err; struct fuse_ring *ring = fch->ring; struct fuse_ring_queue *queue; uint64_t commit_id = READ_ONCE(cmd_req->commit_id); unsigned int qid = READ_ONCE(cmd_req->qid); struct fuse_pqueue *fpq; struct fuse_req *req;
err = -ENOTCONN; if (!ring) return err;
if (qid >= ring->nr_queues) return -EINVAL;
queue = ring->queues[qid]; if (!queue) return err; fpq = &queue->fpq;
if (!READ_ONCE(fch->connected)) return err;
spin_lock(&queue->lock); if (unlikely(queue->stopped)) { spin_unlock(&queue->lock); return err; }
/* Find a request based on the unique ID of the fuse request * This should get revised, as it needs a hash calculation and list * search. And full struct fuse_pqueue is needed (memory overhead). * As well as the link from req to ring_ent. */ req = fuse_request_find(fpq, commit_id); err = -ENOENT; if (!req) { pr_info("qid=%d commit_id %llu not found\n", queue->qid, commit_id); spin_unlock(&queue->lock); return err; } list_del_init(&req->list); ent = req->ring_entry; req->ring_entry = NULL;
err = fuse_ring_ent_set_commit(ent); if (err != 0) { pr_info_ratelimited("qid=%d commit_id %llu state %d", queue->qid, commit_id, ent->state); spin_unlock(&queue->lock); fuse_uring_req_end(ent, req, err); return err; }
ent->cmd = cmd; spin_unlock(&queue->lock);
/* without the queue lock, as other locks are taken */ fuse_uring_prepare_cancel(cmd, issue_flags, ent); fuse_uring_commit(ent, req, issue_flags);
/* * Fetching the next request is absolutely required as queued * fuse requests would otherwise not get processed - committing * and fetching is done in one step vs legacy fuse, which has separated * read (fetch request) and write (commit result). */ if (fuse_uring_get_next_fuse_req(ent, queue)) fuse_uring_send(ent, cmd, 0, issue_flags); return 0;}
static bool is_ring_ready(struct fuse_ring *ring, int current_qid){ int qid; struct fuse_ring_queue *queue; bool ready = true;
for (qid = 0; qid < ring->nr_queues && ready; qid++) { if (current_qid == qid) continue;
queue = ring->queues[qid]; if (!queue) { ready = false; break; }
spin_lock(&queue->lock); if (list_empty(&queue->ent_avail_queue)) ready = false; spin_unlock(&queue->lock); }
return ready;}
/* * fuse_uring_req_fetch command handling */static int fuse_uring_do_register(struct fuse_ring_ent *ent, struct io_uring_cmd *cmd, unsigned int issue_flags){ struct fuse_ring_queue *queue = ent->queue; struct fuse_ring *ring = queue->ring; struct fuse_chan *fch = ring->chan; struct fuse_iqueue *fiq = &fch->iq;
spin_lock(&fch->lock); /* abort teardown path is running or has run */ if (!fch->connected) { spin_unlock(&fch->lock); if (atomic_dec_and_test(&ring->queue_refs)) wake_up_all(&ring->stop_waitq); kfree(ent); return -ECONNABORTED; } spin_unlock(&fch->lock);
fuse_uring_prepare_cancel(cmd, issue_flags, ent);
spin_lock(&queue->lock); ent->cmd = cmd; fuse_uring_ent_avail(ent, queue); spin_unlock(&queue->lock);
if (!READ_ONCE(ring->ready)) { bool ready = is_ring_ready(ring, queue->qid);
if (ready) { WRITE_ONCE(fiq->ops, &fuse_io_uring_ops); smp_store_release(&ring->ready, true); wake_up_all(&fch->blocked_waitq); } } return 0;}
/* * sqe->addr is a ptr to an iovec array, iov[FUSE_URING_IOV_HEADERS] has the * headers, iov[FUSE_URING_IOV_PAYLOAD] the payload */static int fuse_uring_get_iovec_from_sqe(const struct io_uring_sqe *sqe, struct iovec iov[FUSE_URING_IOV_SEGS]){ struct iovec __user *uiov = u64_to_user_ptr(READ_ONCE(sqe->addr)); struct iov_iter iter; ssize_t ret;
if (sqe->len != FUSE_URING_IOV_SEGS) return -EINVAL;
/* * Direction for buffer access will actually be READ and WRITE, * using write for the import should include READ access as well. */ ret = import_iovec(WRITE, uiov, FUSE_URING_IOV_SEGS, FUSE_URING_IOV_SEGS, &iov, &iter); if (ret < 0) return ret;
return 0;}
static struct fuse_ring_ent *fuse_uring_create_ring_ent(struct io_uring_cmd *cmd, struct fuse_ring_queue *queue){ struct fuse_ring *ring = queue->ring; struct fuse_ring_ent *ent; struct iovec iov[FUSE_URING_IOV_SEGS]; struct iovec *headers, *payload; int err;
err = fuse_uring_get_iovec_from_sqe(cmd->sqe, iov); if (err) { pr_info_ratelimited("Failed to get iovec from sqe, err=%d\n", err); return ERR_PTR(err); }
err = -EINVAL; headers = &iov[FUSE_URING_IOV_HEADERS]; if (headers->iov_len < sizeof(struct fuse_uring_req_header)) { pr_info_ratelimited("Invalid header len %zu\n", headers->iov_len); return ERR_PTR(err); }
payload = &iov[FUSE_URING_IOV_PAYLOAD]; if (payload->iov_len < ring->max_payload_sz) { pr_info_ratelimited("Invalid req payload len %zu\n", payload->iov_len); return ERR_PTR(err); }
err = -ENOMEM; ent = kzalloc_obj(*ent, GFP_KERNEL_ACCOUNT); if (!ent) return ERR_PTR(err);
INIT_LIST_HEAD(&ent->list);
ent->queue = queue; ent->headers = headers->iov_base; ent->payload = payload->iov_base;
atomic_inc(&ring->queue_refs); return ent;}
/* * Register header and payload buffer with the kernel and puts the * entry as "ready to get fuse requests" on the queue */static int fuse_uring_register(struct io_uring_cmd *cmd, unsigned int issue_flags, struct fuse_chan *fch){ const struct fuse_uring_cmd_req *cmd_req = io_uring_sqe128_cmd(cmd->sqe, struct fuse_uring_cmd_req); struct fuse_ring *ring = smp_load_acquire(&fch->ring); struct fuse_ring_queue *queue; struct fuse_ring_ent *ent; int err; unsigned int qid = READ_ONCE(cmd_req->qid);
err = -ENOMEM; if (!ring) { ring = fuse_uring_create(fch); if (!ring) return err; }
if (qid >= ring->nr_queues) { pr_info_ratelimited("fuse: Invalid ring qid %u\n", qid); return -EINVAL; }
queue = ring->queues[qid]; if (!queue) { queue = fuse_uring_create_queue(ring, qid); if (!queue) return err; }
/* * The created queue above does not need to be destructed in * case of entry errors below, will be done at ring destruction time. */
ent = fuse_uring_create_ring_ent(cmd, queue); if (IS_ERR(ent)) return PTR_ERR(ent);
return fuse_uring_do_register(ent, cmd, issue_flags);}
/* * Entry function from io_uring to handle the given passthrough command * (op code IORING_OP_URING_CMD) */int fuse_uring_cmd(struct io_uring_cmd *cmd, unsigned int issue_flags){ struct fuse_dev *fud; struct fuse_chan *fch; u32 cmd_op = cmd->cmd_op; int err;
if ((unlikely(issue_flags & IO_URING_F_CANCEL))) { fuse_uring_cancel(cmd, issue_flags); return 0; }
/* This extra SQE size holds struct fuse_uring_cmd_req */ if (!(issue_flags & IO_URING_F_SQE128)) return -EINVAL;
fud = fuse_get_dev(cmd->file); if (IS_ERR(fud)) { pr_info_ratelimited("No fuse device found\n"); return PTR_ERR(fud); } fch = fud->chan;
/* Once a connection has io-uring enabled on it, it can't be disabled */ if (!enable_uring && !fch->io_uring) { pr_info_ratelimited("fuse-io-uring is disabled\n"); return -EOPNOTSUPP; }
if (fch->abort_with_err) return -ECONNABORTED; if (!fch->connected) return -ENOTCONN;
/* * fuse_uring_register() needs the ring to be initialized, * we need to know the max payload size */ if (!fch->initialized) return -EAGAIN;
switch (cmd_op) { case FUSE_IO_URING_CMD_REGISTER: err = fuse_uring_register(cmd, issue_flags, fch); if (err) { pr_info_once("FUSE_IO_URING_CMD_REGISTER failed err=%d\n", err); fch->io_uring = 0; wake_up_all(&fch->blocked_waitq); return err; } break; case FUSE_IO_URING_CMD_COMMIT_AND_FETCH: err = fuse_uring_commit_fetch(cmd, issue_flags, fch); if (err) { pr_info_once("FUSE_IO_URING_COMMIT_AND_FETCH failed err=%d\n", err); return err; } break; default: return -EINVAL; }
return -EIOCBQUEUED;}
/* * This prepares and sends the ring request in fuse-uring task context. * User buffers are not mapped yet - the application does not have permission * to write to it - this has to be executed in ring task context. */static void fuse_uring_send_in_task(struct io_tw_req tw_req, io_tw_token_t tw){ unsigned int issue_flags = IO_URING_CMD_TASK_WORK_ISSUE_FLAGS; struct io_uring_cmd *cmd = io_uring_cmd_from_tw(tw_req); struct fuse_ring_ent *ent = uring_cmd_to_ring_ent(cmd); struct fuse_ring_queue *queue = ent->queue; int err;
if (!tw.cancel) { err = fuse_uring_prepare_send(ent, ent->fuse_req); if (err) { if (!fuse_uring_get_next_fuse_req(ent, queue)) return; err = 0; } fuse_uring_send(ent, cmd, err, issue_flags); } else { err = -ECANCELED;
spin_lock(&queue->lock); list_del_init(&ent->list); spin_unlock(&queue->lock);
io_uring_cmd_done(cmd, err, issue_flags);
fuse_uring_req_end(ent, ent->fuse_req, err); kfree(ent); if (atomic_dec_and_test(&queue->ring->queue_refs)) wake_up_all(&queue->ring->stop_waitq); }}
static struct fuse_ring_queue *fuse_uring_task_to_queue(struct fuse_ring *ring){ unsigned int qid; struct fuse_ring_queue *queue;
qid = task_cpu(current);
if (WARN_ONCE(qid >= ring->nr_queues, "Core number (%u) exceeds nr queues (%zu)\n", qid, ring->nr_queues)) qid = 0;
queue = ring->queues[qid]; WARN_ONCE(!queue, "Missing queue for qid %d\n", qid);
return queue;}
static void fuse_uring_dispatch_ent(struct fuse_ring_ent *ent){ struct io_uring_cmd *cmd = ent->cmd;
uring_cmd_set_ring_ent(cmd, ent); io_uring_cmd_complete_in_task(cmd, fuse_uring_send_in_task);}
/* queue a fuse request and send it if a ring entry is available */void fuse_uring_queue_fuse_req(struct fuse_iqueue *fiq, struct fuse_req *req){ struct fuse_ring *ring = req->chan->ring; struct fuse_ring_queue *queue; struct fuse_ring_ent *ent = NULL; int err;
err = -EINVAL; queue = fuse_uring_task_to_queue(ring); if (!queue) goto err;
fuse_request_assign_unique(fiq, req);
spin_lock(&queue->lock); err = -ENOTCONN; if (unlikely(queue->stopped)) goto err_unlock;
set_bit(FR_URING, &req->flags); req->ring_queue = queue; ent = list_first_entry_or_null(&queue->ent_avail_queue, struct fuse_ring_ent, list); if (ent) fuse_uring_add_req_to_ring_ent(ent, req); else list_add_tail(&req->list, &queue->fuse_req_queue); spin_unlock(&queue->lock);
if (ent) fuse_uring_dispatch_ent(ent);
return;
err_unlock: spin_unlock(&queue->lock);err: req->out.h.error = err; clear_bit(FR_PENDING, &req->flags); fuse_request_end(req);}
bool fuse_uring_queue_bq_req(struct fuse_req *req){ struct fuse_chan *fch = req->chan; struct fuse_ring *ring = fch->ring; struct fuse_ring_queue *queue; struct fuse_ring_ent *ent = NULL;
queue = fuse_uring_task_to_queue(ring); if (!queue) return false;
spin_lock(&queue->lock); if (unlikely(queue->stopped)) { spin_unlock(&queue->lock); return false; }
set_bit(FR_URING, &req->flags); req->ring_queue = queue; list_add_tail(&req->list, &queue->fuse_req_bg_queue);
ent = list_first_entry_or_null(&queue->ent_avail_queue, struct fuse_ring_ent, list); spin_lock(&fch->bg_lock); fch->num_background++; if (fch->num_background == fch->max_background) fch->blocked = 1; fuse_uring_flush_bg(queue); spin_unlock(&fch->bg_lock);
/* * Due to bg_queue flush limits there might be other bg requests * in the queue that need to be handled first. Or no further req * might be available. */ req = list_first_entry_or_null(&queue->fuse_req_queue, struct fuse_req, list); if (ent && req) { fuse_uring_add_req_to_ring_ent(ent, req); spin_unlock(&queue->lock);
fuse_uring_dispatch_ent(ent); } else { spin_unlock(&queue->lock); }
return true;}
bool fuse_uring_remove_pending_req(struct fuse_req *req){ struct fuse_ring_queue *queue = req->ring_queue;
return fuse_remove_pending_req(req, &queue->lock);}
static const struct fuse_iqueue_ops fuse_io_uring_ops = { /* should be send over io-uring as enhancement */ .send_forget = fuse_dev_queue_forget,
/* * could be send over io-uring, but interrupts should be rare, * no need to make the code complex */ .send_interrupt = fuse_dev_queue_interrupt, .send_req = fuse_uring_queue_fuse_req,};