path: root/src/progman.c

/* This file is part of Jumper
   Copyright (C) 2013, 2017, 2020 Sergey Poznyakoff
  
   Jumper is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 3, or (at your option)
   any later version.
  
   Jumper is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.
  
   You should have received a copy of the GNU General Public License
   along with Jumper.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "jumper.h"
#include <sys/wait.h>
#include <time.h>
#include <fcntl.h>

struct process {              /* This describes a single running process */
	struct process *next, *prev;  /* pointers to previous and next process
					 in the list */
	enum process_type type;       /* process type */
	listener_t *lp;               /* corresponding listener */
	pid_t pid;                    /* PID */
	pid_t redir_pid[2];           /* PIDs of redirector processes (0 if
					 the slot is not used */
	/* Timeout information */
	time_t stop;                  /* stop-time for the process, if
					 type == type_action;
					 time of the next heartbeat, if
					 type == type_listener
				      */
	/* Apart from the usual next and prev links, action processes are
	   linked in a doubly-linked list ordered by their stop time.  The
	   ts_prev member points to the process with earlier stop time, and
	   the ts_next member points to the one which should be terminated
	   later. */
	struct process *ts_next, *ts_prev;
};

/* List of running processes */
static struct process *proc_list;
/* Ditto, ordered by stop time */
static struct process *ts_proc_list, *ts_proc_tail;
/* List of available process slots */
static struct process *proc_avail;

static unsigned proc_count; /* Total number of processes in proc_list */


static char *
proc_type_str(int type)
{
	switch (type) {
	case type_listener:
		return "listener";
	case type_redirector:
		return "redirector";
	case type_action:
		return "action";
	}
	return "unknown";
}

/* Basic functions for maintaining the proc_list */
static struct process *
proc_unlink(struct process **root, struct process *p)
{
	if (p->prev)
		p->prev->next = p->next;
	else
		*root = p->next;
	if (p->next)
		p->next->prev = p->prev;
	p->next = p->prev = NULL;
	return p;
}

static struct process *
proc_pop(struct process **pp)
{
	if (*pp)
		return proc_unlink(pp, *pp);
	return NULL;
}

static void
proc_push(struct process **pp, struct process *p)
{
	p->prev = NULL;
	p->next = *pp;
	if (*pp)
		(*pp)->prev = p;
	*pp = p;
}

/* Functions for maintaining the sorted list */
static void
proc_ts_link(struct process *p)
{
	debug(3, ("proc_ts_link: %lu, %d, %lu",
		  (unsigned long) p->pid, p->type,
		  (unsigned long) p->stop));
	if (!ts_proc_list) {
		p->ts_next = p->ts_prev = NULL;
		ts_proc_list = ts_proc_tail = p;
	} else if (p->stop > ts_proc_tail->stop) {
		p->ts_next = NULL;
		p->ts_prev = ts_proc_tail;
		ts_proc_tail->ts_next = p;
		ts_proc_tail = p;
	} else {
		struct process *t;
		
		for (t = ts_proc_list; t && t->stop < p->stop; t = t->ts_next)
			;

		p->ts_next = t;
		if (t->ts_prev)
			t->ts_prev->ts_next = p;
		else
			ts_proc_list = p;
		
		p->ts_prev = t->ts_prev;
		t->ts_prev = p;
	}
}

static void
proc_ts_unlink(struct process *p)
{
	if (p->stop == 0)
		return;
	if (p->ts_next)
		p->ts_next->ts_prev = p->ts_prev;
	else if (ts_proc_tail)
		ts_proc_tail = ts_proc_tail->ts_prev;
	if (p->ts_prev)
		p->ts_prev->ts_next = p->ts_next;
	else
		ts_proc_list = p->ts_next;
}

static void
proc_set_timeout(struct process *proc, unsigned timeout)
{
	proc->stop = time(NULL) + timeout;
	proc_ts_unlink(proc);
	proc_ts_link(proc);
}

/* Terminate the timed-out processes.  Return the number of seconds left
   to the expiration of the earliest available running action. */
