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