atd.c

/***********************************************************************
*                                                                      *
*               This software is part of the ast package               *
*          Copyright (c) 1996-2012 AT&T Intellectual Property          *
*                      and is licensed under the                       *
*                 Eclipse Public License, Version 1.0                  *
*                    by AT&T Intellectual Property                     *
*                                                                      *
*                A copy of the License is available at                 *
*          http://www.eclipse.org/org/documents/epl-v10.html           *
*         (with md5 checksum b35adb5213ca9657e911e9befb180842)         *
*                                                                      *
*              Information and Software Systems Research               *
*                            AT&T Research                             *
*                           Florham Park NJ                            *
*                                                                      *
*                 Glenn Fowler <[email protected]>                  *
*                                                                      *
***********************************************************************/
#pragma prototyped
/*
 * Glenn Fowler
 * AT&T Research
 *
 * at.svc -- the other side of at
 *
 * the at control dir hierarchy, all files owned by one user
 *
 *	$INSTALLROOT/lib/at
 *		jobs			AT_DIR_MODE
 *			<host>		AT_DIR_MODE
 *				<job-i>	AT_JOB_MODE
 *
 * job names are encoded with <uid.gid.time> base 64 where
 * <time> is the earliest absolute time the job can be run
 */

static const char id[] = "\n@(#)$Id: at.svc (AT&T Research) 2012-02-29 $\0\n";

#include "at.h"

#include <cdt.h>
#include <debug.h>
#include <dirent.h>
#include <proc.h>
#include <pwd.h>
#include <sig.h>
#include <wait.h>

#ifndef SIGCHLD
#define SIGCHLD		SIGCLD
#endif

#define HOG		(NPROC*10)
#define LOAD		0
#define MSMAX		(LONG_MAX/1000)
#define NAME		22
#define NICE		2
#define NPROC		10
#define PERUSER		(NPROC/2)
#define SMAX		ULONG_MAX
#define WAIT		0

typedef struct
{
	Dtlink_t	byname;
	Dtlink_t	byuid;
	char*		home;
	int		pending;
	int		running;
	unsigned long	admin;
	unsigned long	uid;
	unsigned long	total;
	char		name[1];
} User_t;

typedef struct
{
	Dtlink_t	byuser;
	User_t*		user;
	int		allow;
	int		pending;
	int		running;
	unsigned long	total;
} Owner_t;

typedef struct
{
	Dtlink_t	byname;
	Dt_t*		owner;
	int		allow;
	int		home;
	int		nproc;
	int		peruser;
	int		load;
	int		nice;
	int		wait;
	int		pending;
	int		running;
	int		specific;
	unsigned long	total;
	char		name[1];
} Queue_t;

typedef struct
{
	Dtlink_t	byname;
	Dtlink_t	bypid;
	unsigned long	start;
	char		name[NAME];
	Queue_t*	queue;
	Owner_t*	owner;
	char*		period;
	char*		repeat;
	char*		shell;
	int		mail;
	Csid_t		id;
	unsigned long	pid;
	unsigned long	run;
	unsigned long	total;
	char		label[11];
	char		data[1];
} Job_t;

typedef struct
{
	int		fd;
	Csid_t		id;
} Connection_t;

typedef struct
{
	Dtdisc_t	discipline;
	Dt_t*		handle;
} Table_t;

typedef struct
{
	Cssdisc_t	disc;		/* css discipline		*/
	Css_t*		css;		/* css handle			*/
	struct
	{
	Table_t		job;		/* job by <start,name>		*/
	Table_t		owner;		/* user by User_t*		*/
	Table_t		pid;		/* job by pid			*/
	Table_t		queue;		/* queue by name		*/
	Table_t		uid;		/* user by uid			*/
	Table_t		user;		/* user by name			*/
	}		table;
	Queue_t*	queue;		/* default queue		*/
	time_t		update;		/* info file mtime		*/
	int		nproc;		/* total proc limit		*/
	int		peruser;	/* total per user limit		*/
	int		pending;	/* total pending jobs		*/
	int		running;	/* total running jobs		*/
	int		bufsiz;		/* sizeof(*buf)			*/
	int		init;		/* initialization		*/
	unsigned long	admin[4];	/* admin uids			*/
	unsigned long	total;		/* total number of jobs run	*/
	unsigned long	sequence;	/* job id sequence		*/
	char*		pwd;		/* jobs dir path		*/
	char*		atx;		/* setuid exec full path	*/
	char*		buf;		/* work buffer			*/
	Sfio_t*		tmp;		/* work string			*/
	Connection_t	con[1];		/* user connections		*/
} State_t;

typedef struct
{
	Connection_t*	con;
	Queue_t*	queue;
	State_t*	state;
} Visit_t;

static const char*	export[] =
{
	CO_ENV_HOST,	CO_ENV_TYPE
};

static const char*	queuedefs[] =
{
	"a.4j1n2u",	"b.2j2n90w2u",		"c.h8j2u60w"
};

static int	schedule(State_t*);

/*
 * return user info given name or uid
 */

static User_t*
user(register State_t* state, char* name, unsigned long uid)
{
	register User_t*	usr;
	register struct passwd*	pwd;
	register int		n;
	char*			home;
	char			buf[16];

	static User_t		nobody;

	if (!(usr = name ? (User_t*)dtmatch(state->table.user.handle, name) : (User_t*)dtmatch(state->table.uid.handle, &uid)))
	{
		if (pwd = name ? getpwnam(name) : getpwuid(uid))
		{
			if (name) uid = pwd->pw_uid;
			else name = pwd->pw_name;
			home = pwd->pw_dir;
		}
		else
		{
			if (!name)
				sfsprintf(name = buf, sizeof(buf), "%lu", uid);
			home = "/tmp";
		}
		n = strlen(name);
		if (!(usr = newof(0, User_t, 1, n + strlen(home) + 2)))
			usr = &nobody;
		else
		{
			usr->uid = uid;
			strcpy(usr->name, name);
			usr->home = strcpy(usr->name + n + 1, home);
			for (n = 0; n < elementsof(state->admin); n++)
				if (uid == state->admin[n])
				{
					usr->admin = 1;
					break;
				}
		}
		dtinsert(state->table.user.handle, usr);
		dtinsert(state->table.uid.handle, usr);
	}
	return usr;
}

