Lines Matching refs:pipe
3 * linux/fs/pipe.c
36 * New pipe buffers will be restricted to this size while the user is exceeding
37 * their pipe buffer quota. The general pipe use case needs at least two
39 * than two, then a write to a non-empty pipe may block even if the pipe is not
42 * pipe before reading tokens: https://lore.kernel.org/lkml/1628086770.5rn8p04n6j.none@localhost/.
44 * Users can reduce their pipe buffers with F_SETPIPE_SZ below this at their
45 * own risk, namely: pipe writes to non-full pipes may block until the pipe is
51 * The max size that a non-root user is allowed to grow the pipe. Can
52 * be set by root in /proc/sys/fs/pipe-max-size
89 void pipe_lock(struct pipe_inode_info *pipe)
91 if (pipe->files)
92 mutex_lock(&pipe->mutex);
96 void pipe_unlock(struct pipe_inode_info *pipe)
98 if (pipe->files)
99 mutex_unlock(&pipe->mutex);
115 static void anon_pipe_buf_release(struct pipe_inode_info *pipe,
125 if (page_count(page) == 1 && !pipe->tmp_page)
126 pipe->tmp_page = page;
131 static bool anon_pipe_buf_try_steal(struct pipe_inode_info *pipe,
145 * @pipe: the pipe that the buffer belongs to
155 bool generic_pipe_buf_try_steal(struct pipe_inode_info *pipe,
175 * @pipe: the pipe that the buffer belongs to
181 * pipe into another.
183 bool generic_pipe_buf_get(struct pipe_inode_info *pipe, struct pipe_buffer *buf)
191 * @pipe: the pipe that the buffer belongs to
197 void generic_pipe_buf_release(struct pipe_inode_info *pipe,
210 /* Done while waiting without holding the pipe lock - thus the READ_ONCE() */
211 static inline bool pipe_readable(const struct pipe_inode_info *pipe)
213 unsigned int head = READ_ONCE(pipe->head);
214 unsigned int tail = READ_ONCE(pipe->tail);
215 unsigned int writers = READ_ONCE(pipe->writers);
220 static inline unsigned int pipe_update_tail(struct pipe_inode_info *pipe,
224 pipe_buf_release(pipe, buf);
227 * If the pipe has a watch_queue, we need additional protection
231 if (pipe_has_watch_queue(pipe)) {
232 spin_lock_irq(&pipe->rd_wait.lock);
235 pipe->note_loss = true;
237 pipe->tail = ++tail;
238 spin_unlock_irq(&pipe->rd_wait.lock);
246 pipe->tail = ++tail;
255 struct pipe_inode_info *pipe = filp->private_data;
264 mutex_lock(&pipe->mutex);
267 * We only wake up writers if the pipe was full when we started
274 was_full = pipe_full(pipe->head, pipe->tail, pipe->max_usage);
277 unsigned int head = smp_load_acquire(&pipe->head);
278 unsigned int tail = pipe->tail;
279 unsigned int mask = pipe->ring_size - 1;
282 if (pipe->note_loss) {
301 pipe->note_loss = false;
306 struct pipe_buffer *buf = &pipe->bufs[tail & mask];
320 error = pipe_buf_confirm(pipe, buf);
344 tail = pipe_update_tail(pipe, buf, tail);
352 if (!pipe->writers)
361 mutex_unlock(&pipe->mutex);
367 * pipe buffer, and might have made space in the buffers
370 * You can't make zero-sized pipe buffers by doing an empty
377 * _very_ unlikely case that the pipe was full, but we got
381 wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
382 kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
390 if (wait_event_interruptible_exclusive(pipe->rd_wait, pipe_readable(pipe)) < 0)
393 mutex_lock(&pipe->mutex);
394 was_full = pipe_full(pipe->head, pipe->tail, pipe->max_usage);
397 if (pipe_empty(pipe->head, pipe->tail))
399 mutex_unlock(&pipe->mutex);
402 wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
404 wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
405 kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
416 /* Done while waiting without holding the pipe lock - thus the READ_ONCE() */
417 static inline bool pipe_writable(const struct pipe_inode_info *pipe)
419 unsigned int head = READ_ONCE(pipe->head);
420 unsigned int tail = READ_ONCE(pipe->tail);
421 unsigned int max_usage = READ_ONCE(pipe->max_usage);
424 !READ_ONCE(pipe->readers);
431 struct pipe_inode_info *pipe = filp->private_data;
441 * the pipe.
443 * pipe locked.
444 * If we had to support locking a normal pipe and a notification pipe at
448 if (pipe_has_watch_queue(pipe))
455 mutex_lock(&pipe->mutex);
457 if (!pipe->readers) {
471 head = pipe->head;
472 was_empty = pipe_empty(head, pipe->tail);
475 unsigned int mask = pipe->ring_size - 1;
476 struct pipe_buffer *buf = &pipe->bufs[(head - 1) & mask];
481 ret = pipe_buf_confirm(pipe, buf);
498 if (!pipe->readers) {
505 head = pipe->head;
506 if (!pipe_full(head, pipe->tail, pipe->max_usage)) {
507 unsigned int mask = pipe->ring_size - 1;
509 struct page *page = pipe->tmp_page;
518 pipe->tmp_page = page;
526 pipe->head = head + 1;
529 buf = &pipe->bufs[head & mask];
538 pipe->tmp_page = NULL;
553 if (!pipe_full(head, pipe->tail, pipe->max_usage))
570 * We're going to release the pipe lock and wait for more
572 * after waiting we need to re-check whether the pipe
575 mutex_unlock(&pipe->mutex);
577 wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
578 kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
579 wait_event_interruptible_exclusive(pipe->wr_wait, pipe_writable(pipe));
580 mutex_lock(&pipe->mutex);
581 was_empty = pipe_empty(pipe->head, pipe->tail);
585 if (pipe_full(pipe->head, pipe->tail, pipe->max_usage))
587 mutex_unlock(&pipe->mutex);
598 * Epoll nonsensically wants a wakeup whether the pipe
601 if (was_empty || pipe->poll_usage)
602 wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
603 kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
605 wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
617 struct pipe_inode_info *pipe = filp->private_data;
622 mutex_lock(&pipe->mutex);
624 head = pipe->head;
625 tail = pipe->tail;
626 mask = pipe->ring_size - 1;
629 count += pipe->bufs[tail & mask].len;
632 mutex_unlock(&pipe->mutex);
639 mutex_lock(&pipe->mutex);
640 ret = watch_queue_set_size(pipe, arg);
641 mutex_unlock(&pipe->mutex);
647 pipe, (struct watch_notification_filter __user *)arg);
660 struct pipe_inode_info *pipe = filp->private_data;
664 WRITE_ONCE(pipe->poll_usage, true);
667 * Reading pipe state only -- no need for acquiring the semaphore.
673 poll_wait(filp, &pipe->rd_wait, wait);
675 poll_wait(filp, &pipe->wr_wait, wait);
682 head = READ_ONCE(pipe->head);
683 tail = READ_ONCE(pipe->tail);
689 if (!pipe->writers && filp->f_version != pipe->w_counter)
694 if (!pipe_full(head, tail, pipe->max_usage))
700 if (!pipe->readers)
707 static void put_pipe_info(struct inode *inode, struct pipe_inode_info *pipe)
712 if (!--pipe->files) {
719 free_pipe_info(pipe);
725 struct pipe_inode_info *pipe = file->private_data;
727 mutex_lock(&pipe->mutex);
729 pipe->readers--;
731 pipe->writers--;
734 if (!pipe->readers != !pipe->writers) {
735 wake_up_interruptible_all(&pipe->rd_wait);
736 wake_up_interruptible_all(&pipe->wr_wait);
737 kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
738 kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
740 mutex_unlock(&pipe->mutex);
742 put_pipe_info(inode, pipe);
749 struct pipe_inode_info *pipe = filp->private_data;
752 mutex_lock(&pipe->mutex);
754 retval = fasync_helper(fd, filp, on, &pipe->fasync_readers);
756 retval = fasync_helper(fd, filp, on, &pipe->fasync_writers);
759 fasync_helper(-1, filp, 0, &pipe->fasync_readers);
761 mutex_unlock(&pipe->mutex);
792 struct pipe_inode_info *pipe;
798 pipe = kzalloc(sizeof(struct pipe_inode_info), GFP_KERNEL_ACCOUNT);
799 if (pipe == NULL)
815 pipe->bufs = kcalloc(pipe_bufs, sizeof(struct pipe_buffer),
818 if (pipe->bufs) {
819 init_waitqueue_head(&pipe->rd_wait);
820 init_waitqueue_head(&pipe->wr_wait);
821 pipe->r_counter = pipe->w_counter = 1;
822 pipe->max_usage = pipe_bufs;
823 pipe->ring_size = pipe_bufs;
824 pipe->nr_accounted = pipe_bufs;
825 pipe->user = user;
826 mutex_init(&pipe->mutex);
827 lock_set_cmp_fn(&pipe->mutex, pipe_lock_cmp_fn, NULL);
828 return pipe;
833 kfree(pipe);
839 void free_pipe_info(struct pipe_inode_info *pipe)
844 if (pipe->watch_queue)
845 watch_queue_clear(pipe->watch_queue);
848 (void) account_pipe_buffers(pipe->user, pipe->nr_accounted, 0);
849 free_uid(pipe->user);
850 for (i = 0; i < pipe->ring_size; i++) {
851 struct pipe_buffer *buf = pipe->bufs + i;
853 pipe_buf_release(pipe, buf);
856 if (pipe->watch_queue)
857 put_watch_queue(pipe->watch_queue);
859 if (pipe->tmp_page)
860 __free_page(pipe->tmp_page);
861 kfree(pipe->bufs);
862 kfree(pipe);
872 return dynamic_dname(buffer, buflen, "pipe:[%lu]",
883 struct pipe_inode_info *pipe;
890 pipe = alloc_pipe_info();
891 if (!pipe)
894 inode->i_pipe = pipe;
895 pipe->files = 2;
896 pipe->readers = pipe->writers = 1;
988 /* pipe groks IOCB_NOWAIT */
1014 * a pipe. It's not the way Unix traditionally does this, though.
1043 SYSCALL_DEFINE1(pipe, int __user *, fildes)
1049 * This is the stupid "wait for pipe to be readable or writable"
1057 void pipe_wait_readable(struct pipe_inode_info *pipe)
1059 pipe_unlock(pipe);
1060 wait_event_interruptible(pipe->rd_wait, pipe_readable(pipe));
1061 pipe_lock(pipe);
1064 void pipe_wait_writable(struct pipe_inode_info *pipe)
1066 pipe_unlock(pipe);
1067 wait_event_interruptible(pipe->wr_wait, pipe_writable(pipe));
1068 pipe_lock(pipe);
1073 * holding the pipe lock, so "*cnt" is stable and we know a wakeup cannot
1078 * because of the pipe lock, we can check the condition before being on
1081 * We use the 'rd_wait' waitqueue for pipe partner waiting.
1083 static int wait_for_partner(struct pipe_inode_info *pipe, unsigned int *cnt)
1089 prepare_to_wait(&pipe->rd_wait, &rdwait, TASK_INTERRUPTIBLE);
1090 pipe_unlock(pipe);
1092 finish_wait(&pipe->rd_wait, &rdwait);
1093 pipe_lock(pipe);
1100 static void wake_up_partner(struct pipe_inode_info *pipe)
1102 wake_up_interruptible_all(&pipe->rd_wait);
1107 struct pipe_inode_info *pipe;
1115 pipe = inode->i_pipe;
1116 pipe->files++;
1120 pipe = alloc_pipe_info();
1121 if (!pipe)
1123 pipe->files = 1;
1128 free_pipe_info(pipe);
1129 pipe = inode->i_pipe;
1131 inode->i_pipe = pipe;
1135 filp->private_data = pipe;
1136 /* OK, we have a pipe and it's pinned down */
1138 mutex_lock(&pipe->mutex);
1150 pipe->r_counter++;
1151 if (pipe->readers++ == 0)
1152 wake_up_partner(pipe);
1154 if (!is_pipe && !pipe->writers) {
1158 filp->f_version = pipe->w_counter;
1160 if (wait_for_partner(pipe, &pipe->w_counter))
1173 if (!is_pipe && (filp->f_flags & O_NONBLOCK) && !pipe->readers)
1176 pipe->w_counter++;
1177 if (!pipe->writers++)
1178 wake_up_partner(pipe);
1180 if (!is_pipe && !pipe->readers) {
1181 if (wait_for_partner(pipe, &pipe->r_counter))
1194 pipe->readers++;
1195 pipe->writers++;
1196 pipe->r_counter++;
1197 pipe->w_counter++;
1198 if (pipe->readers == 1 || pipe->writers == 1)
1199 wake_up_partner(pipe);
1208 mutex_unlock(&pipe->mutex);
1212 if (!--pipe->readers)
1213 wake_up_interruptible(&pipe->wr_wait);
1218 if (!--pipe->writers)
1219 wake_up_interruptible_all(&pipe->rd_wait);
1224 mutex_unlock(&pipe->mutex);
1226 put_pipe_info(inode, pipe);
1243 * Currently we rely on the pipe array holding a power-of-2 number
1251 /* Minimum pipe size, as required by POSIX */
1259 * Resize the pipe ring to a number of slots.
1261 * Note the pipe can be reduced in capacity, but only if the current
1265 int pipe_resize_ring(struct pipe_inode_info *pipe, unsigned int nr_slots)
1275 spin_lock_irq(&pipe->rd_wait.lock);
1276 mask = pipe->ring_size - 1;
1277 head = pipe->head;
1278 tail = pipe->tail;
1282 spin_unlock_irq(&pipe->rd_wait.lock);
1288 * The pipe array wraps around, so just start the new one at zero
1295 memcpy(bufs, pipe->bufs + t,
1298 unsigned int tsize = pipe->ring_size - t;
1300 memcpy(bufs + tsize, pipe->bufs,
1302 memcpy(bufs, pipe->bufs + t,
1310 kfree(pipe->bufs);
1311 pipe->bufs = bufs;
1312 pipe->ring_size = nr_slots;
1313 if (pipe->max_usage > nr_slots)
1314 pipe->max_usage = nr_slots;
1315 pipe->tail = tail;
1316 pipe->head = head;
1318 if (!pipe_has_watch_queue(pipe)) {
1319 pipe->max_usage = nr_slots;
1320 pipe->nr_accounted = nr_slots;
1323 spin_unlock_irq(&pipe->rd_wait.lock);
1326 wake_up_interruptible(&pipe->wr_wait);
1331 * Allocate a new array of pipe buffers and copy the info over. Returns the
1332 * pipe size if successful, or return -ERROR on error.
1334 static long pipe_set_size(struct pipe_inode_info *pipe, unsigned int arg)
1340 if (pipe_has_watch_queue(pipe))
1350 * If trying to increase the pipe capacity, check that an
1353 * Decreasing the pipe capacity is always permitted, even
1356 if (nr_slots > pipe->max_usage &&
1360 user_bufs = account_pipe_buffers(pipe->user, pipe->nr_accounted, nr_slots);
1362 if (nr_slots > pipe->max_usage &&
1370 ret = pipe_resize_ring(pipe, nr_slots);
1374 return pipe->max_usage * PAGE_SIZE;
1377 (void) account_pipe_buffers(pipe->user, nr_slots, pipe->nr_accounted);
1383 * not enough to verify that this is a pipe.
1387 struct pipe_inode_info *pipe = file->private_data;
1389 if (file->f_op != &pipefifo_fops || !pipe)
1391 if (for_splice && pipe_has_watch_queue(pipe))
1393 return pipe;
1398 struct pipe_inode_info *pipe;
1401 pipe = get_pipe_info(file, false);
1402 if (!pipe)
1405 mutex_lock(&pipe->mutex);
1409 ret = pipe_set_size(pipe, arg);
1412 ret = pipe->max_usage * PAGE_SIZE;
1419 mutex_unlock(&pipe->mutex);
1432 * d_name - pipe: will go nicely and kill the special-casing in procfs.
1481 .procname = "pipe-max-size",
1488 .procname = "pipe-user-pages-hard",
1495 .procname = "pipe-user-pages-soft",