tmakin/ack-tegra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge 96f7e448b9 ("Merge tag 'for-6.2/io_uring-next-2022-12-08' of git://git.kernel.dk/linux") into android-mainline

Browse Source 
Steps on the way to 6.2-rc1

Change-Id: I7b32fd3894e49b386ab1d7588279abdddee184b9
Signed-off-by: Greg Kroah-Hartman <gregkh@google.com>
This commit is contained in:
Greg Kroah-Hartman
2023-01-20 14:37:54 +00:00
parent 91b233e570 96f7e448b9
commit ecd7f3d203
24 changed files with 711 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fs/eventfd.c
+21 -16
View File
@@ -43,21 +43,7 @@ struct eventfd_ctx {
int id;
};
/**
* eventfd_signal - Adds @n to the eventfd counter.
* @ctx: [in] Pointer to the eventfd context.
* @n: [in] Value of the counter to be added to the eventfd internal counter.
* The value cannot be negative.
*
* This function is supposed to be called by the kernel in paths that do not
* allow sleeping. In this function we allow the counter to reach the ULLONG_MAX
* value, and we signal this as overflow condition by returning a EPOLLERR
* to poll(2).
*
* Returns the amount by which the counter was incremented. This will be less
* than @n if the counter has overflowed.
*/
__u64 eventfd_signal(struct eventfd_ctx *ctx, __u64 n)
__u64 eventfd_signal_mask(struct eventfd_ctx *ctx, __u64 n, unsigned mask)
{
unsigned long flags;
@@ -78,12 +64,31 @@ __u64 eventfd_signal(struct eventfd_ctx *ctx, __u64 n)
n = ULLONG_MAX - ctx->count;
ctx->count += n;
if (waitqueue_active(&ctx->wqh))
wake_up_locked_poll(&ctx->wqh, EPOLLIN);
wake_up_locked_poll(&ctx->wqh, EPOLLIN | mask);
current->in_eventfd = 0;
spin_unlock_irqrestore(&ctx->wqh.lock, flags);
return n;
}
/**
* eventfd_signal - Adds @n to the eventfd counter.
* @ctx: [in] Pointer to the eventfd context.
* @n: [in] Value of the counter to be added to the eventfd internal counter.
* The value cannot be negative.
*
* This function is supposed to be called by the kernel in paths that do not
* allow sleeping. In this function we allow the counter to reach the ULLONG_MAX
* value, and we signal this as overflow condition by returning a EPOLLERR
* to poll(2).
*
* Returns the amount by which the counter was incremented. This will be less
* than @n if the counter has overflowed.
*/
__u64 eventfd_signal(struct eventfd_ctx *ctx, __u64 n)
{
return eventfd_signal_mask(ctx, n, 0);
}
EXPORT_SYMBOL_GPL(eventfd_signal);
static void eventfd_free_ctx(struct eventfd_ctx *ctx)
fs/eventpoll.c
+10 -8
View File
@@ -491,7 +491,8 @@ static inline void ep_set_busy_poll_napi_id(struct epitem *epi)
*/
#ifdef CONFIG_DEBUG_LOCK_ALLOC
static void ep_poll_safewake(struct eventpoll *ep, struct epitem *epi)
static void ep_poll_safewake(struct eventpoll *ep, struct epitem *epi,
unsigned pollflags)
{
struct eventpoll *ep_src;
unsigned long flags;
@@ -522,16 +523,17 @@ static void ep_poll_safewake(struct eventpoll *ep, struct epitem *epi)
}
spin_lock_irqsave_nested(&ep->poll_wait.lock, flags, nests);
ep->nests = nests + 1;
wake_up_locked_poll(&ep->poll_wait, EPOLLIN);
wake_up_locked_poll(&ep->poll_wait, EPOLLIN | pollflags);
ep->nests = 0;
spin_unlock_irqrestore(&ep->poll_wait.lock, flags);
}
#else
static void ep_poll_safewake(struct eventpoll *ep, struct epitem *epi)
static void ep_poll_safewake(struct eventpoll *ep, struct epitem *epi,
unsigned pollflags)
{
wake_up_poll(&ep->poll_wait, EPOLLIN);
wake_up_poll(&ep->poll_wait, EPOLLIN | pollflags);
}
#endif
@@ -742,7 +744,7 @@ static void ep_free(struct eventpoll *ep)
/* We need to release all tasks waiting for these file */
if (waitqueue_active(&ep->poll_wait))
ep_poll_safewake(ep, NULL);
ep_poll_safewake(ep, NULL, 0);
/*
* We need to lock this because we could be hit by
@@ -1208,7 +1210,7 @@ out_unlock:
/* We have to call this outside the lock */
if (pwake)
ep_poll_safewake(ep, epi);
ep_poll_safewake(ep, epi, pollflags & EPOLL_URING_WAKE);
if (!(epi->event.events & EPOLLEXCLUSIVE))
ewake = 1;
@@ -1553,7 +1555,7 @@ static int ep_insert(struct eventpoll *ep, const struct epoll_event *event,
/* We have to call this outside the lock */
if (pwake)
ep_poll_safewake(ep, NULL);
ep_poll_safewake(ep, NULL, 0);
return 0;
}
@@ -1629,7 +1631,7 @@ static int ep_modify(struct eventpoll *ep, struct epitem *epi,
/* We have to call this outside the lock */
if (pwake)
ep_poll_safewake(ep, NULL);
ep_poll_safewake(ep, NULL, 0);
return 0;
}
fs/iomap/direct-io.c
+1 -2
View File
@@ -240,7 +240,6 @@ static loff_t iomap_dio_bio_iter(const struct iomap_iter *iter,
{
const struct iomap *iomap = &iter->iomap;
struct inode *inode = iter->inode;
unsigned int blkbits = blksize_bits(bdev_logical_block_size(iomap->bdev));
unsigned int fs_block_size = i_blocksize(inode), pad;
loff_t length = iomap_length(iter);
loff_t pos = iter->pos;
@@ -252,7 +251,7 @@ static loff_t iomap_dio_bio_iter(const struct iomap_iter *iter,
size_t copied = 0;
size_t orig_count;
if ((pos | length) & ((1 << blkbits) - 1) ||
if ((pos | length) & (bdev_logical_block_size(iomap->bdev) - 1) ||
!bdev_iter_is_aligned(iomap->bdev, dio->submit.iter))
return -EINVAL;
include/linux/eventfd.h
+7
View File
@@ -40,6 +40,7 @@ struct file *eventfd_fget(int fd);
struct eventfd_ctx *eventfd_ctx_fdget(int fd);
struct eventfd_ctx *eventfd_ctx_fileget(struct file *file);
__u64 eventfd_signal(struct eventfd_ctx *ctx, __u64 n);
__u64 eventfd_signal_mask(struct eventfd_ctx *ctx, __u64 n, unsigned mask);
int eventfd_ctx_remove_wait_queue(struct eventfd_ctx *ctx, wait_queue_entry_t *wait,
__u64 *cnt);
void eventfd_ctx_do_read(struct eventfd_ctx *ctx, __u64 *cnt);
@@ -66,6 +67,12 @@ static inline int eventfd_signal(struct eventfd_ctx *ctx, __u64 n)
return -ENOSYS;
}
static inline int eventfd_signal_mask(struct eventfd_ctx *ctx, __u64 n,
unsigned mask)
{
return -ENOSYS;
}
static inline void eventfd_ctx_put(struct eventfd_ctx *ctx)
{
include/linux/io_uring.h
+7 -6
View File
@@ -9,16 +9,17 @@
enum io_uring_cmd_flags {
IO_URING_F_COMPLETE_DEFER = 1,
IO_URING_F_UNLOCKED = 2,
/* the request is executed from poll, it should not be freed */
IO_URING_F_MULTISHOT = 4,
/* executed by io-wq */
IO_URING_F_IOWQ = 8,
/* int's last bit, sign checks are usually faster than a bit test */
IO_URING_F_NONBLOCK = INT_MIN,
/* ctx state flags, for URING_CMD */
IO_URING_F_SQE128 = 4,
IO_URING_F_CQE32 = 8,
IO_URING_F_IOPOLL = 16,
/* the request is executed from poll, it should not be freed */
IO_URING_F_MULTISHOT = 32,
IO_URING_F_SQE128 = (1 << 8),
IO_URING_F_CQE32 = (1 << 9),
IO_URING_F_IOPOLL = (1 << 10),
};
struct io_uring_cmd {
include/linux/io_uring_types.h
+5
View File
@@ -174,7 +174,9 @@ struct io_submit_state {
bool plug_started;
bool need_plug;
unsigned short submit_nr;
unsigned int cqes_count;
struct blk_plug plug;
struct io_uring_cqe cqes[16];
};
struct io_ev_fd {
@@ -206,6 +208,8 @@ struct io_ring_ctx {
unsigned int drain_disabled: 1;
unsigned int has_evfd: 1;
unsigned int syscall_iopoll: 1;
/* all CQEs should be posted only by the submitter task */
unsigned int task_complete: 1;
} ____cacheline_aligned_in_smp;
/* submission data */
@@ -324,6 +328,7 @@ struct io_ring_ctx {
struct io_rsrc_data *buf_data;
struct delayed_work rsrc_put_work;
struct callback_head rsrc_put_tw;
struct llist_head rsrc_put_llist;
struct list_head rsrc_ref_list;
spinlock_t rsrc_ref_lock;
include/uapi/linux/eventpoll.h
+6
View File
@@ -41,6 +41,12 @@
#define EPOLLMSG (__force __poll_t)0x00000400
#define EPOLLRDHUP (__force __poll_t)0x00002000
/*
* Internal flag - wakeup generated by io_uring, used to detect recursion back
* into the io_uring poll handler.
*/
#define EPOLL_URING_WAKE ((__force __poll_t)(1U << 27))
/* Set exclusive wakeup mode for the target file descriptor */
#define EPOLLEXCLUSIVE ((__force __poll_t)(1U << 28))
include/uapi/linux/io_uring.h
+18
View File
@@ -296,10 +296,28 @@ enum io_uring_op {
*
* IORING_RECVSEND_FIXED_BUF Use registered buffers, the index is stored in
* the buf_index field.
*
* IORING_SEND_ZC_REPORT_USAGE
* If set, SEND[MSG]_ZC should report
* the zerocopy usage in cqe.res
* for the IORING_CQE_F_NOTIF cqe.
* 0 is reported if zerocopy was actually possible.
* IORING_NOTIF_USAGE_ZC_COPIED if data was copied
* (at least partially).
*/
#define IORING_RECVSEND_POLL_FIRST (1U << 0)
#define IORING_RECV_MULTISHOT (1U << 1)
#define IORING_RECVSEND_FIXED_BUF (1U << 2)
#define IORING_SEND_ZC_REPORT_USAGE (1U << 3)
/*
* cqe.res for IORING_CQE_F_NOTIF if
* IORING_SEND_ZC_REPORT_USAGE was requested
*
* It should be treated as a flag, all other
* bits of cqe.res should be treated as reserved!
*/
#define IORING_NOTIF_USAGE_ZC_COPIED (1U << 31)
/*
* accept flags stored in sqe->ioprio
io_uring/io_uring.c
+225 -131
View File
@@ -149,6 +149,7 @@ static void io_clean_op(struct io_kiocb *req);
static void io_queue_sqe(struct io_kiocb *req);
static void io_move_task_work_from_local(struct io_ring_ctx *ctx);
static void __io_submit_flush_completions(struct io_ring_ctx *ctx);
static __cold void io_fallback_tw(struct io_uring_task *tctx);
static struct kmem_cache *req_cachep;
@@ -167,7 +168,8 @@ EXPORT_SYMBOL(io_uring_get_socket);
static inline void io_submit_flush_completions(struct io_ring_ctx *ctx)
{
if (!wq_list_empty(&ctx->submit_state.compl_reqs))
if (!wq_list_empty(&ctx->submit_state.compl_reqs) ||
ctx->submit_state.cqes_count)
__io_submit_flush_completions(ctx);
}
@@ -325,6 +327,7 @@ static __cold struct io_ring_ctx *io_ring_ctx_alloc(struct io_uring_params *p)
spin_lock_init(&ctx->rsrc_ref_lock);
INIT_LIST_HEAD(&ctx->rsrc_ref_list);
INIT_DELAYED_WORK(&ctx->rsrc_put_work, io_rsrc_put_work);
init_task_work(&ctx->rsrc_put_tw, io_rsrc_put_tw);
init_llist_head(&ctx->rsrc_put_llist);
init_llist_head(&ctx->work_llist);
INIT_LIST_HEAD(&ctx->tctx_list);
@@ -495,7 +498,7 @@ static void io_eventfd_ops(struct rcu_head *rcu)
int ops = atomic_xchg(&ev_fd->ops, 0);
if (ops & BIT(IO_EVENTFD_OP_SIGNAL_BIT))
eventfd_signal(ev_fd->cq_ev_fd, 1);
eventfd_signal_mask(ev_fd->cq_ev_fd, 1, EPOLL_URING_WAKE);
/* IO_EVENTFD_OP_FREE_BIT may not be set here depending on callback
* ordering in a race but if references are 0 we know we have to free
@@ -531,7 +534,7 @@ static void io_eventfd_signal(struct io_ring_ctx *ctx)
goto out;
if (likely(eventfd_signal_allowed())) {
eventfd_signal(ev_fd->cq_ev_fd, 1);
eventfd_signal_mask(ev_fd->cq_ev_fd, 1, EPOLL_URING_WAKE);
} else {
atomic_inc(&ev_fd->refs);
if (!atomic_fetch_or(BIT(IO_EVENTFD_OP_SIGNAL_BIT), &ev_fd->ops))
@@ -581,33 +584,63 @@ void __io_commit_cqring_flush(struct io_ring_ctx *ctx)
io_eventfd_flush_signal(ctx);
}
static inline void io_cqring_ev_posted(struct io_ring_ctx *ctx)
static inline void __io_cq_lock(struct io_ring_ctx *ctx)
__acquires(ctx->completion_lock)
{
io_commit_cqring_flush(ctx);
io_cqring_wake(ctx);
if (!ctx->task_complete)
spin_lock(&ctx->completion_lock);
}
static inline void __io_cq_unlock(struct io_ring_ctx *ctx)
{
if (!ctx->task_complete)
spin_unlock(&ctx->completion_lock);
}
/* keep it inlined for io_submit_flush_completions() */
static inline void __io_cq_unlock_post(struct io_ring_ctx *ctx)
__releases(ctx->completion_lock)
{
io_commit_cqring(ctx);
spin_unlock(&ctx->completion_lock);
io_cqring_ev_posted(ctx);
__io_cq_unlock(ctx);
io_commit_cqring_flush(ctx);
io_cqring_wake(ctx);
}
void io_cq_unlock_post(struct io_ring_ctx *ctx)
__releases(ctx->completion_lock)
{
__io_cq_unlock_post(ctx);
io_commit_cqring(ctx);
spin_unlock(&ctx->completion_lock);
io_commit_cqring_flush(ctx);
io_cqring_wake(ctx);
}
/* Returns true if there are no backlogged entries after the flush */
static bool __io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force)
static void io_cqring_overflow_kill(struct io_ring_ctx *ctx)
{
struct io_overflow_cqe *ocqe;
LIST_HEAD(list);
io_cq_lock(ctx);
list_splice_init(&ctx->cq_overflow_list, &list);
clear_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq);
io_cq_unlock(ctx);
while (!list_empty(&list)) {
ocqe = list_first_entry(&list, struct io_overflow_cqe, list);
list_del(&ocqe->list);
kfree(ocqe);
}
}
/* Returns true if there are no backlogged entries after the flush */
static void __io_cqring_overflow_flush(struct io_ring_ctx *ctx)
{
bool all_flushed;
size_t cqe_size = sizeof(struct io_uring_cqe);
if (!force && __io_cqring_events(ctx) == ctx->cq_entries)
return false;
if (__io_cqring_events(ctx) == ctx->cq_entries)
return;
if (ctx->flags & IORING_SETUP_CQE32)
cqe_size <<= 1;
@@ -617,43 +650,32 @@ static bool __io_cqring_overflow_flush(struct io_ring_ctx *ctx, bool force)
struct io_uring_cqe *cqe = io_get_cqe_overflow(ctx, true);
struct io_overflow_cqe *ocqe;
if (!cqe && !force)
if (!cqe)
break;
ocqe = list_first_entry(&ctx->cq_overflow_list,
struct io_overflow_cqe, list);
if (cqe)
memcpy(cqe, &ocqe->cqe, cqe_size);
else
io_account_cq_overflow(ctx);
memcpy(cqe, &ocqe->cqe, cqe_size);
list_del(&ocqe->list);
kfree(ocqe);
}
all_flushed = list_empty(&ctx->cq_overflow_list);
if (all_flushed) {
if (list_empty(&ctx->cq_overflow_list)) {
clear_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq);
atomic_andnot(IORING_SQ_CQ_OVERFLOW, &ctx->rings->sq_flags);
}
io_cq_unlock_post(ctx);
return all_flushed;
}
static bool io_cqring_overflow_flush(struct io_ring_ctx *ctx)
static void io_cqring_overflow_flush(struct io_ring_ctx *ctx)
{
bool ret = true;
if (test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq)) {
/* iopoll syncs against uring_lock, not completion_lock */
if (ctx->flags & IORING_SETUP_IOPOLL)
mutex_lock(&ctx->uring_lock);
ret = __io_cqring_overflow_flush(ctx, false);
__io_cqring_overflow_flush(ctx);
if (ctx->flags & IORING_SETUP_IOPOLL)
mutex_unlock(&ctx->uring_lock);
}
return ret;
}
void __io_put_task(struct task_struct *task, int nr)
@@ -778,11 +800,14 @@ struct io_uring_cqe *__io_get_cqe(struct io_ring_ctx *ctx, bool overflow)
return &rings->cqes[off];
}
bool io_fill_cqe_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags,
bool allow_overflow)
static bool io_fill_cqe_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res,
u32 cflags)
{
struct io_uring_cqe *cqe;
if (!ctx->task_complete)
lockdep_assert_held(&ctx->completion_lock);
ctx->cq_extra++;
/*
@@ -804,34 +829,100 @@ bool io_fill_cqe_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags
}
return true;
}
if (allow_overflow)
return io_cqring_event_overflow(ctx, user_data, res, cflags, 0, 0);
return false;
}
bool io_post_aux_cqe(struct io_ring_ctx *ctx,
u64 user_data, s32 res, u32 cflags,
bool allow_overflow)
static void __io_flush_post_cqes(struct io_ring_ctx *ctx)
__must_hold(&ctx->uring_lock)
{
struct io_submit_state *state = &ctx->submit_state;
unsigned int i;
lockdep_assert_held(&ctx->uring_lock);
for (i = 0; i < state->cqes_count; i++) {
struct io_uring_cqe *cqe = &state->cqes[i];
if (!io_fill_cqe_aux(ctx, cqe->user_data, cqe->res, cqe->flags)) {
if (ctx->task_complete) {
spin_lock(&ctx->completion_lock);
io_cqring_event_overflow(ctx, cqe->user_data,
cqe->res, cqe->flags, 0, 0);
spin_unlock(&ctx->completion_lock);
} else {
io_cqring_event_overflow(ctx, cqe->user_data,
cqe->res, cqe->flags, 0, 0);
}
}
}
state->cqes_count = 0;
}
static bool __io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags,
bool allow_overflow)
{
bool filled;
io_cq_lock(ctx);
filled = io_fill_cqe_aux(ctx, user_data, res, cflags, allow_overflow);
filled = io_fill_cqe_aux(ctx, user_data, res, cflags);
if (!filled && allow_overflow)
filled = io_cqring_event_overflow(ctx, user_data, res, cflags, 0, 0);
io_cq_unlock_post(ctx);
return filled;
}
static void __io_req_complete_put(struct io_kiocb *req)
bool io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags)
{
return __io_post_aux_cqe(ctx, user_data, res, cflags, true);
}
bool io_aux_cqe(struct io_ring_ctx *ctx, bool defer, u64 user_data, s32 res, u32 cflags,
bool allow_overflow)
{
struct io_uring_cqe *cqe;
unsigned int length;
if (!defer)
return __io_post_aux_cqe(ctx, user_data, res, cflags, allow_overflow);
length = ARRAY_SIZE(ctx->submit_state.cqes);
lockdep_assert_held(&ctx->uring_lock);
if (ctx->submit_state.cqes_count == length) {
__io_cq_lock(ctx);
__io_flush_post_cqes(ctx);
/* no need to flush - flush is deferred */
__io_cq_unlock_post(ctx);
}
/* For defered completions this is not as strict as it is otherwise,
* however it's main job is to prevent unbounded posted completions,
* and in that it works just as well.
*/
if (!allow_overflow && test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq))
return false;
cqe = &ctx->submit_state.cqes[ctx->submit_state.cqes_count++];
cqe->user_data = user_data;
cqe->res = res;
cqe->flags = cflags;
return true;
}
static void __io_req_complete_post(struct io_kiocb *req)
{
struct io_ring_ctx *ctx = req->ctx;
io_cq_lock(ctx);
if (!(req->flags & REQ_F_CQE_SKIP))
__io_fill_cqe_req(ctx, req);
/*
* If we're the last reference to this request, add to our locked
* free_list cache.
*/
if (req_ref_put_and_test(req)) {
struct io_ring_ctx *ctx = req->ctx;
if (req->flags & IO_REQ_LINK_FLAGS) {
if (req->flags & IO_DISARM_MASK)
io_disarm_next(req);
@@ -852,38 +943,38 @@ static void __io_req_complete_put(struct io_kiocb *req)
wq_list_add_head(&req->comp_list, &ctx->locked_free_list);
ctx->locked_free_nr++;
}
}
void __io_req_complete_post(struct io_kiocb *req)
{
if (!(req->flags & REQ_F_CQE_SKIP))
__io_fill_cqe_req(req->ctx, req);
__io_req_complete_put(req);
}
void io_req_complete_post(struct io_kiocb *req)
{
struct io_ring_ctx *ctx = req->ctx;
io_cq_lock(ctx);
__io_req_complete_post(req);
io_cq_unlock_post(ctx);
}
inline void __io_req_complete(struct io_kiocb *req, unsigned issue_flags)
void io_req_complete_post(struct io_kiocb *req, unsigned issue_flags)
{
io_req_complete_post(req);
if (req->ctx->task_complete && (issue_flags & IO_URING_F_IOWQ)) {
req->io_task_work.func = io_req_task_complete;
io_req_task_work_add(req);
} else if (!(issue_flags & IO_URING_F_UNLOCKED) ||
!(req->ctx->flags & IORING_SETUP_IOPOLL)) {
__io_req_complete_post(req);
} else {
struct io_ring_ctx *ctx = req->ctx;
mutex_lock(&ctx->uring_lock);
__io_req_complete_post(req);
mutex_unlock(&ctx->uring_lock);
}
}
void io_req_complete_failed(struct io_kiocb *req, s32 res)
void io_req_defer_failed(struct io_kiocb *req, s32 res)
__must_hold(&ctx->uring_lock)
{
const struct io_op_def *def = &io_op_defs[req->opcode];
lockdep_assert_held(&req->ctx->uring_lock);
req_set_fail(req);
io_req_set_res(req, res, io_put_kbuf(req, IO_URING_F_UNLOCKED));
if (def->fail)
def->fail(req);
io_req_complete_post(req);
io_req_complete_defer(req);
}
/*
@@ -1084,10 +1175,17 @@ void tctx_task_work(struct callback_head *cb)
struct io_uring_task *tctx = container_of(cb, struct io_uring_task,
task_work);
struct llist_node fake = {};
struct llist_node *node = io_llist_xchg(&tctx->task_list, &fake);
struct llist_node *node;
unsigned int loops = 1;
unsigned int count = handle_tw_list(node, &ctx, &uring_locked, NULL);
unsigned int count;
if (unlikely(current->flags & PF_EXITING)) {
io_fallback_tw(tctx);
return;
}
node = io_llist_xchg(&tctx->task_list, &fake);
count = handle_tw_list(node, &ctx, &uring_locked, NULL);
node = io_llist_cmpxchg(&tctx->task_list, &fake, NULL);
while (node != &fake) {
loops++;
@@ -1105,6 +1203,20 @@ void tctx_task_work(struct callback_head *cb)
trace_io_uring_task_work_run(tctx, count, loops);
}
static __cold void io_fallback_tw(struct io_uring_task *tctx)
{
struct llist_node *node = llist_del_all(&tctx->task_list);
struct io_kiocb *req;
while (node) {
req = container_of(node, struct io_kiocb, io_task_work.node);
node = node->next;
if (llist_add(&req->io_task_work.node,
&req->ctx->fallback_llist))
schedule_delayed_work(&req->ctx->fallback_work, 1);
}
}
static void io_req_local_work_add(struct io_kiocb *req)
{
struct io_ring_ctx *ctx = req->ctx;
@@ -1127,11 +1239,10 @@ static void io_req_local_work_add(struct io_kiocb *req)
__io_cqring_wake(ctx);
}
static inline void __io_req_task_work_add(struct io_kiocb *req, bool allow_local)
void __io_req_task_work_add(struct io_kiocb *req, bool allow_local)
{
struct io_uring_task *tctx = req->task->io_uring;
struct io_ring_ctx *ctx = req->ctx;
struct llist_node *node;
if (allow_local && ctx->flags & IORING_SETUP_DEFER_TASKRUN) {
io_req_local_work_add(req);
@@ -1148,20 +1259,7 @@ static inline void __io_req_task_work_add(struct io_kiocb *req, bool allow_local
if (likely(!task_work_add(req->task, &tctx->task_work, ctx->notify_method)))
return;
node = llist_del_all(&tctx->task_list);
while (node) {
req = container_of(node, struct io_kiocb, io_task_work.node);
node = node->next;
if (llist_add(&req->io_task_work.node,
&req->ctx->fallback_llist))
schedule_delayed_work(&req->ctx->fallback_work, 1);
}
}
void io_req_task_work_add(struct io_kiocb *req)
{
__io_req_task_work_add(req, true);
io_fallback_tw(tctx);
}
static void __cold io_move_task_work_from_local(struct io_ring_ctx *ctx)
@@ -1237,23 +1335,10 @@ int io_run_local_work(struct io_ring_ctx *ctx)
return ret;
}
static void io_req_tw_post(struct io_kiocb *req, bool *locked)
{
io_req_complete_post(req);
}
void io_req_tw_post_queue(struct io_kiocb *req, s32 res, u32 cflags)
{
io_req_set_res(req, res, cflags);
req->io_task_work.func = io_req_tw_post;
io_req_task_work_add(req);
}
static void io_req_task_cancel(struct io_kiocb *req, bool *locked)
{
/* not needed for normal modes, but SQPOLL depends on it */
io_tw_lock(req->ctx, locked);
io_req_complete_failed(req, req->cqe.res);
io_req_defer_failed(req, req->cqe.res);
}
void io_req_task_submit(struct io_kiocb *req, bool *locked)
@@ -1263,7 +1348,7 @@ void io_req_task_submit(struct io_kiocb *req, bool *locked)
if (likely(!(req->task->flags & PF_EXITING)))
io_queue_sqe(req);
else
io_req_complete_failed(req, -EFAULT);
io_req_defer_failed(req, -EFAULT);
}
void io_req_task_queue_fail(struct io_kiocb *req, int ret)
@@ -1343,18 +1428,31 @@ static void __io_submit_flush_completions(struct io_ring_ctx *ctx)
struct io_wq_work_node *node, *prev;
struct io_submit_state *state = &ctx->submit_state;
io_cq_lock(ctx);
__io_cq_lock(ctx);
/* must come first to preserve CQE ordering in failure cases */
if (state->cqes_count)
__io_flush_post_cqes(ctx);
wq_list_for_each(node, prev, &state->compl_reqs) {
struct io_kiocb *req = container_of(node, struct io_kiocb,
comp_list);
if (!(req->flags & REQ_F_CQE_SKIP))
__io_fill_cqe_req(ctx, req);
if (!(req->flags & REQ_F_CQE_SKIP) &&
unlikely(!__io_fill_cqe_req(ctx, req))) {
if (ctx->task_complete) {
spin_lock(&ctx->completion_lock);
io_req_cqe_overflow(req);
spin_unlock(&ctx->completion_lock);
} else {
io_req_cqe_overflow(req);
}
}
}
__io_cq_unlock_post(ctx);
io_free_batch_list(ctx, state->compl_reqs.first);
INIT_WQ_LIST(&state->compl_reqs);
if (!wq_list_empty(&ctx->submit_state.compl_reqs)) {
io_free_batch_list(ctx, state->compl_reqs.first);
INIT_WQ_LIST(&state->compl_reqs);
}
}
/*
@@ -1420,7 +1518,7 @@ static int io_iopoll_check(struct io_ring_ctx *ctx, long min)
check_cq = READ_ONCE(ctx->check_cq);
if (unlikely(check_cq)) {
if (check_cq & BIT(IO_CHECK_CQ_OVERFLOW_BIT))
__io_cqring_overflow_flush(ctx, false);
__io_cqring_overflow_flush(ctx);
/*
* Similarly do not spin if we have not informed the user of any
* dropped CQE.
@@ -1476,16 +1574,10 @@ static int io_iopoll_check(struct io_ring_ctx *ctx, long min)
void io_req_task_complete(struct io_kiocb *req, bool *locked)
{
if (req->flags & (REQ_F_BUFFER_SELECTED|REQ_F_BUFFER_RING)) {
unsigned issue_flags = *locked ? 0 : IO_URING_F_UNLOCKED;
req->cqe.flags |= io_put_kbuf(req, issue_flags);
}
if (*locked)
io_req_complete_defer(req);
else
io_req_complete_post(req);
io_req_complete_post(req, IO_URING_F_UNLOCKED);
}
/*
@@ -1635,6 +1727,7 @@ static u32 io_get_sequence(struct io_kiocb *req)
}
static __cold void io_drain_req(struct io_kiocb *req)
__must_hold(&ctx->uring_lock)
{
struct io_ring_ctx *ctx = req->ctx;
struct io_defer_entry *de;
@@ -1655,7 +1748,7 @@ queue:
ret = io_req_prep_async(req);
if (ret) {
fail:
io_req_complete_failed(req, ret);
io_req_defer_failed(req, ret);
return;
}
io_prep_async_link(req);
@@ -1752,12 +1845,12 @@ static int io_issue_sqe(struct io_kiocb *req, unsigned int issue_flags)
if (issue_flags & IO_URING_F_COMPLETE_DEFER)
io_req_complete_defer(req);
else
io_req_complete_post(req);
io_req_complete_post(req, issue_flags);
} else if (ret != IOU_ISSUE_SKIP_COMPLETE)
return ret;
/* If the op doesn't have a file, we're not polling for it */
if ((req->ctx->flags & IORING_SETUP_IOPOLL) && req->file)
if ((req->ctx->flags & IORING_SETUP_IOPOLL) && def->iopoll_queue)
io_iopoll_req_issued(req, issue_flags);
return 0;
@@ -1766,9 +1859,8 @@ static int io_issue_sqe(struct io_kiocb *req, unsigned int issue_flags)
int io_poll_issue(struct io_kiocb *req, bool *locked)
{
io_tw_lock(req->ctx, locked);
if (unlikely(req->task->flags & PF_EXITING))
return -EFAULT;
return io_issue_sqe(req, IO_URING_F_NONBLOCK|IO_URING_F_MULTISHOT);
return io_issue_sqe(req, IO_URING_F_NONBLOCK|IO_URING_F_MULTISHOT|
IO_URING_F_COMPLETE_DEFER);
}
struct io_wq_work *io_wq_free_work(struct io_wq_work *work)
@@ -1783,11 +1875,11 @@ void io_wq_submit_work(struct io_wq_work *work)
{
struct io_kiocb *req = container_of(work, struct io_kiocb, work);
const struct io_op_def *def = &io_op_defs[req->opcode];
unsigned int issue_flags = IO_URING_F_UNLOCKED;
unsigned int issue_flags = IO_URING_F_UNLOCKED | IO_URING_F_IOWQ;
bool needs_poll = false;
int ret = 0, err = -ECANCELED;
/* one will be dropped by ->io_free_work() after returning to io-wq */
/* one will be dropped by ->io_wq_free_work() after returning to io-wq */
if (!(req->flags & REQ_F_REFCOUNT))
__io_req_set_refcount(req, 2);
else
@@ -1885,7 +1977,7 @@ static void io_queue_async(struct io_kiocb *req, int ret)
struct io_kiocb *linked_timeout;
if (ret != -EAGAIN || (req->flags & REQ_F_NOWAIT)) {
io_req_complete_failed(req, ret);
io_req_defer_failed(req, ret);
return;
}
@@ -1935,14 +2027,14 @@ static void io_queue_sqe_fallback(struct io_kiocb *req)
*/
req->flags &= ~REQ_F_HARDLINK;
req->flags |= REQ_F_LINK;
io_req_complete_failed(req, req->cqe.res);
io_req_defer_failed(req, req->cqe.res);
} else if (unlikely(req->ctx->drain_active)) {
io_drain_req(req);
} else {
int ret = io_req_prep_async(req);
if (unlikely(ret))
io_req_complete_failed(req, ret);
io_req_defer_failed(req, ret);
else
io_queue_iowq(req, NULL);
}
@@ -2439,11 +2531,7 @@ static int io_cqring_wait(struct io_ring_ctx *ctx, int min_events,
trace_io_uring_cqring_wait(ctx, min_events);
do {
/* if we can't even flush overflow, don't wait for more */
if (!io_cqring_overflow_flush(ctx)) {
ret = -EBUSY;
break;
}
io_cqring_overflow_flush(ctx);
prepare_to_wait_exclusive(&ctx->cq_wait, &iowq.wq,
TASK_INTERRUPTIBLE);
ret = io_cqring_wait_schedule(ctx, &iowq, timeout);
@@ -2594,8 +2682,7 @@ static __cold void io_ring_ctx_free(struct io_ring_ctx *ctx)
__io_sqe_buffers_unregister(ctx);
if (ctx->file_data)
__io_sqe_files_unregister(ctx);
if (ctx->rings)
__io_cqring_overflow_flush(ctx, true);
io_cqring_overflow_kill(ctx);
io_eventfd_unregister(ctx);
io_alloc_cache_free(&ctx->apoll_cache, io_apoll_cache_free);
io_alloc_cache_free(&ctx->netmsg_cache, io_netmsg_cache_free);
@@ -2670,7 +2757,7 @@ static __poll_t io_uring_poll(struct file *file, poll_table *wait)
* lock(&ep->mtx);
*
* Users may get EPOLLIN meanwhile seeing nothing in cqring, this
* pushs them to do the flush.
* pushes them to do the flush.
*/
if (io_cqring_events(ctx) || io_has_work(ctx))
@@ -2738,6 +2825,12 @@ static __cold void io_ring_exit_work(struct work_struct *work)
* as nobody else will be looking for them.
*/
do {
if (test_bit(IO_CHECK_CQ_OVERFLOW_BIT, &ctx->check_cq)) {
mutex_lock(&ctx->uring_lock);
io_cqring_overflow_kill(ctx);
mutex_unlock(&ctx->uring_lock);
}
if (ctx->flags & IORING_SETUP_DEFER_TASKRUN)
io_move_task_work_from_local(ctx);
@@ -2803,8 +2896,6 @@ static __cold void io_ring_ctx_wait_and_kill(struct io_ring_ctx *ctx)
mutex_lock(&ctx->uring_lock);
percpu_ref_kill(&ctx->refs);
if (ctx->rings)
__io_cqring_overflow_flush(ctx, true);
xa_for_each(&ctx->personalities, index, creds)
io_unregister_personality(ctx, index);
if (ctx->rings)
@@ -2871,7 +2962,7 @@ static __cold bool io_cancel_defer_files(struct io_ring_ctx *ctx,
while (!list_empty(&list)) {
de = list_first_entry(&list, struct io_defer_entry, list);
list_del_init(&de->list);
io_req_complete_failed(de->req, -ECANCELED);
io_req_task_queue_fail(de->req, -ECANCELED);
kfree(de);
}
return true;
@@ -3446,6 +3537,11 @@ static __cold int io_uring_create(unsigned entries, struct io_uring_params *p,
if (!ctx)
return -ENOMEM;
if ((ctx->flags & IORING_SETUP_DEFER_TASKRUN) &&
!(ctx->flags & IORING_SETUP_IOPOLL) &&
!(ctx->flags & IORING_SETUP_SQPOLL))
ctx->task_complete = true;
/*
* When SETUP_IOPOLL and SETUP_SQPOLL are both enabled, user
* space applications don't need to do io completion events
@@ -4064,8 +4160,6 @@ SYSCALL_DEFINE4(io_uring_register, unsigned int, fd, unsigned int, opcode,
ctx = f.file->private_data;
io_run_task_work_ctx(ctx);
mutex_lock(&ctx->uring_lock);
ret = __io_uring_register(ctx, opcode, arg, nr_args);
mutex_unlock(&ctx->uring_lock);
io_uring/io_uring.h
+43 -15
View File
@@ -4,6 +4,7 @@
#include <linux/errno.h>
#include <linux/lockdep.h>
#include <linux/io_uring_types.h>
#include <uapi/linux/eventpoll.h>
#include "io-wq.h"
#include "slist.h"
#include "filetable.h"
@@ -29,14 +30,11 @@ bool io_req_cqe_overflow(struct io_kiocb *req);
int io_run_task_work_sig(struct io_ring_ctx *ctx);
int __io_run_local_work(struct io_ring_ctx *ctx, bool *locked);
int io_run_local_work(struct io_ring_ctx *ctx);
void io_req_complete_failed(struct io_kiocb *req, s32 res);
void __io_req_complete(struct io_kiocb *req, unsigned issue_flags);
void io_req_complete_post(struct io_kiocb *req);
void __io_req_complete_post(struct io_kiocb *req);
bool io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags,
bool allow_overflow);
bool io_fill_cqe_aux(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags,
bool allow_overflow);
void io_req_defer_failed(struct io_kiocb *req, s32 res);
void io_req_complete_post(struct io_kiocb *req, unsigned issue_flags);
bool io_post_aux_cqe(struct io_ring_ctx *ctx, u64 user_data, s32 res, u32 cflags);
bool io_aux_cqe(struct io_ring_ctx *ctx, bool defer, u64 user_data, s32 res, u32 cflags,
bool allow_overflow);
void __io_commit_cqring_flush(struct io_ring_ctx *ctx);
struct page **io_pin_pages(unsigned long ubuf, unsigned long len, int *npages);
@@ -50,10 +48,9 @@ static inline bool io_req_ffs_set(struct io_kiocb *req)
return req->flags & REQ_F_FIXED_FILE;
}
void __io_req_task_work_add(struct io_kiocb *req, bool allow_local);
bool io_is_uring_fops(struct file *file);
bool io_alloc_async_data(struct io_kiocb *req);
void io_req_task_work_add(struct io_kiocb *req);
void io_req_tw_post_queue(struct io_kiocb *req, s32 res, u32 cflags);
void io_req_task_queue(struct io_kiocb *req);
void io_queue_iowq(struct io_kiocb *req, bool *dont_use);
void io_req_task_complete(struct io_kiocb *req, bool *locked);
@@ -82,6 +79,11 @@ bool __io_alloc_req_refill(struct io_ring_ctx *ctx);
bool io_match_task_safe(struct io_kiocb *head, struct task_struct *task,
bool cancel_all);
static inline void io_req_task_work_add(struct io_kiocb *req)
{
__io_req_task_work_add(req, true);
}
#define io_for_each_link(pos, head) \
for (pos = (head); pos; pos = pos->link)
@@ -91,6 +93,11 @@ static inline void io_cq_lock(struct io_ring_ctx *ctx)
spin_lock(&ctx->completion_lock);
}
static inline void io_cq_unlock(struct io_ring_ctx *ctx)
{
spin_unlock(&ctx->completion_lock);
}
void io_cq_unlock_post(struct io_ring_ctx *ctx);
static inline struct io_uring_cqe *io_get_cqe_overflow(struct io_ring_ctx *ctx,
@@ -126,7 +133,7 @@ static inline bool __io_fill_cqe_req(struct io_ring_ctx *ctx,
*/
cqe = io_get_cqe(ctx);
if (unlikely(!cqe))
return io_req_cqe_overflow(req);
return false;
trace_io_uring_complete(req->ctx, req, req->cqe.user_data,
req->cqe.res, req->cqe.flags,
@@ -149,6 +156,14 @@ static inline bool __io_fill_cqe_req(struct io_ring_ctx *ctx,
return true;
}
static inline bool io_fill_cqe_req(struct io_ring_ctx *ctx,
struct io_kiocb *req)
{
if (likely(__io_fill_cqe_req(ctx, req)))
return true;
return io_req_cqe_overflow(req);
}
static inline void req_set_fail(struct io_kiocb *req)
{
req->flags |= REQ_F_FAIL;
@@ -207,12 +222,18 @@ static inline void io_commit_cqring(struct io_ring_ctx *ctx)
static inline void __io_cqring_wake(struct io_ring_ctx *ctx)
{
/*
* wake_up_all() may seem excessive, but io_wake_function() and
* io_should_wake() handle the termination of the loop and only
* wake as many waiters as we need to.
* Trigger waitqueue handler on all waiters on our waitqueue. This
* won't necessarily wake up all the tasks, io_should_wake() will make
* that decision.
*
* Pass in EPOLLIN|EPOLL_URING_WAKE as the poll wakeup key. The latter
* set in the mask so that if we recurse back into our own poll
* waitqueue handlers, we know we have a dependency between eventfd or
* epoll and should terminate multishot poll at that point.
*/
if (waitqueue_active(&ctx->cq_wait))
wake_up_all(&ctx->cq_wait);
__wake_up(&ctx->cq_wait, TASK_NORMAL, 0,
poll_to_key(EPOLL_URING_WAKE | EPOLLIN));
}
static inline void io_cqring_wake(struct io_ring_ctx *ctx)
@@ -369,4 +390,11 @@ static inline bool io_allowed_run_tw(struct io_ring_ctx *ctx)
ctx->submitter_task == current);
}
static inline void io_req_queue_tw_complete(struct io_kiocb *req, s32 res)
{
io_req_set_res(req, res, 0);
req->io_task_work.func = io_req_task_complete;
io_req_task_work_add(req);
}
#endif
io_uring/kbuf.c
+5 -9
View File
@@ -306,14 +306,11 @@ int io_remove_buffers(struct io_kiocb *req, unsigned int issue_flags)
if (!bl->buf_nr_pages)
ret = __io_remove_buffers(ctx, bl, p->nbufs);
}
io_ring_submit_unlock(ctx, issue_flags);
if (ret < 0)
req_set_fail(req);
/* complete before unlock, IOPOLL may need the lock */
io_req_set_res(req, ret, 0);
__io_req_complete(req, issue_flags);
io_ring_submit_unlock(ctx, issue_flags);
return IOU_ISSUE_SKIP_COMPLETE;
return IOU_OK;
}
int io_provide_buffers_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
@@ -458,13 +455,12 @@ int io_provide_buffers(struct io_kiocb *req, unsigned int issue_flags)
ret = io_add_buffers(ctx, p, bl);
err:
io_ring_submit_unlock(ctx, issue_flags);
if (ret < 0)
req_set_fail(req);
/* complete before unlock, IOPOLL may need the lock */
io_req_set_res(req, ret, 0);
__io_req_complete(req, issue_flags);
io_ring_submit_unlock(ctx, issue_flags);
return IOU_ISSUE_SKIP_COMPLETE;
return IOU_OK;
}
int io_register_pbuf_ring(struct io_ring_ctx *ctx, void __user *arg)
io_uring/msg_ring.c
+129 -59
View File
@@ -15,6 +15,8 @@
struct io_msg {
struct file *file;
struct file *src_file;
struct callback_head tw;
u64 user_data;
u32 len;
u32 cmd;
@@ -23,6 +25,34 @@ struct io_msg {
u32 flags;
};
void io_msg_ring_cleanup(struct io_kiocb *req)
{
struct io_msg *msg = io_kiocb_to_cmd(req, struct io_msg);
if (WARN_ON_ONCE(!msg->src_file))
return;
fput(msg->src_file);
msg->src_file = NULL;
}
static void io_msg_tw_complete(struct callback_head *head)
{
struct io_msg *msg = container_of(head, struct io_msg, tw);
struct io_kiocb *req = cmd_to_io_kiocb(msg);
struct io_ring_ctx *target_ctx = req->file->private_data;
int ret = 0;
if (current->flags & PF_EXITING)
ret = -EOWNERDEAD;
else if (!io_post_aux_cqe(target_ctx, msg->user_data, msg->len, 0))
ret = -EOVERFLOW;
if (ret < 0)
req_set_fail(req);
io_req_queue_tw_complete(req, ret);
}
static int io_msg_ring_data(struct io_kiocb *req)
{
struct io_ring_ctx *target_ctx = req->file->private_data;
@@ -31,23 +61,29 @@ static int io_msg_ring_data(struct io_kiocb *req)
if (msg->src_fd || msg->dst_fd || msg->flags)
return -EINVAL;
if (io_post_aux_cqe(target_ctx, msg->user_data, msg->len, 0, true))
if (target_ctx->task_complete && current != target_ctx->submitter_task) {
init_task_work(&msg->tw, io_msg_tw_complete);
if (task_work_add(target_ctx->submitter_task, &msg->tw,
TWA_SIGNAL_NO_IPI))
return -EOWNERDEAD;
atomic_or(IORING_SQ_TASKRUN, &target_ctx->rings->sq_flags);
return IOU_ISSUE_SKIP_COMPLETE;
}
if (io_post_aux_cqe(target_ctx, msg->user_data, msg->len, 0))
return 0;
return -EOVERFLOW;
}
static void io_double_unlock_ctx(struct io_ring_ctx *ctx,
struct io_ring_ctx *octx,
static void io_double_unlock_ctx(struct io_ring_ctx *octx,
unsigned int issue_flags)
{
if (issue_flags & IO_URING_F_UNLOCKED)
mutex_unlock(&ctx->uring_lock);
mutex_unlock(&octx->uring_lock);
}
static int io_double_lock_ctx(struct io_ring_ctx *ctx,
struct io_ring_ctx *octx,
static int io_double_lock_ctx(struct io_ring_ctx *octx,
unsigned int issue_flags)
{
/*
@@ -60,67 +96,101 @@ static int io_double_lock_ctx(struct io_ring_ctx *ctx,
return -EAGAIN;
return 0;
}
/* Always grab smallest value ctx first. We know ctx != octx. */
if (ctx < octx) {
mutex_lock(&ctx->uring_lock);
mutex_lock(&octx->uring_lock);
} else {
mutex_lock(&octx->uring_lock);
mutex_lock(&ctx->uring_lock);
}
mutex_lock(&octx->uring_lock);
return 0;
}
static struct file *io_msg_grab_file(struct io_kiocb *req, unsigned int issue_flags)
{
struct io_msg *msg = io_kiocb_to_cmd(req, struct io_msg);
struct io_ring_ctx *ctx = req->ctx;
struct file *file = NULL;
unsigned long file_ptr;
int idx = msg->src_fd;
io_ring_submit_lock(ctx, issue_flags);
if (likely(idx < ctx->nr_user_files)) {
idx = array_index_nospec(idx, ctx->nr_user_files);
file_ptr = io_fixed_file_slot(&ctx->file_table, idx)->file_ptr;
file = (struct file *) (file_ptr & FFS_MASK);
if (file)
get_file(file);
}
io_ring_submit_unlock(ctx, issue_flags);
return file;
}
static int io_msg_install_complete(struct io_kiocb *req, unsigned int issue_flags)
{
struct io_ring_ctx *target_ctx = req->file->private_data;
struct io_msg *msg = io_kiocb_to_cmd(req, struct io_msg);
struct file *src_file = msg->src_file;
int ret;
if (unlikely(io_double_lock_ctx(target_ctx, issue_flags)))
return -EAGAIN;
ret = __io_fixed_fd_install(target_ctx, src_file, msg->dst_fd);
if (ret < 0)
goto out_unlock;
msg->src_file = NULL;
req->flags &= ~REQ_F_NEED_CLEANUP;
if (msg->flags & IORING_MSG_RING_CQE_SKIP)
goto out_unlock;
/*
* If this fails, the target still received the file descriptor but
* wasn't notified of the fact. This means that if this request
* completes with -EOVERFLOW, then the sender must ensure that a
* later IORING_OP_MSG_RING delivers the message.
*/
if (!io_post_aux_cqe(target_ctx, msg->user_data, msg->len, 0))
ret = -EOVERFLOW;
out_unlock:
io_double_unlock_ctx(target_ctx, issue_flags);
return ret;
}
static void io_msg_tw_fd_complete(struct callback_head *head)
{
struct io_msg *msg = container_of(head, struct io_msg, tw);
struct io_kiocb *req = cmd_to_io_kiocb(msg);
int ret = -EOWNERDEAD;
if (!(current->flags & PF_EXITING))
ret = io_msg_install_complete(req, IO_URING_F_UNLOCKED);
if (ret < 0)
req_set_fail(req);
io_req_queue_tw_complete(req, ret);
}
static int io_msg_send_fd(struct io_kiocb *req, unsigned int issue_flags)
{
struct io_ring_ctx *target_ctx = req->file->private_data;
struct io_msg *msg = io_kiocb_to_cmd(req, struct io_msg);
struct io_ring_ctx *ctx = req->ctx;
unsigned long file_ptr;
struct file *src_file;
int ret;
struct file *src_file = msg->src_file;
if (target_ctx == ctx)
return -EINVAL;
ret = io_double_lock_ctx(ctx, target_ctx, issue_flags);
if (unlikely(ret))
return ret;
ret = -EBADF;
if (unlikely(msg->src_fd >= ctx->nr_user_files))
goto out_unlock;
msg->src_fd = array_index_nospec(msg->src_fd, ctx->nr_user_files);
file_ptr = io_fixed_file_slot(&ctx->file_table, msg->src_fd)->file_ptr;
if (!file_ptr)
goto out_unlock;
src_file = (struct file *) (file_ptr & FFS_MASK);
get_file(src_file);
ret = __io_fixed_fd_install(target_ctx, src_file, msg->dst_fd);
if (ret < 0) {
fput(src_file);
goto out_unlock;
if (!src_file) {
src_file = io_msg_grab_file(req, issue_flags);
if (!src_file)
return -EBADF;
msg->src_file = src_file;
req->flags |= REQ_F_NEED_CLEANUP;
}
if (msg->flags & IORING_MSG_RING_CQE_SKIP)
goto out_unlock;
if (target_ctx->task_complete && current != target_ctx->submitter_task) {
init_task_work(&msg->tw, io_msg_tw_fd_complete);
if (task_work_add(target_ctx->submitter_task, &msg->tw,
TWA_SIGNAL))
return -EOWNERDEAD;
/*
* If this fails, the target still received the file descriptor but
* wasn't notified of the fact. This means that if this request
* completes with -EOVERFLOW, then the sender must ensure that a
* later IORING_OP_MSG_RING delivers the message.
*/
if (!io_post_aux_cqe(target_ctx, msg->user_data, msg->len, 0, true))
ret = -EOVERFLOW;
out_unlock:
io_double_unlock_ctx(ctx, target_ctx, issue_flags);
return ret;
return IOU_ISSUE_SKIP_COMPLETE;
}
return io_msg_install_complete(req, issue_flags);
}
int io_msg_ring_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
@@ -130,6 +200,7 @@ int io_msg_ring_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
if (unlikely(sqe->buf_index || sqe->personality))
return -EINVAL;
msg->src_file = NULL;
msg->user_data = READ_ONCE(sqe->off);
msg->len = READ_ONCE(sqe->len);
msg->cmd = READ_ONCE(sqe->addr);
@@ -164,12 +235,11 @@ int io_msg_ring(struct io_kiocb *req, unsigned int issue_flags)
}
done:
if (ret < 0)
if (ret < 0) {
if (ret == -EAGAIN || ret == IOU_ISSUE_SKIP_COMPLETE)
return ret;
req_set_fail(req);
}
io_req_set_res(req, ret, 0);
/* put file to avoid an attempt to IOPOLL the req */
if (!(req->flags & REQ_F_FIXED_FILE))
io_put_file(req->file);
req->file = NULL;
return IOU_OK;
}
io_uring/msg_ring.h
+1
View File
@@ -2,3 +2,4 @@
int io_msg_ring_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe);
int io_msg_ring(struct io_kiocb *req, unsigned int issue_flags);
void io_msg_ring_cleanup(struct io_kiocb *req);
io_uring/net.c
+55 -22
View File
@@ -67,6 +67,19 @@ struct io_sr_msg {
struct io_kiocb *notif;
};
static inline bool io_check_multishot(struct io_kiocb *req,
unsigned int issue_flags)
{
/*
* When ->locked_cq is set we only allow to post CQEs from the original
* task context. Usual request completions will be handled in other
* generic paths but multipoll may decide to post extra cqes.
*/
return !(issue_flags & IO_URING_F_IOWQ) ||
!(issue_flags & IO_URING_F_MULTISHOT) ||
!req->ctx->task_complete;
}
int io_shutdown_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
{
struct io_shutdown *shutdown = io_kiocb_to_cmd(req, struct io_shutdown);
@@ -125,13 +138,15 @@ static struct io_async_msghdr *io_msg_alloc_async(struct io_kiocb *req,
struct io_cache_entry *entry;
struct io_async_msghdr *hdr;
if (!(issue_flags & IO_URING_F_UNLOCKED) &&
(entry = io_alloc_cache_get(&ctx->netmsg_cache)) != NULL) {
hdr = container_of(entry, struct io_async_msghdr, cache);
hdr->free_iov = NULL;
req->flags |= REQ_F_ASYNC_DATA;
req->async_data = hdr;
return hdr;
if (!(issue_flags & IO_URING_F_UNLOCKED)) {
entry = io_alloc_cache_get(&ctx->netmsg_cache);
if (entry) {
hdr = container_of(entry, struct io_async_msghdr, cache);
hdr->free_iov = NULL;
req->flags |= REQ_F_ASYNC_DATA;
req->async_data = hdr;
return hdr;
}
}
if (!io_alloc_async_data(req)) {
@@ -599,16 +614,12 @@ static inline bool io_recv_finish(struct io_kiocb *req, int *ret,
}
if (!mshot_finished) {
if (io_post_aux_cqe(req->ctx, req->cqe.user_data, *ret,
cflags | IORING_CQE_F_MORE, false)) {
if (io_aux_cqe(req->ctx, issue_flags & IO_URING_F_COMPLETE_DEFER,
req->cqe.user_data, *ret, cflags | IORING_CQE_F_MORE, true)) {
io_recv_prep_retry(req);
return false;
}
/*
* Otherwise stop multishot but use the current result.
* Probably will end up going into overflow, but this means
* we cannot trust the ordering anymore
*/
/* Otherwise stop multishot but use the current result. */
}
io_req_set_res(req, *ret, cflags);
@@ -732,6 +743,9 @@ int io_recvmsg(struct io_kiocb *req, unsigned int issue_flags)
(sr->flags & IORING_RECVSEND_POLL_FIRST))
return io_setup_async_msg(req, kmsg, issue_flags);
if (!io_check_multishot(req, issue_flags))
return io_setup_async_msg(req, kmsg, issue_flags);
retry_multishot:
if (io_do_buffer_select(req)) {
void __user *buf;
@@ -831,6 +845,9 @@ int io_recv(struct io_kiocb *req, unsigned int issue_flags)
(sr->flags & IORING_RECVSEND_POLL_FIRST))
return -EAGAIN;
if (!io_check_multishot(req, issue_flags))
return -EAGAIN;
sock = sock_from_file(req->file);
if (unlikely(!sock))
return -ENOTSOCK;
@@ -923,6 +940,9 @@ void io_send_zc_cleanup(struct io_kiocb *req)
}
}
#define IO_ZC_FLAGS_COMMON (IORING_RECVSEND_POLL_FIRST | IORING_RECVSEND_FIXED_BUF)
#define IO_ZC_FLAGS_VALID (IO_ZC_FLAGS_COMMON | IORING_SEND_ZC_REPORT_USAGE)
int io_send_zc_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
{
struct io_sr_msg *zc = io_kiocb_to_cmd(req, struct io_sr_msg);
@@ -935,10 +955,6 @@ int io_send_zc_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
if (req->flags & REQ_F_CQE_SKIP)
return -EINVAL;
zc->flags = READ_ONCE(sqe->ioprio);
if (zc->flags & ~(IORING_RECVSEND_POLL_FIRST |
IORING_RECVSEND_FIXED_BUF))
return -EINVAL;
notif = zc->notif = io_alloc_notif(ctx);
if (!notif)
return -ENOMEM;
@@ -946,6 +962,17 @@ int io_send_zc_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
notif->cqe.res = 0;
notif->cqe.flags = IORING_CQE_F_NOTIF;
req->flags |= REQ_F_NEED_CLEANUP;
zc->flags = READ_ONCE(sqe->ioprio);
if (unlikely(zc->flags & ~IO_ZC_FLAGS_COMMON)) {
if (zc->flags & ~IO_ZC_FLAGS_VALID)
return -EINVAL;
if (zc->flags & IORING_SEND_ZC_REPORT_USAGE) {
io_notif_set_extended(notif);
io_notif_to_data(notif)->zc_report = true;
}
}
if (zc->flags & IORING_RECVSEND_FIXED_BUF) {
unsigned idx = READ_ONCE(sqe->buf_index);
@@ -1087,6 +1114,7 @@ int io_send_zc(struct io_kiocb *req, unsigned int issue_flags)
return ret;
msg.sg_from_iter = io_sg_from_iter;
} else {
io_notif_set_extended(zc->notif);
ret = import_single_range(ITER_SOURCE, zc->buf, zc->len, &iov,
&msg.msg_iter);
if (unlikely(ret))
@@ -1148,6 +1176,8 @@ int io_sendmsg_zc(struct io_kiocb *req, unsigned int issue_flags)
unsigned flags;
int ret, min_ret = 0;
io_notif_set_extended(sr->notif);
sock = sock_from_file(req->file);
if (unlikely(!sock))
return -ENOTSOCK;
@@ -1269,6 +1299,8 @@ int io_accept(struct io_kiocb *req, unsigned int issue_flags)
struct file *file;
int ret, fd;
if (!io_check_multishot(req, issue_flags))
return -EAGAIN;
retry:
if (!fixed) {
fd = __get_unused_fd_flags(accept->flags, accept->nofile);
@@ -1307,12 +1339,13 @@ retry:
return IOU_OK;
}
if (ret >= 0 &&
io_post_aux_cqe(ctx, req->cqe.user_data, ret, IORING_CQE_F_MORE, false))
if (ret < 0)
return ret;
if (io_aux_cqe(ctx, issue_flags & IO_URING_F_COMPLETE_DEFER,
req->cqe.user_data, ret, IORING_CQE_F_MORE, true))
goto retry;
io_req_set_res(req, ret, 0);
return (issue_flags & IO_URING_F_MULTISHOT) ? IOU_STOP_MULTISHOT : IOU_OK;
return -ECANCELED;
}
int io_socket_prep(struct io_kiocb *req, const struct io_uring_sqe *sqe)
io_uring/notif.c
+36 -21
View File
@@ -9,11 +9,14 @@
#include "notif.h"
#include "rsrc.h"
static void __io_notif_complete_tw(struct io_kiocb *notif, bool *locked)
static void io_notif_complete_tw_ext(struct io_kiocb *notif, bool *locked)
{
struct io_notif_data *nd = io_notif_to_data(notif);
struct io_ring_ctx *ctx = notif->ctx;
if (nd->zc_report && (nd->zc_copied || !nd->zc_used))
notif->cqe.res |= IORING_NOTIF_USAGE_ZC_COPIED;
if (nd->account_pages && ctx->user) {
__io_unaccount_mem(ctx->user, nd->account_pages);
nd->account_pages = 0;
@@ -21,16 +24,41 @@ static void __io_notif_complete_tw(struct io_kiocb *notif, bool *locked)
io_req_task_complete(notif, locked);
}
static void io_uring_tx_zerocopy_callback(struct sk_buff *skb,
struct ubuf_info *uarg,
bool success)
static void io_tx_ubuf_callback(struct sk_buff *skb, struct ubuf_info *uarg,
bool success)
{
struct io_notif_data *nd = container_of(uarg, struct io_notif_data, uarg);
struct io_kiocb *notif = cmd_to_io_kiocb(nd);
if (refcount_dec_and_test(&uarg->refcnt)) {
notif->io_task_work.func = __io_notif_complete_tw;
if (refcount_dec_and_test(&uarg->refcnt))
io_req_task_work_add(notif);
}
static void io_tx_ubuf_callback_ext(struct sk_buff *skb, struct ubuf_info *uarg,
bool success)
{
struct io_notif_data *nd = container_of(uarg, struct io_notif_data, uarg);
if (nd->zc_report) {
if (success && !nd->zc_used && skb)
WRITE_ONCE(nd->zc_used, true);
else if (!success && !nd->zc_copied)
WRITE_ONCE(nd->zc_copied, true);
}
io_tx_ubuf_callback(skb, uarg, success);
}
void io_notif_set_extended(struct io_kiocb *notif)
{
struct io_notif_data *nd = io_notif_to_data(notif);
if (nd->uarg.callback != io_tx_ubuf_callback_ext) {
nd->account_pages = 0;
nd->zc_report = false;
nd->zc_used = false;
nd->zc_copied = false;
nd->uarg.callback = io_tx_ubuf_callback_ext;
notif->io_task_work.func = io_notif_complete_tw_ext;
}
}
@@ -49,24 +77,11 @@ struct io_kiocb *io_alloc_notif(struct io_ring_ctx *ctx)
notif->task = current;
io_get_task_refs(1);
notif->rsrc_node = NULL;
io_req_set_rsrc_node(notif, ctx, 0);
notif->io_task_work.func = io_req_task_complete;
nd = io_notif_to_data(notif);
nd->account_pages = 0;
nd->uarg.flags = SKBFL_ZEROCOPY_FRAG | SKBFL_DONT_ORPHAN;
nd->uarg.callback = io_uring_tx_zerocopy_callback;
nd->uarg.callback = io_tx_ubuf_callback;
refcount_set(&nd->uarg.refcnt, 1);
return notif;
}
void io_notif_flush(struct io_kiocb *notif)
__must_hold(&slot->notif->ctx->uring_lock)
{
struct io_notif_data *nd = io_notif_to_data(notif);
/* drop slot's master ref */
if (refcount_dec_and_test(&nd->uarg.refcnt)) {
notif->io_task_work.func = __io_notif_complete_tw;
io_req_task_work_add(notif);
}
}
io_uring/notif.h
+14 -1
View File
@@ -13,16 +13,29 @@ struct io_notif_data {
struct file *file;
struct ubuf_info uarg;
unsigned long account_pages;
bool zc_report;
bool zc_used;
bool zc_copied;
};
void io_notif_flush(struct io_kiocb *notif);
struct io_kiocb *io_alloc_notif(struct io_ring_ctx *ctx);
void io_notif_set_extended(struct io_kiocb *notif);
static inline struct io_notif_data *io_notif_to_data(struct io_kiocb *notif)
{
return io_kiocb_to_cmd(notif, struct io_notif_data);
}
static inline void io_notif_flush(struct io_kiocb *notif)
__must_hold(&notif->ctx->uring_lock)
{
struct io_notif_data *nd = io_notif_to_data(notif);
/* drop slot's master ref */
if (refcount_dec_and_test(&nd->uarg.refcnt))
io_req_task_work_add(notif);
}
static inline int io_notif_account_mem(struct io_kiocb *notif, unsigned len)
{
struct io_ring_ctx *ctx = notif->ctx;
io_uring/opdef.c
+8
View File
@@ -63,6 +63,7 @@ const struct io_op_def io_op_defs[] = {
.audit_skip = 1,
.ioprio = 1,
.iopoll = 1,
.iopoll_queue = 1,
.async_size = sizeof(struct io_async_rw),
.name = "READV",
.prep = io_prep_rw,
@@ -80,6 +81,7 @@ const struct io_op_def io_op_defs[] = {
.audit_skip = 1,
.ioprio = 1,
.iopoll = 1,
.iopoll_queue = 1,
.async_size = sizeof(struct io_async_rw),
.name = "WRITEV",
.prep = io_prep_rw,
@@ -103,6 +105,7 @@ const struct io_op_def io_op_defs[] = {
.audit_skip = 1,
.ioprio = 1,
.iopoll = 1,
.iopoll_queue = 1,
.async_size = sizeof(struct io_async_rw),
.name = "READ_FIXED",
.prep = io_prep_rw,
@@ -118,6 +121,7 @@ const struct io_op_def io_op_defs[] = {
.audit_skip = 1,
.ioprio = 1,
.iopoll = 1,
.iopoll_queue = 1,
.async_size = sizeof(struct io_async_rw),
.name = "WRITE_FIXED",
.prep = io_prep_rw,
@@ -277,6 +281,7 @@ const struct io_op_def io_op_defs[] = {
.audit_skip = 1,
.ioprio = 1,
.iopoll = 1,
.iopoll_queue = 1,
.async_size = sizeof(struct io_async_rw),
.name = "READ",
.prep = io_prep_rw,
@@ -292,6 +297,7 @@ const struct io_op_def io_op_defs[] = {
.audit_skip = 1,
.ioprio = 1,
.iopoll = 1,
.iopoll_queue = 1,
.async_size = sizeof(struct io_async_rw),
.name = "WRITE",
.prep = io_prep_rw,
@@ -439,6 +445,7 @@ const struct io_op_def io_op_defs[] = {
.name = "MSG_RING",
.prep = io_msg_ring_prep,
.issue = io_msg_ring,
.cleanup = io_msg_ring_cleanup,
},
[IORING_OP_FSETXATTR] = {
.needs_file = 1,
@@ -481,6 +488,7 @@ const struct io_op_def io_op_defs[] = {
.plug = 1,
.name = "URING_CMD",
.iopoll = 1,
.iopoll_queue = 1,
.async_size = uring_cmd_pdu_size(1),
.prep = io_uring_cmd_prep,
.issue = io_uring_cmd,
io_uring/opdef.h
+2
View File
@@ -25,6 +25,8 @@ struct io_op_def {
unsigned ioprio : 1;
/* supports iopoll */
unsigned iopoll : 1;
/* have to be put into the iopoll list */
unsigned iopoll_queue : 1;
/* opcode specific path will handle ->async_data allocation if needed */
unsigned manual_alloc : 1;
/* size of async data needed, if any */
io_uring/poll.c
+62 -67
View File
@@ -237,7 +237,6 @@ enum {
*/
static int io_poll_check_events(struct io_kiocb *req, bool *locked)
{
struct io_ring_ctx *ctx = req->ctx;
int v, ret;
/* req->task == current here, checking PF_EXITING is safe */
@@ -247,27 +246,30 @@ static int io_poll_check_events(struct io_kiocb *req, bool *locked)
do {
v = atomic_read(&req->poll_refs);
/* tw handler should be the owner, and so have some references */
if (WARN_ON_ONCE(!(v & IO_POLL_REF_MASK)))
return IOU_POLL_DONE;
if (v & IO_POLL_CANCEL_FLAG)
return -ECANCELED;
/*
* cqe.res contains only events of the first wake up
* and all others are be lost. Redo vfs_poll() to get
* up to date state.
*/
if ((v & IO_POLL_REF_MASK) != 1)
req->cqe.res = 0;
if (v & IO_POLL_RETRY_FLAG) {
req->cqe.res = 0;
if (unlikely(v != 1)) {
/* tw should be the owner and so have some refs */
if (WARN_ON_ONCE(!(v & IO_POLL_REF_MASK)))
return IOU_POLL_NO_ACTION;
if (v & IO_POLL_CANCEL_FLAG)
return -ECANCELED;
/*
* We won't find new events that came in between
* vfs_poll and the ref put unless we clear the flag
* in advance.
* cqe.res contains only events of the first wake up
* and all others are to be lost. Redo vfs_poll() to get
* up to date state.
*/
atomic_andnot(IO_POLL_RETRY_FLAG, &req->poll_refs);
v &= ~IO_POLL_RETRY_FLAG;
if ((v & IO_POLL_REF_MASK) != 1)
req->cqe.res = 0;
if (v & IO_POLL_RETRY_FLAG) {
req->cqe.res = 0;
/*
* We won't find new events that came in between
* vfs_poll and the ref put unless we clear the
* flag in advance.
*/
atomic_andnot(IO_POLL_RETRY_FLAG, &req->poll_refs);
v &= ~IO_POLL_RETRY_FLAG;
}
}
/* the mask was stashed in __io_poll_execute */
@@ -280,16 +282,14 @@ static int io_poll_check_events(struct io_kiocb *req, bool *locked)
continue;
if (req->apoll_events & EPOLLONESHOT)
return IOU_POLL_DONE;
if (io_is_uring_fops(req->file))
return IOU_POLL_DONE;
/* multishot, just fill a CQE and proceed */
if (!(req->flags & REQ_F_APOLL_MULTISHOT)) {
__poll_t mask = mangle_poll(req->cqe.res &
req->apoll_events);
if (!io_post_aux_cqe(ctx, req->cqe.user_data,
mask, IORING_CQE_F_MORE, false)) {
if (!io_aux_cqe(req->ctx, *locked, req->cqe.user_data,
mask, IORING_CQE_F_MORE, false)) {
io_req_set_res(req, mask, 0);
return IOU_POLL_REMOVE_POLL_USE_RES;
}
@@ -321,54 +321,38 @@ static void io_poll_task_func(struct io_kiocb *req, bool *locked)
ret = io_poll_check_events(req, locked);
if (ret == IOU_POLL_NO_ACTION)
return;
io_poll_remove_entries(req);
io_poll_tw_hash_eject(req, locked);
if (ret == IOU_POLL_DONE) {
struct io_poll *poll = io_kiocb_to_cmd(req, struct io_poll);
req->cqe.res = mangle_poll(req->cqe.res & poll->events);
} else if (ret != IOU_POLL_REMOVE_POLL_USE_RES) {
req->cqe.res = ret;
req_set_fail(req);
if (req->opcode == IORING_OP_POLL_ADD) {
if (ret == IOU_POLL_DONE) {
struct io_poll *poll;
poll = io_kiocb_to_cmd(req, struct io_poll);
req->cqe.res = mangle_poll(req->cqe.res & poll->events);
} else if (ret != IOU_POLL_REMOVE_POLL_USE_RES) {
req->cqe.res = ret;
req_set_fail(req);
}
io_req_set_res(req, req->cqe.res, 0);
io_req_task_complete(req, locked);
} else {
io_tw_lock(req->ctx, locked);
if (ret == IOU_POLL_REMOVE_POLL_USE_RES)
io_req_task_complete(req, locked);
else if (ret == IOU_POLL_DONE)
io_req_task_submit(req, locked);
else
io_req_defer_failed(req, ret);
}
io_poll_remove_entries(req);
io_poll_tw_hash_eject(req, locked);
io_req_set_res(req, req->cqe.res, 0);
io_req_task_complete(req, locked);
}
static void io_apoll_task_func(struct io_kiocb *req, bool *locked)
{
int ret;
ret = io_poll_check_events(req, locked);
if (ret == IOU_POLL_NO_ACTION)
return;
io_poll_remove_entries(req);
io_poll_tw_hash_eject(req, locked);
if (ret == IOU_POLL_REMOVE_POLL_USE_RES)
io_req_complete_post(req);
else if (ret == IOU_POLL_DONE)
io_req_task_submit(req, locked);
else
io_req_complete_failed(req, ret);
}
static void __io_poll_execute(struct io_kiocb *req, int mask)
{
io_req_set_res(req, mask, 0);
/*
* This is useful for poll that is armed on behalf of another
* request, and where the wakeup path could be on a different
* CPU. We want to avoid pulling in req->apoll->events for that
* case.
*/
if (req->opcode == IORING_OP_POLL_ADD)
req->io_task_work.func = io_poll_task_func;
else
req->io_task_work.func = io_apoll_task_func;
req->io_task_work.func = io_poll_task_func;
trace_io_uring_task_add(req, mask);
io_req_task_work_add(req);
@@ -429,6 +413,14 @@ static int io_poll_wake(struct wait_queue_entry *wait, unsigned mode, int sync,
return 0;
if (io_poll_get_ownership(req)) {
/*
* If we trigger a multishot poll off our own wakeup path,
* disable multishot as there is a circular dependency between
* CQ posting and triggering the event.
*/
if (mask & EPOLL_URING_WAKE)
poll->events |= EPOLLONESHOT;
/* optional, saves extra locking for removal in tw handler */
if (mask && poll->events & EPOLLONESHOT) {
list_del_init(&poll->wait.entry);
@@ -648,10 +640,13 @@ static struct async_poll *io_req_alloc_apoll(struct io_kiocb *req,
if (req->flags & REQ_F_POLLED) {
apoll = req->apoll;
kfree(apoll->double_poll);
} else if (!(issue_flags & IO_URING_F_UNLOCKED) &&
(entry = io_alloc_cache_get(&ctx->apoll_cache)) != NULL) {
} else if (!(issue_flags & IO_URING_F_UNLOCKED)) {
entry = io_alloc_cache_get(&ctx->apoll_cache);
if (entry == NULL)
goto alloc_apoll;
apoll = container_of(entry, struct async_poll, cache);
} else {
alloc_apoll:
apoll = kmalloc(sizeof(*apoll), GFP_ATOMIC);
if (unlikely(!apoll))
return NULL;
io_uring/rsrc.c
+43 -28
View File
@@ -170,10 +170,10 @@ static void __io_rsrc_put_work(struct io_rsrc_node *ref_node)
if (prsrc->tag) {
if (ctx->flags & IORING_SETUP_IOPOLL) {
mutex_lock(&ctx->uring_lock);
io_post_aux_cqe(ctx, prsrc->tag, 0, 0, true);
io_post_aux_cqe(ctx, prsrc->tag, 0, 0);
mutex_unlock(&ctx->uring_lock);
} else {
io_post_aux_cqe(ctx, prsrc->tag, 0, 0, true);
io_post_aux_cqe(ctx, prsrc->tag, 0, 0);
}
}
@@ -204,6 +204,14 @@ void io_rsrc_put_work(struct work_struct *work)
}
}
void io_rsrc_put_tw(struct callback_head *cb)
{
struct io_ring_ctx *ctx = container_of(cb, struct io_ring_ctx,
rsrc_put_tw);
io_rsrc_put_work(&ctx->rsrc_put_work.work);
}
void io_wait_rsrc_data(struct io_rsrc_data *data)
{
if (data && !atomic_dec_and_test(&data->refs))
@@ -242,8 +250,15 @@ static __cold void io_rsrc_node_ref_zero(struct percpu_ref *ref)
}
spin_unlock_irqrestore(&ctx->rsrc_ref_lock, flags);
if (first_add)
mod_delayed_work(system_wq, &ctx->rsrc_put_work, delay);
if (!first_add)
return;
if (ctx->submitter_task) {
if (!task_work_add(ctx->submitter_task, &ctx->rsrc_put_tw,
ctx->notify_method))
return;
}
mod_delayed_work(system_wq, &ctx->rsrc_put_work, delay);
}
static struct io_rsrc_node *io_rsrc_node_alloc(void)
@@ -309,41 +324,41 @@ __cold static int io_rsrc_ref_quiesce(struct io_rsrc_data *data,
/* As we may drop ->uring_lock, other task may have started quiesce */
if (data->quiesce)
return -ENXIO;
ret = io_rsrc_node_switch_start(ctx);
if (ret)
return ret;
io_rsrc_node_switch(ctx, data);
/* kill initial ref, already quiesced if zero */
if (atomic_dec_and_test(&data->refs))
return 0;
data->quiesce = true;
mutex_unlock(&ctx->uring_lock);
do {
ret = io_rsrc_node_switch_start(ctx);
if (ret)
ret = io_run_task_work_sig(ctx);
if (ret < 0) {
atomic_inc(&data->refs);
/* wait for all works potentially completing data->done */
flush_delayed_work(&ctx->rsrc_put_work);
reinit_completion(&data->done);
mutex_lock(&ctx->uring_lock);
break;
io_rsrc_node_switch(ctx, data);
}
/* kill initial ref, already quiesced if zero */
if (atomic_dec_and_test(&data->refs))
break;
mutex_unlock(&ctx->uring_lock);
flush_delayed_work(&ctx->rsrc_put_work);
ret = wait_for_completion_interruptible(&data->done);
if (!ret) {
mutex_lock(&ctx->uring_lock);
if (atomic_read(&data->refs) > 0) {
/*
* it has been revived by another thread while
* we were unlocked
*/
mutex_unlock(&ctx->uring_lock);
} else {
if (atomic_read(&data->refs) <= 0)
break;
}
/*
* it has been revived by another thread while
* we were unlocked
*/
mutex_unlock(&ctx->uring_lock);
}
atomic_inc(&data->refs);
/* wait for all works potentially completing data->done */
flush_delayed_work(&ctx->rsrc_put_work);
reinit_completion(&data->done);
ret = io_run_task_work_sig(ctx);
mutex_lock(&ctx->uring_lock);
} while (ret >= 0);
} while (1);
data->quiesce = false;
return ret;
io_uring/rsrc.h
+1
View File
@@ -53,6 +53,7 @@ struct io_mapped_ubuf {
struct bio_vec bvec[];
};
void io_rsrc_put_tw(struct callback_head *cb);
void io_rsrc_put_work(struct work_struct *work);
void io_rsrc_refs_refill(struct io_ring_ctx *ctx);
void io_wait_rsrc_data(struct io_rsrc_data *data);
io_uring/rw.c
+6
View File
@@ -286,6 +286,12 @@ static inline int io_fixup_rw_res(struct io_kiocb *req, long res)
static void io_req_rw_complete(struct io_kiocb *req, bool *locked)
{
io_req_io_end(req);
if (req->flags & (REQ_F_BUFFER_SELECTED|REQ_F_BUFFER_RING)) {
unsigned issue_flags = *locked ? 0 : IO_URING_F_UNLOCKED;
req->cqe.flags |= io_put_kbuf(req, issue_flags);
}
io_req_task_complete(req, locked);
}
io_uring/timeout.c
+5 -5
View File
@@ -63,7 +63,7 @@ static bool io_kill_timeout(struct io_kiocb *req, int status)
atomic_set(&req->ctx->cq_timeouts,
atomic_read(&req->ctx->cq_timeouts) + 1);
list_del_init(&timeout->list);
io_req_tw_post_queue(req, status, 0);
io_req_queue_tw_complete(req, status);
return true;
}
return false;
@@ -159,7 +159,7 @@ void io_disarm_next(struct io_kiocb *req)
req->flags &= ~REQ_F_ARM_LTIMEOUT;
if (link && link->opcode == IORING_OP_LINK_TIMEOUT) {
io_remove_next_linked(req);
io_req_tw_post_queue(link, -ECANCELED, 0);
io_req_queue_tw_complete(link, -ECANCELED);
}
} else if (req->flags & REQ_F_LINK_TIMEOUT) {
struct io_ring_ctx *ctx = req->ctx;
@@ -168,7 +168,7 @@ void io_disarm_next(struct io_kiocb *req)
link = io_disarm_linked_timeout(req);
spin_unlock_irq(&ctx->timeout_lock);
if (link)
io_req_tw_post_queue(link, -ECANCELED, 0);
io_req_queue_tw_complete(link, -ECANCELED);
}
if (unlikely((req->flags & REQ_F_FAIL) &&
!(req->flags & REQ_F_HARDLINK)))
@@ -282,11 +282,11 @@ static void io_req_task_link_timeout(struct io_kiocb *req, bool *locked)
ret = io_try_cancel(req->task->io_uring, &cd, issue_flags);
}
io_req_set_res(req, ret ?: -ETIME, 0);
io_req_complete_post(req);
io_req_task_complete(req, locked);
io_put_req(prev);
} else {
io_req_set_res(req, -ETIME, 0);
io_req_complete_post(req);
io_req_task_complete(req, locked);
}
}
io_uring/uring_cmd.c
+1 -1
View File
@@ -56,7 +56,7 @@ void io_uring_cmd_done(struct io_uring_cmd *ioucmd, ssize_t ret, ssize_t res2)
/* order with io_iopoll_req_issued() checking ->iopoll_complete */
smp_store_release(&req->iopoll_completed, 1);
else
__io_req_complete(req, 0);
io_req_complete_post(req, 0);
}
EXPORT_SYMBOL_GPL(io_uring_cmd_done);