/* Copyright (C) 1989, 1990, 1991, 1992, 2000, 2001, 2002, 2003
* Free Software Foundation, Inc.
* Copyright (C) 2003-2017 Colin Watson.
* Written for groff by James Clark (jjc@jclark.com)
* Heavily adapted and extended for man-db by Colin Watson.
*
* This file is part of libpipeline.
*
* libpipeline 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 2 of the License, or (at
* your option) any later version.
*
* libpipeline 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 libpipeline; if not, write to the Free Software
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301
* USA.
*/
#ifdef HAVE_CONFIG_H
# include "config.h"
#endif
#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <errno.h>
#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdarg.h>
#include <assert.h>
#include <string.h>
#include <sys/wait.h>
#include "dirname.h"
#include "full-write.h"
#include "safe-read.h"
#include "safe-write.h"
#include "xalloc.h"
#include "xstrndup.h"
#include "xvasprintf.h"
#include "pipeline-private.h"
#include "error.h"
#include "pipeline.h"
#ifdef USE_SOCKETPAIR_PIPE
# include <netdb.h>
# include <netinet/in.h>
# include <sys/socket.h>
# ifdef CORRECT_SOCKETPAIR_MODE
# include <sys/stat.h>
# endif
# ifndef SHUT_RD
# define SHUT_RD 0
# endif
# ifndef SHUT_WR
# define SHUT_WR 1
# endif
# ifdef CORRECT_SOCKETPAIR_MODE
# define pipe(p) (((socketpair(AF_UNIX,SOCK_STREAM,0,p) < 0) || \
(shutdown((p)[1],SHUT_RD) < 0) || (fchmod((p)[1],S_IWUSR) < 0) || \
(shutdown((p)[0],SHUT_WR) < 0) || (fchmod((p)[0],S_IRUSR) < 0)) ? -1 : 0)
# else
# define pipe(p) (((socketpair(AF_UNIX,SOCK_STREAM,0,p) < 0) || \
(shutdown((p)[1],SHUT_RD) < 0) || (shutdown((p)[0],SHUT_WR) < 0)) ? -1 : 0)
# endif
#endif
#if defined(HAVE_SETENV) && !defined(HAVE_CLEARENV)
int clearenv (void)
{
/* According to:
* http://hg.dovecot.org/dovecot-2.0/file/74d9f61e224d/src/lib/env-util.c#l56
* simply setting "environ = NULL" crashes some systems. Creating a
* new environment consisting of just a terminator is indeed
* probably the best we can do.
*/
environ = XCALLOC (1, char *);
return 0;
}
#endif
/* ---------------------------------------------------------------------- */
/* Functions to build individual commands. */
pipecmd *pipecmd_new (const char *name)
{
pipecmd *cmd = XMALLOC (pipecmd);
struct pipecmd_process *cmdp;
char *name_base;
cmd->tag = PIPECMD_PROCESS;
cmd->name = xstrdup (name);
cmd->nice = 0;
cmd->discard_err = 0;
cmd->cwd_fd = -1;
cmd->cwd = NULL;
cmd->nenv = 0;
cmd->env_max = 4;
cmd->env = xnmalloc (cmd->env_max, sizeof *cmd->env);
cmd->pre_exec_func = NULL;
cmd->pre_exec_free_func = NULL;
cmd->pre_exec_data = NULL;
cmdp = &cmd->u.process;
cmdp->argc = 0;
cmdp->argv_max = 4;
cmdp->argv = xnmalloc (cmdp->argv_max, sizeof *cmdp->argv);
/* argv[0] is the basename of the command name. */
name_base = base_name (name);
pipecmd_arg (cmd, name_base);
free (name_base);
return cmd;
}
pipecmd *pipecmd_new_argv (const char *name, va_list argv)
{
pipecmd *cmd = pipecmd_new (name);
pipecmd_argv (cmd, argv);
return cmd;
}
pipecmd *pipecmd_new_args (const char *name, ...)
{
va_list argv;
pipecmd *cmd;
va_start (argv, name);
cmd = pipecmd_new_argv (name, argv);
va_end (argv);
return cmd;
}
/* As suggested in the header file, this function (for pipecmd_new_argstr()
* and pipecmd_argstr()) is really a wart. If we didn't have to worry about
* old configuration files then it wouldn't be necessary. Worse, the
* definition for tr in man_db.conf currently contains single-quoting, and
* people probably took that as a licence to do similar things, so we're
* obliged to worry about quoting as well!
*
* However, we can mitigate this; shell quoting alone is safe though
* sometimes confusing, but it's other shell constructs that tend to cause
* real security holes. Therefore, rather than punting to 'sh -c' or
* whatever, we parse a safe subset manually. Environment variables are not
* currently handled because of tricky word splitting issues, but in
* principle they could be if there's demand for it.
*
* TODO: Support setting environment variables.
*/
static char *argstr_get_word (const char **argstr)
{
char *out = NULL;
const char *litstart = *argstr;
enum { NONE, SINGLE, DOUBLE } quotemode = NONE;
while (**argstr) {
char backslashed[2];
/* If it's just a literal character, go round again. */
if ((quotemode == NONE && !strchr (" \t'\"\\", **argstr)) ||
/* nothing is special in '; terminated by ' */
(quotemode == SINGLE && **argstr != '\'') ||
/* \ is special in "; terminated by " */
(quotemode == DOUBLE && !strchr ("\"\\", **argstr))) {
++*argstr;
continue;
}
/* Within "", \ is only special when followed by $, `, ", or
* \ (or <newline> in a real shell, but we don't do that).
*/
if (quotemode == DOUBLE && **argstr == '\\' &&
!strchr ("$`\"\\", *(*argstr + 1))) {
++*argstr;
continue;
}
/* Copy any accumulated literal characters. */
if (litstart < *argstr) {
char *tmp = xstrndup (litstart, *argstr - litstart);
out = appendstr (out, tmp, NULL);
free (tmp);
}
switch (**argstr) {
case ' ':
case '\t':
/* End of word; skip over extra whitespace. */
while (*++*argstr)
if (!strchr (" \t", **argstr))
break;
return out;
case '\'':
if (quotemode != NONE)
quotemode = NONE;
else
quotemode = SINGLE;
litstart = ++*argstr;
break;
case '"':
if (quotemode != NONE)
quotemode = NONE;
else
quotemode = DOUBLE;
litstart = ++*argstr;
break;
case '\\':
backslashed[0] = *++*argstr;
if (!backslashed[0]) {
/* Unterminated quoting; give up. */
free (out);
return NULL;
}
backslashed[1] = '\0';
out = appendstr (out, backslashed, NULL);
litstart = ++*argstr;
break;
default:
assert (!"unexpected state parsing argstr");
}
}
if (quotemode != NONE) {
/* Unterminated quoting; give up. */
free (out);
return NULL;
}
/* Copy any accumulated literal characters. */
if (litstart < *argstr) {
char *tmp = xstrndup (litstart, *argstr - litstart);
out = appendstr (out, tmp, NULL);
free (tmp);
}
return out;
}
pipecmd *pipecmd_new_argstr (const char *argstr)
{
pipecmd *cmd;
char *arg;
arg = argstr_get_word (&argstr);
if (!arg)
error (FATAL, 0,
"badly formed configuration directive: '%s'",
argstr);
if (!strcmp (arg, "exec")) {
/* Some old configuration files have "exec command" rather
* than "command"; this worked fine when being evaluated by
* a shell, but since exec is a shell builtin it doesn't
* work when being executed directly. To work around this,
* we just drop "exec" if it appears at the start of argstr.
*/
free (arg);
arg = argstr_get_word (&argstr);
if (!arg)
error (FATAL, 0,
"badly formed configuration directive: '%s'",
argstr);
}
cmd = pipecmd_new (arg);
free (arg);
while ((arg = argstr_get_word (&argstr))) {
pipecmd_arg (cmd, arg);
free (arg);
}
return cmd;
}
pipecmd *pipecmd_new_function (const char *name,
pipecmd_function_type *func,
pipecmd_function_free_type *free_func,
void *data)
{
pipecmd *cmd = XMALLOC (pipecmd);
struct pipecmd_function *cmdf;
cmd->tag = PIPECMD_FUNCTION;
cmd->name = xstrdup (name);
cmd->nice = 0;
cmd->discard_err = 0;
cmd->cwd_fd = -1;
cmd->cwd = NULL;
cmd->nenv = 0;
cmd->env_max = 4;
cmd->env = xnmalloc (cmd->env_max, sizeof *cmd->env);
cmd->pre_exec_func = NULL;
cmd->pre_exec_free_func = NULL;
cmd->pre_exec_data = NULL;
cmdf = &cmd->u.function;
cmdf->func = func;
cmdf->free_func = free_func;
cmdf->data = data;
return cmd;
}
pipecmd *pipecmd_new_sequencev (const char *name, va_list cmdv)
{
pipecmd *cmd = XMALLOC (pipecmd);
struct pipecmd_sequence *cmds;
pipecmd *child;
cmd->tag = PIPECMD_SEQUENCE;
cmd->name = xstrdup (name);
cmd->nice = 0;
cmd->discard_err = 0;
cmd->cwd_fd = -1;
cmd->cwd = NULL;
cmd->nenv = 0;
cmd->env_max = 4;
cmd->env = xnmalloc (cmd->env_max, sizeof *cmd->env);
cmd->pre_exec_func = NULL;
cmd->pre_exec_free_func = NULL;
cmd->pre_exec_data = NULL;
cmds = &cmd->u.sequence;
cmds->ncommands = 0;
cmds->commands_max = 4;
cmds->commands = xnmalloc (cmds->commands_max, sizeof *cmds->commands);
child = va_arg (cmdv, pipecmd *);
while (child) {
pipecmd_sequence_command (cmd, child);
child = va_arg (cmdv, pipecmd *);
}
return cmd;
}
pipecmd *pipecmd_new_sequence (const char *name, ...)
{
va_list cmdv;
pipecmd *cmd;
va_start (cmdv, name);
cmd = pipecmd_new_sequencev (name, cmdv);
va_end (cmdv);
return cmd;
}
static void passthrough (void *data PIPELINE_ATTR_UNUSED)
{
for (;;) {
char buffer[4096];
int r = safe_read (STDIN_FILENO, buffer, 4096);
if (r <= 0)
break;
if (full_write (STDOUT_FILENO, buffer,
(size_t) r) < (size_t) r)
break;
}
return;
}
pipecmd *pipecmd_new_passthrough (void)
{
return pipecmd_new_function ("cat", &passthrough, NULL, NULL);
}
pipecmd *pipecmd_dup (pipecmd *cmd)
{
pipecmd *newcmd = XMALLOC (pipecmd);
int i;
newcmd->tag = cmd->tag;
newcmd->name = xstrdup (cmd->name);
newcmd->nice = cmd->nice;
newcmd->discard_err = cmd->discard_err;
newcmd->cwd_fd = cmd->cwd_fd;
newcmd->cwd = cmd->cwd ? xstrdup (cmd->cwd) : NULL;
newcmd->nenv = cmd->nenv;
newcmd->env_max = cmd->env_max;
assert (newcmd->nenv <= newcmd->env_max);
newcmd->env = xmalloc (newcmd->env_max * sizeof *newcmd->env);
newcmd->pre_exec_func = cmd->pre_exec_func;
newcmd->pre_exec_free_func = cmd->pre_exec_free_func;
newcmd->pre_exec_data = cmd->pre_exec_data;
for (i = 0; i < cmd->nenv; ++i) {
newcmd->env[i].name =
cmd->env[i].name ? xstrdup (cmd->env[i].name) : NULL;
newcmd->env[i].value =
cmd->env[i].value ? xstrdup (cmd->env[i].value) : NULL;
}
switch (newcmd->tag) {
case PIPECMD_PROCESS: {
struct pipecmd_process *cmdp = &cmd->u.process;
struct pipecmd_process *newcmdp = &newcmd->u.process;
newcmdp->argc = cmdp->argc;
newcmdp->argv_max = cmdp->argv_max;
assert (newcmdp->argc < newcmdp->argv_max);
newcmdp->argv = xmalloc
(newcmdp->argv_max * sizeof *newcmdp->argv);
for (i = 0; i < cmdp->argc; ++i)
newcmdp->argv[i] = xstrdup (cmdp->argv[i]);
newcmdp->argv[cmdp->argc] = NULL;
break;
}
case PIPECMD_FUNCTION: {
struct pipecmd_function *cmdf = &cmd->u.function;
struct pipecmd_function *newcmdf = &newcmd->u.function;
newcmdf->func = cmdf->func;
newcmdf->free_func = cmdf->free_func;
newcmdf->data = cmdf->data;
break;
}
case PIPECMD_SEQUENCE: {
struct pipecmd_sequence *cmds = &cmd->u.sequence;
struct pipecmd_sequence *newcmds = &newcmd->u.sequence;
newcmds->ncommands = cmds->ncommands;
newcmds->commands_max = cmds->commands_max;
assert (newcmds->ncommands <= newcmds->commands_max);
newcmds->commands = xmalloc
(newcmds->commands_max *
sizeof *newcmds->commands);
for (i = 0; i < cmds->ncommands; ++i)
newcmds->commands[i] =
pipecmd_dup (cmds->commands[i]);
break;
}
}
return newcmd;
}
void pipecmd_arg (pipecmd *cmd, const char *arg)
{
struct pipecmd_process *cmdp;
assert (cmd->tag == PIPECMD_PROCESS);
cmdp = &cmd->u.process;
if (cmdp->argc + 1 >= cmdp->argv_max) {
cmdp->argv_max *= 2;
cmdp->argv = xrealloc (cmdp->argv,
cmdp->argv_max * sizeof *cmdp->argv);
}
cmdp->argv[cmdp->argc++] = xstrdup (arg);
assert (cmdp->argc < cmdp->argv_max);
cmdp->argv[cmdp->argc] = NULL;
}
void pipecmd_argf (pipecmd *cmd, const char *format, ...)
{
va_list argv;
char *arg;
va_start (argv, format);
arg = xvasprintf (format, argv);
pipecmd_arg (cmd, arg);
free (arg);
va_end (argv);
}
void pipecmd_argv (pipecmd *cmd, va_list argv)
{
const char *arg = va_arg (argv, const char *);
assert (cmd->tag == PIPECMD_PROCESS);
while (arg) {
pipecmd_arg (cmd, arg);
arg = va_arg (argv, const char *);
}
}
void pipecmd_args (pipecmd *cmd, ...)
{
va_list argv;
assert (cmd->tag == PIPECMD_PROCESS);
va_start (argv, cmd);
pipecmd_argv (cmd, argv);
va_end (argv);
}
void pipecmd_argstr (pipecmd *cmd, const char *argstr)
{
char *arg;
assert (cmd->tag == PIPECMD_PROCESS);
while ((arg = argstr_get_word (&argstr))) {
pipecmd_arg (cmd, arg);
free (arg);
}
}
int pipecmd_get_nargs (pipecmd *cmd)
{
struct pipecmd_process *cmdp;
assert (cmd->tag == PIPECMD_PROCESS);
cmdp = &cmd->u.process;
return cmdp->argc;
}
void pipecmd_nice (pipecmd *cmd, int value)
{
cmd->nice = value;
}
void pipecmd_discard_err (pipecmd *cmd, int discard_err)
{
cmd->discard_err = discard_err;
}
void pipecmd_chdir (pipecmd *cmd, const char *directory)
{
free (cmd->cwd);
cmd->cwd = xstrdup (directory);
}
void pipecmd_fchdir (pipecmd *cmd, int directory_fd)
{
cmd->cwd_fd = directory_fd;
}
void pipecmd_setenv (pipecmd *cmd, const char *name, const char *value)
{
if (cmd->nenv >= cmd->env_max) {
cmd->env_max *= 2;
cmd->env = xrealloc (cmd->env,
cmd->env_max * sizeof *cmd->env);
}
cmd->env[cmd->nenv].name = xstrdup (name);
cmd->env[cmd->nenv].value = xstrdup (value);
++cmd->nenv;
}
void pipecmd_unsetenv (pipecmd *cmd, const char *name)
{
if (cmd->nenv >= cmd->env_max) {
cmd->env_max *= 2;
cmd->env = xrealloc (cmd->env,
cmd->env_max * sizeof *cmd->env);
}
cmd->env[cmd->nenv].name = xstrdup (name);
cmd->env[cmd->nenv].value = NULL;
++cmd->nenv;
}
void pipecmd_clearenv (pipecmd *cmd)
{
if (cmd->nenv >= cmd->env_max) {
cmd->env_max *= 2;
cmd->env = xrealloc (cmd->env,
cmd->env_max * sizeof *cmd->env);
}
cmd->env[cmd->nenv].name = NULL;
cmd->env[cmd->nenv].value = NULL;
++cmd->nenv;
}
void pipecmd_pre_exec (pipecmd *cmd,
pipecmd_function_type *func,
pipecmd_function_free_type *free_func,
void *data)
{
cmd->pre_exec_func = func;
cmd->pre_exec_free_func = free_func;
cmd->pre_exec_data = data;
}
void pipecmd_sequence_command (pipecmd *cmd, pipecmd *child)
{
struct pipecmd_sequence *cmds;
assert (cmd->tag == PIPECMD_SEQUENCE);
cmds = &cmd->u.sequence;
if (cmds->ncommands >= cmds->commands_max) {
cmds->commands_max *= 2;
cmds->commands = xrealloc
(cmds->commands,
cmds->commands_max * sizeof *cmds->commands);
}
cmds->commands[cmds->ncommands++] = child;
}
void pipecmd_dump (pipecmd *cmd, FILE *stream)
{
int i;
if (cmd->cwd_fd >= 0)
fprintf (stream, "(cd <fd %d> && ", cmd->cwd_fd);
else if (cmd->cwd)
fprintf (stream, "(cd %s && ", cmd->cwd);
for (i = 0; i < cmd->nenv; ++i) {
if (cmd->env[i].name)
fprintf (stream, "%s=%s ",
cmd->env[i].name,
cmd->env[i].value ? cmd->env[i].value
: "<unset>");
else
fputs ("env -i ", stream);
}
switch (cmd->tag) {
case PIPECMD_PROCESS: {
struct pipecmd_process *cmdp = &cmd->u.process;
fputs (cmd->name, stream);
for (i = 1; i < cmdp->argc; ++i) {
/* TODO: escape_shell()? */
putc (' ', stream);
fputs (cmdp->argv[i], stream);
}
break;
}
case PIPECMD_FUNCTION:
fputs (cmd->name, stream);
break;
case PIPECMD_SEQUENCE: {
struct pipecmd_sequence *cmds = &cmd->u.sequence;
putc ('(', stream);
for (i = 0; i < cmds->ncommands; ++i) {
pipecmd_dump (cmds->commands[i], stream);
if (i < cmds->ncommands - 1)
fputs (" && ", stream);
}
putc (')', stream);
break;
}
}
if (cmd->cwd_fd >= 0 || cmd->cwd)
putc (')', stream);
}
char *pipecmd_tostring (pipecmd *cmd)
{
char *out = NULL;
int i;
if (cmd->cwd_fd >= 0) {
char *cwd_fd_str = xasprintf ("%d", cmd->cwd_fd);
out = appendstr (out, "(cd <fd ", cwd_fd_str, "> && ", NULL);
free (cwd_fd_str);
} else if (cmd->cwd)
out = appendstr (out, "(cd ", cmd->cwd, " && ", NULL);
for (i = 0; i < cmd->nenv; ++i) {
if (cmd->env[i].name)
out = appendstr (out, cmd->env[i].name, "=",
cmd->env[i].value ? cmd->env[i].value
: "<unset>",
" ", NULL);
else
out = appendstr (out, "env -i ", NULL);
}
switch (cmd->tag) {
case PIPECMD_PROCESS: {
struct pipecmd_process *cmdp = &cmd->u.process;
out = appendstr (out, cmd->name, NULL);
for (i = 1; i < cmdp->argc; ++i)
/* TODO: escape_shell()? */
out = appendstr (out, " ", cmdp->argv[i],
NULL);
break;
}
case PIPECMD_FUNCTION:
out = appendstr (out, cmd->name, NULL);
break;
case PIPECMD_SEQUENCE: {
struct pipecmd_sequence *cmds = &cmd->u.sequence;
out = appendstr (out, "(", NULL);
for (i = 0; i < cmds->ncommands; ++i) {
char *subout = pipecmd_tostring
(cmds->commands[i]);
out = appendstr (out, subout, NULL);
free (subout);
if (i < cmds->ncommands - 1)
out = appendstr (out, " && ", NULL);
}
out = appendstr (out, ")", NULL);
break;
}
}
if (cmd->cwd_fd >= 0 || cmd->cwd)
out = appendstr (out, ")", NULL);
return out;
}
/* Children exit with this status if execvp fails. */
#define EXEC_FAILED_EXIT_STATUS 0xff
/* When called internally during pipeline execution, this is called in the
* forked child process, with file descriptors already set up.
*/
void pipecmd_exec (pipecmd *cmd)
{
int i;
if (cmd->nice)
if (nice (cmd->nice) < 0)
/* Don't worry too much. */
debug ("nice failed: %s\n", strerror (errno));
if (cmd->discard_err) {
int devnull = open ("/dev/null", O_WRONLY);
if (devnull != -1) {
dup2 (devnull, 2);
close (devnull);
}
}
if (cmd->cwd_fd >= 0) {
if (fchdir (cmd->cwd_fd) < 0)
error (EXEC_FAILED_EXIT_STATUS, errno,
"can't change directory to fd %d", cmd->cwd_fd);
} else if (cmd->cwd) {
if (chdir (cmd->cwd) < 0)
error (EXEC_FAILED_EXIT_STATUS, errno,
"can't change directory to '%s'", cmd->cwd);
}
for (i = 0; i < cmd->nenv; ++i) {
if (cmd->env[i].name) {
if (cmd->env[i].value)
setenv (cmd->env[i].name,
cmd->env[i].value, 1);
else
unsetenv (cmd->env[i].name);
} else
clearenv ();
}
switch (cmd->tag) {
case PIPECMD_PROCESS: {
struct pipecmd_process *cmdp = &cmd->u.process;
if (cmd->pre_exec_func)
cmd->pre_exec_func (cmd->pre_exec_data);
execvp (cmd->name, cmdp->argv);
break;
}
/* TODO: ideally, could there be a facility
* to execute non-blocking functions without
* needing to fork?
*/
case PIPECMD_FUNCTION: {
struct pipecmd_function *cmdf = &cmd->u.function;
if (cmd->pre_exec_func)
cmd->pre_exec_func (cmd->pre_exec_data);
cmdf->func (cmdf->data);
/* pacify valgrind et al */
if (cmdf->free_func)
cmdf->free_func (cmdf->data);
if (cmd->pre_exec_free_func)
cmd->pre_exec_free_func (cmd->pre_exec_data);
exit (0);
}
case PIPECMD_SEQUENCE: {
struct pipecmd_sequence *cmds = &cmd->u.sequence;
struct sigaction sa;
/* Flush all pending output so that subprocesses
* don't inherit it.
*/
fflush (NULL);
/* pipeline_start will have blocked SIGCHLD. We like
* it that way. Lose the parent's signal handler,
* though.
*/
memset (&sa, 0, sizeof sa);
sa.sa_handler = SIG_DFL;
sigemptyset (&sa.sa_mask);
sa.sa_flags = 0;
if (sigaction (SIGCHLD, &sa, NULL) == -1)
error (FATAL, errno,
"can't install SIGCHLD handler");
for (i = 0; i < cmds->ncommands; ++i) {
pipecmd *child = cmds->commands[i];
pid_t pid = fork ();
int status;
if (pid < 0)
error (FATAL, errno, "fork failed");
if (pid == 0)
pipecmd_exec (child);
debug ("Started \"%s\", pid %d\n",
child->name, pid);
while (waitpid (pid, &status, 0) < 0) {
if (errno == EINTR)
continue;
error (FATAL, errno, "waitpid failed");
}
debug (" \"%s\" (%d) -> %d\n",
child->name, pid, status);
if (WIFSIGNALED (status)) {
int sig = WTERMSIG (status);
#ifdef SIGPIPE
if (sig == SIGPIPE)
status = 0;
else
#endif /* SIGPIPE */
if (getenv ("PIPELINE_QUIET"))
;
else if (WCOREDUMP (status))
error (0, 0,
"%s: %s (core dumped)",
child->name,
strsignal (sig));
else
error (0, 0, "%s: %s",
child->name,
strsignal (sig));
} else if (!WIFEXITED (status))
error (0, 0, "unexpected status %d",
status);
if (child->tag == PIPECMD_FUNCTION) {
struct pipecmd_function *cmdf =
&child->u.function;
if (cmdf->free_func)
(*cmdf->free_func)
(cmdf->data);
}
if (WIFSIGNALED (status)) {
raise (WTERMSIG (status));
exit (1); /* just to make sure */
} else if (status && WIFEXITED (status))
exit (WEXITSTATUS (status));
}
exit (0);
}
}
error (EXEC_FAILED_EXIT_STATUS, errno, "can't execute %s", cmd->name);
/* Never called, but gcc doesn't realise that error with non-zero
* status always exits.
*/
exit (EXEC_FAILED_EXIT_STATUS);
}
void pipecmd_free (pipecmd *cmd)
{
int i;
if (!cmd)
return;
free (cmd->name);
free (cmd->cwd);
for (i = 0; i < cmd->nenv; ++i) {
free (cmd->env[i].name);
free (cmd->env[i].value);
}
free (cmd->env);
switch (cmd->tag) {
case PIPECMD_PROCESS: {
struct pipecmd_process *cmdp = &cmd->u.process;
for (i = 0; i < cmdp->argc; ++i)
free (cmdp->argv[i]);
free (cmdp->argv);
break;
}
case PIPECMD_FUNCTION:
break;
case PIPECMD_SEQUENCE: {
struct pipecmd_sequence *cmds = &cmd->u.sequence;
for (i = 0; i < cmds->ncommands; ++i)
pipecmd_free (cmds->commands[i]);
free (cmds->commands);
break;
}
}
free (cmd);
}
/* ---------------------------------------------------------------------- */
/* Functions to build pipelines. */
pipeline *pipeline_new (void)
{
pipeline *p = XMALLOC (pipeline);
p->ncommands = 0;
p->commands_max = 4;
p->commands = xnmalloc (p->commands_max, sizeof *p->commands);
p->pids = NULL;
p->statuses = NULL;
p->redirect_in = p->redirect_out = REDIRECT_NONE;
p->want_in = p->want_out = 0;
p->want_infile = p->want_outfile = NULL;
p->infd = p->outfd = -1;
p->infile = p->outfile = NULL;
p->source = NULL;
p->buffer = NULL;
p->buflen = p->bufmax = 0;
p->line_cache = NULL;
p->peek_offset = 0;
p->ignore_signals = 0;
return p;
}
pipeline *pipeline_new_commandv (pipecmd *cmd1, va_list cmdv)
{
pipeline *p = pipeline_new ();
pipeline_command (p, cmd1);
pipeline_commandv (p, cmdv);
return p;
}
pipeline *pipeline_new_commands (pipecmd *cmd1, ...)
{
va_list cmdv;
pipeline *p;
va_start (cmdv, cmd1);
p = pipeline_new_commandv (cmd1, cmdv);
va_end (cmdv);
return p;
}
pipeline *pipeline_new_command_argv (const char *name, va_list argv)
{
pipeline *p;
pipecmd *cmd;
p = pipeline_new ();
cmd = pipecmd_new_argv (name, argv);
pipeline_command (p, cmd);
return p;
}
pipeline *pipeline_new_command_args (const char *name, ...)
{
va_list argv;
pipeline *p;
va_start (argv, name);
p = pipeline_new_command_argv (name, argv);
va_end (argv);
return p;
}
pipeline *pipeline_join (pipeline *p1, pipeline *p2)
{
pipeline *p = XMALLOC (pipeline);
int i;
assert (!p1->pids);
assert (!p2->pids);
assert (!p1->statuses);
assert (!p2->statuses);
p->ncommands = p1->ncommands + p2->ncommands;
p->commands_max = p1->ncommands + p2->ncommands;
p->commands = xnmalloc (p->commands_max, sizeof *p->commands);
p->pids = NULL;
p->statuses = NULL;
p->redirect_in = p1->redirect_in;
p->want_in = p1->want_in;
p->want_infile = p1->want_infile ? xstrdup (p1->want_infile) : NULL;
p->redirect_out = p2->redirect_out;
p->want_out = p2->want_out;
p->want_outfile = p2->want_outfile ? xstrdup (p2->want_outfile) : NULL;
p->infd = p1->infd;
p->outfd = p2->outfd;
p->infile = p1->infile;
p->outfile = p2->outfile;
p->source = NULL;
p->buffer = NULL;
p->buflen = p->bufmax = 0;
p->line_cache = NULL;
p->peek_offset = 0;
p->ignore_signals = (p1->ignore_signals || p2->ignore_signals);
for (i = 0; i < p1->ncommands; ++i)
p->commands[i] = pipecmd_dup (p1->commands[i]);
for (i = 0; i < p2->ncommands; ++i)
p->commands[p1->ncommands + i] = pipecmd_dup (p2->commands[i]);
return p;
}
void pipeline_connect (pipeline *source, pipeline *sink, ...)
{
va_list argv;
pipeline *arg;
/* We must be in control of output from the source pipeline. If the
* source isn't started, we can force this.
*/
if (!source->pids)
pipeline_want_out (source, -1);
assert (source->redirect_out == REDIRECT_FD);
assert (source->want_out < 0);
va_start (argv, sink);
for (arg = sink; arg; arg = va_arg (argv, pipeline *)) {
assert (!arg->pids); /* not started */
arg->source = source;
pipeline_want_in (arg, -1);
/* Zero-command sinks should represent data being passed
* straight through from the input to the output.
* Unfortunately pipeline_start and pipeline_pump don't
* handle this very well between them; a zero-command
* pipeline has the write end of its input pipe wrongly
* stashed in outfd and then pipeline_pump can't handle it
* because it has nowhere to send output. Until this is
* fixed, this kludge is necessary.
*/
if (arg->ncommands == 0)
pipeline_command (arg, pipecmd_new_passthrough ());
}
va_end (argv);
}
void pipeline_command (pipeline *p, pipecmd *cmd)
{
if (p->ncommands >= p->commands_max) {
p->commands_max *= 2;
p->commands = xrealloc (p->commands,
p->commands_max * sizeof *p->commands);
}
p->commands[p->ncommands++] = cmd;
}
void pipeline_command_argv (pipeline *p, const char *name, va_list argv)
{
pipecmd *cmd;
cmd = pipecmd_new_argv (name, argv);
pipeline_command (p, cmd);
}
void pipeline_command_args (pipeline *p, const char *name, ...)
{
va_list argv;
va_start (argv, name);
pipeline_command_argv (p, name, argv);
va_end (argv);
}
void pipeline_command_argstr (pipeline *p, const char *argstr)
{
pipeline_command (p, pipecmd_new_argstr (argstr));
}
void pipeline_commandv (pipeline *p, va_list cmdv)
{
pipecmd *cmd = va_arg (cmdv, pipecmd *);
while (cmd) {
pipeline_command (p, cmd);
cmd = va_arg (cmdv, pipecmd *);
}
}
void pipeline_commands (pipeline *p, ...)
{
va_list cmdv;
va_start (cmdv, p);
pipeline_commandv (p, cmdv);
va_end (cmdv);
}
int pipeline_get_ncommands (pipeline *p)
{
return p->ncommands;
}
pipecmd *pipeline_get_command (pipeline *p, int n)
{
if (n < 0 || n >= p->ncommands)
return NULL;
return p->commands[n];
}
pipecmd *pipeline_set_command (pipeline *p, int n, pipecmd *cmd)
{
pipecmd *prev;
if (n < 0 || n >= p->ncommands)
return NULL;
prev = p->commands[n];
p->commands[n] = cmd;
return prev;
}
pid_t pipeline_get_pid (pipeline *p, int n)
{
assert (p->pids); /* pipeline started */
if (n < 0 || n >= p->ncommands)
return -1;
return p->pids[n];
}
void pipeline_want_in (pipeline *p, int fd)
{
p->redirect_in = REDIRECT_FD;
p->want_in = fd;
p->want_infile = NULL;
}
void pipeline_want_out (pipeline *p, int fd)
{
p->redirect_out = REDIRECT_FD;
p->want_out = fd;
p->want_outfile = NULL;
}
void pipeline_want_infile (pipeline *p, const char *file)
{
p->redirect_in = (file != NULL) ? REDIRECT_FILE_NAME : REDIRECT_NONE;
p->want_in = 0;
p->want_infile = file ? xstrdup (file) : NULL;
}
void pipeline_want_outfile (pipeline *p, const char *file)
{
p->redirect_out = (file != NULL) ? REDIRECT_FILE_NAME : REDIRECT_NONE;
p->want_out = 0;
p->want_outfile = file ? xstrdup (file) : NULL;
}
void pipeline_ignore_signals (pipeline *p, int ignore_signals)
{
p->ignore_signals = ignore_signals;
}
FILE *pipeline_get_infile (pipeline *p)
{
assert (p->pids); /* pipeline started */
assert (p->statuses);
if (p->infile)
return p->infile;
else if (p->infd == -1) {
error (0, 0, "pipeline input not open");
return NULL;
} else
return p->infile = fdopen (p->infd, "w");
}
FILE *pipeline_get_outfile (pipeline *p)
{
assert (p->pids); /* pipeline started */
assert (p->statuses);
if (p->outfile)
return p->outfile;
else if (p->outfd == -1) {
error (0, 0, "pipeline output not open");
return NULL;
} else
return p->outfile = fdopen (p->outfd, "r");
}
void pipeline_dump (pipeline *p, FILE *stream)
{
int i;
for (i = 0; i < p->ncommands; ++i) {
pipecmd_dump (p->commands[i], stream);
if (i < p->ncommands - 1)
fputs (" | ", stream);
}
fprintf (stream, " [input: {%d, %s}, output: {%d, %s}]\n",
p->want_in, p->want_infile ? p->want_infile : "NULL",
p->want_out, p->want_outfile ? p->want_outfile : "NULL");
}
char *pipeline_tostring (pipeline *p)
{
char *out = NULL;
int i;
for (i = 0; i < p->ncommands; ++i) {
char *cmdout = pipecmd_tostring (p->commands[i]);
out = appendstr (out, cmdout, NULL);
free (cmdout);
if (i < p->ncommands - 1)
out = appendstr (out, " | ", NULL);
}
return out;
}
void pipeline_free (pipeline *p)
{
int i;
if (!p)
return;
if (p->pids)
pipeline_wait (p);
for (i = 0; i < p->ncommands; ++i)
pipecmd_free (p->commands[i]);
free (p->commands);
free (p->pids);
free (p->statuses);
free (p->want_infile);
free (p->want_outfile);
free (p->buffer);
free (p->line_cache);
free (p);
}
/* ---------------------------------------------------------------------- */
/* Functions to run pipelines and handle signals. */
static pipeline **active_pipelines = NULL;
static int n_active_pipelines = 0, max_active_pipelines = 0;
static int sigchld = 0;
static int queue_sigchld = 0;
static int reap_children (int block)
{
pid_t pid;
int status;
int collected = 0;
do {
int i;
if (sigchld) {
/* Deal with a SIGCHLD delivery. */
pid = waitpid (-1, &status, WNOHANG);
--sigchld;
} else
pid = waitpid (-1, &status, block ? 0 : WNOHANG);
if (pid < 0 && errno == EINTR) {
/* Try again. */
pid = 0;
continue;
}
if (pid <= 0)
/* We've run out of children to reap. */
break;
++collected;
/* Deliver the command status if possible. */
for (i = 0; i < n_active_pipelines; ++i) {
pipeline *p = active_pipelines[i];
int j;
if (!p || !p->pids || !p->statuses)
continue;
for (j = 0; j < p->ncommands; ++j) {
if (p->pids[j] == pid) {
p->statuses[j] = status;
i = n_active_pipelines;
break;
}
}
}
} while ((sigchld || block == 0) && pid >= 0);
if (collected)
return collected;
else
return -1;
}
static void pipeline_sigchld (int signum)
{
/* really an assert, but that's not async-signal-safe */
if (signum == SIGCHLD) {
++sigchld;
if (!queue_sigchld) {
int save_errno = errno;
reap_children (0);
errno = save_errno;
}
}
}
static void pipeline_install_sigchld (void)
{
struct sigaction act;
static int installed = 0;
if (installed)
return;
memset (&act, 0, sizeof act);
act.sa_handler = &pipeline_sigchld;
sigemptyset (&act.sa_mask);
sigaddset (&act.sa_mask, SIGINT);
sigaddset (&act.sa_mask, SIGTERM);
sigaddset (&act.sa_mask, SIGHUP);
sigaddset (&act.sa_mask, SIGCHLD);
act.sa_flags = 0;
#ifdef SA_NOCLDSTOP
act.sa_flags |= SA_NOCLDSTOP;
#endif
#ifdef SA_RESTART
act.sa_flags |= SA_RESTART;
#endif
if (sigaction (SIGCHLD, &act, NULL) == -1)
error (FATAL, errno, "can't install SIGCHLD handler");
installed = 1;
}
static pipeline_post_fork_fn *post_fork = NULL;
void pipeline_install_post_fork (pipeline_post_fork_fn *fn)
{
post_fork = fn;
}
static int ignored_signals = 0;
static struct sigaction osa_sigint, osa_sigquit;
void pipeline_start (pipeline *p)
{
int i, j;
int last_input = -1;
int infd[2];
sigset_t set, oset;
/* Make sure our SIGCHLD handler is installed. */
pipeline_install_sigchld ();
assert (!p->pids); /* pipeline not started already */
assert (!p->statuses);
init_debug ();
if (debug_level) {
debug ("Starting pipeline: ");
pipeline_dump (p, stderr);
}
/* Flush all pending output so that subprocesses don't inherit it. */
fflush (NULL);
if (p->ignore_signals && !ignored_signals++) {
struct sigaction sa;
/* Ignore SIGINT and SIGQUIT while subprocesses are running,
* just like system().
*/
memset (&sa, 0, sizeof sa);
sa.sa_handler = SIG_IGN;
sigemptyset (&sa.sa_mask);
sa.sa_flags = 0;
if (sigaction (SIGINT, &sa, &osa_sigint) < 0)
error (FATAL, errno, "Couldn't ignore SIGINT");
if (sigaction (SIGQUIT, &sa, &osa_sigquit) < 0)
error (FATAL, errno, "Couldn't ignore SIGQUIT");
}
/* Add to the table of active pipelines, so that signal handlers
* know what to do with exit statuses. Block SIGCHLD so that we can
* do this safely.
*/
sigemptyset (&set);
sigaddset (&set, SIGCHLD);
sigemptyset (&oset);
while (sigprocmask (SIG_BLOCK, &set, &oset) == -1 && errno == EINTR)
;
/* Grow the table if necessary. */
if (n_active_pipelines >= max_active_pipelines) {
int filled = max_active_pipelines;
if (max_active_pipelines)
max_active_pipelines *= 2;
else
max_active_pipelines = 4;
/* reduces to xmalloc (...) if active_pipelines == NULL */
active_pipelines = xrealloc
(active_pipelines,
max_active_pipelines * sizeof *active_pipelines);
memset (active_pipelines + filled, 0,
(max_active_pipelines - filled) *
sizeof *active_pipelines);
}
for (i = 0; i < max_active_pipelines; ++i)
if (!active_pipelines[i]) {
active_pipelines[i] = p;
break;
}
assert (i < max_active_pipelines);
++n_active_pipelines;
p->pids = xcalloc (p->ncommands, sizeof *p->pids);
p->statuses = xcalloc (p->ncommands, sizeof *p->statuses);
/* Unblock SIGCHLD. */
while (sigprocmask (SIG_SETMASK, &oset, NULL) == -1 && errno == EINTR)
;
if (p->redirect_in == REDIRECT_FD && p->want_in < 0) {
if (pipe (infd) < 0)
error (FATAL, errno, "pipe failed");
last_input = infd[0];
p->infd = infd[1];
} else if (p->redirect_in == REDIRECT_FD)
last_input = p->want_in;
else if (p->redirect_in == REDIRECT_FILE_NAME) {
assert (p->want_infile);
last_input = open (p->want_infile, O_RDONLY);
if (last_input < 0)
error (FATAL, errno, "can't open %s", p->want_infile);
}
for (i = 0; i < p->ncommands; i++) {
int pdes[2];
pid_t pid;
int output_read = -1, output_write = -1;
if (i != p->ncommands - 1 ||
(p->redirect_out == REDIRECT_FD && p->want_out < 0)) {
if (pipe (pdes) < 0)
error (FATAL, errno, "pipe failed");
if (i == p->ncommands - 1)
p->outfd = pdes[0];
output_read = pdes[0];
output_write = pdes[1];
} else {
if (p->redirect_out == REDIRECT_FD)
output_write = p->want_out;
else if (p->redirect_out == REDIRECT_FILE_NAME) {
assert (p->want_outfile);
output_write = open (p->want_outfile,
O_WRONLY | O_CREAT |
O_TRUNC, 0666);
if (output_write < 0)
error (FATAL, errno, "can't open %s",
p->want_outfile);
}
}
/* Block SIGCHLD so that the signal handler doesn't collect
* the exit status before we've filled in the pids array.
*/
sigemptyset (&set);
sigaddset (&set, SIGCHLD);
sigemptyset (&oset);
while (sigprocmask (SIG_BLOCK, &set, &oset) == -1 &&
errno == EINTR)
;
pid = fork ();
if (pid < 0)
error (FATAL, errno, "fork failed");
if (pid == 0) {
/* child */
if (post_fork)
post_fork ();
/* input, reading side */
if (last_input != -1) {
if (dup2 (last_input, 0) < 0)
error (FATAL, errno, "dup2 failed");
if (close (last_input) < 0)
error (FATAL, errno, "close failed");
}
/* output, writing side */
if (output_write != -1) {
if (dup2 (output_write, 1) < 0)
error (FATAL, errno, "dup2 failed");
if (close (output_write) < 0)
error (FATAL, errno, "close failed");
}
/* output, reading side */
if (output_read != -1)
if (close (output_read))
error (FATAL, errno, "close failed");
/* input from first command, writing side; must close
* it in every child because it has to be created
* before forking anything
*/
if (p->infd != -1)
if (close (p->infd))
error (FATAL, errno, "close failed");
/* inputs and outputs from other active pipelines */
for (j = 0; j < n_active_pipelines; ++j) {
pipeline *active = active_pipelines[j];
if (!active || active == p)
continue;
/* ignore failures */
if (active->infd != -1)
close (active->infd);
if (active->outfd != -1)
close (active->outfd);
}
/* Restore signals. */
if (p->ignore_signals) {
sigaction (SIGINT, &osa_sigint, NULL);
sigaction (SIGQUIT, &osa_sigquit, NULL);
}
pipecmd_exec (p->commands[i]);
/* never returns */
}
/* in the parent */
if (last_input != -1) {
if (close (last_input) < 0)
error (FATAL, errno, "close failed");
}
if (output_write != -1) {
if (close (output_write) < 0)
error (FATAL, errno, "close failed");
}
if (output_read != -1)
last_input = output_read;
p->pids[i] = pid;
p->statuses[i] = -1;
/* Unblock SIGCHLD. */
while (sigprocmask (SIG_SETMASK, &oset, NULL) == -1 &&
errno == EINTR)
;
debug ("Started \"%s\", pid %d\n", p->commands[i]->name, pid);
}
if (p->ncommands == 0)
p->outfd = last_input;
}
int pipeline_wait_all (pipeline *p, int **statuses, int *n_statuses)
{
int ret = 0;
int proc_count = p->ncommands;
int i;
int raise_signal = 0;
init_debug ();
if (debug_level) {
debug ("Waiting for pipeline: ");
pipeline_dump (p, stderr);
}
assert (p->pids); /* pipeline started */
assert (p->statuses);
if (p->infile) {
if (fclose (p->infile))
error (0, errno,
"closing pipeline input stream failed");
p->infile = NULL;
p->infd = -1;
} else if (p->infd != -1) {
if (close (p->infd))
error (0, errno, "closing pipeline input failed");
p->infd = -1;
}
if (p->outfile) {
if (fclose (p->outfile)) {
error (0, errno,
"closing pipeline output stream failed");
ret = 127;
}
p->outfile = NULL;
p->outfd = -1;
} else if (p->outfd != -1) {
if (close (p->outfd)) {
error (0, errno, "closing pipeline output failed");
ret = 127;
}
p->outfd = -1;
}
/* Tell the SIGCHLD handler not to get in our way. */
queue_sigchld = 1;
while (proc_count > 0) {
int r;
debug ("Active processes (%d):\n", proc_count);
/* Check for any statuses already collected by SIGCHLD
* handlers or the previous iteration before calling
* reap_children() again.
*/
for (i = 0; i < p->ncommands; ++i) {
int status;
if (p->pids[i] == -1)
continue;
debug (" \"%s\" (%d) -> %d\n",
p->commands[i]->name, p->pids[i],
p->statuses[i]);
if (p->statuses[i] == -1)
continue;
status = p->statuses[i];
p->pids[i] = -1;
--proc_count;
if (WIFSIGNALED (status)) {
int sig = WTERMSIG (status);
#ifdef SIGPIPE
if (sig == SIGPIPE)
status = 0;
else {
#endif /* SIGPIPE */
/* signals currently blocked,
* re-raise later
*/
if (sig == SIGINT || sig == SIGQUIT)
raise_signal = sig;
else if (getenv ("PIPELINE_QUIET"))
;
else if (WCOREDUMP (status))
error (0, 0,
"%s: %s (core dumped)",
p->commands[i]->name,
strsignal (sig));
else
error (0, 0, "%s: %s",
p->commands[i]->name,
strsignal (sig));
#ifdef SIGPIPE
}
#endif /* SIGPIPE */
} else if (!WIFEXITED (status))
error (0, 0, "unexpected status %d",
status);
if (p->commands[i]->tag == PIPECMD_FUNCTION) {
struct pipecmd_function *cmdf =
&p->commands[i]->u.function;
if (cmdf->free_func)
(*cmdf->free_func) (cmdf->data);
}
if (i == p->ncommands - 1) {
if (WIFSIGNALED (status))
ret = 128 + WTERMSIG (status);
else if (WEXITSTATUS (status))
ret = WEXITSTATUS (status);
} else if (!ret &&
(WIFSIGNALED (status) ||
WEXITSTATUS (status)))
ret = 127;
}
assert (proc_count >= 0);
if (proc_count == 0)
break;
errno = 0;
r = reap_children (1);
if (r == -1 && errno == ECHILD)
/* Eh? The pipeline was allegedly still running, so
* we shouldn't have got ECHILD.
*/
error (FATAL, errno, "waitpid failed");
}
queue_sigchld = 0;
for (i = 0; i < n_active_pipelines; ++i)
if (active_pipelines[i] == p)
active_pipelines[i] = NULL;
/* It isn't normally safe to compactify active_pipelines as we go,
* because it's used by a signal handler. However, if it's entirely
* empty, we can safely clean it up now. This prevents the table
* growing without bound, not to mention pacifying valgrind.
*/
for (i = 0; i < n_active_pipelines; ++i)
if (active_pipelines[i])
break;
if (i == n_active_pipelines) {
n_active_pipelines = 0;
max_active_pipelines = 0;
free (active_pipelines);
active_pipelines = NULL;
}
if (statuses && n_statuses) {
*statuses = xnmalloc (p->ncommands, sizeof **statuses);
*n_statuses = p->ncommands;
for (i = 0; i < p->ncommands; ++i)
(*statuses)[i] = p->statuses[i];
}
free (p->pids);
p->pids = NULL;
free (p->statuses);
p->statuses = NULL;
if (p->ignore_signals && !--ignored_signals) {
/* Restore signals. */
sigaction (SIGINT, &osa_sigint, NULL);
sigaction (SIGQUIT, &osa_sigquit, NULL);
}
if (raise_signal)
raise (raise_signal);
return ret;
}
int pipeline_wait (pipeline *p)
{
return pipeline_wait_all (p, NULL, NULL);
}
int pipeline_run (pipeline *p)
{
int status;
pipeline_start (p);
status = pipeline_wait (p);
pipeline_free (p);
return status;
}
void pipeline_pump (pipeline *p, ...)
{
va_list argv;
int argc, i, j;
pipeline *arg, **pieces;
size_t *pos;
int *known_source, *blocking_in, *blocking_out,
*dying_source, *waiting, *write_error;
struct sigaction sa, osa_sigpipe;
/* Count pipelines and allocate space for arrays. */
va_start (argv, p);
argc = 0;
for (arg = p; arg; arg = va_arg (argv, pipeline *))
++argc;
va_end (argv);
pieces = xnmalloc (argc, sizeof *pieces);
pos = xnmalloc (argc, sizeof *pos);
known_source = xcalloc (argc, sizeof *known_source);
blocking_in = xcalloc (argc, sizeof *blocking_in);
blocking_out = xcalloc (argc, sizeof *blocking_out);
dying_source = xcalloc (argc, sizeof *dying_source);
waiting = xcalloc (argc, sizeof *waiting);
write_error = xcalloc (argc, sizeof *write_error);
/* Set up arrays of pipelines and their read positions. Start all
* pipelines if necessary.
*/
va_start (argv, p);
for (arg = p, i = 0; i < argc; arg = va_arg (argv, pipeline *), ++i) {
pieces[i] = arg;
pos[i] = 0;
if (!pieces[i]->pids)
pipeline_start (pieces[i]);
}
assert (arg == NULL);
va_end (argv);
/* All source pipelines must be supplied as arguments. */
for (i = 0; i < argc; ++i) {
int found = 0;
if (!pieces[i]->source)
continue;
for (j = 0; j < argc; ++j) {
if (pieces[i]->source == pieces[j]) {
known_source[j] = found = 1;
break;
}
}
assert (found);
}
for (i = 0; i < argc; ++i) {
int flags;
if (pieces[i]->infd != -1) {
flags = fcntl (pieces[i]->infd, F_GETFL);
if (!(flags & O_NONBLOCK)) {
blocking_in[i] = 1;
fcntl (pieces[i]->infd, F_SETFL,
flags | O_NONBLOCK);
}
}
if (pieces[i]->outfd != -1) {
flags = fcntl (pieces[i]->outfd, F_GETFL);
if (!(flags & O_NONBLOCK)) {
blocking_out[i] = 1;
fcntl (pieces[i]->outfd, F_SETFL,
flags | O_NONBLOCK);
}
}
}
#ifdef SIGPIPE
memset (&sa, 0, sizeof sa);
sa.sa_handler = SIG_IGN;
sigemptyset (&sa.sa_mask);
sa.sa_flags = 0;
sigaction (SIGPIPE, &sa, &osa_sigpipe);
#endif
#ifdef SA_RESTART
/* We rely on getting EINTR from select. */
sigaction (SIGCHLD, NULL, &sa);
sa.sa_flags &= ~SA_RESTART;
sigaction (SIGCHLD, &sa, NULL);
#endif
for (;;) {
fd_set rfds, wfds;
int maxfd = -1;
int ret;
/* If a source dies and all data from it has been written to
* all sinks, close the writing end of the pipe to each of
* its sinks.
*/
for (i = 0; i < argc; ++i) {
if (!known_source[i] || pieces[i]->outfd != -1 ||
pipeline_peek_size (pieces[i]))
continue;
for (j = 0; j < argc; ++j) {
if (pieces[j]->source == pieces[i] &&
pieces[j]->infd != -1) {
if (close (pieces[j]->infd))
error (0, errno,
"closing pipeline "
"input failed");
pieces[j]->infd = -1;
}
}
}
/* If all sinks on a source have died, close the reading end
* of the pipe from that source.
*/
for (i = 0; i < argc; ++i) {
int got_sink = 0;
if (!known_source[i] || pieces[i]->outfd == -1)
continue;
for (j = 0; j < argc; ++j) {
if (pieces[j]->source == pieces[i] &&
pieces[j]->infd != -1) {
got_sink = 1;
break;
}
}
if (got_sink)
continue;
if (close (pieces[i]->outfd))
error (0, errno,
"closing pipeline output failed");
pieces[i]->outfd = -1;
}
FD_ZERO (&rfds);
FD_ZERO (&wfds);
for (i = 0; i < argc; ++i) {
/* Input to sink pipeline. */
if (pieces[i]->source && pieces[i]->infd != -1 &&
!waiting[i]) {
FD_SET (pieces[i]->infd, &wfds);
if (pieces[i]->infd > maxfd)
maxfd = pieces[i]->infd;
}
/* Output from source pipeline. */
if (known_source[i] && pieces[i]->outfd != -1) {
FD_SET (pieces[i]->outfd, &rfds);
if (pieces[i]->outfd > maxfd)
maxfd = pieces[i]->outfd;
}
}
if (maxfd == -1)
break; /* nothing meaningful left to do */
ret = select (maxfd + 1, &rfds, &wfds, NULL, NULL);
if (ret < 0 && errno == EINTR) {
/* Did a source or sink pipeline die? */
for (i = 0; i < argc; ++i) {
if (pieces[i]->ncommands == 0)
continue;
if (known_source[i] && !dying_source[i] &&
pieces[i]->outfd != -1) {
int last = pieces[i]->ncommands - 1;
assert (pieces[i]->statuses);
if (pieces[i]->statuses[last] != -1) {
debug ("source pipeline %d "
"died\n", i);
dying_source[i] = 1;
}
}
if (pieces[i]->source &&
pieces[i]->infd != -1) {
assert (pieces[i]->statuses);
if (pieces[i]->statuses[0] != -1) {
debug ("sink pipeline %d "
"died\n", i);
close (pieces[i]->infd);
pieces[i]->infd = -1;
}
}
}
continue;
} else if (ret < 0)
error (FATAL, errno, "select");
/* Read a block of data from each available source pipeline. */
for (i = 0; i < argc; ++i) {
size_t peek_size, len;
if (!known_source[i] || pieces[i]->outfd == -1)
continue;
if (!FD_ISSET (pieces[i]->outfd, &rfds))
continue;
peek_size = pipeline_peek_size (pieces[i]);
len = peek_size + 4096;
if (!pipeline_peek (pieces[i], &len) ||
len == peek_size) {
/* Error or end-of-file; skip this pipeline
* from now on.
*/
debug ("source pipeline %d returned error "
"or EOF\n", i);
close (pieces[i]->outfd);
pieces[i]->outfd = -1;
} else
/* This is rather a large hammer. Whenever
* any data is read from any source
* pipeline, we go through and retry all
* sink pipelines, even if they aren't
* receiving data from the source in
* question. This probably results in a few
* more passes around the select() loop, but
* it eliminates some annoyingly fiddly
* bookkeeping.
*/
memset (waiting, 0, argc * sizeof *waiting);
}
/* Write as much data as we can to each available sink
* pipeline.
*/
for (i = 0; i < argc; ++i) {
const char *block;
size_t peek_size;
ssize_t w;
size_t minpos;
if (!pieces[i]->source || pieces[i]->infd == -1)
continue;
if (!FD_ISSET (pieces[i]->infd, &wfds))
continue;
peek_size = pipeline_peek_size (pieces[i]->source);
if (peek_size <= pos[i]) {
/* Disable reading until data is read from a
* source fd or a child process exits, so
* that we neither spin nor block if the
* source is slow.
*/
waiting[i] = 1;
continue;
}
/* peek a block from the source */
block = pipeline_peek (pieces[i]->source, &peek_size);
/* should all already be in the peek cache */
assert (block);
assert (peek_size);
/* write as much of it as will fit to the sink */
for (;;) {
w = safe_write (pieces[i]->infd,
block + pos[i],
peek_size - pos[i]);
if (w >= 0)
break;
if (errno == EAGAIN) {
w = 0;
break;
}
/* It may be useful for other processes to
* continue even though this one fails, so
* don't FATAL yet.
*/
if (errno != EPIPE)
write_error[i] = errno;
close (pieces[i]->infd);
pieces[i]->infd = -1;
goto next_sink;
}
pos[i] += w;
minpos = pos[i];
/* check other sinks on the same source, and update
* the source's read position if earlier data is no
* longer needed by any sink
*/
for (j = 0; j < argc; ++j) {
if (pieces[i]->source != pieces[j]->source ||
pieces[j]->infd == -1)
continue;
if (pos[j] < minpos)
minpos = pos[j];
/* If the source is dead and all data has
* been written to this sink, close the
* writing end of the pipe to the sink.
*/
if (pieces[j]->source->outfd == -1 &&
pos[j] >= peek_size) {
close (pieces[j]->infd);
pieces[j]->infd = -1;
}
}
/* If some data has been written to all sinks,
* discard it from the source's peek cache.
*/
pipeline_peek_skip (pieces[i]->source, minpos);
for (j = 0; j < argc; ++j) {
if (pieces[i]->source == pieces[j]->source)
pos[j] -= minpos;
}
next_sink: ;
}
}
#ifdef SA_RESTART
sigaction (SIGCHLD, NULL, &sa);
sa.sa_flags |= SA_RESTART;
sigaction (SIGCHLD, &sa, NULL);
#endif
#ifdef SIGPIPE
sigaction (SIGPIPE, &osa_sigpipe, NULL);
#endif
for (i = 0; i < argc; ++i) {
int flags;
if (blocking_in[i] && pieces[i]->infd != -1) {
flags = fcntl (pieces[i]->infd, F_GETFL);
fcntl (pieces[i]->infd, F_SETFL, flags & ~O_NONBLOCK);
}
if (blocking_out[i] && pieces[i]->outfd != -1) {
flags = fcntl (pieces[i]->outfd, F_GETFL);
fcntl (pieces[i]->outfd, F_SETFL, flags & ~O_NONBLOCK);
}
}
for (i = 0; i < argc; ++i) {
if (write_error[i])
error (FATAL, write_error[i], "write to sink %d", i);
}
free (write_error);
free (waiting);
free (dying_source);
free (blocking_out);
free (blocking_in);
free (known_source);
free (pieces);
free (pos);
}
/* ---------------------------------------------------------------------- */
/* Functions to read output from pipelines. */
static const char *get_block (pipeline *p, size_t *len, int peek)
{
size_t readstart = 0, retstart = 0;
size_t space = p->bufmax;
size_t toread = *len;
ssize_t r;
if (p->buffer && p->peek_offset) {
if (p->peek_offset >= toread) {
/* We've got the whole thing in the peek cache; just
* return it.
*/
const char *buffer;
assert (p->peek_offset <= p->buflen);
buffer = p->buffer + p->buflen - p->peek_offset;
if (!peek)
p->peek_offset -= toread;
return buffer;
} else {
readstart = p->buflen;
retstart = p->buflen - p->peek_offset;
space -= p->buflen;
toread -= p->peek_offset;
}
}
if (toread > space) {
if (p->buffer)
p->bufmax = readstart + toread;
else
p->bufmax = toread;
p->buffer = xrealloc (p->buffer, p->bufmax + 1);
}
if (!peek)
p->peek_offset = 0;
assert (p->outfd != -1);
r = safe_read (p->outfd, p->buffer + readstart, toread);
if (r == -1)
return NULL;
p->buflen = readstart + r;
if (peek)
p->peek_offset += r;
*len -= (toread - r);
return p->buffer + retstart;
}
const char *pipeline_read (pipeline *p, size_t *len)
{
return get_block (p, len, 0);
}
const char *pipeline_peek (pipeline *p, size_t *len)
{
return get_block (p, len, 1);
}
size_t pipeline_peek_size (pipeline *p)
{
if (!p->buffer)
return 0;
return p->peek_offset;
}
void pipeline_peek_skip (pipeline *p, size_t len)
{
if (len > 0) {
assert (p->buffer);
assert (len <= p->peek_offset);
p->peek_offset -= len;
}
}
/* readline and peekline repeatedly peek larger and larger buffers until
* they find a newline or they fail. readline then adjusts the peek offset.
*/
static const char *get_line (pipeline *p, size_t *outlen)
{
const size_t block = 4096;
const char *buffer = NULL, *end = NULL;
int i;
size_t previous_len = 0;
if (p->line_cache) {
free (p->line_cache);
p->line_cache = NULL;
}
if (outlen)
*outlen = 0;
for (i = 0; ; ++i) {
size_t len = block * (i + 1);
buffer = get_block (p, &len, 1);
if (!buffer || len == 0)
return NULL;
if (len == previous_len)
/* end of file, no newline found */
end = buffer + len - 1;
else
end = memchr (buffer + previous_len, '\n',
len - previous_len);
if (end)
break;
previous_len = len;
}
if (end) {
p->line_cache = xstrndup (buffer, end - buffer + 1);
if (outlen)
*outlen = end - buffer + 1;
return p->line_cache;
} else
return NULL;
}
const char *pipeline_readline (pipeline *p)
{
size_t buflen;
const char *buffer = get_line (p, &buflen);
if (buffer)
p->peek_offset -= buflen;
return buffer;
}
const char *pipeline_peekline (pipeline *p)
{
return get_line (p, NULL);
}