> I think this is misguided. Not just the special-casing of eventfd but
> also letting file operations opt-in how locking is done in another
> subystem. That's almost guaranteed to wreak havoc... and sashiko is
> already proving this.

Thanks for the feedback. I agree that the fast-path logic should not be
connected to other subsystems. I have reworked the proposal so that the
optimization is entirely local to eventpoll. I removed the special-casing
of eventfd and the opt-in bit in FOP.

The new version tracks two bits in each epitem:

  - FAST marks an item that is eligible for the lockless callback path.
    Nested epoll and interests using EPOLLEXCLUSIVE, EPOLLET,
    EPOLLONESHOT, or EPOLLWAKEUP do not use the fast path.

  - ACCOUNTED means that the ready-processing path already owns the item:
    it is on rdllist, in the current scan batch, or queued on ovflist,
	the same as the 'notified' before.

As for the question sashiko raised about the fatal branch, I do not modify
the flag in the branch any more. Instead, I deliver the wakeup directly to
avoid list traversal race.

---
 fs/eventpoll.c | 100 ++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 98 insertions(+), 2 deletions(-)

diff --git a/fs/eventpoll.c b/fs/eventpoll.c
index e0c4bf88a838..600d3d2d4763 100644
--- a/fs/eventpoll.c
+++ b/fs/eventpoll.c
@@ -243,6 +243,14 @@ struct eppoll_entry {
 	wait_queue_head_t *whead;
 };
 
+/*
+ * FAST is set for eligible items at insertion. ACCOUNTED tracks FAST items
+ * while the ready path owns them and is cleared immediately before re-poll.
+ */
+enum {
+	EP_STATE_ACCOUNTED	= BIT(0),
+	EP_STATE_FAST		= BIT(1),
+};
 /*
  * Each file descriptor added to the eventpoll interface will
  * have an entry of this type linked to the "rbr" RB tree.
@@ -271,6 +279,9 @@ struct epitem {
 	/* The file descriptor information this item refers to */
 	struct epoll_key ffd;
 
+	/* Atomic state bits, see the EP_STATE_* enum above. */
+	atomic_t state;
+
 	/* List containing poll wait queues */
 	struct eppoll_entry *pwqlist;
 
@@ -620,6 +631,65 @@ static inline bool ep_events_available(struct eventpoll *ep)
 		read_seqcount_retry(&ep->seq, seq);
 }
 
+/*
+ * Let A be ACCOUNTED and B be file readiness. Fully ordered atomic RMWs
+ * pair W(A=0) before R(B) with W(B) before R(A), so a scanner cannot miss
+ * B while its callback skips an accounted item. ep_done_scan() and the
+ * fatal-signal handoff pass queued events between exclusive waiters.
+ * poll_wait users, private modes and special callbacks take the slow path.
+ */
+static inline bool ep_state_test(const struct epitem *epi, unsigned int bit)
+{
+	return atomic_read(&epi->state) & bit;
+}
+
+static inline bool ep_ready_fast_enabled(const struct epitem *epi)
+{
+	return ep_state_test(epi, EP_STATE_FAST);
+}
+
+static inline void ep_set_ready_accounted(struct epitem *epi)
+{
+	if (ep_ready_fast_enabled(epi))
+		atomic_or(EP_STATE_ACCOUNTED, &epi->state);
+}
+
+static inline void ep_prepare_repoll(struct epitem *epi)
+{
+	if (ep_ready_fast_enabled(epi))
+		atomic_fetch_andnot(EP_STATE_ACCOUNTED, &epi->state);
+}
+
+static bool ep_callback_can_skip(struct epitem *epi, __poll_t pollflags)
+{
+	__poll_t events;
+
+	events = READ_ONCE(epi->event.events);
+	
+	/* Only insertion-time candidates may use the lockless path. */
+	if (!ep_ready_fast_enabled(epi))
+		return false;
+	
+	/* ep_modify() can select a private mode that requires the slow path. */
+	if (events & EP_PRIVATE_BITS)
+		return false;
+
+	/* poll_wait users must be woken by this callback. */
+	if (waitqueue_active(&epi->ep->poll_wait))
+		return false;
+
+	/* POLLFREE tears down wait entries; URING_WAKE must propagate. */
+	if (pollflags & (POLLFREE | EPOLL_URING_WAKE))
+		return false;
+
+	/* Unmatched wake keys retain the slow-path callback semantics. */
+	if (pollflags && !(pollflags & events))
+		return false;
+
+	/* Fully ordered RMW pairs the B update with the scanner's A clear. */
+	return atomic_fetch_or(0, &epi->state) & EP_STATE_ACCOUNTED;
+}
+
 #ifdef CONFIG_NET_RX_BUSY_POLL
 /**
  * busy_loop_ep_timeout - check if busy poll has timed out. The timeout value
@@ -1007,6 +1077,7 @@ static void ep_done_scan(struct eventpoll *ep,
 			 * reverses the iteration order into FIFO.
 			 */
 			list_add(&epi->rdllink, &ep->rdllist);
+			ep_set_ready_accounted(epi);
 			ep_pm_stay_awake(epi);
 		}
 	}
