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