sys_pipe.c revision 51474
1275988Sngie/*
2240116Smarcel * Copyright (c) 1996 John S. Dyson
3240116Smarcel * All rights reserved.
4240116Smarcel *
5240116Smarcel * Redistribution and use in source and binary forms, with or without
6240116Smarcel * modification, are permitted provided that the following conditions
7240116Smarcel * are met:
8240116Smarcel * 1. Redistributions of source code must retain the above copyright
9240116Smarcel *    notice immediately at the beginning of the file, without modification,
10240116Smarcel *    this list of conditions, and the following disclaimer.
11240116Smarcel * 2. Redistributions in binary form must reproduce the above copyright
12240116Smarcel *    notice, this list of conditions and the following disclaimer in the
13240116Smarcel *    documentation and/or other materials provided with the distribution.
14240116Smarcel * 3. Absolutely no warranty of function or purpose is made by the author
15240116Smarcel *    John S. Dyson.
16240116Smarcel * 4. Modifications may be freely made to this file if the above conditions
17240116Smarcel *    are met.
18240116Smarcel *
19240116Smarcel * $FreeBSD: head/sys/kern/sys_pipe.c 51474 1999-09-20 19:08:48Z dillon $
20240116Smarcel */
21240116Smarcel
22240116Smarcel/*
23240116Smarcel * This file contains a high-performance replacement for the socket-based
24275988Sngie * pipes scheme originally used in FreeBSD/4.4Lite.  It does not support
25240116Smarcel * all features of sockets, but does do everything that pipes normally
26240116Smarcel * do.
27240116Smarcel */
28240116Smarcel
29240116Smarcel/*
30240116Smarcel * This code has two modes of operation, a small write mode and a large
31260029Sjmmv * write mode.  The small write mode acts like conventional pipes with
32240116Smarcel * a kernel buffer.  If the buffer is less than PIPE_MINDIRECT, then the
33240116Smarcel * "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 */
90static int pipe_read __P((struct file *fp, struct uio *uio,
91		struct ucred *cred, int flags, struct proc *p));
92static int pipe_write __P((struct file *fp, struct uio *uio,
93		struct ucred *cred, int flags, struct proc *p));
94static int pipe_close __P((struct file *fp, struct proc *p));
95static int pipe_poll __P((struct file *fp, int events, struct ucred *cred,
96		struct proc *p));
97static int pipe_ioctl __P((struct file *fp, u_long cmd, caddr_t data, struct proc *p));
98
99static 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
127static int nbigpipe;
128
129static int amountpipekva;
130
131static void pipeclose __P((struct pipe *cpipe));
132static void pipeinit __P((struct pipe *cpipe));
133static __inline int pipelock __P((struct pipe *cpipe, int catch));
134static __inline void pipeunlock __P((struct pipe *cpipe));
135static __inline void pipeselwakeup __P((struct pipe *cpipe));
136#ifndef PIPE_NODIRECT
137static int pipe_build_write_buffer __P((struct pipe *wpipe, struct uio *uio));
138static void pipe_destroy_write_buffer __P((struct pipe *wpipe));
139static int pipe_direct_write __P((struct pipe *wpipe, struct uio *uio));
140static void pipe_clone_write_buffer __P((struct pipe *wpipe));
141#endif
142static void pipespace __P((struct pipe *cpipe));
143
144static vm_zone_t pipe_zone;
145
146/*
147 * The pipe system call for the DTYPE_PIPE type of pipes
148 */
149
150/* ARGSUSED */
151int
152pipe(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_data = (caddr_t)rpipe;
180	rf->f_ops = &pipeops;
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_data = (caddr_t)wpipe;
187	wf->f_ops = &pipeops;
188	p->p_retval[1] = fd;
189
190	rpipe->pipe_peer = wpipe;
191	wpipe->pipe_peer = rpipe;
192
193	return (0);
194free3:
195	fdp->fd_ofiles[p->p_retval[0]] = 0;
196	ffree(rf);
197free2:
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 */
206static void
207pipespace(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 */
239static void
240pipeinit(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 */
277static __inline int
278pipelock(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 */
297static __inline void
298pipeunlock(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
308static __inline void
309pipeselwakeup(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 */
321static int
322pipe_read(fp, uio, cred, flags, p)
323	struct file *fp;
324	struct uio *uio;
325	struct ucred *cred;
326	struct proc *p;
327	int flags;
328{
329
330	struct pipe *rpipe = (struct pipe *) fp->f_data;
331	int error;
332	int nread = 0;
333	u_int size;
334
335	++rpipe->pipe_busy;
336	error = pipelock(rpipe, 1);
337	if (error)
338		goto unlocked_error;
339
340	while (uio->uio_resid) {
341		/*
342		 * normal pipe buffer receive
343		 */
344		if (rpipe->pipe_buffer.cnt > 0) {
345			size = rpipe->pipe_buffer.size - rpipe->pipe_buffer.out;
346			if (size > rpipe->pipe_buffer.cnt)
347				size = rpipe->pipe_buffer.cnt;
348			if (size > (u_int) uio->uio_resid)
349				size = (u_int) uio->uio_resid;
350
351			error = uiomove(&rpipe->pipe_buffer.buffer[rpipe->pipe_buffer.out],
352					size, uio);
353			if (error) {
354				break;
355			}
356			rpipe->pipe_buffer.out += size;
357			if (rpipe->pipe_buffer.out >= rpipe->pipe_buffer.size)
358				rpipe->pipe_buffer.out = 0;
359
360			rpipe->pipe_buffer.cnt -= size;
361
362			/*
363			 * If there is no more to read in the pipe, reset
364			 * its pointers to the beginning.  This improves
365			 * cache hit stats.
366			 */
367			if (rpipe->pipe_buffer.cnt == 0) {
368				rpipe->pipe_buffer.in = 0;
369				rpipe->pipe_buffer.out = 0;
370			}
371			nread += size;
372#ifndef PIPE_NODIRECT
373		/*
374		 * Direct copy, bypassing a kernel buffer.
375		 */
376		} else if ((size = rpipe->pipe_map.cnt) &&
377			   (rpipe->pipe_state & PIPE_DIRECTW)) {
378			caddr_t	va;
379			if (size > (u_int) uio->uio_resid)
380				size = (u_int) uio->uio_resid;
381
382			va = (caddr_t) rpipe->pipe_map.kva + rpipe->pipe_map.pos;
383			error = uiomove(va, size, uio);
384			if (error)
385				break;
386			nread += size;
387			rpipe->pipe_map.pos += size;
388			rpipe->pipe_map.cnt -= size;
389			if (rpipe->pipe_map.cnt == 0) {
390				rpipe->pipe_state &= ~PIPE_DIRECTW;
391				wakeup(rpipe);
392			}
393#endif
394		} else {
395			/*
396			 * detect EOF condition
397			 */
398			if (rpipe->pipe_state & PIPE_EOF) {
399				/* XXX error = ? */
400				break;
401			}
402
403			/*
404			 * If the "write-side" has been blocked, wake it up now.
405			 */
406			if (rpipe->pipe_state & PIPE_WANTW) {
407				rpipe->pipe_state &= ~PIPE_WANTW;
408				wakeup(rpipe);
409			}
410
411			/*
412			 * Break if some data was read.
413			 */
414			if (nread > 0)
415				break;
416
417			/*
418			 * Unlock the pipe buffer for our remaining processing.  We
419			 * will either break out with an error or we will sleep and
420			 * relock to loop.
421			 */
422			pipeunlock(rpipe);
423
424			/*
425			 * Handle non-blocking mode operation or
426			 * wait for more data.
427			 */
428			if (fp->f_flag & FNONBLOCK)
429				error = EAGAIN;
430			else {
431				rpipe->pipe_state |= PIPE_WANTR;
432				if ((error = tsleep(rpipe, PRIBIO|PCATCH, "piperd", 0)) == 0)
433					error = pipelock(rpipe, 1);
434			}
435			if (error)
436				goto unlocked_error;
437		}
438	}
439	pipeunlock(rpipe);
440
441	if (error == 0)
442		getnanotime(&rpipe->pipe_atime);
443unlocked_error:
444	--rpipe->pipe_busy;
445
446	/*
447	 * PIPE_WANT processing only makes sense if pipe_busy is 0.
448	 */
449	if ((rpipe->pipe_busy == 0) && (rpipe->pipe_state & PIPE_WANT)) {
450		rpipe->pipe_state &= ~(PIPE_WANT|PIPE_WANTW);
451		wakeup(rpipe);
452	} else if (rpipe->pipe_buffer.cnt < MINPIPESIZE) {
453		/*
454		 * Handle write blocking hysteresis.
455		 */
456		if (rpipe->pipe_state & PIPE_WANTW) {
457			rpipe->pipe_state &= ~PIPE_WANTW;
458			wakeup(rpipe);
459		}
460	}
461
462	if ((rpipe->pipe_buffer.size - rpipe->pipe_buffer.cnt) >= PIPE_BUF)
463		pipeselwakeup(rpipe);
464
465	return error;
466}
467
468#ifndef PIPE_NODIRECT
469/*
470 * Map the sending processes' buffer into kernel space and wire it.
471 * This is similar to a physical write operation.
472 */
473static int
474pipe_build_write_buffer(wpipe, uio)
475	struct pipe *wpipe;
476	struct uio *uio;
477{
478	u_int size;
479	int i;
480	vm_offset_t addr, endaddr, paddr;
481
482	size = (u_int) uio->uio_iov->iov_len;
483	if (size > wpipe->pipe_buffer.size)
484		size = wpipe->pipe_buffer.size;
485
486	endaddr = round_page((vm_offset_t)uio->uio_iov->iov_base + size);
487	for(i = 0, addr = trunc_page((vm_offset_t)uio->uio_iov->iov_base);
488		addr < endaddr;
489		addr += PAGE_SIZE, i+=1) {
490
491		vm_page_t m;
492
493		if (vm_fault_quick((caddr_t)addr, VM_PROT_READ) < 0 ||
494		    (paddr = pmap_kextract(addr)) == 0) {
495			int j;
496			for(j=0;j<i;j++)
497				vm_page_unwire(wpipe->pipe_map.ms[j], 1);
498			return EFAULT;
499		}
500
501		m = PHYS_TO_VM_PAGE(paddr);
502		vm_page_wire(m);
503		wpipe->pipe_map.ms[i] = m;
504	}
505
506/*
507 * set up the control block
508 */
509	wpipe->pipe_map.npages = i;
510	wpipe->pipe_map.pos = ((vm_offset_t) uio->uio_iov->iov_base) & PAGE_MASK;
511	wpipe->pipe_map.cnt = size;
512
513/*
514 * and map the buffer
515 */
516	if (wpipe->pipe_map.kva == 0) {
517		/*
518		 * We need to allocate space for an extra page because the
519		 * address range might (will) span pages at times.
520		 */
521		wpipe->pipe_map.kva = kmem_alloc_pageable(kernel_map,
522			wpipe->pipe_buffer.size + PAGE_SIZE);
523		amountpipekva += wpipe->pipe_buffer.size + PAGE_SIZE;
524	}
525	pmap_qenter(wpipe->pipe_map.kva, wpipe->pipe_map.ms,
526		wpipe->pipe_map.npages);
527
528/*
529 * and update the uio data
530 */
531
532	uio->uio_iov->iov_len -= size;
533	uio->uio_iov->iov_base += size;
534	if (uio->uio_iov->iov_len == 0)
535		uio->uio_iov++;
536	uio->uio_resid -= size;
537	uio->uio_offset += size;
538	return 0;
539}
540
541/*
542 * unmap and unwire the process buffer
543 */
544static void
545pipe_destroy_write_buffer(wpipe)
546struct pipe *wpipe;
547{
548	int i;
549	if (wpipe->pipe_map.kva) {
550		pmap_qremove(wpipe->pipe_map.kva, wpipe->pipe_map.npages);
551
552		if (amountpipekva > MAXPIPEKVA) {
553			vm_offset_t kva = wpipe->pipe_map.kva;
554			wpipe->pipe_map.kva = 0;
555			kmem_free(kernel_map, kva,
556				wpipe->pipe_buffer.size + PAGE_SIZE);
557			amountpipekva -= wpipe->pipe_buffer.size + PAGE_SIZE;
558		}
559	}
560	for (i=0;i<wpipe->pipe_map.npages;i++)
561		vm_page_unwire(wpipe->pipe_map.ms[i], 1);
562}
563
564/*
565 * In the case of a signal, the writing process might go away.  This
566 * code copies the data into the circular buffer so that the source
567 * pages can be freed without loss of data.
568 */
569static void
570pipe_clone_write_buffer(wpipe)
571struct pipe *wpipe;
572{
573	int size;
574	int pos;
575
576	size = wpipe->pipe_map.cnt;
577	pos = wpipe->pipe_map.pos;
578	bcopy((caddr_t) wpipe->pipe_map.kva+pos,
579			(caddr_t) wpipe->pipe_buffer.buffer,
580			size);
581
582	wpipe->pipe_buffer.in = size;
583	wpipe->pipe_buffer.out = 0;
584	wpipe->pipe_buffer.cnt = size;
585	wpipe->pipe_state &= ~PIPE_DIRECTW;
586
587	pipe_destroy_write_buffer(wpipe);
588}
589
590/*
591 * This implements the pipe buffer write mechanism.  Note that only
592 * a direct write OR a normal pipe write can be pending at any given time.
593 * If there are any characters in the pipe buffer, the direct write will
594 * be deferred until the receiving process grabs all of the bytes from
595 * the pipe buffer.  Then the direct mapping write is set-up.
596 */
597static int
598pipe_direct_write(wpipe, uio)
599	struct pipe *wpipe;
600	struct uio *uio;
601{
602	int error;
603retry:
604	while (wpipe->pipe_state & PIPE_DIRECTW) {
605		if ( wpipe->pipe_state & PIPE_WANTR) {
606			wpipe->pipe_state &= ~PIPE_WANTR;
607			wakeup(wpipe);
608		}
609		wpipe->pipe_state |= PIPE_WANTW;
610		error = tsleep(wpipe,
611				PRIBIO|PCATCH, "pipdww", 0);
612		if (error)
613			goto error1;
614		if (wpipe->pipe_state & PIPE_EOF) {
615			error = EPIPE;
616			goto error1;
617		}
618	}
619	wpipe->pipe_map.cnt = 0;	/* transfer not ready yet */
620	if (wpipe->pipe_buffer.cnt > 0) {
621		if ( wpipe->pipe_state & PIPE_WANTR) {
622			wpipe->pipe_state &= ~PIPE_WANTR;
623			wakeup(wpipe);
624		}
625
626		wpipe->pipe_state |= PIPE_WANTW;
627		error = tsleep(wpipe,
628				PRIBIO|PCATCH, "pipdwc", 0);
629		if (error)
630			goto error1;
631		if (wpipe->pipe_state & PIPE_EOF) {
632			error = EPIPE;
633			goto error1;
634		}
635		goto retry;
636	}
637
638	wpipe->pipe_state |= PIPE_DIRECTW;
639
640	error = pipe_build_write_buffer(wpipe, uio);
641	if (error) {
642		wpipe->pipe_state &= ~PIPE_DIRECTW;
643		goto error1;
644	}
645
646	error = 0;
647	while (!error && (wpipe->pipe_state & PIPE_DIRECTW)) {
648		if (wpipe->pipe_state & PIPE_EOF) {
649			pipelock(wpipe, 0);
650			pipe_destroy_write_buffer(wpipe);
651			pipeunlock(wpipe);
652			pipeselwakeup(wpipe);
653			error = EPIPE;
654			goto error1;
655		}
656		if (wpipe->pipe_state & PIPE_WANTR) {
657			wpipe->pipe_state &= ~PIPE_WANTR;
658			wakeup(wpipe);
659		}
660		pipeselwakeup(wpipe);
661		error = tsleep(wpipe, PRIBIO|PCATCH, "pipdwt", 0);
662	}
663
664	pipelock(wpipe,0);
665	if (wpipe->pipe_state & PIPE_DIRECTW) {
666		/*
667		 * this bit of trickery substitutes a kernel buffer for
668		 * the process that might be going away.
669		 */
670		pipe_clone_write_buffer(wpipe);
671	} else {
672		pipe_destroy_write_buffer(wpipe);
673	}
674	pipeunlock(wpipe);
675	return error;
676
677error1:
678	wakeup(wpipe);
679	return error;
680}
681#endif
682
683static int
684pipe_write(fp, uio, cred, flags, p)
685	struct file *fp;
686	struct uio *uio;
687	struct ucred *cred;
688	struct proc *p;
689	int flags;
690{
691	int error = 0;
692	int orig_resid;
693
694	struct pipe *wpipe, *rpipe;
695
696	rpipe = (struct pipe *) fp->f_data;
697	wpipe = rpipe->pipe_peer;
698
699	/*
700	 * detect loss of pipe read side, issue SIGPIPE if lost.
701	 */
702	if ((wpipe == NULL) || (wpipe->pipe_state & PIPE_EOF)) {
703		return EPIPE;
704	}
705
706	/*
707	 * If it is advantageous to resize the pipe buffer, do
708	 * so.
709	 */
710	if ((uio->uio_resid > PIPE_SIZE) &&
711		(nbigpipe < LIMITBIGPIPES) &&
712		(wpipe->pipe_state & PIPE_DIRECTW) == 0 &&
713		(wpipe->pipe_buffer.size <= PIPE_SIZE) &&
714		(wpipe->pipe_buffer.cnt == 0)) {
715
716		if (wpipe->pipe_buffer.buffer) {
717			amountpipekva -= wpipe->pipe_buffer.size;
718			kmem_free(kernel_map,
719				(vm_offset_t)wpipe->pipe_buffer.buffer,
720				wpipe->pipe_buffer.size);
721		}
722
723#ifndef PIPE_NODIRECT
724		if (wpipe->pipe_map.kva) {
725			amountpipekva -= wpipe->pipe_buffer.size + PAGE_SIZE;
726			kmem_free(kernel_map,
727				wpipe->pipe_map.kva,
728				wpipe->pipe_buffer.size + PAGE_SIZE);
729		}
730#endif
731
732		wpipe->pipe_buffer.in = 0;
733		wpipe->pipe_buffer.out = 0;
734		wpipe->pipe_buffer.cnt = 0;
735		wpipe->pipe_buffer.size = BIG_PIPE_SIZE;
736		wpipe->pipe_buffer.buffer = NULL;
737		++nbigpipe;
738
739#ifndef PIPE_NODIRECT
740		wpipe->pipe_map.cnt = 0;
741		wpipe->pipe_map.kva = 0;
742		wpipe->pipe_map.pos = 0;
743		wpipe->pipe_map.npages = 0;
744#endif
745
746	}
747
748
749	if( wpipe->pipe_buffer.buffer == NULL) {
750		if ((error = pipelock(wpipe,1)) == 0) {
751			pipespace(wpipe);
752			pipeunlock(wpipe);
753		} else {
754			return error;
755		}
756	}
757
758	++wpipe->pipe_busy;
759	orig_resid = uio->uio_resid;
760	while (uio->uio_resid) {
761		int space;
762#ifndef PIPE_NODIRECT
763		/*
764		 * If the transfer is large, we can gain performance if
765		 * we do process-to-process copies directly.
766		 * If the write is non-blocking, we don't use the
767		 * direct write mechanism.
768		 */
769		if ((uio->uio_iov->iov_len >= PIPE_MINDIRECT) &&
770		    (fp->f_flag & FNONBLOCK) == 0 &&
771			(wpipe->pipe_map.kva || (amountpipekva < LIMITPIPEKVA)) &&
772			(uio->uio_iov->iov_len >= PIPE_MINDIRECT)) {
773			error = pipe_direct_write( wpipe, uio);
774			if (error) {
775				break;
776			}
777			continue;
778		}
779#endif
780
781		/*
782		 * Pipe buffered writes cannot be coincidental with
783		 * direct writes.  We wait until the currently executing
784		 * direct write is completed before we start filling the
785		 * pipe buffer.
786		 */
787	retrywrite:
788		while (wpipe->pipe_state & PIPE_DIRECTW) {
789			if (wpipe->pipe_state & PIPE_WANTR) {
790				wpipe->pipe_state &= ~PIPE_WANTR;
791				wakeup(wpipe);
792			}
793			error = tsleep(wpipe,
794					PRIBIO|PCATCH, "pipbww", 0);
795			if (error)
796				break;
797		}
798
799		space = wpipe->pipe_buffer.size - wpipe->pipe_buffer.cnt;
800
801		/* Writes of size <= PIPE_BUF must be atomic. */
802		/* XXX perhaps they need to be contiguous to be atomic? */
803		if ((space < uio->uio_resid) && (orig_resid <= PIPE_BUF))
804			space = 0;
805
806		if (space > 0 && (wpipe->pipe_buffer.cnt < PIPE_SIZE)) {
807			/*
808			 * This set the maximum transfer as a segment of
809			 * the buffer.
810			 */
811			int size = wpipe->pipe_buffer.size - wpipe->pipe_buffer.in;
812			/*
813			 * space is the size left in the buffer
814			 */
815			if (size > space)
816				size = space;
817			/*
818			 * now limit it to the size of the uio transfer
819			 */
820			if (size > uio->uio_resid)
821				size = uio->uio_resid;
822			if ((error = pipelock(wpipe,1)) == 0) {
823				/*
824				 * It is possible for a direct write to
825				 * slip in on us... handle it here...
826				 */
827				if (wpipe->pipe_state & PIPE_DIRECTW) {
828					pipeunlock(wpipe);
829					goto retrywrite;
830				}
831				error = uiomove( &wpipe->pipe_buffer.buffer[wpipe->pipe_buffer.in],
832					size, uio);
833				pipeunlock(wpipe);
834			}
835			if (error)
836				break;
837
838			wpipe->pipe_buffer.in += size;
839			if (wpipe->pipe_buffer.in >= wpipe->pipe_buffer.size)
840				wpipe->pipe_buffer.in = 0;
841
842			wpipe->pipe_buffer.cnt += size;
843		} else {
844			/*
845			 * If the "read-side" has been blocked, wake it up now.
846			 */
847			if (wpipe->pipe_state & PIPE_WANTR) {
848				wpipe->pipe_state &= ~PIPE_WANTR;
849				wakeup(wpipe);
850			}
851
852			/*
853			 * don't block on non-blocking I/O
854			 */
855			if (fp->f_flag & FNONBLOCK) {
856				error = EAGAIN;
857				break;
858			}
859
860			/*
861			 * We have no more space and have something to offer,
862			 * wake up select/poll.
863			 */
864			pipeselwakeup(wpipe);
865
866			wpipe->pipe_state |= PIPE_WANTW;
867			if ((error = tsleep(wpipe, (PRIBIO+1)|PCATCH, "pipewr", 0)) != 0) {
868				break;
869			}
870			/*
871			 * If read side wants to go away, we just issue a signal
872			 * to ourselves.
873			 */
874			if (wpipe->pipe_state & PIPE_EOF) {
875				error = EPIPE;
876				break;
877			}
878		}
879	}
880
881	--wpipe->pipe_busy;
882	if ((wpipe->pipe_busy == 0) &&
883		(wpipe->pipe_state & PIPE_WANT)) {
884		wpipe->pipe_state &= ~(PIPE_WANT|PIPE_WANTR);
885		wakeup(wpipe);
886	} else if (wpipe->pipe_buffer.cnt > 0) {
887		/*
888		 * If we have put any characters in the buffer, we wake up
889		 * the reader.
890		 */
891		if (wpipe->pipe_state & PIPE_WANTR) {
892			wpipe->pipe_state &= ~PIPE_WANTR;
893			wakeup(wpipe);
894		}
895	}
896
897	/*
898	 * Don't return EPIPE if I/O was successful
899	 */
900	if ((wpipe->pipe_buffer.cnt == 0) &&
901		(uio->uio_resid == 0) &&
902		(error == EPIPE))
903		error = 0;
904
905	if (error == 0)
906		getnanotime(&wpipe->pipe_mtime);
907
908	/*
909	 * We have something to offer,
910	 * wake up select/poll.
911	 */
912	if (wpipe->pipe_buffer.cnt)
913		pipeselwakeup(wpipe);
914
915	return error;
916}
917
918/*
919 * we implement a very minimal set of ioctls for compatibility with sockets.
920 */
921int
922pipe_ioctl(fp, cmd, data, p)
923	struct file *fp;
924	u_long cmd;
925	register caddr_t data;
926	struct proc *p;
927{
928	register struct pipe *mpipe = (struct pipe *)fp->f_data;
929
930	switch (cmd) {
931
932	case FIONBIO:
933		return (0);
934
935	case FIOASYNC:
936		if (*(int *)data) {
937			mpipe->pipe_state |= PIPE_ASYNC;
938		} else {
939			mpipe->pipe_state &= ~PIPE_ASYNC;
940		}
941		return (0);
942
943	case FIONREAD:
944		if (mpipe->pipe_state & PIPE_DIRECTW)
945			*(int *)data = mpipe->pipe_map.cnt;
946		else
947			*(int *)data = mpipe->pipe_buffer.cnt;
948		return (0);
949
950	case FIOSETOWN:
951		return (fsetown(*(int *)data, &mpipe->pipe_sigio));
952
953	case FIOGETOWN:
954		*(int *)data = fgetown(mpipe->pipe_sigio);
955		return (0);
956
957	/* This is deprecated, FIOSETOWN should be used instead. */
958	case TIOCSPGRP:
959		return (fsetown(-(*(int *)data), &mpipe->pipe_sigio));
960
961	/* This is deprecated, FIOGETOWN should be used instead. */
962	case TIOCGPGRP:
963		*(int *)data = -fgetown(mpipe->pipe_sigio);
964		return (0);
965
966	}
967	return (ENOTTY);
968}
969
970int
971pipe_poll(fp, events, cred, p)
972	struct file *fp;
973	int events;
974	struct ucred *cred;
975	struct proc *p;
976{
977	register struct pipe *rpipe = (struct pipe *)fp->f_data;
978	struct pipe *wpipe;
979	int revents = 0;
980
981	wpipe = rpipe->pipe_peer;
982	if (events & (POLLIN | POLLRDNORM))
983		if ((rpipe->pipe_state & PIPE_DIRECTW) ||
984		    (rpipe->pipe_buffer.cnt > 0) ||
985		    (rpipe->pipe_state & PIPE_EOF))
986			revents |= events & (POLLIN | POLLRDNORM);
987
988	if (events & (POLLOUT | POLLWRNORM))
989		if (wpipe == NULL || (wpipe->pipe_state & PIPE_EOF) ||
990		    (((wpipe->pipe_state & PIPE_DIRECTW) == 0) &&
991		     (wpipe->pipe_buffer.size - wpipe->pipe_buffer.cnt) >= PIPE_BUF))
992			revents |= events & (POLLOUT | POLLWRNORM);
993
994	if ((rpipe->pipe_state & PIPE_EOF) ||
995	    (wpipe == NULL) ||
996	    (wpipe->pipe_state & PIPE_EOF))
997		revents |= POLLHUP;
998
999	if (revents == 0) {
1000		if (events & (POLLIN | POLLRDNORM)) {
1001			selrecord(p, &rpipe->pipe_sel);
1002			rpipe->pipe_state |= PIPE_SEL;
1003		}
1004
1005		if (events & (POLLOUT | POLLWRNORM)) {
1006			selrecord(p, &wpipe->pipe_sel);
1007			wpipe->pipe_state |= PIPE_SEL;
1008		}
1009	}
1010
1011	return (revents);
1012}
1013
1014int
1015pipe_stat(pipe, ub)
1016	register struct pipe *pipe;
1017	register struct stat *ub;
1018{
1019	bzero((caddr_t)ub, sizeof (*ub));
1020	ub->st_mode = S_IFIFO;
1021	ub->st_blksize = pipe->pipe_buffer.size;
1022	ub->st_size = pipe->pipe_buffer.cnt;
1023	ub->st_blocks = (ub->st_size + ub->st_blksize - 1) / ub->st_blksize;
1024	ub->st_atimespec = pipe->pipe_atime;
1025	ub->st_mtimespec = pipe->pipe_mtime;
1026	ub->st_ctimespec = pipe->pipe_ctime;
1027	/*
1028	 * Left as 0: st_dev, st_ino, st_nlink, st_uid, st_gid, st_rdev,
1029	 * st_flags, st_gen.
1030	 * XXX (st_dev, st_ino) should be unique.
1031	 */
1032	return 0;
1033}
1034
1035/* ARGSUSED */
1036static int
1037pipe_close(fp, p)
1038	struct file *fp;
1039	struct proc *p;
1040{
1041	struct pipe *cpipe = (struct pipe *)fp->f_data;
1042
1043	fp->f_ops = &badfileops;
1044	fp->f_data = NULL;
1045	funsetown(cpipe->pipe_sigio);
1046	pipeclose(cpipe);
1047	return 0;
1048}
1049
1050/*
1051 * shutdown the pipe
1052 */
1053static void
1054pipeclose(cpipe)
1055	struct pipe *cpipe;
1056{
1057	struct pipe *ppipe;
1058	if (cpipe) {
1059
1060		pipeselwakeup(cpipe);
1061
1062		/*
1063		 * If the other side is blocked, wake it up saying that
1064		 * we want to close it down.
1065		 */
1066		while (cpipe->pipe_busy) {
1067			wakeup(cpipe);
1068			cpipe->pipe_state |= PIPE_WANT|PIPE_EOF;
1069			tsleep(cpipe, PRIBIO, "pipecl", 0);
1070		}
1071
1072		/*
1073		 * Disconnect from peer
1074		 */
1075		if ((ppipe = cpipe->pipe_peer) != NULL) {
1076			pipeselwakeup(ppipe);
1077
1078			ppipe->pipe_state |= PIPE_EOF;
1079			wakeup(ppipe);
1080			ppipe->pipe_peer = NULL;
1081		}
1082
1083		/*
1084		 * free resources
1085		 */
1086		if (cpipe->pipe_buffer.buffer) {
1087			if (cpipe->pipe_buffer.size > PIPE_SIZE)
1088				--nbigpipe;
1089			amountpipekva -= cpipe->pipe_buffer.size;
1090			kmem_free(kernel_map,
1091				(vm_offset_t)cpipe->pipe_buffer.buffer,
1092				cpipe->pipe_buffer.size);
1093		}
1094#ifndef PIPE_NODIRECT
1095		if (cpipe->pipe_map.kva) {
1096			amountpipekva -= cpipe->pipe_buffer.size + PAGE_SIZE;
1097			kmem_free(kernel_map,
1098				cpipe->pipe_map.kva,
1099				cpipe->pipe_buffer.size + PAGE_SIZE);
1100		}
1101#endif
1102		zfree(pipe_zone, cpipe);
1103	}
1104}
1105