unsigned
progman_expire(void)
{
	time_t t = time(NULL);
	struct process *p = ts_proc_list;

	while (p && p->stop <= t) {
		struct process *next = p->ts_next;
		
		switch (p->type) {
		case type_listener:
			debug(1, ("%s: heartbeat for pid %lu",
				  p->lp->id, (unsigned long) p->pid));
			listener_run_action(p->lp, event_heartbeat);
			proc_set_timeout(p, config.heartbeat);
			break;
		default:
			debug(1, ("terminating timed out process %lu (%d,%lu)",
				  (unsigned long) p->pid, p->type,
				  (unsigned long) p->stop));
			kill(p->pid, SIGKILL);
			proc_ts_unlink(p);
			if (p->redir_pid[REDIR_OUT])
				kill(p->redir_pid[REDIR_OUT], SIGKILL);
			if (p->redir_pid[REDIR_ERR])
				kill(p->redir_pid[REDIR_ERR], SIGKILL);
		}

		p = next;
	}
	if (p)
		return p->stop - t;
	return 0;
}

/* Fill in TV with the time left to the expiration of the earliest
   available running action.  Return TV. */
struct timeval *
progman_timeout(struct timeval *tv)
{
	if (ts_proc_list) {
		tv->tv_sec = progman_expire();
		tv->tv_usec = 0;
		return tv;
	}
	return NULL;
}


/* Register new process */
static struct process *
process_register(listener_t *lp, int type, unsigned timeout,
		 pid_t pid, pid_t *redir_pid)
{
	struct process *p;

	if (proc_avail)
		p = proc_pop(&proc_avail);
	else
		p = emalloc(sizeof(*p));
	memset(p, 0, sizeof(*p));
	p->lp = lp;
	if (type == type_listener) {
		lp->pid = pid;
		lp->status = stat_up;
	}
	p->type = type;
	p->pid = pid;
	if (timeout == 0)
		p->stop = 0;
	else {
		p->stop = time(NULL) + timeout;
		proc_ts_link(p);
	}
	if (redir_pid)
		memcpy(p->redir_pid, redir_pid, sizeof(p->redir_pid));
	else
		memset(p->redir_pid, 0, sizeof(p->redir_pid));
	proc_push(&proc_list, p);
	proc_count++;
	return p;
}

/* Find and return the process slot with the given PID. */
static struct process *
proc_lookup(pid_t pid)
{
	struct process *p;

	for (p = proc_list; p; p = p->next)
		if (p->pid == pid)
			return p;
	return NULL;
}

/* Kill all processes belonging to the given listener */
static void
proc_kill(listener_t *lp)
{
	struct process *p;

	for (p = proc_list; p; p = p->next)
		if (p->lp == lp && p->type != type_listener) {
			debug(1, ("killing pid %lu (%s)",
				  (unsigned long) p->pid,
				  proc_type_str(p->type)));
			kill(p->pid, SIGKILL);
		}
}

/* Collect terminated children. Safe to use in signal handlers. */
void
progman_cleanup(void)
{
	pid_t pid;
	int status;
	
	while ((pid = waitpid(-1, &status, WNOHANG)) > 0) {
		struct process *p = proc_lookup(pid);
		if (!p)
			continue;

		proc_ts_unlink(p);
		proc_unlink(&proc_list, p);
		proc_push(&proc_avail, p);
			
		proc_count--;

		switch (p->type) {
		case type_listener:
			proc_kill(p->lp);
			p->lp->progstat = status;
			p->lp->status = stat_term;
			listener_proc_report(p->lp);
			break;

		case type_redirector:
			break;

		case type_action:
			if (p->redir_pid[REDIR_OUT])
				kill(p->redir_pid[REDIR_OUT], SIGKILL);
			if (p->redir_pid[REDIR_ERR])
				kill(p->redir_pid[REDIR_ERR], SIGKILL);
			if (--p->lp->act_count == 0) {
				if (p->lp->status == stat_onexit)
					p->lp->status = stat_down;
			}
			break;
		}

	}
}

/* Terminate all running processes, first with SIGTERM and, if any of them
   are still running SHUTDOWN_TIMEOUT seconds after that, with SIGKILL.
*/
void
progman_terminate(void)
{
	struct process *p;
	time_t start;
	
	if (proc_count == 0)
		return;
	
	diag(LOG_INFO, "terminating active processes");
	
	debug(1, ("sending running processes the TERM signal"));
	for (p = proc_list; p; p = p->next)
		if (p->type != type_redirector)
			kill(p->pid, SIGTERM);
	
	start = time(NULL);
	while (proc_list) {
		progman_cleanup();
		if (time(NULL) - start > config.shutdown_timeout) {
			diag(LOG_NOTICE,
			     "sending running processes the KILL signal");
			for (p = proc_list; p; p = p->next)
				kill(p->pid, SIGKILL);
		}
		sleep(1);
	}
}

