53 |
#include "error.h" |
#include "error.h" |
54 |
#include "pipeline.h" |
#include "pipeline.h" |
55 |
|
|
56 |
|
extern int debug; |
57 |
|
|
58 |
|
/* ---------------------------------------------------------------------- */ |
59 |
|
|
60 |
|
/* Functions to build individual commands. */ |
61 |
|
|
62 |
command *command_new (const char *name) |
command *command_new (const char *name) |
63 |
{ |
{ |
64 |
command *cmd = xmalloc (sizeof *cmd); |
command *cmd = xmalloc (sizeof *cmd); |
298 |
free (cmd); |
free (cmd); |
299 |
} |
} |
300 |
|
|
301 |
|
/* ---------------------------------------------------------------------- */ |
302 |
|
|
303 |
|
/* Functions to build pipelines. */ |
304 |
|
|
305 |
pipeline *pipeline_new (void) |
pipeline *pipeline_new (void) |
306 |
{ |
{ |
307 |
pipeline *p = xmalloc (sizeof *p); |
pipeline *p = xmalloc (sizeof *p); |
309 |
p->commands_max = 4; |
p->commands_max = 4; |
310 |
p->commands = xmalloc (p->commands_max * sizeof *p->commands); |
p->commands = xmalloc (p->commands_max * sizeof *p->commands); |
311 |
p->pids = NULL; |
p->pids = NULL; |
312 |
|
p->statuses = NULL; |
313 |
p->want_in = p->want_out = 0; |
p->want_in = p->want_out = 0; |
314 |
p->infd = p->outfd = -1; |
p->infd = p->outfd = -1; |
315 |
p->infile = p->outfile = NULL; |
p->infile = p->outfile = NULL; |
343 |
|
|
344 |
assert (!p1->pids); |
assert (!p1->pids); |
345 |
assert (!p2->pids); |
assert (!p2->pids); |
346 |
|
assert (!p1->statuses); |
347 |
|
assert (!p2->statuses); |
348 |
|
|
349 |
p->ncommands = p1->ncommands + p2->ncommands; |
p->ncommands = p1->ncommands + p2->ncommands; |
350 |
p->commands_max = p1->ncommands + p2->ncommands; |
p->commands_max = p1->ncommands + p2->ncommands; |
351 |
p->commands = xmalloc (p->commands_max * sizeof *p->commands); |
p->commands = xmalloc (p->commands_max * sizeof *p->commands); |
352 |
p->pids = NULL; |
p->pids = NULL; |
353 |
|
p->statuses = NULL; |
354 |
p->want_in = p1->want_in; |
p->want_in = p1->want_in; |
355 |
p->want_out = p2->want_out; |
p->want_out = p2->want_out; |
356 |
p->infd = p1->infd; |
p->infd = p1->infd; |
415 |
FILE *pipeline_get_infile (pipeline *p) |
FILE *pipeline_get_infile (pipeline *p) |
416 |
{ |
{ |
417 |
assert (p->pids); /* pipeline started */ |
assert (p->pids); /* pipeline started */ |
418 |
|
assert (p->statuses); |
419 |
if (p->infile) |
if (p->infile) |
420 |
return p->infile; |
return p->infile; |
421 |
else if (p->infd == -1) { |
else if (p->infd == -1) { |
428 |
FILE *pipeline_get_outfile (pipeline *p) |
FILE *pipeline_get_outfile (pipeline *p) |
429 |
{ |
{ |
430 |
assert (p->pids); /* pipeline started */ |
assert (p->pids); /* pipeline started */ |
431 |
|
assert (p->statuses); |
432 |
if (p->outfile) |
if (p->outfile) |
433 |
return p->outfile; |
return p->outfile; |
434 |
else if (p->outfd == -1) { |
else if (p->outfd == -1) { |
473 |
return out; |
return out; |
474 |
} |
} |
475 |
|
|
476 |
|
void pipeline_free (pipeline *p) |
477 |
|
{ |
478 |
|
int i; |
479 |
|
|
480 |
|
for (i = 0; i < p->ncommands; ++i) |
481 |
|
command_free (p->commands[i]); |
482 |
|
free (p->commands); |
483 |
|
if (p->pids) |
484 |
|
free (p->pids); |
485 |
|
if (p->statuses) |
486 |
|
free (p->statuses); |
487 |
|
free (p); |
488 |
|
} |
489 |
|
|
490 |
|
/* ---------------------------------------------------------------------- */ |
491 |
|
|
492 |
|
/* Functions to run pipelines and handle signals. */ |
493 |
|
|
494 |
|
static pipeline **active_pipelines = NULL; |
495 |
|
static int n_active_pipelines = 0, max_active_pipelines = 0; |
496 |
|
|
497 |
/* Children exit with this status if execvp fails. */ |
/* Children exit with this status if execvp fails. */ |
498 |
#define EXEC_FAILED_EXIT_STATUS 0xff |
#define EXEC_FAILED_EXIT_STATUS 0xff |
499 |
|
|
502 |
int i; |
int i; |
503 |
int last_input = -1; |
int last_input = -1; |
504 |
int infd[2]; |
int infd[2]; |
505 |
|
sigset_t set, oset; |
506 |
|
|
507 |
assert (!p->pids); /* pipeline not started already */ |
assert (!p->pids); /* pipeline not started already */ |
508 |
|
assert (!p->statuses); |
509 |
|
|
510 |
|
/* Add to the table of active pipelines, so that signal handlers |
511 |
|
* know what to do with exit statuses. Block SIGCHLD so that we can |
512 |
|
* do this safely. |
513 |
|
*/ |
514 |
|
sigemptyset (&set); |
515 |
|
sigaddset (&set, SIGCHLD); |
516 |
|
sigemptyset (&oset); |
517 |
|
while (sigprocmask (SIG_BLOCK, &set, &oset) == -1 && errno == EINTR) |
518 |
|
; |
519 |
|
|
520 |
|
/* Grow the table if necessary. */ |
521 |
|
if (n_active_pipelines >= max_active_pipelines) { |
522 |
|
int filled = max_active_pipelines; |
523 |
|
if (max_active_pipelines) |
524 |
|
max_active_pipelines *= 2; |
525 |
|
else |
526 |
|
max_active_pipelines = 4; |
527 |
|
/* reduces to xmalloc (...) if active_pipelines == NULL */ |
528 |
|
active_pipelines = xrealloc |
529 |
|
(active_pipelines, |
530 |
|
max_active_pipelines * sizeof *active_pipelines); |
531 |
|
memset (active_pipelines + filled, 0, |
532 |
|
(max_active_pipelines - filled) * |
533 |
|
sizeof *active_pipelines); |
534 |
|
} |
535 |
|
|
536 |
|
for (i = 0; i < max_active_pipelines; ++i) |
537 |
|
if (!active_pipelines[i]) { |
538 |
|
active_pipelines[i] = p; |
539 |
|
break; |
540 |
|
} |
541 |
|
assert (i < max_active_pipelines); |
542 |
|
++n_active_pipelines; |
543 |
|
|
544 |
|
/* Unblock SIGCHLD. */ |
545 |
|
while (sigprocmask (SIG_SETMASK, &oset, NULL) == -1 && errno == EINTR) |
546 |
|
; |
547 |
|
|
548 |
p->pids = xmalloc (p->ncommands * sizeof *p->pids); |
p->pids = xmalloc (p->ncommands * sizeof *p->pids); |
549 |
|
p->statuses = xmalloc (p->ncommands * sizeof *p->statuses); |
550 |
|
|
551 |
if (p->want_in < 0) { |
if (p->want_in < 0) { |
552 |
if (pipe (infd) < 0) |
if (pipe (infd) < 0) |
560 |
int pdes[2]; |
int pdes[2]; |
561 |
pid_t pid; |
pid_t pid; |
562 |
int output_read = -1, output_write = -1; |
int output_read = -1, output_write = -1; |
563 |
|
sigset_t set, oset; |
564 |
|
|
565 |
if (i != p->ncommands - 1 || p->want_out < 0) { |
if (i != p->ncommands - 1 || p->want_out < 0) { |
566 |
if (pipe (pdes) < 0) |
if (pipe (pdes) < 0) |
572 |
} else if (i == p->ncommands - 1 && p->want_out > 0) |
} else if (i == p->ncommands - 1 && p->want_out > 0) |
573 |
output_write = p->want_out; |
output_write = p->want_out; |
574 |
|
|
575 |
|
/* Block SIGCHLD so that the signal handler doesn't collect |
576 |
|
* the exit status before we've filled in the pids array. |
577 |
|
*/ |
578 |
|
sigemptyset (&set); |
579 |
|
sigaddset (&set, SIGCHLD); |
580 |
|
sigemptyset (&oset); |
581 |
|
while (sigprocmask (SIG_BLOCK, &set, &oset) == -1 && |
582 |
|
errno == EINTR) |
583 |
|
; |
584 |
|
|
585 |
pid = fork (); |
pid = fork (); |
586 |
if (pid < 0) |
if (pid < 0) |
587 |
error (FATAL, errno, _("fork failed")); |
error (FATAL, errno, _("fork failed")); |
642 |
if (output_read != -1) |
if (output_read != -1) |
643 |
last_input = output_read; |
last_input = output_read; |
644 |
p->pids[i] = pid; |
p->pids[i] = pid; |
645 |
|
p->statuses[i] = -1; |
646 |
|
|
647 |
|
/* Unblock SIGCHLD. */ |
648 |
|
while (sigprocmask (SIG_SETMASK, &oset, NULL) == -1 && |
649 |
|
errno == EINTR) |
650 |
|
; |
651 |
|
|
652 |
|
if (debug) |
653 |
|
fprintf (stderr, "Started \"%s\", pid %d\n", |
654 |
|
p->commands[i]->name, pid); |
655 |
} |
} |
656 |
} |
} |
657 |
|
|
658 |
/* TODO: Perhaps it would be useful to wait on multiple pipelines |
static int sigchld = 0; |
659 |
* simultaneously? Then you could do: |
static int queue_sigchld = 0; |
660 |
* |
|
661 |
* p1->want_out = -1; |
static int reap_children (int block) |
662 |
* p2->want_in = -1; |
{ |
663 |
* p3->want_in = -1; |
pid_t pid; |
664 |
* pipeline_start (p1); |
int status; |
665 |
* pipeline_start (p2); |
int collected = 0; |
666 |
* pipeline_start (p3); |
|
667 |
* ... select() on p1's output, p2's input, and p3's input, and glue them |
do { |
668 |
* together ... |
int i; |
669 |
* pipeline_wait (p1, p2, p3); |
|
670 |
* |
if (sigchld) { |
671 |
* ... and have processes exit as their input is closed by other processes |
/* Deal with a SIGCHLD delivery. */ |
672 |
* exiting, etc. |
pid = waitpid (-1, &status, WNOHANG); |
673 |
* |
--sigchld; |
674 |
* This function is kind of broken if multiple pipelines exist, anyway, or |
} else |
675 |
* even if there are other child processes not created by the pipeline |
pid = waitpid (-1, &status, block ? 0 : WNOHANG); |
676 |
* library. Unfortunately, spinning and polling non-blocking waitpid() is |
|
677 |
* going to suck. Perhaps ignoring the second case is OK. |
if (pid < 0 && errno == EINTR) { |
678 |
*/ |
/* Try again. */ |
679 |
|
pid = 0; |
680 |
|
continue; |
681 |
|
} |
682 |
|
|
683 |
|
if (pid <= 0) |
684 |
|
/* We've run out of children to reap. */ |
685 |
|
break; |
686 |
|
|
687 |
|
++collected; |
688 |
|
|
689 |
|
/* Deliver the command status if possible. */ |
690 |
|
for (i = 0; i < n_active_pipelines; ++i) { |
691 |
|
pipeline *p = active_pipelines[i]; |
692 |
|
int j; |
693 |
|
|
694 |
|
if (!p || !p->pids || !p->statuses) |
695 |
|
continue; |
696 |
|
|
697 |
|
for (j = 0; j < p->ncommands; ++j) { |
698 |
|
if (p->pids[j] == pid) { |
699 |
|
p->statuses[j] = status; |
700 |
|
i = n_active_pipelines; |
701 |
|
break; |
702 |
|
} |
703 |
|
} |
704 |
|
} |
705 |
|
} while ((sigchld || block == 0) && pid >= 0); |
706 |
|
|
707 |
|
if (collected) |
708 |
|
return collected; |
709 |
|
else |
710 |
|
return -1; |
711 |
|
} |
712 |
|
|
713 |
int pipeline_wait (pipeline *p) |
int pipeline_wait (pipeline *p) |
714 |
{ |
{ |
715 |
int ret = 0; |
int ret = 0; |
716 |
int proc_count = p->ncommands; |
int proc_count = p->ncommands; |
717 |
|
int i; |
718 |
|
|
719 |
assert (p->pids); /* pipeline started */ |
assert (p->pids); /* pipeline started */ |
720 |
|
assert (p->statuses); |
721 |
|
|
722 |
if (p->infile) { |
if (p->infile) { |
723 |
if (fclose (p->infile)) |
if (fclose (p->infile)) |
732 |
} |
} |
733 |
|
|
734 |
while (proc_count > 0) { |
while (proc_count > 0) { |
735 |
int i; |
int r; |
|
int status; |
|
736 |
|
|
737 |
pid_t pid = wait (&status); |
if (debug) |
738 |
|
fprintf (stderr, "Active processes (%d):\n", |
739 |
|
proc_count); |
740 |
|
|
741 |
|
/* Check for any statuses already collected by SIGCHLD |
742 |
|
* handlers or the previous iteration before calling |
743 |
|
* reap_children() again. |
744 |
|
*/ |
745 |
|
for (i = 0; i < p->ncommands; ++i) { |
746 |
|
if (p->pids[i] == -1) |
747 |
|
continue; /* already collected */ |
748 |
|
|
749 |
|
if (debug) |
750 |
|
fprintf (stderr, " \"%s\" (%d) -> %d\n", |
751 |
|
p->commands[i]->name, p->pids[i], |
752 |
|
p->statuses[i]); |
753 |
|
|
754 |
if (pid < 0) |
if (p->statuses[i] != -1) { |
755 |
error (FATAL, errno, _("wait failed")); |
int status = p->statuses[i]; |
|
for (i = 0; i < p->ncommands; i++) |
|
|
if (p->pids[i] == pid) { |
|
756 |
p->pids[i] = -1; |
p->pids[i] = -1; |
757 |
--proc_count; |
--proc_count; |
758 |
if (WIFSIGNALED (status)) { |
if (WIFSIGNALED (status)) { |
772 |
|
|
773 |
if (i == p->ncommands - 1) |
if (i == p->ncommands - 1) |
774 |
ret = status; |
ret = status; |
|
|
|
|
break; |
|
775 |
} |
} |
776 |
|
} |
777 |
|
|
778 |
|
assert (proc_count >= 0); |
779 |
|
if (proc_count == 0) |
780 |
|
break; |
781 |
|
|
782 |
|
/* Tell the SIGCHLD handler not to get in our way. */ |
783 |
|
queue_sigchld = 1; |
784 |
|
r = reap_children (1); |
785 |
|
queue_sigchld = 0; |
786 |
|
|
787 |
|
if (r == -1 && errno == ECHILD) |
788 |
|
/* Eh? The pipeline was allegedly still running, so |
789 |
|
* we shouldn't have got ECHILD. |
790 |
|
*/ |
791 |
|
error (FATAL, errno, _("waitpid failed")); |
792 |
} |
} |
793 |
|
|
794 |
if (p->outfile) { |
if (p->outfile) { |
803 |
p->outfd = -1; |
p->outfd = -1; |
804 |
} |
} |
805 |
|
|
806 |
|
for (i = 0; i < n_active_pipelines; ++i) |
807 |
|
if (active_pipelines[i] == p) |
808 |
|
active_pipelines[i] = NULL; |
809 |
|
|
810 |
free (p->pids); |
free (p->pids); |
811 |
p->pids = NULL; |
p->pids = NULL; |
812 |
|
free (p->statuses); |
813 |
|
p->statuses = NULL; |
814 |
|
|
815 |
return ret; |
return ret; |
816 |
} |
} |
817 |
|
|
818 |
void pipeline_free (pipeline *p) |
static void pipeline_sigchld (int signum) |
819 |
{ |
{ |
820 |
int i; |
int save_errno = errno; |
821 |
|
|
822 |
for (i = 0; i < p->ncommands; ++i) |
assert (signum == SIGCHLD); |
823 |
command_free (p->commands[i]); |
|
824 |
free (p->commands); |
++sigchld; |
825 |
if (p->pids) |
if (!queue_sigchld) |
826 |
free (p->pids); |
reap_children (0); |
827 |
free (p); |
|
828 |
|
errno = save_errno; |
829 |
|
} |
830 |
|
|
831 |
|
void pipeline_install_sigchld (void) |
832 |
|
{ |
833 |
|
struct sigaction act; |
834 |
|
|
835 |
|
memset (&act, 0, sizeof act); |
836 |
|
act.sa_handler = &pipeline_sigchld; |
837 |
|
sigemptyset (&act.sa_mask); |
838 |
|
sigaddset (&act.sa_mask, SIGINT); |
839 |
|
sigaddset (&act.sa_mask, SIGTERM); |
840 |
|
sigaddset (&act.sa_mask, SIGHUP); |
841 |
|
sigaddset (&act.sa_mask, SIGCHLD); |
842 |
|
act.sa_flags = 0; |
843 |
|
#ifdef SA_NOCLDSTOP |
844 |
|
act.sa_flags |= SA_NOCLDSTOP; |
845 |
|
#endif |
846 |
|
#ifdef SA_RESTART |
847 |
|
act.sa_flags |= SA_RESTART; |
848 |
|
#endif |
849 |
|
while (sigaction (SIGCHLD, &act, NULL) == -1) { |
850 |
|
if (errno == EINTR) |
851 |
|
continue; |
852 |
|
error (FATAL, errno, _("can't install SIGCHLD handler")); |
853 |
|
} |
854 |
} |
} |