/*
 * check if usr is permitted on que
 * Owner_t allocated and returned if ok
 */

static Owner_t*
permit(register Queue_t* que, User_t* usr)
{
	register Owner_t*	own;

	if (!(own = (Owner_t*)dtmatch(que->owner, &usr)))
	{
		if (que->allow > 0)
			return 0;
		if (own = newof(0, Owner_t, 1, 0))
		{
			own->allow = 1;
			own->user = usr;
			dtinsert(que->owner, own);
		}
	}
	else if (!own->allow)
		return 0;
	return own;
}

/*
 * initialize owner.allow 
 */

static int
allow(Dt_t* dt, void* object, void* handle)
{
	NoP(dt);
	((Owner_t*)object)->allow = !((Queue_t*)handle)->allow;
	return 0;
}

/*
 * add queue defined in s
 */

static void
queue(register State_t* state, register char* s)
{
	register Queue_t*	que;
	register Owner_t*	own;
	User_t*			usr;
	char*			t;
	long			n;

	while (isspace(*s))
		s++;
	if (*s && *s != '#')
	{
		if (t = strchr(s, '.'))
			*t++ = 0;
		if (!(que = (Queue_t*)dtmatch(state->table.queue.handle, s)))
		{
			if (!(que = newof(0, Queue_t, 1, strlen(s) + 1)))
				error(ERROR_SYSTEM|3, "out of space [queue]");
			strcpy(que->name, s);
			que->nproc = NPROC;
			que->peruser = PERUSER;
			que->load = LOAD;
			que->nice = NICE;
			que->wait = WAIT;
			que = (Queue_t*)dtinsert(state->table.queue.handle, que);
		}
		if (!state->queue)
			state->queue = que;
		if (!que->owner && !(que->owner = dtopen(&state->table.owner.discipline, Dtset)))
			error(ERROR_SYSTEM|3, "out of space [queue %s owner hash]", que->name);
		que->allow = -1;
		que->specific = 0;
		while ((s = t) && *s)
		{
			n = strtol(s, &t, 10);
			if (t == s)
				n = 1;
			switch (*t++)
			{
			case 0:
			case '#':
				t = 0;
				break;
			case 'h':
				que->home = n;
				break;
			case 'j':
				que->nproc = n;
				break;
			case 'l':
				que->load = n;
				break;
			case 'n':
				que->nice = n;
				break;
			case 'u':
				que->peruser = n;
				break;
			case 'w':
				que->wait = n;
				break;
			case ' ':
			case '\t':
				que->allow = *(t - 2) == '+';
				que->specific = 1;
				dtwalk(que->owner, allow, que);
				s = t;
				for (;;)
				{
					while (isspace(*s))
						s++;
					if (*(t = s) == '#')
						break;
					while (*s && !isspace(*s))
						s++;
					if (n = *s)
						*s++ = 0;
					if (!*t)
						break;
					usr = user(state, t, 0);
					if (!(own = (Owner_t*)dtmatch(que->owner, &usr)))
					{
						if (!(own = newof(0, Owner_t, 1, 0)))
							error(ERROR_SYSTEM|3, "out of space [queue owner]");
						own->user = usr;
						dtinsert(que->owner, own);
					}
					own->allow = que->allow;
				}
				t = 0;
				break;
			}
		}
	}
}

/*
 * update state from the queue, allow and deny files
 */

static void
update(register State_t* state)
{
	register char*		s;
	register Sfio_t*	sp;
	register Queue_t*	que;
	User_t*			usr;
	Owner_t*		own;
	int			permit;
	char*			file;
	char*			path;
	struct stat		st;

	sfprintf(state->tmp, "%s/%s/%s", state->pwd, state->pwd, AT_QUEUE_FILE);
	if (!(path = sfstruse(state->tmp)))
		error(ERROR_SYSTEM|3, "out of space");
	pathcanon(path, 0, 0);
	if (sp = sfopen(NiL, path, "r"))
	{
		error(0, "scan %s queue list", path);
		if (fstat(sffileno(sp), &st) || st.st_mtime != state->update)
		{
			state->update = st.st_mtime;
			if (que = state->queue)
				state->queue = 0;
			while (s = sfgetr(sp, '\n', 1))
				queue(state, s);
			if (!state->queue && !(state->queue = que))
				error(ERROR_SYSTEM|3, "%s: no queues", path);
		}
		sfclose(sp);
	}

	/*
	 * apply the global allow/deny files to queues
	 * with no specific allow/deny overrides
	 */

	for (que = (Queue_t*)dtfirst(state->table.queue.handle); que && que->specific; que = (Queue_t*)dtnext(state->table.queue.handle, que));
	if (que && !stat(file = AT_CRON_DIR, &st) && S_ISDIR(st.st_mode))
	{
		sfprintf(state->tmp, "%s/%s/%s", state->pwd, file, AT_ALLOW_FILE);
		if (!(path = sfstruse(state->tmp)))
			error(ERROR_SYSTEM|3, "out of space");
		pathcanon(path, 0, 0);
		if (sp = sfopen(NiL, path, "r"))
			permit = 1;
		else
		{
			permit = 0;
			sfprintf(state->tmp, "%s/%s/%s", state->pwd, file, AT_DENY_FILE);
			if (!(path = sfstruse(state->tmp)))
				error(ERROR_SYSTEM|3, "out of space");
			pathcanon(path, 0, 0);
			sp = sfopen(NiL, path, "r");
		}

		/*
		 * reset the queue and associated user access
		 */

		for (que = (Queue_t*)dtfirst(state->table.queue.handle); que; que = (Queue_t*)dtnext(state->table.queue.handle, que))
			if (!que->specific)
			{
				que->allow = permit;
				dtwalk(que->owner, allow, que);
			}

		/*
		 * scan and update the access
		 */

		if (sp)
		{
			error(0, "scan %s access list", path);
			while (s = sfgetr(sp, '\n', 1))
			{
				usr = user(state, s, 0);
				for (que = (Queue_t*)dtfirst(state->table.queue.handle); que; que = (Queue_t*)dtnext(state->table.queue.handle, que))
					if (!que->specific)
					{
						if (!(own = (Owner_t*)dtmatch(que->owner, &usr)))
						{
							if (!(own = newof(0, Owner_t, 1, 0)))
								error(ERROR_SYSTEM|3, "out of space [queue owner]");
							own->user = usr;
							dtinsert(que->owner, own);
						}
						own->allow = que->allow;
					}
			}
			sfclose(sp);
		}
	}
}

