coexec.c

/***********************************************************************
*                                                                      *
*               This software is part of the ast package               *
*          Copyright (c) 1990-2011 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
 *
 * send an action to the coshell for execution
 */

#include "colib.h"

#include <proc.h>
#include <ls.h>

static Cojob_t*
service(register Coshell_t* co, Coservice_t* cs, Cojob_t* cj, int flags, Sfio_t* sp)
{
	Proc_t*		proc;
	size_t		n;
	int		i;
	int		j;
	int		fds[2];
	long		ops[4];
	char*		s;
	char**		a;

	if (flags & CO_DEBUG)
	{
		for (a = cs->argv; *a; a++)
			sfprintf(sp, " %s", *a);
		if (!(s = costash(sp)))
			goto nospace;
		errormsg(state.lib, ERROR_LIBRARY|2, "service %s:%s", cs->path, s);
	}
	if (pipe(fds) < 0)
	{
		errormsg(state.lib, ERROR_LIBRARY|ERROR_SYSTEM|2, "%s: cannot allocate service pipe", cs->name);
		return 0;
	}
	if (co->flags & CO_SHELL)
		for (i = 0; i < elementsof(fds); i++)
			if (fds[i] < 10 && (j = fcntl(fds[i], F_DUPFD, 10)) >= 0)
			{
				close(fds[i]);
				fds[i] = j;
			}
	cs->fd = fds[1];
	ops[0] = PROC_FD_DUP(fds[0], 0, PROC_FD_PARENT);
	ops[1] = PROC_FD_CLOSE(fds[1], PROC_FD_CHILD);
	ops[2] = PROC_FD_DUP(co->gsmfd, 1, 0);
	ops[3] = 0;
	if (!(proc = procopen(cs->path, cs->argv, NiL, ops, PROC_DAEMON|PROC_IGNORE)))
	{
		errormsg(state.lib, ERROR_LIBRARY|ERROR_SYSTEM|2, "%s: cannot connect to %s service", cs->path, cs->name);
		close(fds[0]);	
		close(fds[1]);	
		return 0;
	}
	fcntl(cs->fd, F_SETFD, FD_CLOEXEC);
	cs->pid = proc->pid;
	procfree(proc);
	sfprintf(sp, "id=%d info\n", cj->id);
	n = sfstrtell(sp);
	if (!(s = costash(sp)))
		goto bad;
	if (write(cs->fd, s, n) != n || sfpoll(&co->msgfp, 1, 5 * 1000) <= 0)
		goto bad;
	cj->pid = 0;
	cj->status = 0;
	cj->local = 0;
	cj->service = cs;
	co->svc_outstanding++;
	co->svc_running++;
	if (!cowait(co, cj, -1))
		goto bad;
	return cj;
 bad:
	errormsg(state.lib, ERROR_LIBRARY|ERROR_SYSTEM|2, "%s: service not responding", cs->name);
 nospace:
	cj->pid = CO_PID_FREE;
	cs->pid = 0;
	close(cs->fd);
	cs->fd = -1;
	return 0;
}

static Cojob_t*
request(register Coshell_t* co, Cojob_t* cj, Coservice_t* cs, const char* action, int flags)
{
	ssize_t		n;
	ssize_t		i;
	Sfio_t*		sp;

	if (!(sp = sfstropen()))
	{
		errormsg(state.lib, ERROR_LIBRARY|2, "out of space");
		return 0;
	}
	if (!cs->fd && !service(co, cs, cj, flags, sp))
		goto bad;
	if (!cs->pid)
		goto bad;
	if (flags & CO_DEBUG)
		errormsg(state.lib, ERROR_LIBRARY|2, "job %d commands:\n\n%s %s\n", cj->id, cs->name, action);
	if (!(flags & CO_SILENT))
		sfprintf(sfstderr, "+ %s %s\n", cs->name, action);
	sfprintf(sp, "id=%d %s\n", cj->id, action);
	n = sfstrtell(sp);
	action = sfstrbase(sp);
	while ((i = write(cs->fd, action, n)) > 0 && (n -= i) > 0)
		action += i;
	sfstrclose(sp);
	if (n)
		goto bad;
	sfclose(sp);
	cj->pid = 0;
	cj->status = 0;
	cj->local = 0;
	cj->service = cs;
	co->svc_outstanding++;
	co->svc_running++;
	co->total++;
	return cj;
 bad:
	cj->pid = CO_PID_FREE;
	sfclose(sp);
	return 0;
}

