[Toybox] toybox: xargs: add actual support for -P max-proc flag

[Mark Salyzyn]

-P flag was fake just to provide compatibility.  Add support for -P
maximum process count to actually create parallelism.  Does not
support SIGUSR1 or SIGUSR2 increment/decrement signals as publicly
documented for other variants.  If max-proc is 0, run as many
processes as possible simultaneously.

Signed-off-by: Mark Salyzyn <salyzyn at android.com>
---
 tests/xargs.test   | 15 ++++++++++
 toys/posix/xargs.c | 70 ++++++++++++++++++++++++++++++++--------------
 2 files changed, 64 insertions(+), 21 deletions(-)

diff --git a/tests/xargs.test b/tests/xargs.test
index afed8a17..dc3c7b32 100644
--- a/tests/xargs.test
+++ b/tests/xargs.test
@@ -61,6 +61,21 @@ testing "big input forces split" \
   'for i in $(seq 1 100);do echo $X;done|{ xargs >/dev/null && echo yes;}' \
   "yes\n" "" ""
 
+# -P max-proc
+testing "max-proc=1" "xargs -n 1 -P 1" "one\ntwo\nthree\n" "" "one\ntwo\nthree\n"
+testing "max-proc=2" "xargs -n 1 -P 2" "y\ny\ny\n" "" "y\ny\ny\n"
+testing "max-proc=3" "xargs -n 1 -P 3" "y\ny\ny\n" "" "y\ny\ny\n"
+# 3 seconds vs 2 seconds vs 1 second parallel sleep
+testing "parallel sleep" "
+  xargs_sleep() {
+    ( yes | head -3 | tr y 1 | time -p xargs -n 1 \${*} sleep ) 2>&1 |
+      sed -n 's/^real *\([0-9]*\)[.]\(.*\)/\1\2/p'
+  } &&
+  A=\`xargs_sleep\` &&
+  B=\`xargs_sleep -P 2\` &&
+  C=\`xargs_sleep -P 0\` &&
+  [ \${A} -gt \${B} -a \${B} -gt \${C} ] && echo OK || echo FAIL" "OK\n" "" ""
+
 # TODO: what exactly is -x supposed to do? why does coreutils output "one"?
 #testing "-x" "xargs -x -s 9 || echo expected" "one\nexpected\n" "" "one\ntwo\nthree"
 
diff --git a/toys/posix/xargs.c b/toys/posix/xargs.c
index 79e7dea5..a1707ea7 100644
--- a/toys/posix/xargs.c
+++ b/toys/posix/xargs.c
@@ -8,15 +8,15 @@
  * TODO: -I	Insert mode
  * TODO: -L	Max number of lines of input per command
  * TODO: -x	Exit if can't fit everything in one command
- * TODO: -P NUM	Run up to NUM processes at once
+ * TODO: -P NUM	Support SIGUSR1 and SIGUSR2
 
-USE_XARGS(NEWTOY(xargs, "^E:P#optrn#<1(max-args)s#0[!0E]", TOYFLAG_USR|TOYFLAG_BIN))
+USE_XARGS(NEWTOY(xargs, "^E:P#<0=1optrn#<1(max-args)s#0[!0E]", TOYFLAG_USR|TOYFLAG_BIN))
 
 config XARGS
   bool "xargs"
   default y
   help
-    usage: xargs [-0prt] [-s NUM] [-n NUM] [-E STR] COMMAND...
+    usage: xargs [-0prt] [-s NUM] [-n NUM] [-P NUM] [-E STR] COMMAND...
 
     Run command line one or more times, appending arguments from stdin.
 
@@ -27,6 +27,7 @@ config XARGS
     -n	Max number of arguments per command
     -o	Open tty for COMMAND's stdin (default /dev/null)
     -p	Prompt for y/n from tty before running each command
+    -P	Max number of parallel processes (default 1)
     -r	Don't run command with empty input (otherwise always run command once)
     -s	Size in bytes per command line
     -t	Trace, print command line to stderr
@@ -42,6 +43,7 @@ GLOBALS(
   long entries, bytes;
   char delim;
   FILE *tty;
+  long np;
 )
 
 // If !entry count TT.bytes and TT.entries, stopping at max.
@@ -91,10 +93,47 @@ static char *handle_entries(char *data, char **entry)
   return 0;
 }
 
+static void wait_for_pid_flush() {
+  int status;
+
+  while (TT.np > 0 && waitpid(-1, &status, 0)) --TT.np;
+}
+
+// Handle process completions and report.
+//
+// pid: process to wait for, or 0 non-blocking all, -1 blocking all
+// return pid, negative pid if return code requires immediate exit
+static pid_t wait_for_pid(pid_t pid) {
+  int status;
+
+  if (TT.np <= 0) return 0;
+  pid = waitpid(pid > 0 ? pid : -1, &status, pid ? 0 : WNOHANG);
+  if (pid == 0) return pid;
+  --TT.np;
+
+  // xargs is yet another weird collection of exit value special cases,
+  // different from all the others.
+  if (WIFEXITED(status)) {
+    if (WEXITSTATUS(status) == 126 || WEXITSTATUS(status) == 127) {
+      toys.exitval = WEXITSTATUS(status);
+      wait_for_pid_flush();
+      return -pid;
+    } else if (WEXITSTATUS(status) >= 1 && WEXITSTATUS(status) <= 125) {
+      toys.exitval = 123;
+    } else if (WEXITSTATUS(status) == 255) {
+      error_msg("%s: exited with status 255; aborting", toys.optargs[0]);
+      toys.exitval = 124;
+      wait_for_pid_flush();
+      return -pid;
+    }
+  } else toys.exitval = 127;
+  return pid;
+}
+
 void xargs_main(void)
 {
   struct double_list *dlist = 0, *dtemp;
-  int entries, bytes, done = 0, status;
+  int entries, bytes, done = 0;
   char *data = 0, **out;
   pid_t pid = 0;
 
@@ -150,7 +189,7 @@ void xargs_main(void)
 
     if (!TT.entries) {
       if (data) error_exit("argument too long");
-      else if (pid) return;
+      else if (pid) break;
       else if (FLAG(r)) continue;
     }
 
@@ -174,27 +213,15 @@ void xargs_main(void)
       } else fprintf(stderr, "\n");
     }
 
+    while ((pid = wait_for_pid((TT.P > 0 && TT.np >= TT.P) ? -1 : 0)) > 0);
+    if (pid < 0) break;
+    ++TT.np;
     if (!(pid = XVFORK())) {
       close(0);
       xopen_stdio(FLAG(o) ? "/dev/tty" : "/dev/null", O_RDONLY);
       xexec(out);
     }
-    waitpid(pid, &status, 0);
-
-    // xargs is yet another weird collection of exit value special cases,
-    // different from all the others.
-    if (WIFEXITED(status)) {
-      if (WEXITSTATUS(status) == 126 || WEXITSTATUS(status) == 127) {
-        toys.exitval = WEXITSTATUS(status);
-        return;
-      } else if (WEXITSTATUS(status) >= 1 && WEXITSTATUS(status) <= 125) {
-        toys.exitval = 123;
-      } else if (WEXITSTATUS(status) == 255) {
-        error_msg("%s: exited with status 255; aborting", out[0]);
-        toys.exitval = 124;
-        return;
-      }
-    } else toys.exitval = 127;
+    if (TT.P == 1 && wait_for_pid(pid) < 0) break;
 
     // Abritrary number of execs, can't just leak memory each time...
 skip:
@@ -202,5 +229,6 @@ skip:
     dlist = 0;
     free(out);
   }
+  while (wait_for_pid(-1) > 0);
   if (TT.tty) fclose(TT.tty);
 }
-- 
2.28.0.163.g6104cc2f0b6-goog