/*
 * add a client connection
 */

static int
client(Css_t* css, Cssfd_t* fp, Csid_t* id, char** args, Cssdisc_t* disc)
{
	register State_t*	state = (State_t*)disc;
	register Connection_t*	con = state->con + fp->fd;

	NoP(args);
	con->id = *id;
	return con->fd = fp->fd;
}

/*
 * insert a job into the run list
 */

static void
submit(State_t* state, register Job_t* job)
{
	dtinsert(state->table.job.handle, job);
	job->owner->pending++;
	job->owner->user->pending++;
	job->queue->pending++;
	state->pending++;
	error(0, "%s %s que %s at %s \"%s\"", job->name, fmtuid(job->owner->user->uid), job->queue->name, fmttime(AT_TIME_FORMAT, job->start), job->label);
}

/*
 * remove job from the run list
 */

static void
complete(State_t* state, register Job_t* job)
{
	dtdelete(state->table.job.handle, job);
	if (job->pid)
	{
		dtdelete(state->table.pid.handle, job);
		job->pid = 0;
	}
	if (job->owner)
	{
		if (job->run)
		{
			job->owner->running--;
			job->owner->user->running--;
		}
		job->owner->pending--;
		job->owner->user->pending--;
	}
	if (job->queue)
	{
		if (job->run)
		{
			job->queue->running--;
			state->running--;
		}
		job->queue->pending--;
		state->pending--;
	}
	job->run = 0;
}

/*
 * drop a job
 */

static void
drop(register State_t* state, register Job_t* job)
{
	complete(state, job);
	error(0, "%s %s que %s drop \"%s\"", job->name, fmtuid(job->owner->user->uid), job->queue->name, job->label);
	remove(job->name);
	free(job);
}

/*
 * pass job to atx for impersonation and execution
 */

static unsigned long
execute(register State_t* state, register Job_t* job)
{
	Proc_t*		proc;
	unsigned long	pid;
	char*		argv[4];
	long		ops[3];

	argv[0] = state->atx;
	argv[1] = job->shell;
	argv[2] = job->name;
	argv[3] = 0;
	ops[0] = PROC_SYS_PGRP(1);
	ops[1] = 0;
	message((-2, "%s %s %s", argv[0], argv[1], argv[2]));
	if (!(proc = procopen(argv[0], argv, NiL, ops, 0)))
		return 0;
	pid = proc->pid;
	procfree(proc);
	return pid;
}

/*
 * reap the state for a job that has completed
 */

static void
reap(register State_t* state, register Job_t* job, int status)
{
	char*		e;
	time_t		t;

	error(0, "%s %s %lu exit %d \"%s\"", job->name, fmtuid(job->owner->user->uid), job->pid, status, job->label);
	if (job->repeat && (t = job->start) && (t = tmdate(job->repeat, &e, &t)) && !*e)
	{
		complete(state, job);
		job->start = t;
		submit(state, job);
	}
	else
		drop(state, job);
	schedule(state);
}

/*
 * execute ready jobs
 */

static int
schedule(register State_t* state)
{
	register Job_t*		job;
	register Queue_t*	que;
	Csstat_t		st;

	unsigned long		x = SMAX;

	csstat(state->css->state, NiL, &st);
	for (job = (Job_t*)dtfirst(state->table.job.handle); job; job = (Job_t*)dtnext(state->table.job.handle, job))
	{
		message((-2, "schedule job=%s start=%lu time=%lu queue=%s load=%d.%02d/%d.%02d%s", job->name, job->start, cs.time, job->queue->name, job->queue->load / 100, job->queue->load % 100, st.load / 100, st.load % 100, job->run ? " RUNNING" : job->start <= cs.time ? " READY" : ""));
		if (!job->run)
		{
			if (job->start <= cs.time)
			{
				que = job->queue;
				if ((que->nproc <= 0 || que->running < que->nproc) &&
				    (que->load <= 0 || st.load < que->load) &&
				    (state->nproc <= 0 || state->running < state->nproc) &&
				    (que->peruser <= 0 || job->owner->running < que->peruser) &&
				    (state->peruser <= 0 || job->owner->running < state->peruser))
				{
					if (job->pid = execute(state, job))
					{
						job->run = cs.time;
						job->owner->running++;
						job->owner->total++;
						job->owner->user->running++;
						job->owner->user->total++;
						que->running++;
						que->total++;
						state->running++;
						state->total++;
						dtinsert(state->table.pid.handle, job);
						message((-2, "exec job=%s pid=%lu", job->name, job->pid));
						error(0, "%s %s %lu exec \"%s\"", job->name, fmtuid(job->owner->user->uid), job->pid, job->label);
					}
					else if (x > (job->start + que->wait))
						x = job->start + que->wait;
				}
				else if (x > (job->start + que->wait))
					x = job->start + que->wait;
			}
			else
			{
				if (x > job->start)
					x = job->start;
				break;
			}
		}
	}
	if (x == SMAX)
		state->disc.wakeup = 0;
	else if (x > cs.time)
	{
		x -= cs.time;
		x = (x >= MSMAX) ? MSMAX : (x * 1000);
		state->disc.wakeup = x;
		message((-2, "wakeup %s", fmtelapsed(x, 1000)));
	}
	return 0;
}

