path: root/sys/src/cmd/upas/q/runq.c

#include "common.h"
#include <ctype.h>

typedef struct Job Job;
typedef struct Wdir Wdir;

struct Wdir {
	Dir	*d;
	int	nd;
	char	*name;
};

struct Job {
	Job	*next;
	int	pid;
	int	ac;
	int	dfd;
	char	**av;
	char	*buf;	/* backing for av */
	Wdir	*wdir;	/* work dir */
	Dir	*dp;	/* not owned */
	Mlock	*l;
	Biobuf	*b;
};

void	doalldirs(void);
void	dodir(char*);
Job*	dofile(Wdir*, Dir*);
Job*	donefile(Job*, Waitmsg*);
void	freejob(Job*);
void	rundir(char*);
char*	file(char*, char);
void	warning(char*, void*);
void	error(char*, void*);
int	returnmail(char**, Wdir*, char*, char*);
void	logit(char*, Wdir*, char*, char**);
void	doload(int);

char	*cmd;
char	*root;
int	debug;
int	giveup = 2*24*60*60;
int	limit;

char *runqlog = "runq";

char	**badsys;		/* array of recalcitrant systems */
int	nbad;
int	njob = 1;		/* number of concurrent jobs to invoke */
int	Eflag;			/* ignore E.xxxxxx dates */
int	Rflag;			/* no giving up, ever */
int	aflag;			/* do all dirs */

void
usage(void)
{
	fprint(2, "usage: runq [-dE] [-q dir] [-l load] [-t time] [-r nfiles] [-n nprocs] q-root cmd\n");
	exits("usage");
}

void
main(int argc, char **argv)
{
	char *qdir;

	qdir = 0;

	ARGBEGIN{
	case 'E':
		Eflag++;
		break;
	case 'R':	/* no giving up -- just leave stuff in the queue */
		Rflag++;
		break;
	case 'd':
		debug++;
		break;
	case 'r':
		limit = atoi(EARGF(usage()));
		break;
	case 't':
		giveup = 60*60*atoi(EARGF(usage()));
		break;
	case 'q':
		qdir = EARGF(usage());
		break;
	case 'a':
		aflag++;
		break;
	case 'n':
		njob = atoi(EARGF(usage()));
		if(njob == 0)
			usage();
		break;
	default:
		usage();
		break;
	}ARGEND;

	if(argc != 2)
		usage();

	if(!aflag && qdir == nil){
		qdir = getuser();
		if(qdir == nil)
			error("unknown user", 0);
	}
	root = argv[0];
	cmd = argv[1];

	if(chdir(root) < 0)
		error("can't cd to %s", root);

	if(aflag)
		doalldirs();
	else
		dodir(qdir);
	exits(0);
}

int
emptydir(char *name)
{
	int fd;
	long n;
	char buf[2048];

	fd = open(name, OREAD);
	if(fd < 0)
		return 1;
	n = read(fd, buf, sizeof(buf));
	close(fd);
	if(n <= 0) {
		if(debug)
			fprint(2, "removing directory %s\n", name);
		syslog(0, runqlog, "rmdir %s", name);
		remove(name);
		return 1;
	}
	return 0;
}

/*
 *  run all user directories, must be bootes (or root on unix) to do this
 */
void
doalldirs(void)
{
	Dir *db;
	int fd;
	long i, n;


	if((fd = open(".", OREAD)) == -1){
		warning("opening %s", root);
		return;
	}
	if((n = dirreadall(fd, &db)) == -1){
		warning("reading %s: ", root);
		close(fd);
		return;
	}
	for(i=0; i<n; i++){
		if((db[i].qid.type & QTDIR) == 0)
			continue;
		if(emptydir(db[i].name))
			continue;
		dodir(db[i].name);
	}
	close(fd);
	free(db);
}

/*
 *  cd to a user directory and run it
 */
void
dodir(char *name)
{
	if(chdir(name) < 0){
		warning("cd to %s", name);
		return;
	}
	if(debug)
		fprint(2, "running %s\n", name);
	rundir(name);
	chdir("..");
}

/*
 *  run the current directory
 */
