summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--MAINTAINERS8
-rw-r--r--include/linux/io_uring_types.h46
-rw-r--r--include/uapi/linux/io_uring/bpf_filter.h16
-rw-r--r--include/uapi/linux/io_uring/query.h12
-rw-r--r--include/uapi/linux/io_uring/zcrx.h36
-rw-r--r--io_uring/bpf-ops.c9
-rw-r--r--io_uring/bpf-ops.h2
-rw-r--r--io_uring/cancel.c2
-rw-r--r--io_uring/fdinfo.c2
-rw-r--r--io_uring/io-wq.c2
-rw-r--r--io_uring/io_uring.c14
-rw-r--r--io_uring/io_uring.h3
-rw-r--r--io_uring/kbuf.c18
-rw-r--r--io_uring/loop.c2
-rw-r--r--io_uring/loop.h10
-rw-r--r--io_uring/mpscq.h125
-rw-r--r--io_uring/net.c137
-rw-r--r--io_uring/net.h7
-rw-r--r--io_uring/nop.c1
-rw-r--r--io_uring/opdef.c7
-rw-r--r--io_uring/query.c16
-rw-r--r--io_uring/rsrc.c269
-rw-r--r--io_uring/rsrc.h7
-rw-r--r--io_uring/sqpoll.c30
-rw-r--r--io_uring/tctx.c3
-rw-r--r--io_uring/tw.c315
-rw-r--r--io_uring/tw.h11
-rw-r--r--io_uring/wait.c2
-rw-r--r--io_uring/wait.h12
-rw-r--r--io_uring/zcrx.c229
-rw-r--r--io_uring/zcrx.h11
31 files changed, 1004 insertions, 360 deletions
diff --git a/MAINTAINERS b/MAINTAINERS
index 51a99eb86233..025e566e813a 100644
--- a/MAINTAINERS
+++ b/MAINTAINERS
@@ -13542,6 +13542,14 @@ T: git git://git.kernel.org/pub/scm/linux/kernel/git/axboe/linux.git
S: Maintained
F: io_uring/zcrx.*
+IO_URING BPF-OPS
+M: Pavel Begunkov <asml.silence@gmail.com>
+L: io-uring@vger.kernel.org
+T: git git://git.kernel.org/pub/scm/linux/kernel/git/axboe/linux.git
+S: Maintained
+F: io_uring/bpf-ops.*
+F: io_uring/loop.*
+
IPMI SUBSYSTEM
M: Corey Minyard <corey@minyard.net>
L: openipmi-developer@lists.sourceforge.net (moderated for non-subscribers)
diff --git a/include/linux/io_uring_types.h b/include/linux/io_uring_types.h
index 244392026c6d..6415a3353ee0 100644
--- a/include/linux/io_uring_types.h
+++ b/include/linux/io_uring_types.h
@@ -55,6 +55,18 @@ struct io_wq_work_list {
struct io_wq_work_node *last;
};
+/*
+ * Lockless multi-producer, single-consumer FIFO queue, see
+ * io_uring/mpscq.h for the implementation and rules. Defined here so
+ * that it can be embedded in io_ring_ctx. This is the producer side
+ * only - the consumer cursor is kept separately, on a cacheline that
+ * isn't dirtied by the producers.
+ */
+struct mpscq {
+ struct llist_node *tail; /* producers */
+ struct llist_node stub;
+};
+
struct io_wq_work {
struct io_wq_work_node list;
atomic_t flags;
@@ -119,6 +131,11 @@ struct io_uring_task {
const struct io_ring_ctx *last;
struct task_struct *task;
struct io_wq *io_wq;
+ /*
+ * Consumer cursor for ->task_list. Only popped by the task itself,
+ * or by ->fallback_work once the task can no longer run task_work.
+ */
+ struct llist_node *task_head;
struct file *registered_rings[IO_RINGFD_REG_MAX];
struct xarray xa;
@@ -127,8 +144,13 @@ struct io_uring_task {
atomic_t inflight_tracked;
struct percpu_counter inflight;
+ /* drains ->task_list once the task can no longer run task_work */
+ struct work_struct fallback_work;
+
struct { /* task_work */
- struct llist_head task_list;
+ struct mpscq task_list;
+ /* BIT(0) guards adding tw only once */
+ unsigned long tw_pending;
struct callback_head task_work;
} ____cacheline_aligned_in_smp;
};
@@ -290,6 +312,8 @@ enum {
IO_RING_F_IOWQ_LIMITS_SET = BIT(12),
};
+struct iou_ctx {};
+
struct io_ring_ctx {
/* const or read-mostly hot data */
struct {
@@ -346,6 +370,14 @@ struct io_ring_ctx {
bool poll_multi_queue;
struct list_head iopoll_list;
+ /*
+ * Consumer cursor for ->work_list, protected by ->uring_lock.
+ * Deliberately kept away from the producer side of the queue,
+ * as it's written for every popped entry, and the producer
+ * cacheline is contended enough as it is.
+ */
+ struct llist_node *work_head;
+
struct io_file_table file_table;
struct io_rsrc_data buf_table;
struct io_alloc_cache node_cache;
@@ -366,7 +398,7 @@ struct io_ring_ctx {
struct io_alloc_cache rw_cache;
struct io_alloc_cache cmd_cache;
- int (*loop_step)(struct io_ring_ctx *ctx,
+ int (*loop_step)(struct iou_ctx *,
struct iou_loop_params *);
/*
@@ -403,8 +435,7 @@ struct io_ring_ctx {
*/
struct {
struct io_rings __rcu *rings_rcu;
- struct llist_head work_llist;
- struct llist_head retry_llist;
+ struct mpscq work_list;
unsigned long check_cq;
atomic_t cq_wait_nr;
atomic_t cq_timeouts;
@@ -446,6 +477,9 @@ struct io_ring_ctx {
/* Stores zcrx object pointers of type struct io_zcrx_ifq */
struct xarray zcrx_ctxs;
+ /* Used for accounting references on pages in registered buffers */
+ struct xarray hpage_acct;
+
u32 pers_next;
struct xarray personalities;
@@ -464,8 +498,6 @@ struct io_ring_ctx {
struct mutex tctx_lock;
/* ctx exit and cancelation */
- struct llist_head fallback_llist;
- struct delayed_work fallback_work;
struct work_struct exit_work;
struct completion ref_comp;
@@ -725,8 +757,6 @@ struct io_kiocb {
*/
u16 buf_index;
- unsigned nr_tw;
-
/* REQ_F_* flags */
io_req_flags_t flags;
diff --git a/include/uapi/linux/io_uring/bpf_filter.h b/include/uapi/linux/io_uring/bpf_filter.h
index 1b461d792a7b..ce7d78ab13b3 100644
--- a/include/uapi/linux/io_uring/bpf_filter.h
+++ b/include/uapi/linux/io_uring/bpf_filter.h
@@ -27,6 +27,22 @@ struct io_uring_bpf_ctx {
__u64 mode;
__u64 resolve;
} open;
+ /*
+ * For CONNECT: fields are populated only when addr_len covers
+ * them; unpopulated fields are zero from the caller-side memset
+ * in io_uring_populate_bpf_ctx(). port and v4_addr are network
+ * byte order. Filters may only issue BPF_LD|BPF_W|BPF_ABS at
+ * 4-byte aligned offsets; load + mask for sub-word fields.
+ */
+ struct {
+ __u32 family; /* sa_family_t zero-extended */
+ __be16 port;
+ __u8 pad[2];
+ union {
+ __be32 v4_addr;
+ __u8 v6_addr[16];
+ };
+ } connect;
};
};
diff --git a/include/uapi/linux/io_uring/query.h b/include/uapi/linux/io_uring/query.h
index 95500759cc13..1a68eca7c6b4 100644
--- a/include/uapi/linux/io_uring/query.h
+++ b/include/uapi/linux/io_uring/query.h
@@ -23,6 +23,7 @@ enum {
IO_URING_QUERY_OPCODES = 0,
IO_URING_QUERY_ZCRX = 1,
IO_URING_QUERY_SCQ = 2,
+ IO_URING_QUERY_ZCRX_NOTIF = 3,
__IO_URING_QUERY_MAX,
};
@@ -62,6 +63,17 @@ struct io_uring_query_zcrx {
__u64 __resv2;
};
+struct io_uring_query_zcrx_notif {
+ /* Bitmask of supported ZCRX_NOTIF_* flags */
+ __u32 notif_flags;
+ /* Size of io_uring_zcrx_notif_stats */
+ __u32 notif_stats_size;
+ /* Required alignment for the stats struct within the region (ie stats_offset) */
+ __u32 notif_stats_off_alignment;
+ __u32 __resv1;
+ __u64 __resv2[4];
+};
+
struct io_uring_query_scq {
/* The SQ/CQ rings header size */
__u64 hdr_size;
diff --git a/include/uapi/linux/io_uring/zcrx.h b/include/uapi/linux/io_uring/zcrx.h
index 5ce02c7a6096..15c05c45ce36 100644
--- a/include/uapi/linux/io_uring/zcrx.h
+++ b/include/uapi/linux/io_uring/zcrx.h
@@ -65,6 +65,32 @@ enum zcrx_features {
* value in struct io_uring_zcrx_ifq_reg::rx_buf_len.
*/
ZCRX_FEATURE_RX_PAGE_SIZE = 1 << 0,
+ ZCRX_FEATURE_NOTIFICATION = 1 << 1,
+};
+
+enum zcrx_notification_type {
+ ZCRX_NOTIF_NO_BUFFERS,
+ ZCRX_NOTIF_COPY,
+
+ __ZCRX_NOTIF_TYPE_LAST,
+};
+
+enum zcrx_notification_desc_flags {
+ /* If set, stats_offset holds a valid offset to a notif_stats struct */
+ ZCRX_NOTIF_DESC_FLAG_STATS = 1 << 0,
+};
+
+struct zcrx_notif_stats {
+ __u64 copy_count; /* cumulative copy-fallback CQEs */
+ __u64 copy_bytes; /* cumulative bytes copied */
+};
+
+struct zcrx_notification_desc {
+ __u64 user_data;
+ __u32 type_mask;
+ __u32 flags; /* see enum zcrx_notification_desc_flags */
+ __u64 stats_offset; /* offset from the beginning of refill ring region for stats */
+ __u64 __resv2[9];
};
/*
@@ -82,12 +108,14 @@ struct io_uring_zcrx_ifq_reg {
struct io_uring_zcrx_offsets offsets;
__u32 zcrx_id;
__u32 rx_buf_len;
- __u64 __resv[3];
+ __u64 notif_desc; /* see struct zcrx_notification_desc */
+ __u64 __resv[2];
};
enum zcrx_ctrl_op {
ZCRX_CTRL_FLUSH_RQ,
ZCRX_CTRL_EXPORT,
+ ZCRX_CTRL_ARM_NOTIFICATION,
__ZCRX_CTRL_LAST,
};
@@ -101,6 +129,11 @@ struct zcrx_ctrl_export {
__u32 __resv1[11];
};
+struct zcrx_ctrl_arm_notif {
+ __u32 notif_type;
+ __u32 __resv[11];
+};
+
struct zcrx_ctrl {
__u32 zcrx_id;
__u32 op; /* see enum zcrx_ctrl_op */
@@ -109,6 +142,7 @@ struct zcrx_ctrl {
union {
struct zcrx_ctrl_export zc_export;
struct zcrx_ctrl_flush_rq zc_flush;
+ struct zcrx_ctrl_arm_notif zc_arm_notif;
};
};
diff --git a/io_uring/bpf-ops.c b/io_uring/bpf-ops.c
index 937e48bef40b..5a50f0675fe5 100644
--- a/io_uring/bpf-ops.c
+++ b/io_uring/bpf-ops.c
@@ -14,15 +14,18 @@ static const struct btf_type *loop_params_type;
__bpf_kfunc_start_defs();
-__bpf_kfunc int bpf_io_uring_submit_sqes(struct io_ring_ctx *ctx, u32 nr)
+__bpf_kfunc int bpf_io_uring_submit_sqes(struct iou_ctx *loop_ctx, u32 nr)
{
+ struct io_ring_ctx *ctx = io_loop_demangle_ctx(loop_ctx);
+
return io_submit_sqes(ctx, nr);
}
__bpf_kfunc
-__u8 *bpf_io_uring_get_region(struct io_ring_ctx *ctx, __u32 region_id,
+__u8 *bpf_io_uring_get_region(struct iou_ctx *loop_ctx, __u32 region_id,
const size_t rdwr_buf_size)
{
+ struct io_ring_ctx *ctx = io_loop_demangle_ctx(loop_ctx);
struct io_mapped_region *r;
lockdep_assert_held(&ctx->uring_lock);
@@ -58,7 +61,7 @@ static const struct btf_kfunc_id_set bpf_io_uring_kfunc_set = {
.set = &io_uring_kfunc_set,
};
-static int io_bpf_ops__loop_step(struct io_ring_ctx *ctx,
+static int io_bpf_ops__loop_step(struct iou_ctx *ctx,
struct iou_loop_params *lp)
{
return IOU_LOOP_STOP;
diff --git a/io_uring/bpf-ops.h b/io_uring/bpf-ops.h
index b39b3fd3acda..0b6d7894915e 100644
--- a/io_uring/bpf-ops.h
+++ b/io_uring/bpf-ops.h
@@ -11,7 +11,7 @@ enum {
};
struct io_uring_bpf_ops {
- int (*loop_step)(struct io_ring_ctx *ctx, struct iou_loop_params *lp);
+ int (*loop_step)(struct iou_ctx *, struct iou_loop_params *lp);
__u32 ring_fd;
void *priv;
diff --git a/io_uring/cancel.c b/io_uring/cancel.c
index 4aa3103ba9c3..8c6fa6f367e4 100644
--- a/io_uring/cancel.c
+++ b/io_uring/cancel.c
@@ -565,8 +565,6 @@ __cold bool io_uring_try_cancel_requests(struct io_ring_ctx *ctx,
mutex_unlock(&ctx->uring_lock);
if (tctx)
ret |= io_run_task_work() > 0;
- else
- ret |= flush_delayed_work(&ctx->fallback_work);
return ret;
}
diff --git a/io_uring/fdinfo.c b/io_uring/fdinfo.c
index 001fb542dc11..3ae4804765b9 100644
--- a/io_uring/fdinfo.c
+++ b/io_uring/fdinfo.c
@@ -224,7 +224,7 @@ static void __io_uring_show_fdinfo(struct io_ring_ctx *ctx, struct seq_file *m)
if (ctx->buf_table.nodes[i])
buf = ctx->buf_table.nodes[i]->buf;
if (buf)
- seq_printf(m, "%5u: 0x%llx/%u\n", i, buf->ubuf, buf->len);
+ seq_printf(m, "%5u: 0x%llx/%zu\n", i, buf->ubuf, buf->len);
else
seq_printf(m, "%5u: <none>\n", i);
}
diff --git a/io_uring/io-wq.c b/io_uring/io-wq.c
index 8cc7b47d3089..2e14880eef92 100644
--- a/io_uring/io-wq.c
+++ b/io_uring/io-wq.c
@@ -602,7 +602,6 @@ static void io_worker_handle_work(struct io_wq_acct *acct,
struct io_wq *wq = worker->wq;
do {
- bool do_kill = test_bit(IO_WQ_BIT_EXIT, &wq->state);
struct io_wq_work *work;
/*
@@ -638,6 +637,7 @@ static void io_worker_handle_work(struct io_wq_acct *acct,
/* handle a whole dependent link */
do {
+ bool do_kill = test_bit(IO_WQ_BIT_EXIT, &wq->state);
struct io_wq_work *next_hashed, *linked;
unsigned int work_flags = atomic_read(&work->flags);
unsigned int hash = __io_wq_is_hashed(work_flags)
diff --git a/io_uring/io_uring.c b/io_uring/io_uring.c
index bf847ca823f7..1ea2fca34a36 100644
--- a/io_uring/io_uring.c
+++ b/io_uring/io_uring.c
@@ -160,7 +160,7 @@ static void io_poison_cached_req(struct io_kiocb *req)
req->apoll = IO_URING_PTR_POISON;
}
-static void io_poison_req(struct io_kiocb *req)
+void io_poison_req(struct io_kiocb *req)
{
io_poison_cached_req(req);
req->async_data = IO_URING_PTR_POISON;
@@ -233,6 +233,7 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
return NULL;
xa_init(&ctx->io_bl_xa);
+ xa_init(&ctx->hpage_acct);
/*
* Use 5 bits less than the max cq entries, that should give us around
@@ -279,7 +280,7 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
INIT_LIST_HEAD(&ctx->defer_list);
INIT_LIST_HEAD(&ctx->timeout_list);
INIT_LIST_HEAD(&ctx->ltimeout_list);
- init_llist_head(&ctx->work_llist);
+ mpscq_init(&ctx->work_list, &ctx->work_head);
INIT_LIST_HEAD(&ctx->tctx_list);
mutex_init(&ctx->tctx_lock);
ctx->submit_state.free_list.next = NULL;
@@ -288,7 +289,6 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
#ifdef CONFIG_FUTEX
INIT_HLIST_HEAD(&ctx->futex_list);
#endif
- INIT_DELAYED_WORK(&ctx->fallback_work, io_fallback_req_func);
INIT_WQ_LIST(&ctx->submit_state.compl_reqs);
INIT_HLIST_HEAD(&ctx->cancelable_uring_cmd);
io_napi_init(ctx);
@@ -302,6 +302,7 @@ err:
io_free_alloc_caches(ctx);
kvfree(ctx->cancel_table.hbs);
xa_destroy(&ctx->io_bl_xa);
+ xa_destroy(&ctx->hpage_acct);
kfree(ctx);
return NULL;
}
@@ -1197,7 +1198,7 @@ __cold void io_iopoll_try_reap_events(struct io_ring_ctx *ctx)
mutex_unlock(&ctx->uring_lock);
if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
- io_move_task_work_from_local(ctx);
+ io_cancel_local_task_work(ctx);
}
static int io_iopoll_check(struct io_ring_ctx *ctx, unsigned int min_events)
@@ -2209,6 +2210,7 @@ static __cold void io_ring_ctx_free(struct io_ring_ctx *ctx)
io_napi_free(ctx);
kvfree(ctx->cancel_table.hbs);
xa_destroy(&ctx->io_bl_xa);
+ xa_destroy(&ctx->hpage_acct);
kfree(ctx);
}
@@ -2342,7 +2344,7 @@ static __cold void io_ring_exit_work(struct work_struct *work)
/* The SQPOLL thread never reaches this path */
do {
if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
- io_move_task_work_from_local(ctx);
+ io_cancel_local_task_work(ctx);
cond_resched();
} while (io_uring_try_cancel_requests(ctx, NULL, true, false));
@@ -2428,8 +2430,6 @@ static __cold void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
io_unregister_personality(ctx, index);
mutex_unlock(&ctx->uring_lock);
- flush_delayed_work(&ctx->fallback_work);
-
INIT_WORK(&ctx->exit_work, io_ring_exit_work);
/*
* Use system_dfl_wq to avoid spawning tons of event kworkers
diff --git a/io_uring/io_uring.h b/io_uring/io_uring.h
index e612a66ee80e..cb736b815422 100644
--- a/io_uring/io_uring.h
+++ b/io_uring/io_uring.h
@@ -213,6 +213,7 @@ bool __io_alloc_req_refill(struct io_ring_ctx *ctx);
void io_activate_pollwq(struct io_ring_ctx *ctx);
void io_restriction_clone(struct io_restriction *dst, struct io_restriction *src);
+void io_poison_req(struct io_kiocb *req);
static inline void io_lockdep_assert_cq_locked(struct io_ring_ctx *ctx)
{
@@ -312,7 +313,7 @@ static __always_inline bool io_fill_cqe_req(struct io_ring_ctx *ctx,
}
if (trace_io_uring_complete_enabled())
- trace_io_uring_complete(req->ctx, req, cqe);
+ trace_call__io_uring_complete(req->ctx, req, cqe);
return true;
}
diff --git a/io_uring/kbuf.c b/io_uring/kbuf.c
index 926254b6898f..3cd29477fff2 100644
--- a/io_uring/kbuf.c
+++ b/io_uring/kbuf.c
@@ -21,7 +21,7 @@
#define MAX_BIDS_PER_BGID (1 << 16)
/* Mapped buffer ring, return io_uring_buf from head */
-#define io_ring_head_to_buf(br, head, mask) &(br)->bufs[(head) & (mask)]
+#define io_ring_head_to_buf(br, head, mask) (&(br)->bufs[(head) & (mask)])
struct io_provide_buf {
struct file *file;
@@ -210,10 +210,14 @@ static struct io_br_sel io_ring_buffer_select(struct io_kiocb *req, size_t *len,
buf_len = READ_ONCE(buf->len);
if (*len == 0 || *len > buf_len)
*len = buf_len;
+ sel.addr = u64_to_user_ptr(READ_ONCE(buf->addr));
+ if (unlikely(!access_ok(sel.addr, *len))) {
+ sel.addr = NULL;
+ return sel;
+ }
req->flags |= REQ_F_BUFFER_RING | REQ_F_BUFFERS_COMMIT;
req->buf_index = READ_ONCE(buf->bid);
sel.buf_list = bl;
- sel.addr = u64_to_user_ptr(READ_ONCE(buf->addr));
if (io_should_commit(req, issue_flags)) {
if (!io_kbuf_commit(req, sel.buf_list, *len, 1))
@@ -250,6 +254,7 @@ static int io_ring_buffers_peek(struct io_kiocb *req, struct buf_sel_arg *arg,
struct io_buffer_list *bl)
{
struct io_uring_buf_ring *br = bl->buf_ring;
+ struct iovec *org_iovs = arg->iovs;
struct iovec *iov = arg->iovs;
int nr_iovs = arg->nr_iovs;
__u16 nr_avail, tail, head;
@@ -310,6 +315,11 @@ static int io_ring_buffers_peek(struct io_kiocb *req, struct buf_sel_arg *arg,
iov->iov_base = u64_to_user_ptr(READ_ONCE(buf->addr));
iov->iov_len = len;
+ if (unlikely(!access_ok(iov->iov_base, len))) {
+ if (arg->iovs != org_iovs)
+ kfree(arg->iovs);
+ return -EFAULT;
+ }
iov++;
arg->out_len += len;
@@ -540,11 +550,11 @@ static int io_add_buffers(struct io_ring_ctx *ctx, struct io_provide_buf *pbuf,
for (i = 0; i < pbuf->nbufs; i++) {
/*
- * Nonsensical to have more than sizeof(bid) buffers in a
+ * Nonsensical to have more than MAX_BIDS_PER_BGID buffers in a
* buffer list, as the application then has no way of knowing
* which duplicate bid refers to what buffer.
*/
- if (bl->nbufs == USHRT_MAX) {
+ if (bl->nbufs == MAX_BIDS_PER_BGID) {
ret = -EOVERFLOW;
break;
}
diff --git a/io_uring/loop.c b/io_uring/loop.c
index 31843cc3e451..bbbb6ef14e6a 100644
--- a/io_uring/loop.c
+++ b/io_uring/loop.c
@@ -49,7 +49,7 @@ static int __io_run_loop(struct io_ring_ctx *ctx)
if (unlikely(!ctx->loop_step))
return -EFAULT;
- step_res = ctx->loop_step(ctx, &lp);
+ step_res = ctx->loop_step(io_loop_mangle_ctx(ctx), &lp);
if (step_res == IOU_LOOP_STOP)
break;
if (step_res != IOU_LOOP_CONTINUE)
diff --git a/io_uring/loop.h b/io_uring/loop.h
index d7718b9ce61e..4dd4fb3aefef 100644
--- a/io_uring/loop.h
+++ b/io_uring/loop.h
@@ -24,4 +24,14 @@ static inline bool io_has_loop_ops(struct io_ring_ctx *ctx)
int io_run_loop(struct io_ring_ctx *ctx);
+static inline struct iou_ctx *io_loop_mangle_ctx(struct io_ring_ctx *ctx)
+{
+ return (struct iou_ctx *)ctx;
+}
+
+static inline struct io_ring_ctx *io_loop_demangle_ctx(struct iou_ctx *ctx)
+{
+ return (struct io_ring_ctx *)ctx;
+}
+
#endif
diff --git a/io_uring/mpscq.h b/io_uring/mpscq.h
new file mode 100644
index 000000000000..c801384c6a0a
--- /dev/null
+++ b/io_uring/mpscq.h
@@ -0,0 +1,125 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+#ifndef IOU_MPSCQ_H
+#define IOU_MPSCQ_H
+
+#include <linux/io_uring_types.h>
+
+/*
+ * mpscq - lockless multi-producer, single-consumer FIFO queue
+ *
+ * Unlike llist, which is LIFO ordered and hence needs an O(n)
+ * llist_reverse_order() pass before entries can be processed in queue order,
+ * this queue hands out nodes in the order they were pushed.
+ *
+ * The consumer cursor is held by the caller rather than in the queue struct
+ * (see below), and with the stub reinsertion done as a single cmpxchg attempt
+ * instead of an unconditional push, keeping tail == stub a reliable empty test
+ * while a producer is in the middle of a push.
+ *
+ * Producers may run in any context (task, softirq, hardirq) and are wait-free:
+ * a push is one xchg() plus one store, with no retry loops. FIFO order between
+ * producers is the order in which the xchg() on ->tail serializes them.
+ *
+ * The price for linked-list FIFO is that a push publishes the node in two
+ * steps: the xchg() makes it the new tail, and the subsequent store links it to
+ * its predecessor. In between, the tail end of the queue is not yet reachable
+ * from the head. mpscq_pop() detects this and returns NULL, while mpscq_empty()
+ * reports false. The consumer must not treat such a NULL as "queue empty" - it
+ * should retry later. The window is two instructions wide, but a producer can
+ * be preempted inside it, so the consumer must not spin on it while holding
+ * resources the producer might need to make progress.
+ *
+ * The consumer side only supports a single consumer at a time, callers must
+ * provide their own serialization for it. The stub node is what allows the
+ * consumer to detach the final node without racing with the link stores of
+ * producers. This scheme also guarantees that the previous tail observed by
+ * mpscq_push() cannot be freed by the consumer until the push has linked it,
+ * which is what makes the deferred link store safe.
+ *
+ * The queue struct only holds the producer side. The consumer keeps its cursor
+ * (the oldest not yet handed out node) externally and passes it to mpscq_pop(),
+ * so that it can be placed on a different cacheline: the cursor is written for
+ * every pop, and having it share a line with ->tail would have the consumer
+ * invalidating the line that producers need for every push.
+ */
+static inline void mpscq_init(struct mpscq *q, struct llist_node **headp)
+{
+ q->tail = *headp = &q->stub;
+ q->stub.next = NULL;
+}
+
+/*
+ * Returns true if the queue holds no entries that mpscq_pop() hasn't handed out
+ * yet. May be called from any context. Note that !empty doesn't guarantee that
+ * mpscq_pop() will return an entry yet, see the in-flight producer window
+ * above.
+ */
+static inline bool mpscq_empty(struct mpscq *q)
+{
+ return READ_ONCE(q->tail) == &q->stub;
+}
+
+/*
+ * Push a node onto the queue. Safe against concurrent pushes from any context,
+ * and against the (single) consumer. Returns true if the queue was empty
+ * before this push.
+ */
+static inline bool mpscq_push(struct mpscq *q, struct llist_node *node)
+{
+ struct llist_node *prev;
+
+ node->next = NULL;
+ /*
+ * xchg() implies a full barrier, so the initialization of the
+ * entry (including ->next above) is visible before the node can
+ * be reached, either via ->tail or via ->next chasing from the
+ * head once the store below has linked it.
+ */
+ prev = xchg(&q->tail, node);
+ WRITE_ONCE(prev->next, node);
+ return prev == &q->stub;
+}
+
+/*
+ * Pop the oldest node off the queue, or return NULL if no node is available.
+ * NULL is returned both when the queue is empty and when a producer has
+ * published a node via ->tail but hasn't linked it yet; use mpscq_empty() to
+ * tell the two apart. Single consumer only, with headp being the consumer
+ * cursor that mpscq_init() set up.
+ */
+static inline struct llist_node *mpscq_pop(struct mpscq *q,
+ struct llist_node **headp)
+{
+ struct llist_node *head = *headp, *next;
+
+ if (head == &q->stub) {
+ head = READ_ONCE(head->next);
+ if (!head)
+ return NULL;
+ /*
+ * The stub is now detached and stays quiescent until the
+ * consumer reinserts it as the tail, so reset its ->next here,
+ * ready for that.
+ */
+ q->stub.next = NULL;
+ *headp = head;
+ }
+ next = READ_ONCE(head->next);
+ if (next) {
+ *headp = next;
+ return head;
+ }
+ /*
+ * 'head' is the last linked node, it can only be handed out once the
+ * stub has taken its place as the tail. If the cmpxchg fails, a
+ * producer has made a new node the tail but hasn't linked 'head' to
+ * it yet - bail and let the caller retry.
+ */
+ if (try_cmpxchg(&q->tail, &head, &q->stub)) {
+ *headp = &q->stub;
+ return head;
+ }
+ return NULL;
+}
+
+#endif /* IOU_MPSCQ_H */
diff --git a/io_uring/net.c b/io_uring/net.c
index 34ab1f97cc5a..00a7df803b99 100644
--- a/io_uring/net.c
+++ b/io_uring/net.c
@@ -418,7 +418,8 @@ static int io_sendmsg_setup(struct io_kiocb *req, const struct io_uring_sqe *sqe
return io_net_import_vec(req, kmsg, msg.msg_iov, msg.msg_iovlen, ITER_SOURCE);
}
-#define SENDMSG_FLAGS (IORING_RECVSEND_POLL_FIRST | IORING_RECVSEND_BUNDLE | IORING_SEND_VECTORIZED)
+#define SENDMSG_FLAGS (IORING_RECVSEND_POLL_FIRST | IORING_RECVSEND_BUNDLE | \
+ IORING_SEND_VECTORIZED | IORING_RECVSEND_FIXED_BUF)
int io_sendmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
{
@@ -431,6 +432,14 @@ int io_sendmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
sr->flags = READ_ONCE(sqe->ioprio);
if (sr->flags & ~SENDMSG_FLAGS)
return -EINVAL;
+ if (sr->flags & IORING_RECVSEND_FIXED_BUF) {
+ /* registered buffer send only supported for plain IORING_OP_SEND */
+ if (req->opcode != IORING_OP_SEND ||
+ (req->flags & REQ_F_BUFFER_SELECT) ||
+ (sr->flags & (IORING_RECVSEND_BUNDLE|IORING_SEND_VECTORIZED)))
+ return -EINVAL;
+ req->buf_index = READ_ONCE(sqe->buf_index);
+ }
sr->msg_flags = READ_ONCE(sqe->msg_flags) | MSG_NOSIGNAL;
if (sr->msg_flags & MSG_DONTWAIT)
req->flags |= REQ_F_NOWAIT;
@@ -662,6 +671,15 @@ int io_send(struct io_kiocb *req, unsigned int issue_flags)
(sr->flags & IORING_RECVSEND_POLL_FIRST))
return -EAGAIN;
+ if (req->flags & REQ_F_IMPORT_BUFFER) {
+ ret = io_import_reg_buf(req, &kmsg->msg.msg_iter,
+ (u64)(uintptr_t)sr->buf, sr->len,
+ ITER_SOURCE, issue_flags);
+ if (unlikely(ret))
+ return ret;
+ req->flags &= ~REQ_F_IMPORT_BUFFER;
+ }
+
flags = sr->msg_flags;
if (issue_flags & IO_URING_F_NONBLOCK)
flags |= MSG_DONTWAIT;
@@ -776,6 +794,10 @@ static int io_recvmsg_prep_setup(struct io_kiocb *req)
if (req->flags & REQ_F_BUFFER_SELECT)
return 0;
+ if (sr->flags & IORING_RECVSEND_FIXED_BUF) {
+ req->flags |= REQ_F_IMPORT_BUFFER;
+ return 0;
+ }
return import_ubuf(ITER_DEST, sr->buf, sr->len,
&kmsg->msg.msg_iter);
}
@@ -784,7 +806,7 @@ static int io_recvmsg_prep_setup(struct io_kiocb *req)
}
#define RECVMSG_FLAGS (IORING_RECVSEND_POLL_FIRST | IORING_RECV_MULTISHOT | \
- IORING_RECVSEND_BUNDLE)
+ IORING_RECVSEND_BUNDLE | IORING_RECVSEND_FIXED_BUF)
int io_recvmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
{
@@ -802,6 +824,14 @@ int io_recvmsg_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
sr->flags = READ_ONCE(sqe->ioprio);
if (sr->flags & ~RECVMSG_FLAGS)
return -EINVAL;
+ if (sr->flags & IORING_RECVSEND_FIXED_BUF) {
+ /* registered buffer recv only for plain IORING_OP_RECV */
+ if (req->opcode != IORING_OP_RECV ||
+ (req->flags & REQ_F_BUFFER_SELECT) ||
+ (sr->flags & (IORING_RECV_MULTISHOT | IORING_RECVSEND_BUNDLE)))
+ return -EINVAL;
+ req->buf_index = READ_ONCE(sqe->buf_index);
+ }
sr->msg_flags = READ_ONCE(sqe->msg_flags);
if (sr->msg_flags & MSG_DONTWAIT)
req->flags |= REQ_F_NOWAIT;
@@ -1187,18 +1217,30 @@ int io_recv(struct io_kiocb *req, unsigned int issue_flags)
bool force_nonblock = issue_flags & IO_URING_F_NONBLOCK;
bool mshot_finished;
- if (!(req->flags & REQ_F_POLLED) &&
- (sr->flags & IORING_RECVSEND_POLL_FIRST))
- return -EAGAIN;
-
sock = sock_from_file(req->file);
if (unlikely(!sock))
return -ENOTSOCK;
+ if (!(req->flags & REQ_F_POLLED) &&
+ (sr->flags & IORING_RECVSEND_POLL_FIRST))
+ return -EAGAIN;
+
flags = sr->msg_flags;
if (force_nonblock)
flags |= MSG_DONTWAIT;
+ if (req->flags & REQ_F_IMPORT_BUFFER) {
+ ret = io_import_reg_buf(req, &kmsg->msg.msg_iter,
+ (u64)(uintptr_t)sr->buf, sr->len,
+ ITER_DEST, issue_flags);
+ if (unlikely(ret)) {
+ kmsg->msg.msg_inq = -1;
+ sel.buf_list = NULL;
+ goto out_free;
+ }
+ req->flags &= ~REQ_F_IMPORT_BUFFER;
+ }
+
retry_multishot:
sel.buf_list = NULL;
if (io_do_buffer_select(req)) {
@@ -1287,14 +1329,14 @@ int io_recvzc(struct io_kiocb *req, unsigned int issue_flags)
unsigned int len;
int ret;
- if (!(req->flags & REQ_F_POLLED) &&
- (zc->flags & IORING_RECVSEND_POLL_FIRST))
- return -EAGAIN;
-
sock = sock_from_file(req->file);
if (unlikely(!sock))
return -ENOTSOCK;
+ if (!(req->flags & REQ_F_POLLED) &&
+ (zc->flags & IORING_RECVSEND_POLL_FIRST))
+ return -EAGAIN;
+
len = zc->len;
ret = io_zcrx_recv(req, zc->ifq, sock, 0, issue_flags, &zc->len);
if (len && zc->len == 0) {
@@ -1675,6 +1717,46 @@ void io_socket_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req)
bctx->socket.protocol = sock->protocol;
}
+void io_connect_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req)
+{
+ struct io_connect *conn = io_kiocb_to_cmd(req, struct io_connect);
+ struct sockaddr_storage *ss = req->async_data;
+
+ /*
+ * move_addr_to_kernel() skips the copy for addr_len == 0, so
+ * iomsg->addr may hold stale data from a prior CONNECT. Bail
+ * unless addr_len covers the family discriminator.
+ */
+ if (conn->addr_len < (int)sizeof(sa_family_t))
+ return;
+
+ bctx->connect.family = ss->ss_family;
+ switch (ss->ss_family) {
+ case AF_INET: {
+ struct sockaddr_in *sin = (struct sockaddr_in *)ss;
+
+ if (conn->addr_len < (int)sizeof(*sin))
+ break;
+ bctx->connect.port = sin->sin_port;
+ bctx->connect.v4_addr = sin->sin_addr.s_addr;
+ break;
+ }
+ case AF_INET6: {
+ struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)ss;
+
+ if (conn->addr_len < (int)sizeof(*sin6))
+ break;
+ bctx->connect.port = sin6->sin6_port;
+ memcpy(bctx->connect.v6_addr, &sin6->sin6_addr,
+ sizeof(bctx->connect.v6_addr));
+ break;
+ }
+ default:
+ /* family is set; per-family fields stay zero - family-only filtering */
+ break;
+ }
+}
+
int io_socket_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
{
struct io_socket *sock = io_kiocb_to_cmd(req, struct io_socket);
@@ -1732,7 +1814,7 @@ int io_socket(struct io_kiocb *req, unsigned int issue_flags)
int io_connect_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
{
struct io_connect *conn = io_kiocb_to_cmd(req, struct io_connect);
- struct io_async_msghdr *io;
+ struct sockaddr_storage *addr;
if (sqe->len || sqe->buf_index || sqe->rw_flags || sqe->splice_fd_in)
return -EINVAL;
@@ -1741,17 +1823,17 @@ int io_connect_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
conn->addr_len = READ_ONCE(sqe->addr2);
conn->in_progress = conn->seen_econnaborted = false;
- io = io_msg_alloc_async(req);
- if (unlikely(!io))
+ addr = io_uring_alloc_async_data(NULL, req);
+ if (unlikely(!addr))
return -ENOMEM;
- return move_addr_to_kernel(conn->addr, conn->addr_len, &io->addr);
+ return move_addr_to_kernel(conn->addr, conn->addr_len, addr);
}
int io_connect(struct io_kiocb *req, unsigned int issue_flags)
{
struct io_connect *connect = io_kiocb_to_cmd(req, struct io_connect);
- struct io_async_msghdr *io = req->async_data;
+ struct sockaddr_storage *addr = req->async_data;
unsigned file_flags;
int ret;
bool force_nonblock = issue_flags & IO_URING_F_NONBLOCK;
@@ -1765,8 +1847,7 @@ int io_connect(struct io_kiocb *req, unsigned int issue_flags)
file_flags = force_nonblock ? O_NONBLOCK : 0;
- ret = __sys_connect_file(req->file, &io->addr, connect->addr_len,
- file_flags);
+ ret = __sys_connect_file(req->file, addr, connect->addr_len, file_flags);
if ((ret == -EAGAIN || ret == -EINPROGRESS || ret == -ECONNABORTED)
&& force_nonblock) {
if (ret == -EINPROGRESS) {
@@ -1795,7 +1876,6 @@ get_sock_err:
out:
if (ret < 0)
req_set_fail(req);
- io_req_msg_cleanup(req, issue_flags);
io_req_set_res(req, ret, 0);
return IOU_COMPLETE;
}
@@ -1805,15 +1885,15 @@ out:
* which in turn end up in mnt_want_write() which will grab the fs
* percpu start write sem. This can trigger a lockdep warning.
*/
-static int io_bind_file_create(const struct io_async_msghdr *io, int addr_len)
+static int io_bind_file_create(const struct sockaddr_storage *addr, int addr_len)
{
const struct sockaddr_un *sun;
- if (io->addr.ss_family != AF_UNIX)
+ if (addr->ss_family != AF_UNIX)
return 0;
if (addr_len <= offsetof(struct sockaddr_un, sun_path))
return 0;
- sun = (const struct sockaddr_un *) &io->addr;
+ sun = (const struct sockaddr_un *) addr;
return sun->sun_path[0] != '\0';
}
@@ -1821,7 +1901,7 @@ int io_bind_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
{
struct io_bind *bind = io_kiocb_to_cmd(req, struct io_bind);
struct sockaddr __user *uaddr;
- struct io_async_msghdr *io;
+ struct sockaddr_storage *addr;
int ret;
if (sqe->len || sqe->buf_index || sqe->rw_flags || sqe->splice_fd_in)
@@ -1830,21 +1910,22 @@ int io_bind_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
uaddr = u64_to_user_ptr(READ_ONCE(sqe->addr));
bind->addr_len = READ_ONCE(sqe->addr2);
- io = io_msg_alloc_async(req);
- if (unlikely(!io))
+ addr = io_uring_alloc_async_data(NULL, req);
+ if (unlikely(!addr))
return -ENOMEM;
- ret = move_addr_to_kernel(uaddr, bind->addr_len, &io->addr);
+ ret = move_addr_to_kernel(uaddr, bind->addr_len, addr);
if (unlikely(ret))
return ret;
- if (io_bind_file_create(io, bind->addr_len))
+ if (io_bind_file_create(addr, bind->addr_len))
req->flags |= REQ_F_FORCE_ASYNC;
return 0;
}
+
int io_bind(struct io_kiocb *req, unsigned int issue_flags)
{
struct io_bind *bind = io_kiocb_to_cmd(req, struct io_bind);
- struct io_async_msghdr *io = req->async_data;
+ struct sockaddr_storage *addr = req->async_data;
struct socket *sock;
int ret;
@@ -1852,7 +1933,7 @@ int io_bind(struct io_kiocb *req, unsigned int issue_flags)
if (unlikely(!sock))
return -ENOTSOCK;
- ret = __sys_bind_socket(sock, &io->addr, bind->addr_len);
+ ret = __sys_bind_socket(sock, addr, bind->addr_len);
if (ret < 0)
req_set_fail(req);
io_req_set_res(req, ret, 0);
diff --git a/io_uring/net.h b/io_uring/net.h
index d4d1ddce50e3..51fda715d3c0 100644
--- a/io_uring/net.h
+++ b/io_uring/net.h
@@ -46,6 +46,7 @@ int io_accept(struct io_kiocb *req, unsigned int issue_flags);
int io_socket_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
int io_socket(struct io_kiocb *req, unsigned int issue_flags);
void io_socket_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req);
+void io_connect_bpf_populate(struct io_uring_bpf_ctx *bctx, struct io_kiocb *req);
int io_connect_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
int io_connect(struct io_kiocb *req, unsigned int issue_flags);
@@ -69,4 +70,10 @@ static inline void io_socket_bpf_populate(struct io_uring_bpf_ctx *bctx,
struct io_kiocb *req)
{
}
+
+static inline void io_connect_bpf_populate(struct io_uring_bpf_ctx *bctx,
+ struct io_kiocb *req)
+{
+}
+
#endif
diff --git a/io_uring/nop.c b/io_uring/nop.c
index f5c9969e7f64..91ae0b2e7e55 100644
--- a/io_uring/nop.c
+++ b/io_uring/nop.c
@@ -12,7 +12,6 @@
#include "nop.h"
struct io_nop {
- /* NOTE: kiocb has the file as the first member, so don't do it here */
struct file *file;
int result;
int fd;
diff --git a/io_uring/opdef.c b/io_uring/opdef.c
index c3ef52b70811..88a45c7d897f 100644
--- a/io_uring/opdef.c
+++ b/io_uring/opdef.c
@@ -203,9 +203,11 @@ const struct io_issue_def io_issue_defs[] = {
.unbound_nonreg_file = 1,
.pollout = 1,
#if defined(CONFIG_NET)
- .async_size = sizeof(struct io_async_msghdr),
+ .filter_pdu_size = sizeof_field(struct io_uring_bpf_ctx, connect),
+ .async_size = sizeof(struct sockaddr_storage),
.prep = io_connect_prep,
.issue = io_connect,
+ .filter_populate = io_connect_bpf_populate,
#else
.prep = io_eopnotsupp_prep,
#endif
@@ -503,7 +505,7 @@ const struct io_issue_def io_issue_defs[] = {
.needs_file = 1,
.prep = io_bind_prep,
.issue = io_bind,
- .async_size = sizeof(struct io_async_msghdr),
+ .async_size = sizeof(struct sockaddr_storage),
#else
.prep = io_eopnotsupp_prep,
#endif
@@ -513,7 +515,6 @@ const struct io_issue_def io_issue_defs[] = {
.needs_file = 1,
.prep = io_listen_prep,
.issue = io_listen,
- .async_size = sizeof(struct io_async_msghdr),
#else
.prep = io_eopnotsupp_prep,
#endif
diff --git a/io_uring/query.c b/io_uring/query.c
index c1704d088374..d529d94aa8f4 100644
--- a/io_uring/query.c
+++ b/io_uring/query.c
@@ -9,6 +9,7 @@
union io_query_data {
struct io_uring_query_opcode opcodes;
struct io_uring_query_zcrx zcrx;
+ struct io_uring_query_zcrx_notif zcrx_notif;
struct io_uring_query_scq scq;
};
@@ -44,6 +45,18 @@ static ssize_t io_query_zcrx(union io_query_data *data)
return sizeof(*e);
}
+static ssize_t io_query_zcrx_notif(union io_query_data *data)
+{
+ struct io_uring_query_zcrx_notif *e = &data->zcrx_notif;
+
+ e->notif_flags = ZCRX_NOTIF_TYPE_MASK;
+ e->notif_stats_size = sizeof(struct zcrx_notif_stats);
+ e->notif_stats_off_alignment = __alignof__(struct zcrx_notif_stats);
+ e->__resv1 = 0;
+ memset(&e->__resv2, 0, sizeof(e->__resv2));
+ return sizeof(*e);
+}
+
static ssize_t io_query_scq(union io_query_data *data)
{
struct io_uring_query_scq *e = &data->scq;
@@ -83,6 +96,9 @@ static int io_handle_query_entry(union io_query_data *data, void __user *uhdr,
case IO_URING_QUERY_ZCRX:
ret = io_query_zcrx(data);
break;
+ case IO_URING_QUERY_ZCRX_NOTIF:
+ ret = io_query_zcrx_notif(data);
+ break;
case IO_URING_QUERY_SCQ:
ret = io_query_scq(data);
break;
diff --git a/io_uring/rsrc.c b/io_uring/rsrc.c
index 650303626be6..7f553d115e36 100644
--- a/io_uring/rsrc.c
+++ b/io_uring/rsrc.c
@@ -28,7 +28,52 @@ struct io_rsrc_update {
};
static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx,
- struct iovec *iov, struct page **last_hpage);
+ struct iovec *iov);
+
+static int hpage_acct_ref(struct io_ring_ctx *ctx, struct page *hpage,
+ bool *acct_new)
+{
+ unsigned long key = (unsigned long) hpage;
+ unsigned long count;
+ void *entry;
+ int ret;
+
+ lockdep_assert_held(&ctx->uring_lock);
+
+ entry = xa_load(&ctx->hpage_acct, key);
+ if (entry) {
+ *acct_new = false;
+ count = xa_to_value(entry) + 1;
+ } else {
+ ret = xa_reserve(&ctx->hpage_acct, key, GFP_KERNEL_ACCOUNT);
+ if (ret)
+ return ret;
+ *acct_new = true;
+ count = 1;
+ }
+ xa_store(&ctx->hpage_acct, key, xa_mk_value(count), GFP_KERNEL_ACCOUNT);
+ return 0;
+}
+
+static bool hpage_acct_unref(struct io_ring_ctx *ctx, struct page *hpage)
+{
+ unsigned long key = (unsigned long) hpage;
+ unsigned long count;
+ void *entry;
+
+ lockdep_assert_held(&ctx->uring_lock);
+
+ entry = xa_load(&ctx->hpage_acct, key);
+ if (WARN_ON_ONCE(!entry))
+ return false;
+ count = xa_to_value(entry);
+ if (count == 1) {
+ xa_erase(&ctx->hpage_acct, key);
+ return true;
+ }
+ xa_store(&ctx->hpage_acct, key, xa_mk_value(count - 1), GFP_KERNEL_ACCOUNT);
+ return false;
+}
/* only define max */
#define IORING_MAX_FIXED_FILES (1U << 20)
@@ -88,9 +133,14 @@ int io_validate_user_buf_range(u64 uaddr, u64 ulen)
unsigned long tmp, base = (unsigned long)uaddr;
unsigned long acct_len = (unsigned long)PAGE_ALIGN(ulen);
- /* arbitrary limit, but we need something */
- if (ulen > SZ_1G || !ulen)
+ if (!ulen)
return -EFAULT;
+ /* 32-bit sanity checking */
+ if (ulen > ULONG_MAX || uaddr > ULONG_MAX)
+ return -EFAULT;
+ /* cap to 1TB for 64-bit */
+ if (ulen > SZ_1T)
+ return -EINVAL;
if (check_add_overflow(base, acct_len, &tmp))
return -EOVERFLOW;
return 0;
@@ -124,15 +174,53 @@ static void io_free_imu(struct io_ring_ctx *ctx, struct io_mapped_ubuf *imu)
kvfree(imu);
}
+static unsigned long io_buffer_unaccount_pages(struct io_ring_ctx *ctx,
+ struct io_mapped_ubuf *imu)
+{
+ struct page *seen = NULL;
+ unsigned long acct = 0;
+ int i;
+
+ if (imu->flags & IO_REGBUF_F_KBUF || !ctx->user)
+ return 0;
+
+ for (i = 0; i < imu->nr_bvecs; i++) {
+ struct page *page = imu->bvec[i].bv_page;
+ struct page *hpage;
+
+ if (!PageCompound(page)) {
+ acct++;
+ continue;
+ }
+
+ hpage = compound_head(page);
+ if (hpage == seen)
+ continue;
+ seen = hpage;
+
+ /* Unaccount on last reference */
+ if (hpage_acct_unref(ctx, hpage))
+ acct += page_size(hpage) >> PAGE_SHIFT;
+ cond_resched();
+ }
+
+ return acct;
+}
+
static void io_buffer_unmap(struct io_ring_ctx *ctx, struct io_mapped_ubuf *imu)
{
+ unsigned long acct_pages = 0;
+
+ /* Always decrement, so it works for cloned buffers too */
+ acct_pages = io_buffer_unaccount_pages(ctx, imu);
+
if (unlikely(refcount_read(&imu->refs) > 1)) {
if (!refcount_dec_and_test(&imu->refs))
return;
}
- if (imu->acct_pages)
- io_unaccount_mem(ctx->user, ctx->mm_account, imu->acct_pages);
+ if (acct_pages)
+ io_unaccount_mem(ctx->user, ctx->mm_account, acct_pages);
imu->release(imu->priv);
io_free_imu(ctx, imu);
}
@@ -282,7 +370,6 @@ static int __io_sqe_buffers_update(struct io_ring_ctx *ctx,
{
u64 __user *tags = u64_to_user_ptr(up->tags);
struct iovec fast_iov, *iov;
- struct page *last_hpage = NULL;
struct iovec __user *uvec;
u64 user_data = up->data;
__u32 done;
@@ -307,7 +394,7 @@ static int __io_sqe_buffers_update(struct io_ring_ctx *ctx,
err = -EFAULT;
break;
}
- node = io_sqe_buffer_register(ctx, iov, &last_hpage);
+ node = io_sqe_buffer_register(ctx, iov);
if (IS_ERR(node)) {
err = PTR_ERR(node);
break;
@@ -605,76 +692,79 @@ int io_sqe_buffers_unregister(struct io_ring_ctx *ctx)
}
/*
- * Not super efficient, but this is just a registration time. And we do cache
- * the last compound head, so generally we'll only do a full search if we don't
- * match that one.
- *
- * We check if the given compound head page has already been accounted, to
- * avoid double accounting it. This allows us to account the full size of the
- * page, not just the constituent pages of a huge page.
+ * Undo hpage_acct_ref() calls made during io_buffer_account_pin() on failure.
+ * This operates on the pages array since imu->bvec isn't populated yet.
*/
-static bool headpage_already_acct(struct io_ring_ctx *ctx, struct page **pages,
- int nr_pages, struct page *hpage)
+static void io_buffer_unaccount_hpages(struct io_ring_ctx *ctx,
+ struct page **pages, int nr_pages)
{
- int i, j;
+ struct page *seen = NULL;
+ int i;
+
+ if (!ctx->user)
+ return;
- /* check current page array */
for (i = 0; i < nr_pages; i++) {
+ struct page *hpage;
+
if (!PageCompound(pages[i]))
continue;
- if (compound_head(pages[i]) == hpage)
- return true;
- }
- /* check previously registered pages */
- for (i = 0; i < ctx->buf_table.nr; i++) {
- struct io_rsrc_node *node = ctx->buf_table.nodes[i];
- struct io_mapped_ubuf *imu;
-
- if (!node)
+ hpage = compound_head(pages[i]);
+ if (hpage == seen)
continue;
- imu = node->buf;
- for (j = 0; j < imu->nr_bvecs; j++) {
- if (!PageCompound(imu->bvec[j].bv_page))
- continue;
- if (compound_head(imu->bvec[j].bv_page) == hpage)
- return true;
- }
- }
+ seen = hpage;
- return false;
+ hpage_acct_unref(ctx, hpage);
+ cond_resched();
+ }
}
static int io_buffer_account_pin(struct io_ring_ctx *ctx, struct page **pages,
- int nr_pages, struct io_mapped_ubuf *imu,
- struct page **last_hpage)
+ int nr_pages)
{
+ unsigned long acct_pages = 0;
+ struct page *seen = NULL;
int i, ret;
- imu->acct_pages = 0;
+ if (!ctx->user)
+ return 0;
+
for (i = 0; i < nr_pages; i++) {
+ struct page *hpage;
+ bool acct_new;
+
if (!PageCompound(pages[i])) {
- imu->acct_pages++;
- } else {
- struct page *hpage;
-
- hpage = compound_head(pages[i]);
- if (hpage == *last_hpage)
- continue;
- *last_hpage = hpage;
- if (headpage_already_acct(ctx, pages, i, hpage))
- continue;
- imu->acct_pages += page_size(hpage) >> PAGE_SHIFT;
+ acct_pages++;
+ continue;
+ }
+
+ hpage = compound_head(pages[i]);
+ if (hpage == seen)
+ continue;
+ seen = hpage;
+
+ ret = hpage_acct_ref(ctx, hpage, &acct_new);
+ if (ret) {
+ io_buffer_unaccount_hpages(ctx, pages, i);
+ return ret;
}
+ if (acct_new)
+ acct_pages += page_size(hpage) >> PAGE_SHIFT;
+ cond_resched();
}
- if (!imu->acct_pages)
- return 0;
+ /* Try to account the memory */
+ if (acct_pages) {
+ ret = io_account_mem(ctx->user, ctx->mm_account, acct_pages);
+ if (ret) {
+ /* Undo the refs we just added */
+ io_buffer_unaccount_hpages(ctx, pages, nr_pages);
+ return ret;
+ }
+ }
- ret = io_account_mem(ctx->user, ctx->mm_account, imu->acct_pages);
- if (ret)
- imu->acct_pages = 0;
- return ret;
+ return 0;
}
static bool io_coalesce_buffer(struct page ***pages, int *nr_pages,
@@ -763,8 +853,7 @@ bool io_check_coalesce_buffer(struct page **page_array, int nr_pages,
}
static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx,
- struct iovec *iov,
- struct page **last_hpage)
+ struct iovec *iov)
{
struct io_mapped_ubuf *imu = NULL;
struct page **pages = NULL;
@@ -811,7 +900,7 @@ static struct io_rsrc_node *io_sqe_buffer_register(struct io_ring_ctx *ctx,
goto done;
imu->nr_bvecs = nr_pages;
- ret = io_buffer_account_pin(ctx, pages, nr_pages, imu, last_hpage);
+ ret = io_buffer_account_pin(ctx, pages, nr_pages);
if (ret)
goto done;
@@ -861,7 +950,6 @@ done:
int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
unsigned int nr_args, u64 __user *tags)
{
- struct page *last_hpage = NULL;
struct io_rsrc_data data;
struct iovec fast_iov, *iov = &fast_iov;
const struct iovec __user *uvec;
@@ -904,7 +992,7 @@ int io_sqe_buffers_register(struct io_ring_ctx *ctx, void __user *arg,
}
}
- node = io_sqe_buffer_register(ctx, iov, &last_hpage);
+ node = io_sqe_buffer_register(ctx, iov);
if (IS_ERR(node)) {
ret = PTR_ERR(node);
break;
@@ -971,7 +1059,6 @@ int io_buffer_register_bvec(struct io_uring_cmd *cmd, struct request *rq,
imu->ubuf = 0;
imu->len = blk_rq_bytes(rq);
- imu->acct_pages = 0;
imu->folio_shift = PAGE_SHIFT;
refcount_set(&imu->refs, 1);
imu->release = release;
@@ -1136,6 +1223,56 @@ int io_import_reg_buf(struct io_kiocb *req, struct iov_iter *iter,
return io_import_fixed(ddir, iter, node->buf, buf_addr, len);
}
+static int io_buffer_acct_cloned_hpages(struct io_ring_ctx *ctx,
+ struct io_mapped_ubuf *imu)
+{
+ struct page *seen = NULL;
+ int i, ret = 0;
+
+ if (imu->flags & IO_REGBUF_F_KBUF || !ctx->user)
+ return 0;
+
+ for (i = 0; i < imu->nr_bvecs; i++) {
+ struct page *page = imu->bvec[i].bv_page;
+ struct page *hpage;
+ bool acct_new;
+
+ if (!PageCompound(page))
+ continue;
+
+ hpage = compound_head(page);
+ if (hpage == seen)
+ continue;
+ seen = hpage;
+
+ /* Atomically add reference for cloned buffer */
+ ret = hpage_acct_ref(ctx, hpage, &acct_new);
+ if (ret)
+ break;
+
+ cond_resched();
+ }
+
+ if (!ret)
+ return 0;
+
+ /* Undo refs we added for bvecs [0..i) */
+ seen = NULL;
+ for (int j = 0; j < i; j++) {
+ struct page *p = imu->bvec[j].bv_page;
+ struct page *hp;
+
+ if (!PageCompound(p))
+ continue;
+ hp = compound_head(p);
+ if (hp == seen)
+ continue;
+ seen = hp;
+ hpage_acct_unref(ctx, hp);
+ }
+ return ret;
+}
+
/* Lock two rings at once. The rings must be different! */
static void lock_two_rings(struct io_ring_ctx *ctx1, struct io_ring_ctx *ctx2)
{
@@ -1218,6 +1355,14 @@ static int io_clone_buffers(struct io_ring_ctx *ctx, struct io_ring_ctx *src_ctx
refcount_inc(&src_node->buf->refs);
dst_node->buf = src_node->buf;
+ /* track compound references to clones */
+ ret = io_buffer_acct_cloned_hpages(ctx, src_node->buf);
+ if (ret) {
+ refcount_dec(&src_node->buf->refs);
+ io_cache_free(&ctx->node_cache, dst_node);
+ io_rsrc_data_free(ctx, &data);
+ return ret;
+ }
}
data.nodes[off++] = dst_node;
i++;
diff --git a/io_uring/rsrc.h b/io_uring/rsrc.h
index 44e3386f7c1c..98ae8ef51009 100644
--- a/io_uring/rsrc.h
+++ b/io_uring/rsrc.h
@@ -34,15 +34,14 @@ enum {
struct io_mapped_ubuf {
u64 ubuf;
- unsigned int len;
+ size_t len;
unsigned int nr_bvecs;
unsigned int folio_shift;
refcount_t refs;
- unsigned long acct_pages;
- void (*release)(void *);
- void *priv;
u8 flags;
u8 dir;
+ void (*release)(void *);
+ void *priv;
struct bio_vec bvec[] __counted_by(nr_bvecs);
};
diff --git a/io_uring/sqpoll.c b/io_uring/sqpoll.c
index 46c12afec73e..2460bd605266 100644
--- a/io_uring/sqpoll.c
+++ b/io_uring/sqpoll.c
@@ -260,39 +260,29 @@ static bool io_sqd_handle_event(struct io_sq_data *sqd)
}
/*
- * Run task_work, processing the retry_list first. The retry_list holds
- * entries that we passed on in the previous run, if we had more task_work
- * than we were asked to process. Newly queued task_work isn't run until the
- * retry list has been fully processed.
+ * Run task_work, processing no more than max_entries at a time. If more
+ * than that is pending, it simply stays on the queue for the next run.
*/
-static unsigned int io_sq_tw(struct llist_node **retry_list, int max_entries)
+static unsigned int io_sq_tw(int max_entries)
{
struct io_uring_task *tctx = current->io_uring;
unsigned int count = 0;
- if (*retry_list) {
- *retry_list = io_handle_tw_list(*retry_list, &count, max_entries);
- if (count >= max_entries)
- goto out;
- max_entries -= count;
- }
- *retry_list = tctx_task_work_run(tctx, max_entries, &count);
-out:
+ tctx_task_work_run(tctx, max_entries, &count);
if (task_work_pending(current))
task_work_run();
return count;
}
-static bool io_sq_tw_pending(struct llist_node *retry_list)
+static bool io_sq_tw_pending(void)
{
struct io_uring_task *tctx = current->io_uring;
- return retry_list || !llist_empty(&tctx->task_list);
+ return !mpscq_empty(&tctx->task_list);
}
static int io_sq_thread(void *data)
{
- struct llist_node *retry_list = NULL;
struct io_sq_data *sqd = data;
struct io_ring_ctx *ctx;
unsigned long timeout = 0;
@@ -347,7 +337,7 @@ static int io_sq_thread(void *data)
if (!sqt_spin && (ret > 0 || !list_empty(&ctx->iopoll_list)))
sqt_spin = true;
}
- if (io_sq_tw(&retry_list, IORING_TW_CAP_ENTRIES_VALUE))
+ if (io_sq_tw(IORING_TW_CAP_ENTRIES_VALUE))
sqt_spin = true;
list_for_each_entry(ctx, &sqd->ctx_list, sqd_list) {
@@ -372,7 +362,7 @@ static int io_sq_thread(void *data)
}
prepare_to_wait(&sqd->wait, &wait, TASK_INTERRUPTIBLE);
- if (!io_sqd_events_pending(sqd) && !io_sq_tw_pending(retry_list)) {
+ if (!io_sqd_events_pending(sqd) && !io_sq_tw_pending()) {
bool needs_sched = true;
list_for_each_entry(ctx, &sqd->ctx_list, sqd_list) {
@@ -411,8 +401,8 @@ static int io_sq_thread(void *data)
timeout = jiffies + sqd->sq_thread_idle;
}
- if (retry_list)
- io_sq_tw(&retry_list, UINT_MAX);
+ if (io_sq_tw_pending())
+ io_sq_tw(UINT_MAX);
io_uring_cancel_generic(true, sqd);
rcu_assign_pointer(sqd->thread, NULL);
diff --git a/io_uring/tctx.c b/io_uring/tctx.c
index 42b219b34aa8..cc3bf2b3bdbc 100644
--- a/io_uring/tctx.c
+++ b/io_uring/tctx.c
@@ -103,7 +103,8 @@ __cold struct io_uring_task *io_uring_alloc_task_context(struct task_struct *tas
init_waitqueue_head(&tctx->wait);
atomic_set(&tctx->in_cancel, 0);
atomic_set(&tctx->inflight_tracked, 0);
- init_llist_head(&tctx->task_list);
+ mpscq_init(&tctx->task_list, &tctx->task_head);
+ INIT_WORK(&tctx->fallback_work, io_tctx_fallback_work);
init_task_work(&tctx->task_work, tctx_task_work);
return tctx;
}
diff --git a/io_uring/tw.c b/io_uring/tw.c
index 023d5e6bc491..e74372233f40 100644
--- a/io_uring/tw.c
+++ b/io_uring/tw.c
@@ -14,24 +14,7 @@
#include "rw.h"
#include "eventfd.h"
#include "wait.h"
-
-void io_fallback_req_func(struct work_struct *work)
-{
- struct io_ring_ctx *ctx = container_of(work, struct io_ring_ctx,
- fallback_work.work);
- struct llist_node *node = llist_del_all(&ctx->fallback_llist);
- struct io_kiocb *req, *tmp;
- struct io_tw_state ts = {};
-
- percpu_ref_get(&ctx->refs);
- mutex_lock(&ctx->uring_lock);
- ts.cancel = io_should_terminate_tw(ctx);
- llist_for_each_entry_safe(req, tmp, node, io_task_work.node)
- req->io_task_work.func((struct io_tw_req){req}, ts);
- io_submit_flush_completions(ctx);
- mutex_unlock(&ctx->uring_lock);
- percpu_ref_put(&ctx->refs);
-}
+#include "mpscq.h"
static void ctx_flush_and_put(struct io_ring_ctx *ctx, io_tw_token_t tw)
{
@@ -45,23 +28,68 @@ static void ctx_flush_and_put(struct io_ring_ctx *ctx, io_tw_token_t tw)
percpu_ref_put(&ctx->refs);
}
+void io_tctx_fallback_work(struct work_struct *work)
+{
+ struct io_uring_task *tctx = container_of(work, struct io_uring_task,
+ fallback_work);
+ unsigned int count = 0;
+
+ /* see tctx_task_work() - a set bit must always have a run coming */
+ clear_bit(0, &tctx->tw_pending);
+ smp_mb__after_atomic();
+
+ /*
+ * Run the entries directly. We're in PF_KTHRED context, hence
+ * io_should_terminate_tw() is true and they will be marked as
+ * canceled.
+ */
+ tctx_task_work_run(tctx, UINT_MAX, &count);
+ put_task_struct(tctx->task);
+}
+
+static void io_fallback_tw(struct io_uring_task *tctx)
+{
+ /*
+ * The task ref both keeps ->task valid and, as __io_uring_free() is
+ * only called when the task itself is freed, ensures the tctx (and
+ * the queued work) stay around until the drain has run.
+ */
+ get_task_struct(tctx->task);
+ if (!queue_work(system_unbound_wq, &tctx->fallback_work))
+ put_task_struct(tctx->task);
+}
+
/*
- * Run queued task_work, returning the number of entries processed in *count.
- * If more entries than max_entries are available, stop processing once this
- * is reached and return the rest of the list.
+ * Run queued task_work, processing no more than max_entries, with the number
+ * of entries processed added to *count. If more entries than max_entries are
+ * available, the remainder simply stay on the queue for the next run.
*/
-struct llist_node *io_handle_tw_list(struct llist_node *node,
- unsigned int *count,
- unsigned int max_entries)
+void tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries,
+ unsigned int *count)
{
struct io_ring_ctx *ctx = NULL;
struct io_tw_state ts = { };
- do {
- struct llist_node *next = node->next;
- struct io_kiocb *req = container_of(node, struct io_kiocb,
- io_task_work.node);
+ while (*count < max_entries) {
+ struct llist_node *node = mpscq_pop(&tctx->task_list,
+ &tctx->task_head);
+ struct io_kiocb *req;
+ if (!node) {
+ if (mpscq_empty(&tctx->task_list))
+ break;
+ /*
+ * A producer has published a node but hasn't
+ * linked it into the queue yet (see mpscq_pop()).
+ * Give it a chance to finish rather than spinning,
+ * and don't sit on the ctx lock while doing so.
+ */
+ ctx_flush_and_put(ctx, ts);
+ ctx = NULL;
+ cond_resched();
+ continue;
+ }
+ req = container_of(node, struct io_kiocb, io_task_work.node);
if (req->ctx != ctx) {
ctx_flush_and_put(ctx, ts);
ctx = req->ctx;
@@ -72,84 +100,36 @@ struct llist_node *io_handle_tw_list(struct llist_node *node,
INDIRECT_CALL_2(req->io_task_work.func,
io_poll_task_func, io_req_rw_complete,
(struct io_tw_req){req}, ts);
- node = next;
(*count)++;
if (unlikely(need_resched())) {
ctx_flush_and_put(ctx, ts);
ctx = NULL;
cond_resched();
}
- } while (node && *count < max_entries);
-
- ctx_flush_and_put(ctx, ts);
- return node;
-}
-
-static __cold void __io_fallback_tw(struct llist_node *node, bool sync)
-{
- struct io_ring_ctx *last_ctx = NULL;
- struct io_kiocb *req;
-
- while (node) {
- req = container_of(node, struct io_kiocb, io_task_work.node);
- node = node->next;
- if (last_ctx != req->ctx) {
- if (last_ctx) {
- if (sync)
- flush_delayed_work(&last_ctx->fallback_work);
- percpu_ref_put(&last_ctx->refs);
- }
- last_ctx = req->ctx;
- percpu_ref_get(&last_ctx->refs);
- }
- if (llist_add(&req->io_task_work.node, &last_ctx->fallback_llist))
- schedule_delayed_work(&last_ctx->fallback_work, 1);
- }
-
- if (last_ctx) {
- if (sync)
- flush_delayed_work(&last_ctx->fallback_work);
- percpu_ref_put(&last_ctx->refs);
- }
-}
-
-static void io_fallback_tw(struct io_uring_task *tctx, bool sync)
-{
- struct llist_node *node = llist_del_all(&tctx->task_list);
-
- __io_fallback_tw(node, sync);
-}
-
-struct llist_node *tctx_task_work_run(struct io_uring_task *tctx,
- unsigned int max_entries,
- unsigned int *count)
-{
- struct llist_node *node;
-
- node = llist_del_all(&tctx->task_list);
- if (node) {
- node = llist_reverse_order(node);
- node = io_handle_tw_list(node, count, max_entries);
}
+ ctx_flush_and_put(ctx, ts);
- /* relaxed read is enough as only the task itself sets ->in_cancel */
- if (unlikely(atomic_read(&tctx->in_cancel)))
+ /*
+ * Relaxed read is enough as only the task itself sets ->in_cancel.
+ * The tctx may also be drained by io_tctx_fallback_work(), in which
+ * case current is a kworker that has no tctx refs to drop.
+ */
+ if (unlikely(atomic_read(&tctx->in_cancel)) &&
+ current->io_uring == tctx)
io_uring_drop_tctx_refs(current);
trace_io_uring_task_work_run(tctx, *count);
- return node;
}
void tctx_task_work(struct callback_head *cb)
{
struct io_uring_task *tctx;
- struct llist_node *ret;
unsigned int count = 0;
tctx = container_of(cb, struct io_uring_task, task_work);
- ret = tctx_task_work_run(tctx, UINT_MAX, &count);
- /* can't happen */
- WARN_ON_ONCE(ret);
+ clear_bit(0, &tctx->tw_pending);
+ smp_mb__after_atomic();
+ tctx_task_work_run(tctx, UINT_MAX, &count);
}
/*
@@ -158,11 +138,11 @@ void tctx_task_work(struct callback_head *cb)
*/
static void io_ctx_mark_taskrun(struct io_ring_ctx *ctx)
{
- lockdep_assert_in_rcu_read_lock();
-
if (ctx->flags & IORING_SETUP_TASKRUN_FLAG) {
- struct io_rings *rings = rcu_dereference(ctx->rings_rcu);
+ struct io_rings *rings;
+ guard(rcu)();
+ rings = rcu_dereference(ctx->rings_rcu);
atomic_or(IORING_SQ_TASKRUN, &rings->sq_flags);
}
}
@@ -170,11 +150,7 @@ static void io_ctx_mark_taskrun(struct io_ring_ctx *ctx)
void io_req_local_work_add(struct io_kiocb *req, unsigned flags)
{
struct io_ring_ctx *ctx = req->ctx;
- unsigned nr_wait, nr_tw, nr_tw_prev;
- struct llist_node *head;
-
- /* See comment above IO_CQ_WAKE_INIT */
- BUILD_BUG_ON(IO_CQ_WAKE_FORCE <= IORING_MAX_CQ_ENTRIES);
+ int nr_wait;
/*
* We don't know how many requests there are in the link and whether
@@ -183,56 +159,45 @@ void io_req_local_work_add(struct io_kiocb *req, unsigned flags)
if (req->flags & IO_REQ_LINK_FLAGS)
flags &= ~IOU_F_TWQ_LAZY_WAKE;
- guard(rcu)();
-
- head = READ_ONCE(ctx->work_llist.first);
- do {
- nr_tw_prev = 0;
- if (head) {
- struct io_kiocb *first_req = container_of(head,
- struct io_kiocb,
- io_task_work.node);
- /*
- * Might be executed at any moment, rely on
- * SLAB_TYPESAFE_BY_RCU to keep it alive.
- */
- nr_tw_prev = READ_ONCE(first_req->nr_tw);
- }
-
- /*
- * Theoretically, it can overflow, but that's fine as one of
- * previous adds should've tried to wake the task.
- */
- nr_tw = nr_tw_prev + 1;
- if (!(flags & IOU_F_TWQ_LAZY_WAKE))
- nr_tw = IO_CQ_WAKE_FORCE;
-
- req->nr_tw = nr_tw;
- req->io_task_work.node.next = head;
- } while (!try_cmpxchg(&ctx->work_llist.first, &head,
- &req->io_task_work.node));
-
/*
- * cmpxchg implies a full barrier, which pairs with the barrier
- * in set_current_state() on the io_cqring_wait() side. It's used
- * to ensure that either we see updated ->cq_wait_nr, or waiters
- * going to sleep will observe the work added to the list, which
- * is similar to the wait/wawke task state sync.
+ * The xchg() in mpscq_push() implies a full barrier, which pairs with
+ * the barrier in set_current_state() on the io_cqring_wait() side. This
+ * ensures that either we see the updated ->cq_wait_nr, or waiters going
+ * to sleep will observe the work added to the list, which is similar to
+ * the wait/wake task state sync.
*/
-
- if (!head) {
+ if (mpscq_push(&ctx->work_list, &req->io_task_work.node)) {
io_ctx_mark_taskrun(ctx);
if (data_race(ctx->int_flags) & IO_RING_F_HAS_EVFD)
io_eventfd_signal(ctx, false);
}
+ /*
+ * No one is waiting (IO_CQ_WAKE_INIT), or this cycle's wake up has
+ * already been issued (zero or negative, see below).
+ */
nr_wait = atomic_read(&ctx->cq_wait_nr);
- /* not enough or no one is waiting */
- if (nr_tw < nr_wait)
+ if (nr_wait <= 0)
return;
- /* the previous add has already woken it up */
- if (nr_tw_prev >= nr_wait)
+ if (flags & IOU_F_TWQ_LAZY_WAKE) {
+ /*
+ * ->cq_wait_nr counts down the number of lazy adds, once it
+ * hits zero we're good to wake the waiter. A producer that
+ * gets delayed between pushing its entry and getting here
+ * may count down a later wait cycle. That's OK, it'll be an
+ * early wake, not a lost one.
+ */
+ if (!atomic_dec_and_test(&ctx->cq_wait_nr))
+ return;
+ } else if (atomic_xchg(&ctx->cq_wait_nr, IO_CQ_WAKE_INIT) <= 0) {
+ /*
+ * Potentially raced with lazy add, claim the wake. A value
+ * <= 0 means a lazy add hit zero or another forced add
+ * claimed IO_CQ_WAKE_INIT. Either way, the wake up for this
+ * wait cycle has already been done.
+ */
return;
+ }
wake_up_state(ctx->submitter_task, TASK_INTERRUPTIBLE);
}
@@ -242,7 +207,7 @@ void io_req_normal_work_add(struct io_kiocb *req)
struct io_ring_ctx *ctx = req->ctx;
/* task_work already pending, we're done */
- if (!llist_add(&req->io_task_work.node, &tctx->task_list))
+ if (!mpscq_push(&tctx->task_list, &req->io_task_work.node))
return;
/*
@@ -258,10 +223,14 @@ void io_req_normal_work_add(struct io_kiocb *req)
return;
}
+ /* task_work must only be added once */
+ if (test_and_set_bit(0, &tctx->tw_pending))
+ return;
+
if (likely(!task_work_add(tctx->task, &tctx->task_work, ctx->notify_method)))
return;
- io_fallback_tw(tctx, false);
+ io_fallback_tw(tctx);
}
void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags)
@@ -271,23 +240,34 @@ void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags)
__io_req_task_work_add(req, flags);
}
-void __cold io_move_task_work_from_local(struct io_ring_ctx *ctx)
+void __cold io_cancel_local_task_work(struct io_ring_ctx *ctx)
{
+ struct io_tw_state ts = { .cancel = true };
struct llist_node *node;
/*
- * Running the work items may utilize ->retry_llist as a means
- * for capping the number of task_work entries run at the same
- * time. But that list can potentially race with moving the work
- * from here, if the task is exiting. As any normal task_work
- * running holds ->uring_lock already, just guard this slow path
- * with ->uring_lock to avoid racing on ->retry_llist.
+ * The work list consumer side is serialized by ->uring_lock, see
+ * __io_run_local_work(). Grab it to guard against racing with normal
+ * task_work running, as the task may be exiting. The ring is going
+ * away, run the entries in cancel mode right here - the callers
+ * provide the same process context the per-ctx fallback work that
+ * they were previously punted to ran in.
*/
guard(mutex)(&ctx->uring_lock);
- node = llist_del_all(&ctx->work_llist);
- __io_fallback_tw(node, false);
- node = llist_del_all(&ctx->retry_llist);
- __io_fallback_tw(node, false);
+
+ while (!mpscq_empty(&ctx->work_list)) {
+ struct io_kiocb *req;
+
+ node = mpscq_pop(&ctx->work_list, &ctx->work_head);
+ if (!node) {
+ /* a producer is mid-push, wait for it to link */
+ cond_resched();
+ continue;
+ }
+ req = container_of(node, struct io_kiocb, io_task_work.node);
+ req->io_task_work.func((struct io_tw_req){req}, ts);
+ }
+ io_submit_flush_completions(ctx);
}
static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events,
@@ -302,22 +282,23 @@ static bool io_run_local_work_continue(struct io_ring_ctx *ctx, int events,
return false;
}
-static int __io_run_local_work_loop(struct llist_node **node,
+static int __io_run_local_work_loop(struct io_ring_ctx *ctx,
io_tw_token_t tw,
int events)
{
int ret = 0;
- while (*node) {
- struct llist_node *next = (*node)->next;
- struct io_kiocb *req = container_of(*node, struct io_kiocb,
- io_task_work.node);
+ while (ret < events) {
+ struct llist_node *node = mpscq_pop(&ctx->work_list, &ctx->work_head);
+ struct io_kiocb *req;
+
+ if (!node)
+ break;
+ req = container_of(node, struct io_kiocb, io_task_work.node);
INDIRECT_CALL_2(req->io_task_work.func,
io_poll_task_func, io_req_rw_complete,
(struct io_tw_req){req}, tw);
- *node = next;
- if (++ret >= events)
- break;
+ ret++;
}
return ret;
@@ -326,7 +307,6 @@ static int __io_run_local_work_loop(struct llist_node **node,
static int __io_run_local_work(struct io_ring_ctx *ctx, io_tw_token_t tw,
int min_events, int max_events)
{
- struct llist_node *node;
unsigned int loops = 0;
int ret = 0;
@@ -335,24 +315,21 @@ static int __io_run_local_work(struct io_ring_ctx *ctx, io_tw_token_t tw,
if (ctx->flags & IORING_SETUP_TASKRUN_FLAG)
atomic_andnot(IORING_SQ_TASKRUN, &ctx->rings->sq_flags);
again:
- tw.cancel = io_should_terminate_tw(ctx);
- min_events -= ret;
- ret = __io_run_local_work_loop(&ctx->retry_llist.first, tw, max_events);
- if (ctx->retry_llist.first)
- goto retry_done;
-
/*
- * llists are in reverse order, flip it back the right way before
- * running the pending items.
+ * If the last loop made no progress while work is still pending,
+ * a producer has published a node but hasn't linked it into the
+ * queue yet (see mpscq_pop()). Give it a chance to finish rather
+ * than spinning on the queue.
*/
- node = llist_reverse_order(llist_del_all(&ctx->work_llist));
- ret += __io_run_local_work_loop(&node, tw, max_events - ret);
- ctx->retry_llist.first = node;
+ if (unlikely(loops && !ret))
+ cond_resched();
+ tw.cancel = io_should_terminate_tw(ctx);
+ min_events -= ret;
+ ret = __io_run_local_work_loop(ctx, tw, max_events);
loops++;
if (io_run_local_work_continue(ctx, ret, min_events))
goto again;
-retry_done:
io_submit_flush_completions(ctx);
if (io_run_local_work_continue(ctx, ret, min_events))
goto again;
diff --git a/io_uring/tw.h b/io_uring/tw.h
index 415e330fabde..3ade5ad577fd 100644
--- a/io_uring/tw.h
+++ b/io_uring/tw.h
@@ -6,6 +6,8 @@
#include <linux/percpu-refcount.h>
#include <linux/io_uring_types.h>
+#include "mpscq.h"
+
#define IO_LOCAL_TW_DEFAULT_MAX 20
/*
@@ -23,18 +25,17 @@ static inline bool io_should_terminate_tw(struct io_ring_ctx *ctx)
}
void io_req_task_work_add_remote(struct io_kiocb *req, unsigned flags);
-struct llist_node *io_handle_tw_list(struct llist_node *node, unsigned int *count, unsigned int max_entries);
void tctx_task_work(struct callback_head *cb);
+void io_tctx_fallback_work(struct work_struct *work);
int io_run_local_work(struct io_ring_ctx *ctx, int min_events, int max_events);
int io_run_task_work_sig(struct io_ring_ctx *ctx);
-__cold void io_fallback_req_func(struct work_struct *work);
-__cold void io_move_task_work_from_local(struct io_ring_ctx *ctx);
+__cold void io_cancel_local_task_work(struct io_ring_ctx *ctx);
int io_run_local_work_locked(struct io_ring_ctx *ctx, int min_events);
void io_req_local_work_add(struct io_kiocb *req, unsigned flags);
void io_req_normal_work_add(struct io_kiocb *req);
-struct llist_node *tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries, unsigned int *count);
+void tctx_task_work_run(struct io_uring_task *tctx, unsigned int max_entries, unsigned int *count);
static inline void __io_req_task_work_add(struct io_kiocb *req, unsigned flags)
{
@@ -89,7 +90,7 @@ static inline int io_run_task_work(void)
static inline bool io_local_work_pending(struct io_ring_ctx *ctx)
{
- return !llist_empty(&ctx->work_llist) || !llist_empty(&ctx->retry_llist);
+ return !mpscq_empty(&ctx->work_list);
}
static inline bool io_task_work_pending(struct io_ring_ctx *ctx)
diff --git a/io_uring/wait.c b/io_uring/wait.c
index d005ea17b35f..a48523c2f356 100644
--- a/io_uring/wait.c
+++ b/io_uring/wait.c
@@ -98,7 +98,7 @@ static enum hrtimer_restart io_cqring_min_timer_wakeup(struct hrtimer *timer)
if (ctx->flags & IORING_SETUP_DEFER_TASKRUN) {
atomic_set(&ctx->cq_wait_nr, 1);
smp_mb();
- if (!llist_empty(&ctx->work_llist))
+ if (io_local_work_pending(ctx))
goto out_wake;
}
diff --git a/io_uring/wait.h b/io_uring/wait.h
index a4274b137f81..b7b9c46b1b01 100644
--- a/io_uring/wait.h
+++ b/io_uring/wait.h
@@ -5,12 +5,14 @@
#include <linux/io_uring_types.h>
/*
- * No waiters. It's larger than any valid value of the tw counter
- * so that tests against ->cq_wait_nr would fail and skip wake_up().
+ * ->cq_wait_nr is armed with the number of lazy task_work adds the waiter
+ * still needs, and counted down by the add side, with the add reaching zero
+ * issuing the (single) wake up for this wait cycle. Zero and below means no
+ * wake up is to be issued: IO_CQ_WAKE_INIT when no task is waiting (also
+ * what a forced wake up resets it to when claiming one), zero once the
+ * countdown has fired.
*/
-#define IO_CQ_WAKE_INIT (-1U)
-/* Forced wake up if there is a waiter regardless of ->cq_wait_nr */
-#define IO_CQ_WAKE_FORCE (IO_CQ_WAKE_INIT >> 1)
+#define IO_CQ_WAKE_INIT (-1)
struct ext_arg {
size_t argsz;
diff --git a/io_uring/zcrx.c b/io_uring/zcrx.c
index 19837e0b5e91..49163f9c39df 100644
--- a/io_uring/zcrx.c
+++ b/io_uring/zcrx.c
@@ -44,6 +44,17 @@ static inline struct io_zcrx_area *io_zcrx_iov_to_area(const struct net_iov *nio
return container_of(owner, struct io_zcrx_area, nia);
}
+static bool zcrx_set_ring_ctx(struct io_zcrx_ifq *zcrx,
+ struct io_ring_ctx *ctx)
+{
+ guard(spinlock_bh)(&zcrx->ctx_lock);
+ if (zcrx->master_ctx)
+ return false;
+ percpu_ref_get(&ctx->refs);
+ zcrx->master_ctx = ctx;
+ return true;
+}
+
static inline struct page *io_zcrx_iov_page(const struct net_iov *niov)
{
struct io_zcrx_area *area = io_zcrx_iov_to_area(niov);
@@ -245,14 +256,13 @@ static void io_release_area_mem(struct io_zcrx_mem *mem)
{
if (mem->is_dmabuf) {
io_release_dmabuf(mem);
- return;
- }
- if (mem->pages) {
+ } else if (mem->pages) {
unpin_user_pages(mem->pages, mem->nr_folios);
sg_free_table(mem->sgt);
- mem->sgt = NULL;
kvfree(mem->pages);
}
+ mem->pages = IO_URING_PTR_POISON;
+ mem->sgt = IO_URING_PTR_POISON;
}
static int io_import_area(struct io_zcrx_ifq *ifq,
@@ -329,7 +339,6 @@ static void zcrx_sync_for_device(struct page_pool *pp, struct io_zcrx_ifq *zcrx,
struct io_zcrx_args {
struct io_kiocb *req;
struct io_zcrx_ifq *ifq;
- struct socket *sock;
unsigned nr_skbs;
};
@@ -403,8 +412,9 @@ static int io_allocate_rbuf_ring(struct io_ring_ctx *ctx,
static void io_free_rbuf_ring(struct io_zcrx_ifq *ifq)
{
io_free_region(ifq->user, &ifq->rq_region);
- ifq->rq.ring = NULL;
- ifq->rq.rqes = NULL;
+ ifq->rq.ring = IO_URING_PTR_POISON;
+ ifq->rq.rqes = IO_URING_PTR_POISON;
+ ifq->notif_stats = IO_URING_PTR_POISON;
}
static void io_zcrx_free_area(struct io_zcrx_ifq *ifq,
@@ -530,6 +540,7 @@ static struct io_zcrx_ifq *io_zcrx_ifq_alloc(struct io_ring_ctx *ctx)
return NULL;
ifq->if_rxq = -1;
+ spin_lock_init(&ifq->ctx_lock);
spin_lock_init(&ifq->rq.lock);
mutex_init(&ifq->pp_lock);
refcount_set(&ifq->refs, 1);
@@ -575,7 +586,12 @@ static void io_close_queue(struct io_zcrx_ifq *ifq)
static void io_zcrx_ifq_free(struct io_zcrx_ifq *ifq)
{
- io_close_queue(ifq);
+ if (WARN_ON_ONCE(ifq->if_rxq != -1))
+ return;
+ if (WARN_ON_ONCE(ifq->netdev != NULL))
+ return;
+ if (WARN_ON_ONCE(ifq->master_ctx))
+ return;
if (ifq->area)
io_zcrx_free_area(ifq, ifq->area);
@@ -652,17 +668,24 @@ static void io_zcrx_scrub(struct io_zcrx_ifq *ifq)
}
}
-static void zcrx_unregister_user(struct io_zcrx_ifq *ifq)
+static void zcrx_unregister_user(struct io_zcrx_ifq *ifq, struct io_ring_ctx *ctx)
{
+ scoped_guard(spinlock_bh, &ifq->ctx_lock) {
+ if (ctx && ifq->master_ctx == ctx) {
+ ifq->master_ctx = NULL;
+ percpu_ref_put(&ctx->refs);
+ }
+ }
+
if (refcount_dec_and_test(&ifq->user_refs)) {
io_close_queue(ifq);
io_zcrx_scrub(ifq);
}
}
-static void zcrx_unregister(struct io_zcrx_ifq *ifq)
+static void zcrx_unregister(struct io_zcrx_ifq *ifq, struct io_ring_ctx *ctx)
{
- zcrx_unregister_user(ifq);
+ zcrx_unregister_user(ifq, ctx);
io_put_zcrx_ifq(ifq);
}
@@ -682,7 +705,7 @@ static int zcrx_box_release(struct inode *inode, struct file *file)
if (WARN_ON_ONCE(!ifq))
return -EFAULT;
- zcrx_unregister(ifq);
+ zcrx_unregister(ifq, NULL);
return 0;
}
@@ -696,19 +719,10 @@ static int zcrx_export(struct io_ring_ctx *ctx, struct io_zcrx_ifq *ifq,
{
struct zcrx_ctrl_export *ce = &ctrl->zc_export;
struct file *file;
- int fd = -1;
+ int fd;
if (!mem_is_zero(ce, sizeof(*ce)))
return -EINVAL;
- fd = get_unused_fd_flags(O_CLOEXEC);
- if (fd < 0)
- return fd;
-
- ce->zcrx_fd = fd;
- if (copy_to_user(arg, ctrl, sizeof(*ctrl))) {
- put_unused_fd(fd);
- return -EFAULT;
- }
refcount_inc(&ifq->refs);
refcount_inc(&ifq->user_refs);
@@ -716,11 +730,23 @@ static int zcrx_export(struct io_ring_ctx *ctx, struct io_zcrx_ifq *ifq,
file = anon_inode_create_getfile("[zcrx]", &zcrx_box_fops,
ifq, O_CLOEXEC, NULL);
if (IS_ERR(file)) {
- put_unused_fd(fd);
- zcrx_unregister(ifq);
+ zcrx_unregister(ifq, NULL);
return PTR_ERR(file);
}
+ fd = get_unused_fd_flags(O_CLOEXEC);
+ if (fd < 0) {
+ fput(file);
+ return fd;
+ }
+
+ ce->zcrx_fd = fd;
+ if (copy_to_user(arg, ctrl, sizeof(*ctrl))) {
+ fput(file);
+ put_unused_fd(fd);
+ return -EFAULT;
+ }
+
fd_install(fd, file);
return 0;
}
@@ -740,6 +766,8 @@ static int import_zcrx(struct io_ring_ctx *ctx,
return -EINVAL;
if (reg->if_rxq || reg->rq_entries || reg->area_ptr || reg->region_ptr)
return -EINVAL;
+ if (reg->notif_desc)
+ return -EINVAL;
if (reg->flags & ~ZCRX_REG_IMPORT)
return -EINVAL;
@@ -780,7 +808,7 @@ err_xa_erase:
scoped_guard(mutex, &ctx->mmap_lock)
xa_erase(&ctx->zcrx_ctxs, id);
err:
- zcrx_unregister(ifq);
+ zcrx_unregister(ifq, ctx);
return ret;
}
@@ -825,9 +853,37 @@ netdev_put_unlock:
return ret;
}
+static int zcrx_validate_notif_stats(struct io_zcrx_ifq *ifq,
+ const struct io_uring_zcrx_ifq_reg *reg,
+ const struct zcrx_notification_desc *notif)
+{
+ size_t stats_off = notif->stats_offset;
+ size_t used, end;
+
+ used = reg->offsets.rqes +
+ sizeof(struct io_uring_zcrx_rqe) * reg->rq_entries;
+
+ if (!IS_ALIGNED(stats_off, __alignof__(struct zcrx_notif_stats)))
+ return -EINVAL;
+ if (stats_off < used)
+ return -ERANGE;
+ if (check_add_overflow(stats_off,
+ sizeof(struct zcrx_notif_stats),
+ &end))
+ return -ERANGE;
+ if (end > io_region_size(&ifq->rq_region))
+ return -ERANGE;
+
+ ifq->notif_stats = io_region_get_ptr(&ifq->rq_region) + stats_off;
+ memset(ifq->notif_stats, 0, sizeof(*ifq->notif_stats));
+
+ return 0;
+}
+
int io_register_zcrx(struct io_ring_ctx *ctx,
struct io_uring_zcrx_ifq_reg __user *arg)
{
+ struct zcrx_notification_desc notif;
struct io_uring_zcrx_area_reg area;
struct io_uring_zcrx_ifq_reg reg;
struct io_uring_region_desc rd;
@@ -871,10 +927,28 @@ int io_register_zcrx(struct io_ring_ctx *ctx,
if (copy_from_user(&area, u64_to_user_ptr(reg.area_ptr), sizeof(area)))
return -EFAULT;
+ memset(&notif, 0, sizeof(notif));
+ if (reg.notif_desc && copy_from_user(&notif, u64_to_user_ptr(reg.notif_desc),
+ sizeof(notif)))
+ return -EFAULT;
+ if (notif.type_mask & ~ZCRX_NOTIF_TYPE_MASK)
+ return -EINVAL;
+ if (notif.flags & ~ZCRX_NOTIF_DESC_FLAG_STATS)
+ return -EINVAL;
+ if (!(notif.flags & ZCRX_NOTIF_DESC_FLAG_STATS)) {
+ if (notif.stats_offset)
+ return -EINVAL;
+ }
+ if (!mem_is_zero(&notif.__resv2, sizeof(notif.__resv2)))
+ return -EINVAL;
+
ifq = io_zcrx_ifq_alloc(ctx);
if (!ifq)
return -ENOMEM;
+ ifq->notif_data = notif.user_data;
+ ifq->allowed_notif_mask = notif.type_mask;
+
if (ctx->user) {
get_uid(ctx->user);
ifq->user = ctx->user;
@@ -896,6 +970,12 @@ int io_register_zcrx(struct io_ring_ctx *ctx,
if (ret)
goto err;
+ if (notif.flags & ZCRX_NOTIF_DESC_FLAG_STATS) {
+ ret = zcrx_validate_notif_stats(ifq, &reg, &notif);
+ if (ret)
+ goto err;
+ }
+
ifq->kern_readable = !(area.flags & IORING_ZCRX_AREA_DMABUF);
if (!(reg.flags & ZCRX_REG_NODEV)) {
@@ -925,12 +1005,15 @@ int io_register_zcrx(struct io_ring_ctx *ctx,
ret = -EFAULT;
goto err;
}
+
+ if (notif.type_mask)
+ zcrx_set_ring_ctx(ifq, ctx);
return 0;
err:
scoped_guard(mutex, &ctx->mmap_lock)
xa_erase(&ctx->zcrx_ctxs, id);
ifq_free:
- zcrx_unregister(ifq);
+ zcrx_unregister(ifq, ctx);
return ret;
}
@@ -960,7 +1043,7 @@ void io_terminate_zcrx(struct io_ring_ctx *ctx)
break;
set_zcrx_entry_mark(ctx, id);
id++;
- zcrx_unregister_user(ifq);
+ zcrx_unregister_user(ifq, ctx);
}
}
@@ -985,6 +1068,12 @@ void io_unregister_zcrx(struct io_ring_ctx *ctx)
}
if (!ifq)
break;
+ /*
+ * io_uring can run requests and return buffers to the user
+ * after termination, scrub it again.
+ */
+ if (refcount_read(&ifq->user_refs) == 0)
+ io_zcrx_scrub(ifq);
io_put_zcrx_ifq(ifq);
}
@@ -1091,6 +1180,53 @@ static unsigned io_zcrx_refill_slow(struct page_pool *pp, struct io_zcrx_ifq *if
return allocated;
}
+static void zcrx_notif_tw(struct io_tw_req tw_req, io_tw_token_t tw)
+{
+ struct io_kiocb *req = tw_req.req;
+ struct io_ring_ctx *ctx = req->ctx;
+
+ io_post_aux_cqe(ctx, req->cqe.user_data, req->cqe.res, 0);
+ percpu_ref_put(&ctx->refs);
+ io_poison_req(req);
+ kmem_cache_free(req_cachep, req);
+}
+
+static void zcrx_stat_add(__u64 *p, s64 v)
+{
+ WRITE_ONCE(*p, READ_ONCE(*p) + v);
+}
+
+static void zcrx_send_notif(struct io_zcrx_ifq *ifq, unsigned type)
+{
+ gfp_t gfp = GFP_ATOMIC | __GFP_NOWARN | __GFP_ZERO;
+ u32 type_mask = 1 << type;
+ struct io_kiocb *req;
+
+ if (!(type_mask & ifq->allowed_notif_mask))
+ return;
+
+ guard(spinlock_bh)(&ifq->ctx_lock);
+ if (!ifq->master_ctx)
+ return;
+ if (type_mask & ifq->fired_notifs)
+ return;
+
+ req = kmem_cache_alloc(req_cachep, gfp);
+ if (unlikely(!req))
+ return;
+
+ ifq->fired_notifs |= type_mask;
+
+ req->opcode = IORING_OP_NOP;
+ req->cqe.user_data = ifq->notif_data;
+ req->cqe.res = type;
+ req->ctx = ifq->master_ctx;
+ percpu_ref_get(&req->ctx->refs);
+ req->tctx = NULL;
+ req->io_task_work.func = zcrx_notif_tw;
+ io_req_task_work_add(req);
+}
+
static netmem_ref io_pp_zc_alloc_netmems(struct page_pool *pp, gfp_t gfp)
{
struct io_zcrx_ifq *ifq = io_pp_to_ifq(pp);
@@ -1107,8 +1243,10 @@ static netmem_ref io_pp_zc_alloc_netmems(struct page_pool *pp, gfp_t gfp)
goto out_return;
allocated = io_zcrx_refill_slow(pp, ifq, netmems, to_alloc);
- if (!allocated)
+ if (!allocated) {
+ zcrx_send_notif(ifq, ZCRX_NOTIF_NO_BUFFERS);
return 0;
+ }
out_return:
zcrx_sync_for_device(pp, ifq, netmems, allocated);
allocated--;
@@ -1257,12 +1395,32 @@ static int zcrx_flush_rq(struct io_ring_ctx *ctx, struct io_zcrx_ifq *zcrx,
return 0;
}
+static int zcrx_arm_notif(struct io_ring_ctx *ctx, struct io_zcrx_ifq *zcrx,
+ struct zcrx_ctrl *ctrl)
+{
+ const struct zcrx_ctrl_arm_notif *an = &ctrl->zc_arm_notif;
+ unsigned type_mask;
+
+ if (an->notif_type >= __ZCRX_NOTIF_TYPE_LAST)
+ return -EINVAL;
+ if (!mem_is_zero(&an->__resv, sizeof(an->__resv)))
+ return -EINVAL;
+
+ guard(spinlock_bh)(&zcrx->ctx_lock);
+ type_mask = 1U << an->notif_type;
+ if (type_mask & ~zcrx->fired_notifs)
+ return -EINVAL;
+ zcrx->fired_notifs &= ~type_mask;
+ return 0;
+}
+
int io_zcrx_ctrl(struct io_ring_ctx *ctx, void __user *arg, unsigned nr_args)
{
struct zcrx_ctrl ctrl;
struct io_zcrx_ifq *zcrx;
BUILD_BUG_ON(sizeof(ctrl.zc_export) != sizeof(ctrl.zc_flush));
+ BUILD_BUG_ON(sizeof(ctrl.zc_export) != sizeof(ctrl.zc_arm_notif));
if (nr_args)
return -EINVAL;
@@ -1280,6 +1438,8 @@ int io_zcrx_ctrl(struct io_ring_ctx *ctx, void __user *arg, unsigned nr_args)
return zcrx_flush_rq(ctx, zcrx, &ctrl);
case ZCRX_CTRL_EXPORT:
return zcrx_export(ctx, zcrx, &ctrl, arg);
+ case ZCRX_CTRL_ARM_NOTIFICATION:
+ return zcrx_arm_notif(ctx, zcrx, &ctrl);
}
return -EOPNOTSUPP;
@@ -1416,8 +1576,18 @@ static int io_zcrx_copy_frag(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
const skb_frag_t *frag, int off, int len)
{
struct page *page = skb_frag_page(frag);
+ int ret;
- return io_zcrx_copy_chunk(req, ifq, page, off + skb_frag_off(frag), len);
+ ret = io_zcrx_copy_chunk(req, ifq, page, off + skb_frag_off(frag), len);
+ if (ret > 0) {
+ if (ifq->notif_stats) {
+ zcrx_stat_add(&ifq->notif_stats->copy_count, 1);
+ zcrx_stat_add(&ifq->notif_stats->copy_bytes, ret);
+ }
+ zcrx_send_notif(ifq, ZCRX_NOTIF_COPY);
+ }
+
+ return ret;
}
static int io_zcrx_recv_frag(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
@@ -1562,7 +1732,6 @@ static int io_zcrx_tcp_recvmsg(struct io_kiocb *req, struct io_zcrx_ifq *ifq,
struct io_zcrx_args args = {
.req = req,
.ifq = ifq,
- .sock = sk->sk_socket,
};
read_descriptor_t rd_desc = {
.count = len ? len : UINT_MAX,
diff --git a/io_uring/zcrx.h b/io_uring/zcrx.h
index 75e0a4e6ef6e..fa00900e479e 100644
--- a/io_uring/zcrx.h
+++ b/io_uring/zcrx.h
@@ -9,7 +9,9 @@
#include <net/net_trackers.h>
#define ZCRX_SUPPORTED_REG_FLAGS (ZCRX_REG_IMPORT | ZCRX_REG_NODEV)
-#define ZCRX_FEATURES (ZCRX_FEATURE_RX_PAGE_SIZE)
+#define ZCRX_FEATURES (ZCRX_FEATURE_RX_PAGE_SIZE |\
+ ZCRX_FEATURE_NOTIFICATION)
+#define ZCRX_NOTIF_TYPE_MASK ((1U << ZCRX_NOTIF_NO_BUFFERS) | (1U << ZCRX_NOTIF_COPY))
struct io_zcrx_mem {
unsigned long size;
@@ -72,6 +74,13 @@ struct io_zcrx_ifq {
*/
struct mutex pp_lock;
struct io_mapped_region rq_region;
+
+ spinlock_t ctx_lock;
+ struct io_ring_ctx *master_ctx;
+ u32 allowed_notif_mask;
+ u32 fired_notifs;
+ u64 notif_data;
+ struct zcrx_notif_stats *notif_stats;
};
#if defined(CONFIG_IO_URING_ZCRX)