static int
exception(Css_t* css, unsigned long op, unsigned long arg, Cssdisc_t* disc)
{
	register State_t*	state = (State_t*)disc;
	int			status;
	pid_t			pid;
	Job_t*			job;

	switch (op)
	{
	case CSS_INTERRUPT:
		if (arg != SIGCHLD)
			error(ERROR_SYSTEM|3, "%s: interrupt exit", fmtsignal(arg));
		for (;;)
		{
			switch (pid = waitpid(-1, &status, WNOHANG))
			{
			case -1:
				if (errno == EINTR)
					continue;
				break;
			case 0:
				break;
			default:
				status = WIFSIGNALED(status) ?
					EXIT_TERM(WTERMSIG(status)) :
					EXIT_CODE(WEXITSTATUS(status));
				message((-2, "wait pid=%lu status=%d", pid, status));
				if (job = (Job_t*)dtmatch(state->table.pid.handle, &pid))
					reap(state, job, status);
				continue;
			}
			break;
		}
		return 1;
	case CSS_WAKEUP:
		schedule(state);
		return 1;
	}
	error(ERROR_SYSTEM|3, "poll error");
	return -1;
}

/*
 * list owner names in handle
 */

static int
listowner(Dt_t* dt, void* object, void* handle)
{
	register Owner_t*	own = (Owner_t*)object;
	register Visit_t*	vis = (Visit_t*)handle;

	NoP(dt);
	if (own->allow == vis->queue->allow)
		sfprintf(vis->state->tmp, " %s", own->user->name);
	return 0;
}

/*
 * list queue status
 */

static int
listqueue(Dt_t* dt, void* object, void* handle)
{
	register Queue_t*	que = (Queue_t*)object;
	Connection_t*		con = ((Visit_t*)handle)->con;
	State_t*		state = ((Visit_t*)handle)->state;
	char*			s;
	Visit_t			visit;

	NoP(dt);
	sfprintf(state->tmp, " %c", que->allow ? '+' : '-');
	visit.queue = que;
	visit.state = state;
	dtwalk(que->owner, listowner, &visit);
	if (!(s = sfstruse(state->tmp)))
		error(ERROR_SYSTEM|3, "out of space");
	if (!s[2])
		*s = 0;
	error(ERROR_OUTPUT|0, con->fd, "%-3s %5lu %3d %3d %3d %3d %3d %2d.%02d %5.5s%s", que->name, que->total, que->pending, que->running, que->nproc, que->peruser, que->nice, que->load / 100, que->load % 100, fmtelapsed(que->wait, 1), s);
	return 0;
}

/*
 * output name='value'
 */

static void
neqv(register Sfio_t* sp, const char* name, register char* v)
{
	register int	c;

	sfprintf(sp, " \\\n %s='", name);
	while (c = *v++)
	{
		if (c == '\'')
			sfputr(sp, "'\\'", -1);
		sfputc(sp, c);
	}
	sfputc(sp, '\'');
}

/*
 * execute the at command in s
 */