void
rundir(char *name)
{
	int nlive, fidx, fd, found;
	Job *hd, *j, **p;
	Waitmsg *w;
	Wdir wd;

	fd = open(".", OREAD);
	if(fd == -1){
		warning("reading %s", name);
		return;
	}
	fidx= 0;
	hd = nil;
	nlive = 0;
	wd.name = name;
	wd.nd = dirreadall(fd, &wd.d);
	while(nlive > 0 ||  fidx< wd.nd){
		while(fidx< wd.nd && nlive < njob){
			if(strncmp(wd.d[fidx].name, "C.", 2) != 0){
				fidx++;
				continue;
			}
			if((j = dofile(&wd, &wd.d[fidx])) != nil){
				nlive++;
				j->next = hd;
				hd = j;
			}
			fidx++;
		}
		if(nlive == 0){
			fprint(2, "nothing live\n");
			break;
		}
rescan:
		if((w = wait()) == nil){
			syslog(0, "runq", "wait error: %r");
			break;
		}
		found = 0;
		for(p = &hd; *p != nil; p = &(*p)->next){
			if(w->pid == (*p)->pid){
				*p = donefile(*p, w);
				found++;
				nlive--;
				break;
			}
		}
		free(w);
		if(!found)
			goto rescan;
	}
	assert(hd == nil);
	free(wd.d);
	close(fd);
}

/*
 *  free files matching name in the current directory
 */
void
remmatch(Wdir *w, char *name)
{
	long i;

	syslog(0, runqlog, "removing %s/%s", w->name, name);

	for(i=0; i<w->nd; i++){
		if(strcmp(&w->d[i].name[1], &name[1]) == 0)
			remove(w->d[i].name);
	}

	/* error file (may have) appeared after we read the directory */
	/* stomp on data file in case of phase error */
	remove(file(name, 'D'));
	remove(file(name, 'E'));
}

/*
 *  like trylock, but we've already got the lock on fd,
 *  and don't want an L. lock file.
 */
static Mlock *
keeplockalive(char *path, int fd)
{
	char buf[1];
	Mlock *l;

	l = malloc(sizeof(Mlock));
	if(l == 0)
		return 0;
	l->fd = fd;
	snprint(l->name, sizeof l->name, "%s", path);

	/* fork process to keep lock alive until sysunlock(l) */
	switch(l->pid = rfork(RFPROC)){
	default:
		break;
	case 0:
		fd = l->fd;
		for(;;){
			sleep(1000*60);
			if(pread(fd, buf, 1, 0) < 0)
				break;
		}
		_exits(0);
	}
	return l;
}

/*
 *  Launch trying a message, returning a job
 *  tracks the running pid.
 */
