Re: [RFC PATCH v2 0/1] pipe: only enable the extra wake_up(rd_wait) for edge-triggered consumers

[Oleg Nesterov <oleg@redhat.com>]

Damn, sorry, the subject is wrong... ignore, will resend

On 07/27, Oleg Nesterov wrote:
>
> pipe_poll() unconditionally sets poll_usage on the first call, forcing
> anon_pipe_write() to wake up readers on every write even if the pipe was
> not empty. But this is only needed for edge-triggered consumers: epoll
> with EPOLLET and io_uring without (unsupported) IORING_POLL_ADD_LEVEL.
> poll() and select() users pay for it for no reason.
> 
> Change pipe_poll() to set ->pipe_usage only if wait->_key & EPOLLET is
> true, this check should catch both users.
> 
> While at it, add READ_ONCE() in anon_pipe_write() to pair with WRITE_ONCE()
> in pipe_poll().
> 
> Test-case for epoll:
> 
> 	#include <unistd.h>
> 	#include <sys/epoll.h>
> 	#include <assert.h>
> 
> 	int main(void)
> 	{
> 		int pfd[2], efd;
> 		struct epoll_event evt = { .events = EPOLLIN | EPOLLET };
> 
> 		pipe(pfd);
> 		efd = epoll_create1(0);
> 		epoll_ctl(efd, EPOLL_CTL_ADD, pfd[0], &evt);
> 
> 		for (int i = 0; i < 2; ++i) {
> 			write(pfd[1], "", 1);
> 			assert(epoll_wait(efd, &evt, 1, 0) == 1);
> 		}
> 
> 		return 0;
> 	}
> 
> Test-case for io_uring:
> 
> 	#include <unistd.h>
> 	#include <sys/mman.h>
> 	#include <sys/epoll.h>
> 	#include <sys/syscall.h>
> 	#include <linux/io_uring.h>
> 	#include <assert.h>
> 
> 	int main(void)
> 	{
> 		struct io_uring_params p = {};
> 		int fd, pfd[2];
> 
> 		pipe(pfd);
> 
> 		fd = syscall(SYS_io_uring_setup, 2, &p);
> 		assert(fd >= 0);
> 
> 		void *ring = mmap(0, p.cq_off.cqes + p.cq_entries * sizeof(struct io_uring_cqe),
> 				  PROT_READ | PROT_WRITE, MAP_SHARED, fd, IORING_OFF_SQ_RING);
> 		assert(ring != MAP_FAILED);
> 		*(unsigned *)(ring + p.sq_off.tail) = 1;
> 
> 		struct io_uring_sqe *sqes = mmap(0, p.sq_entries * sizeof(*sqes),
> 				  PROT_READ | PROT_WRITE, MAP_SHARED, fd, IORING_OFF_SQES);
> 		assert(sqes != MAP_FAILED);
> 		sqes[0].opcode = IORING_OP_POLL_ADD;
> 		sqes[0].fd = pfd[0];
> 		sqes[0].len = IORING_POLL_ADD_MULTI;
> 		sqes[0].poll32_events = EPOLLIN;
> 
> 		syscall(SYS_io_uring_enter, fd, 1, 0, 0, 0, 0);
> 
> 		unsigned *cq_head = ring + p.cq_off.head;
> 		unsigned *cq_tail = ring + p.cq_off.tail;
> 		for (int i = 0; i < 2; ++i) {
> 			write(pfd[1], "", 1);
> 			syscall(SYS_io_uring_enter, fd, 0, 0, IORING_ENTER_GETEVENTS, 0, 0);
> 			assert(*cq_tail == ++*cq_head);
> 		}
> 
> 		return 0;
> 	}
> 
> Signed-off-by: Oleg Nesterov <oleg@redhat.com>
> ---
>  fs/pipe.c | 7 ++++---
>  1 file changed, 4 insertions(+), 3 deletions(-)
> 
> diff --git a/fs/pipe.c b/fs/pipe.c
> index 429b0714ec57..98b1e2385103 100644
> --- a/fs/pipe.c
> +++ b/fs/pipe.c
> @@ -689,7 +689,7 @@ anon_pipe_write(struct kiocb *iocb, struct iov_iter *from)
>  	 * Epoll nonsensically wants a wakeup whether the pipe
>  	 * was already empty or not.
>  	 */
> -	if (was_empty || pipe->poll_usage)
> +	if (was_empty || READ_ONCE(pipe->poll_usage))
>  		wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
>  	kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
>  	if (wake_next_writer)
> @@ -752,7 +752,6 @@ static long pipe_ioctl(struct file *filp, unsigned int cmd, unsigned long arg)
>  	}
>  }
>  
> -/* No kernel lock held - fine */
>  static __poll_t
>  pipe_poll(struct file *filp, poll_table *wait)
>  {
> @@ -761,7 +760,9 @@ pipe_poll(struct file *filp, poll_table *wait)
>  	union pipe_index idx;
>  
>  	/* Epoll has some historical nasty semantics, this enables them */
> -	if (unlikely(!READ_ONCE(pipe->poll_usage)))
> +	if ((filp->f_mode & FMODE_READ) &&
> +	    wait && (wait->_key & EPOLLET) &&
> +	    unlikely(!READ_ONCE(pipe->poll_usage)))
>  		WRITE_ONCE(pipe->poll_usage, true);
>  
>  	/*
> -- 
> 2.52.0
>