static int
command(register State_t* state, Connection_t* con, register char* s, int n, char* data)
{
	register Queue_t*	que;
	register Job_t*		job;
	Owner_t*		own;
	User_t*			usr;
	char*			t;
	char*			u;
	char*			b;
	char*			h;
	int			c;
	int			admin;
	int			mail;
	int			skip;
	unsigned long		m;
	unsigned long		w;
	long			x;
	Job_t*			next;
	Sfio_t*			sp;
	Visit_t			visit;

	usr = user(state, NiL, con->id.uid);
	b = s;
	admin = 0;
	if (*++s == AT_ADMIN)
	{
		s++;
		if (!usr->admin)
			goto denied;
		if (!++usr->admin)
			usr->admin = 1;
		admin = 1;
	}
	if (*s == AT_QUEUE)
	{
		t = ++s;
		if (!(s = strchr(s, ' ')))
		{
			error(ERROR_OUTPUT|2, con->fd, "%s: invalid queue name", t);
			return -1;
		}
		*s = 0;
		if (!(que = (Queue_t*)dtmatch(state->table.queue.handle, t)))
		{
			error(ERROR_OUTPUT|2, con->fd, "%s: unknown queue", t);
			return -1;
		}
		*s++ = ' ';
		if (!permit(que, usr))
			goto noqueue;
	}
	else
		que = 0;
	if (mail = *s == AT_MAIL)
		s++;
	switch (c = *s++)
	{
	case AT_ACCESS:
		if (!que && !permit(que = state->queue, usr))
			goto noqueue;
		break;
	case AT_DEBUG:
		if (!usr->admin)
			goto denied;
		if (!++usr->admin)
			usr->admin = 1;
		error_info.trace = -strtol(s, &t, 0);
		message((error_info.trace, "%s", fmtident(id)));
		break;
	case AT_INFO:
		error(ERROR_OUTPUT|0, con->fd, "at service daemon pid %ld user %s", state->con[0].id.pid, fmtuid(state->admin[0]));
		error(ERROR_OUTPUT|0, con->fd, "QUE TOTAL SUB RUN MAX USR PRI  LOAD  WAIT ACCESS");
		visit.state = state;
		visit.con = con;
		if (que)
			listqueue(NiL, que, &visit);
		else
			dtwalk(state->table.queue.handle, listqueue, &visit);
		break;
	case AT_JOB:
		if (!que)
			que = state->queue;
		if (!(own = permit(que, usr)))
			goto noqueue;
		if (own->user->pending >= HOG)
		{
			error(ERROR_OUTPUT|2, con->fd, "%s: hog", own->user->name);
			return -1;
		}
		skip = strtol(s, &t, 10);
		if (*t++ != ' ' || !state->init && skip > n)
		{
			error(ERROR_OUTPUT|2, con->fd, "garbled job message");
			return -1;
		}
		for (s = t; *s && *s != '\n'; s++);
		h = s + 1;
		if (!(job = newof(0, Job_t, 1, s - t)))
		{
			error(ERROR_SYSTEM|ERROR_OUTPUT|2, con->fd, "out of space [job]");
			return -1;
		}
		if (state->sequence < cs.time)
			state->sequence = cs.time;
		if (data)
			sfsprintf(job->name, sizeof(job->name), "%s", data);
		else
			sfsprintf(job->name, sizeof(job->name), "%..36lu.%..36lu.%..36lu", con->id.uid, con->id.gid, state->sequence++);
		s = (char*)memcpy(job->data, t, s - t);
		job->start = cs.time;
		job->shell = s;
		while (*s)
			if (*s++ == ' ')
			{
				*(s - 1) = 0;
				break;
			}
		job->id = con->id;
		job->queue = que;
		job->owner = own;
		job->mail = mail;
		while (c = *s++)
		{
			switch (c)
			{
			case AT_LABEL:
				t = job->label;
				for (t = job->label; *s && !isspace(*s); s++)
					if (t < &job->label[sizeof(job->label)-1])
						*t++ = *s;
				if (*s)
					s++;
				continue;
			case AT_TIME:
				job->start = strtol(s, &t, 0);
				s = t;
				if (*s == ' ')
					s++;
				if (*s)
				{
					job->repeat = s;
					if (*s == '+')
					{
						while (*++s && !isspace(*s));
						while (isspace(*s))
							s++;
					}
				}
				job->period = s;
				break;
			default:
				break;
			}
			break;
		}
		if (state->init && job->repeat)
			job->start = tmdate(job->repeat, NiL, NiL);
		if (!state->init && !*(t = job->label))
		{
			m = 0;
			t = b + skip;
			if (*t == ':')
			{
				while (isspace(*++t));
				if (u = strchr(t, ';'))
				{
					while (u > t && isspace(*(u - 1)))
						u--;
					m = u - t;
				}
			}
			if (m)
			{
				if (m >= sizeof(job->label))
					m = sizeof(job->label) - 1;
				memcpy(job->label, t, m);
				job->label[m] = 0;
			}
			else
			{
				for (t = b + skip; *t; t++)
					if (isalnum(*t))
					{
						u = t;
						while (isalnum(*++t));
						c = t - u;
						if (c == 3 && strneq(u, "for", 3) || c == 2 && strneq(u, "if", 2) || c == 5 && strneq(u, "while", 5))
							continue;
						break;
					}
				m = 0;
				x = -1;
				while (c = *t++)
				{
					if (isalnum(c))
					{
						if (x)
						{
							if (x > 0)
							{
								w = sfstrtell(state->tmp);
								if (++m >= sizeof(job->label))
									break;
								sfputc(state->tmp, '.');
							}
							x = 0;
						}
						if (++m >= sizeof(job->label))
							break;
						sfputc(state->tmp, c);
					}
					else if (!x)
						x = 1;
				}
				if (m >= sizeof(job->label))
					sfstrseek(state->tmp, w, SEEK_SET);
				if (!(t = sfstruse(state->tmp)))
					error(ERROR_SYSTEM|3, "out of space");
				strcpy(job->label, t);
			}
		}
		submit(state, job);
		if (!state->init)
		{
			if (!(sp = sfopen(NiL, job->name, "w")))
				goto noexec;
			chmod(job->name, AT_JOB_MODE);
			sfprintf(sp, "#%c%s %c0 %s %c%s %c%lu", AT_QUEUE, job->queue->name, AT_JOB, job->shell, AT_LABEL, t, AT_TIME, job->start);
			if (job->repeat)
				sfprintf(sp, " %s", job->repeat);
			sfputc(sp, '\n');
			c = h - b;
			b += c;
			n -= c;
			skip -= c;
			sfprintf(sp, "tmp=/tmp/at$$\ntrap \"rm -f $tmp\" 0 1 2 3 15\n");
			sfprintf(sp, "{\necho at job %s exec $(date)\n{\n", job->name);
			sfprintf(sp, "export");
			x = 0;
			for (c = 0; c < elementsof(export); c++)
				if ((s = getenv(export[c])) && *s)
				{
					neqv(sp, export[c], s);
					x = 1;
				}
			if (que->home)
			{
				b += skip;
				n -= skip;
				neqv(sp, "HOME", job->owner->user->home);
				neqv(sp, "LOGNAME", job->owner->user->name);
				neqv(sp, "USER", job->owner->user->name);
				neqv(sp, "PATH", pathbin());
				neqv(sp, "SHELL", job->shell);
				sfputr(sp, "\ncd \"$HOME\"\n", -1);
				x = 1;
			}
			sfputr(sp, x ? "" : " HOME", '\n');
			b[n - 1] = '\n';
			if (sfwrite(sp, b, n) != n)
				goto noexec;
			sfprintf(sp, "\n} </dev/null\necho at job %s exit $(date) status $?\n} >$tmp 2>&1\n", job->name);
			if (!job->mail)
				sfprintf(sp, "test \"$(wc -l < $tmp)\" -le 2 || ");
			sfprintf(sp, "mailx -s \"at job status\" %s < $tmp\n", job->owner->user->name);
			if (sfclose(sp))
			{
				sp = 0;
				goto noexec;
			}
			error(ERROR_OUTPUT|0, con->fd, "job %s at %s%s%s", job->name, fmttime("%a %b %e %T %Y", job->start), job->repeat ? " repeat " : "", job->period);
			schedule(state);
		}
		break;
	case AT_LIST:
	case AT_REMOVE:
	case AT_STATUS:
		if (*s++ != ' ')
			s = 0;
		m = 0;
		for (job = (Job_t*)dtfirst(state->table.job.handle); job; job = next)
		{
			next = (Job_t*)dtnext(state->table.job.handle, job);
			if ((!que || que == job->queue) && (s && strmatch(job->name, s) || !s && (admin || con->id.uid == job->id.uid)) && job->owner->allow)
				switch (c)
				{
				case AT_LIST:
					error(ERROR_OUTPUT|0, con->fd, "%s\t%s", job->name, fmttime("%a %b %e %T %Y", job->start));
					break;
				case AT_STATUS:
					if (!m++)
						error(ERROR_OUTPUT|0, con->fd, "JOB                   LABEL        PID   Q USER     START               REPEAT");
					error(ERROR_OUTPUT|0, con->fd, "%-21s %-*s%7d %-1s %.-8s %s %s", job->name, sizeof(job->label), job->label, job->pid, job->queue->name, job->owner->user->name, fmttime(AT_TIME_FORMAT, job->start), job->period);
					break;
				case AT_REMOVE:
					if (con->id.uid == job->id.uid || !con->id.uid)
					{
						drop(state, job);
						m++;
					}
					else
						error(ERROR_OUTPUT|2, con->fd, "%s: only %s can remove this job", job->name, fmtuid(job->id.uid));
					break;
				}
		}
		if (s && !m)
		{
			error(ERROR_OUTPUT|0, con->fd, "%s: no matching jobs", s ? s : "*");
			return -1;
		}
		break;
	case AT_LOG:
		sfprintf(state->tmp, "%s/%s", state->pwd, AT_LOG_FILE);
		if (!(s = sfstruse(state->tmp)))
			error(ERROR_SYSTEM|3, "out of space");
		pathcanon(s, 0, 0);
		error(ERROR_OUTPUT|0, con->fd, "%s", s);
		break;
	case AT_QUIT:
		if (!usr->admin)
			goto denied;
		if (!++usr->admin)
			usr->admin = 1;
		error(0, "daemon quit by %s", usr->name);
		exit(0);
		break;
	case AT_UPDATE:
		if (!usr->admin)
			goto denied;
		if (!++usr->admin)
			usr->admin = 1;
		if (*s)
			queue(state, s);
		update(state);
		break;
	case AT_USER:
		error(ERROR_OUTPUT|0, con->fd, "USER      ADMIN TOTAL SUB RUN HOME");
		for (usr = (User_t*)dtfirst(state->table.user.handle); usr; usr = (User_t*)dtnext(state->table.user.handle, usr))
			if (admin || con->id.uid == usr->uid)
				error(ERROR_OUTPUT|0, con->fd, "%-9.9s %5lu %5lu %3d %3d %s", usr->name, usr->admin, usr->total, usr->pending, usr->running, usr->home);
		break;
	case AT_VERSION:
		error(ERROR_OUTPUT|0, con->fd, "%s", fmtident(id));
		break;
	default:
		error(ERROR_OUTPUT|2, con->fd, "%c: unknown command", *(s - 1));
		return -1;
	}
	return 0;
 denied:
	error(ERROR_OUTPUT|2, con->fd, "%s: access denied", usr->name);
	return -1;
 noexec:
	if (sp)
		sfclose(sp);
	error(ERROR_SYSTEM|ERROR_OUTPUT|2, con->fd, "%s: cannot save action", job->name);
	drop(state, job);
	return -1;
 noqueue:
	error(ERROR_OUTPUT|2, con->fd, "%s: queue %s access denied", usr->name, que->name);
	return -1;
}

