From: Reuben Thomas <rrt@sc3d.org>
Date: Sun, 21 Jan 2018 23:17:04 +0000 (+0000)
Subject: task.c: make “correct” version of perform_pipe_sequence work
X-Git-Tag: v3.7~83
X-Git-Url: https://granicus.if.org/sourcecode?a=commitdiff_plain;h=e41c5443e9559741613e8b122076d6ce8b107e5c;p=recode

task.c: make “correct” version of perform_pipe_sequence work

Finish converting the right-to-left pipeline into a left-to-right version.
---

diff --git a/src/task.c b/src/task.c
index db7aa2f..b096dd2 100644
--- a/src/task.c
+++ b/src/task.c
@@ -385,25 +385,13 @@ perform_sequence (RECODE_TASK task, enum recode_sequence_strategy strategy)
 | that more than one recoding step is needed.				   |
 `-------------------------------------------------------------------------*/
 
-#if 1
-
-/* FIXME: This is no good.  The main process might open too many files for
-   one thing.  All of it should create children from left to right, instead
-   of all children to a single parent right to left.
-
-   The code to do it correctly is below, but doesn't yet work for multiple
-   steps.
-*/
-
 static bool
 perform_pipe_sequence (RECODE_TASK task)
 {
   RECODE_CONST_REQUEST request = task->request;
-  RECODE_OUTER outer = request->outer;
   struct recode_subtask subtask_block;
   RECODE_SUBTASK subtask = &subtask_block;
 
-  unsigned sequence_index;	/* index into sequence */
   RECODE_CONST_STEP step;	/* pointer into single_steps */
 
   int pipe_pair[2];		/* pair of file descriptors for a pipe */
@@ -415,67 +403,66 @@ perform_pipe_sequence (RECODE_TASK task)
   subtask->input = task->input;
   subtask->output = task->output;
 
-  /* Prepare the final output file.  */
+  /* Prepare the final input file.  */
 
-  if (!*subtask->output.name)
-    subtask->output.file = stdout;
-  else if (subtask->output.file = fopen (subtask->output.name, "w"),
-	   subtask->output.file == NULL)
+  if (!*subtask->input.name)
+    subtask->input.file = stdin;
+  else if (subtask->input.file = fopen (subtask->input.name, "r"),
+	   subtask->input.file == NULL)
     {
-      recode_perror (outer, "fopen (%s)", subtask->output.name);
-      return false;
+      recode_perror (NULL, "fopen (%s)", subtask->input.name);
+      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
+      SUBTASK_RETURN (subtask);
     }
 
-  /* Create all subprocesses, from the last to the first, and
+  /* Create all subprocesses, from the first to the last, and
      interconnect them.  */
 
-  for (sequence_index = request->sequence_length - 1;
-       sequence_index > 0;
-       sequence_index--)
+  for (unsigned sequence_index = 0;
+       sequence_index < (unsigned)request->sequence_length - 1;
+       sequence_index++)
     {
       if (pipe (pipe_pair) < 0)
 	{
-	  recode_perror (outer, "pipe ()");
+	  recode_perror (NULL, "pipe ()");
 	  recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
 	  SUBTASK_RETURN (subtask);
 	}
       if (child_process = fork (), child_process < 0)
 	{
-	  recode_perror (outer, "fork ()");
+	  recode_perror (NULL, "fork ()");
 	  recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
 	  SUBTASK_RETURN (subtask);
 	}
       if (child_process == 0)
 	{
-          /* The child executes its recoding step, reading from the pipe
-             and writing to the current output file; then it exits.  */
+          /* The child executes its recoding step, reading from the
+             current input file and writing to the pipe; then it exits.  */
 
-	  if (close (pipe_pair[1]) < 0)
+	  if (close (pipe_pair[0]) < 0)
 	    {
-	      recode_perror (outer, "close ()");
-	      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-	      SUBTASK_RETURN (subtask);
+	      recode_perror (NULL, "close ()");
+              recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
 	    }
-	  if (subtask->input.file = fdopen (pipe_pair[0], "r"),
-	      subtask->input.file == NULL)
+	  if (subtask->output.file = fdopen (pipe_pair[1], "w"),
+	      subtask->output.file == NULL)
 	    {
-	      recode_perror (outer, "fdopen ()");
-	      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-	      SUBTASK_RETURN (subtask);
+	      recode_perror (NULL, "fdopen ()");
+              recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
 	    }
 
 	  step = request->sequence_array + sequence_index;
 	  subtask->step = step;
 	  (*step->transform_routine) (subtask);
 
-	  if (fclose (subtask->input.file) != 0)
+	  if (subtask->input.file && fclose (subtask->input.file) != 0)
 	    {
-              recode_perror (NULL, "fclose (%s)", subtask->input.name);
+              recode_perror (NULL, "fclose (%s)", subtask->input.name ? subtask->input.name : "stdin");
               recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
 	    }
-	  if (subtask->output.file && fclose (subtask->output.file) != 0)
+	  if (fclose (subtask->output.file) != 0)
             {
-              recode_perror (NULL, "fclose (%s)", subtask->output.name);
+              recode_perror (NULL, "fclose (%s)", subtask->output.name ? subtask->output.name : "stdout");
               recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
             }
 
@@ -484,252 +471,57 @@ perform_pipe_sequence (RECODE_TASK task)
 	}
       else
 	{
-          /* The parent redirects the current output file to the pipe.  */
+          /* The parent redirects the current input file to the pipe.  */
 
-	  if (dup2 (pipe_pair[1], fileno (subtask->output.file)) < 0)
+	  if (dup2 (pipe_pair[0], fileno (subtask->input.file)) < 0)
 	    {
-	      recode_perror (outer, "dup2 ()");
+	      recode_perror (NULL, "dup2 ()");
 	      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
 	      SUBTASK_RETURN (subtask);
 	    }
 	  if (close (pipe_pair[0]) < 0)
 	    {
-	      recode_perror (outer, "close ()");
+	      recode_perror (NULL, "close ()");
 	      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
 	      SUBTASK_RETURN (subtask);
 	    }
 	  if (close (pipe_pair[1]) < 0)
 	    {
-	      recode_perror (outer, "close ()");
+	      recode_perror (NULL, "close ()");
 	      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
 	      SUBTASK_RETURN (subtask);
 	    }
 	}
     }
 
-  /* All the children are created, blocked on read.  Now, feed the whole
-     chain of processes with the output of the first recoding step.  */
-
-  if (!*subtask->input.name)
-    subtask->input.file = stdin;
-  else if (subtask->input.file = fopen (subtask->input.name, "r"),
-	   subtask->input.file == NULL)
-    {
-      recode_perror (outer, "fopen (%s)", subtask->input.name);
-      SUBTASK_RETURN (subtask);
-    }
-
-  step = request->sequence_array;
-  subtask->step = step;
-  (*step->transform_routine) (subtask);
-
-  if (subtask->input.file && fclose (subtask->input.file) != 0)
-    {
-      recode_perror (NULL, "fclose (%s)", subtask->input.name);
-      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-    }
-  if (subtask->output.file && fclose (subtask->output.file) != 0)
-    {
-      recode_perror (NULL, "fclose (%s)", subtask->output.name);
-      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-    }
-
-  /* Wait on all children, mainly to avoid synchronisation problems on
-     output file contents, but also to reduce the number of zombie
-     processes in case the user recodes many files at once.  */
-
-  while (wait (&wait_status) > 0)
-    {
-      /* Diagnose and abort on any abnormally terminating child.  */
-
-      if (!(WIFEXITED (wait_status)
-	    || (WIFSIGNALED (wait_status)
-		&& WTERMSIG (wait_status) == SIGPIPE)))
-	{
-	  recode_error (outer, _("Child process wait status is 0x%0.2x"),
-			wait_status);
-	  return false;
-	}
-
-      /* Check for a nonzero exit from the terminating child.  */
-
-      if (WIFEXITED (wait_status)
-	  ? WEXITSTATUS (wait_status) != 0
-	  : WTERMSIG (wait_status) != 0)
-	/* FIXME: It is not very clear what happened in sub-processes.  */
-	if (task->error_so_far < task->fail_level)
-	  {
-	    task->error_so_far = task->fail_level;
-	    task->error_at_step = step;
-	  }
-    }
-
-#if 0
-  if (interrupted)
-    /* FIXME: It is not very clear what happened in sub-processes.  */
-    if (task->error_so_far < task->fail_level)
-      {
-	task->error_so_far = task->fail_level;
-	task->error_at_step = step;
-      }
-#endif
-
-  SUBTASK_RETURN (subtask);
-}
-
-#else /* not 1 */
-
-static bool
-perform_pipe_sequence (RECODE_TASK task)
-{
-  RECODE_CONST_REQUEST request = task->request;
-  RECODE_OUTER outer = request->outer;
-  struct recode_subtask subtask_block;
-  RECODE_SUBTASK subtask = &subtask_block;
-
-  unsigned sequence_index;	/* index into sequence */
-  RECODE_CONST_STEP step;	/* pointer into single_steps */
-
-  int pipe_pair[2];		/* pair of file descriptors for a pipe */
-  int child_process;		/* child process number, zero if child */
-  pid_t wait_status;		/* status returned by wait() */
-
-  memset (subtask, 0, sizeof (struct recode_subtask));
-  subtask->task = task;
-  subtask->input = task->input;
-  subtask->output = task->output;
-
-  /* Prepare the final files.  */
-
-  if (!*subtask->input.name)
-    subtask->input.file = stdin;
-  else if (subtask->input.file = fopen (subtask->input.name, "r"),
-	   subtask->input.file == NULL)
-    {
-      recode_perror (outer, "fopen (%s)", subtask->input.name);
-      return false;
-    }
+  /* Perform the last recoding step and read its output.  */
 
   if (!*subtask->output.name)
     subtask->output.file = stdout;
   else if (subtask->output.file = fopen (subtask->output.name, "w"),
 	   subtask->output.file == NULL)
     {
-      recode_perror (outer, "fopen (%s)", subtask->output.name);
-      return false;
-    }
-
-  /* Create all subprocesses, from the first to the last, and
-     interconnect them.  */
-
-  for (sequence_index = 0;
-       sequence_index < request->sequence_length - 1;
-       sequence_index++)
-    {
-      if (pipe (pipe_pair) < 0)
-	{
-	  recode_perror (outer, "pipe ()");
-	  return false;
-	}
-      if (child_process = fork (), child_process < 0)
-	{
-	  recode_perror (outer, "fork ()");
-	  return false;
-	}
-      if (child_process == 0)
-	{
-          /* The child executes its recoding step, reading from the pipe
-             and writing to the current output file; then it exits.  */
-
-	  if (close (pipe_pair[1]) < 0)
-	    {
-	      recode_perror (outer, "close ()");
-	      return false;
-	    }
-	  if (subtask->input.file = fdopen (pipe_pair[0], "r"),
-	      subtask->input.file == NULL)
-	    {
-	      recode_perror (outer, "fdopen ()");
-	      return false;
-	    }
-
-	  step = request->sequence_array + sequence_index;
-	  subtask->step = step;
-	  (*step->transform_routine) (subtask);
-
-	  if (subtask->input.file && fclose (subtask->input.file) != 0)
-	    {
-              recode_perror (NULL, "fclose (%s)", subtask->input.name);
-              recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-              return false;
-	    }
-	  if (subtask->output.file && fclose (subtask->output.file) != 0)
-            {
-              recode_perror (NULL, "fclose (%s)", subtask->output.name);
-              recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-              return false;
-            }
-
-	  exit (task->error_so_far < task->fail_level ? EXIT_SUCCESS
-		: EXIT_FAILURE);
-	}
-      else
-	{
-          /* The parent redirects the current output file to the pipe.  */
-
-	  if (dup2 (pipe_pair[1], fileno (subtask->output.file)) < 0)
-	    {
-	      recode_perror (outer, "dup2 ()");
-	      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-	      return false;
-	    }
-	  if (close (pipe_pair[0]) < 0)
-	    {
-	      recode_perror (outer, "close ()");
-	      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-	      return false;
-	    }
-	  if (close (pipe_pair[1]) < 0)
-	    {
-	      recode_perror (outer, "close ()");
-	      recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-	      return false;
-	    }
-	}
-    }
-
-  /* All processes execute the following common code, each with its proper
-     value for SEQUENCE_INDEX, CHILD_PROCESS, etc.  */
-
-  /* All the children are created, blocked on read.  Now, feed the whole
-     chain of processes with the output of the first recoding step.  */
-
-  if (!*subtask->input.name)
-    subtask->input.file = stdin;
-  else if (subtask->input.file = fopen (subtask->input.name, "r"),
-	   subtask->input.file == NULL)
-    {
-      recode_perror (outer, "fopen (%s)", subtask->input.name);
+      recode_perror (NULL, "fopen (%s)", subtask->output.name);
       recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-      return false;
+      SUBTASK_RETURN (subtask);
     }
 
-  step = request->sequence_array;
+  step = request->sequence_array + (unsigned)request->sequence_length - 1;
   subtask->step = step;
   (*step->transform_routine) (subtask);
 
   if (subtask->input.file && fclose (subtask->input.file) != 0)
     {
-      recode_perror (NULL, "fclose (%s)", subtask->input.name);
+      recode_perror (NULL, "fclose (%s)", subtask->input.name ? subtask->input.name : "stdin");
       recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-      return false;
+      SUBTASK_RETURN (subtask);
     }
 
-  if (subtask->output.file && fclose (subtask->output.file) != 0)
+  if (fclose (subtask->output.file) != 0)
     {
-      recode_perror (NULL, "fclose (%s)", subtask->output.name);
+      recode_perror (NULL, "fclose (%s)", subtask->output.name ? subtask->output.name : "stdout");
       recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-      return false;
+      SUBTASK_RETURN (subtask);
     }
 
   /* Wait on all children, mainly to avoid synchronisation problems on
@@ -744,10 +536,10 @@ perform_pipe_sequence (RECODE_TASK task)
 	    || (WIFSIGNALED (wait_status)
 		&& WTERMSIG (wait_status) == SIGPIPE)))
 	{
-	  recode_error (outer, _("Child process wait status is 0x%0.2x"),
+	  recode_error (NULL, _("Child process wait status is 0x%0.2x"),
 			wait_status);
 	  recode_if_nogo (RECODE_SYSTEM_ERROR, subtask);
-	  return false;
+	  SUBTASK_RETURN (subtask);
 	}
 
       /* Check for a nonzero exit from the terminating child.  */
@@ -776,8 +568,6 @@ perform_pipe_sequence (RECODE_TASK task)
   SUBTASK_RETURN (subtask);
 }
 
-#endif /* not 1 */
-
 #endif /* HAVE_PIPE */
 
 /* Library interface.  */