@@ -1303,8 +1374,10 @@ static __poll_t __ep_eventpoll_poll(struct file *file, poll_table *wait, int dep
 	mutex_lock_nested(&ep->mtx, depth);
 	ep_start_scan(ep, &scan_batch);
 	list_for_each_entry_safe(epi, tmp, &scan_batch, rdllink) {
+		ep_prepare_repoll(epi);
 		if (ep_item_poll(epi, &pt, depth + 1)) {
 			res = EPOLLIN | EPOLLRDNORM;
+			ep_set_ready_accounted(epi);
 			break;
 		} else {
 			/*
@@ -1497,6 +1570,9 @@ static int ep_poll_callback(wait_queue_entry_t *wait, unsigned mode, int sync, v
 	unsigned long flags;
 	int ewake = 0;
 
+	if (ep_callback_can_skip(epi, pollflags))
+		return 1;
+
 	spin_lock_irqsave(&ep->lock, flags);
 
 	ep_set_busy_poll_napi_id(epi);
@@ -1529,11 +1605,13 @@ static int ep_poll_callback(wait_queue_entry_t *wait, unsigned mode, int sync, v
 		if (!epi_on_ovflist(epi)) {
 			epi->ovflist_next = READ_ONCE(ep->ovflist);
 			WRITE_ONCE(ep->ovflist, epi);
+			ep_set_ready_accounted(epi);
 			ep_pm_stay_awake_rcu(epi);
 		}
 	} else if (!ep_is_linked(epi)) {
 		/* In the usual case, add event to ready list. */
 		list_add_tail(&epi->rdllink, &ep->rdllist);
+		ep_set_ready_accounted(epi);
 		ep_pm_stay_awake_rcu(epi);
 	}
 
@@ -1912,6 +1990,13 @@ static int ep_insert(struct ep_ctl_ctx *ctx, struct eventpoll *ep,
 	if (IS_ERR(epi))
 		return PTR_ERR(epi);
 
+	/*
+	 * Lockless candidates are plain level-triggered interests;
+	 * private modes and nested epoll change the callback contract.
+	 */
+	if (!tep && !(event->events & EP_PRIVATE_BITS))
+		atomic_set(&epi->state, EP_STATE_FAST);
+
 	error = ep_register_epitem(ctx, ep, epi, tep, full_check);
 	if (error)
 		return error;
@@ -1956,6 +2041,7 @@ static int ep_insert(struct ep_ctl_ctx *ctx, struct eventpoll *ep,
 
 	if (revents && !ep_is_linked(epi)) {
 		list_add_tail(&epi->rdllink, &ep->rdllist);
+		ep_set_ready_accounted(epi);
 		ep_pm_stay_awake(epi);
 
 		if (waitqueue_active(&ep->wq))
@@ -1992,7 +2078,7 @@ static int ep_modify(struct eventpoll *ep, struct epitem *epi,
 	 * otherwise we might miss an event that happens between the
 	 * f_op->poll() call and the new event set registering.
 	 */
-	epi->event.events = event->events; /* need barrier below */
+	WRITE_ONCE(epi->event.events, event->events); /* need barrier below */
 	epi->event.data = event->data; /* protected by mtx */
 	if (epi->event.events & EPOLLWAKEUP) {
 		if (!ep_has_wakeup_source(epi))
@@ -2031,6 +2117,7 @@ static int ep_modify(struct eventpoll *ep, struct epitem *epi,
 		spin_lock_irq(&ep->lock);
 		if (!ep_is_linked(epi)) {
 			list_add_tail(&epi->rdllink, &ep->rdllist);
+			ep_set_ready_accounted(epi);
 			ep_pm_stay_awake(epi);
 
 			/* Notify waiting tasks that events are available */
@@ -2084,6 +2171,7 @@ static int ep_deliver_event(struct eventpoll *ep, struct epitem *epi,
 		__pm_relax(ws);
 	}
 
+	ep_prepare_repoll(epi);
 	list_del_init(&epi->rdllink);
 
 	/*
@@ -2104,13 +2192,15 @@ static int ep_deliver_event(struct eventpoll *ep, struct epitem *epi,
 		 * attempt.
 		 */
 		list_add(&epi->rdllink, scan_batch);
+		ep_set_ready_accounted(epi);
 		ep_pm_stay_awake(epi);
 		return -EFAULT;
 	}
 	*uevents = next;
 
 	if (epi->event.events & EPOLLONESHOT) {
-		epi->event.events &= EP_PRIVATE_BITS;
+		WRITE_ONCE(epi->event.events,
+			   READ_ONCE(epi->event.events) & EP_PRIVATE_BITS);
 	} else if (!(epi->event.events & EPOLLET)) {
 		/*
 		 * Level-triggered: re-queue so the next epoll_wait()
@@ -2120,6 +2210,7 @@ static int ep_deliver_event(struct eventpoll *ep, struct epitem *epi,
 		 * during scans.
 		 */
 		list_add_tail(&epi->rdllink, &ep->rdllist);
+		ep_set_ready_accounted(epi);
 		ep_pm_stay_awake(epi);
 	}
 	return 1;
@@ -2287,6 +2378,11 @@ static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
 	while (1) {
 		if (eavail) {
 			res = ep_try_send_events(ep, events, maxevents);
+			if (res == -EINTR) {
+				spin_lock_irq(&ep->lock);
+				wake_up(&ep->wq);
+				spin_unlock_irq(&ep->lock);
+			}
 			if (res)
 				return res;
 		}
-- 
2.43.0