/*
 * order job by <start,name>
 */

static int
order(Dt_t* dt, void* a, void* b, Dtdisc_t* disc)
{
	register long	r1;
	register int	r2;

	NoP(dt);
	NoP(disc);
	if (!(r2 = strcmp((char*)a + sizeof(unsigned long), (char*)b + sizeof(unsigned long))))
		return 0;
	if (!(r1 = *(unsigned long*)a - *(unsigned long*)b))
		return r2;
	return r1;
}

static unsigned long	rollover;	/* XXX stampwrite() has no discipline	*/

/*
 * commit to the log file
 * if its too big then rename to .old and start fresh
 */

static void
commit(void)
{
	int		rolled;
	int		fd;
	time_t		t;
	unsigned long	now;
	struct stat	st;
	char		buf[PATH_MAX];
	Tm_t*		tm;

	static int	commiting = 0;

	if (commiting++)
	{
		commiting--;
		return;
	}
	now = NOW;
	if (!rollover)
	{
		t = stat(AT_LOG_FILE, &st) ? now : st.st_mtime;
		tm = tmmake(&t);
		tm->tm_mon++;
		tm->tm_mday = 1;
		tm->tm_hour = 0;
		tm->tm_min = 0;
		tm->tm_sec = 0;
		rollover = tmtime(tm, TM_LOCALZONE);
	}
	if (now >= rollover)
	{
		rolled = 1;
		error(0, "log file rollover");
		sfsprintf(buf, sizeof(buf), "%s.old", AT_LOG_FILE);
		remove(buf);
		if (rename(AT_LOG_FILE, buf))
			error(ERROR_SYSTEM|AT_STRICT, "%s: cannot rename log file to %s", AT_LOG_FILE, buf);
		t = rollover;
		tm = tmmake(&t);
		tm->tm_mon++;
		rollover = tmtime(tm, TM_LOCALZONE);
	}
	else
		rolled = 0;
	if ((fd = open(AT_LOG_FILE, O_CREAT|O_WRONLY|O_APPEND, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH)) < 0 || fd != 2 && (dup2(fd, 2) != 2 || close(fd)))
		error(ERROR_SYSTEM|AT_STRICT, "%s: cannot append to log file", AT_LOG_FILE);
	if (rolled)
		error(0, "log file rollover");
	commiting--;
}

/*
 * prepend current date-time to buffer on fd==2
 * and drop the initial command label if any
 */

