ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/radiance/ray/src/util/netproc.c
Revision: 2.1
Committed: Thu Jan 18 11:24:53 1996 UTC (28 years, 2 months ago) by greg
Content type: text/plain
Branch: MAIN
Log Message:
Initial revision

File Contents

# Content
1 /* Copyright (c) 1996 Regents of the University of California */
2
3 #ifndef lint
4 static char SCCSid[] = "$SunId$ LBL";
5 #endif
6
7 /*
8 * Parallel network process handling routines
9 */
10
11 #include <stdio.h>
12 #include <sys/types.h>
13 #include <signal.h>
14 #include <fcntl.h>
15 #include "netproc.h"
16 #include "vfork.h"
17 /* Remote shell location */
18 #ifdef sgi
19 #define RSH "/usr/bsd/rsh"
20 #endif
21 #ifdef _AUX_SOURCE
22 #define RSH "/usr/bin/remsh"
23 #endif
24 #ifndef RSH
25 #define RSH "/usr/ucb/rsh"
26 #endif
27 /* select call compatibility stuff */
28 #ifndef FD_SETSIZE
29 #include <sys/param.h>
30 #define FD_SETSIZE NOFILE /* maximum # select file descriptors */
31 #endif
32 #ifndef FD_SET
33 #ifndef NFDBITS
34 #define NFDBITS (8*sizeof(int)) /* number of bits per fd_mask */
35 #endif
36 #define FD_SET(n, p) ((p)->fds_bits[(n)/NFDBITS] |= (1 << ((n) % NFDBITS)))
37 #define FD_CLR(n, p) ((p)->fds_bits[(n)/NFDBITS] &= ~(1 << ((n) % NFDBITS)))
38 #define FD_ISSET(n, p) ((p)->fds_bits[(n)/NFDBITS] & (1 << ((n) % NFDBITS)))
39 #ifndef BSD
40 #define bzero(d,n) (void)memset(d,0,n)
41 #endif
42 #define FD_ZERO(p) bzero((char *)(p), sizeof(*(p)))
43 #endif
44
45 PSERVER *pslist = NULL; /* global process server list */
46
47 static PROC *pindex[FD_SETSIZE]; /* process index table */
48
49 static char ourhost[64]; /* this host name */
50 static char ourdir[128]; /* our working directory */
51 static char ouruser[32]; /* our user name */
52 static char *ourshell; /* our user's shell */
53
54 static fd_set errdesc; /* error file descriptors */
55 static int maxfd; /* maximum assigned descriptor */
56
57 extern char *malloc(), *realloc();
58 extern char *getenv();
59
60
61 PSERVER *
62 addpserver(host, dir, usr, np) /* add a new process server */
63 char *host, *dir, *usr;
64 int np;
65 {
66 register PSERVER *ps;
67 /* allocate the struct */
68 if (np < 1)
69 return(NULL);
70 ps = (PSERVER *)malloc(sizeof(PSERVER)+(np-1)*sizeof(PROC));
71 if (ps == NULL)
72 return(NULL);
73 if (!ourhost[0]) { /* initialize */
74 char dirtmp[128];
75 register char *cp;
76 register int len;
77
78 gethostname(ourhost, sizeof(ourhost));
79 getcwd(dirtmp, sizeof(dirtmp));
80 if ((cp = getenv("HOME")) != NULL) {
81 if (!strcmp(cp, dirtmp))
82 ourdir[0] = '\0';
83 else if (!strncmp(cp, dirtmp, len=strlen(cp)) &&
84 dirtmp[len] == '/')
85 strcpy(ourdir, dirtmp+len+1);
86 else
87 strcpy(ourdir, dirtmp);
88 } else
89 strcpy(ourdir, dirtmp);
90 if ((cp = getenv("USER")) != NULL)
91 strcpy(ouruser, cp);
92 if ((ourshell = getenv("SHELL")) == NULL)
93 ourshell = "/bin/sh";
94 FD_ZERO(&errdesc);
95 maxfd = -1;
96 }
97 /* assign host, directory, user */
98 if (host == NULL || !strcmp(host, ourhost) ||
99 !strcmp(host, LHOSTNAME))
100 ps->hostname[0] = '\0';
101 else
102 strcpy(ps->hostname, host);
103 if (dir == NULL)
104 strcpy(ps->directory, ourdir);
105 else
106 strcpy(ps->directory, dir);
107 if (usr == NULL || !strcmp(usr, ouruser))
108 ps->username[0] = '\0';
109 else
110 strcpy(ps->username, usr);
111 /* clear process slots */
112 ps->nprocs = np;
113 while (np--) {
114 ps->proc[np].com = NULL;
115 ps->proc[np].pid = -1;
116 ps->proc[np].efd = -1;
117 ps->proc[np].errs = NULL;
118 ps->proc[np].elen = 0;
119 ps->proc[np].cf = NULL;
120 }
121 /* insert in our list */
122 ps->next = pslist;
123 pslist = ps;
124 /* check for signs of life */
125 if (!pserverOK(ps)) {
126 delpserver(ps); /* failure -- abort */
127 return(NULL);
128 }
129 return(ps);
130 }
131
132
133 delpserver(ps) /* delete a process server */
134 PSERVER *ps;
135 {
136 PSERVER pstart;
137 register PSERVER *psp;
138 register int i;
139 /* find server in our list */
140 pstart.next = pslist;
141 for (psp = &pstart; ps != psp->next; psp = psp->next)
142 if (psp->next == NULL)
143 return; /* not in our list! */
144 /* kill any running jobs */
145 for (i = 0; i < ps->nprocs; i++)
146 if (ps->proc[i].com != NULL) {
147 kill(SIGTERM, ps->proc[i].pid);
148 wait4job(ps, ps->proc[i].pid);
149 }
150 /* remove server from list */
151 psp->next = ps->next;
152 pslist = pstart.next;
153 free((char *)ps); /* free associated memory */
154 }
155
156
157 PSERVER *
158 findjob(pnp) /* find out where process is running */
159 register int *pnp; /* modified */
160 {
161 register PSERVER *ps;
162 register int i;
163
164 for (ps = pslist; ps != NULL; ps = ps->next)
165 for (i = 0; i < ps->nprocs; i++)
166 if (ps->proc[i].pid == *pnp) {
167 *pnp = i;
168 return(ps);
169 }
170 return(NULL); /* not found */
171 }
172
173
174 int
175 startjob(ps, command, compf) /* start a job on a process server */
176 register PSERVER *ps;
177 char *command;
178 int (*compf)();
179 {
180 char *av[12];
181 int pfd[2], pid;
182 register int i;
183
184 if (ps == NULL) { /* find a server */
185 for (ps = pslist; ps != NULL; ps = ps->next)
186 if ((i = startjob(ps, command, compf)) != -1)
187 return(i); /* got one */
188 return(-1); /* no slots anywhere */
189 }
190 for (i = 0; i < ps->nprocs; i++)
191 if (ps->proc[i].com == NULL)
192 break;
193 if (i >= ps->nprocs)
194 return(-1); /* out of process slots */
195 /* open pipe */
196 if (pipe(pfd) < 0) {
197 perror("cannot open pipe");
198 exit(1);
199 }
200 /* start child process */
201 if ((pid = vfork()) == 0) {
202 close(pfd[0]); /* connect stderr to pipe */
203 if (pfd[1] != 2) {
204 dup2(pfd[1], 2);
205 close(pfd[1]);
206 }
207 if (ps->hostname[0]) { /* rsh command */
208 av[i=0] = RSH;
209 av[++i] = ps->hostname;
210 av[++i] = "-n"; /* no stdin */
211 if (ps->username[0]) {
212 av[++i] = "-l";
213 av[++i] = ps->username;
214 }
215 if (ps->directory[0]) {
216 av[++i] = "cd";
217 av[++i] = ps->directory;
218 av[++i] = ";";
219 }
220 av[++i] = command;
221 av[++i] = NULL;
222 } else { /* shell command */
223 av[0] = ourshell;
224 av[1] = "-c";
225 av[2] = command;
226 av[3] = NULL;
227 }
228 execv(av[0], av);
229 _exit(1);
230 }
231 if (pid == -1) {
232 perror("fork failed");
233 exit(1);
234 }
235 ps->proc[i].com = command; /* assign process slot */
236 ps->proc[i].cf = compf;
237 ps->proc[i].pid = pid;
238 close(pfd[1]); /* get piped stderr file descriptor */
239 ps->proc[i].efd = pfd[0];
240 fcntl(pfd[0], F_SETFD, 1); /* set close on exec flag */
241 pindex[pfd[0]] = ps->proc + i; /* assign error fd index */
242 FD_SET(pfd[0], &errdesc); /* add to select call parameter */
243 if (pfd[0] > maxfd)
244 maxfd = pfd[0];
245 return(pid); /* return to parent process */
246 }
247
248
249 static int
250 readerrs(fd) /* read error output from fd */
251 int fd;
252 {
253 char errbuf[BUFSIZ];
254 int nr;
255 register PROC *pp;
256 /* look up associated process */
257 if ((pp = pindex[fd]) == NULL)
258 abort(); /* serious consistency error */
259 nr = read(fd, errbuf, BUFSIZ-1);
260 if (nr < 0) {
261 perror("read error");
262 exit(1);
263 }
264 if (nr == 0) /* stream closed (process finished) */
265 return(0);
266 errbuf[nr] = '\0'; /* add to error buffer */
267 if (pp->elen == 0)
268 pp->errs = (char *)malloc(nr+1);
269 else
270 pp->errs = (char *)realloc(pp->errs, pp->elen+nr+1);
271 if (pp->errs == NULL) {
272 perror("malloc failed");
273 exit(1);
274 }
275 strcpy(pp->errs+pp->elen, errbuf);
276 pp->elen += nr;
277 return(nr);
278 }
279
280
281 static
282 wait4end() /* read error streams until someone is done */
283 {
284 fd_set readfds, excepfds;
285 register int i;
286 /* find end of descriptor set */
287 for ( ; maxfd >= 0; maxfd--)
288 if (FD_ISSET(maxfd, &errdesc))
289 break;
290 if (maxfd < 0)
291 return; /* nothing to read */
292 readfds = excepfds = errdesc;
293 while (select(maxfd+1, &readfds, NULL, &excepfds, NULL) > 0)
294 for (i = 0; i <= maxfd; i++) /* get pending i/o */
295 if (FD_ISSET(i, &readfds) || FD_ISSET(i, &excepfds))
296 if (readerrs(i) == 0)
297 return; /* finished process */
298 perror("select call failed");
299 exit(1);
300 }
301
302
303 static int
304 finishjob(ps, pn, status) /* clean up finished process */
305 PSERVER *ps;
306 int pn;
307 int status;
308 {
309 register PROC *pp;
310
311 pp = ps->proc + pn;
312 if (pp->cf != NULL) /* client cleanup */
313 status = (*pp->cf)(ps, pn, status);
314 close(pp->efd); /* close error stream */
315 pindex[pp->efd] = NULL;
316 FD_CLR(pp->efd, &errdesc);
317 free((char *)pp->errs);
318 pp->com = NULL; /* clear settings */
319 pp->pid = -1;
320 pp->efd = -1;
321 pp->errs = NULL;
322 pp->elen = 0;
323 pp->cf = NULL;
324 return(status);
325 }
326
327
328 int
329 wait4job(ps, pid) /* wait for process to finish */
330 PSERVER *ps;
331 int pid;
332 {
333 int status, psn, psn2;
334 PSERVER *ps2;
335
336 if (pid == -1) { /* wait for first job */
337 if (ps != NULL) {
338 for (psn = ps->nprocs; psn--; )
339 if (ps->proc[psn].com != NULL)
340 break;
341 if (psn < 0)
342 return(-1); /* no processes this server */
343 }
344 do {
345 wait4end(); /* wait for something to end */
346 if ((psn2 = wait(&status)) == -1)
347 return(-1); /* none left */
348 ps2 = findjob(&psn2);
349 if (ps2 != NULL) /* clean up job if ours */
350 status = finishjob(ps2, psn2, status);
351 } while (ps2 == NULL || (ps != NULL && ps2 != ps));
352 return(status); /* return job status */
353 }
354 psn = pid; /* else find specific job */
355 ps2 = findjob(&psn); /* find process slot */
356 if (ps2 == NULL || (ps != NULL && ps2 != ps))
357 return(-1); /* inconsistent target */
358 ps = ps2;
359 do {
360 wait4end(); /* wait for something to end */
361 if ((psn2 = wait(&status)) == -1)
362 return(-1); /* none left */
363 ps2 = findjob(&psn2);
364 if (ps2 != NULL) /* clean up job if ours */
365 status = finishjob(ps2, psn2, status);
366 } while (ps2 != ps || psn2 != psn);
367 return(status); /* return job status */
368 }