servers/vfs/pipe.c

   1 /* This file deals with the suspension and revival of processes.  A process can
   2  * be suspended because it wants to read or write from a pipe and can't, or
   3  * because it wants to read or write from a special file and can't.  When a
   4  * process can't continue it is suspended, and revived later when it is able
   5  * to continue.
   6  *
   7  * The entry points into this file are
   8  *   do_pipe:     perform the PIPE system call
   9  *   pipe_check:  check to see that a read or write on a pipe is feasible now
  10  *   suspend:     suspend a process that cannot do a requested read or write
  11  *   release:     check to see if a suspended process can be released and do
  12  *                it
  13  *   revive:      mark a suspended process as able to run again
  14  *   unsuspend_by_endpt: revive all processes blocking on a given process
  15  *   do_unpause:  a signal has been sent to a process; see if it suspended
  16  */
  17
  18 #include "fs.h"
  19 #include <fcntl.h>
  20 #include <signal.h>
  21 #include <assert.h>
  22 #include <minix/callnr.h>
  23 #include <minix/endpoint.h>
  24 #include <minix/com.h>
  25 #include <minix/u64.h>
  26 #include <sys/select.h>
  27 #include <sys/time.h>
  28 #include "file.h"
  29 #include "fproc.h"
  30 #include "scratchpad.h"
  31 #include "dmap.h"
  32 #include "param.h"
  33 #include <minix/vfsif.h>
  34 #include "vnode.h"
  35 #include "vmnt.h"
  36
  37
  38 /*===========================================================================*
  39  *                              do_pipe                                      *
  40  *===========================================================================*/
  41 int do_pipe()
  42 {
  43 /* Perform the pipe(fil_des) system call. */
  44
  45   register struct fproc *rfp;
  46   int r;
  47   struct filp *fil_ptr0, *fil_ptr1;
  48   int fil_des[2];               /* reply goes here */
  49   struct vnode *vp;
  50   struct vmnt *vmp;
  51   struct node_details res;
  52
  53   /* Get a lock on PFS */
  54   if ((vmp = find_vmnt(PFS_PROC_NR)) == NULL) panic("PFS gone");
  55   if ((r = lock_vmnt(vmp, VMNT_WRITE)) != OK) return(r);
  56
  57   /* See if a free vnode is available */
  58   if ((vp = get_free_vnode()) == NULL) {
  59         unlock_vmnt(vmp);
  60         return(err_code);
  61   }
  62   lock_vnode(vp, VNODE_OPCL);
  63
  64   /* Acquire two file descriptors. */
  65   rfp = fp;
  66   if ((r = get_fd(0, R_BIT, &fil_des[0], &fil_ptr0)) != OK) {
  67         unlock_vnode(vp);
  68         unlock_vmnt(vmp);
  69         return(r);
  70   }
  71   rfp->fp_filp[fil_des[0]] = fil_ptr0;
  72   FD_SET(fil_des[0], &rfp->fp_filp_inuse);
  73   fil_ptr0->filp_count = 1;             /* mark filp in use */
  74   if ((r = get_fd(0, W_BIT, &fil_des[1], &fil_ptr1)) != OK) {
  75         rfp->fp_filp[fil_des[0]] = NULL;
  76         FD_CLR(fil_des[0], &rfp->fp_filp_inuse);
  77         fil_ptr0->filp_count = 0;       /* mark filp free */
  78         unlock_filp(fil_ptr0);
  79         unlock_vnode(vp);
  80         unlock_vmnt(vmp);
  81         return(r);
  82   }
  83   rfp->fp_filp[fil_des[1]] = fil_ptr1;
  84   FD_SET(fil_des[1], &rfp->fp_filp_inuse);
  85   fil_ptr1->filp_count = 1;
  86
  87   /* Create a named pipe inode on PipeFS */
  88   r = req_newnode(PFS_PROC_NR, fp->fp_effuid, fp->fp_effgid, I_NAMED_PIPE,
  89                   NO_DEV, &res);
  90
  91   if (r != OK) {
  92         rfp->fp_filp[fil_des[0]] = NULL;
  93         FD_CLR(fil_des[0], &rfp->fp_filp_inuse);
  94         fil_ptr0->filp_count = 0;
  95         rfp->fp_filp[fil_des[1]] = NULL;
  96         FD_CLR(fil_des[1], &rfp->fp_filp_inuse);
  97         fil_ptr1->filp_count = 0;
  98         unlock_filp(fil_ptr1);
  99         unlock_filp(fil_ptr0);
 100         unlock_vnode(vp);
 101         unlock_vmnt(vmp);
 102         return(r);
 103   }
 104
 105   /* Fill in vnode */
 106   vp->v_fs_e = res.fs_e;
 107   vp->v_mapfs_e = res.fs_e;
 108   vp->v_inode_nr = res.inode_nr;
 109   vp->v_mapinode_nr = res.inode_nr;
 110   vp->v_mode = res.fmode;
 111   vp->v_pipe_rd_pos= 0;
 112   vp->v_pipe_wr_pos= 0;
 113   vp->v_fs_count = 1;
 114   vp->v_mapfs_count = 1;
 115   vp->v_ref_count = 1;
 116   vp->v_size = 0;
 117   vp->v_vmnt = NULL;
 118   vp->v_dev = NO_DEV;
 119
 120   /* Fill in filp objects */
 121   fil_ptr0->filp_vno = vp;
 122   dup_vnode(vp);
 123   fil_ptr1->filp_vno = vp;
 124   fil_ptr0->filp_flags = O_RDONLY;
 125   fil_ptr1->filp_flags = O_WRONLY;
 126
 127   m_out.reply_i1 = fil_des[0];
 128   m_out.reply_i2 = fil_des[1];
 129
 130   unlock_filps(fil_ptr0, fil_ptr1);
 131   unlock_vmnt(vmp);
 132
 133   return(OK);
 134 }
 135
 136
 137 /*===========================================================================*
 138  *                              map_vnode                                    *
 139  *===========================================================================*/
 140 int map_vnode(vp, map_to_fs_e)
 141 struct vnode *vp;
 142 endpoint_t map_to_fs_e;
 143 {
 144   int r;
 145   struct vmnt *vmp;
 146   struct node_details res;
 147
 148   if(vp->v_mapfs_e != NONE) return(OK); /* Already mapped; nothing to do. */
 149
 150   if ((vmp = find_vmnt(map_to_fs_e)) == NULL)
 151         panic("Can't map to unknown endpoint");
 152   if ((r = lock_vmnt(vmp, VMNT_WRITE)) != OK) {
 153         if (r == EBUSY)
 154                 vmp = NULL;     /* Already locked, do not unlock */
 155         else
 156                 return(r);
 157
 158   }
 159
 160   /* Create a temporary mapping of this inode to another FS. Read and write
 161    * operations on data will be handled by that FS. The rest by the 'original'
 162    * FS that holds the inode. */
 163   if ((r = req_newnode(map_to_fs_e, fp->fp_effuid, fp->fp_effgid, I_NAMED_PIPE,
 164                        vp->v_dev, &res)) == OK) {
 165         vp->v_mapfs_e = res.fs_e;
 166         vp->v_mapinode_nr = res.inode_nr;
 167         vp->v_mapfs_count = 1;
 168   }
 169
 170   if (vmp) unlock_vmnt(vmp);
 171
 172   return(r);
 173 }
 174
 175 /*===========================================================================*
 176  *                              pipe_check                                   *
 177  *===========================================================================*/
 178 int pipe_check(vp, rw_flag, oflags, bytes, position, notouch)
 179 register struct vnode *vp;      /* the inode of the pipe */
 180 int rw_flag;                    /* READING or WRITING */
 181 int oflags;                     /* flags set by open or fcntl */
 182 register int bytes;             /* bytes to be read or written (all chunks) */
 183 u64_t position;                 /* current file position */
 184 int notouch;                    /* check only */
 185 {
 186 /* Pipes are a little different.  If a process reads from an empty pipe for
 187  * which a writer still exists, suspend the reader.  If the pipe is empty
 188  * and there is no writer, return 0 bytes.  If a process is writing to a
 189  * pipe and no one is reading from it, give a broken pipe error.
 190  */
 191   off_t pos;
 192   int r = OK;
 193
 194   if (ex64hi(position) != 0)
 195         panic("pipe_check: position too large in pipe");
 196   pos = ex64lo(position);
 197
 198   /* If reading, check for empty pipe. */
 199   if (rw_flag == READING) {
 200         if (pos >= vp->v_size) {
 201                 /* Process is reading from an empty pipe. */
 202                 if (find_filp(vp, W_BIT) != NULL) {
 203                         /* Writer exists */
 204                         if (oflags & O_NONBLOCK)
 205                                 r = EAGAIN;
 206                         else
 207                                 r = SUSPEND;
 208
 209                         /* If need be, activate sleeping writers. */
 210                         if (susp_count > 0)
 211                                 release(vp, WRITE, susp_count);
 212                 }
 213                 return(r);
 214         }
 215         return(bytes);
 216   }
 217
 218   /* Process is writing to a pipe. */
 219   if (find_filp(vp, R_BIT) == NULL) {
 220         /* Process is writing, but there is no reader. Tell kernel to generate
 221          * a SIGPIPE signal. */
 222         if (!notouch) sys_kill(fp->fp_endpoint, SIGPIPE);
 223
 224         return(EPIPE);
 225   }
 226
 227   /* Calculate how many bytes can be written. */
 228   if (pos + bytes > PIPE_BUF) {
 229         if (oflags & O_NONBLOCK) {
 230                 if (bytes <= PIPE_BUF) {
 231                         /* Write has to be atomic */
 232                         return(EAGAIN);
 233                 }
 234
 235                 /* Compute available space */
 236                 bytes = PIPE_BUF - pos;
 237
 238                 if (bytes > 0)  {
 239                         /* Do a partial write. Need to wakeup reader */
 240                         if (!notouch)
 241                                 release(vp, READ, susp_count);
 242                         return(bytes);
 243                 } else {
 244                         /* Pipe is full */
 245                         return(EAGAIN);
 246                 }
 247         }
 248
 249         if (bytes > PIPE_BUF) {
 250                 /* Compute available space */
 251                 bytes = PIPE_BUF - pos;
 252
 253                 if (bytes > 0) {
 254                         /* Do a partial write. Need to wakeup reader
 255                          * since we'll suspend ourself in read_write()
 256                          */
 257                         if (!notouch)
 258                                 release(vp, READ, susp_count);
 259                         return(bytes);
 260                 }
 261         }
 262
 263         /* Pipe is full */
 264         return(SUSPEND);
 265   }
 266
 267   /* Writing to an empty pipe.  Search for suspended reader. */
 268   if (pos == 0 && !notouch)
 269         release(vp, READ, susp_count);
 270
 271   /* Requested amount fits */
 272   return(bytes);
 273 }
 274
 275
 276 /*===========================================================================*
 277  *                              suspend                                      *
 278  *===========================================================================*/
 279 void suspend(int why)
 280 {
 281 /* Take measures to suspend the processing of the present system call.
 282  * Store the parameters to be used upon resuming in the process table.
 283  * (Actually they are not used when a process is waiting for an I/O device,
 284  * but they are needed for pipes, and it is not worth making the distinction.)
 285  * The SUSPEND pseudo error should be returned after calling suspend().
 286  */
 287
 288   if (why == FP_BLOCKED_ON_POPEN || why == FP_BLOCKED_ON_PIPE)
 289         /* #procs susp'ed on pipe*/
 290         susp_count++;
 291
 292   fp->fp_blocked_on = why;
 293   assert(fp->fp_grant == GRANT_INVALID || !GRANT_VALID(fp->fp_grant));
 294   fp->fp_block_callnr = job_call_nr;
 295   fp->fp_flags &= ~FP_SUSP_REOPEN;              /* Clear this flag. The caller
 296                                                  * can set it when needed.
 297                                                  */
 298 }
 299
 300 /*===========================================================================*
 301  *                              wait_for                                     *
 302  *===========================================================================*/
 303 void wait_for(endpoint_t who)
 304 {
 305   if(who == NONE || who == ANY)
 306         panic("suspend on NONE or ANY");
 307   suspend(FP_BLOCKED_ON_OTHER);
 308   fp->fp_task = who;
 309 }
 310
 311
 312 /*===========================================================================*
 313  *                              pipe_suspend                                 *
 314  *===========================================================================*/
 315 void pipe_suspend(filp, buf, size)
 316 struct filp *filp;
 317 char *buf;
 318 size_t size;
 319 {
 320 /* Take measures to suspend the processing of the present system call.
 321  * Store the parameters to be used upon resuming in the process table.
 322  */
 323
 324   scratch(fp).file.filp = filp;
 325   scratch(fp).io.io_buffer = buf;
 326   scratch(fp).io.io_nbytes = size;
 327   suspend(FP_BLOCKED_ON_PIPE);
 328 }
 329
 330
 331 /*===========================================================================*
 332  *                              unsuspend_by_endpt                           *
 333  *===========================================================================*/
 334 void unsuspend_by_endpt(endpoint_t proc_e)
 335 {
 336 /* Revive processes waiting for drivers (SUSPENDed) that have disappeared with
 337  * return code EAGAIN.
 338  */
 339   struct fproc *rp;
 340
 341   for (rp = &fproc[0]; rp < &fproc[NR_PROCS]; rp++) {
 342         if (rp->fp_pid == PID_FREE) continue;
 343         if (rp->fp_blocked_on == FP_BLOCKED_ON_OTHER && rp->fp_task == proc_e)
 344                 revive(rp->fp_endpoint, EAGAIN);
 345   }
 346
 347   /* Revive processes waiting in drivers on select()s with EAGAIN too */
 348   select_unsuspend_by_endpt(proc_e);
 349
 350   return;
 351 }
 352
 353
 354 /*===========================================================================*
 355  *                              release                                      *
 356  *===========================================================================*/
 357 void release(vp, op, count)
 358 register struct vnode *vp;      /* inode of pipe */
 359 int op;                         /* READ, WRITE, OPEN or CREAT */
 360 int count;                      /* max number of processes to release */
 361 {
 362 /* Check to see if any process is hanging on vnode 'vp'. If one is, and it
 363  * was trying to perform the call indicated by 'call_nr', release it.
 364  */
 365
 366   register struct fproc *rp;
 367   struct filp *f;
 368   int selop;
 369
 370   /* Trying to perform the call also includes SELECTing on it with that
 371    * operation.
 372    */
 373   if (op == READ || op == WRITE) {
 374         if (op == READ)
 375                 selop = SEL_RD;
 376         else
 377                 selop = SEL_WR;
 378
 379         for (f = &filp[0]; f < &filp[NR_FILPS]; f++) {
 380                 if (f->filp_count < 1 || !(f->filp_pipe_select_ops & selop) ||
 381                     f->filp_vno != vp)
 382                         continue;
 383                 select_callback(f, selop);
 384                 f->filp_pipe_select_ops &= ~selop;
 385         }
 386   }
 387
 388   /* Search the proc table. */
 389   for (rp = &fproc[0]; rp < &fproc[NR_PROCS] && count > 0; rp++) {
 390         if (rp->fp_pid != PID_FREE && fp_is_blocked(rp) &&
 391             !(rp->fp_flags & FP_REVIVED) && rp->fp_block_callnr == op) {
 392                 /* Find the vnode. Depending on the reason the process was
 393                  * suspended, there are different ways of finding it.
 394                  */
 395
 396                 if (rp->fp_blocked_on == FP_BLOCKED_ON_POPEN ||
 397                     rp->fp_blocked_on == FP_BLOCKED_ON_DOPEN ||
 398                     rp->fp_blocked_on == FP_BLOCKED_ON_LOCK ||
 399                     rp->fp_blocked_on == FP_BLOCKED_ON_OTHER) {
 400                         if (!FD_ISSET(scratch(rp).file.fd_nr,
 401                                                         &rp->fp_filp_inuse))
 402                                 continue;
 403                         if (rp->fp_filp[scratch(rp).file.fd_nr]->filp_vno != vp)
 404                                 continue;
 405                 } else if (rp->fp_blocked_on == FP_BLOCKED_ON_PIPE) {
 406                         if (scratch(rp).file.filp == NULL)
 407                                 continue;
 408                         if (scratch(rp).file.filp->filp_vno != vp)
 409                                 continue;
 410                 } else
 411                         continue;
 412
 413                 /* We found the vnode. Revive process. */
 414                 revive(rp->fp_endpoint, 0);
 415                 susp_count--;   /* keep track of who is suspended */
 416                 if(susp_count < 0)
 417                         panic("susp_count now negative: %d", susp_count);
 418                 if (--count == 0) return;
 419         }
 420   }
 421 }
 422
 423
 424 /*===========================================================================*
 425  *                              revive                                       *
 426  *===========================================================================*/
 427 void revive(endpoint_t proc_e, int returned)
 428 {
 429 /* Revive a previously blocked process. When a process hangs on tty, this
 430  * is the way it is eventually released.
 431  */
 432   struct fproc *rfp;
 433   int blocked_on;
 434   int fd_nr, slot;
 435   struct filp *fil_ptr;
 436
 437   if (proc_e == NONE || isokendpt(proc_e, &slot) != OK) return;
 438
 439   rfp = &fproc[slot];
 440   if (!fp_is_blocked(rfp) || (rfp->fp_flags & FP_REVIVED)) return;
 441
 442   /* The 'reviving' flag only applies to pipes.  Processes waiting for TTY get
 443    * a message right away.  The revival process is different for TTY and pipes.
 444    * For select and TTY revival, the work is already done, for pipes it is not:
 445    * the proc must be restarted so it can try again.
 446    */
 447   blocked_on = rfp->fp_blocked_on;
 448   fd_nr = scratch(rfp).file.fd_nr;
 449   if (blocked_on == FP_BLOCKED_ON_PIPE || blocked_on == FP_BLOCKED_ON_LOCK) {
 450         /* Revive a process suspended on a pipe or lock. */
 451         rfp->fp_flags |= FP_REVIVED;
 452         reviving++;             /* process was waiting on pipe or lock */
 453   } else if (blocked_on == FP_BLOCKED_ON_DOPEN) {
 454         rfp->fp_blocked_on = FP_BLOCKED_ON_NONE;
 455         scratch(rfp).file.fd_nr = 0;
 456         if (returned < 0) {
 457                 fil_ptr = rfp->fp_filp[fd_nr];
 458                 lock_filp(fil_ptr, VNODE_OPCL);
 459                 rfp->fp_filp[fd_nr] = NULL;
 460                 FD_CLR(fd_nr, &rfp->fp_filp_inuse);
 461                 if (fil_ptr->filp_count != 1) {
 462                         panic("VFS: revive: bad count in filp: %d",
 463                                 fil_ptr->filp_count);
 464                 }
 465                 fil_ptr->filp_count = 0;
 466                 unlock_filp(fil_ptr);
 467                 put_vnode(fil_ptr->filp_vno);
 468                 fil_ptr->filp_vno = NULL;
 469                 reply(proc_e, returned);
 470         } else {
 471                 reply(proc_e, fd_nr);
 472         }
 473   } else {
 474         rfp->fp_blocked_on = FP_BLOCKED_ON_NONE;
 475         scratch(rfp).file.fd_nr = 0;
 476         if (blocked_on == FP_BLOCKED_ON_POPEN) {
 477                 /* process blocked in open or create */
 478                 reply(proc_e, fd_nr);
 479         } else if (blocked_on == FP_BLOCKED_ON_SELECT) {
 480                 reply(proc_e, returned);
 481         } else {
 482                 /* Revive a process suspended on TTY or other device.
 483                  * Pretend it wants only what there is.
 484                  */
 485                 scratch(rfp).io.io_nbytes = returned;
 486                 /* If a grant has been issued by FS for this I/O, revoke
 487                  * it again now that I/O is done.
 488                  */
 489                 if (GRANT_VALID(rfp->fp_grant)) {
 490                         if(cpf_revoke(rfp->fp_grant)) {
 491                                 panic("VFS: revoke failed for grant: %d",
 492                                         rfp->fp_grant);
 493                         }
 494                         rfp->fp_grant = GRANT_INVALID;
 495                 }
 496                 reply(proc_e, returned);/* unblock the process */
 497         }
 498   }
 499 }
 500
 501
 502 /*===========================================================================*
 503  *                              unpause                                      *
 504  *===========================================================================*/
 505 void unpause(endpoint_t proc_e)
 506 {
 507 /* A signal has been sent to a user who is paused on the file system.
 508  * Abort the system call with the EINTR error message.
 509  */
 510
 511   register struct fproc *rfp, *org_fp;
 512   int slot, blocked_on, fild, status = EINTR, major_dev, minor_dev;
 513   struct filp *f;
 514   dev_t dev;
 515   message mess;
 516   int wasreviving = 0;
 517
 518   if (isokendpt(proc_e, &slot) != OK) {
 519         printf("VFS: ignoring unpause for bogus endpoint %d\n", proc_e);
 520         return;
 521   }
 522
 523   rfp = &fproc[slot];
 524   if (!fp_is_blocked(rfp)) return;
 525   blocked_on = rfp->fp_blocked_on;
 526
 527   if (rfp->fp_flags & FP_REVIVED) {
 528         rfp->fp_flags &= ~FP_REVIVED;
 529         reviving--;
 530         wasreviving = 1;
 531   }
 532
 533   switch (blocked_on) {
 534         case FP_BLOCKED_ON_PIPE:/* process trying to read or write a pipe */
 535                 break;
 536
 537         case FP_BLOCKED_ON_LOCK:/* process trying to set a lock with FCNTL */
 538                 break;
 539
 540         case FP_BLOCKED_ON_SELECT:/* process blocking on select() */
 541                 select_forget(proc_e);
 542                 break;
 543
 544         case FP_BLOCKED_ON_POPEN:       /* process trying to open a fifo */
 545                 break;
 546
 547         case FP_BLOCKED_ON_DOPEN:/* process trying to open a device */
 548                 /* Don't cancel OPEN. Just wait until the open completes. */
 549                 return;
 550
 551         case FP_BLOCKED_ON_OTHER:/* process trying to do device I/O (e.g. tty)*/
 552                 if (rfp->fp_flags & FP_SUSP_REOPEN) {
 553                         /* Process is suspended while waiting for a reopen.
 554                          * Just reply EINTR.
 555                          */
 556                         rfp->fp_flags &= ~FP_SUSP_REOPEN;
 557                         status = EINTR;
 558                         break;
 559                 }
 560
 561                 fild = scratch(rfp).file.fd_nr;
 562                 if (fild < 0 || fild >= OPEN_MAX)
 563                         panic("file descriptor out-of-range");
 564                 f = rfp->fp_filp[fild];
 565                 dev = (dev_t) f->filp_vno->v_sdev;      /* device hung on */
 566                 major_dev = major(dev);
 567                 minor_dev = minor(dev);
 568                 mess.TTY_LINE = minor_dev;
 569                 mess.USER_ENDPT = rfp->fp_ioproc;
 570                 mess.IO_GRANT = (char *) rfp->fp_grant;
 571
 572                 /* Tell kernel R or W. Mode is from current call, not open. */
 573                 mess.COUNT = rfp->fp_block_callnr == READ ? R_BIT : W_BIT;
 574                 mess.m_type = CANCEL;
 575
 576                 org_fp = fp;
 577                 fp = rfp;       /* hack - ctty_io uses fp */
 578                 (*dmap[major_dev].dmap_io)(rfp->fp_task, &mess);
 579                 fp = org_fp;
 580                 status = mess.REP_STATUS;
 581                 if (status == SUSPEND)
 582                         return;         /* Process will be revived at a
 583                                          * later time.
 584                                          */
 585
 586                 if (status == EAGAIN) status = EINTR;
 587                 if (GRANT_VALID(rfp->fp_grant)) {
 588                         (void) cpf_revoke(rfp->fp_grant);
 589                         rfp->fp_grant = GRANT_INVALID;
 590                 }
 591                 break;
 592         default :
 593                 panic("VFS: unknown block reason: %d", blocked_on);
 594   }
 595
 596   rfp->fp_blocked_on = FP_BLOCKED_ON_NONE;
 597
 598   if ((blocked_on == FP_BLOCKED_ON_PIPE || blocked_on == FP_BLOCKED_ON_POPEN)&&
 599         !wasreviving) {
 600         susp_count--;
 601   }
 602
 603   reply(proc_e, status);        /* signal interrupted call */
 604 }
 605