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 */
90static int pipe_read __P((struct file *fp, struct uio *uio,
91 struct ucred *cred, int flags));
92static int pipe_write __P((struct file *fp, struct uio *uio,
93 struct ucred *cred, int flags));
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;
| 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 */
90static int pipe_read __P((struct file *fp, struct uio *uio,
91 struct ucred *cred, int flags));
92static int pipe_write __P((struct file *fp, struct uio *uio,
93 struct ucred *cred, int flags));
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;
|
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)
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;
331 int nread = 0;
332 u_int size;
333
334 ++rpipe->pipe_busy;
335 error = pipelock(rpipe, 1);
336 if (error)
337 goto unlocked_error;
338
339 while (uio->uio_resid) {
340 /*
341 * normal pipe buffer receive
342 */
343 if (rpipe->pipe_buffer.cnt > 0) {
344 size = rpipe->pipe_buffer.size - rpipe->pipe_buffer.out;
345 if (size > rpipe->pipe_buffer.cnt)
346 size = rpipe->pipe_buffer.cnt;
347 if (size > (u_int) uio->uio_resid)
348 size = (u_int) uio->uio_resid;
349
350 error = uiomove(&rpipe->pipe_buffer.buffer[rpipe->pipe_buffer.out],
351 size, uio);
352 if (error) {
353 break;
354 }
355 rpipe->pipe_buffer.out += size;
356 if (rpipe->pipe_buffer.out >= rpipe->pipe_buffer.size)
357 rpipe->pipe_buffer.out = 0;
358
359 rpipe->pipe_buffer.cnt -= size;
360
361 /*
362 * If there is no more to read in the pipe, reset
363 * its pointers to the beginning. This improves
364 * cache hit stats.
365 */
366 if (rpipe->pipe_buffer.cnt == 0) {
367 rpipe->pipe_buffer.in = 0;
368 rpipe->pipe_buffer.out = 0;
369 }
370 nread += size;
371#ifndef PIPE_NODIRECT
372 /*
373 * Direct copy, bypassing a kernel buffer.
374 */
375 } else if ((size = rpipe->pipe_map.cnt) &&
376 (rpipe->pipe_state & PIPE_DIRECTW)) {
377 caddr_t va;
378 if (size > (u_int) uio->uio_resid)
379 size = (u_int) uio->uio_resid;
380
381 va = (caddr_t) rpipe->pipe_map.kva + rpipe->pipe_map.pos;
382 error = uiomove(va, size, uio);
383 if (error)
384 break;
385 nread += size;
386 rpipe->pipe_map.pos += size;
387 rpipe->pipe_map.cnt -= size;
388 if (rpipe->pipe_map.cnt == 0) {
389 rpipe->pipe_state &= ~PIPE_DIRECTW;
390 wakeup(rpipe);
391 }
392#endif
393 } else {
394 /*
395 * detect EOF condition
396 */
397 if (rpipe->pipe_state & PIPE_EOF) {
398 /* XXX error = ? */
399 break;
400 }
401
402 /*
403 * If the "write-side" has been blocked, wake it up now.
404 */
405 if (rpipe->pipe_state & PIPE_WANTW) {
406 rpipe->pipe_state &= ~PIPE_WANTW;
407 wakeup(rpipe);
408 }
409
410 /*
411 * Break if some data was read.
412 */
413 if (nread > 0)
414 break;
415
416 /*
417 * Unlock the pipe buffer for our remaining processing. We
418 * will either break out with an error or we will sleep and
419 * relock to loop.
420 */
421 pipeunlock(rpipe);
422
423 /*
424 * Handle non-blocking mode operation or
425 * wait for more data.
426 */
427 if (fp->f_flag & FNONBLOCK)
428 error = EAGAIN;
429 else {
430 rpipe->pipe_state |= PIPE_WANTR;
431 if ((error = tsleep(rpipe, PRIBIO|PCATCH, "piperd", 0)) == 0)
432 error = pipelock(rpipe, 1);
433 }
434 if (error)
435 goto unlocked_error;
436 }
437 }
438 pipeunlock(rpipe);
439
440 if (error == 0)
441 getnanotime(&rpipe->pipe_atime);
442unlocked_error:
443 --rpipe->pipe_busy;
444
445 /*
446 * PIPE_WANT processing only makes sense if pipe_busy is 0.
447 */
448 if ((rpipe->pipe_busy == 0) && (rpipe->pipe_state & PIPE_WANT)) {
449 rpipe->pipe_state &= ~(PIPE_WANT|PIPE_WANTW);
450 wakeup(rpipe);
451 } else if (rpipe->pipe_buffer.cnt < MINPIPESIZE) {
452 /*
453 * Handle write blocking hysteresis.
454 */
455 if (rpipe->pipe_state & PIPE_WANTW) {
456 rpipe->pipe_state &= ~PIPE_WANTW;
457 wakeup(rpipe);
458 }
459 }
460
461 if ((rpipe->pipe_buffer.size - rpipe->pipe_buffer.cnt) >= PIPE_BUF)
462 pipeselwakeup(rpipe);
463
464 return error;
465}
466
467#ifndef PIPE_NODIRECT
468/*
469 * Map the sending processes' buffer into kernel space and wire it.
470 * This is similar to a physical write operation.
471 */
472static int
473pipe_build_write_buffer(wpipe, uio)
474 struct pipe *wpipe;
475 struct uio *uio;
476{
477 u_int size;
478 int i;
479 vm_offset_t addr, endaddr, paddr;
480
481 size = (u_int) uio->uio_iov->iov_len;
482 if (size > wpipe->pipe_buffer.size)
483 size = wpipe->pipe_buffer.size;
484
485 endaddr = round_page((vm_offset_t)uio->uio_iov->iov_base + size);
486 for(i = 0, addr = trunc_page((vm_offset_t)uio->uio_iov->iov_base);
487 addr < endaddr;
488 addr += PAGE_SIZE, i+=1) {
489
490 vm_page_t m;
491
492 vm_fault_quick( (caddr_t) addr, VM_PROT_READ);
493 paddr = pmap_kextract(addr);
494 if (!paddr) {
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)
685 struct file *fp;
686 struct uio *uio;
687 struct ucred *cred;
688 int flags;
689{
690 int error = 0;
691 int orig_resid;
692
693 struct pipe *wpipe, *rpipe;
694
695 rpipe = (struct pipe *) fp->f_data;
696 wpipe = rpipe->pipe_peer;
697
698 /*
699 * detect loss of pipe read side, issue SIGPIPE if lost.
700 */
701 if ((wpipe == NULL) || (wpipe->pipe_state & PIPE_EOF)) {
702 return EPIPE;
703 }
704
705 /*
706 * If it is advantageous to resize the pipe buffer, do
707 * so.
708 */
709 if ((uio->uio_resid > PIPE_SIZE) &&
710 (nbigpipe < LIMITBIGPIPES) &&
711 (wpipe->pipe_state & PIPE_DIRECTW) == 0 &&
712 (wpipe->pipe_buffer.size <= PIPE_SIZE) &&
713 (wpipe->pipe_buffer.cnt == 0)) {
714
715 if (wpipe->pipe_buffer.buffer) {
716 amountpipekva -= wpipe->pipe_buffer.size;
717 kmem_free(kernel_map,
718 (vm_offset_t)wpipe->pipe_buffer.buffer,
719 wpipe->pipe_buffer.size);
720 }
721
722#ifndef PIPE_NODIRECT
723 if (wpipe->pipe_map.kva) {
724 amountpipekva -= wpipe->pipe_buffer.size + PAGE_SIZE;
725 kmem_free(kernel_map,
726 wpipe->pipe_map.kva,
727 wpipe->pipe_buffer.size + PAGE_SIZE);
728 }
729#endif
730
731 wpipe->pipe_buffer.in = 0;
732 wpipe->pipe_buffer.out = 0;
733 wpipe->pipe_buffer.cnt = 0;
734 wpipe->pipe_buffer.size = BIG_PIPE_SIZE;
735 wpipe->pipe_buffer.buffer = NULL;
736 ++nbigpipe;
737
738#ifndef PIPE_NODIRECT
739 wpipe->pipe_map.cnt = 0;
740 wpipe->pipe_map.kva = 0;
741 wpipe->pipe_map.pos = 0;
742 wpipe->pipe_map.npages = 0;
743#endif
744
745 }
746
747
748 if( wpipe->pipe_buffer.buffer == NULL) {
749 if ((error = pipelock(wpipe,1)) == 0) {
750 pipespace(wpipe);
751 pipeunlock(wpipe);
752 } else {
753 return error;
754 }
755 }
756
757 ++wpipe->pipe_busy;
758 orig_resid = uio->uio_resid;
759 while (uio->uio_resid) {
760 int space;
761#ifndef PIPE_NODIRECT
762 /*
763 * If the transfer is large, we can gain performance if
764 * we do process-to-process copies directly.
765 * If the write is non-blocking, we don't use the
766 * direct write mechanism.
767 */
768 if ((uio->uio_iov->iov_len >= PIPE_MINDIRECT) &&
769 (fp->f_flag & FNONBLOCK) == 0 &&
770 (wpipe->pipe_map.kva || (amountpipekva < LIMITPIPEKVA)) &&
771 (uio->uio_iov->iov_len >= PIPE_MINDIRECT)) {
772 error = pipe_direct_write( wpipe, uio);
773 if (error) {
774 break;
775 }
776 continue;
777 }
778#endif
779
780 /*
781 * Pipe buffered writes cannot be coincidental with
782 * direct writes. We wait until the currently executing
783 * direct write is completed before we start filling the
784 * pipe buffer.
785 */
786 retrywrite:
787 while (wpipe->pipe_state & PIPE_DIRECTW) {
788 if (wpipe->pipe_state & PIPE_WANTR) {
789 wpipe->pipe_state &= ~PIPE_WANTR;
790 wakeup(wpipe);
791 }
792 error = tsleep(wpipe,
793 PRIBIO|PCATCH, "pipbww", 0);
794 if (error)
795 break;
796 }
797
798 space = wpipe->pipe_buffer.size - wpipe->pipe_buffer.cnt;
799
800 /* Writes of size <= PIPE_BUF must be atomic. */
801 /* XXX perhaps they need to be contiguous to be atomic? */
802 if ((space < uio->uio_resid) && (orig_resid <= PIPE_BUF))
803 space = 0;
804
805 if (space > 0 && (wpipe->pipe_buffer.cnt < PIPE_SIZE)) {
806 /*
807 * This set the maximum transfer as a segment of
808 * the buffer.
809 */
810 int size = wpipe->pipe_buffer.size - wpipe->pipe_buffer.in;
811 /*
812 * space is the size left in the buffer
813 */
814 if (size > space)
815 size = space;
816 /*
817 * now limit it to the size of the uio transfer
818 */
819 if (size > uio->uio_resid)
820 size = uio->uio_resid;
821 if ((error = pipelock(wpipe,1)) == 0) {
822 /*
823 * It is possible for a direct write to
824 * slip in on us... handle it here...
825 */
826 if (wpipe->pipe_state & PIPE_DIRECTW) {
827 pipeunlock(wpipe);
828 goto retrywrite;
829 }
830 error = uiomove( &wpipe->pipe_buffer.buffer[wpipe->pipe_buffer.in],
831 size, uio);
832 pipeunlock(wpipe);
833 }
834 if (error)
835 break;
836
837 wpipe->pipe_buffer.in += size;
838 if (wpipe->pipe_buffer.in >= wpipe->pipe_buffer.size)
839 wpipe->pipe_buffer.in = 0;
840
841 wpipe->pipe_buffer.cnt += size;
842 } else {
843 /*
844 * If the "read-side" has been blocked, wake it up now.
845 */
846 if (wpipe->pipe_state & PIPE_WANTR) {
847 wpipe->pipe_state &= ~PIPE_WANTR;
848 wakeup(wpipe);
849 }
850
851 /*
852 * don't block on non-blocking I/O
853 */
854 if (fp->f_flag & FNONBLOCK) {
855 error = EAGAIN;
856 break;
857 }
858
859 /*
860 * We have no more space and have something to offer,
861 * wake up select/poll.
862 */
863 pipeselwakeup(wpipe);
864
865 wpipe->pipe_state |= PIPE_WANTW;
866 if ((error = tsleep(wpipe, (PRIBIO+1)|PCATCH, "pipewr", 0)) != 0) {
867 break;
868 }
869 /*
870 * If read side wants to go away, we just issue a signal
871 * to ourselves.
872 */
873 if (wpipe->pipe_state & PIPE_EOF) {
874 error = EPIPE;
875 break;
876 }
877 }
878 }
879
880 --wpipe->pipe_busy;
881 if ((wpipe->pipe_busy == 0) &&
882 (wpipe->pipe_state & PIPE_WANT)) {
883 wpipe->pipe_state &= ~(PIPE_WANT|PIPE_WANTR);
884 wakeup(wpipe);
885 } else if (wpipe->pipe_buffer.cnt > 0) {
886 /*
887 * If we have put any characters in the buffer, we wake up
888 * the reader.
889 */
890 if (wpipe->pipe_state & PIPE_WANTR) {
891 wpipe->pipe_state &= ~PIPE_WANTR;
892 wakeup(wpipe);
893 }
894 }
895
896 /*
897 * Don't return EPIPE if I/O was successful
898 */
899 if ((wpipe->pipe_buffer.cnt == 0) &&
900 (uio->uio_resid == 0) &&
901 (error == EPIPE))
902 error = 0;
903
904 if (error == 0)
905 getnanotime(&wpipe->pipe_mtime);
906
907 /*
908 * We have something to offer,
909 * wake up select/poll.
910 */
911 if (wpipe->pipe_buffer.cnt)
912 pipeselwakeup(wpipe);
913
914 return error;
915}
916
917/*
918 * we implement a very minimal set of ioctls for compatibility with sockets.
919 */
920int
921pipe_ioctl(fp, cmd, data, p)
922 struct file *fp;
923 u_long cmd;
924 register caddr_t data;
925 struct proc *p;
926{
927 register struct pipe *mpipe = (struct pipe *)fp->f_data;
928
929 switch (cmd) {
930
931 case FIONBIO:
932 return (0);
933
934 case FIOASYNC:
935 if (*(int *)data) {
936 mpipe->pipe_state |= PIPE_ASYNC;
937 } else {
938 mpipe->pipe_state &= ~PIPE_ASYNC;
939 }
940 return (0);
941
942 case FIONREAD:
943 if (mpipe->pipe_state & PIPE_DIRECTW)
944 *(int *)data = mpipe->pipe_map.cnt;
945 else
946 *(int *)data = mpipe->pipe_buffer.cnt;
947 return (0);
948
949 case FIOSETOWN:
950 return (fsetown(*(int *)data, &mpipe->pipe_sigio));
951
952 case FIOGETOWN:
953 *(int *)data = fgetown(mpipe->pipe_sigio);
954 return (0);
955
956 /* This is deprecated, FIOSETOWN should be used instead. */
957 case TIOCSPGRP:
958 return (fsetown(-(*(int *)data), &mpipe->pipe_sigio));
959
960 /* This is deprecated, FIOGETOWN should be used instead. */
961 case TIOCGPGRP:
962 *(int *)data = -fgetown(mpipe->pipe_sigio);
963 return (0);
964
965 }
966 return (ENOTTY);
967}
968
969int
970pipe_poll(fp, events, cred, p)
971 struct file *fp;
972 int events;
973 struct ucred *cred;
974 struct proc *p;
975{
976 register struct pipe *rpipe = (struct pipe *)fp->f_data;
977 struct pipe *wpipe;
978 int revents = 0;
979
980 wpipe = rpipe->pipe_peer;
981 if (events & (POLLIN | POLLRDNORM))
982 if ((rpipe->pipe_state & PIPE_DIRECTW) ||
983 (rpipe->pipe_buffer.cnt > 0) ||
984 (rpipe->pipe_state & PIPE_EOF))
985 revents |= events & (POLLIN | POLLRDNORM);
986
987 if (events & (POLLOUT | POLLWRNORM))
988 if (wpipe == NULL || (wpipe->pipe_state & PIPE_EOF) ||
989 (((wpipe->pipe_state & PIPE_DIRECTW) == 0) &&
990 (wpipe->pipe_buffer.size - wpipe->pipe_buffer.cnt) >= PIPE_BUF))
991 revents |= events & (POLLOUT | POLLWRNORM);
992
993 if ((rpipe->pipe_state & PIPE_EOF) ||
994 (wpipe == NULL) ||
995 (wpipe->pipe_state & PIPE_EOF))
996 revents |= POLLHUP;
997
998 if (revents == 0) {
999 if (events & (POLLIN | POLLRDNORM)) {
1000 selrecord(p, &rpipe->pipe_sel);
1001 rpipe->pipe_state |= PIPE_SEL;
1002 }
1003
1004 if (events & (POLLOUT | POLLWRNORM)) {
1005 selrecord(p, &wpipe->pipe_sel);
1006 wpipe->pipe_state |= PIPE_SEL;
1007 }
1008 }
1009
1010 return (revents);
1011}
1012
1013int
1014pipe_stat(pipe, ub)
1015 register struct pipe *pipe;
1016 register struct stat *ub;
1017{
1018 bzero((caddr_t)ub, sizeof (*ub));
1019 ub->st_mode = S_IFIFO;
1020 ub->st_blksize = pipe->pipe_buffer.size;
1021 ub->st_size = pipe->pipe_buffer.cnt;
1022 ub->st_blocks = (ub->st_size + ub->st_blksize - 1) / ub->st_blksize;
1023 ub->st_atimespec = pipe->pipe_atime;
1024 ub->st_mtimespec = pipe->pipe_mtime;
1025 ub->st_ctimespec = pipe->pipe_ctime;
1026 /*
1027 * Left as 0: st_dev, st_ino, st_nlink, st_uid, st_gid, st_rdev,
1028 * st_flags, st_gen.
1029 * XXX (st_dev, st_ino) should be unique.
1030 */
1031 return 0;
1032}
1033
1034/* ARGSUSED */
1035static int
1036pipe_close(fp, p)
1037 struct file *fp;
1038 struct proc *p;
1039{
1040 struct pipe *cpipe = (struct pipe *)fp->f_data;
1041
| 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)
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;
331 int nread = 0;
332 u_int size;
333
334 ++rpipe->pipe_busy;
335 error = pipelock(rpipe, 1);
336 if (error)
337 goto unlocked_error;
338
339 while (uio->uio_resid) {
340 /*
341 * normal pipe buffer receive
342 */
343 if (rpipe->pipe_buffer.cnt > 0) {
344 size = rpipe->pipe_buffer.size - rpipe->pipe_buffer.out;
345 if (size > rpipe->pipe_buffer.cnt)
346 size = rpipe->pipe_buffer.cnt;
347 if (size > (u_int) uio->uio_resid)
348 size = (u_int) uio->uio_resid;
349
350 error = uiomove(&rpipe->pipe_buffer.buffer[rpipe->pipe_buffer.out],
351 size, uio);
352 if (error) {
353 break;
354 }
355 rpipe->pipe_buffer.out += size;
356 if (rpipe->pipe_buffer.out >= rpipe->pipe_buffer.size)
357 rpipe->pipe_buffer.out = 0;
358
359 rpipe->pipe_buffer.cnt -= size;
360
361 /*
362 * If there is no more to read in the pipe, reset
363 * its pointers to the beginning. This improves
364 * cache hit stats.
365 */
366 if (rpipe->pipe_buffer.cnt == 0) {
367 rpipe->pipe_buffer.in = 0;
368 rpipe->pipe_buffer.out = 0;
369 }
370 nread += size;
371#ifndef PIPE_NODIRECT
372 /*
373 * Direct copy, bypassing a kernel buffer.
374 */
375 } else if ((size = rpipe->pipe_map.cnt) &&
376 (rpipe->pipe_state & PIPE_DIRECTW)) {
377 caddr_t va;
378 if (size > (u_int) uio->uio_resid)
379 size = (u_int) uio->uio_resid;
380
381 va = (caddr_t) rpipe->pipe_map.kva + rpipe->pipe_map.pos;
382 error = uiomove(va, size, uio);
383 if (error)
384 break;
385 nread += size;
386 rpipe->pipe_map.pos += size;
387 rpipe->pipe_map.cnt -= size;
388 if (rpipe->pipe_map.cnt == 0) {
389 rpipe->pipe_state &= ~PIPE_DIRECTW;
390 wakeup(rpipe);
391 }
392#endif
393 } else {
394 /*
395 * detect EOF condition
396 */
397 if (rpipe->pipe_state & PIPE_EOF) {
398 /* XXX error = ? */
399 break;
400 }
401
402 /*
403 * If the "write-side" has been blocked, wake it up now.
404 */
405 if (rpipe->pipe_state & PIPE_WANTW) {
406 rpipe->pipe_state &= ~PIPE_WANTW;
407 wakeup(rpipe);
408 }
409
410 /*
411 * Break if some data was read.
412 */
413 if (nread > 0)
414 break;
415
416 /*
417 * Unlock the pipe buffer for our remaining processing. We
418 * will either break out with an error or we will sleep and
419 * relock to loop.
420 */
421 pipeunlock(rpipe);
422
423 /*
424 * Handle non-blocking mode operation or
425 * wait for more data.
426 */
427 if (fp->f_flag & FNONBLOCK)
428 error = EAGAIN;
429 else {
430 rpipe->pipe_state |= PIPE_WANTR;
431 if ((error = tsleep(rpipe, PRIBIO|PCATCH, "piperd", 0)) == 0)
432 error = pipelock(rpipe, 1);
433 }
434 if (error)
435 goto unlocked_error;
436 }
437 }
438 pipeunlock(rpipe);
439
440 if (error == 0)
441 getnanotime(&rpipe->pipe_atime);
442unlocked_error:
443 --rpipe->pipe_busy;
444
445 /*
446 * PIPE_WANT processing only makes sense if pipe_busy is 0.
447 */
448 if ((rpipe->pipe_busy == 0) && (rpipe->pipe_state & PIPE_WANT)) {
449 rpipe->pipe_state &= ~(PIPE_WANT|PIPE_WANTW);
450 wakeup(rpipe);
451 } else if (rpipe->pipe_buffer.cnt < MINPIPESIZE) {
452 /*
453 * Handle write blocking hysteresis.
454 */
455 if (rpipe->pipe_state & PIPE_WANTW) {
456 rpipe->pipe_state &= ~PIPE_WANTW;
457 wakeup(rpipe);
458 }
459 }
460
461 if ((rpipe->pipe_buffer.size - rpipe->pipe_buffer.cnt) >= PIPE_BUF)
462 pipeselwakeup(rpipe);
463
464 return error;
465}
466
467#ifndef PIPE_NODIRECT
468/*
469 * Map the sending processes' buffer into kernel space and wire it.
470 * This is similar to a physical write operation.
471 */
472static int
473pipe_build_write_buffer(wpipe, uio)
474 struct pipe *wpipe;
475 struct uio *uio;
476{
477 u_int size;
478 int i;
479 vm_offset_t addr, endaddr, paddr;
480
481 size = (u_int) uio->uio_iov->iov_len;
482 if (size > wpipe->pipe_buffer.size)
483 size = wpipe->pipe_buffer.size;
484
485 endaddr = round_page((vm_offset_t)uio->uio_iov->iov_base + size);
486 for(i = 0, addr = trunc_page((vm_offset_t)uio->uio_iov->iov_base);
487 addr < endaddr;
488 addr += PAGE_SIZE, i+=1) {
489
490 vm_page_t m;
491
492 vm_fault_quick( (caddr_t) addr, VM_PROT_READ);
493 paddr = pmap_kextract(addr);
494 if (!paddr) {
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)
685 struct file *fp;
686 struct uio *uio;
687 struct ucred *cred;
688 int flags;
689{
690 int error = 0;
691 int orig_resid;
692
693 struct pipe *wpipe, *rpipe;
694
695 rpipe = (struct pipe *) fp->f_data;
696 wpipe = rpipe->pipe_peer;
697
698 /*
699 * detect loss of pipe read side, issue SIGPIPE if lost.
700 */
701 if ((wpipe == NULL) || (wpipe->pipe_state & PIPE_EOF)) {
702 return EPIPE;
703 }
704
705 /*
706 * If it is advantageous to resize the pipe buffer, do
707 * so.
708 */
709 if ((uio->uio_resid > PIPE_SIZE) &&
710 (nbigpipe < LIMITBIGPIPES) &&
711 (wpipe->pipe_state & PIPE_DIRECTW) == 0 &&
712 (wpipe->pipe_buffer.size <= PIPE_SIZE) &&
713 (wpipe->pipe_buffer.cnt == 0)) {
714
715 if (wpipe->pipe_buffer.buffer) {
716 amountpipekva -= wpipe->pipe_buffer.size;
717 kmem_free(kernel_map,
718 (vm_offset_t)wpipe->pipe_buffer.buffer,
719 wpipe->pipe_buffer.size);
720 }
721
722#ifndef PIPE_NODIRECT
723 if (wpipe->pipe_map.kva) {
724 amountpipekva -= wpipe->pipe_buffer.size + PAGE_SIZE;
725 kmem_free(kernel_map,
726 wpipe->pipe_map.kva,
727 wpipe->pipe_buffer.size + PAGE_SIZE);
728 }
729#endif
730
731 wpipe->pipe_buffer.in = 0;
732 wpipe->pipe_buffer.out = 0;
733 wpipe->pipe_buffer.cnt = 0;
734 wpipe->pipe_buffer.size = BIG_PIPE_SIZE;
735 wpipe->pipe_buffer.buffer = NULL;
736 ++nbigpipe;
737
738#ifndef PIPE_NODIRECT
739 wpipe->pipe_map.cnt = 0;
740 wpipe->pipe_map.kva = 0;
741 wpipe->pipe_map.pos = 0;
742 wpipe->pipe_map.npages = 0;
743#endif
744
745 }
746
747
748 if( wpipe->pipe_buffer.buffer == NULL) {
749 if ((error = pipelock(wpipe,1)) == 0) {
750 pipespace(wpipe);
751 pipeunlock(wpipe);
752 } else {
753 return error;
754 }
755 }
756
757 ++wpipe->pipe_busy;
758 orig_resid = uio->uio_resid;
759 while (uio->uio_resid) {
760 int space;
761#ifndef PIPE_NODIRECT
762 /*
763 * If the transfer is large, we can gain performance if
764 * we do process-to-process copies directly.
765 * If the write is non-blocking, we don't use the
766 * direct write mechanism.
767 */
768 if ((uio->uio_iov->iov_len >= PIPE_MINDIRECT) &&
769 (fp->f_flag & FNONBLOCK) == 0 &&
770 (wpipe->pipe_map.kva || (amountpipekva < LIMITPIPEKVA)) &&
771 (uio->uio_iov->iov_len >= PIPE_MINDIRECT)) {
772 error = pipe_direct_write( wpipe, uio);
773 if (error) {
774 break;
775 }
776 continue;
777 }
778#endif
779
780 /*
781 * Pipe buffered writes cannot be coincidental with
782 * direct writes. We wait until the currently executing
783 * direct write is completed before we start filling the
784 * pipe buffer.
785 */
786 retrywrite:
787 while (wpipe->pipe_state & PIPE_DIRECTW) {
788 if (wpipe->pipe_state & PIPE_WANTR) {
789 wpipe->pipe_state &= ~PIPE_WANTR;
790 wakeup(wpipe);
791 }
792 error = tsleep(wpipe,
793 PRIBIO|PCATCH, "pipbww", 0);
794 if (error)
795 break;
796 }
797
798 space = wpipe->pipe_buffer.size - wpipe->pipe_buffer.cnt;
799
800 /* Writes of size <= PIPE_BUF must be atomic. */
801 /* XXX perhaps they need to be contiguous to be atomic? */
802 if ((space < uio->uio_resid) && (orig_resid <= PIPE_BUF))
803 space = 0;
804
805 if (space > 0 && (wpipe->pipe_buffer.cnt < PIPE_SIZE)) {
806 /*
807 * This set the maximum transfer as a segment of
808 * the buffer.
809 */
810 int size = wpipe->pipe_buffer.size - wpipe->pipe_buffer.in;
811 /*
812 * space is the size left in the buffer
813 */
814 if (size > space)
815 size = space;
816 /*
817 * now limit it to the size of the uio transfer
818 */
819 if (size > uio->uio_resid)
820 size = uio->uio_resid;
821 if ((error = pipelock(wpipe,1)) == 0) {
822 /*
823 * It is possible for a direct write to
824 * slip in on us... handle it here...
825 */
826 if (wpipe->pipe_state & PIPE_DIRECTW) {
827 pipeunlock(wpipe);
828 goto retrywrite;
829 }
830 error = uiomove( &wpipe->pipe_buffer.buffer[wpipe->pipe_buffer.in],
831 size, uio);
832 pipeunlock(wpipe);
833 }
834 if (error)
835 break;
836
837 wpipe->pipe_buffer.in += size;
838 if (wpipe->pipe_buffer.in >= wpipe->pipe_buffer.size)
839 wpipe->pipe_buffer.in = 0;
840
841 wpipe->pipe_buffer.cnt += size;
842 } else {
843 /*
844 * If the "read-side" has been blocked, wake it up now.
845 */
846 if (wpipe->pipe_state & PIPE_WANTR) {
847 wpipe->pipe_state &= ~PIPE_WANTR;
848 wakeup(wpipe);
849 }
850
851 /*
852 * don't block on non-blocking I/O
853 */
854 if (fp->f_flag & FNONBLOCK) {
855 error = EAGAIN;
856 break;
857 }
858
859 /*
860 * We have no more space and have something to offer,
861 * wake up select/poll.
862 */
863 pipeselwakeup(wpipe);
864
865 wpipe->pipe_state |= PIPE_WANTW;
866 if ((error = tsleep(wpipe, (PRIBIO+1)|PCATCH, "pipewr", 0)) != 0) {
867 break;
868 }
869 /*
870 * If read side wants to go away, we just issue a signal
871 * to ourselves.
872 */
873 if (wpipe->pipe_state & PIPE_EOF) {
874 error = EPIPE;
875 break;
876 }
877 }
878 }
879
880 --wpipe->pipe_busy;
881 if ((wpipe->pipe_busy == 0) &&
882 (wpipe->pipe_state & PIPE_WANT)) {
883 wpipe->pipe_state &= ~(PIPE_WANT|PIPE_WANTR);
884 wakeup(wpipe);
885 } else if (wpipe->pipe_buffer.cnt > 0) {
886 /*
887 * If we have put any characters in the buffer, we wake up
888 * the reader.
889 */
890 if (wpipe->pipe_state & PIPE_WANTR) {
891 wpipe->pipe_state &= ~PIPE_WANTR;
892 wakeup(wpipe);
893 }
894 }
895
896 /*
897 * Don't return EPIPE if I/O was successful
898 */
899 if ((wpipe->pipe_buffer.cnt == 0) &&
900 (uio->uio_resid == 0) &&
901 (error == EPIPE))
902 error = 0;
903
904 if (error == 0)
905 getnanotime(&wpipe->pipe_mtime);
906
907 /*
908 * We have something to offer,
909 * wake up select/poll.
910 */
911 if (wpipe->pipe_buffer.cnt)
912 pipeselwakeup(wpipe);
913
914 return error;
915}
916
917/*
918 * we implement a very minimal set of ioctls for compatibility with sockets.
919 */
920int
921pipe_ioctl(fp, cmd, data, p)
922 struct file *fp;
923 u_long cmd;
924 register caddr_t data;
925 struct proc *p;
926{
927 register struct pipe *mpipe = (struct pipe *)fp->f_data;
928
929 switch (cmd) {
930
931 case FIONBIO:
932 return (0);
933
934 case FIOASYNC:
935 if (*(int *)data) {
936 mpipe->pipe_state |= PIPE_ASYNC;
937 } else {
938 mpipe->pipe_state &= ~PIPE_ASYNC;
939 }
940 return (0);
941
942 case FIONREAD:
943 if (mpipe->pipe_state & PIPE_DIRECTW)
944 *(int *)data = mpipe->pipe_map.cnt;
945 else
946 *(int *)data = mpipe->pipe_buffer.cnt;
947 return (0);
948
949 case FIOSETOWN:
950 return (fsetown(*(int *)data, &mpipe->pipe_sigio));
951
952 case FIOGETOWN:
953 *(int *)data = fgetown(mpipe->pipe_sigio);
954 return (0);
955
956 /* This is deprecated, FIOSETOWN should be used instead. */
957 case TIOCSPGRP:
958 return (fsetown(-(*(int *)data), &mpipe->pipe_sigio));
959
960 /* This is deprecated, FIOGETOWN should be used instead. */
961 case TIOCGPGRP:
962 *(int *)data = -fgetown(mpipe->pipe_sigio);
963 return (0);
964
965 }
966 return (ENOTTY);
967}
968
969int
970pipe_poll(fp, events, cred, p)
971 struct file *fp;
972 int events;
973 struct ucred *cred;
974 struct proc *p;
975{
976 register struct pipe *rpipe = (struct pipe *)fp->f_data;
977 struct pipe *wpipe;
978 int revents = 0;
979
980 wpipe = rpipe->pipe_peer;
981 if (events & (POLLIN | POLLRDNORM))
982 if ((rpipe->pipe_state & PIPE_DIRECTW) ||
983 (rpipe->pipe_buffer.cnt > 0) ||
984 (rpipe->pipe_state & PIPE_EOF))
985 revents |= events & (POLLIN | POLLRDNORM);
986
987 if (events & (POLLOUT | POLLWRNORM))
988 if (wpipe == NULL || (wpipe->pipe_state & PIPE_EOF) ||
989 (((wpipe->pipe_state & PIPE_DIRECTW) == 0) &&
990 (wpipe->pipe_buffer.size - wpipe->pipe_buffer.cnt) >= PIPE_BUF))
991 revents |= events & (POLLOUT | POLLWRNORM);
992
993 if ((rpipe->pipe_state & PIPE_EOF) ||
994 (wpipe == NULL) ||
995 (wpipe->pipe_state & PIPE_EOF))
996 revents |= POLLHUP;
997
998 if (revents == 0) {
999 if (events & (POLLIN | POLLRDNORM)) {
1000 selrecord(p, &rpipe->pipe_sel);
1001 rpipe->pipe_state |= PIPE_SEL;
1002 }
1003
1004 if (events & (POLLOUT | POLLWRNORM)) {
1005 selrecord(p, &wpipe->pipe_sel);
1006 wpipe->pipe_state |= PIPE_SEL;
1007 }
1008 }
1009
1010 return (revents);
1011}
1012
1013int
1014pipe_stat(pipe, ub)
1015 register struct pipe *pipe;
1016 register struct stat *ub;
1017{
1018 bzero((caddr_t)ub, sizeof (*ub));
1019 ub->st_mode = S_IFIFO;
1020 ub->st_blksize = pipe->pipe_buffer.size;
1021 ub->st_size = pipe->pipe_buffer.cnt;
1022 ub->st_blocks = (ub->st_size + ub->st_blksize - 1) / ub->st_blksize;
1023 ub->st_atimespec = pipe->pipe_atime;
1024 ub->st_mtimespec = pipe->pipe_mtime;
1025 ub->st_ctimespec = pipe->pipe_ctime;
1026 /*
1027 * Left as 0: st_dev, st_ino, st_nlink, st_uid, st_gid, st_rdev,
1028 * st_flags, st_gen.
1029 * XXX (st_dev, st_ino) should be unique.
1030 */
1031 return 0;
1032}
1033
1034/* ARGSUSED */
1035static int
1036pipe_close(fp, p)
1037 struct file *fp;
1038 struct proc *p;
1039{
1040 struct pipe *cpipe = (struct pipe *)fp->f_data;
1041
|