/[man-db]/man-db/lib/pipeline.c
ViewVC logotype

Diff of /man-db/lib/pipeline.c

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.15 by cjwatson, Sat Aug 9 16:00:27 2003 UTC revision 1.16 by cjwatson, Sat Aug 9 17:28:18 2003 UTC
# Line 53  extern char *strerror (); Line 53  extern char *strerror ();
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);
# Line 292  void command_free (command *cmd) Line 298  void command_free (command *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);
# Line 299  pipeline *pipeline_new (void) Line 309  pipeline *pipeline_new (void)
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;
# Line 332  pipeline *pipeline_join (pipeline *p1, p Line 343  pipeline *pipeline_join (pipeline *p1, p
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;
# Line 401  void pipeline_commands (pipeline *p, ... Line 415  void pipeline_commands (pipeline *p, ...
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) {
# Line 413  FILE *pipeline_get_infile (pipeline *p) Line 428  FILE *pipeline_get_infile (pipeline *p)
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) {
# Line 457  char *pipeline_tostring (pipeline *p) Line 473  char *pipeline_tostring (pipeline *p)
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    
# Line 465  void pipeline_start (pipeline *p) Line 502  void pipeline_start (pipeline *p)
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)
# Line 481  void pipeline_start (pipeline *p) Line 560  void pipeline_start (pipeline *p)
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)
# Line 492  void pipeline_start (pipeline *p) Line 572  void pipeline_start (pipeline *p)
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"));
# Line 552  void pipeline_start (pipeline *p) Line 642  void pipeline_start (pipeline *p)
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))
# Line 596  int pipeline_wait (pipeline *p) Line 732  int pipeline_wait (pipeline *p)
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)) {
# Line 624  int pipeline_wait (pipeline *p) Line 772  int pipeline_wait (pipeline *p)
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) {
# Line 641  int pipeline_wait (pipeline *p) Line 803  int pipeline_wait (pipeline *p)
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  }  }

Legend:
Removed from v.1.15  
changed lines
  Added in v.1.16

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26