source: trunk/minix/servers/fs/pipe.c@ 9

Last change on this file since 9 was 9, checked in by Mattia Monga, 13 years ago

Minix 3.1.2a
File size: 13.3 KB
Line 
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 <minix/callnr.h>
22#include <minix/endpoint.h>
23#include <minix/com.h>
24#include <sys/select.h>
25#include <sys/time.h>
26#include "file.h"
27#include "fproc.h"
28#include "inode.h"
29#include "param.h"
30#include "super.h"
31#include "select.h"
32
33/*===========================================================================*
34 * do_pipe *
35 *===========================================================================*/
36PUBLIC int do_pipe()
37{
38/* Perform the pipe(fil_des) system call. */
39
40 register struct fproc *rfp;
41 register struct inode *rip;
42 int r;
43 struct filp *fil_ptr0, *fil_ptr1;
44 int fil_des[2]; /* reply goes here */
45
46 /* Acquire two file descriptors. */
47 rfp = fp;
48 if ( (r = get_fd(0, R_BIT, &fil_des[0], &fil_ptr0)) != OK) return(r);
49 rfp->fp_filp[fil_des[0]] = fil_ptr0;
50 FD_SET(fil_des[0], &rfp->fp_filp_inuse);
51 fil_ptr0->filp_count = 1;
52 if ( (r = get_fd(0, W_BIT, &fil_des[1], &fil_ptr1)) != OK) {
53 rfp->fp_filp[fil_des[0]] = NIL_FILP;
54 FD_CLR(fil_des[0], &rfp->fp_filp_inuse);
55 fil_ptr0->filp_count = 0;
56 return(r);
57 }
58 rfp->fp_filp[fil_des[1]] = fil_ptr1;
59 FD_SET(fil_des[1], &rfp->fp_filp_inuse);
60 fil_ptr1->filp_count = 1;
61
62 /* Make the inode on the pipe device. */
63 if ( (rip = alloc_inode(root_dev, I_REGULAR) ) == NIL_INODE) {
64 rfp->fp_filp[fil_des[0]] = NIL_FILP;
65 FD_CLR(fil_des[0], &rfp->fp_filp_inuse);
66 fil_ptr0->filp_count = 0;
67 rfp->fp_filp[fil_des[1]] = NIL_FILP;
68 FD_CLR(fil_des[1], &rfp->fp_filp_inuse);
69 fil_ptr1->filp_count = 0;
70 return(err_code);
71 }
72
73 if (read_only(rip) != OK)
74 panic(__FILE__,"pipe device is read only", NO_NUM);
75
76 rip->i_pipe = I_PIPE;
77 rip->i_mode &= ~I_REGULAR;
78 rip->i_mode |= I_NAMED_PIPE; /* pipes and FIFOs have this bit set */
79 fil_ptr0->filp_ino = rip;
80 fil_ptr0->filp_flags = O_RDONLY;
81 dup_inode(rip); /* for double usage */
82 fil_ptr1->filp_ino = rip;
83 fil_ptr1->filp_flags = O_WRONLY;
84 rw_inode(rip, WRITING); /* mark inode as allocated */
85 m_out.reply_i1 = fil_des[0];
86 m_out.reply_i2 = fil_des[1];
87 rip->i_update = ATIME | CTIME | MTIME;
88 return(OK);
89}
90
91/*===========================================================================*
92 * pipe_check *
93 *===========================================================================*/
94PUBLIC int pipe_check(rip, rw_flag, oflags, bytes, position, canwrite, notouch)
95register struct inode *rip; /* the inode of the pipe */
96int rw_flag; /* READING or WRITING */
97int oflags; /* flags set by open or fcntl */
98register int bytes; /* bytes to be read or written (all chunks) */
99register off_t position; /* current file position */
100int *canwrite; /* return: number of bytes we can write */
101int notouch; /* check only */
102{
103/* Pipes are a little different. If a process reads from an empty pipe for
104 * which a writer still exists, suspend the reader. If the pipe is empty
105 * and there is no writer, return 0 bytes. If a process is writing to a
106 * pipe and no one is reading from it, give a broken pipe error.
107 */
108
109 /* If reading, check for empty pipe. */
110 if (rw_flag == READING) {
111 if (position >= rip->i_size) {
112 /* Process is reading from an empty pipe. */
113 int r = 0;
114 if (find_filp(rip, W_BIT) != NIL_FILP) {
115 /* Writer exists */
116 if (oflags & O_NONBLOCK) {
117 r = EAGAIN;
118 } else {
119 if (!notouch)
120 suspend(XPIPE); /* block reader */
121 r = SUSPEND;
122 }
123 /* If need be, activate sleeping writers. */
124 if (susp_count > 0 && !notouch)
125 release(rip, WRITE, susp_count);
126 }
127 return(r);
128 }
129 } else {
130 /* Process is writing to a pipe. */
131 if (find_filp(rip, R_BIT) == NIL_FILP) {
132 /* Tell kernel to generate a SIGPIPE signal. */
133 if (!notouch) {
134 sys_kill(fp->fp_endpoint, SIGPIPE);
135 }
136 return(EPIPE);
137 }
138
139 if (position + bytes > PIPE_SIZE(rip->i_sp->s_block_size)) {
140 if ((oflags & O_NONBLOCK)
141 && bytes < PIPE_SIZE(rip->i_sp->s_block_size))
142 return(EAGAIN);
143 else if ((oflags & O_NONBLOCK)
144 && bytes > PIPE_SIZE(rip->i_sp->s_block_size)) {
145 if ( (*canwrite = (PIPE_SIZE(rip->i_sp->s_block_size)
146 - position)) > 0) {
147 /* Do a partial write. Need to wakeup reader */
148 if (!notouch)
149 release(rip, READ, susp_count);
150 return(1);
151 } else {
152 return(EAGAIN);
153 }
154 }
155 if (bytes > PIPE_SIZE(rip->i_sp->s_block_size)) {
156 if ((*canwrite = PIPE_SIZE(rip->i_sp->s_block_size)
157 - position) > 0) {
158 /* Do a partial write. Need to wakeup reader
159 * since we'll suspend ourself in read_write()
160 */
161 if (!notouch)
162 release(rip, READ, susp_count);
163 return(1);
164 }
165 }
166 if (!notouch)
167 suspend(XPIPE); /* stop writer -- pipe full */
168 return(SUSPEND);
169 }
170
171 /* Writing to an empty pipe. Search for suspended reader. */
172 if (position == 0 && !notouch)
173 release(rip, READ, susp_count);
174 }
175
176 *canwrite = 0;
177 return(1);
178}
179
180/*===========================================================================*
181 * suspend *
182 *===========================================================================*/
183PUBLIC void suspend(task)
184int task; /* who is proc waiting for? (PIPE = pipe) */
185{
186/* Take measures to suspend the processing of the present system call.
187 * Store the parameters to be used upon resuming in the process table.
188 * (Actually they are not used when a process is waiting for an I/O device,
189 * but they are needed for pipes, and it is not worth making the distinction.)
190 * The SUSPEND pseudo error should be returned after calling suspend().
191 */
192
193 if (task == XPIPE || task == XPOPEN) susp_count++;/* #procs susp'ed on pipe*/
194 fp->fp_suspended = SUSPENDED;
195 fp->fp_fd = m_in.fd << 8 | call_nr;
196 if(task == NONE)
197 panic(__FILE__,"suspend on NONE",NO_NUM);
198 fp->fp_task = -task;
199 if (task == XLOCK) {
200 fp->fp_buffer = (char *) m_in.name1; /* third arg to fcntl() */
201 fp->fp_nbytes = m_in.request; /* second arg to fcntl() */
202 } else {
203 fp->fp_buffer = m_in.buffer; /* for reads and writes */
204 fp->fp_nbytes = m_in.nbytes;
205 }
206}
207
208/*===========================================================================*
209 * unsuspend_by_endpt *
210 *===========================================================================*/
211PUBLIC void unsuspend_by_endpt(int proc_e)
212{
213 struct fproc *rp;
214 int client = 0;
215
216 /* Revive processes waiting for drivers (SUSPENDed) that have
217 * disappeared with return code EAGAIN.
218 */
219 for (rp = &fproc[0]; rp < &fproc[NR_PROCS]; rp++, client++)
220 if(rp->fp_pid != PID_FREE &&
221 rp->fp_suspended == SUSPENDED && rp->fp_task == -proc_e) {
222 revive(rp->fp_endpoint, EAGAIN);
223 }
224
225 /* Revive processes waiting in drivers on select()s
226 * with EAGAIN too.
227 */
228 select_unsuspend_by_endpt(proc_e);
229
230 return;
231}
232
233
234/*===========================================================================*
235 * release *
236 *===========================================================================*/
237PUBLIC void release(ip, call_nr, count)
238register struct inode *ip; /* inode of pipe */
239int call_nr; /* READ, WRITE, OPEN or CREAT */
240int count; /* max number of processes to release */
241{
242/* Check to see if any process is hanging on the pipe whose inode is in 'ip'.
243 * If one is, and it was trying to perform the call indicated by 'call_nr',
244 * release it.
245 */
246
247 register struct fproc *rp;
248 struct filp *f;
249
250 /* Trying to perform the call also includes SELECTing on it with that
251 * operation.
252 */
253 if (call_nr == READ || call_nr == WRITE) {
254 int op;
255 if (call_nr == READ)
256 op = SEL_RD;
257 else
258 op = SEL_WR;
259 for(f = &filp[0]; f < &filp[NR_FILPS]; f++) {
260 if (f->filp_count < 1 || !(f->filp_pipe_select_ops & op) ||
261 f->filp_ino != ip)
262 continue;
263 select_callback(f, op);
264 f->filp_pipe_select_ops &= ~op;
265 }
266 }
267
268 /* Search the proc table. */
269 for (rp = &fproc[0]; rp < &fproc[NR_PROCS]; rp++) {
270 if (rp->fp_pid != PID_FREE && rp->fp_suspended == SUSPENDED &&
271 rp->fp_revived == NOT_REVIVING &&
272 (rp->fp_fd & BYTE) == call_nr &&
273 rp->fp_filp[rp->fp_fd>>8]->filp_ino == ip) {
274 revive(rp->fp_endpoint, 0);
275 susp_count--; /* keep track of who is suspended */
276 if (--count == 0) return;
277 }
278 }
279}
280
281/*===========================================================================*
282 * revive *
283 *===========================================================================*/
284PUBLIC void revive(proc_nr_e, returned)
285int proc_nr_e; /* process to revive */
286int returned; /* if hanging on task, how many bytes read */
287{
288/* Revive a previously blocked process. When a process hangs on tty, this
289 * is the way it is eventually released.
290 */
291
292 register struct fproc *rfp;
293 register int task;
294 int proc_nr;
295
296 if(isokendpt(proc_nr_e, &proc_nr) != OK)
297 return;
298
299 rfp = &fproc[proc_nr];
300 if (rfp->fp_suspended == NOT_SUSPENDED || rfp->fp_revived == REVIVING)return;
301
302 /* The 'reviving' flag only applies to pipes. Processes waiting for TTY get
303 * a message right away. The revival process is different for TTY and pipes.
304 * For select and TTY revival, the work is already done, for pipes it is not:
305 * the proc must be restarted so it can try again.
306 */
307 task = -rfp->fp_task;
308 if (task == XPIPE || task == XLOCK) {
309 /* Revive a process suspended on a pipe or lock. */
310 rfp->fp_revived = REVIVING;
311 reviving++; /* process was waiting on pipe or lock */
312 } else {
313 rfp->fp_suspended = NOT_SUSPENDED;
314 if (task == XPOPEN) /* process blocked in open or create */
315 reply(proc_nr_e, rfp->fp_fd>>8);
316 else if (task == XSELECT) {
317 reply(proc_nr_e, returned);
318 } else {
319 /* Revive a process suspended on TTY or other device. */
320 rfp->fp_nbytes = returned; /*pretend it wants only what there is*/
321 reply(proc_nr_e, returned); /* unblock the process */
322 }
323 }
324}
325
326/*===========================================================================*
327 * do_unpause *
328 *===========================================================================*/
329PUBLIC int do_unpause()
330{
331/* A signal has been sent to a user who is paused on the file system.
332 * Abort the system call with the EINTR error message.
333 */
334
335 register struct fproc *rfp;
336 int proc_nr_e, proc_nr_p, task, fild;
337 struct filp *f;
338 dev_t dev;
339 message mess;
340
341 if (who_e != PM_PROC_NR) return(EPERM);
342 proc_nr_e = m_in.ENDPT;
343 okendpt(proc_nr_e, &proc_nr_p);
344 rfp = &fproc[proc_nr_p];
345 if (rfp->fp_suspended == NOT_SUSPENDED) return(OK);
346 task = -rfp->fp_task;
347
348 switch (task) {
349 case XPIPE: /* process trying to read or write a pipe */
350 break;
351
352 case XLOCK: /* process trying to set a lock with FCNTL */
353 break;
354
355 case XSELECT: /* process blocking on select() */
356 select_forget(proc_nr_e);
357 break;
358
359 case XPOPEN: /* process trying to open a fifo */
360 break;
361
362 default: /* process trying to do device I/O (e.g. tty)*/
363 fild = (rfp->fp_fd >> 8) & BYTE;/* extract file descriptor */
364 if (fild < 0 || fild >= OPEN_MAX)
365 panic(__FILE__,"unpause err 2",NO_NUM);
366 f = rfp->fp_filp[fild];
367 dev = (dev_t) f->filp_ino->i_zone[0]; /* device hung on */
368 mess.TTY_LINE = (dev >> MINOR) & BYTE;
369 mess.IO_ENDPT = proc_nr_e;
370
371 /* Tell kernel R or W. Mode is from current call, not open. */
372 mess.COUNT = (rfp->fp_fd & BYTE) == READ ? R_BIT : W_BIT;
373 mess.m_type = CANCEL;
374 fp = rfp; /* hack - ctty_io uses fp */
375 (*dmap[(dev >> MAJOR) & BYTE].dmap_io)(task, &mess);
376 }
377
378 rfp->fp_suspended = NOT_SUSPENDED;
379 reply(proc_nr_e, EINTR); /* signal interrupted call */
380 return(OK);
381}
382
383/*===========================================================================*
384 * select_request_pipe *
385 *===========================================================================*/
386PUBLIC int select_request_pipe(struct filp *f, int *ops, int block)
387{
388 int orig_ops, r = 0, err, canwrite;
389 orig_ops = *ops;
390 if ((*ops & (SEL_RD|SEL_ERR))) {
391 if ((err = pipe_check(f->filp_ino, READING, 0,
392 1, f->filp_pos, &canwrite, 1)) != SUSPEND && err > 0)
393 r |= SEL_RD;
394 if (err < 0 && err != SUSPEND)
395 r |= SEL_ERR;
396 }
397 if ((*ops & (SEL_WR|SEL_ERR))) {
398 if ((err = pipe_check(f->filp_ino, WRITING, 0,
399 1, f->filp_pos, &canwrite, 1)) != SUSPEND &&
400 err > 0 && canwrite > 0)
401 r |= SEL_WR;
402 if (err < 0 && err != SUSPEND)
403 r |= SEL_ERR;
404 }
405
406 /* Some options we collected might not be requested. */
407 *ops = r & orig_ops;
408
409 if (!*ops && block) {
410 f->filp_pipe_select_ops |= orig_ops;
411 }
412
413 return SEL_OK;
414}
415
416/*===========================================================================*
417 * select_match_pipe *
418 *===========================================================================*/
419PUBLIC int select_match_pipe(struct filp *f)
420{
421 /* recognize either pipe or named pipe (FIFO) */
422 if (f && f->filp_ino && (f->filp_ino->i_mode & I_NAMED_PIPE))
423 return 1;
424 return 0;
425}
426
Note: See TracBrowser for help on using the repository browser.