path: root/libexec_run_pipeline.c

/* See LICENSE file for copyright and license details. */
#include "common.h"
#ifndef TEST


struct reap_context {
	pid_t *pids;
	size_t npids;
	size_t procs_rem;
	int *exit_statuses_out; /* nullable */
};


struct reaped {
	pid_t pid;
	int status;
};


static struct reaped *reaped_elsewhere = NULL;
static size_t nreaped_elsewhere = 0;
static size_t reaped_elsewhere_size = 0;


static void
sigchld_handler(int signo)
{
	(void) signo;
}


static int
unlist_reap(pid_t pid, int *status)
{
	size_t i;

	for (i = 0; i < nreaped_elsewhere; i++)
		if (reaped_elsewhere[i].pid == pid)
			goto found;
	return 0;

found:
	*status = reaped_elsewhere[i].status;
	reaped_elsewhere[i] = reaped_elsewhere[--nreaped_elsewhere];

	if (nreaped_elsewhere <= reaped_elsewhere_size / 4) {
		free(reaped_elsewhere);
		reaped_elsewhere = NULL;
		nreaped_elsewhere = 0;
		reaped_elsewhere_size = 0;
	}

	return 1;
}


static int
reap_child(struct reap_context *ctx, pid_t pid)
{
	size_t i;
	int status;

	if (!ctx->procs_rem)
		return 0;

	for (i = 0; i < ctx->npids; i++) {
		if (ctx->pids[i] == pid) {
			if (waitpid(pid, &status, WNOHANG) != pid)
				return -1;
			if (ctx->exit_statuses_out)
				ctx->exit_statuses_out[i] = status;
			ctx->procs_rem -= 1;
			ctx->pids[i] = -1;
			while (ctx->npids && ctx->pids[ctx->npids - 1] == -1)
				ctx->npids -= 1;
			return 1;
		}
	}

	return 0;
}


static int
reap_children(struct reap_context *reapctx, int flags,
              int (*reap_mutex_control)(int action /* -1 = acquire, +1 = release */, void *user4), void *user4,
              int (*on_alien_child_death)(pid_t pid, void *user2), void *user2)
{
	siginfo_t info;
	int r, status;
	void *new;
	size_t new_size;
	size_t i;

	if (reap_mutex_control)
		if ((*reap_mutex_control)(-1, user4))
			goto fail;

	if (nreaped_elsewhere) {
		for (i = 0; i < reapctx->npids; i++) {
			if (unlist_reap(reapctx->pids[i], &status)) {
				if (reapctx->exit_statuses_out)
					reapctx->exit_statuses_out[i] = status;
				reapctx->procs_rem -= 1;
				reapctx->pids[i] = -1;
			}
		}
		while (reapctx->npids && reapctx->pids[reapctx->npids - 1] == -1)
			reapctx->npids -= 1;
	}

	for (;;) {
		info.si_pid = 0;
		if (waitid(P_ALL, 0, &info, flags | WNOWAIT))
			goto fail;
		if (info.si_pid < 1)
			break;

		r = reap_child(reapctx, info.si_pid);
		if (r < 0)
			goto fail;
		if (r > 0)
			continue;

		if (on_alien_child_death) {
			r = (*on_alien_child_death)(info.si_pid, user2);
			if (r < 0)
				goto fail;
			if (r > 0)
				continue;
		}

		if (nreaped_elsewhere == reaped_elsewhere_size) {
			new_size = reaped_elsewhere_size + 4;
			new = realloc(reaped_elsewhere, new_size * sizeof(*reaped_elsewhere));
			if (!new)
				goto fail;
			reaped_elsewhere = new;
			reaped_elsewhere_size = new_size;
		}

		if (waitpid(info.si_pid, &status, WNOHANG) != info.si_pid) {
			if (!nreaped_elsewhere) {
				free(reaped_elsewhere);
				reaped_elsewhere = NULL;
				reaped_elsewhere_size = 0;
			}
			goto fail;
		}

		reaped_elsewhere[nreaped_elsewhere].pid = info.si_pid;
		reaped_elsewhere[nreaped_elsewhere].status = status;
		nreaped_elsewhere++;
	}

	if (reap_mutex_control)
		if ((*reap_mutex_control)(+1, user4))
			goto fail;
	return 0;

fail:
	if (reap_mutex_control)
		if ((*reap_mutex_control)(+1, user4))
			goto fail;
	return -1;
}


int
libexec_run_pipeline(int (*on_alien_epoll)(int alien_epoll, uint32_t events, void *user1), int alien_epoll, void *user1,
                     int (*on_alien_child_death)(pid_t pid, void *user2), void *user2,
                     int (*after_fork)(struct libexec_command *cmd, int new_fd, void *user3), void *user3,
                     int (*reap_mutex_control)(int action /* -1 = acquire, +1 = release */, void *user4), void *user4,
		     int (*on_interrupt)(void *user5), void *user5,
                     const sigset_t *sigmask /* nullable */,
                     struct libexec_document *const *docs, size_t ndocs, int docs_1_level,
                     struct libexec_command *const *cmds, int *exit_statuses_out /* nullable */, size_t ncmds)
{
	struct libexec_document *input_docs = NULL;
	size_t i, ninput_docs = 0, files_open;
	struct epoll_event evs[8];
	int epfd, n, sigchld_set = 0, error, r;
	struct libexec_document *doc;
	struct sigaction sigchld, old_sigchld;
	struct reap_context reapctx = {
		.pids = NULL,
		.npids = 0,
		.procs_rem = 0,
		.exit_statuses_out = exit_statuses_out
	};