Cojob_t*
coexec(register Coshell_t* co, const char* action, int flags, const char* out, const char* err, const char* att)
{
	register Cojob_t*	cj;
	register Sfio_t*	sp;
	register Coservice_t*	cs;
	int			n;
	int			i;
	int			og;
	int			cg;
	char*			s;
	char*			t;
	char*			env;
	char*			red;
	char*			sh[4];
	struct stat		sto;
	struct stat		ste;

	/*
	 * get a free job slot
	 */

	for (cj = co->jobs; cj; cj = cj->next)
		if (cj->pid == CO_PID_FREE)
			break;
	if (cj)
		cj->service = 0;
	else if (!(cj = vmnewof(co->vm, 0, Cojob_t, 1, 0)))
		return 0;
	else
	{
		cj->coshell = co;
		cj->pid = CO_PID_FREE;
		cj->id = ++co->slots;
		cj->next = co->jobs;
		co->jobs = cj;
	}

	/*
	 * set the flags
	 */

	flags &= ~co->mask;
	flags |= co->flags;
	cj->flags = flags;

	/*
	 * check service intercepts
	 */

	for (cs = co->service; cs; cs = cs->next)
	{
		for (s = cs->name, t = (char*)action; *s && *s == *t; s++, t++);
		if (!*s && *t == ' ')
			return request(co, cj, cs, t + 1, flags);
	}
	cj->flags &= ~CO_SERVICE;
	red = (cj->flags & CO_APPEND) ? ">>" : ">";

	/*
	 * package the action
	 */

	if (!(env = coinitialize(co, co->flags)))
		return 0;
	if (!(sp = sfstropen()))
		return 0;
	n = strlen(action);
	if (co->flags & CO_SERVER)
	{
		/*
		 * leave it to server
		 */

		sfprintf(sp, "#%05d\ne %d %d %s %s %s",
			0,
			cj->id,
			cj->flags,
			state.pwd,
			out,
			err);
		if (att)
			sfprintf(sp, " (%d:%s)", strlen(att), att);
		else
			sfprintf(sp, " %s", att);
		sfprintf(sp, " (%d:%s) (%d:%s)\n", strlen(env), env, n, action);
	}
	else if (co->flags & CO_INIT)
	{
		if (flags & CO_DEBUG)
			sfprintf(sp, "set -x\n");
		sfprintf(sp, "%s%s\necho x %d $? >&$%s\n",
			env,
			action,
			cj->id,
			CO_ENV_MSGFD);
	}
	else if (flags & CO_KSH)
	{
#if !_lib_fork && defined(_map_spawnve)
		Sfio_t*	tp;

		tp = sp;
		if (!(sp = sfstropen()))
			sp = tp;
#endif
		sfprintf(sp, "{\ntrap 'set %s$?; trap \"\" 0; IFS=\"\n\"; print -u$%s x %d $1 $(times); exit $1' 0 HUP INT QUIT TERM%s\n%s%s%s",
			(flags & CO_SILENT) ? "" : "+x ",
			CO_ENV_MSGFD,
			cj->id,
			(flags & CO_IGNORE) ? "" : " ERR",
			env,
			n > CO_MAXEVAL ? "" : "eval '",
			(flags & CO_SILENT) ? "" : "set -x\n");
		if (n > CO_MAXEVAL)
			sfputr(sp, action, -1);
		else
		{
			coquote(sp, action, 0);
			sfprintf(sp, "\n'");
		}
		sfprintf(sp, "\n} </dev/null");
		if (out)
		{
			if (*out == '/')
				sfprintf(sp, " %s%s", red, out);
			else
				sfprintf(sp, " %s%s/%s", red, state.pwd, out);
		}
		else if ((flags & CO_SERIALIZE) && (cj->out = pathtemp(NiL, 64, NiL, "coo", NiL)))
			sfprintf(sp, " >%s", cj->out);
		if (err)
		{
			if (out && streq(out, err))
				sfprintf(sp, " 2>&1");
			else if (*err == '/')
				sfprintf(sp, " 2%s%s", red, err);
			else
				sfprintf(sp, " 2%s%s/%s", red, state.pwd, err);
		}
		else if (flags & CO_SERIALIZE)
		{
			if (!out && !fstat(1, &sto) && !fstat(2, &ste) && sto.st_dev == ste.st_dev && sto.st_ino == ste.st_ino)
				sfprintf(sp, " 2>&1");
			else if (cj->err = pathtemp(NiL, 64, NiL, "coe", NiL))
				sfprintf(sp, " 2>%s", cj->err);
		}
#if !_lib_fork && defined(_map_spawnve)
		if (sp != tp)
		{
			sfprintf(tp, "%s -c '", state.sh);
			if (!(s = costash(sp)))
				return 0;
			coquote(tp, s, 0);
			sfprintf(tp, "'");
			sfstrclose(sp);
			sp = tp;
		}
#endif
		sfprintf(sp, " &\nprint -u$%s j %d $!\n",
			CO_ENV_MSGFD,
			cj->id);
	}
	else
	{
#if !_lib_fork && defined(_map_spawnve)
		Sfio_t*	tp;

		tp = sp;
		if (!(sp = sfstropen())) sp = tp;
#endif
		flags |= CO_IGNORE;
		if (co->mode & CO_MODE_SEPARATE)
		{
			flags &= ~CO_SERIALIZE;
			og = '{';
			cg = '}';
		}
		else
		{
			og = '(';
			cg = ')';
		}
		sfprintf(sp, "%c\n%s%sset -%s%s\n",
			og,
			env,
			n > CO_MAXEVAL ? "" : "eval '",
			(flags & CO_IGNORE) ? "" : "e",
			(flags & CO_SILENT) ? "" : "x");
		if (n > CO_MAXEVAL)
			sfprintf(sp, "%s", action);
		else
		{
			coquote(sp, action, 0);
			sfprintf(sp, "\n'");
		}
		sfprintf(sp, "\n%c </dev/null", cg);
		if (out)
		{
			if (*out == '/')
				sfprintf(sp, " %s%s", red, out);
			else
				sfprintf(sp, " %s%s/%s", red, state.pwd, out);
		}
		else if ((flags & CO_SERIALIZE) && (cj->out = pathtemp(NiL, 64, NiL, "coo", NiL)))
			sfprintf(sp, " >%s", cj->out);
		if (err)
		{
			if (out && streq(out, err))
				sfprintf(sp, " 2>&1");
			else if (*err == '/')
				sfprintf(sp, " 2%s%s", red, err);
			else
				sfprintf(sp, " 2%s%s/%s", red, state.pwd, err);
		}
		else if (flags & CO_SERIALIZE)
		{
			if (out)
				sfprintf(sp, " 2>&1");
			else if (cj->err = pathtemp(NiL, 64, NiL, "coe", NiL))
				sfprintf(sp, " 2>%s", cj->err);
		}
		if (!(co->mode & CO_MODE_SEPARATE))
		{
			if (flags & CO_OSH)
				sfprintf(sp, " && echo x %d 0 >&$%s || echo x %d $? >&$%s",
					cj->id,
					CO_ENV_MSGFD,
					cj->id,
					CO_ENV_MSGFD);
			else
				sfprintf(sp, " && echo x %d 0 `times` >&$%s || echo x %d $? `times` >&$%s",
					cj->id,
					CO_ENV_MSGFD,
					cj->id,
					CO_ENV_MSGFD);
		}
#if !_lib_fork && defined(_map_spawnve)
		if (sp != tp)
		{
			sfprintf(tp, "%s -c '", state.sh);
			if (!(s = costash(sp)))
				return 0;
			coquote(tp, s, 0);
			sfprintf(tp, "'");
			sfstrclose(sp);
			sp = tp;
		}
#endif
		if (!(co->mode & CO_MODE_SEPARATE))
			sfprintf(sp, " &\necho j %d $! >&$%s\n",
				cj->id,
				CO_ENV_MSGFD);
	}
	n = sfstrtell(sp);
	if (!costash(sp))
		return 0;
	if (flags & CO_SERVER)
		sfprintf(sp, "#%05d\n", n - 7);
	s = sfstrseek(sp, 0, SEEK_SET);
	if (flags & CO_DEBUG)
		errormsg(state.lib, ERROR_LIBRARY|2, "job %d commands:\n\n%s\n", cj->id, s);
	if (co->mode & CO_MODE_SEPARATE)
	{
		sh[0] = state.sh;
		sh[1] = "-c";
		sh[2] = s;
		sh[3] = 0;
		cj->status = procrun(state.sh, sh, 0);
		sfstrclose(sp);
		cj->pid = CO_PID_ZOMBIE;
		cj->local = 0;
		co->outstanding++;
		co->total++;
	}
	else
	{
		/*
		 * send it off
		 */

		while ((i = write(co->cmdfd, s, n)) > 0 && (n -= i) > 0)
			s += i;
		sfstrclose(sp);
		if (n)
			return 0;

		/*
		 * it's a job
		 */

		cj->pid = 0;
		cj->status = 0;
		cj->local = 0;
		co->outstanding++;
		co->running++;
		co->total++;
		if (co->mode & CO_MODE_ACK)
			cj = cowait(co, cj, -1);
	}
	return cj;
}