static ssize_t
stampwrite(int fd, const void* buf, size_t n)
{
	register char*		s;
	register int		i;
	register ssize_t	r;
	register ssize_t	z;
	unsigned long		now;

	r = 0;
	if (rollover)
	{
		now = NOW;
		if (now >= rollover)
			commit();
	}
	else if (!now)
		now = NOW;
	if (fd == 2 && (s = fmttime(AT_TIME_FORMAT, now)))
	{
		i = strlen(s);
		s[i++] = ' ';
		if ((z = write(fd, s, i)) < 0)
			r = -1;
		else
			r += z;
		for (s = (char*)buf; s < ((char*)buf + n - 1) && !isspace(*s); s++)
			if (*s == ':')
			{
				while (++s < ((char*)buf + n - 1) && isspace(*s));
				n -= s - (char*)buf;
				buf = (void*)s;
				break;
			}
	}
	if ((z = write(fd, buf, n)) < 0)
		r = -1;
	else if (r >= 0)
		r += z;
	return r;
}

/*
 * service a request
 */

static int
request(Css_t* css, Cssfd_t* fp, Cssdisc_t* disc)
{
	register State_t*	state = (State_t*)disc;
	register Connection_t*	con = state->con + fp->fd;
	register char*		s;
	char*			t;
	unsigned long		n;
	int			c;
	
	if (fp->status != CS_POLL_READ)
		return -1;
	if ((c = csread(state->css->state, fp->fd, state->buf, 7, CS_EXACT|CS_RESTART)) != 7)
	{
		if (c)
			error(ERROR_OUTPUT|2, con->fd, "message size read error");
		return -1;
	}
	state->buf[6] = 0;
	if ((n = strtol(state->buf + 1, &t, 10)) <= 0 || *t)
	{
		error(ERROR_OUTPUT|2, con->fd, "invalid message size");
		return -1;
	}
	if (n > state->bufsiz)
	{
		if (n > INT_MAX || (c = roundof(n, PATH_MAX)) <= 0 || !(s = newof(state->buf, char, c, 0)))
		{
			error(ERROR_OUTPUT|2, con->fd, "%ld: message too big", n);
			return -1;
		}
		state->buf = s;
		state->bufsiz = c;
	}
	if (csread(state->css->state, fp->fd, s = state->buf, n, CS_EXACT|CS_RESTART) != n)
	{
		error(ERROR_OUTPUT|2, con->fd, "message body read error");
		return -1;
	}
	s[n - 1] = 0;
	command(state, con, s, n, NiL);
	return -1;
}

/*
 * initialize the state
 */