	if (!ncmds)
		return 0;

	memset(evs, 0, sizeof(evs));

	epfd = epoll_create1(EPOLL_CLOEXEC);
	if (epfd < 0)
		return -1;

	if (sigaction(SIGCHLD, NULL, &old_sigchld))
		goto fail;
	if (!(old_sigchld.sa_flags & SA_SIGINFO)) {
		if (old_sigchld.sa_handler == SIG_IGN || old_sigchld.sa_handler == SIG_DFL) {
			memset(&sigchld, 0, sizeof(sigchld));
			sigchld.sa_handler = &sigchld_handler;
			sigemptyset(&sigchld.sa_mask);
			if (sigaction(SIGCHLD, &sigchld, NULL))
				goto fail;
			sigchld_set = 1;
		}
	}

	if (alien_epoll >= 0) {
		evs[0].events = EPOLLIN;
		evs[0].data.ptr = NULL;
		if (epoll_ctl(epfd, EPOLL_CTL_ADD, alien_epoll, &evs[0]))
			goto fail;
	}

	reapctx.pids = calloc(ncmds, sizeof(*reapctx.pids));
	if (!reapctx.pids)
		goto fail;

	for (i = 0; i < ncmds; i++)
		if (libexec_spawn(&reapctx.pids[reapctx.npids++ /* may be set on failure */], cmds[i],
		                  after_fork, user3, &input_docs, &ninput_docs, O_CLOEXEC | O_NONBLOCK))
			goto fail;

	for (i = 0; i < ninput_docs; i++) {
		evs[0].events = EPOLLOUT;
		evs[0].data.ptr = &input_docs[i];
		if (epoll_ctl(epfd, EPOLL_CTL_ADD, input_docs[i].fd, &evs[0]))
			goto fail;
	}

	for (i = 0; i < ndocs; i++) {
		doc = docs_1_level ? &(*docs)[i] : docs[i];
		evs[0].events = EPOLLIN;
		evs[0].data.ptr = doc;
		doc->user = 1;
		if (epoll_ctl(epfd, EPOLL_CTL_ADD, doc->fd, &evs[0]))
			goto fail;
	}

	reapctx.procs_rem = ncmds;

	files_open = ninput_docs + ndocs + (size_t)(alien_epoll >= 0);
	while (files_open) {
		if (reap_children(&reapctx, WNOHANG, reap_mutex_control, user4, on_alien_child_death, user2))
			goto fail;

		n = epoll_pwait(epfd, evs, sizeof(evs) / sizeof(*evs), -1, sigmask);
		if (n < 0) {
			if (errno != EINTR)
				goto fail;
			if (on_interrupt)
				if ((*on_interrupt)(user5))
					goto fail;
			continue;
		}

		for (i = 0; i < (size_t)n; i++) {
			if (evs[i].data.ptr == NULL) {
				r = (*on_alien_epoll)(alien_epoll, evs[i].events, user1);
				if (r < 0) {
					goto fail;
				} else if (r > 0) {
					if (epoll_ctl(epfd, EPOLL_CTL_DEL, alien_epoll, &evs[i]))
						goto fail;
					files_open -= 1;
				}
			} else {
				doc = evs[i].data.ptr;
			send_recv_again:
				if (doc->user ? libexec_recv_document(doc) : libexec_send_document(doc)) {
					if (errno == EAGAIN)
						continue;
					if (errno == EINTR) {
						if (on_interrupt)
							if ((*on_interrupt)(user5))
								goto fail;
						goto send_recv_again;
					}
				} else {
					files_open -= 1;
					if (doc->user) {
						close(doc->fd);
						doc->fd = -1;
					} else {
						libexec_destroy_document(doc);
					}
				}
			}
		}
	}
	free(input_docs);
	close(epfd);

	while (reapctx.procs_rem)
		if (reap_children(&reapctx, 0, reap_mutex_control, user4, on_alien_child_death, user2))
			goto fail;

	free(reapctx.pids);
	if (sigchld_set)
		if (sigaction(SIGCHLD, &old_sigchld, NULL))
			abort();

	return 0;

fail:
	error = errno;
	while (ninput_docs)
		libexec_destroy_document(&input_docs[--ninput_docs]);
	free(input_docs);
	close(epfd);
	if (reapctx.npids) {
		if (reap_mutex_control)
			if ((*reap_mutex_control)(+1, user4))
				abort();
		for (i = 0; i < reapctx.npids; i++)
			if (reapctx.pids[i] >= 1)
				if (unlist_reap(reapctx.pids[i], &(int){0}) || kill(reapctx.pids[i], SIGKILL))
					reapctx.pids[i] = -1;
		for (i = 0; i < reapctx.npids; i++)
			if (reapctx.pids[i] >= 1)
				waitpid(reapctx.pids[i], &(int){0}, 0);
		if (reap_mutex_control)
			if ((*reap_mutex_control)(-1, user4))
				abort();
	}
	free(reapctx.pids);
	if (sigchld_set)
		if (sigaction(SIGCHLD, &old_sigchld, NULL))
			abort();
	errno = error;
	return -1;
}


#else
LIBEXEC_CONST__ int main(void) {return 0;} /* TODO test */
#endif