source: branches/minix3-book/servers/fs/pipe.c@ 11

Last change on this file since 11 was 4, checked in by Mattia Monga, 14 years ago

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