mirror of
https://github.com/libfuse/libfuse.git
synced 2026-07-07 14:47:41 +08:00
fuse_uring: restore inline-reply batching, add single-issuer fast path
Part 1 dropped cqe_processing and made the serving loop wait-only, so the default path now issues one io_uring_submit() per inline reply instead of coalescing a cqe batch into a single submit. Restore the batching: the loop splits submit_and_wait so the flush stays under ring_lock (part 1's race fix), and cqe_processing - now an _Atomic bool set lock-free around handle_cqes() - again gates the per-reply submit. The SQ stays serialised, so a batched reply is always flushed before the next wait, never stranded. Add a conn flag, io_uring_single_issuer, for a filesystem that replies only from the per-queue uring thread: it skips ring_lock entirely and runs the combined io_uring_submit_and_wait() loop, sound because only that thread touches the ring. The flag reuses a reserved padding bit (ABI unchanged; flag 0 keeps the multi-issuer path) and is mirrored into the ring pool after op.init(), before serving starts. A filesystem enables it with fuse_set_conn_flag(conn, FUSE_CONN_FLAG_SINGLE_ISSUER), a new exported setter modelled on fuse_set_feature_flag(); the io_uring_single_issuer bitfield is the internal storage. Signed-off-by: Bernd Schubert <bernd@bsbernd.com>
This commit is contained in:
+28
-1
@@ -698,8 +698,19 @@ struct fuse_conn_info {
|
||||
*/
|
||||
uint32_t no_interrupt : 1;
|
||||
|
||||
/**
|
||||
* Only meaningful with io-uring (FUSE_CAP_OVER_IO_URING).
|
||||
|
||||
* The filesystem promises that every reply to an io-uring request is
|
||||
* sent from the same thread that received the request (the per-queue
|
||||
* io-uring worker), i.e. replies are never deferred to another thread.
|
||||
* The flag is used for io-uring optimizations.
|
||||
*
|
||||
*/
|
||||
uint32_t io_uring_single_issuer : 1;
|
||||
|
||||
/* reserved bits for future use */
|
||||
uint32_t padding : 31;
|
||||
uint32_t padding : 30;
|
||||
|
||||
/**
|
||||
* Extended capability flags that the kernel supports (read-only)
|
||||
@@ -1140,6 +1151,22 @@ void fuse_unset_feature_flag(struct fuse_conn_info *conn, uint64_t flag);
|
||||
*/
|
||||
bool fuse_get_feature_flag(const struct fuse_conn_info *conn, uint64_t flag);
|
||||
|
||||
/* The file system replies to requests from the same thread that received them,
|
||||
* allowing io-uring optimizations
|
||||
*/
|
||||
#define FUSE_CONN_FLAG_SINGLE_ISSUER (1u << 0)
|
||||
|
||||
/**
|
||||
* Set a libfuse connection flag (a FUSE_CONN_FLAG_* value). Unlike
|
||||
* fuse_set_feature_flag(), these are libfuse-side hints, not negotiated
|
||||
* with the kernel.
|
||||
*
|
||||
* @param conn connection information
|
||||
* @param flag a single FUSE_CONN_FLAG_* value
|
||||
* @return true if the flag is known to this libfuse and was set, false otherwise
|
||||
*/
|
||||
bool fuse_set_conn_flag(struct fuse_conn_info *conn, uint64_t flag);
|
||||
|
||||
/*
|
||||
* DO NOT USE: Not part of public API, for internal test use only.
|
||||
* The function signature or any use of it is not guaranteeed to
|
||||
|
||||
@@ -2700,6 +2700,17 @@ bool fuse_get_feature_flag(const struct fuse_conn_info *conn,
|
||||
return (conn->capable_ext & flag) ? true : false;
|
||||
}
|
||||
|
||||
bool fuse_set_conn_flag(struct fuse_conn_info *conn, uint64_t flag)
|
||||
{
|
||||
switch (flag) {
|
||||
case FUSE_CONN_FLAG_SINGLE_ISSUER:
|
||||
conn->io_uring_single_issuer = 1;
|
||||
return true;
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/* Prevent bogus data races (bogus since "init" is called before
|
||||
* multi-threading becomes relevant */
|
||||
static __attribute__((no_sanitize("thread"))) void
|
||||
|
||||
+55
-16
@@ -75,6 +75,9 @@ struct fuse_ring_queue {
|
||||
struct fuse_ring_pool {
|
||||
struct fuse_session *se;
|
||||
|
||||
/* mirror of se->conn.io_uring_single_issuer, fixed at ring creation */
|
||||
bool single_issuer;
|
||||
|
||||
/* number of queues */
|
||||
size_t nr_queues;
|
||||
|
||||
@@ -162,6 +165,7 @@ static int fuse_uring_commit_sqe(struct fuse_ring_pool *ring_pool,
|
||||
struct fuse_ring_queue *queue,
|
||||
struct fuse_ring_ent *ring_ent)
|
||||
{
|
||||
const bool locked = !ring_pool->single_issuer;
|
||||
struct fuse_session *se = ring_pool->se;
|
||||
struct fuse_uring_req_header *rrh = ring_ent->req_header;
|
||||
struct fuse_out_header *out = (struct fuse_out_header *)&rrh->in_out;
|
||||
@@ -169,6 +173,13 @@ static int fuse_uring_commit_sqe(struct fuse_ring_pool *ring_pool,
|
||||
(struct fuse_uring_ent_in_out *)&rrh->ring_ent_in_out;
|
||||
struct io_uring_sqe *sqe;
|
||||
|
||||
/*
|
||||
* Multi-issuer: serialise every submission-side SQ access under
|
||||
* ring_lock. Single-issuer: only the uring thread submits, so skip the
|
||||
* lock and batch inline replies (cqe_processing), flushed by the next
|
||||
* submit_and_wait().
|
||||
*/
|
||||
if (locked)
|
||||
pthread_mutex_lock(&queue->ring_lock);
|
||||
|
||||
sqe = io_uring_get_sqe(&queue->ring);
|
||||
@@ -180,6 +191,7 @@ static int fuse_uring_commit_sqe(struct fuse_ring_pool *ring_pool,
|
||||
* SQEs matches the number tof requests.
|
||||
*/
|
||||
|
||||
if (locked)
|
||||
pthread_mutex_unlock(&queue->ring_lock);
|
||||
se->error = -EIO;
|
||||
fuse_log(FUSE_LOG_ERR, "Failed to get a ring SQEs\n");
|
||||
@@ -197,13 +209,10 @@ static int fuse_uring_commit_sqe(struct fuse_ring_pool *ring_pool,
|
||||
out->unique, ent_in_out->payload_sz);
|
||||
}
|
||||
|
||||
/*
|
||||
* only submit when the main thread processing is not running. I.e.
|
||||
* No submission per request, but aggregated submission, if possible
|
||||
*/
|
||||
if (!queue->cqe_processing)
|
||||
if (!atomic_load_explicit(&queue->cqe_processing, memory_order_relaxed))
|
||||
io_uring_submit(&queue->ring);
|
||||
|
||||
if (locked)
|
||||
pthread_mutex_unlock(&queue->ring_lock);
|
||||
|
||||
return 0;
|
||||
@@ -542,6 +551,7 @@ static struct fuse_ring_pool *fuse_create_ring(struct fuse_session *se)
|
||||
fuse_ring->queue_depth = se->uring.q_depth;
|
||||
fuse_ring->max_req_payload_sz = payload_sz;
|
||||
fuse_ring->queue_mem_size = queue_sz;
|
||||
fuse_ring->single_issuer = se->conn.io_uring_single_issuer;
|
||||
|
||||
/*
|
||||
* very basic queue initialization, that cannot fail and will
|
||||
@@ -575,8 +585,10 @@ err:
|
||||
static void fuse_uring_resubmit(struct fuse_ring_queue *queue,
|
||||
struct fuse_ring_ent *ent)
|
||||
{
|
||||
const bool locked = !queue->ring_pool->single_issuer;
|
||||
struct io_uring_sqe *sqe;
|
||||
|
||||
if (locked)
|
||||
pthread_mutex_lock(&queue->ring_lock);
|
||||
|
||||
sqe = io_uring_get_sqe(&queue->ring);
|
||||
@@ -587,6 +599,7 @@ static void fuse_uring_resubmit(struct fuse_ring_queue *queue,
|
||||
* SQEs matches the number tof requests.
|
||||
*/
|
||||
|
||||
if (locked)
|
||||
pthread_mutex_unlock(&queue->ring_lock);
|
||||
queue->ring_pool->se->error = -EIO;
|
||||
fuse_log(FUSE_LOG_ERR, "Failed to get a ring SQEs\n");
|
||||
@@ -614,7 +627,9 @@ static void fuse_uring_resubmit(struct fuse_ring_queue *queue,
|
||||
break;
|
||||
}
|
||||
|
||||
if (!atomic_load_explicit(&queue->cqe_processing, memory_order_relaxed))
|
||||
io_uring_submit(&queue->ring);
|
||||
if (locked)
|
||||
pthread_mutex_unlock(&queue->ring_lock);
|
||||
}
|
||||
|
||||
@@ -833,6 +848,7 @@ static void *fuse_uring_thread(void *arg)
|
||||
struct fuse_ring_queue *queue = arg;
|
||||
struct fuse_ring_pool *ring_pool = queue->ring_pool;
|
||||
struct fuse_session *se = ring_pool->se;
|
||||
const bool single_issuer = ring_pool->single_issuer;
|
||||
int err;
|
||||
char thread_name[16] = { 0 };
|
||||
|
||||
@@ -859,35 +875,58 @@ static void *fuse_uring_thread(void *arg)
|
||||
sem_wait(&ring_pool->init_sem);
|
||||
|
||||
/*
|
||||
* Submit the registration SQEs once. Safe without ring_lock: no
|
||||
* request can reach this queue until the kernel sees these, so there
|
||||
* is no concurrent submitter yet.
|
||||
* Multi-issuer flushes the registration SQEs here - safe without
|
||||
* ring_lock, no request can reach this queue yet. Single-issuer's
|
||||
* first submit_and_wait() below flushes them instead.
|
||||
*/
|
||||
if (!single_issuer)
|
||||
io_uring_submit(&queue->ring);
|
||||
|
||||
/* Not using fuse_session_exited(se), as that cannot be inlined */
|
||||
while (!atomic_load_explicit(&se->mt_exited, memory_order_relaxed)) {
|
||||
/*
|
||||
* Single-issuer: one combined submit_and_wait() flushes the
|
||||
* previous iteration's batched replies and waits. Multi-issuer:
|
||||
* split it - wait only here (no ring_lock) so off-thread
|
||||
* repliers keep submitting; the batched inline replies are
|
||||
* flushed below. The returned cqe is ignored; handle_cqes()
|
||||
* re-scans and advances the CQ.
|
||||
*/
|
||||
if (single_issuer) {
|
||||
io_uring_submit_and_wait(&queue->ring, 1);
|
||||
} else {
|
||||
struct io_uring_cqe *cqe;
|
||||
|
||||
/*
|
||||
* Wait only, not submit_and_wait: the serving loop must not
|
||||
* flush the SQ unlocked. All SQ flushing happens under ring_lock
|
||||
* - in commit_sqe()/resubmit() and the batched submit below. The
|
||||
* returned cqe is ignored; handle_cqes() re-scans the CQ.
|
||||
*/
|
||||
io_uring_wait_cqe(&queue->ring, &cqe);
|
||||
}
|
||||
|
||||
queue->cqe_processing = true;
|
||||
/*
|
||||
* Batch inline replies (commit_sqe()/resubmit()) across cqe
|
||||
* handling. Lock-free: the flag only gates who submits, while
|
||||
* the SQ stays serialised by ring_lock, so a reply that batched
|
||||
* here is always flushed by the submit below before the next
|
||||
* wait - it is never stranded.
|
||||
*/
|
||||
atomic_store_explicit(&queue->cqe_processing, true,
|
||||
memory_order_relaxed);
|
||||
|
||||
err = fuse_uring_queue_handle_cqes(queue);
|
||||
if (err < 0)
|
||||
goto err;
|
||||
|
||||
atomic_store_explicit(&queue->cqe_processing, false,
|
||||
memory_order_relaxed);
|
||||
|
||||
/*
|
||||
* Multi-issuer does not use io_uring_submit_and_wait(),
|
||||
* but io_uring_wait_cqe() and locked io_uring_submit().
|
||||
*/
|
||||
if (!single_issuer) {
|
||||
pthread_mutex_lock(&queue->ring_lock);
|
||||
queue->cqe_processing = false;
|
||||
io_uring_submit(&queue->ring);
|
||||
pthread_mutex_unlock(&queue->ring_lock);
|
||||
}
|
||||
}
|
||||
|
||||
return NULL;
|
||||
|
||||
|
||||
@@ -224,6 +224,7 @@ FUSE_3.18 {
|
||||
|
||||
FUSE_3.19 {
|
||||
global:
|
||||
fuse_set_conn_flag;
|
||||
fuse_session_start_teardown_watchdog;
|
||||
fuse_session_stop_teardown_watchdog;
|
||||
fuse_lowlevel_notify_prune;
|
||||
|
||||
Reference in New Issue
Block a user