xref: /freebsd/sys/kern/sys_pipe.c (revision 601752d5a7bef087e755da5a2b158fa35cb51ccb)
1 /*
2  * Copyright (c) 1996 John S. Dyson
3  * All rights reserved.
4  *
5  * Redistribution and use in source and binary forms, with or without
6  * modification, are permitted provided that the following conditions
7  * are met:
8  * 1. Redistributions of source code must retain the above copyright
9  *    notice immediately at the beginning of the file, without modification,
10  *    this list of conditions, and the following disclaimer.
11  * 2. Redistributions in binary form must reproduce the above copyright
12  *    notice, this list of conditions and the following disclaimer in the
13  *    documentation and/or other materials provided with the distribution.
14  * 3. Absolutely no warranty of function or purpose is made by the author
15  *    John S. Dyson.
16  * 4. Modifications may be freely made to this file if the above conditions
17  *    are met.
18  *
19  * $Id: sys_pipe.c,v 1.50 1999/02/04 23:50:49 dillon Exp $
20  */
21 
22 /*
23  * This file contains a high-performance replacement for the socket-based
24  * pipes scheme originally used in FreeBSD/4.4Lite.  It does not support
25  * all features of sockets, but does do everything that pipes normally
26  * do.
27  */
28 
29 /*
30  * This code has two modes of operation, a small write mode and a large
31  * write mode.  The small write mode acts like conventional pipes with
32  * a kernel buffer.  If the buffer is less than PIPE_MINDIRECT, then the
33  * "normal" pipe buffering is done.  If the buffer is between PIPE_MINDIRECT
34  * and PIPE_SIZE in size, it is fully mapped and wired into the kernel, and
35  * the receiving process can copy it directly from the pages in the sending
36  * process.
37  *
38  * If the sending process receives a signal, it is possible that it will
39  * go away, and certainly its address space can change, because control
40  * is returned back to the user-mode side.  In that case, the pipe code
41  * arranges to copy the buffer supplied by the user process, to a pageable
42  * kernel buffer, and the receiving process will grab the data from the
43  * pageable kernel buffer.  Since signals don't happen all that often,
44  * the copy operation is normally eliminated.
45  *
46  * The constant PIPE_MINDIRECT is chosen to make sure that buffering will
47  * happen for small transfers so that the system will not spend all of
48  * its time context switching.  PIPE_SIZE is constrained by the
49  * amount of kernel virtual memory.
50  */
51 
52 #include <sys/param.h>
53 #include <sys/systm.h>
54 #include <sys/proc.h>
55 #include <sys/fcntl.h>
56 #include <sys/file.h>
57 #include <sys/filedesc.h>
58 #include <sys/filio.h>
59 #include <sys/ttycom.h>
60 #include <sys/stat.h>
61 #include <sys/poll.h>
62 #include <sys/select.h>
63 #include <sys/signalvar.h>
64 #include <sys/sysproto.h>
65 #include <sys/pipe.h>
66 #include <sys/uio.h>
67 
68 #include <vm/vm.h>
69 #include <vm/vm_prot.h>
70 #include <vm/vm_param.h>
71 #include <sys/lock.h>
72 #include <vm/vm_object.h>
73 #include <vm/vm_kern.h>
74 #include <vm/vm_extern.h>
75 #include <vm/pmap.h>
76 #include <vm/vm_map.h>
77 #include <vm/vm_page.h>
78 #include <vm/vm_zone.h>
79 
80 /*
81  * Use this define if you want to disable *fancy* VM things.  Expect an
82  * approx 30% decrease in transfer rate.  This could be useful for
83  * NetBSD or OpenBSD.
84  */
85 /* #define PIPE_NODIRECT */
86 
87 /*
88  * interfaces to the outside world
89  */
90 static int pipe_read __P((struct file *fp, struct uio *uio,
91 		struct ucred *cred, int flags));
92 static int pipe_write __P((struct file *fp, struct uio *uio,
93 		struct ucred *cred, int flags));
94 static int pipe_close __P((struct file *fp, struct proc *p));
95 static int pipe_poll __P((struct file *fp, int events, struct ucred *cred,
96 		struct proc *p));
97 static int pipe_ioctl __P((struct file *fp, u_long cmd, caddr_t data, struct proc *p));
98 
99 static struct fileops pipeops =
100     { pipe_read, pipe_write, pipe_ioctl, pipe_poll, pipe_close };
101 
102 /*
103  * Default pipe buffer size(s), this can be kind-of large now because pipe
104  * space is pageable.  The pipe code will try to maintain locality of
105  * reference for performance reasons, so small amounts of outstanding I/O
106  * will not wipe the cache.
107  */
108 #define MINPIPESIZE (PIPE_SIZE/3)
109 #define MAXPIPESIZE (2*PIPE_SIZE/3)
110 
111 /*
112  * Maximum amount of kva for pipes -- this is kind-of a soft limit, but
113  * is there so that on large systems, we don't exhaust it.
114  */
115 #define MAXPIPEKVA (8*1024*1024)
116 
117 /*
118  * Limit for direct transfers, we cannot, of course limit
119  * the amount of kva for pipes in general though.
120  */
121 #define LIMITPIPEKVA (16*1024*1024)
122 
123 /*
124  * Limit the number of "big" pipes
125  */
126 #define LIMITBIGPIPES	32
127 static int nbigpipe;
128 
129 static int amountpipekva;
130 
131 static void pipeclose __P((struct pipe *cpipe));
132 static void pipeinit __P((struct pipe *cpipe));
133 static __inline int pipelock __P((struct pipe *cpipe, int catch));
134 static __inline void pipeunlock __P((struct pipe *cpipe));
135 static __inline void pipeselwakeup __P((struct pipe *cpipe));
136 #ifndef PIPE_NODIRECT
137 static int pipe_build_write_buffer __P((struct pipe *wpipe, struct uio *uio));
138 static void pipe_destroy_write_buffer __P((struct pipe *wpipe));
139 static int pipe_direct_write __P((struct pipe *wpipe, struct uio *uio));
140 static void pipe_clone_write_buffer __P((struct pipe *wpipe));
141 #endif
142 static void pipespace __P((struct pipe *cpipe));
143 
144 static vm_zone_t pipe_zone;
145 
146 /*
147  * The pipe system call for the DTYPE_PIPE type of pipes
148  */
149 
150 /* ARGSUSED */
151 int
152 pipe(p, uap)
153 	struct proc *p;
154 	struct pipe_args /* {
155 		int	dummy;
156 	} */ *uap;
157 {
158 	register struct filedesc *fdp = p->p_fd;
159 	struct file *rf, *wf;
160 	struct pipe *rpipe, *wpipe;
161 	int fd, error;
162 
163 	if (pipe_zone == NULL)
164 		pipe_zone = zinit("PIPE", sizeof (struct pipe), 0, 0, 4);
165 
166 	rpipe = zalloc( pipe_zone);
167 	pipeinit(rpipe);
168 	rpipe->pipe_state |= PIPE_DIRECTOK;
169 	wpipe = zalloc( pipe_zone);
170 	pipeinit(wpipe);
171 	wpipe->pipe_state |= PIPE_DIRECTOK;
172 
173 	error = falloc(p, &rf, &fd);
174 	if (error)
175 		goto free2;
176 	p->p_retval[0] = fd;
177 	rf->f_flag = FREAD | FWRITE;
178 	rf->f_type = DTYPE_PIPE;
179 	rf->f_ops = &pipeops;
180 	rf->f_data = (caddr_t)rpipe;
181 	error = falloc(p, &wf, &fd);
182 	if (error)
183 		goto free3;
184 	wf->f_flag = FREAD | FWRITE;
185 	wf->f_type = DTYPE_PIPE;
186 	wf->f_ops = &pipeops;
187 	wf->f_data = (caddr_t)wpipe;
188 	p->p_retval[1] = fd;
189 
190 	rpipe->pipe_peer = wpipe;
191 	wpipe->pipe_peer = rpipe;
192 
193 	return (0);
194 free3:
195 	ffree(rf);
196 	fdp->fd_ofiles[p->p_retval[0]] = 0;
197 free2:
198 	(void)pipeclose(wpipe);
199 	(void)pipeclose(rpipe);
200 	return (error);
201 }
202 
203 /*
204  * Allocate kva for pipe circular buffer, the space is pageable
205  */
206 static void
207 pipespace(cpipe)
208 	struct pipe *cpipe;
209 {
210 	int npages, error;
211 
212 	npages = round_page(cpipe->pipe_buffer.size)/PAGE_SIZE;
213 	/*
214 	 * Create an object, I don't like the idea of paging to/from
215 	 * kernel_object.
216 	 * XXX -- minor change needed here for NetBSD/OpenBSD VM systems.
217 	 */
218 	cpipe->pipe_buffer.object = vm_object_allocate(OBJT_DEFAULT, npages);
219 	cpipe->pipe_buffer.buffer = (caddr_t) vm_map_min(kernel_map);
220 
221 	/*
222 	 * Insert the object into the kernel map, and allocate kva for it.
223 	 * The map entry is, by default, pageable.
224 	 * XXX -- minor change needed here for NetBSD/OpenBSD VM systems.
225 	 */
226 	error = vm_map_find(kernel_map, cpipe->pipe_buffer.object, 0,
227 		(vm_offset_t *) &cpipe->pipe_buffer.buffer,
228 		cpipe->pipe_buffer.size, 1,
229 		VM_PROT_ALL, VM_PROT_ALL, 0);
230 
231 	if (error != KERN_SUCCESS)
232 		panic("pipeinit: cannot allocate pipe -- out of kvm -- code = %d", error);
233 	amountpipekva += cpipe->pipe_buffer.size;
234 }
235 
236 /*
237  * initialize and allocate VM and memory for pipe
238  */
239 static void
240 pipeinit(cpipe)
241 	struct pipe *cpipe;
242 {
243 
244 	cpipe->pipe_buffer.in = 0;
245 	cpipe->pipe_buffer.out = 0;
246 	cpipe->pipe_buffer.cnt = 0;
247 	cpipe->pipe_buffer.size = PIPE_SIZE;
248 
249 	/* Buffer kva gets dynamically allocated */
250 	cpipe->pipe_buffer.buffer = NULL;
251 	/* cpipe->pipe_buffer.object = invalid */
252 
253 	cpipe->pipe_state = 0;
254 	cpipe->pipe_peer = NULL;
255 	cpipe->pipe_busy = 0;
256 	getnanotime(&cpipe->pipe_ctime);
257 	cpipe->pipe_atime = cpipe->pipe_ctime;
258 	cpipe->pipe_mtime = cpipe->pipe_ctime;
259 	bzero(&cpipe->pipe_sel, sizeof cpipe->pipe_sel);
260 
261 #ifndef PIPE_NODIRECT
262 	/*
263 	 * pipe data structure initializations to support direct pipe I/O
264 	 */
265 	cpipe->pipe_map.cnt = 0;
266 	cpipe->pipe_map.kva = 0;
267 	cpipe->pipe_map.pos = 0;
268 	cpipe->pipe_map.npages = 0;
269 	/* cpipe->pipe_map.ms[] = invalid */
270 #endif
271 }
272 
273 
274 /*
275  * lock a pipe for I/O, blocking other access
276  */
277 static __inline int
278 pipelock(cpipe, catch)
279 	struct pipe *cpipe;
280 	int catch;
281 {
282 	int error;
283 	while (cpipe->pipe_state & PIPE_LOCK) {
284 		cpipe->pipe_state |= PIPE_LWANT;
285 		if ((error = tsleep( cpipe,
286 			catch?(PRIBIO|PCATCH):PRIBIO, "pipelk", 0)) != 0) {
287 			return error;
288 		}
289 	}
290 	cpipe->pipe_state |= PIPE_LOCK;
291 	return 0;
292 }
293 
294 /*
295  * unlock a pipe I/O lock
296  */
297 static __inline void
298 pipeunlock(cpipe)
299 	struct pipe *cpipe;
300 {
301 	cpipe->pipe_state &= ~PIPE_LOCK;
302 	if (cpipe->pipe_state & PIPE_LWANT) {
303 		cpipe->pipe_state &= ~PIPE_LWANT;
304 		wakeup(cpipe);
305 	}
306 }
307 
308 static __inline void
309 pipeselwakeup(cpipe)
310 	struct pipe *cpipe;
311 {
312 	if (cpipe->pipe_state & PIPE_SEL) {
313 		cpipe->pipe_state &= ~PIPE_SEL;
314 		selwakeup(&cpipe->pipe_sel);
315 	}
316 	if ((cpipe->pipe_state & PIPE_ASYNC) && cpipe->pipe_sigio)
317 		pgsigio(cpipe->pipe_sigio, SIGIO, 0);
318 }
319 
320 /* ARGSUSED */
321 static int
322 pipe_read(fp, uio, cred, flags)
323 	struct file *fp;
324 	struct uio *uio;
325 	struct ucred *cred;
326 	int flags;
327 {
328 
329 	struct pipe *rpipe = (struct pipe *) fp->f_data;
330 	int error = 0;
331 	int nread = 0;
332 	u_int size;
333 
334 	++rpipe->pipe_busy;
335 	while (uio->uio_resid) {
336 		/*
337 		 * normal pipe buffer receive
338 		 */
339 		if (rpipe->pipe_buffer.cnt > 0) {
340 			size = rpipe->pipe_buffer.size - rpipe->pipe_buffer.out;
341 			if (size > rpipe->pipe_buffer.cnt)
342 				size = rpipe->pipe_buffer.cnt;
343 			if (size > (u_int) uio->uio_resid)
344 				size = (u_int) uio->uio_resid;
345 			if ((error = pipelock(rpipe,1)) == 0) {
346 				error = uiomove( &rpipe->pipe_buffer.buffer[rpipe->pipe_buffer.out],
347 					size, uio);
348 				pipeunlock(rpipe);
349 			}
350 			if (error) {
351 				break;
352 			}
353 			rpipe->pipe_buffer.out += size;
354 			if (rpipe->pipe_buffer.out >= rpipe->pipe_buffer.size)
355 				rpipe->pipe_buffer.out = 0;
356 
357 			rpipe->pipe_buffer.cnt -= size;
358 			nread += size;
359 #ifndef PIPE_NODIRECT
360 		/*
361 		 * Direct copy, bypassing a kernel buffer.
362 		 */
363 		} else if ((size = rpipe->pipe_map.cnt) &&
364 			(rpipe->pipe_state & PIPE_DIRECTW)) {
365 			caddr_t va;
366 			if (size > (u_int) uio->uio_resid)
367 				size = (u_int) uio->uio_resid;
368 			if ((error = pipelock(rpipe,1)) == 0) {
369 				va = (caddr_t) rpipe->pipe_map.kva + rpipe->pipe_map.pos;
370 				error = uiomove(va, size, uio);
371 				pipeunlock(rpipe);
372 			}
373 			if (error)
374 				break;
375 			nread += size;
376 			rpipe->pipe_map.pos += size;
377 			rpipe->pipe_map.cnt -= size;
378 			if (rpipe->pipe_map.cnt == 0) {
379 				rpipe->pipe_state &= ~PIPE_DIRECTW;
380 				wakeup(rpipe);
381 			}
382 #endif
383 		} else {
384 			/*
385 			 * If there is no more to read in the pipe, reset
386 			 * its pointers to the beginning.  This improves
387 			 * cache hit stats.
388 			 *
389 			 * We get this over with now because it may block
390 			 * and cause the state to change out from under us,
391 			 * rather then have to re-test the state both before
392 			 * and after this fragment.
393 			 */
394 
395 			if ((error = pipelock(rpipe,1)) == 0) {
396 				if (rpipe->pipe_buffer.cnt == 0) {
397 					rpipe->pipe_buffer.in = 0;
398 					rpipe->pipe_buffer.out = 0;
399 				}
400 				pipeunlock(rpipe);
401 
402 				/*
403 				 * If pipe filled up due to pipelock
404 				 * blocking, loop back up.
405 				 */
406 				if (rpipe->pipe_buffer.cnt > 0)
407 					continue;
408 			}
409 
410 			/*
411 			 * detect EOF condition
412 			 */
413 			if (rpipe->pipe_state & PIPE_EOF) {
414 				/* XXX error = ? */
415 				break;
416 			}
417 
418 			/*
419 			 * If the "write-side" has been blocked, wake it up now.
420 			 */
421 			if (rpipe->pipe_state & PIPE_WANTW) {
422 				rpipe->pipe_state &= ~PIPE_WANTW;
423 				wakeup(rpipe);
424 			}
425 
426 			/*
427 			 * break if error (signal via pipelock), or if some
428 			 * data was read
429 			 */
430 			if (error || nread > 0)
431 				break;
432 
433 			/*
434 			 * Handle non-blocking mode operation
435 			 */
436 
437 			if (fp->f_flag & FNONBLOCK) {
438 				error = EAGAIN;
439 				break;
440 			}
441 
442 			/*
443 			 * Wait for more data
444 			 */
445 
446 			rpipe->pipe_state |= PIPE_WANTR;
447 			if ((error = tsleep(rpipe, PRIBIO|PCATCH, "piperd", 0)) != 0) {
448 				break;
449 			}
450 		}
451 	}
452 
453 	if (error == 0)
454 		getnanotime(&rpipe->pipe_atime);
455 
456 	--rpipe->pipe_busy;
457 	if ((rpipe->pipe_busy == 0) && (rpipe->pipe_state & PIPE_WANT)) {
458 		rpipe->pipe_state &= ~(PIPE_WANT|PIPE_WANTW);
459 		wakeup(rpipe);
460 	} else if (rpipe->pipe_buffer.cnt < MINPIPESIZE) {
461 		/*
462 		 * If there is no more to read in the pipe, reset
463 		 * its pointers to the beginning.  This improves
464 		 * cache hit stats.
465 		 */
466 		if (rpipe->pipe_buffer.cnt == 0) {
467 			if ((error == 0) && (error = pipelock(rpipe,1)) == 0) {
468 				rpipe->pipe_buffer.in = 0;
469 				rpipe->pipe_buffer.out = 0;
470 				pipeunlock(rpipe);
471 			}
472 		}
473 
474 		/*
475 		 * If the "write-side" has been blocked, wake it up now.
476 		 */
477 		if (rpipe->pipe_state & PIPE_WANTW) {
478 			rpipe->pipe_state &= ~PIPE_WANTW;
479 			wakeup(rpipe);
480 		}
481 	}
482 
483 	if ((rpipe->pipe_buffer.size - rpipe->pipe_buffer.cnt) >= PIPE_BUF)
484 		pipeselwakeup(rpipe);
485 
486 	return error;
487 }
488 
489 #ifndef PIPE_NODIRECT
490 /*
491  * Map the sending processes' buffer into kernel space and wire it.
492  * This is similar to a physical write operation.
493  */
494 static int
495 pipe_build_write_buffer(wpipe, uio)
496 	struct pipe *wpipe;
497 	struct uio *uio;
498 {
499 	u_int size;
500 	int i;
501 	vm_offset_t addr, endaddr, paddr;
502 
503 	size = (u_int) uio->uio_iov->iov_len;
504 	if (size > wpipe->pipe_buffer.size)
505 		size = wpipe->pipe_buffer.size;
506 
507 	endaddr = round_page((vm_offset_t)uio->uio_iov->iov_base + size);
508 	for(i = 0, addr = trunc_page((vm_offset_t)uio->uio_iov->iov_base);
509 		addr < endaddr;
510 		addr += PAGE_SIZE, i+=1) {
511 
512 		vm_page_t m;
513 
514 		vm_fault_quick( (caddr_t) addr, VM_PROT_READ);
515 		paddr = pmap_kextract(addr);
516 		if (!paddr) {
517 			int j;
518 			for(j=0;j<i;j++)
519 				vm_page_unwire(wpipe->pipe_map.ms[j], 1);
520 			return EFAULT;
521 		}
522 
523 		m = PHYS_TO_VM_PAGE(paddr);
524 		vm_page_wire(m);
525 		wpipe->pipe_map.ms[i] = m;
526 	}
527 
528 /*
529  * set up the control block
530  */
531 	wpipe->pipe_map.npages = i;
532 	wpipe->pipe_map.pos = ((vm_offset_t) uio->uio_iov->iov_base) & PAGE_MASK;
533 	wpipe->pipe_map.cnt = size;
534 
535 /*
536  * and map the buffer
537  */
538 	if (wpipe->pipe_map.kva == 0) {
539 		/*
540 		 * We need to allocate space for an extra page because the
541 		 * address range might (will) span pages at times.
542 		 */
543 		wpipe->pipe_map.kva = kmem_alloc_pageable(kernel_map,
544 			wpipe->pipe_buffer.size + PAGE_SIZE);
545 		amountpipekva += wpipe->pipe_buffer.size + PAGE_SIZE;
546 	}
547 	pmap_qenter(wpipe->pipe_map.kva, wpipe->pipe_map.ms,
548 		wpipe->pipe_map.npages);
549 
550 /*
551  * and update the uio data
552  */
553 
554 	uio->uio_iov->iov_len -= size;
555 	uio->uio_iov->iov_base += size;
556 	if (uio->uio_iov->iov_len == 0)
557 		uio->uio_iov++;
558 	uio->uio_resid -= size;
559 	uio->uio_offset += size;
560 	return 0;
561 }
562 
563 /*
564  * unmap and unwire the process buffer
565  */
566 static void
567 pipe_destroy_write_buffer(wpipe)
568 struct pipe *wpipe;
569 {
570 	int i;
571 	if (wpipe->pipe_map.kva) {
572 		pmap_qremove(wpipe->pipe_map.kva, wpipe->pipe_map.npages);
573 
574 		if (amountpipekva > MAXPIPEKVA) {
575 			vm_offset_t kva = wpipe->pipe_map.kva;
576 			wpipe->pipe_map.kva = 0;
577 			kmem_free(kernel_map, kva,
578 				wpipe->pipe_buffer.size + PAGE_SIZE);
579 			amountpipekva -= wpipe->pipe_buffer.size + PAGE_SIZE;
580 		}
581 	}
582 	for (i=0;i<wpipe->pipe_map.npages;i++)
583 		vm_page_unwire(wpipe->pipe_map.ms[i], 1);
584 }
585 
586 /*
587  * In the case of a signal, the writing process might go away.  This
588  * code copies the data into the circular buffer so that the source
589  * pages can be freed without loss of data.
590  */
591 static void
592 pipe_clone_write_buffer(wpipe)
593 struct pipe *wpipe;
594 {
595 	int size;
596 	int pos;
597 
598 	size = wpipe->pipe_map.cnt;
599 	pos = wpipe->pipe_map.pos;
600 	bcopy((caddr_t) wpipe->pipe_map.kva+pos,
601 			(caddr_t) wpipe->pipe_buffer.buffer,
602 			size);
603 
604 	wpipe->pipe_buffer.in = size;
605 	wpipe->pipe_buffer.out = 0;
606 	wpipe->pipe_buffer.cnt = size;
607 	wpipe->pipe_state &= ~PIPE_DIRECTW;
608 
609 	pipe_destroy_write_buffer(wpipe);
610 }
611 
612 /*
613  * This implements the pipe buffer write mechanism.  Note that only
614  * a direct write OR a normal pipe write can be pending at any given time.
615  * If there are any characters in the pipe buffer, the direct write will
616  * be deferred until the receiving process grabs all of the bytes from
617  * the pipe buffer.  Then the direct mapping write is set-up.
618  */
619 static int
620 pipe_direct_write(wpipe, uio)
621 	struct pipe *wpipe;
622 	struct uio *uio;
623 {
624 	int error;
625 retry:
626 	while (wpipe->pipe_state & PIPE_DIRECTW) {
627 		if ( wpipe->pipe_state & PIPE_WANTR) {
628 			wpipe->pipe_state &= ~PIPE_WANTR;
629 			wakeup(wpipe);
630 		}
631 		wpipe->pipe_state |= PIPE_WANTW;
632 		error = tsleep(wpipe,
633 				PRIBIO|PCATCH, "pipdww", 0);
634 		if (error)
635 			goto error1;
636 		if (wpipe->pipe_state & PIPE_EOF) {
637 			error = EPIPE;
638 			goto error1;
639 		}
640 	}
641 	wpipe->pipe_map.cnt = 0;	/* transfer not ready yet */
642 	if (wpipe->pipe_buffer.cnt > 0) {
643 		if ( wpipe->pipe_state & PIPE_WANTR) {
644 			wpipe->pipe_state &= ~PIPE_WANTR;
645 			wakeup(wpipe);
646 		}
647 
648 		wpipe->pipe_state |= PIPE_WANTW;
649 		error = tsleep(wpipe,
650 				PRIBIO|PCATCH, "pipdwc", 0);
651 		if (error)
652 			goto error1;
653 		if (wpipe->pipe_state & PIPE_EOF) {
654 			error = EPIPE;
655 			goto error1;
656 		}
657 		goto retry;
658 	}
659 
660 	wpipe->pipe_state |= PIPE_DIRECTW;
661 
662 	error = pipe_build_write_buffer(wpipe, uio);
663 	if (error) {
664 		wpipe->pipe_state &= ~PIPE_DIRECTW;
665 		goto error1;
666 	}
667 
668 	error = 0;
669 	while (!error && (wpipe->pipe_state & PIPE_DIRECTW)) {
670 		if (wpipe->pipe_state & PIPE_EOF) {
671 			pipelock(wpipe, 0);
672 			pipe_destroy_write_buffer(wpipe);
673 			pipeunlock(wpipe);
674 			pipeselwakeup(wpipe);
675 			error = EPIPE;
676 			goto error1;
677 		}
678 		if (wpipe->pipe_state & PIPE_WANTR) {
679 			wpipe->pipe_state &= ~PIPE_WANTR;
680 			wakeup(wpipe);
681 		}
682 		pipeselwakeup(wpipe);
683 		error = tsleep(wpipe, PRIBIO|PCATCH, "pipdwt", 0);
684 	}
685 
686 	pipelock(wpipe,0);
687 	if (wpipe->pipe_state & PIPE_DIRECTW) {
688 		/*
689 		 * this bit of trickery substitutes a kernel buffer for
690 		 * the process that might be going away.
691 		 */
692 		pipe_clone_write_buffer(wpipe);
693 	} else {
694 		pipe_destroy_write_buffer(wpipe);
695 	}
696 	pipeunlock(wpipe);
697 	return error;
698 
699 error1:
700 	wakeup(wpipe);
701 	return error;
702 }
703 #endif
704 
705 static int
706 pipe_write(fp, uio, cred, flags)
707 	struct file *fp;
708 	struct uio *uio;
709 	struct ucred *cred;
710 	int flags;
711 {
712 	int error = 0;
713 	int orig_resid;
714 
715 	struct pipe *wpipe, *rpipe;
716 
717 	rpipe = (struct pipe *) fp->f_data;
718 	wpipe = rpipe->pipe_peer;
719 
720 	/*
721 	 * detect loss of pipe read side, issue SIGPIPE if lost.
722 	 */
723 	if ((wpipe == NULL) || (wpipe->pipe_state & PIPE_EOF)) {
724 		return EPIPE;
725 	}
726 
727 	/*
728 	 * If it is advantageous to resize the pipe buffer, do
729 	 * so.
730 	 */
731 	if ((uio->uio_resid > PIPE_SIZE) &&
732 		(nbigpipe < LIMITBIGPIPES) &&
733 		(wpipe->pipe_state & PIPE_DIRECTW) == 0 &&
734 		(wpipe->pipe_buffer.size <= PIPE_SIZE) &&
735 		(wpipe->pipe_buffer.cnt == 0)) {
736 
737 		if (wpipe->pipe_buffer.buffer) {
738 			amountpipekva -= wpipe->pipe_buffer.size;
739 			kmem_free(kernel_map,
740 				(vm_offset_t)wpipe->pipe_buffer.buffer,
741 				wpipe->pipe_buffer.size);
742 		}
743 
744 #ifndef PIPE_NODIRECT
745 		if (wpipe->pipe_map.kva) {
746 			amountpipekva -= wpipe->pipe_buffer.size + PAGE_SIZE;
747 			kmem_free(kernel_map,
748 				wpipe->pipe_map.kva,
749 				wpipe->pipe_buffer.size + PAGE_SIZE);
750 		}
751 #endif
752 
753 		wpipe->pipe_buffer.in = 0;
754 		wpipe->pipe_buffer.out = 0;
755 		wpipe->pipe_buffer.cnt = 0;
756 		wpipe->pipe_buffer.size = BIG_PIPE_SIZE;
757 		wpipe->pipe_buffer.buffer = NULL;
758 		++nbigpipe;
759 
760 #ifndef PIPE_NODIRECT
761 		wpipe->pipe_map.cnt = 0;
762 		wpipe->pipe_map.kva = 0;
763 		wpipe->pipe_map.pos = 0;
764 		wpipe->pipe_map.npages = 0;
765 #endif
766 
767 	}
768 
769 
770 	if( wpipe->pipe_buffer.buffer == NULL) {
771 		if ((error = pipelock(wpipe,1)) == 0) {
772 			pipespace(wpipe);
773 			pipeunlock(wpipe);
774 		} else {
775 			return error;
776 		}
777 	}
778 
779 	++wpipe->pipe_busy;
780 	orig_resid = uio->uio_resid;
781 	while (uio->uio_resid) {
782 		int space;
783 #ifndef PIPE_NODIRECT
784 		/*
785 		 * If the transfer is large, we can gain performance if
786 		 * we do process-to-process copies directly.
787 		 * If the write is non-blocking, we don't use the
788 		 * direct write mechanism.
789 		 */
790 		if ((uio->uio_iov->iov_len >= PIPE_MINDIRECT) &&
791 		    (fp->f_flag & FNONBLOCK) == 0 &&
792 			(wpipe->pipe_map.kva || (amountpipekva < LIMITPIPEKVA)) &&
793 			(uio->uio_iov->iov_len >= PIPE_MINDIRECT)) {
794 			error = pipe_direct_write( wpipe, uio);
795 			if (error) {
796 				break;
797 			}
798 			continue;
799 		}
800 #endif
801 
802 		/*
803 		 * Pipe buffered writes cannot be coincidental with
804 		 * direct writes.  We wait until the currently executing
805 		 * direct write is completed before we start filling the
806 		 * pipe buffer.
807 		 */
808 	retrywrite:
809 		while (wpipe->pipe_state & PIPE_DIRECTW) {
810 			if (wpipe->pipe_state & PIPE_WANTR) {
811 				wpipe->pipe_state &= ~PIPE_WANTR;
812 				wakeup(wpipe);
813 			}
814 			error = tsleep(wpipe,
815 					PRIBIO|PCATCH, "pipbww", 0);
816 			if (error)
817 				break;
818 		}
819 
820 		space = wpipe->pipe_buffer.size - wpipe->pipe_buffer.cnt;
821 
822 		/* Writes of size <= PIPE_BUF must be atomic. */
823 		/* XXX perhaps they need to be contiguous to be atomic? */
824 		if ((space < uio->uio_resid) && (orig_resid <= PIPE_BUF))
825 			space = 0;
826 
827 		if (space > 0 && (wpipe->pipe_buffer.cnt < PIPE_SIZE)) {
828 			/*
829 			 * This set the maximum transfer as a segment of
830 			 * the buffer.
831 			 */
832 			int size = wpipe->pipe_buffer.size - wpipe->pipe_buffer.in;
833 			/*
834 			 * space is the size left in the buffer
835 			 */
836 			if (size > space)
837 				size = space;
838 			/*
839 			 * now limit it to the size of the uio transfer
840 			 */
841 			if (size > uio->uio_resid)
842 				size = uio->uio_resid;
843 			if ((error = pipelock(wpipe,1)) == 0) {
844 				/*
845 				 * It is possible for a direct write to
846 				 * slip in on us... handle it here...
847 				 */
848 				if (wpipe->pipe_state & PIPE_DIRECTW) {
849 					pipeunlock(wpipe);
850 					goto retrywrite;
851 				}
852 				error = uiomove( &wpipe->pipe_buffer.buffer[wpipe->pipe_buffer.in],
853 					size, uio);
854 				pipeunlock(wpipe);
855 			}
856 			if (error)
857 				break;
858 
859 			wpipe->pipe_buffer.in += size;
860 			if (wpipe->pipe_buffer.in >= wpipe->pipe_buffer.size)
861 				wpipe->pipe_buffer.in = 0;
862 
863 			wpipe->pipe_buffer.cnt += size;
864 		} else {
865 			/*
866 			 * If the "read-side" has been blocked, wake it up now.
867 			 */
868 			if (wpipe->pipe_state & PIPE_WANTR) {
869 				wpipe->pipe_state &= ~PIPE_WANTR;
870 				wakeup(wpipe);
871 			}
872 
873 			/*
874 			 * don't block on non-blocking I/O
875 			 */
876 			if (fp->f_flag & FNONBLOCK) {
877 				error = EAGAIN;
878 				break;
879 			}
880 
881 			/*
882 			 * We have no more space and have something to offer,
883 			 * wake up select/poll.
884 			 */
885 			pipeselwakeup(wpipe);
886 
887 			wpipe->pipe_state |= PIPE_WANTW;
888 			if ((error = tsleep(wpipe, (PRIBIO+1)|PCATCH, "pipewr", 0)) != 0) {
889 				break;
890 			}
891 			/*
892 			 * If read side wants to go away, we just issue a signal
893 			 * to ourselves.
894 			 */
895 			if (wpipe->pipe_state & PIPE_EOF) {
896 				error = EPIPE;
897 				break;
898 			}
899 		}
900 	}
901 
902 	--wpipe->pipe_busy;
903 	if ((wpipe->pipe_busy == 0) &&
904 		(wpipe->pipe_state & PIPE_WANT)) {
905 		wpipe->pipe_state &= ~(PIPE_WANT|PIPE_WANTR);
906 		wakeup(wpipe);
907 	} else if (wpipe->pipe_buffer.cnt > 0) {
908 		/*
909 		 * If we have put any characters in the buffer, we wake up
910 		 * the reader.
911 		 */
912 		if (wpipe->pipe_state & PIPE_WANTR) {
913 			wpipe->pipe_state &= ~PIPE_WANTR;
914 			wakeup(wpipe);
915 		}
916 	}
917 
918 	/*
919 	 * Don't return EPIPE if I/O was successful
920 	 */
921 	if ((wpipe->pipe_buffer.cnt == 0) &&
922 		(uio->uio_resid == 0) &&
923 		(error == EPIPE))
924 		error = 0;
925 
926 	if (error == 0)
927 		getnanotime(&wpipe->pipe_mtime);
928 
929 	/*
930 	 * We have something to offer,
931 	 * wake up select/poll.
932 	 */
933 	if (wpipe->pipe_buffer.cnt)
934 		pipeselwakeup(wpipe);
935 
936 	return error;
937 }
938 
939 /*
940  * we implement a very minimal set of ioctls for compatibility with sockets.
941  */
942 int
943 pipe_ioctl(fp, cmd, data, p)
944 	struct file *fp;
945 	u_long cmd;
946 	register caddr_t data;
947 	struct proc *p;
948 {
949 	register struct pipe *mpipe = (struct pipe *)fp->f_data;
950 
951 	switch (cmd) {
952 
953 	case FIONBIO:
954 		return (0);
955 
956 	case FIOASYNC:
957 		if (*(int *)data) {
958 			mpipe->pipe_state |= PIPE_ASYNC;
959 		} else {
960 			mpipe->pipe_state &= ~PIPE_ASYNC;
961 		}
962 		return (0);
963 
964 	case FIONREAD:
965 		if (mpipe->pipe_state & PIPE_DIRECTW)
966 			*(int *)data = mpipe->pipe_map.cnt;
967 		else
968 			*(int *)data = mpipe->pipe_buffer.cnt;
969 		return (0);
970 
971 	case FIOSETOWN:
972 		return (fsetown(*(int *)data, &mpipe->pipe_sigio));
973 
974 	case FIOGETOWN:
975 		*(int *)data = fgetown(mpipe->pipe_sigio);
976 		return (0);
977 
978 	/* This is deprecated, FIOSETOWN should be used instead. */
979 	case TIOCSPGRP:
980 		return (fsetown(-(*(int *)data), &mpipe->pipe_sigio));
981 
982 	/* This is deprecated, FIOGETOWN should be used instead. */
983 	case TIOCGPGRP:
984 		*(int *)data = -fgetown(mpipe->pipe_sigio);
985 		return (0);
986 
987 	}
988 	return (ENOTTY);
989 }
990 
991 int
992 pipe_poll(fp, events, cred, p)
993 	struct file *fp;
994 	int events;
995 	struct ucred *cred;
996 	struct proc *p;
997 {
998 	register struct pipe *rpipe = (struct pipe *)fp->f_data;
999 	struct pipe *wpipe;
1000 	int revents = 0;
1001 
1002 	wpipe = rpipe->pipe_peer;
1003 	if (events & (POLLIN | POLLRDNORM))
1004 		if ((rpipe->pipe_state & PIPE_DIRECTW) ||
1005 		    (rpipe->pipe_buffer.cnt > 0) ||
1006 		    (rpipe->pipe_state & PIPE_EOF))
1007 			revents |= events & (POLLIN | POLLRDNORM);
1008 
1009 	if (events & (POLLOUT | POLLWRNORM))
1010 		if (wpipe == NULL || (wpipe->pipe_state & PIPE_EOF) ||
1011 		    (((wpipe->pipe_state & PIPE_DIRECTW) == 0) &&
1012 		     (wpipe->pipe_buffer.size - wpipe->pipe_buffer.cnt) >= PIPE_BUF))
1013 			revents |= events & (POLLOUT | POLLWRNORM);
1014 
1015 	if ((rpipe->pipe_state & PIPE_EOF) ||
1016 	    (wpipe == NULL) ||
1017 	    (wpipe->pipe_state & PIPE_EOF))
1018 		revents |= POLLHUP;
1019 
1020 	if (revents == 0) {
1021 		if (events & (POLLIN | POLLRDNORM)) {
1022 			selrecord(p, &rpipe->pipe_sel);
1023 			rpipe->pipe_state |= PIPE_SEL;
1024 		}
1025 
1026 		if (events & (POLLOUT | POLLWRNORM)) {
1027 			selrecord(p, &wpipe->pipe_sel);
1028 			wpipe->pipe_state |= PIPE_SEL;
1029 		}
1030 	}
1031 
1032 	return (revents);
1033 }
1034 
1035 int
1036 pipe_stat(pipe, ub)
1037 	register struct pipe *pipe;
1038 	register struct stat *ub;
1039 {
1040 	bzero((caddr_t)ub, sizeof (*ub));
1041 	ub->st_mode = S_IFIFO;
1042 	ub->st_blksize = pipe->pipe_buffer.size;
1043 	ub->st_size = pipe->pipe_buffer.cnt;
1044 	ub->st_blocks = (ub->st_size + ub->st_blksize - 1) / ub->st_blksize;
1045 	ub->st_atimespec = pipe->pipe_atime;
1046 	ub->st_mtimespec = pipe->pipe_mtime;
1047 	ub->st_ctimespec = pipe->pipe_ctime;
1048 	/*
1049 	 * Left as 0: st_dev, st_ino, st_nlink, st_uid, st_gid, st_rdev,
1050 	 * st_flags, st_gen.
1051 	 * XXX (st_dev, st_ino) should be unique.
1052 	 */
1053 	return 0;
1054 }
1055 
1056 /* ARGSUSED */
1057 static int
1058 pipe_close(fp, p)
1059 	struct file *fp;
1060 	struct proc *p;
1061 {
1062 	struct pipe *cpipe = (struct pipe *)fp->f_data;
1063 
1064 	funsetown(cpipe->pipe_sigio);
1065 	pipeclose(cpipe);
1066 	fp->f_data = NULL;
1067 	return 0;
1068 }
1069 
1070 /*
1071  * shutdown the pipe
1072  */
1073 static void
1074 pipeclose(cpipe)
1075 	struct pipe *cpipe;
1076 {
1077 	struct pipe *ppipe;
1078 	if (cpipe) {
1079 
1080 		pipeselwakeup(cpipe);
1081 
1082 		/*
1083 		 * If the other side is blocked, wake it up saying that
1084 		 * we want to close it down.
1085 		 */
1086 		while (cpipe->pipe_busy) {
1087 			wakeup(cpipe);
1088 			cpipe->pipe_state |= PIPE_WANT|PIPE_EOF;
1089 			tsleep(cpipe, PRIBIO, "pipecl", 0);
1090 		}
1091 
1092 		/*
1093 		 * Disconnect from peer
1094 		 */
1095 		if ((ppipe = cpipe->pipe_peer) != NULL) {
1096 			pipeselwakeup(ppipe);
1097 
1098 			ppipe->pipe_state |= PIPE_EOF;
1099 			wakeup(ppipe);
1100 			ppipe->pipe_peer = NULL;
1101 		}
1102 
1103 		/*
1104 		 * free resources
1105 		 */
1106 		if (cpipe->pipe_buffer.buffer) {
1107 			if (cpipe->pipe_buffer.size > PIPE_SIZE)
1108 				--nbigpipe;
1109 			amountpipekva -= cpipe->pipe_buffer.size;
1110 			kmem_free(kernel_map,
1111 				(vm_offset_t)cpipe->pipe_buffer.buffer,
1112 				cpipe->pipe_buffer.size);
1113 		}
1114 #ifndef PIPE_NODIRECT
1115 		if (cpipe->pipe_map.kva) {
1116 			amountpipekva -= cpipe->pipe_buffer.size + PAGE_SIZE;
1117 			kmem_free(kernel_map,
1118 				cpipe->pipe_map.kva,
1119 				cpipe->pipe_buffer.size + PAGE_SIZE);
1120 		}
1121 #endif
1122 		zfree(pipe_zone, cpipe);
1123 	}
1124 }
1125