static int
init(const char* path)
{
	register State_t*	state;
	register DIR*		dir;
	register struct dirent*	ent;
	char*			s;
	char*			b;
	Sfio_t*			sp;
	int			i;
	unsigned long		limited;
	unsigned long*		ap;
	struct stat		ds;
	struct stat		hs;
	struct stat		js;
	struct stat		xs;

	umask(S_IWGRP|S_IWOTH);
	if ((i = (int)strtol(astconf("OPEN_MAX", NiL, NiL), NiL, 0)) < 20)
		i = 20;
	if (!(state = newof(0, State_t, 1, (i - 1) * sizeof(Connection_t))))
		error(ERROR_SYSTEM|3, "out of space [state]");
	state->bufsiz = 4 * PATH_MAX;
	if (!(state->buf = newof(0, char, state->bufsiz, 0)))
		error(ERROR_SYSTEM|3, "out of space [buf]");
	if (!(state->tmp = sfstropen()))
		error(ERROR_SYSTEM|3, "out of space [tmp]");
	ap = state->admin;
	*ap++ = state->con[0].id.uid = geteuid();
	*ap++ = 0;
	s = state->buf;
	if (!pathpath(AT_JOB_DIR, "", PATH_ABSOLUTE|PATH_EXECUTE, s, state->bufsiz) || lstat(s, &ds) || !AT_DIR_OK(&ds))
		error(ERROR_SYSTEM|3, "%s: job directory not found", AT_JOB_DIR);
	if (ds.st_uid != state->admin[0])
		error(ERROR_SYSTEM|3, "%s: job directory uid %d != effective uid %d", s, ds.st_uid, state->admin[0]);
	state->disc.version = CSS_VERSION;
	state->disc.flags = CSS_DAEMON|CSS_ERROR|CSS_INTERRUPT|CSS_WAKEUP;
	state->disc.errorf = errorf;
	state->disc.acceptf = client;
	state->disc.actionf = request;
	state->disc.exceptf = exception;
	if (!(state->css = cssopen(path, &state->disc)))
		return -1;
	state->con[0].id.gid = getegid();
	state->con[0].id.pid = getpid();
	state->con[0].id.hid = csaddr(state->css->state, NiL);
	state->con[0].fd = 2;
	state->init = 1;
	b = s + strlen(s);
	*b++ = '/';
	sfsprintf(b, state->bufsiz - (b - s), "%s", csname(state->css->state, 0L));
	if (lstat(s, &hs) && (mkdir(s, AT_DIR_MODE) || chmod(s, AT_DIR_MODE) || lstat(s, &hs)))
		error(ERROR_SYSTEM|3, "%s: cannot create job host directory", s);
	if (!AT_DIR_OK(&hs) || ds.st_uid != hs.st_uid || chdir(s))
		error(ERROR_SYSTEM|3, "%s: invalid job host directory %s [dir.uid=%d host.uid=%d]", s, fmtmode(hs.st_mode, 0), ds.st_uid, hs.st_uid);
	if (!(state->pwd = strdup(s)))
		error(ERROR_SYSTEM|3, "out of space [pwd]");
	if (hs.st_uid != state->admin[0])
		error(AT_STRICT, "%s: directory owner %s does not match daemon %s", s, fmtuid(hs.st_uid), fmtuid(state->admin[0]));
	sfsprintf(s, state->bufsiz, "%s/%s", state->pwd, AT_EXEC_FILE);
	pathcanon(s, 0, 0);
	if (lstat(s, &xs))
		error(ERROR_SYSTEM|3, "%s: job exec command not found", s);
	if (!S_ISREG(xs.st_mode))
		error(3, "%s: invalid mode %s -- regular file expected", s, fmtmode(xs.st_mode, 0));
	if ((xs.st_mode&(S_IXUSR|S_IXGRP|S_IWOTH|S_IXOTH)) != (S_IXUSR|S_IXGRP|S_IXOTH))
		error(3, "%s: invalid mode %s", s, fmtmode(xs.st_mode, 0));
	if (!(xs.st_mode&S_ISUID) && geteuid() != 0 && geteuid() != xs.st_uid)
		error(3, "%s: invalid euid %d -- %d expected", s, geteuid(), xs.st_uid);
#if 0
	if (!AT_EXEC_OK(&ds, &xs))
		error(3, "%s: invalid [ mode=%04o uid=%d euid=%d t1=%04o t2=%04o==%04o ]", s, xs.st_mode, xs.st_uid, geteuid(), S_ISREG(xs.st_mode), xs.st_mode&(S_IXUSR|S_IXGRP|S_IWOTH|S_IXOTH), (S_IXUSR|S_IXGRP|S_IXOTH));
#endif
	*ap++ = ds.st_uid;
	*ap = xs.st_uid;
	limited = (xs.st_mode & S_ISUID) ? (unsigned long)xs.st_uid : state->admin[0];
	if (!(state->atx = strdup(s)))
		error(ERROR_SYSTEM|3, "out of space [atx]");
	state->table.job.discipline.key = offsetof(Job_t, start);
	state->table.job.discipline.comparf = order;
	if (!(state->table.job.handle = dtopen(&state->table.job.discipline, Dtoset)))
		error(ERROR_SYSTEM|3, "out of space [job table]");
	state->table.owner.discipline.key = offsetof(Owner_t, user);
	state->table.owner.discipline.size = sizeof(User_t*);
	state->table.pid.discipline.key = offsetof(Job_t, pid);
	state->table.pid.discipline.size = sizeof(long);
	state->table.pid.discipline.link = offsetof(Job_t, bypid);
	if (!(state->table.pid.handle = dtopen(&state->table.pid.discipline, Dtset)))
		error(ERROR_SYSTEM|3, "out of space [pid table]");
	state->table.queue.discipline.key = offsetof(Queue_t, name);
	if (!(state->table.queue.handle = dtopen(&state->table.queue.discipline, Dtoset)))
		error(ERROR_SYSTEM|3, "out of space [queue table]");
	state->table.uid.discipline.key = offsetof(User_t, uid);
	state->table.uid.discipline.size = sizeof(long);
	state->table.uid.discipline.link = offsetof(User_t, byuid);
	if (!(state->table.uid.handle = dtopen(&state->table.uid.discipline, Dtset)))
		error(ERROR_SYSTEM|3, "out of space [uid table]");
	state->table.user.discipline.key = offsetof(User_t, name);
	if (!(state->table.user.handle = dtopen(&state->table.user.discipline, Dtoset)))
		error(ERROR_SYSTEM|3, "out of space [user table]");
	for (i = 0; i < elementsof(queuedefs); i++)
		queue(state, strcpy(state->buf, queuedefs[i]));
	commit();
	error(0, "daemon restart pid %ld user %s", state->con[0].id.pid, fmtuid(state->admin[0]));
	if (limited)
		error(0, "service limited to user %s", fmtuid(limited));

	/*
	 * update the queue definitions
	 */

	update(state);

	/*
	 * resubmit old jobs
	 */

	if (dir = opendir("."))
	{
		while (ent = readdir(dir))
			if (lstat(s = ent->d_name, &js))
			{
				error(0, "cannot stat old job %s", s);
				remove(s);
			}
			else if (!S_ISREG(js.st_mode))
			{
				if (!streq(s, ".") && !streq(s, ".."))
				{
					error(0, "invalid old job %s type %s rejected", s, fmtmode(js.st_mode, 0));
					if (S_ISDIR(js.st_mode))
						rmdir(s);
					else
						remove(s);
				}
			}
			else if (sfsscanf(s, "%..36lu.%..36lu.%..36lu", &state->con[0].id.uid, &state->con[0].id.gid, &cs.time) != 3)
			{
				error(0, "invalid old job %s name rejected", s);
				remove(s);
			}
			else if (!AT_OLD_OK(&hs, &js))
			{
				error(0, "invalid old job %s mode %s rejected [dir.uid=%d job.uid=%d]", s, fmtmode(js.st_mode, 0), hs.st_uid, js.st_uid);
				remove(s);
			}
			else if (!(sp = sfopen(NiL, s, "r")))
			{
				error(0, "cannot read old job %s", s);
				remove(s);
			}
			else
			{
				if (b = (char*)sfreserve(sp, SF_UNBOUND, SF_LOCKR))
					command(state, state->con, b, sfvalue(sp), s);
				sfclose(sp);
			}
		closedir(dir);
	}
	CSTIME();
	state->con[0].id.uid = geteuid();
	state->con[0].id.gid = getegid();
	state->init = 0;
	schedule(state);
	return 0;
}

int
main(int argc, char** argv)
{
	char*	path;
	pid_t	pid;
	int	status;

	NoP(argc);
	NoP(argv);
	error_info.id = "at.svc";
	error_info.write = stampwrite;
	if ((path = argv[1]) && !(path = strdup(path)))
		path = argv[1];

	/*
	 * monitor the daemon and restart if it dies
	 */

	csdaemon(&cs, (1<<0)|(1<<1)|(1<<2));
	for (;;)
	{
		if ((pid = fork()) <= 0)
		{
			if (!init(path))
				csspoll(CS_NEVER, 0);
			return 1;
		}
		while (waitpid(pid, &status, 0) != pid);
		if (!status)
			break;
		sleep(60);
	}
	return 0;
}