sys/kern/sys_pipe.c

   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  * $FreeBSD: src/sys/kern/sys_pipe.c,v 1.60.2.13 2002/08/05 15:05:15 des Exp $
  20  * $DragonFly: src/sys/kern/sys_pipe.c,v 1.50 2008/09/09 04:06:13 dillon Exp $
  21  */
  22
  23 /*
  24  * This file contains a high-performance replacement for the socket-based
  25  * pipes scheme originally used in FreeBSD/4.4Lite.  It does not support
  26  * all features of sockets, but does do everything that pipes normally
  27  * do.
  28  */
  29 #include <sys/param.h>
  30 #include <sys/systm.h>
  31 #include <sys/kernel.h>
  32 #include <sys/proc.h>
  33 #include <sys/fcntl.h>
  34 #include <sys/file.h>
  35 #include <sys/filedesc.h>
  36 #include <sys/filio.h>
  37 #include <sys/ttycom.h>
  38 #include <sys/stat.h>
  39 #include <sys/poll.h>
  40 #include <sys/select.h>
  41 #include <sys/signalvar.h>
  42 #include <sys/sysproto.h>
  43 #include <sys/pipe.h>
  44 #include <sys/vnode.h>
  45 #include <sys/uio.h>
  46 #include <sys/event.h>
  47 #include <sys/globaldata.h>
  48 #include <sys/module.h>
  49 #include <sys/malloc.h>
  50 #include <sys/sysctl.h>
  51 #include <sys/socket.h>
  52
  53 #include <vm/vm.h>
  54 #include <vm/vm_param.h>
  55 #include <sys/lock.h>
  56 #include <vm/vm_object.h>
  57 #include <vm/vm_kern.h>
  58 #include <vm/vm_extern.h>
  59 #include <vm/pmap.h>
  60 #include <vm/vm_map.h>
  61 #include <vm/vm_page.h>
  62 #include <vm/vm_zone.h>
  63
  64 #include <sys/file2.h>
  65 #include <sys/signal2.h>
  66
  67 #include <machine/cpufunc.h>
  68
  69 /*
  70  * interfaces to the outside world
  71  */
  72 static int pipe_read (struct file *fp, struct uio *uio,
  73                 struct ucred *cred, int flags);
  74 static int pipe_write (struct file *fp, struct uio *uio,
  75                 struct ucred *cred, int flags);
  76 static int pipe_close (struct file *fp);
  77 static int pipe_shutdown (struct file *fp, int how);
  78 static int pipe_poll (struct file *fp, int events, struct ucred *cred);
  79 static int pipe_kqfilter (struct file *fp, struct knote *kn);
  80 static int pipe_stat (struct file *fp, struct stat *sb, struct ucred *cred);
  81 static int pipe_ioctl (struct file *fp, u_long cmd, caddr_t data,
  82                 struct ucred *cred, struct sysmsg *msg);
  83
  84 static struct fileops pipeops = {
  85         .fo_read = pipe_read,
  86         .fo_write = pipe_write,
  87         .fo_ioctl = pipe_ioctl,
  88         .fo_poll = pipe_poll,
  89         .fo_kqfilter = pipe_kqfilter,
  90         .fo_stat = pipe_stat,
  91         .fo_close = pipe_close,
  92         .fo_shutdown = pipe_shutdown
  93 };
  94
  95 static void     filt_pipedetach(struct knote *kn);
  96 static int      filt_piperead(struct knote *kn, long hint);
  97 static int      filt_pipewrite(struct knote *kn, long hint);
  98
  99 static struct filterops pipe_rfiltops =
 100         { 1, NULL, filt_pipedetach, filt_piperead };
 101 static struct filterops pipe_wfiltops =
 102         { 1, NULL, filt_pipedetach, filt_pipewrite };
 103
 104 MALLOC_DEFINE(M_PIPE, "pipe", "pipe structures");
 105
 106 /*
 107  * Default pipe buffer size(s), this can be kind-of large now because pipe
 108  * space is pageable.  The pipe code will try to maintain locality of
 109  * reference for performance reasons, so small amounts of outstanding I/O
 110  * will not wipe the cache.
 111  */
 112 #define MINPIPESIZE (PIPE_SIZE/3)
 113 #define MAXPIPESIZE (2*PIPE_SIZE/3)
 114
 115 /*
 116  * Limit the number of "big" pipes
 117  */
 118 #define LIMITBIGPIPES   64
 119 #define PIPEQ_MAX_CACHE 16      /* per-cpu pipe structure cache */
 120
 121 static int pipe_maxbig = LIMITBIGPIPES;
 122 static int pipe_maxcache = PIPEQ_MAX_CACHE;
 123 static int pipe_bigcount;
 124 static int pipe_nbig;
 125 static int pipe_bcache_alloc;
 126 static int pipe_bkmem_alloc;
 127 static int pipe_rblocked_count;
 128 static int pipe_wblocked_count;
 129
 130 SYSCTL_NODE(_kern, OID_AUTO, pipe, CTLFLAG_RW, 0, "Pipe operation");
 131 SYSCTL_INT(_kern_pipe, OID_AUTO, nbig,
 132         CTLFLAG_RD, &pipe_nbig, 0, "numer of big pipes allocated");
 133 SYSCTL_INT(_kern_pipe, OID_AUTO, bigcount,
 134         CTLFLAG_RW, &pipe_bigcount, 0, "number of times pipe expanded");
 135 SYSCTL_INT(_kern_pipe, OID_AUTO, rblocked,
 136         CTLFLAG_RW, &pipe_rblocked_count, 0, "number of times pipe expanded");
 137 SYSCTL_INT(_kern_pipe, OID_AUTO, wblocked,
 138         CTLFLAG_RW, &pipe_wblocked_count, 0, "number of times pipe expanded");
 139 SYSCTL_INT(_kern_pipe, OID_AUTO, maxcache,
 140         CTLFLAG_RW, &pipe_maxcache, 0, "max pipes cached per-cpu");
 141 SYSCTL_INT(_kern_pipe, OID_AUTO, maxbig,
 142         CTLFLAG_RW, &pipe_maxbig, 0, "max number of big pipes");
 143 #ifdef SMP
 144 static int pipe_delay = 5000;   /* 5uS default */
 145 SYSCTL_INT(_kern_pipe, OID_AUTO, delay,
 146         CTLFLAG_RW, &pipe_delay, 0, "SMP delay optimization in ns");
 147 static int pipe_mpsafe = 1;
 148 SYSCTL_INT(_kern_pipe, OID_AUTO, mpsafe,
 149         CTLFLAG_RW, &pipe_mpsafe, 0, "");
 150 #endif
 151 #if !defined(NO_PIPE_SYSCTL_STATS)
 152 SYSCTL_INT(_kern_pipe, OID_AUTO, bcache_alloc,
 153         CTLFLAG_RW, &pipe_bcache_alloc, 0, "pipe buffer from pcpu cache");
 154 SYSCTL_INT(_kern_pipe, OID_AUTO, bkmem_alloc,
 155         CTLFLAG_RW, &pipe_bkmem_alloc, 0, "pipe buffer from kmem");
 156 #endif
 157
 158 static void pipeclose (struct pipe *cpipe);
 159 static void pipe_free_kmem (struct pipe *cpipe);
 160 static int pipe_create (struct pipe **cpipep);
 161 static __inline void pipeselwakeup (struct pipe *cpipe);
 162 static int pipespace (struct pipe *cpipe, int size);
 163
 164 static __inline int
 165 pipeseltest(struct pipe *cpipe)
 166 {
 167         return ((cpipe->pipe_state & PIPE_SEL) ||
 168                 ((cpipe->pipe_state & PIPE_ASYNC) && cpipe->pipe_sigio) ||
 169                 SLIST_FIRST(&cpipe->pipe_sel.si_note));
 170 }
 171
 172 static __inline void
 173 pipeselwakeup(struct pipe *cpipe)
 174 {
 175         if (cpipe->pipe_state & PIPE_SEL) {
 176                 get_mplock();
 177                 cpipe->pipe_state &= ~PIPE_SEL;
 178                 selwakeup(&cpipe->pipe_sel);
 179                 rel_mplock();
 180         }
 181         if ((cpipe->pipe_state & PIPE_ASYNC) && cpipe->pipe_sigio) {
 182                 get_mplock();
 183                 pgsigio(cpipe->pipe_sigio, SIGIO, 0);
 184                 rel_mplock();
 185         }
 186         if (SLIST_FIRST(&cpipe->pipe_sel.si_note)) {
 187                 get_mplock();
 188                 KNOTE(&cpipe->pipe_sel.si_note, 0);
 189                 rel_mplock();
 190         }
 191 }
 192
 193 /*
 194  * These routines are called before and after a UIO.  The UIO
 195  * may block, causing our held tokens to be lost temporarily.
 196  *
 197  * We use these routines to serialize reads against other reads
 198  * and writes against other writes.
 199  *
 200  * The read token is held on entry so *ipp does not race.
 201  */
 202 static __inline int
 203 pipe_start_uio(struct pipe *cpipe, int *ipp)
 204 {
 205         int error;
 206
 207         while (*ipp) {
 208                 *ipp = -1;
 209                 error = tsleep(ipp, PCATCH, "pipexx", 0);
 210                 if (error)
 211                         return (error);
 212         }
 213         *ipp = 1;
 214         return (0);
 215 }
 216
 217 static __inline void
 218 pipe_end_uio(struct pipe *cpipe, int *ipp)
 219 {
 220         if (*ipp < 0) {
 221                 *ipp = 0;
 222                 wakeup(ipp);
 223         } else {
 224                 KKASSERT(*ipp > 0);
 225                 *ipp = 0;
 226         }
 227 }
 228
 229 static __inline void
 230 pipe_get_mplock(int *save)
 231 {
 232 #ifdef SMP
 233         if (pipe_mpsafe == 0) {
 234                 get_mplock();
 235                 *save = 1;
 236         } else
 237 #endif
 238         {
 239                 *save = 0;
 240         }
 241 }
 242
 243 static __inline void
 244 pipe_rel_mplock(int *save)
 245 {
 246 #ifdef SMP
 247         if (*save)
 248                 rel_mplock();
 249 #endif
 250 }
 251
 252
 253 /*
 254  * The pipe system call for the DTYPE_PIPE type of pipes
 255  *
 256  * pipe_args(int dummy)
 257  *
 258  * MPSAFE
 259  */
 260 int
 261 sys_pipe(struct pipe_args *uap)
 262 {
 263         struct thread *td = curthread;
 264         struct filedesc *fdp = td->td_proc->p_fd;
 265         struct file *rf, *wf;
 266         struct pipe *rpipe, *wpipe;
 267         int fd1, fd2, error;
 268
 269         rpipe = wpipe = NULL;
 270         if (pipe_create(&rpipe) || pipe_create(&wpipe)) {
 271                 pipeclose(rpipe);
 272                 pipeclose(wpipe);
 273                 return (ENFILE);
 274         }
 275
 276         error = falloc(td->td_lwp, &rf, &fd1);
 277         if (error) {
 278                 pipeclose(rpipe);
 279                 pipeclose(wpipe);
 280                 return (error);
 281         }
 282         uap->sysmsg_fds[0] = fd1;
 283
 284         /*
 285          * Warning: once we've gotten past allocation of the fd for the
 286          * read-side, we can only drop the read side via fdrop() in order
 287          * to avoid races against processes which manage to dup() the read
 288          * side while we are blocked trying to allocate the write side.
 289          */
 290         rf->f_type = DTYPE_PIPE;
 291         rf->f_flag = FREAD | FWRITE;
 292         rf->f_ops = &pipeops;
 293         rf->f_data = rpipe;
 294         error = falloc(td->td_lwp, &wf, &fd2);
 295         if (error) {
 296                 fsetfd(fdp, NULL, fd1);
 297                 fdrop(rf);
 298                 /* rpipe has been closed by fdrop(). */
 299                 pipeclose(wpipe);
 300                 return (error);
 301         }
 302         wf->f_type = DTYPE_PIPE;
 303         wf->f_flag = FREAD | FWRITE;
 304         wf->f_ops = &pipeops;
 305         wf->f_data = wpipe;
 306         uap->sysmsg_fds[1] = fd2;
 307
 308         rpipe->pipe_slock = kmalloc(sizeof(struct lock),
 309                                     M_PIPE, M_WAITOK|M_ZERO);
 310         wpipe->pipe_slock = rpipe->pipe_slock;
 311         rpipe->pipe_peer = wpipe;
 312         wpipe->pipe_peer = rpipe;
 313         lockinit(rpipe->pipe_slock, "pipecl", 0, 0);
 314
 315         /*
 316          * Once activated the peer relationship remains valid until
 317          * both sides are closed.
 318          */
 319         fsetfd(fdp, rf, fd1);
 320         fsetfd(fdp, wf, fd2);
 321         fdrop(rf);
 322         fdrop(wf);
 323
 324         return (0);
 325 }
 326
 327 /*
 328  * Allocate kva for pipe circular buffer, the space is pageable
 329  * This routine will 'realloc' the size of a pipe safely, if it fails
 330  * it will retain the old buffer.
 331  * If it fails it will return ENOMEM.
 332  */
 333 static int
 334 pipespace(struct pipe *cpipe, int size)
 335 {
 336         struct vm_object *object;
 337         caddr_t buffer;
 338         int npages, error;
 339
 340         npages = round_page(size) / PAGE_SIZE;
 341         object = cpipe->pipe_buffer.object;
 342
 343         /*
 344          * [re]create the object if necessary and reserve space for it
 345          * in the kernel_map.  The object and memory are pageable.  On
 346          * success, free the old resources before assigning the new
 347          * ones.
 348          */
 349         if (object == NULL || object->size != npages) {
 350                 get_mplock();
 351                 object = vm_object_allocate(OBJT_DEFAULT, npages);
 352                 buffer = (caddr_t)vm_map_min(&kernel_map);
 353
 354                 error = vm_map_find(&kernel_map, object, 0,
 355                                     (vm_offset_t *)&buffer, size,
 356                                     1,
 357                                     VM_MAPTYPE_NORMAL,
 358                                     VM_PROT_ALL, VM_PROT_ALL,
 359                                     0);
 360
 361                 if (error != KERN_SUCCESS) {
 362                         vm_object_deallocate(object);
 363                         rel_mplock();
 364                         return (ENOMEM);
 365                 }
 366                 pipe_free_kmem(cpipe);
 367                 rel_mplock();
 368                 cpipe->pipe_buffer.object = object;
 369                 cpipe->pipe_buffer.buffer = buffer;
 370                 cpipe->pipe_buffer.size = size;
 371                 ++pipe_bkmem_alloc;
 372         } else {
 373                 ++pipe_bcache_alloc;
 374         }
 375         cpipe->pipe_buffer.rindex = 0;
 376         cpipe->pipe_buffer.windex = 0;
 377         return (0);
 378 }
 379
 380 /*
 381  * Initialize and allocate VM and memory for pipe, pulling the pipe from
 382  * our per-cpu cache if possible.  For now make sure it is sized for the
 383  * smaller PIPE_SIZE default.
 384  */
 385 static int
 386 pipe_create(struct pipe **cpipep)
 387 {
 388         globaldata_t gd = mycpu;
 389         struct pipe *cpipe;
 390         int error;
 391
 392         if ((cpipe = gd->gd_pipeq) != NULL) {
 393                 gd->gd_pipeq = cpipe->pipe_peer;
 394                 --gd->gd_pipeqcount;
 395                 cpipe->pipe_peer = NULL;
 396                 cpipe->pipe_wantwcnt = 0;
 397         } else {
 398                 cpipe = kmalloc(sizeof(struct pipe), M_PIPE, M_WAITOK|M_ZERO);
 399         }
 400         *cpipep = cpipe;
 401         if ((error = pipespace(cpipe, PIPE_SIZE)) != 0)
 402                 return (error);
 403         vfs_timestamp(&cpipe->pipe_ctime);
 404         cpipe->pipe_atime = cpipe->pipe_ctime;
 405         cpipe->pipe_mtime = cpipe->pipe_ctime;
 406         lwkt_token_init(&cpipe->pipe_rlock);
 407         lwkt_token_init(&cpipe->pipe_wlock);
 408         return (0);
 409 }
 410
 411 /*
 412  * MPALMOSTSAFE (acquires mplock)
 413  */
 414 static int
 415 pipe_read(struct file *fp, struct uio *uio, struct ucred *cred, int fflags)
 416 {
 417         struct pipe *rpipe;
 418         int error;
 419         size_t nread = 0;
 420         int nbio;
 421         u_int size;     /* total bytes available */
 422         u_int nsize;    /* total bytes to read */
 423         u_int rindex;   /* contiguous bytes available */
 424         int notify_writer;
 425         lwkt_tokref rlock;
 426         lwkt_tokref wlock;
 427         int mpsave;
 428         int bigread;
 429         int bigcount;
 430
 431         if (uio->uio_resid == 0)
 432                 return(0);
 433
 434         /*
 435          * Setup locks, calculate nbio
 436          */
 437         pipe_get_mplock(&mpsave);
 438         rpipe = (struct pipe *)fp->f_data;
 439         lwkt_gettoken(&rlock, &rpipe->pipe_rlock);
 440
 441         if (fflags & O_FBLOCKING)
 442                 nbio = 0;
 443         else if (fflags & O_FNONBLOCKING)
 444                 nbio = 1;
 445         else if (fp->f_flag & O_NONBLOCK)
 446                 nbio = 1;
 447         else
 448                 nbio = 0;
 449
 450         /*
 451          * Reads are serialized.  Note howeverthat pipe_buffer.buffer and
 452          * pipe_buffer.size can change out from under us when the number
 453          * of bytes in the buffer are zero due to the write-side doing a
 454          * pipespace().
 455          */
 456         error = pipe_start_uio(rpipe, &rpipe->pipe_rip);
 457         if (error) {
 458                 pipe_rel_mplock(&mpsave);
 459                 lwkt_reltoken(&rlock);
 460                 return (error);
 461         }
 462         notify_writer = 0;
 463
 464         bigread = (uio->uio_resid > 10 * 1024 * 1024);
 465         bigcount = 10;
 466
 467         while (uio->uio_resid) {
 468                 /*
 469                  * Don't hog the cpu.
 470                  */
 471                 if (bigread && --bigcount == 0) {
 472                         lwkt_user_yield();
 473                         bigcount = 10;
 474                         if (CURSIG(curthread->td_lwp)) {
 475                                 error = EINTR;
 476                                 break;
 477                         }
 478                 }
 479
 480                 size = rpipe->pipe_buffer.windex - rpipe->pipe_buffer.rindex;
 481                 cpu_lfence();
 482                 if (size) {
 483                         rindex = rpipe->pipe_buffer.rindex &
 484                                  (rpipe->pipe_buffer.size - 1);
 485                         nsize = size;
 486                         if (nsize > rpipe->pipe_buffer.size - rindex)
 487                                 nsize = rpipe->pipe_buffer.size - rindex;
 488                         nsize = szmin(nsize, uio->uio_resid);
 489
 490                         error = uiomove(&rpipe->pipe_buffer.buffer[rindex],
 491                                         nsize, uio);
 492                         if (error)
 493                                 break;
 494                         cpu_mfence();
 495                         rpipe->pipe_buffer.rindex += nsize;
 496                         nread += nsize;
 497
 498                         /*
 499                          * If the FIFO is still over half full just continue
 500                          * and do not try to notify the writer yet.
 501                          */
 502                         if (size - nsize >= (rpipe->pipe_buffer.size >> 1)) {
 503                                 notify_writer = 0;
 504                                 continue;
 505                         }
 506
 507                         /*
 508                          * When the FIFO is less then half full notify any
 509                          * waiting writer.  WANTW can be checked while
 510                          * holding just the rlock.
 511                          */
 512                         notify_writer = 1;
 513                         if ((rpipe->pipe_state & PIPE_WANTW) == 0)
 514                                 continue;
 515                 }
 516
 517                 /*
 518                  * If the "write-side" was blocked we wake it up.  This code
 519                  * is reached either when the buffer is completely emptied
 520                  * or if it becomes more then half-empty.
 521                  *
 522                  * Pipe_state can only be modified if both the rlock and
 523                  * wlock are held.
 524                  */
 525                 if (rpipe->pipe_state & PIPE_WANTW) {
 526                         lwkt_gettoken(&wlock, &rpipe->pipe_wlock);
 527                         if (rpipe->pipe_state & PIPE_WANTW) {
 528                                 notify_writer = 0;
 529                                 rpipe->pipe_state &= ~PIPE_WANTW;
 530                                 lwkt_reltoken(&wlock);
 531                                 wakeup(rpipe);
 532                         } else {
 533                                 lwkt_reltoken(&wlock);
 534                         }
 535                 }
 536
 537                 /*
 538                  * Pick up our copy loop again if the writer sent data to
 539                  * us while we were messing around.
 540                  *
 541                  * On a SMP box poll up to pipe_delay nanoseconds for new
 542                  * data.  Typically a value of 2000 to 4000 is sufficient
 543                  * to eradicate most IPIs/tsleeps/wakeups when a pipe
 544                  * is used for synchronous communications with small packets,
 545                  * and 8000 or so (8uS) will pipeline large buffer xfers
 546                  * between cpus over a pipe.
 547                  *
 548                  * For synchronous communications a hit means doing a
 549                  * full Awrite-Bread-Bwrite-Aread cycle in less then 2uS,
 550                  * where as miss requiring a tsleep/wakeup sequence
 551                  * will take 7uS or more.
 552                  */
 553                 if (rpipe->pipe_buffer.windex != rpipe->pipe_buffer.rindex)
 554                         continue;
 555
 556 #if defined(SMP) && defined(_RDTSC_SUPPORTED_)
 557                 if (pipe_delay) {
 558                         int64_t tsc_target;
 559                         int good = 0;
 560
 561                         tsc_target = tsc_get_target(pipe_delay);
 562                         while (tsc_test_target(tsc_target) == 0) {
 563                                 if (rpipe->pipe_buffer.windex !=
 564                                     rpipe->pipe_buffer.rindex) {
 565                                         good = 1;
 566                                         break;
 567                                 }
 568                         }
 569                         if (good)
 570                                 continue;
 571                 }
 572 #endif
 573
 574                 /*
 575                  * Detect EOF condition, do not set error.
 576                  */
 577                 if (rpipe->pipe_state & PIPE_REOF)
 578                         break;
 579
 580                 /*
 581                  * Break if some data was read, or if this was a non-blocking
 582                  * read.
 583                  */
 584                 if (nread > 0)
 585                         break;
 586
 587                 if (nbio) {
 588                         error = EAGAIN;
 589                         break;
 590                 }
 591
 592                 /*
 593                  * Last chance, interlock with WANTR.
 594                  */
 595                 lwkt_gettoken(&wlock, &rpipe->pipe_wlock);
 596                 size = rpipe->pipe_buffer.windex - rpipe->pipe_buffer.rindex;
 597                 if (size) {
 598                         lwkt_reltoken(&wlock);
 599                         continue;
 600                 }
 601
 602                 /*
 603                  * Retest EOF - acquiring a new token can temporarily release
 604                  * tokens already held.
 605                  */
 606                 if (rpipe->pipe_state & PIPE_REOF) {
 607                         lwkt_reltoken(&wlock);
 608                         break;
 609                 }
 610
 611                 /*
 612                  * If there is no more to read in the pipe, reset its
 613                  * pointers to the beginning.  This improves cache hit
 614                  * stats.
 615                  *
 616                  * We need both locks to modify both pointers, and there
 617                  * must also not be a write in progress or the uiomove()
 618                  * in the write might block and temporarily release
 619                  * its wlock, then reacquire and update windex.  We are
 620                  * only serialized against reads, not writes.
 621                  *
 622                  * XXX should we even bother resetting the indices?  It
 623                  *     might actually be more cache efficient not to.
 624                  */
 625                 if (rpipe->pipe_buffer.rindex == rpipe->pipe_buffer.windex &&
 626                     rpipe->pipe_wip == 0) {
 627                         rpipe->pipe_buffer.rindex = 0;
 628                         rpipe->pipe_buffer.windex = 0;
 629                 }
 630
 631                 /*
 632                  * Wait for more data.
 633                  *
 634                  * Pipe_state can only be set if both the rlock and wlock
 635                  * are held.
 636                  */
 637                 rpipe->pipe_state |= PIPE_WANTR;
 638                 tsleep_interlock(rpipe, PCATCH);
 639                 lwkt_reltoken(&wlock);
 640                 error = tsleep(rpipe, PCATCH | PINTERLOCKED, "piperd", 0);
 641                 ++pipe_rblocked_count;
 642                 if (error)
 643                         break;
 644         }
 645         pipe_end_uio(rpipe, &rpipe->pipe_rip);
 646
 647         /*
 648          * Uptime last access time
 649          */
 650         if (error == 0 && nread)
 651                 vfs_timestamp(&rpipe->pipe_atime);
 652
 653         /*
 654          * If we drained the FIFO more then half way then handle
 655          * write blocking hysteresis.
 656          *
 657          * Note that PIPE_WANTW cannot be set by the writer without
 658          * it holding both rlock and wlock, so we can test it
 659          * while holding just rlock.
 660          */
 661         if (notify_writer) {
 662                 if (rpipe->pipe_state & PIPE_WANTW) {
 663                         lwkt_gettoken(&wlock, &rpipe->pipe_wlock);
 664                         if (rpipe->pipe_state & PIPE_WANTW) {
 665                                 rpipe->pipe_state &= ~PIPE_WANTW;
 666                                 lwkt_reltoken(&wlock);
 667                                 wakeup(rpipe);
 668                         } else {
 669                                 lwkt_reltoken(&wlock);
 670                         }
 671                 }
 672                 if (pipeseltest(rpipe)) {
 673                         lwkt_gettoken(&wlock, &rpipe->pipe_wlock);
 674                         pipeselwakeup(rpipe);
 675                         lwkt_reltoken(&wlock);
 676                 }
 677         }
 678         /*size = rpipe->pipe_buffer.windex - rpipe->pipe_buffer.rindex;*/
 679         lwkt_reltoken(&rlock);
 680
 681         pipe_rel_mplock(&mpsave);
 682         return (error);
 683 }
 684
 685 /*
 686  * MPALMOSTSAFE - acquires mplock
 687  */
 688 static int
 689 pipe_write(struct file *fp, struct uio *uio, struct ucred *cred, int fflags)
 690 {
 691         int error;
 692         int orig_resid;
 693         int nbio;
 694         struct pipe *wpipe, *rpipe;
 695         lwkt_tokref rlock;
 696         lwkt_tokref wlock;
 697         u_int windex;
 698         u_int space;
 699         u_int wcount;
 700         int mpsave;
 701         int bigwrite;
 702         int bigcount;
 703
 704         pipe_get_mplock(&mpsave);
 705
 706         /*
 707          * Writes go to the peer.  The peer will always exist.
 708          */
 709         rpipe = (struct pipe *) fp->f_data;
 710         wpipe = rpipe->pipe_peer;
 711         lwkt_gettoken(&wlock, &wpipe->pipe_wlock);
 712         if (wpipe->pipe_state & PIPE_WEOF) {
 713                 pipe_rel_mplock(&mpsave);
 714                 lwkt_reltoken(&wlock);
 715                 return (EPIPE);
 716         }
 717
 718         /*
 719          * Degenerate case (EPIPE takes prec)
 720          */
 721         if (uio->uio_resid == 0) {
 722                 pipe_rel_mplock(&mpsave);
 723                 lwkt_reltoken(&wlock);
 724                 return(0);
 725         }
 726
 727         /*
 728          * Writes are serialized (start_uio must be called with wlock)
 729          */
 730         error = pipe_start_uio(wpipe, &wpipe->pipe_wip);
 731         if (error) {
 732                 pipe_rel_mplock(&mpsave);
 733                 lwkt_reltoken(&wlock);
 734                 return (error);
 735         }
 736
 737         if (fflags & O_FBLOCKING)
 738                 nbio = 0;
 739         else if (fflags & O_FNONBLOCKING)
 740                 nbio = 1;
 741         else if (fp->f_flag & O_NONBLOCK)
 742                 nbio = 1;
 743         else
 744                 nbio = 0;
 745
 746         /*
 747          * If it is advantageous to resize the pipe buffer, do
 748          * so.  We are write-serialized so we can block safely.
 749          */
 750         if ((wpipe->pipe_buffer.size <= PIPE_SIZE) &&
 751             (pipe_nbig < pipe_maxbig) &&
 752             wpipe->pipe_wantwcnt > 4 &&
 753             (wpipe->pipe_buffer.rindex == wpipe->pipe_buffer.windex)) {
 754                 /*
 755                  * Recheck after lock.
 756                  */
 757                 lwkt_gettoken(&rlock, &wpipe->pipe_rlock);
 758                 if ((wpipe->pipe_buffer.size <= PIPE_SIZE) &&
 759                     (pipe_nbig < pipe_maxbig) &&
 760                     (wpipe->pipe_buffer.rindex == wpipe->pipe_buffer.windex)) {
 761                         atomic_add_int(&pipe_nbig, 1);
 762                         if (pipespace(wpipe, BIG_PIPE_SIZE) == 0)
 763                                 ++pipe_bigcount;
 764                         else
 765                                 atomic_subtract_int(&pipe_nbig, 1);
 766                 }
 767                 lwkt_reltoken(&rlock);
 768         }
 769
 770         orig_resid = uio->uio_resid;
 771         wcount = 0;
 772
 773         bigwrite = (uio->uio_resid > 10 * 1024 * 1024);
 774         bigcount = 10;
 775
 776         while (uio->uio_resid) {
 777                 if (wpipe->pipe_state & PIPE_WEOF) {
 778                         error = EPIPE;
 779                         break;
 780                 }
 781
 782                 /*
 783                  * Don't hog the cpu.
 784                  */
 785                 if (bigwrite && --bigcount == 0) {
 786                         lwkt_user_yield();
 787                         bigcount = 10;
 788                         if (CURSIG(curthread->td_lwp)) {
 789                                 error = EINTR;
 790                                 break;
 791                         }
 792                 }
 793
 794                 windex = wpipe->pipe_buffer.windex &
 795                          (wpipe->pipe_buffer.size - 1);
 796                 space = wpipe->pipe_buffer.size -
 797                         (wpipe->pipe_buffer.windex - wpipe->pipe_buffer.rindex);
 798                 cpu_lfence();
 799
 800                 /* Writes of size <= PIPE_BUF must be atomic. */
 801                 if ((space < uio->uio_resid) && (orig_resid <= PIPE_BUF))
 802                         space = 0;
 803
 804                 /*
 805                  * Write to fill, read size handles write hysteresis.  Also
 806                  * additional restrictions can cause select-based non-blocking
 807                  * writes to spin.
 808                  */
 809                 if (space > 0) {
 810                         u_int segsize;
 811
 812                         /*
 813                          * Transfer size is minimum of uio transfer
 814                          * and free space in pipe buffer.
 815                          *
 816                          * Limit each uiocopy to no more then PIPE_SIZE
 817                          * so we can keep the gravy train going on a
 818                          * SMP box.  This doubles the performance for
 819                          * write sizes > 16K.  Otherwise large writes
 820                          * wind up doing an inefficient synchronous
 821                          * ping-pong.
 822                          */
 823                         space = szmin(space, uio->uio_resid);
 824                         if (space > PIPE_SIZE)
 825                                 space = PIPE_SIZE;
 826
 827                         /*
 828                          * First segment to transfer is minimum of
 829                          * transfer size and contiguous space in
 830                          * pipe buffer.  If first segment to transfer
 831                          * is less than the transfer size, we've got
 832                          * a wraparound in the buffer.
 833                          */
 834                         segsize = wpipe->pipe_buffer.size - windex;
 835                         if (segsize > space)
 836                                 segsize = space;
 837
 838 #ifdef SMP
 839                         /*
 840                          * If this is the first loop and the reader is
 841                          * blocked, do a preemptive wakeup of the reader.
 842                          *
 843                          * On SMP the IPI latency plus the wlock interlock
 844                          * on the reader side is the fastest way to get the
 845                          * reader going.  (The scheduler will hard loop on
 846                          * lock tokens).
 847                          *
 848                          * NOTE: We can't clear WANTR here without acquiring
 849                          * the rlock, which we don't want to do here!
 850                          */
 851                         if ((wpipe->pipe_state & PIPE_WANTR) && pipe_mpsafe > 1)
 852                                 wakeup(wpipe);
 853 #endif
 854
 855                         /*
 856                          * Transfer segment, which may include a wrap-around.
 857                          * Update windex to account for both all in one go
 858                          * so the reader can read() the data atomically.
 859                          */
 860                         error = uiomove(&wpipe->pipe_buffer.buffer[windex],
 861                                         segsize, uio);
 862                         if (error == 0 && segsize < space) {
 863                                 segsize = space - segsize;
 864                                 error = uiomove(&wpipe->pipe_buffer.buffer[0],
 865                                                 segsize, uio);
 866                         }
 867                         if (error)
 868                                 break;
 869                         cpu_mfence();
 870                         wpipe->pipe_buffer.windex += space;
 871                         wcount += space;
 872                         continue;
 873                 }
 874
 875                 /*
 876                  * We need both the rlock and the wlock to interlock against
 877                  * the EOF, WANTW, and size checks, and to modify pipe_state.
 878                  *
 879                  * These are token locks so we do not have to worry about
 880                  * deadlocks.
 881                  */
 882                 lwkt_gettoken(&rlock, &wpipe->pipe_rlock);
 883
 884                 /*
 885                  * If the "read-side" has been blocked, wake it up now
 886                  * and yield to let it drain synchronously rather
 887                  * then block.
 888                  */
 889                 if (wpipe->pipe_state & PIPE_WANTR) {
 890                         wpipe->pipe_state &= ~PIPE_WANTR;
 891                         wakeup(wpipe);
 892                 }
 893
 894                 /*
 895                  * don't block on non-blocking I/O
 896                  */
 897                 if (nbio) {
 898                         lwkt_reltoken(&rlock);
 899                         error = EAGAIN;
 900                         break;
 901                 }
 902
 903                 /*
 904                  * re-test whether we have to block in the writer after
 905                  * acquiring both locks, in case the reader opened up
 906                  * some space.
 907                  */
 908                 space = wpipe->pipe_buffer.size -
 909                         (wpipe->pipe_buffer.windex - wpipe->pipe_buffer.rindex);
 910                 cpu_lfence();
 911                 if ((space < uio->uio_resid) && (orig_resid <= PIPE_BUF))
 912                         space = 0;
 913
 914                 /*
 915                  * Retest EOF - acquiring a new token can temporarily release
 916                  * tokens already held.
 917                  */
 918                 if (wpipe->pipe_state & PIPE_WEOF) {
 919                         lwkt_reltoken(&rlock);
 920                         error = EPIPE;
 921                         break;
 922                 }
 923
 924                 /*
 925                  * We have no more space and have something to offer,
 926                  * wake up select/poll.
 927                  */
 928                 if (space == 0) {
 929                         wpipe->pipe_state |= PIPE_WANTW;
 930                         ++wpipe->pipe_wantwcnt;
 931                         pipeselwakeup(wpipe);
 932                         if (wpipe->pipe_state & PIPE_WANTW)
 933                                 error = tsleep(wpipe, PCATCH, "pipewr", 0);
 934                         ++pipe_wblocked_count;
 935                 }
 936                 lwkt_reltoken(&rlock);
 937
 938                 /*
 939                  * Break out if we errored or the read side wants us to go
 940                  * away.
 941                  */
 942                 if (error)
 943                         break;
 944                 if (wpipe->pipe_state & PIPE_WEOF) {
 945                         error = EPIPE;
 946                         break;
 947                 }
 948         }
 949         pipe_end_uio(wpipe, &wpipe->pipe_wip);
 950
 951         /*
 952          * If we have put any characters in the buffer, we wake up
 953          * the reader.
 954          *
 955          * Both rlock and wlock are required to be able to modify pipe_state.
 956          */
 957         if (wpipe->pipe_buffer.windex != wpipe->pipe_buffer.rindex) {
 958                 if (wpipe->pipe_state & PIPE_WANTR) {
 959                         lwkt_gettoken(&rlock, &wpipe->pipe_rlock);
 960                         if (wpipe->pipe_state & PIPE_WANTR) {
 961                                 wpipe->pipe_state &= ~PIPE_WANTR;
 962                                 lwkt_reltoken(&rlock);
 963                                 wakeup(wpipe);
 964                         } else {
 965                                 lwkt_reltoken(&rlock);
 966                         }
 967                 }
 968                 if (pipeseltest(wpipe)) {
 969                         lwkt_gettoken(&rlock, &wpipe->pipe_rlock);
 970                         pipeselwakeup(wpipe);
 971                         lwkt_reltoken(&rlock);
 972                 }
 973         }
 974
 975         /*
 976          * Don't return EPIPE if I/O was successful
 977          */
 978         if ((wpipe->pipe_buffer.rindex == wpipe->pipe_buffer.windex) &&
 979             (uio->uio_resid == 0) &&
 980             (error == EPIPE)) {
 981                 error = 0;
 982         }
 983
 984         if (error == 0)
 985                 vfs_timestamp(&wpipe->pipe_mtime);
 986
 987         /*
 988          * We have something to offer,
 989          * wake up select/poll.
 990          */
 991         /*space = wpipe->pipe_buffer.windex - wpipe->pipe_buffer.rindex;*/
 992         lwkt_reltoken(&wlock);
 993         pipe_rel_mplock(&mpsave);
 994         return (error);
 995 }
 996
 997 /*
 998  * MPALMOSTSAFE - acquires mplock
 999  *
1000  * we implement a very minimal set of ioctls for compatibility with sockets.
1001  */
1002 int
1003 pipe_ioctl(struct file *fp, u_long cmd, caddr_t data,
1004            struct ucred *cred, struct sysmsg *msg)
1005 {
1006         struct pipe *mpipe;
1007         lwkt_tokref rlock;
1008         lwkt_tokref wlock;
1009         int error;
1010         int mpsave;
1011
1012         pipe_get_mplock(&mpsave);
1013         mpipe = (struct pipe *)fp->f_data;
1014
1015         lwkt_gettoken(&rlock, &mpipe->pipe_rlock);
1016         lwkt_gettoken(&wlock, &mpipe->pipe_wlock);
1017
1018         switch (cmd) {
1019         case FIOASYNC:
1020                 if (*(int *)data) {
1021                         mpipe->pipe_state |= PIPE_ASYNC;
1022                 } else {
1023                         mpipe->pipe_state &= ~PIPE_ASYNC;
1024                 }
1025                 error = 0;
1026                 break;
1027         case FIONREAD:
1028                 *(int *)data = mpipe->pipe_buffer.windex -
1029                                 mpipe->pipe_buffer.rindex;
1030                 error = 0;
1031                 break;
1032         case FIOSETOWN:
1033                 get_mplock();
1034                 error = fsetown(*(int *)data, &mpipe->pipe_sigio);
1035                 rel_mplock();
1036                 break;
1037         case FIOGETOWN:
1038                 *(int *)data = fgetown(mpipe->pipe_sigio);
1039                 error = 0;
1040                 break;
1041         case TIOCSPGRP:
1042                 /* This is deprecated, FIOSETOWN should be used instead. */
1043                 get_mplock();
1044                 error = fsetown(-(*(int *)data), &mpipe->pipe_sigio);
1045                 rel_mplock();
1046                 break;
1047
1048         case TIOCGPGRP:
1049                 /* This is deprecated, FIOGETOWN should be used instead. */
1050                 *(int *)data = -fgetown(mpipe->pipe_sigio);
1051                 error = 0;
1052                 break;
1053         default:
1054                 error = ENOTTY;
1055                 break;
1056         }
1057         lwkt_reltoken(&rlock);
1058         lwkt_reltoken(&wlock);
1059         pipe_rel_mplock(&mpsave);
1060
1061         return (error);
1062 }
1063
1064 /*
1065  * MPALMOSTSAFE - acquires mplock
1066  *
1067  * poll for events (helper)
1068  */
1069 static int
1070 pipe_poll_events(struct pipe *rpipe, struct pipe *wpipe, int events)
1071 {
1072         int revents = 0;
1073         u_int space;
1074
1075         if (events & (POLLIN | POLLRDNORM)) {
1076                 if ((rpipe->pipe_buffer.windex != rpipe->pipe_buffer.rindex) ||
1077                     (rpipe->pipe_state & PIPE_REOF)) {
1078                         revents |= events & (POLLIN | POLLRDNORM);
1079                 }
1080         }
1081
1082         if (events & (POLLOUT | POLLWRNORM)) {
1083                 if (wpipe == NULL || (wpipe->pipe_state & PIPE_WEOF)) {
1084                         revents |= events & (POLLOUT | POLLWRNORM);
1085                 } else {
1086                         space = wpipe->pipe_buffer.windex -
1087                                 wpipe->pipe_buffer.rindex;
1088                         space = wpipe->pipe_buffer.size - space;
1089                         if (space >= PIPE_BUF)
1090                                 revents |= events & (POLLOUT | POLLWRNORM);
1091                 }
1092         }
1093
1094         if ((rpipe->pipe_state & PIPE_REOF) ||
1095             (wpipe == NULL) ||
1096             (wpipe->pipe_state & PIPE_WEOF)) {
1097                 revents |= POLLHUP;
1098         }
1099         return (revents);
1100 }
1101
1102 /*
1103  * Poll for events from file pointer.
1104  */
1105 int
1106 pipe_poll(struct file *fp, int events, struct ucred *cred)
1107 {
1108         lwkt_tokref rpipe_rlock;
1109         lwkt_tokref rpipe_wlock;
1110         lwkt_tokref wpipe_rlock;
1111         lwkt_tokref wpipe_wlock;
1112         struct pipe *rpipe;
1113         struct pipe *wpipe;
1114         int revents = 0;
1115         int mpsave;
1116
1117         pipe_get_mplock(&mpsave);
1118         rpipe = (struct pipe *)fp->f_data;
1119         wpipe = rpipe->pipe_peer;
1120
1121         revents = pipe_poll_events(rpipe, wpipe, events);
1122         if (revents == 0) {
1123                 if (events & (POLLIN | POLLRDNORM)) {
1124                         lwkt_gettoken(&rpipe_rlock, &rpipe->pipe_rlock);
1125                         lwkt_gettoken(&rpipe_wlock, &rpipe->pipe_wlock);
1126                 }
1127                 if (events & (POLLOUT | POLLWRNORM)) {
1128                         lwkt_gettoken(&wpipe_rlock, &wpipe->pipe_rlock);
1129                         lwkt_gettoken(&wpipe_wlock, &wpipe->pipe_wlock);
1130                 }
1131                 revents = pipe_poll_events(rpipe, wpipe, events);
1132                 if (revents == 0) {
1133                         if (events & (POLLIN | POLLRDNORM)) {
1134                                 selrecord(curthread, &rpipe->pipe_sel);
1135                                 rpipe->pipe_state |= PIPE_SEL;
1136                         }
1137
1138                         if (events & (POLLOUT | POLLWRNORM)) {
1139                                 selrecord(curthread, &wpipe->pipe_sel);
1140                                 wpipe->pipe_state |= PIPE_SEL;
1141                         }
1142                 }
1143                 if (events & (POLLIN | POLLRDNORM)) {
1144                         lwkt_reltoken(&rpipe_rlock);
1145                         lwkt_reltoken(&rpipe_wlock);
1146                 }
1147                 if (events & (POLLOUT | POLLWRNORM)) {
1148                         lwkt_reltoken(&wpipe_rlock);
1149                         lwkt_reltoken(&wpipe_wlock);
1150                 }
1151         }
1152         pipe_rel_mplock(&mpsave);
1153         return (revents);
1154 }
1155
1156 /*
1157  * MPSAFE
1158  */
1159 static int
1160 pipe_stat(struct file *fp, struct stat *ub, struct ucred *cred)
1161 {
1162         struct pipe *pipe;
1163         int mpsave;
1164
1165         pipe_get_mplock(&mpsave);
1166         pipe = (struct pipe *)fp->f_data;
1167
1168         bzero((caddr_t)ub, sizeof(*ub));
1169         ub->st_mode = S_IFIFO;
1170         ub->st_blksize = pipe->pipe_buffer.size;
1171         ub->st_size = pipe->pipe_buffer.windex - pipe->pipe_buffer.rindex;
1172         ub->st_blocks = (ub->st_size + ub->st_blksize - 1) / ub->st_blksize;
1173         ub->st_atimespec = pipe->pipe_atime;
1174         ub->st_mtimespec = pipe->pipe_mtime;
1175         ub->st_ctimespec = pipe->pipe_ctime;
1176         /*
1177          * Left as 0: st_dev, st_ino, st_nlink, st_uid, st_gid, st_rdev,
1178          * st_flags, st_gen.
1179          * XXX (st_dev, st_ino) should be unique.
1180          */
1181         pipe_rel_mplock(&mpsave);
1182         return (0);
1183 }
1184
1185 /*
1186  * MPALMOSTSAFE - acquires mplock
1187  */
1188 static int
1189 pipe_close(struct file *fp)
1190 {
1191         struct pipe *cpipe;
1192
1193         get_mplock();
1194         cpipe = (struct pipe *)fp->f_data;
1195         fp->f_ops = &badfileops;
1196         fp->f_data = NULL;
1197         funsetown(cpipe->pipe_sigio);
1198         pipeclose(cpipe);
1199         rel_mplock();
1200         return (0);
1201 }
1202
1203 /*
1204  * Shutdown one or both directions of a full-duplex pipe.
1205  *
1206  * MPALMOSTSAFE - acquires mplock
1207  */
1208 static int
1209 pipe_shutdown(struct file *fp, int how)
1210 {
1211         struct pipe *rpipe;
1212         struct pipe *wpipe;
1213         int error = EPIPE;
1214         lwkt_tokref rpipe_rlock;
1215         lwkt_tokref rpipe_wlock;
1216         lwkt_tokref wpipe_rlock;
1217         lwkt_tokref wpipe_wlock;
1218         int mpsave;
1219
1220         pipe_get_mplock(&mpsave);
1221         rpipe = (struct pipe *)fp->f_data;
1222         wpipe = rpipe->pipe_peer;
1223
1224         /*
1225          * We modify pipe_state on both pipes, which means we need
1226          * all four tokens!
1227          */
1228         lwkt_gettoken(&rpipe_rlock, &rpipe->pipe_rlock);
1229         lwkt_gettoken(&rpipe_wlock, &rpipe->pipe_wlock);
1230         lwkt_gettoken(&wpipe_rlock, &wpipe->pipe_rlock);
1231         lwkt_gettoken(&wpipe_wlock, &wpipe->pipe_wlock);
1232
1233         switch(how) {
1234         case SHUT_RDWR:
1235         case SHUT_RD:
1236                 rpipe->pipe_state |= PIPE_REOF;         /* my reads */
1237                 rpipe->pipe_state |= PIPE_WEOF;         /* peer writes */
1238                 if (rpipe->pipe_state & PIPE_WANTR) {
1239                         rpipe->pipe_state &= ~PIPE_WANTR;
1240                         wakeup(rpipe);
1241                 }
1242                 if (rpipe->pipe_state & PIPE_WANTW) {
1243                         rpipe->pipe_state &= ~PIPE_WANTW;
1244                         wakeup(rpipe);
1245                 }
1246                 error = 0;
1247                 if (how == SHUT_RD)
1248                         break;
1249                 /* fall through */
1250         case SHUT_WR:
1251                 wpipe->pipe_state |= PIPE_REOF;         /* peer reads */
1252                 wpipe->pipe_state |= PIPE_WEOF;         /* my writes */
1253                 if (wpipe->pipe_state & PIPE_WANTR) {
1254                         wpipe->pipe_state &= ~PIPE_WANTR;
1255                         wakeup(wpipe);
1256                 }
1257                 if (wpipe->pipe_state & PIPE_WANTW) {
1258                         wpipe->pipe_state &= ~PIPE_WANTW;
1259                         wakeup(wpipe);
1260                 }
1261                 error = 0;
1262                 break;
1263         }
1264         pipeselwakeup(rpipe);
1265         pipeselwakeup(wpipe);
1266
1267         lwkt_reltoken(&rpipe_rlock);
1268         lwkt_reltoken(&rpipe_wlock);
1269         lwkt_reltoken(&wpipe_rlock);
1270         lwkt_reltoken(&wpipe_wlock);
1271
1272         pipe_rel_mplock(&mpsave);
1273         return (error);
1274 }
1275
1276 static void
1277 pipe_free_kmem(struct pipe *cpipe)
1278 {
1279         if (cpipe->pipe_buffer.buffer != NULL) {
1280                 if (cpipe->pipe_buffer.size > PIPE_SIZE)
1281                         atomic_subtract_int(&pipe_nbig, 1);
1282                 kmem_free(&kernel_map,
1283                         (vm_offset_t)cpipe->pipe_buffer.buffer,
1284                         cpipe->pipe_buffer.size);
1285                 cpipe->pipe_buffer.buffer = NULL;
1286                 cpipe->pipe_buffer.object = NULL;
1287         }
1288 }
1289
1290 /*
1291  * Close the pipe.  The slock must be held to interlock against simultanious
1292  * closes.  The rlock and wlock must be held to adjust the pipe_state.
1293  */
1294 static void
1295 pipeclose(struct pipe *cpipe)
1296 {
1297         globaldata_t gd;
1298         struct pipe *ppipe;
1299         lwkt_tokref cpipe_rlock;
1300         lwkt_tokref cpipe_wlock;
1301         lwkt_tokref ppipe_rlock;
1302         lwkt_tokref ppipe_wlock;
1303
1304         if (cpipe == NULL)
1305                 return;
1306
1307         /*
1308          * The slock may not have been allocated yet (close during
1309          * initialization)
1310          *
1311          * We need both the read and write tokens to modify pipe_state.
1312          */
1313         if (cpipe->pipe_slock)
1314                 lockmgr(cpipe->pipe_slock, LK_EXCLUSIVE);
1315         lwkt_gettoken(&cpipe_rlock, &cpipe->pipe_rlock);
1316         lwkt_gettoken(&cpipe_wlock, &cpipe->pipe_wlock);
1317
1318         /*
1319          * Set our state, wakeup anyone waiting in select, and
1320          * wakeup anyone blocked on our pipe.
1321          */
1322         cpipe->pipe_state |= PIPE_CLOSED | PIPE_REOF | PIPE_WEOF;
1323         pipeselwakeup(cpipe);
1324         if (cpipe->pipe_state & (PIPE_WANTR | PIPE_WANTW)) {
1325                 cpipe->pipe_state &= ~(PIPE_WANTR | PIPE_WANTW);
1326                 wakeup(cpipe);
1327         }
1328
1329         /*
1330          * Disconnect from peer.
1331          */
1332         if ((ppipe = cpipe->pipe_peer) != NULL) {
1333                 lwkt_gettoken(&ppipe_rlock, &ppipe->pipe_rlock);
1334                 lwkt_gettoken(&ppipe_wlock, &ppipe->pipe_wlock);
1335                 ppipe->pipe_state |= PIPE_REOF | PIPE_WEOF;
1336                 pipeselwakeup(ppipe);
1337                 if (ppipe->pipe_state & (PIPE_WANTR | PIPE_WANTW)) {
1338                         ppipe->pipe_state &= ~(PIPE_WANTR | PIPE_WANTW);
1339                         wakeup(ppipe);
1340                 }
1341                 if (SLIST_FIRST(&ppipe->pipe_sel.si_note)) {
1342                         get_mplock();
1343                         KNOTE(&ppipe->pipe_sel.si_note, 0);
1344                         rel_mplock();
1345                 }
1346                 lwkt_reltoken(&ppipe_rlock);
1347                 lwkt_reltoken(&ppipe_wlock);
1348         }
1349
1350         /*
1351          * If the peer is also closed we can free resources for both
1352          * sides, otherwise we leave our side intact to deal with any
1353          * races (since we only have the slock).
1354          */
1355         if (ppipe && (ppipe->pipe_state & PIPE_CLOSED)) {
1356                 cpipe->pipe_peer = NULL;
1357                 ppipe->pipe_peer = NULL;
1358                 ppipe->pipe_slock = NULL;       /* we will free the slock */
1359                 pipeclose(ppipe);
1360                 ppipe = NULL;
1361         }
1362
1363         lwkt_reltoken(&cpipe_rlock);
1364         lwkt_reltoken(&cpipe_wlock);
1365         if (cpipe->pipe_slock)
1366                 lockmgr(cpipe->pipe_slock, LK_RELEASE);
1367
1368         /*
1369          * If we disassociated from our peer we can free resources
1370          */
1371         if (ppipe == NULL) {
1372                 gd = mycpu;
1373                 if (cpipe->pipe_slock) {
1374                         kfree(cpipe->pipe_slock, M_PIPE);
1375                         cpipe->pipe_slock = NULL;
1376                 }
1377                 if (gd->gd_pipeqcount >= pipe_maxcache ||
1378                     cpipe->pipe_buffer.size != PIPE_SIZE
1379                 ) {
1380                         pipe_free_kmem(cpipe);
1381                         kfree(cpipe, M_PIPE);
1382                 } else {
1383                         cpipe->pipe_state = 0;
1384                         cpipe->pipe_peer = gd->gd_pipeq;
1385                         gd->gd_pipeq = cpipe;
1386                         ++gd->gd_pipeqcount;
1387                 }
1388         }
1389 }
1390
1391 /*
1392  * MPALMOSTSAFE - acquires mplock
1393  */
1394 static int
1395 pipe_kqfilter(struct file *fp, struct knote *kn)
1396 {
1397         struct pipe *cpipe;
1398
1399         get_mplock();
1400         cpipe = (struct pipe *)kn->kn_fp->f_data;
1401
1402         switch (kn->kn_filter) {
1403         case EVFILT_READ:
1404                 kn->kn_fop = &pipe_rfiltops;
1405                 break;
1406         case EVFILT_WRITE:
1407                 kn->kn_fop = &pipe_wfiltops;
1408                 cpipe = cpipe->pipe_peer;
1409                 if (cpipe == NULL) {
1410                         /* other end of pipe has been closed */
1411                         rel_mplock();
1412                         return (EPIPE);
1413                 }
1414                 break;
1415         default:
1416                 return (1);
1417         }
1418         kn->kn_hook = (caddr_t)cpipe;
1419
1420         SLIST_INSERT_HEAD(&cpipe->pipe_sel.si_note, kn, kn_selnext);
1421         rel_mplock();
1422         return (0);
1423 }
1424
1425 static void
1426 filt_pipedetach(struct knote *kn)
1427 {
1428         struct pipe *cpipe = (struct pipe *)kn->kn_hook;
1429
1430         SLIST_REMOVE(&cpipe->pipe_sel.si_note, kn, knote, kn_selnext);
1431 }
1432
1433 /*ARGSUSED*/
1434 static int
1435 filt_piperead(struct knote *kn, long hint)
1436 {
1437         struct pipe *rpipe = (struct pipe *)kn->kn_fp->f_data;
1438
1439         kn->kn_data = rpipe->pipe_buffer.windex - rpipe->pipe_buffer.rindex;
1440
1441         /* XXX RACE */
1442         if (rpipe->pipe_state & PIPE_REOF) {
1443                 kn->kn_flags |= EV_EOF;
1444                 return (1);
1445         }
1446         return (kn->kn_data > 0);
1447 }
1448
1449 /*ARGSUSED*/
1450 static int
1451 filt_pipewrite(struct knote *kn, long hint)
1452 {
1453         struct pipe *rpipe = (struct pipe *)kn->kn_fp->f_data;
1454         struct pipe *wpipe = rpipe->pipe_peer;
1455         u_int32_t space;
1456
1457         /* XXX RACE */
1458         if ((wpipe == NULL) || (wpipe->pipe_state & PIPE_WEOF)) {
1459                 kn->kn_data = 0;
1460                 kn->kn_flags |= EV_EOF;
1461                 return (1);
1462         }
1463         space = wpipe->pipe_buffer.windex -
1464                 wpipe->pipe_buffer.rindex;
1465         space = wpipe->pipe_buffer.size - space;
1466         kn->kn_data = space;
1467         return (kn->kn_data >= PIPE_BUF);
1468 }