Job*
dofile(Wdir *w, Dir *dp)
{
	int dtime, efd, i, etime;
	Job *j;
	Dir *d;
	char *cp;

	if(debug)
		fprint(2, "dofile %s\n", dp->name);
	/*
	 *  if no data file or empty control or data file, just clean up
	 *  the empty control file must be 15 minutes old, to minimize the
	 *  chance of a race.
	 */
	d = dirstat(file(dp->name, 'D'));
	if(d == nil){
		syslog(0, runqlog, "no data file for %s", dp->name);
		remmatch(w, dp->name);
		return nil;
	}
	if(dp->length == 0){
		if(time(0)-dp->mtime > 15*60){
			syslog(0, runqlog, "empty ctl file for %s", dp->name);
			remmatch(w, dp->name);
		}
		return nil;
	}
	dtime = d->mtime;
	free(d);

	/*
	 *  retry times depend on the age of the errors file
	 */
	if(!Eflag && (d = dirstat(file(dp->name, 'E'))) != nil){
		etime = d->mtime;
		free(d);
		if(etime - dtime < 60*60){
			/* up to the first hour, try every 15 minutes */
			if(time(0) - etime < 15*60)
				return nil;
		} else {
			/* after the first hour, try once an hour */
			if(time(0) - etime < 60*60)
				return nil;
		}
	}

	/*
	 *  open control and data
	 */
	j = malloc(sizeof(Job));
	if(j == nil)
		return nil;
	memset(j, 0, sizeof(Job));
	j->dp = dp;
	j->dfd = -1;
	j->b = sysopen(file(dp->name, 'C'), "rl", 0660);
	if(j->b == 0) {
		if(debug)
			fprint(2, "can't open %s: %r\n", file(dp->name, 'C'));
		return nil;
	}
	j->dfd = open(file(dp->name, 'D'), OREAD);
	if(j->dfd < 0){
		if(debug)
			fprint(2, "can't open %s: %r\n", file(dp->name, 'D'));
		freejob(j);
		return nil;
	}

	/*
	 *  make arg list
	 *	- read args into (malloc'd) buffer
	 *	- malloc a vector and copy pointers to args into it
	 */
	j->wdir = w;
	j->buf = malloc(dp->length+1);
	if(j->buf == nil){
		warning("buffer allocation", 0);
		freejob(j);
		return nil;
	}
	if(Bread(j->b, j->buf, dp->length) != dp->length){
		warning("reading control file %s\n", dp->name);
		freejob(j);
		return nil;
	}
	j->buf[dp->length] = 0;
	j->av = malloc(2*sizeof(char*));
	if(j->av == 0){
		warning("argv allocation", 0);
		freejob(j);
		return nil;
	}
	for(j->ac = 1, cp = j->buf; *cp; j->ac++){
		while(isspace(*cp))
			*cp++ = 0;
		if(*cp == 0)
			break;

		j->av = realloc(j->av, (j->ac+2)*sizeof(char*));
		if(j->av == 0){
			warning("argv allocation", 0);
		}
		j->av[j->ac] = cp;
		while(*cp && !isspace(*cp)){
			if(*cp++ == '"'){
				while(*cp && *cp != '"')
					cp++;
				if(*cp)
					cp++;
			}
		}
	}
	j->av[0] = cmd;
	j->av[j->ac] = 0;

	if(!Eflag &&time(0) - dtime > giveup){
		if(returnmail(j->av, w, dp->name, "Giveup") != 0)
			logit("returnmail failed", w, dp->name, j->av);
		remmatch(w, dp->name);
		goto done;
	}

	for(i = 0; i < nbad; i++){
		if(j->ac > 3 && strcmp(j->av[3], badsys[i]) == 0)
			goto done;
	}

	/*
	 * Ken's fs, for example, gives us 5 minutes of inactivity before
	 * the lock goes stale, so we have to keep reading it.
 	 */
	j->l = keeplockalive(file(dp->name, 'C'), Bfildes(j->b));

	/*
	 *  transfer
	 */
	j->pid = fork();
	switch(j->pid){
	case -1:
		sysunlock(j->l);
		sysunlockfile(Bfildes(j->b));
		syslog(0, runqlog, "out of procs");
		exits(0);
	case 0:
		if(debug) {
			fprint(2, "Starting %s\n", cmd);
			for(i = 0; j->av[i]; i++)
				fprint(2, " %s", j->av[i]);
			fprint(2, "\n");
		}
		logit("execing", w, dp->name, j->av);
		close(0);
		dup(j->dfd, 0);
		close(j->dfd);
		close(2);
		efd = open(file(dp->name, 'E'), OWRITE);
		if(efd < 0){
			if(debug)
				syslog(0, "runq", "open %s as %s: %r", file(dp->name,'E'), getuser());
			efd = create(file(dp->name, 'E'), OWRITE, 0666);
			if(efd < 0){
				if(debug) syslog(0, "runq", "create %s as %s: %r", file(dp->name, 'E'), getuser());
				exits("could not open error file - Retry");
			}
		}
		seek(efd, 0, 2);
		exec(cmd, j->av);
		error("can't exec %s", cmd);
		break;
	default:
		return j;
	}
done:
	freejob(j);
	return nil;
}

/*
 * Handle the completion of a job.
 * Wait for the pid, check its status,
 * and then pop the job off the list.
 * Return the next running job.
 */
Job*
donefile(Job *j, Waitmsg *wm)
{
	Job *n;

	if(debug)
		fprint(2, "wm->pid %d wm->msg == %s\n", wm->pid, wm->msg);

	if(wm->msg[0]){
		if(debug)
			fprint(2, "[%d] wm->msg == %s\n", getpid(), wm->msg);
		if(!Rflag && strstr(wm->msg, "Retry")==0){
			/* return the message and remove it */
			if(returnmail(j->av, j->wdir, j->dp->name, wm->msg) != 0)
				logit("returnmail failed", j->wdir, j->dp->name, j->av);
			remmatch(j->wdir, j->dp->name);
		} else {
			/* add sys to bad list and try again later */
			nbad++;
			badsys = realloc(badsys, nbad*sizeof(char*));
			badsys[nbad-1] = strdup(j->av[3]);
		}
	} else {
		/* it worked remove the message */
		remmatch(j->wdir, j->dp->name);
	}
	n = j->next;
	freejob(j);
	return n;
}

