src/util/netproc.c

#ifndef lint
static const char       RCSid[] = "$Id: netproc.c,v 2.10 2003/04/23 00:52:34 greg Exp $";
#endif
/*
 * Parallel network process handling routines
 */

#include <stdio.h>
#include <signal.h>
#include <fcntl.h>
#include <unistd.h>

#include "selcall.h"
#include "netproc.h"
#include "paths.h"
#include "vfork.h"

PSERVER *pslist = NULL;         /* global process server list */

static PROC     *pindex[FD_SETSIZE];    /* process index table */

static char     ourhost[64];    /* this host name */
static char     ourdir[PATH_MAX];       /* our working directory */
static char     ouruser[32];    /* our user name */
static char     *ourshell;      /* our user's shell */

static fd_set   errdesc;        /* error file descriptors */
static int      maxfd;          /* maximum assigned descriptor */

extern char     *remsh;         /* externally defined remote shell program */

extern char     *getenv();


PSERVER *
addpserver(host, dir, usr, np)          /* add a new process server */
char    *host, *dir, *usr;
int     np;
{
        register PSERVER        *ps;
                                        /* allocate the struct */
        if (np < 1)
                return(NULL);
        ps = (PSERVER *)malloc(sizeof(PSERVER)+(np-1)*sizeof(PROC));
        if (ps == NULL)
                return(NULL);
        if (!ourhost[0]) {              /* initialize */
                char    dirtmp[PATH_MAX];
                register char   *cp;
                register int    len;

                strcpy(ourhost, myhostname());
                getcwd(dirtmp, sizeof(dirtmp));
                if ((cp = getenv("HOME")) != NULL) {
                        if (!strcmp(cp, dirtmp))
                                ourdir[0] = '\0';
                        else if (!strncmp(cp, dirtmp, len=strlen(cp)) &&
                                        dirtmp[len] == '/')
                                strcpy(ourdir, dirtmp+len+1);
                        else
                                strcpy(ourdir, dirtmp);
                } else
                        strcpy(ourdir, dirtmp);
                if ((cp = getenv("USER")) != NULL)
                        strcpy(ouruser, cp);
                if ((ourshell = getenv("SHELL")) == NULL)
                        ourshell = "/bin/sh";
                FD_ZERO(&errdesc);
                maxfd = -1;
        }
                                        /* assign host, directory, user */
        if (host == NULL || !strcmp(host, ourhost) ||
                        !strcmp(host, LHOSTNAME))
                ps->hostname[0] = '\0';
        else
                strcpy(ps->hostname, host);
        if (dir == NULL)
                strcpy(ps->directory, ourdir);
        else
                strcpy(ps->directory, dir);
        if (usr == NULL || !strcmp(usr, ouruser))
                ps->username[0] = '\0';
        else
                strcpy(ps->username, usr);
                                        /* clear process slots */
        ps->nprocs = np;
        while (np--) {
                ps->proc[np].com = NULL;
                ps->proc[np].pid = -1;
                ps->proc[np].efd = -1;
                ps->proc[np].errs = NULL;
                ps->proc[np].elen = 0;
                ps->proc[np].cf = NULL;
        }
                                        /* insert in our list */
        ps->next = pslist;
        pslist = ps;
                                        /* check for signs of life */
        if (!pserverOK(ps)) {
                delpserver(ps);                 /* failure -- abort */
                return(NULL);
        }
        return(ps);
}


delpserver(ps)                          /* delete a process server */
PSERVER *ps;
{
        PSERVER pstart;
        register PSERVER        *psp;
        register int    i;
                                        /* find server in our list */
        pstart.next = pslist;
        for (psp = &pstart; ps != psp->next; psp = psp->next)
                if (psp->next == NULL)
                        return;                 /* not in our list! */
                                        /* kill any running jobs */
        for (i = 0; i < ps->nprocs; i++)
                if (ps->proc[i].com != NULL) {
                        kill(SIGTERM, ps->proc[i].pid);
                        wait4job(ps, ps->proc[i].pid);
                }
                                        /* remove server from list */
        psp->next = ps->next;
        pslist = pstart.next;
        free((void *)ps);               /* free associated memory */
}


PSERVER *
findjob(pnp)                    /* find out where process is running */
register int    *pnp;   /* modified */
{
        register PSERVER        *ps;
        register int    i;

        for (ps = pslist; ps != NULL; ps = ps->next)
                for (i = 0; i < ps->nprocs; i++)
                        if (ps->proc[i].pid == *pnp) {
                                *pnp = i;
                                return(ps);
                        }
        return(NULL);           /* not found */
}