void
progman_decommission(void)
{
	struct process *p;
	time_t start;
	size_t count = 0;
	int info_shown = 0;
	
	if (proc_count == 0)
		return;
	
	for (p = proc_list; p; p = p->next)
		if (p->type != type_redirector && p->lp->decommission) {
			if (!info_shown) {
				diag(LOG_INFO,
				     "terminating decommissioned processes");
				debug(1,
				      ("sending running decommissioned processes the TERM signal"));
				info_shown = 1;
			}
			kill(p->pid, SIGTERM);
			count++;
		}

	start = time(NULL);
	while (count) {
		if (time(NULL) - start > config.shutdown_timeout) {
			diag(LOG_NOTICE,
			     "sending running processes the KILL signal");
			for (p = proc_list; p; p = p->next)
				if (p->lp->decommission)
					kill(p->pid, SIGKILL);
		}
		progman_cleanup();
		sleep(1);
		count = 0;
		for (p = proc_list; p; p = p->next)
			if (p->lp->decommission && p->lp->status != stat_down)
				++count;
	}
}

/* Close all file descriptors, except those in fdset. */
static void
close_fds(fd_set *fdset)
{
	int i;

	for (i = FD_SETSIZE-1; i >= 0; i--) {
		if (FD_ISSET(i, fdset))
			continue;
		close(i);
	}
}

static void
redir_exit(int sig)
{
	_exit(0);
}


/* Create a redirector, i.e. a process that reads its standard input and
   sends it to syslog with the priority PRIO.  Syslog records will be marked
   with TAG.  Upon successful return, stores the PID of the created process
   in RETURN_PID and returns the file descriptor sutable for redirecting
   output to it. */
static int
open_redirector(const char *tag, int prio, pid_t *return_pid)
{
	int p[2];
	FILE *fp;
	char buf[512];
	pid_t pid;
	fd_set fdset;

	if (pipe(p)) {
		diag(LOG_ERR,
		     "cannot start redirector for %s, pipe failed: %s",
		     tag, strerror(errno));
		return -1;
	}
	switch (pid = fork()) {
	case 0:
		/* Redirector process */
		FD_ZERO(&fdset);
		FD_SET(p[0], &fdset);
		if (log_to_stderr >= 0 || config.facility <= 0)
			FD_SET(2, &fdset);
		close_fds(&fdset);
		signal_setup(redir_exit);

		fp = fdopen(p[0], "r");
		if (fp == NULL)
			_exit(1);
		if (config.facility > 0) 
			openlog(tag, LOG_PID, config.facility);

		while (fgets(buf, sizeof(buf), fp) > 0) {
			int len = strlen(buf);
			if (len && buf[len-1] == '\n')
				buf[len-1] = 0;
			diag(prio, "%s", buf);
		}
		_exit(0);
      
	case -1:
		diag(LOG_CRIT,
		     "cannot run redirector `%s': fork failed: %s",
		     tag, strerror(errno));
		return -1;

	default:
		debug(2, ("redirector for %s started, pid=%lu",
			  tag, (unsigned long) pid));
		close(p[0]);

		*return_pid = pid;
		return p[1];
	}
}

/* Start the command PROG with the environment modified as per ENV.  Use
   optional KVE pairs for variable expansion in PROG and ENV. TYPE and LP
   supply  data necessary for registering in the process list.
   OPTIONS instruct what should be redirected to syslog.  TIMEOUT sets the
   maximum execution time (for type_action processes); 0 means no limit. */