/*
 * Release resources associated with
 * a job.
 */
void
freejob(Job *j)
{
	if(j->b != nil){
		sysunlockfile(Bfildes(j->b));
		Bterm(j->b);
	}
	if(j->dfd != -1)
		close(j->dfd);
	if(j->l != nil)
		sysunlock(j->l);
	free(j->buf);
	free(j->av);
	free(j);
}


/*
 *  return a name starting with the given character
 */
char*
file(char *name, char type)
{
	static char nname[Elemlen+1];

	strncpy(nname, name, Elemlen);
	nname[Elemlen] = 0;
	nname[0] = type;
	return nname;
}

/*
 *  send back the mail with an error message
 *
 *  return 0 if successful
 */
int
returnmail(char **av, Wdir *w, char *name, char *msg)
{
	char buf[256], attachment[Pathlen], *sender;
	int i, fd, pfd[2];
	long n;
	Waitmsg *wm;
	String *s;

	if(av[1] == 0 || av[2] == 0){
		logit("runq - dumping bad file", w, name, av);
		return 0;
	}

	s = unescapespecial(s_copy(av[2]));
	sender = s_to_c(s);

	if(!returnable(sender) || strcmp(sender, "postmaster") == 0) {
		logit("runq - dumping p to p mail", w, name, av);
		return 0;
	}

	if(pipe(pfd) < 0){
		logit("runq - pipe failed", w, name, av);
		return -1;
	}

	switch(rfork(RFFDG|RFPROC|RFENVG)){
	case -1:
		logit("runq - fork failed", w, name, av);
		return -1;
	case 0:
		logit("returning", w, name, av);
		close(pfd[1]);
		close(0);
		dup(pfd[0], 0);
		close(pfd[0]);
		putenv("upasname", "/dev/null");
		snprint(buf, sizeof(buf), "%s/marshal", UPASBIN);
		snprint(attachment, sizeof(attachment), "%s", file(name, 'D'));
		execl(buf, "send", "-A", attachment, "-s", "permanent failure", sender, nil);
		error("can't exec", 0);
		break;
	default:
		break;
	}

	close(pfd[0]);
	fprint(pfd[1], "\n");	/* get out of headers */
	if(av[1]){
		fprint(pfd[1], "Your request ``%.20s ", av[1]);
		for(n = 3; av[n]; n++)
			fprint(pfd[1], "%s ", av[n]);
	}
	fprint(pfd[1], "'' failed (code %s).\nThe symptom was:\n\n", msg);
	fd = open(file(name, 'E'), OREAD);
	if(fd >= 0){
		for(;;){
			n = read(fd, buf, sizeof(buf));
			if(n <= 0)
				break;
			if(write(pfd[1], buf, n) != n){
				close(fd);
				goto out;
			}
		}
		close(fd);
	}
	close(pfd[1]);
out:
	wm = wait();
	if(wm == nil){
		syslog(0, "runq", "wait: %r");
		logit("wait failed", w, name, av);
		return -1;
	}
	i = 0;
	if(wm->msg[0]){
		i = -1;
		syslog(0, "runq", "returnmail child: %s", wm->msg);
		logit("returnmail child failed", w, name, av);
	}
	free(wm);
	return i;
}

/*
 *  print a warning and continue
 */
void
warning(char *f, void *a)
{
	char err[ERRMAX];
	char buf[256];

	rerrstr(err, sizeof(err));
	snprint(buf, sizeof(buf), f, a);
	fprint(2, "runq: %s: %s\n", buf, err);
}

/*
 *  print an error and die
 */
void
error(char *f, void *a)
{
	char err[ERRMAX];
	char buf[256];

	rerrstr(err, sizeof(err));
	snprint(buf, sizeof(buf), f, a);
	fprint(2, "runq: %s: %s\n", buf, err);
	exits(buf);
}

void
logit(char *msg, Wdir *w, char *file, char **av)
{
	int n, m;
	char buf[256];

	n = snprint(buf, sizeof(buf), "%s/%s: %s", w->name, file, msg);
	for(; *av; av++){
		m = strlen(*av);
		if(n + m + 4 > sizeof(buf))
			break;
		sprint(buf + n, " '%s'", *av);
		n += m + 3;
	}
	syslog(0, runqlog, "%s", buf);
}