sanchayanmaity/minix
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity
minix/servers/vfs/pipe.c

514 lines
15 KiB
C
Raw Blame History

/* This file deals with the suspension and revival of processes. A process can
* be suspended because it wants to read or write from a pipe and can't, or
* because it wants to read or write from a special file and can't. When a
* process can't continue it is suspended, and revived later when it is able
* to continue.
*
* The entry points into this file are
* do_pipe: perform the PIPE system call
* pipe_check: check to see that a read or write on a pipe is feasible now
* suspend: suspend a process that cannot do a requested read or write
* release: check to see if a suspended process can be released and do
* it
* revive: mark a suspended process as able to run again
* unsuspend_by_endpt: revive all processes blocking on a given process
* do_unpause: a signal has been sent to a process; see if it suspended
*
* Changes for VFS:
* Jul 2006 (Balazs Gerofi)
*/
#include "fs.h"
#include <fcntl.h>
#include <signal.h>
#include <assert.h>
#include <minix/callnr.h>
#include <minix/endpoint.h>
#include <minix/com.h>
#include <sys/select.h>
#include <sys/time.h>
#include "file.h"
#include "fproc.h"
#include "param.h"
#include "select.h"
#include <minix/vfsif.h>
#include "vnode.h"
#include "vmnt.h"
/*===========================================================================*
* do_pipe *
*===========================================================================*/
PUBLIC int do_pipe()
{
/* Perform the pipe(fil_des) system call. */
register struct fproc *rfp;
int r;
struct filp *fil_ptr0, *fil_ptr1;
int fil_des[2]; /* reply goes here */
struct vnode *vp;
struct vmnt *vmp;
struct pipe_req req;
struct node_details res;
/* See if a free vnode is available */
if ( (vp = get_free_vnode()) == NIL_VNODE) {
printf("VFS: no vnode available!\n");
return err_code;
}
/* Acquire two file descriptors. */
rfp = fp;
if ( (r = get_fd(0, R_BIT, &fil_des[0], &fil_ptr0)) != OK) return(r);
rfp->fp_filp[fil_des[0]] = fil_ptr0;
FD_SET(fil_des[0], &rfp->fp_filp_inuse);
fil_ptr0->filp_count = 1;
if ( (r = get_fd(0, W_BIT, &fil_des[1], &fil_ptr1)) != OK) {
rfp->fp_filp[fil_des[0]] = NIL_FILP;
FD_CLR(fil_des[0], &rfp->fp_filp_inuse);
fil_ptr0->filp_count = 0;
return(r);
}
rfp->fp_filp[fil_des[1]] = fil_ptr1;
FD_SET(fil_des[1], &rfp->fp_filp_inuse);
fil_ptr1->filp_count = 1;
/* Fill in FS request */
req.fs_e = ROOT_FS_E;
req.uid = fp->fp_effuid;
req.gid = fp->fp_effgid;
/* Send request */
r = req_pipe(&req, &res);
/* Handle error */
if (r != OK) {
rfp->fp_filp[fil_des[0]] = NIL_FILP;
FD_CLR(fil_des[0], &rfp->fp_filp_inuse);
fil_ptr0->filp_count = 0;
rfp->fp_filp[fil_des[1]] = NIL_FILP;
FD_CLR(fil_des[1], &rfp->fp_filp_inuse);
fil_ptr1->filp_count = 0;
return r;
}
/* Fill in vnode */
vp->v_fs_e = res.fs_e;
vp->v_inode_nr = res.inode_nr;
vp->v_mode = res.fmode;
vp->v_index = res.inode_index;
vp->v_pipe = I_PIPE;
vp->v_count = 2; /* Double usage */
vp->v_size = 0;
if ( (vmp = find_vmnt(vp->v_fs_e)) == NIL_VMNT) {
printf("VFS: vmnt not found by pipe() ==>> USING ROOT VMNT\n");
vp->v_vmnt = &vmnt[0];
}
else {
vp->v_vmnt = vmp;
vp->v_dev = vmp->m_dev;
}
/* Fill in filp objects */
fil_ptr0->filp_vno = vp;
fil_ptr1->filp_vno = vp;
fil_ptr0->filp_flags = O_RDONLY;
fil_ptr1->filp_flags = O_WRONLY;
m_out.reply_i1 = fil_des[0];
m_out.reply_i2 = fil_des[1];
return(OK);
}
/*===========================================================================*
* pipe_check *
*===========================================================================*/
PUBLIC int pipe_check(vp, rw_flag, oflags, bytes, position, canwrite, notouch)
register struct vnode *vp; /* the inode of the pipe */
int rw_flag; /* READING or WRITING */
int oflags; /* flags set by open or fcntl */
register int bytes; /* bytes to be read or written (all chunks) */
register off_t position; /* current file position */
int *canwrite; /* return: number of bytes we can write */
int notouch; /* check only */
{
/* Pipes are a little different. If a process reads from an empty pipe for
* which a writer still exists, suspend the reader. If the pipe is empty
* and there is no writer, return 0 bytes. If a process is writing to a
* pipe and no one is reading from it, give a broken pipe error.
*/
/* If reading, check for empty pipe. */
if (rw_flag == READING) {
if (position >= vp->v_size) {
/* Process is reading from an empty pipe. */
int r = 0;
if (find_filp(vp, W_BIT) != NIL_FILP) {
/* Writer exists */
if (oflags & O_NONBLOCK) {
r = EAGAIN;
} else {
if (!notouch)
suspend(XPIPE); /* block reader */
r = SUSPEND;
}
/* If need be, activate sleeping writers. */
if (susp_count > 0)
release(vp, WRITE, susp_count);
}
return(r);
}
} else {
/* Process is writing to a pipe. */
if (find_filp(vp, R_BIT) == NIL_FILP) {
/* Tell kernel to generate a SIGPIPE signal. */
if (!notouch) {
sys_kill(fp->fp_endpoint, SIGPIPE);
}
return(EPIPE);
}
if (position + bytes > PIPE_SIZE(vp->v_vmnt->m_block_size)) {
if ((oflags & O_NONBLOCK)
&& bytes <= PIPE_SIZE(vp->v_vmnt->m_block_size)) {
return(EAGAIN);
}
else if ((oflags & O_NONBLOCK)
&& bytes > PIPE_SIZE(vp->v_vmnt->m_block_size)) {
if ( (*canwrite = (PIPE_SIZE(vp->v_vmnt->m_block_size)
- position)) > 0) {
/* Do a partial write. Need to wakeup reader */
if (!notouch)
release(vp, READ, susp_count);
return(1);
} else {
return(EAGAIN);
}
}
if (bytes > PIPE_SIZE(vp->v_vmnt->m_block_size)) {
if ((*canwrite = PIPE_SIZE(vp->v_vmnt->m_block_size)
- position) > 0) {
/* Do a partial write. Need to wakeup reader
* since we'll suspend ourself in read_write()
*/
if (!notouch)
release(vp, READ, susp_count);
return(1);
}
}
if (!notouch)
suspend(XPIPE); /* stop writer -- pipe full */
return(SUSPEND);
}
/* Writing to an empty pipe. Search for suspended reader. */
if (position == 0 && !notouch)
release(vp, READ, susp_count);
}
*canwrite = 0;
return(1);
}
/*===========================================================================*
* suspend *
*===========================================================================*/
PUBLIC void suspend(task)
int task; /* who is proc waiting for? (PIPE = pipe) */
{
/* Take measures to suspend the processing of the present system call.
* Store the parameters to be used upon resuming in the process table.
* (Actually they are not used when a process is waiting for an I/O device,
* but they are needed for pipes, and it is not worth making the distinction.)
* The SUSPEND pseudo error should be returned after calling suspend().
*/
if (task == XPIPE || task == XPOPEN) susp_count++;/* #procs susp'ed on pipe*/
fp->fp_suspended = SUSPENDED;
assert(!GRANT_VALID(fp->fp_grant));
fp->fp_fd = m_in.fd << 8 | call_nr;
if(task == NONE)
panic(__FILE__,"suspend on NONE",NO_NUM);
fp->fp_task = -task;
if (task == XLOCK) {
fp->fp_buffer = (char *) m_in.name1; /* third arg to fcntl() */
fp->fp_nbytes = m_in.request; /* second arg to fcntl() */
} else {
fp->fp_buffer = m_in.buffer; /* for reads and writes */
fp->fp_nbytes = m_in.nbytes;
}
}
/*===========================================================================*
* unsuspend_by_endpt *
*===========================================================================*/
PUBLIC void unsuspend_by_endpt(int proc_e)
{
struct fproc *rp;
int client = 0;
/* Revive processes waiting for drivers (SUSPENDed) that have
* disappeared with return code EAGAIN.
*/
for (rp = &fproc[0]; rp < &fproc[NR_PROCS]; rp++, client++)
if(rp->fp_pid != PID_FREE &&
rp->fp_suspended == SUSPENDED && rp->fp_task == -proc_e) {
revive(rp->fp_endpoint, EAGAIN);
}
/* Revive processes waiting in drivers on select()s
* with EAGAIN too.
*/
select_unsuspend_by_endpt(proc_e);
return;
}
/*===========================================================================*
* release *
*===========================================================================*/
PUBLIC void release(vp, call_nr, count)
register struct vnode *vp; /* inode of pipe */
int call_nr; /* READ, WRITE, OPEN or CREAT */
int count; /* max number of processes to release */
{
/* Check to see if any process is hanging on the pipe whose inode is in 'ip'.
* If one is, and it was trying to perform the call indicated by 'call_nr',
* release it.
*/
register struct fproc *rp;
struct filp *f;
/* Trying to perform the call also includes SELECTing on it with that
* operation.
*/
if (call_nr == READ || call_nr == WRITE) {
int op;
if (call_nr == READ)
op = SEL_RD;
else
op = SEL_WR;
for(f = &filp[0]; f < &filp[NR_FILPS]; f++) {
if (f->filp_count < 1 || !(f->filp_pipe_select_ops & op) ||
f->filp_vno != vp)
continue;
select_callback(f, op);
f->filp_pipe_select_ops &= ~op;
}
}
/* Search the proc table. */
for (rp = &fproc[0]; rp < &fproc[NR_PROCS]; rp++) {
if (rp->fp_pid != PID_FREE && rp->fp_suspended == SUSPENDED &&
rp->fp_revived == NOT_REVIVING &&
(rp->fp_fd & BYTE) == call_nr &&
rp->fp_filp[rp->fp_fd>>8]->filp_vno == vp) {
revive(rp->fp_endpoint, 0);
susp_count--; /* keep track of who is suspended */
if (--count == 0) return;
}
}
}
/*===========================================================================*
* revive *
*===========================================================================*/
PUBLIC void revive(proc_nr_e, returned)
int proc_nr_e; /* process to revive */
int returned; /* if hanging on task, how many bytes read */
{
/* Revive a previously blocked process. When a process hangs on tty, this
* is the way it is eventually released.
*/
register struct fproc *rfp;
register int task;
int proc_nr;
if(isokendpt(proc_nr_e, &proc_nr) != OK)
return;
rfp = &fproc[proc_nr];
if (rfp->fp_suspended == NOT_SUSPENDED || rfp->fp_revived == REVIVING)return;
/* The 'reviving' flag only applies to pipes. Processes waiting for TTY get
* a message right away. The revival process is different for TTY and pipes.
* For select and TTY revival, the work is already done, for pipes it is not:
* the proc must be restarted so it can try again.
*/
task = -rfp->fp_task;
if (task == XPIPE || task == XLOCK) {
/* Revive a process suspended on a pipe or lock. */
rfp->fp_revived = REVIVING;
reviving++; /* process was waiting on pipe or lock */
} else {
rfp->fp_suspended = NOT_SUSPENDED;
if (task == XPOPEN) /* process blocked in open or create */
reply(proc_nr_e, rfp->fp_fd>>8);
else if (task == XSELECT) {
reply(proc_nr_e, returned);
} else {
/* Revive a process suspended on TTY or other device.
* Pretend it wants only what there is.
*/
rfp->fp_nbytes = returned;
/* If a grant has been issued by FS for this I/O, revoke
* it again now that I/O is done.
*/
if(GRANT_VALID(rfp->fp_grant)) {
if(cpf_revoke(rfp->fp_grant)) {
panic(__FILE__,"FS: revoke failed for grant",
rfp->fp_grant);
}
rfp->fp_grant = GRANT_INVALID;
}
reply(proc_nr_e, returned); /* unblock the process */
}
}
}
/*===========================================================================*
* do_unpause *
*===========================================================================*/
PUBLIC int do_unpause()
{
/* A signal has been sent to a user who is paused on the file system.
* Abort the system call with the EINTR error message.
*/
int proc_nr_e;
if (who_e != PM_PROC_NR) return(EPERM);
proc_nr_e = m_in.ENDPT;
return unpause(proc_nr_e);
}
/*===========================================================================*
* unpause *
*===========================================================================*/
PUBLIC int unpause(proc_nr_e)
int proc_nr_e;
{
/* A signal has been sent to a user who is paused on the file system.
* Abort the system call with the EINTR error message.
*/
register struct fproc *rfp;
int proc_nr_p, task, fild, status = EINTR;
struct filp *f;
dev_t dev;
message mess;
okendpt(proc_nr_e, &proc_nr_p);
rfp = &fproc[proc_nr_p];
if (rfp->fp_suspended == NOT_SUSPENDED) return(OK);
task = -rfp->fp_task;
if (rfp->fp_revived == REVIVING)
{
rfp->fp_revived = NOT_REVIVING;
reviving--;
}
switch (task) {
case XPIPE: /* process trying to read or write a pipe */
break;
case XLOCK: /* process trying to set a lock with FCNTL */
break;
case XSELECT: /* process blocking on select() */
select_forget(proc_nr_e);
break;
case XPOPEN: /* process trying to open a fifo */
break;
default: /* process trying to do device I/O (e.g. tty)*/
fild = (rfp->fp_fd >> 8) & BYTE;/* extract file descriptor */
if (fild < 0 || fild >= OPEN_MAX)
panic(__FILE__,"unpause err 2",NO_NUM);
f = rfp->fp_filp[fild];
dev = (dev_t) f->filp_vno->v_sdev; /* device hung on */
mess.TTY_LINE = (dev >> MINOR) & BYTE;
mess.IO_ENDPT = rfp->fp_ioproc;
mess.IO_GRANT = (char *) rfp->fp_grant;
/* Tell kernel R or W. Mode is from current call, not open. */
mess.COUNT = (rfp->fp_fd & BYTE) == READ ? R_BIT : W_BIT;
mess.m_type = CANCEL;
fp = rfp; /* hack - ctty_io uses fp */
(*dmap[(dev >> MAJOR) & BYTE].dmap_io)(task, &mess);
status = mess.REP_STATUS;
if(status == EAGAIN) status = EINTR;
if(GRANT_VALID(rfp->fp_grant)) {
if(cpf_revoke(rfp->fp_grant)) {
panic(__FILE__,"FS: revoke failed for grant (cancel)",
rfp->fp_grant);
}
rfp->fp_grant = GRANT_INVALID;
}
}
rfp->fp_suspended = NOT_SUSPENDED;
reply(proc_nr_e, status); /* signal interrupted call */
return(OK);
}
/*===========================================================================*
* select_request_pipe *
*===========================================================================*/
PUBLIC int select_request_pipe(struct filp *f, int *ops, int block)
{
int orig_ops, r = 0, err, canwrite;
orig_ops = *ops;
if ((*ops & (SEL_RD|SEL_ERR))) {
if ((err = pipe_check(f->filp_vno, READING, 0,
1, f->filp_pos, &canwrite, 1)) != SUSPEND)
r |= SEL_RD;
if (err < 0 && err != SUSPEND)
r |= SEL_ERR;
}
if ((*ops & (SEL_WR|SEL_ERR))) {
if ((err = pipe_check(f->filp_vno, WRITING, 0,
1, f->filp_pos, &canwrite, 1)) != SUSPEND)
r |= SEL_WR;
if (err < 0 && err != SUSPEND)
r |= SEL_ERR;
}
/* Some options we collected might not be requested. */
*ops = r & orig_ops;
if (!*ops && block) {
f->filp_pipe_select_ops |= orig_ops;
}
return SEL_OK;
}
/*===========================================================================*
* select_match_pipe *
*===========================================================================*/
PUBLIC int select_match_pipe(struct filp *f)
{
/* recognize either pipe or named pipe (FIFO) */
if (f && f->filp_vno && (f->filp_vno->v_mode & I_NAMED_PIPE))
return 1;
return 0;
}
Reference in a new issue View git blame Copy permalink