int
startjob(ps, command, compf)    /* start a job on a process server */
register PSERVER        *ps;
char    *command;
int     (*compf)();
{
        char    udirt[PATH_MAX];
        char    *av[16];
        int     pfd[2], pid;
        register int    i;

        if (ps == NULL) {               /* find a server */
                for (ps = pslist; ps != NULL; ps = ps->next)
                        if ((i = startjob(ps, command, compf)) != -1)
                                return(i);              /* got one */
                return(-1);     /* no slots anywhere */
        }
        for (i = 0; i < ps->nprocs; i++)
                if (ps->proc[i].com == NULL)
                        break;
        if (i >= ps->nprocs)
                return(-1);             /* out of process slots */
                                        /* open pipe */
        if (pipe(pfd) < 0) {
                perror("cannot open pipe");
                exit(1);
        }
                                        /* start child process */
        if ((pid = vfork()) == 0) {
                close(pfd[0]);                  /* connect stderr to pipe */
                if (pfd[1] != 2) {
                        dup2(pfd[1], 2);
                        close(pfd[1]);
                }
                if (ps->hostname[0]) {          /* rsh command */
                        av[i=0] = remsh;
                        av[++i] = ps->hostname;
                        av[++i] = "-n";                 /* no stdin */
                        if (ps->username[0]) {          /* different user */
                                av[++i] = "-l";
                                av[++i] = ps->username;
                                av[++i] = "cd";
                                udirt[0] = '~';
                                strcpy(udirt+1, ouruser);
                                av[++i] = udirt;
                                av[++i] = ";";
                        }
                        if (ps->directory[0]) {         /* change directory */
                                av[++i] = "cd";
                                av[++i] = ps->directory;
                                av[++i] = ";";
                        }
                        av[++i] = command;
                        av[++i] = NULL;
                } else {                        /* shell command */
                        av[0] = ourshell;
                        av[1] = "-c";
                        av[2] = command;
                        av[3] = NULL;
                }
                execv(av[0], av);
                _exit(1);
        }
        if (pid == -1) {
                perror("fork failed");
                exit(1);
        }
        ps->proc[i].com = command;      /* assign process slot */
        ps->proc[i].cf = compf;
        ps->proc[i].pid = pid;
        close(pfd[1]);                  /* get piped stderr file descriptor */
        ps->proc[i].efd = pfd[0];
        fcntl(pfd[0], F_SETFD, 1);      /* set close on exec flag */
        pindex[pfd[0]] = ps->proc + i;  /* assign error fd index */
        FD_SET(pfd[0], &errdesc);       /* add to select call parameter */
        if (pfd[0] > maxfd)
                maxfd = pfd[0];
        return(pid);                    /* return to parent process */
}


static int
readerrs(fd)                    /* read error output from fd */
int     fd;
{
        char    errbuf[BUFSIZ];
        int     nr;
        register PROC   *pp;
                                /* look up associated process */
        if ((pp = pindex[fd]) == NULL)
                abort();                /* serious consistency error */
        nr = read(fd, errbuf, BUFSIZ-1);
        if (nr < 0) {
                perror("read error");
                exit(1);
        }
        if (nr == 0)            /* stream closed (process finished) */
                return(0);
        errbuf[nr] = '\0';      /* add to error buffer */
        if (pp->elen == 0)
                pp->errs = (char *)malloc(nr+1);
        else
                pp->errs = (char *)realloc((void *)pp->errs, pp->elen+nr+1);
        if (pp->errs == NULL) {
                perror("malloc failed");
                exit(1);
        }
        strcpy(pp->errs+pp->elen, errbuf);
        pp->elen += nr;
        return(nr);
}


static
wait4end()                      /* read error streams until someone is done */
{
        fd_set  readfds, excepfds;
        register int    i;
                                /* find end of descriptor set */
        for ( ; maxfd >= 0; maxfd--)
                if (FD_ISSET(maxfd, &errdesc))
                        break;
        if (maxfd < 0)
                return;         /* nothing to read */
        readfds = excepfds = errdesc;
        while (select(maxfd+1, &readfds, NULL, &excepfds, NULL) > 0)
                for (i = 0; i <= maxfd; i++)            /* get pending i/o */
                        if (FD_ISSET(i, &readfds) || FD_ISSET(i, &excepfds))
                                if (readerrs(i) == 0)
                                        return;         /* finished process */
        perror("select call failed");
        exit(1);
}


static int
finishjob(ps, pn, status)       /* clean up finished process */
PSERVER *ps;
int     pn;
int     status;
{
        register PROC   *pp;

        pp = ps->proc + pn;
        if (pp->cf != NULL)                     /* client cleanup */
                status = (*pp->cf)(ps, pn, status);
        close(pp->efd);                         /* close error stream */
        pindex[pp->efd] = NULL;
        FD_CLR(pp->efd, &errdesc);
        free((void *)pp->errs);
        pp->com = NULL;                         /* clear settings */
        pp->pid = -1;
        pp->efd = -1;
        pp->errs = NULL;
        pp->elen = 0;
        pp->cf = NULL;
        return(status);
}


int
wait4job(ps, pid)               /* wait for process to finish */
PSERVER *ps;
int     pid;
{
        int     status, psn, psn2;
        PSERVER *ps2;

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