Commit 0ddad21d authored by Linus Torvalds's avatar Linus Torvalds

pipe: use exclusive waits when reading or writing

This makes the pipe code use separate wait-queues and exclusive waiting
for readers and writers, avoiding a nasty thundering herd problem when
there are lots of readers waiting for data on a pipe (or, less commonly,
lots of writers waiting for a pipe to have space).

While this isn't a common occurrence in the traditional "use a pipe as a
data transport" case, where you typically only have a single reader and
a single writer process, there is one common special case: using a pipe
as a source of "locking tokens" rather than for data communication.

In particular, the GNU make jobserver code ends up using a pipe as a way
to limit parallelism, where each job consumes a token by reading a byte
from the jobserver pipe, and releases the token by writing a byte back
to the pipe.

This pattern is fairly traditional on Unix, and works very well, but
will waste a lot of time waking up a lot of processes when only a single
reader needs to be woken up when a writer releases a new token.

A simplified test-case of just this pipe interaction is to create 64
processes, and then pass a single token around between them (this
test-case also intentionally passes another token that gets ignored to
test the "wake up next" logic too, in case anybody wonders about it):

    #include <unistd.h>

    int main(int argc, char **argv)
    {
        int fd[2], counters[2];

        pipe(fd);
        counters[0] = 0;
        counters[1] = -1;
        write(fd[1], counters, sizeof(counters));

        /* 64 processes */
        fork(); fork(); fork(); fork(); fork(); fork();

        do {
                int i;
                read(fd[0], &i, sizeof(i));
                if (i < 0)
                        continue;
                counters[0] = i+1;
                write(fd[1], counters, (1+(i & 1)) *sizeof(int));
        } while (counters[0] < 1000000);
        return 0;
    }

and in a perfect world, passing that token around should only cause one
context switch per transfer, when the writer of a token causes a
directed wakeup of just a single reader.

But with the "writer wakes all readers" model we traditionally had, on
my test box the above case causes more than an order of magnitude more
scheduling: instead of the expected ~1M context switches, "perf stat"
shows

        231,852.37 msec task-clock                #   15.857 CPUs utilized
        11,250,961      context-switches          #    0.049 M/sec
           616,304      cpu-migrations            #    0.003 M/sec
             1,648      page-faults               #    0.007 K/sec
 1,097,903,998,514      cycles                    #    4.735 GHz
   120,781,778,352      instructions              #    0.11  insn per cycle
    27,997,056,043      branches                  #  120.754 M/sec
       283,581,233      branch-misses             #    1.01% of all branches

      14.621273891 seconds time elapsed

       0.018243000 seconds user
       3.611468000 seconds sys

before this commit.

After this commit, I get

          5,229.55 msec task-clock                #    3.072 CPUs utilized
         1,212,233      context-switches          #    0.232 M/sec
           103,951      cpu-migrations            #    0.020 M/sec
             1,328      page-faults               #    0.254 K/sec
    21,307,456,166      cycles                    #    4.074 GHz
    12,947,819,999      instructions              #    0.61  insn per cycle
     2,881,985,678      branches                  #  551.096 M/sec
        64,267,015      branch-misses             #    2.23% of all branches

       1.702148350 seconds time elapsed

       0.004868000 seconds user
       0.110786000 seconds sys

instead. Much better.

[ Note! This kernel improvement seems to be very good at triggering a
  race condition in the make jobserver (in GNU make 4.2.1) for me. It's
  a long known bug that was fixed back in June 2017 by GNU make commit
  b552b0525198 ("[SV 51159] Use a non-blocking read with pselect to
  avoid hangs.").

  But there wasn't a new release of GNU make until 4.3 on Jan 19 2020,
  so a number of distributions may still have the buggy version. Some
  have backported the fix to their 4.2.1 release, though, and even
  without the fix it's quite timing-dependent whether the bug actually
  is hit. ]

Josh Triplett says:
 "I've been hammering on your pipe fix patch (switching to exclusive
  wait queues) for a month or so, on several different systems, and I've
  run into no issues with it. The patch *substantially* improves
  parallel build times on large (~100 CPU) systems, both with parallel
  make and with other things that use make's pipe-based jobserver.

  All current distributions (including stable and long-term stable
  distributions) have versions of GNU make that no longer have the
  jobserver bug"