int
progman_start(int type, listener_t *lp, struct grecs_locus *loc,
	      int options, unsigned timeout, char *prog, char **env,
	      char **kve)
{
	pid_t pid;
	int redir_fd[2] = { -1, -1 };
	pid_t redir_pid[2] = { 0, 0 };
	fd_set fdset;
	
	struct wordsplit ws;
	int flags = WRDSF_NOCMD;

	if (kve) {
		ws.ws_env = (const char **) kve;
		flags |= WRDSF_ENV | WRDSF_ENV_KV;
	}
	
	if (options & OPT_SHELL) {
		ws.ws_offs = 2;
		flags |= WRDSF_NOSPLIT | WRDSF_DOOFFS;
	} else {
		flags |= WRDSF_QUOTE | WRDSF_SQUEEZE_DELIMS |
			 WRDSF_CESCAPES;
	}

	if (wordsplit(prog, &ws, flags)) {
		diag(LOG_CRIT, "%s:%d: wordsplit: %s",
		     loc->beg.file, loc->beg.line,
		     wordsplit_strerror(&ws));
		return -1;
	}
	if (options & OPT_SHELL) {
		ws.ws_wordv[0] = "/bin/sh";
		ws.ws_wordv[1] = "-c";
	}

	if (config.debug_level >= 1) {
		int i;
		struct grecs_txtacc *acc = grecs_txtacc_create();
		char *locstr = NULL;
		size_t locsize = 0;
		
		grecs_txtacc_grow_string(acc, ws.ws_wordv[0]);
		for (i = 1; i < ws.ws_wordc + ws.ws_offs; i++) {
			grecs_txtacc_grow_char(acc, ' ');
			grecs_txtacc_grow_string_escape(acc, ws.ws_wordv[i]);
		}
		grecs_txtacc_grow_char(acc, 0);

		grecs_asprint_locus(&locstr, &locsize, loc);
		debugprt("%s: starting %s", locstr,
			 grecs_txtacc_finish(acc, 0));
		free(locstr);
		grecs_txtacc_free(acc);
	}
	
	if (options & OPT_STDERR) {
		redir_fd[REDIR_ERR] = open_redirector(ws.ws_wordv[0], LOG_ERR,
						      &redir_pid[REDIR_ERR]);
		process_register(lp, type_redirector, 0,
				 redir_pid[REDIR_ERR], NULL);
	}
	      
	if (options & OPT_STDOUT) {
		redir_fd[REDIR_OUT] = open_redirector(ws.ws_wordv[0], LOG_INFO,
						      &redir_pid[REDIR_OUT]);
		process_register(lp, type_redirector, 0,
				 redir_pid[REDIR_OUT], NULL);
	}
	
	
	pid = fork();
	if (pid == -1) {
		diag(LOG_ERR, "fork: %s", strerror(errno));
		close(redir_fd[REDIR_OUT]);
		close(redir_fd[REDIR_ERR]);
		kill(redir_pid[REDIR_OUT], SIGKILL);
		kill(redir_pid[REDIR_ERR], SIGKILL);
		return -1;
	}

	if (pid) {
		/* master */
		if (options & OPT_STDERR)
			close(redir_fd[REDIR_ERR]);
		if (options & OPT_STDOUT)
			close(redir_fd[REDIR_OUT]);
		process_register(lp, type, timeout, pid, redir_pid);
		wordsplit_free(&ws);
		return 0;
	}
		
	/* child */
	FD_ZERO(&fdset);
	if (options & OPT_NULLIN) {
		close(0);
		if (open("/dev/null", O_RDONLY) != 0) {
			diag(LOG_ERR, "can't open /dev/null: %s",
			     strerror(errno));
			_exit(127);
		}
		FD_SET(0, &fdset);
	}
	
	if (redir_fd[REDIR_OUT] != -1) {
		if (redir_fd[REDIR_OUT] != 1 &&
		    dup2(redir_fd[REDIR_OUT], 1) == -1) {
			diag(LOG_ERR, "dup2: %s", strerror(errno));
			_exit(127);
		}
		FD_SET(1, &fdset);
	}
	if (redir_fd[REDIR_ERR] != -1) {
		if (redir_fd[REDIR_ERR] != 2 &&
		    dup2(redir_fd[REDIR_ERR], 2) == -1) {
			diag(LOG_ERR, "dup2: %s", strerror(errno));
			_exit(127);
		}
		FD_SET(2, &fdset);
	}
	close_fds(&fdset);
	signal_setup(SIG_DFL);
	execve(ws.ws_wordv[0], ws.ws_wordv, environ_setup(env, kve));
	diag(LOG_CRIT, "execve(%s): %s", ws.ws_wordv[0], strerror(errno));
	_exit(127);
}