lib/pipe-filter-ii.c

   1 /* Filtering of data through a subprocess.
   2    Copyright (C) 2001-2003, 2008-2016 Free Software Foundation, Inc.
   3    Written by Bruno Haible <bruno@clisp.org>, 2009.
   4
   5    This program is free software: you can redistribute it and/or modify
   6    it under the terms of the GNU General Public License as published by
   7    the Free Software Foundation; either version 3 of the License, or
   8    (at your option) any later version.
   9
  10    This program is distributed in the hope that it will be useful,
  11    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13    GNU General Public License for more details.
  14
  15    You should have received a copy of the GNU General Public License
  16    along with this program.  If not, see <http://www.gnu.org/licenses/>.  */
  17
  18 #include <config.h>
  19
  20 #include "pipe-filter.h"
  21
  22 #include <errno.h>
  23 #include <fcntl.h>
  24 #include <stdbool.h>
  25 #include <stdint.h>
  26 #include <stdlib.h>
  27 #include <unistd.h>
  28 #if (defined _WIN32 || defined __WIN32__) && ! defined __CYGWIN__
  29 # include <windows.h>
  30 #elif defined __KLIBC__
  31 # define INCL_DOS
  32 # include <os2.h>
  33
  34 /* Simple implementation of Win32 APIs */
  35
  36 # define WINAPI
  37
  38 typedef struct _HANDLE
  39 {
  40   TID tid;
  41   HEV hevDone;
  42   unsigned int WINAPI (*start) (void *);
  43   void *arg;
  44 } *HANDLE;
  45
  46 typedef ULONG DWORD;
  47
  48 static void
  49 start_wrapper (void *arg)
  50 {
  51   HANDLE h = (HANDLE) arg;
  52
  53   h->start (h->arg);
  54
  55   DosPostEventSem (h->hevDone);
  56   _endthread ();
  57 }
  58
  59 static HANDLE
  60 _beginthreadex (void *s, unsigned n, unsigned int WINAPI (*start) (void *),
  61                 void *arg, unsigned fl, unsigned *th)
  62 {
  63   HANDLE h;
  64
  65   h = malloc (sizeof (*h));
  66   if (!h)
  67     return NULL;
  68
  69   if (DosCreateEventSem (NULL, &h->hevDone, 0, FALSE))
  70     goto exit_free;
  71
  72   h->start = start;
  73   h->arg = arg;
  74
  75   h->tid = _beginthread (start_wrapper, NULL, n, (void *) h);
  76   if (h->tid == -1)
  77     goto exit_close_event_sem;
  78
  79   return h;
  80
  81  exit_close_event_sem:
  82   DosCloseEventSem (h->hevDone);
  83
  84  exit_free:
  85   free (h);
  86
  87   return NULL;
  88 }
  89
  90 static BOOL
  91 CloseHandle (HANDLE h)
  92 {
  93   DosCloseEventSem (h->hevDone);
  94   free (h);
  95 }
  96
  97 # define _endthreadex(x) return (x)
  98 # define TerminateThread(h, e) DosKillThread (h->tid)
  99
 100 # define GetLastError()  -1
 101
 102 # ifndef ERROR_NO_DATA
 103 #  define ERROR_NO_DATA 232
 104 # endif
 105
 106 # define INFINITE SEM_INDEFINITE_WAIT
 107 # define WAIT_OBJECT_0  0
 108
 109 static DWORD
 110 WaitForSingleObject (HANDLE h, DWORD ms)
 111 {
 112   return DosWaitEventSem (h->hevDone, ms) == 0 ? WAIT_OBJECT_0 : (DWORD) -1;
 113 }
 114
 115 static DWORD
 116 WaitForMultipleObjects (DWORD nCount, const HANDLE *pHandles, BOOL bWaitAll,
 117                         DWORD ms)
 118 {
 119   HMUX hmux;
 120   PSEMRECORD psr;
 121   ULONG ulUser;
 122   ULONG rc = (ULONG) -1;
 123   DWORD i;
 124
 125   psr = malloc (sizeof (*psr) * nCount);
 126   if (!psr)
 127     goto exit_return;
 128
 129   for (i = 0; i < nCount; ++i)
 130     {
 131       psr[i].hsemCur = (HSEM) pHandles[i]->hevDone;
 132       psr[i].ulUser  = WAIT_OBJECT_0 + i;
 133     }
 134
 135   if (DosCreateMuxWaitSem (NULL, &hmux, nCount, psr,
 136                            bWaitAll ? DCMW_WAIT_ALL : DCMW_WAIT_ANY))
 137     goto exit_free;
 138
 139   rc = DosWaitMuxWaitSem (hmux, ms, &ulUser);
 140   DosCloseMuxWaitSem (hmux);
 141
 142  exit_free:
 143   free (psr);
 144
 145  exit_return:
 146   if (rc)
 147     return (DWORD) -1;
 148
 149   return ulUser;
 150 }
 151 #else
 152 # include <signal.h>
 153 # include <sys/select.h>
 154 #endif
 155
 156 #include "error.h"
 157 #include "spawn-pipe.h"
 158 #include "wait-process.h"
 159 #include "gettext.h"
 160
 161 #define _(str) gettext (str)
 162
 163 #include "pipe-filter-aux.h"
 164
 165 #if (((defined _WIN32 || defined __WIN32__) && ! defined __CYGWIN__) \
 166      || defined __KLIBC__)
 167
 168 struct locals
 169 {
 170   /* Arguments passed to pipe_filter_ii_execute.  */
 171   prepare_write_fn prepare_write;
 172   done_write_fn done_write;
 173   prepare_read_fn prepare_read;
 174   done_read_fn done_read;
 175
 176   /* Management of the subprocess.  */
 177   void *private_data;
 178   int fd[2];
 179
 180   /* Status of the writer part.  */
 181   volatile bool writer_terminated;
 182   volatile int writer_errno;
 183   /* Status of the reader part.  */
 184   volatile bool reader_terminated;
 185   volatile int reader_errno;
 186 };
 187
 188 static unsigned int WINAPI
 189 writer_thread_func (void *thread_arg)
 190 {
 191   struct locals *l = (struct locals *) thread_arg;
 192
 193   for (;;)
 194     {
 195       size_t bufsize;
 196       const void *buf = l->prepare_write (&bufsize, l->private_data);
 197       if (buf != NULL)
 198         {
 199           ssize_t nwritten =
 200             write (l->fd[1], buf, bufsize > SSIZE_MAX ? SSIZE_MAX : bufsize);
 201           if (nwritten < 0)
 202             {
 203               /* Don't assume that the gnulib modules 'write' and 'sigpipe' are
 204                  used.  */
 205               if (GetLastError () == ERROR_NO_DATA)
 206                 errno = EPIPE;
 207               l->writer_errno = errno;
 208               break;
 209             }
 210           else if (nwritten > 0)
 211             l->done_write ((void *) buf, nwritten, l->private_data);
 212         }
 213       else
 214         break;
 215     }
 216
 217   l->writer_terminated = true;
 218   _endthreadex (0); /* calls ExitThread (0) */
 219   abort ();
 220 }
 221
 222 static unsigned int WINAPI
 223 reader_thread_func (void *thread_arg)
 224 {
 225   struct locals *l = (struct locals *) thread_arg;
 226
 227   for (;;)
 228     {
 229       size_t bufsize;
 230       void *buf = l->prepare_read (&bufsize, l->private_data);
 231       if (!(buf != NULL && bufsize > 0))
 232         /* prepare_read returned wrong values.  */
 233         abort ();
 234       {
 235         ssize_t nread =
 236           read (l->fd[0], buf, bufsize > SSIZE_MAX ? SSIZE_MAX : bufsize);
 237         if (nread < 0)
 238           {
 239             l->reader_errno = errno;
 240             break;
 241           }
 242         else if (nread > 0)
 243           l->done_read (buf, nread, l->private_data);
 244         else /* nread == 0 */
 245           break;
 246       }
 247     }
 248
 249   l->reader_terminated = true;
 250   _endthreadex (0); /* calls ExitThread (0) */
 251   abort ();
 252 }
 253
 254 #endif
 255
 256 int
 257 pipe_filter_ii_execute (const char *progname,
 258                         const char *prog_path, const char **prog_argv,
 259                         bool null_stderr, bool exit_on_error,
 260                         prepare_write_fn prepare_write,
 261                         done_write_fn done_write,
 262                         prepare_read_fn prepare_read,
 263                         done_read_fn done_read,
 264                         void *private_data)
 265 {
 266   pid_t child;
 267   int fd[2];
 268 #if !(((defined _WIN32 || defined __WIN32__) && ! defined __CYGWIN__) \
 269       || defined __KLIBC__)
 270   struct sigaction orig_sigpipe_action;
 271 #endif
 272
 273   /* Open a bidirectional pipe to a subprocess.  */
 274   child = create_pipe_bidi (progname, prog_path, (char **) prog_argv,
 275                             null_stderr, true, exit_on_error,
 276                             fd);
 277   if (child == -1)
 278     return -1;
 279
 280 #if (((defined _WIN32 || defined __WIN32__) && ! defined __CYGWIN__) \
 281      || defined __KLIBC__)
 282   /* Native Windows API.  */
 283   /* Pipes have a non-blocking mode, see function SetNamedPipeHandleState and
 284      the article "Named Pipe Type, Read, and Wait Modes", but Microsoft's
 285      documentation discourages its use.  So don't use it.
 286      Asynchronous I/O is also not suitable because it notifies the caller only
 287      about completion of the I/O request, not about intermediate progress.
 288      So do the writing and the reading in separate threads.  */
 289   {
 290     struct locals l;
 291     HANDLE handles[2];
 292     #define writer_thread_handle handles[0]
 293     #define reader_thread_handle handles[1]
 294     bool writer_cleaned_up;
 295     bool reader_cleaned_up;
 296
 297     l.prepare_write = prepare_write;
 298     l.done_write = done_write;
 299     l.prepare_read = prepare_read;
 300     l.done_read = done_read;
 301     l.private_data = private_data;
 302     l.fd[0] = fd[0];
 303     l.fd[1] = fd[1];
 304     l.writer_terminated = false;
 305     l.writer_errno = 0;
 306     l.reader_terminated = false;
 307     l.reader_errno = 0;
 308
 309     writer_thread_handle =
 310       (HANDLE) _beginthreadex (NULL, 100000, writer_thread_func, &l, 0, NULL);
 311     reader_thread_handle =
 312       (HANDLE) _beginthreadex (NULL, 100000, reader_thread_func, &l, 0, NULL);
 313     if (writer_thread_handle == NULL || reader_thread_handle == NULL)
 314       {
 315         if (exit_on_error)
 316           error (EXIT_FAILURE, 0, _("creation of threads failed"));
 317         if (reader_thread_handle != NULL)
 318           CloseHandle (reader_thread_handle);
 319         if (writer_thread_handle != NULL)
 320           CloseHandle (writer_thread_handle);
 321         goto fail;
 322       }
 323     writer_cleaned_up = false;
 324     reader_cleaned_up = false;
 325     for (;;)
 326       {
 327         DWORD ret;
 328
 329         /* Here !(writer_cleaned_up && reader_cleaned_up).  */
 330         if (writer_cleaned_up)
 331           ret = WaitForSingleObject (reader_thread_handle, INFINITE);
 332         else if (reader_cleaned_up)
 333           ret = WaitForSingleObject (writer_thread_handle, INFINITE);
 334         else
 335           ret = WaitForMultipleObjects (2, handles, FALSE, INFINITE);
 336         if (!(ret == WAIT_OBJECT_0 + 0 || ret == WAIT_OBJECT_0 + 1))
 337           abort ();
 338
 339         if (l.writer_terminated)
 340           {
 341             /* The writer thread has just terminated.  */
 342             l.writer_terminated = false;
 343             CloseHandle (writer_thread_handle);
 344             if (l.writer_errno)
 345               {
 346                 if (exit_on_error)
 347                   error (EXIT_FAILURE, l.writer_errno,
 348                          _("write to %s subprocess failed"), progname);
 349                 if (!reader_cleaned_up)
 350                   {
 351                     TerminateThread (reader_thread_handle, 1);
 352                     CloseHandle (reader_thread_handle);
 353                   }
 354                 goto fail;
 355               }
 356             /* Tell the child there is nothing more the parent will send.  */
 357             close (fd[1]);
 358             writer_cleaned_up = true;
 359           }
 360         if (l.reader_terminated)
 361           {
 362             /* The reader thread has just terminated.  */
 363             l.reader_terminated = false;
 364             CloseHandle (reader_thread_handle);
 365             if (l.reader_errno)
 366               {
 367                 if (exit_on_error)
 368                   error (EXIT_FAILURE, l.reader_errno,
 369                          _("read from %s subprocess failed"), progname);
 370                 if (!writer_cleaned_up)
 371                   {
 372                     TerminateThread (writer_thread_handle, 1);
 373                     CloseHandle (writer_thread_handle);
 374                   }
 375                 goto fail;
 376               }
 377             reader_cleaned_up = true;
 378           }
 379         if (writer_cleaned_up && reader_cleaned_up)
 380           break;
 381       }
 382   }
 383 #else
 384   /* When we write to the child process and it has just terminated,
 385      we don't want to die from a SIGPIPE signal.  So set the SIGPIPE
 386      handler to SIG_IGN, and handle EPIPE error codes in write().  */
 387   {
 388     struct sigaction sigpipe_action;
 389
 390     sigpipe_action.sa_handler = SIG_IGN;
 391     sigpipe_action.sa_flags = 0;
 392     sigemptyset (&sigpipe_action.sa_mask);
 393     if (sigaction (SIGPIPE, &sigpipe_action, &orig_sigpipe_action) < 0)
 394       abort ();
 395   }
 396
 397   {
 398 # if HAVE_SELECT
 399     fd_set readfds;  /* All bits except fd[0] are always cleared.  */
 400     fd_set writefds; /* All bits except fd[1] are always cleared.  */
 401 # endif
 402     bool done_writing;
 403
 404     /* Enable non-blocking I/O.  This permits the read() and write() calls
 405        to return -1/EAGAIN without blocking; this is important for polling
 406        if HAVE_SELECT is not defined.  It also permits the read() and write()
 407        calls to return after partial reads/writes; this is important if
 408        HAVE_SELECT is defined, because select() only says that some data
 409        can be read or written, not how many.  Without non-blocking I/O,
 410        Linux 2.2.17 and BSD systems prefer to block instead of returning
 411        with partial results.  */
 412     {
 413       int fcntl_flags;
 414
 415       if ((fcntl_flags = fcntl (fd[1], F_GETFL, 0)) < 0
 416           || fcntl (fd[1], F_SETFL, fcntl_flags | O_NONBLOCK) == -1
 417           || (fcntl_flags = fcntl (fd[0], F_GETFL, 0)) < 0
 418           || fcntl (fd[0], F_SETFL, fcntl_flags | O_NONBLOCK) == -1)
 419         {
 420           if (exit_on_error)
 421             error (EXIT_FAILURE, errno,
 422                    _("cannot set up nonblocking I/O to %s subprocess"),
 423                    progname);
 424           goto fail;
 425         }
 426     }
 427
 428 # if HAVE_SELECT
 429     FD_ZERO (&readfds);
 430     FD_ZERO (&writefds);
 431 # endif
 432     done_writing = false;
 433     for (;;)
 434       {
 435 # if HAVE_SELECT
 436         int n, retval;
 437
 438         FD_SET (fd[0], &readfds);
 439         n = fd[0] + 1;
 440         if (!done_writing)
 441           {
 442             FD_SET (fd[1], &writefds);
 443             if (n <= fd[1])
 444               n = fd[1] + 1;
 445           }
 446
 447         /* Do EINTR handling here instead of in pipe-filter-aux.h,
 448            because select() cannot be referred to from an inline
 449            function on AIX 7.1.  */
 450         do
 451           retval = select (n, &readfds, (!done_writing ? &writefds : NULL),
 452                            NULL, NULL);
 453         while (retval < 0 && errno == EINTR);
 454         n = retval;
 455
 456         if (n < 0)
 457           {
 458             if (exit_on_error)
 459               error (EXIT_FAILURE, errno,
 460                      _("communication with %s subprocess failed"), progname);
 461             goto fail;
 462           }
 463         if (!done_writing && FD_ISSET (fd[1], &writefds))
 464           goto try_write;
 465         if (FD_ISSET (fd[0], &readfds))
 466           goto try_read;
 467         /* How could select() return if none of the two descriptors is ready?  */
 468         abort ();
 469 # endif
 470
 471         /* Attempt to write.  */
 472 # if HAVE_SELECT
 473       try_write:
 474 # endif
 475         if (!done_writing)
 476           {
 477             size_t bufsize;
 478             const void *buf = prepare_write (&bufsize, private_data);
 479             if (buf != NULL)
 480               {
 481                 /* Writing to a pipe in non-blocking mode is tricky: The
 482                    write() call may fail with EAGAIN, simply because sufficient
 483                    space is not available in the pipe. See POSIX:2008
 484                    <http://pubs.opengroup.org/onlinepubs/9699919799/functions/write.html>.
 485                    This happens actually on AIX and IRIX, when bufsize >= 8192
 486                    (even though PIPE_BUF and pathconf ("/", _PC_PIPE_BUF) are
 487                    both 32768).  */
 488                 size_t attempt_to_write =
 489                   (bufsize > SSIZE_MAX ? SSIZE_MAX : bufsize);
 490                 for (;;)
 491                   {
 492                     ssize_t nwritten = write (fd[1], buf, attempt_to_write);
 493                     if (nwritten < 0)
 494                       {
 495                         if (errno == EAGAIN)
 496                           {
 497                             attempt_to_write = attempt_to_write / 2;
 498                             if (attempt_to_write == 0)
 499                               break;
 500                           }
 501                         else if (!IS_EAGAIN (errno))
 502                           {
 503                             if (exit_on_error)
 504                               error (EXIT_FAILURE, errno,
 505                                      _("write to %s subprocess failed"),
 506                                      progname);
 507                             goto fail;
 508                           }
 509                       }
 510                     else
 511                       {
 512                         if (nwritten > 0)
 513                           done_write ((void *) buf, nwritten, private_data);
 514                         break;
 515                       }
 516                   }
 517               }
 518             else
 519               {
 520                 /* Tell the child there is nothing more the parent will send.  */
 521                 close (fd[1]);
 522                 done_writing = true;
 523               }
 524           }
 525 # if HAVE_SELECT
 526         continue;
 527 # endif
 528
 529         /* Attempt to read.  */
 530 # if HAVE_SELECT
 531       try_read:
 532 # endif
 533         {
 534           size_t bufsize;
 535           void *buf = prepare_read (&bufsize, private_data);
 536           if (!(buf != NULL && bufsize > 0))
 537             /* prepare_read returned wrong values.  */
 538             abort ();
 539           {
 540             ssize_t nread =
 541               read (fd[0], buf, bufsize > SSIZE_MAX ? SSIZE_MAX : bufsize);
 542             if (nread < 0)
 543               {
 544                 if (!IS_EAGAIN (errno))
 545                   {
 546                     if (exit_on_error)
 547                       error (EXIT_FAILURE, errno,
 548                              _("read from %s subprocess failed"), progname);
 549                     goto fail;
 550                   }
 551               }
 552             else if (nread > 0)
 553               done_read (buf, nread, private_data);
 554             else /* nread == 0 */
 555               {
 556                 if (done_writing)
 557                   break;
 558               }
 559           }
 560         }
 561 # if HAVE_SELECT
 562         continue;
 563 # endif
 564       }
 565   }
 566
 567   /* Restore SIGPIPE signal handler.  */
 568   if (sigaction (SIGPIPE, &orig_sigpipe_action, NULL) < 0)
 569     abort ();
 570 #endif
 571
 572   close (fd[0]);
 573
 574   /* Remove zombie process from process list.  */
 575   {
 576     int exitstatus =
 577       wait_subprocess (child, progname, false, null_stderr,
 578                        true, exit_on_error, NULL);
 579     if (exitstatus != 0 && exit_on_error)
 580       error (EXIT_FAILURE, 0, _("%s subprocess terminated with exit code %d"),
 581              progname, exitstatus);
 582     return exitstatus;
 583   }
 584
 585  fail:
 586   {
 587     int saved_errno = errno;
 588     close (fd[1]);
 589 #if !(((defined _WIN32 || defined __WIN32__) && ! defined __CYGWIN__) \
 590       || defined __KLIBC__)
 591     if (sigaction (SIGPIPE, &orig_sigpipe_action, NULL) < 0)
 592       abort ();
 593 #endif
 594     close (fd[0]);
 595     wait_subprocess (child, progname, true, true, true, false, NULL);
 596     errno = saved_errno;
 597     return -1;
 598   }
 599 }