Tested-by: default avatarJosh Triplett <josh@joshtriplett.org>
Signed-off-by: default avatarLinus Torvalds <torvalds@linux-foundation.org>
parent d5226fa6
...@@ -517,7 +517,7 @@ static void wait_for_dump_helpers(struct file *file) ...@@ -517,7 +517,7 @@ static void wait_for_dump_helpers(struct file *file)
pipe_lock(pipe); pipe_lock(pipe);
pipe->readers++; pipe->readers++;
pipe->writers--; pipe->writers--;
wake_up_interruptible_sync(&pipe->wait); wake_up_interruptible_sync(&pipe->rd_wait);
kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN); kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
pipe_unlock(pipe); pipe_unlock(pipe);
...@@ -525,7 +525,7 @@ static void wait_for_dump_helpers(struct file *file) ...@@ -525,7 +525,7 @@ static void wait_for_dump_helpers(struct file *file)
* We actually want wait_event_freezable() but then we need * We actually want wait_event_freezable() but then we need
* to clear TIF_SIGPENDING and improve dump_interrupted(). * to clear TIF_SIGPENDING and improve dump_interrupted().
*/ */
wait_event_interruptible(pipe->wait, pipe->readers == 1); wait_event_interruptible(pipe->rd_wait, pipe->readers == 1);
pipe_lock(pipe); pipe_lock(pipe);
pipe->readers--; pipe->readers--;
......
...@@ -108,16 +108,19 @@ void pipe_double_lock(struct pipe_inode_info *pipe1, ...@@ -108,16 +108,19 @@ void pipe_double_lock(struct pipe_inode_info *pipe1,
/* Drop the inode semaphore and wait for a pipe event, atomically */ /* Drop the inode semaphore and wait for a pipe event, atomically */
void pipe_wait(struct pipe_inode_info *pipe) void pipe_wait(struct pipe_inode_info *pipe)
{ {
DEFINE_WAIT(wait); DEFINE_WAIT(rdwait);
DEFINE_WAIT(wrwait);
/* /*
* Pipes are system-local resources, so sleeping on them * Pipes are system-local resources, so sleeping on them
* is considered a noninteractive wait: * is considered a noninteractive wait:
*/ */
prepare_to_wait(&pipe->wait, &wait, TASK_INTERRUPTIBLE); prepare_to_wait(&pipe->rd_wait, &rdwait, TASK_INTERRUPTIBLE);
prepare_to_wait(&pipe->wr_wait, &wrwait, TASK_INTERRUPTIBLE);
pipe_unlock(pipe); pipe_unlock(pipe);
schedule(); schedule();
finish_wait(&pipe->wait, &wait); finish_wait(&pipe->rd_wait, &rdwait);
finish_wait(&pipe->wr_wait, &wrwait);
pipe_lock(pipe); pipe_lock(pipe);
} }
...@@ -286,7 +289,7 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to) ...@@ -286,7 +289,7 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to)
size_t total_len = iov_iter_count(to); size_t total_len = iov_iter_count(to);
struct file *filp = iocb->ki_filp; struct file *filp = iocb->ki_filp;
struct pipe_inode_info *pipe = filp->private_data; struct pipe_inode_info *pipe = filp->private_data;
bool was_full; bool was_full, wake_next_reader = false;
ssize_t ret; ssize_t ret;
/* Null read succeeds. */ /* Null read succeeds. */
...@@ -344,10 +347,10 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to) ...@@ -344,10 +347,10 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to)
if (!buf->len) { if (!buf->len) {
pipe_buf_release(pipe, buf); pipe_buf_release(pipe, buf);
spin_lock_irq(&pipe->wait.lock); spin_lock_irq(&pipe->rd_wait.lock);
tail++; tail++;
pipe->tail = tail; pipe->tail = tail;
spin_unlock_irq(&pipe->wait.lock); spin_unlock_irq(&pipe->rd_wait.lock);
} }
total_len -= chars; total_len -= chars;
if (!total_len) if (!total_len)
...@@ -384,7 +387,7 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to) ...@@ -384,7 +387,7 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to)
* no data. * no data.
*/ */
if (unlikely(was_full)) { if (unlikely(was_full)) {
wake_up_interruptible_sync_poll(&pipe->wait, EPOLLOUT | EPOLLWRNORM); wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT); kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
} }
...@@ -394,18 +397,23 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to) ...@@ -394,18 +397,23 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to)
* since we've done any required wakeups and there's no need * since we've done any required wakeups and there's no need
* to mark anything accessed. And we've dropped the lock. * to mark anything accessed. And we've dropped the lock.
*/ */
if (wait_event_interruptible(pipe->wait, pipe_readable(pipe)) < 0) if (wait_event_interruptible_exclusive(pipe->rd_wait, pipe_readable(pipe)) < 0)
return -ERESTARTSYS; return -ERESTARTSYS;
__pipe_lock(pipe); __pipe_lock(pipe);
was_full = pipe_full(pipe->head, pipe->tail, pipe->max_usage); was_full = pipe_full(pipe->head, pipe->tail, pipe->max_usage);
wake_next_reader = true;
} }
if (pipe_empty(pipe->head, pipe->tail))
wake_next_reader = false;
__pipe_unlock(pipe); __pipe_unlock(pipe);
if (was_full) { if (was_full) {
wake_up_interruptible_sync_poll(&pipe->wait, EPOLLOUT | EPOLLWRNORM); wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT); kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
} }
if (wake_next_reader)
wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
if (ret > 0) if (ret > 0)
file_accessed(filp); file_accessed(filp);
return ret; return ret;
...@@ -437,6 +445,7 @@ pipe_write(struct kiocb *iocb, struct iov_iter *from) ...@@ -437,6 +445,7 @@ pipe_write(struct kiocb *iocb, struct iov_iter *from)
size_t total_len = iov_iter_count(from); size_t total_len = iov_iter_count(from);
ssize_t chars; ssize_t chars;
bool was_empty = false; bool was_empty = false;
bool wake_next_writer = false;
/* Null write succeeds. */ /* Null write succeeds. */
if (unlikely(total_len == 0)) if (unlikely(total_len == 0))
...@@ -515,16 +524,16 @@ pipe_write(struct kiocb *iocb, struct iov_iter *from) ...@@ -515,16 +524,16 @@ pipe_write(struct kiocb *iocb, struct iov_iter *from)
* it, either the reader will consume it or it'll still * it, either the reader will consume it or it'll still
* be there for the next write. * be there for the next write.
*/ */
spin_lock_irq(&pipe->wait.lock); spin_lock_irq(&pipe->rd_wait.lock);
head = pipe->head; head = pipe->head;
if (pipe_full(head, pipe->tail, pipe->max_usage)) { if (pipe_full(head, pipe->tail, pipe->max_usage)) {
spin_unlock_irq(&pipe->wait.lock); spin_unlock_irq(&pipe->rd_wait.lock);
continue; continue;
} }
pipe->head = head + 1; pipe->head = head + 1;
spin_unlock_irq(&pipe->wait.lock); spin_unlock_irq(&pipe->rd_wait.lock);
/* Insert it into the buffer array */ /* Insert it into the buffer array */
buf = &pipe->bufs[head & mask]; buf = &pipe->bufs[head & mask];
...@@ -576,14 +585,17 @@ pipe_write(struct kiocb *iocb, struct iov_iter *from) ...@@ -576,14 +585,17 @@ pipe_write(struct kiocb *iocb, struct iov_iter *from)
*/ */
__pipe_unlock(pipe); __pipe_unlock(pipe);
if (was_empty) { if (was_empty) {
wake_up_interruptible_sync_poll(&pipe->wait, EPOLLIN | EPOLLRDNORM); wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN); kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
} }
wait_event_interruptible(pipe->wait, pipe_writable(pipe)); wait_event_interruptible_exclusive(pipe->wr_wait, pipe_writable(pipe));
__pipe_lock(pipe); __pipe_lock(pipe);
was_empty = pipe_empty(pipe->head, pipe->tail); was_empty = pipe_empty(pipe->head, pipe->tail);
wake_next_writer = true;
} }
out: out:
if (pipe_full(pipe->head, pipe->tail, pipe->max_usage))
wake_next_writer = false;
__pipe_unlock(pipe); __pipe_unlock(pipe);
/* /*
...@@ -596,9 +608,11 @@ pipe_write(struct kiocb *iocb, struct iov_iter *from) ...@@ -596,9 +608,11 @@ pipe_write(struct kiocb *iocb, struct iov_iter *from)
* wake up pending jobs * wake up pending jobs
*/ */
if (was_empty) { if (was_empty) {
wake_up_interruptible_sync_poll(&pipe->wait, EPOLLIN | EPOLLRDNORM); wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN); kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
} }
if (wake_next_writer)
wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
if (ret > 0 && sb_start_write_trylock(file_inode(filp)->i_sb)) { if (ret > 0 && sb_start_write_trylock(file_inode(filp)->i_sb)) {
int err = file_update_time(filp); int err = file_update_time(filp);
if (err) if (err)
...@@ -642,12 +656,15 @@ pipe_poll(struct file *filp, poll_table *wait) ...@@ -642,12 +656,15 @@ pipe_poll(struct file *filp, poll_table *wait)
unsigned int head, tail; unsigned int head, tail;
/* /*
* Reading only -- no need for acquiring the semaphore. * Reading pipe state only -- no need for acquiring the semaphore.
* *
* But because this is racy, the code has to add the * But because this is racy, the code has to add the
* entry to the poll table _first_ .. * entry to the poll table _first_ ..
*/ */
poll_wait(filp, &pipe->wait, wait); if (filp->f_mode & FMODE_READ)
poll_wait(filp, &pipe->rd_wait, wait);
if (filp->f_mode & FMODE_WRITE)
poll_wait(filp, &pipe->wr_wait, wait);
/* /*
* .. and only then can you do the racy tests. That way, * .. and only then can you do the racy tests. That way,
...@@ -706,7 +723,8 @@ pipe_release(struct inode *inode, struct file *file) ...@@ -706,7 +723,8 @@ pipe_release(struct inode *inode, struct file *file)
pipe->writers--; pipe->writers--;
if (pipe->readers || pipe->writers) { if (pipe->readers || pipe->writers) {
wake_up_interruptible_sync_poll(&pipe->wait, EPOLLIN | EPOLLOUT | EPOLLRDNORM | EPOLLWRNORM | EPOLLERR | EPOLLHUP); wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM | EPOLLERR | EPOLLHUP);
wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM | EPOLLERR | EPOLLHUP);
kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN); kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT); kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
} }
...@@ -789,7 +807,8 @@ struct pipe_inode_info *alloc_pipe_info(void) ...@@ -789,7 +807,8 @@ struct pipe_inode_info *alloc_pipe_info(void)
GFP_KERNEL_ACCOUNT); GFP_KERNEL_ACCOUNT);
if (pipe->bufs) { if (pipe->bufs) {
init_waitqueue_head(&pipe->wait); init_waitqueue_head(&pipe->rd_wait);
init_waitqueue_head(&pipe->wr_wait);
pipe->r_counter = pipe->w_counter = 1; pipe->r_counter = pipe->w_counter = 1;
pipe->max_usage = pipe_bufs; pipe->max_usage = pipe_bufs;
pipe->ring_size = pipe_bufs; pipe->ring_size = pipe_bufs;
...@@ -1007,7 +1026,8 @@ static int wait_for_partner(struct pipe_inode_info *pipe, unsigned int *cnt) ...@@ -1007,7 +1026,8 @@ static int wait_for_partner(struct pipe_inode_info *pipe, unsigned int *cnt)
static void wake_up_partner(struct pipe_inode_info *pipe) static void wake_up_partner(struct pipe_inode_info *pipe)
{ {
wake_up_interruptible(&pipe->wait); wake_up_interruptible(&pipe->rd_wait);
wake_up_interruptible(&pipe->wr_wait);
} }
static int fifo_open(struct inode *inode, struct file *filp) static int fifo_open(struct inode *inode, struct file *filp)
...@@ -1118,13 +1138,13 @@ static int fifo_open(struct inode *inode, struct file *filp) ...@@ -1118,13 +1138,13 @@ static int fifo_open(struct inode *inode, struct file *filp)
err_rd: err_rd:
if (!--pipe->readers) if (!--pipe->readers)
wake_up_interruptible(&pipe->wait); wake_up_interruptible(&pipe->wr_wait);
ret = -ERESTARTSYS; ret = -ERESTARTSYS;
goto err; goto err;
err_wr: err_wr:
if (!--pipe->writers) if (!--pipe->writers)
wake_up_interruptible(&pipe->wait); wake_up_interruptible(&pipe->rd_wait);
ret = -ERESTARTSYS; ret = -ERESTARTSYS;
goto err; goto err;
...@@ -1251,7 +1271,8 @@ static long pipe_set_size(struct pipe_inode_info *pipe, unsigned long arg) ...@@ -1251,7 +1271,8 @@ static long pipe_set_size(struct pipe_inode_info *pipe, unsigned long arg)
pipe->max_usage = nr_slots; pipe->max_usage = nr_slots;
pipe->tail = tail; pipe->tail = tail;
pipe->head = head; pipe->head = head;
wake_up_interruptible_all(&pipe->wait); wake_up_interruptible_all(&pipe->rd_wait);
wake_up_interruptible_all(&pipe->wr_wait);
return pipe->max_usage * PAGE_SIZE; return pipe->max_usage * PAGE_SIZE;
out_revert_acct: out_revert_acct:
......
...@@ -165,8 +165,8 @@ static const struct pipe_buf_operations user_page_pipe_buf_ops = { ...@@ -165,8 +165,8 @@ static const struct pipe_buf_operations user_page_pipe_buf_ops = {
static void wakeup_pipe_readers(struct pipe_inode_info *pipe) static void wakeup_pipe_readers(struct pipe_inode_info *pipe)
{ {
smp_mb(); smp_mb();
if (waitqueue_active(&pipe->wait)) if (waitqueue_active(&pipe->rd_wait))
wake_up_interruptible(&pipe->wait); wake_up_interruptible(&pipe->rd_wait);
kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN); kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
} }
...@@ -462,8 +462,8 @@ static int pipe_to_sendpage(struct pipe_inode_info *pipe, ...@@ -462,8 +462,8 @@ static int pipe_to_sendpage(struct pipe_inode_info *pipe,
static void wakeup_pipe_writers(struct pipe_inode_info *pipe) static void wakeup_pipe_writers(struct pipe_inode_info *pipe)
{ {
smp_mb(); smp_mb();
if (waitqueue_active(&pipe->wait)) if (waitqueue_active(&pipe->wr_wait))
wake_up_interruptible(&pipe->wait); wake_up_interruptible(&pipe->wr_wait);
kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT); kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
} }
......
...@@ -47,7 +47,7 @@ struct pipe_buffer { ...@@ -47,7 +47,7 @@ struct pipe_buffer {
**/ **/
struct pipe_inode_info { struct pipe_inode_info {
struct mutex mutex; struct mutex mutex;
wait_queue_head_t wait; wait_queue_head_t rd_wait, wr_wait;
unsigned int head; unsigned int head;
unsigned int tail; unsigned int tail;
unsigned int max_usage; unsigned int max_usage;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment