lib/pipe-filter-ii.c

   1 /* Filtering of data through a subprocess.
   2    Copyright (C) 2001-2003, 2008-2024 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 <https://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 <stdint.h>
  25 #include <stdlib.h>
  26 #include <unistd.h>
  27 #if defined _WIN32 && ! defined __CYGWIN__
  28 # include <windows.h>
  29 # include <process.h> /* _beginthreadex, _endthreadex */
  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 __CYGWIN__) || defined __KLIBC__
 166
 167 struct locals
 168 {
 169   /* Arguments passed to pipe_filter_ii_execute.  */
 170   prepare_write_fn prepare_write;
 171   done_write_fn done_write;
 172   prepare_read_fn prepare_read;
 173   done_read_fn done_read;
 174
 175   /* Management of the subprocess.  */
 176   void *private_data;
 177   int fd[2];
 178
 179   /* Status of the writer part.  */
 180   volatile bool writer_terminated;
 181   volatile int writer_errno;
 182   /* Status of the reader part.  */
 183   volatile bool reader_terminated;
 184   volatile int reader_errno;
 185 };
 186
 187 static unsigned int WINAPI
 188 writer_thread_func (void *thread_arg)
 189 {
 190   struct locals *l = (struct locals *) thread_arg;
 191
 192   for (;;)
 193     {
 194       size_t bufsize;
 195       const void *buf = l->prepare_write (&bufsize, l->private_data);
 196       if (buf != NULL)
 197         {
 198           ssize_t nwritten =
 199             write (l->fd[1], buf, bufsize > SSIZE_MAX ? SSIZE_MAX : bufsize);
 200           if (nwritten < 0)
 201             {
 202               /* Don't assume that the gnulib modules 'write' and 'sigpipe' are
 203                  used.  */
 204               if (GetLastError () == ERROR_NO_DATA)
 205                 errno = EPIPE;
 206               l->writer_errno = errno;
 207               break;
 208             }
 209           else if (nwritten > 0)
 210             l->done_write ((void *) buf, nwritten, l->private_data);
 211         }
 212       else
 213         break;
 214     }
 215
 216   l->writer_terminated = true;
 217   _endthreadex (0); /* calls ExitThread (0) */
 218   abort ();
 219 }
 220
 221 static unsigned int WINAPI
 222 reader_thread_func (void *thread_arg)
 223 {
 224   struct locals *l = (struct locals *) thread_arg;
 225
 226   for (;;)
 227     {
 228       size_t bufsize;
 229       void *buf = l->prepare_read (&bufsize, l->private_data);
 230       if (!(buf != NULL && bufsize > 0))
 231         /* prepare_read returned wrong values.  */
 232         abort ();
 233       {
 234         ssize_t nread =
 235           read (l->fd[0], buf, bufsize > SSIZE_MAX ? SSIZE_MAX : bufsize);
 236         if (nread < 0)
 237           {
 238             l->reader_errno = errno;
 239             break;
 240           }
 241         else if (nread > 0)
 242           l->done_read (buf, nread, l->private_data);
 243         else /* nread == 0 */
 244           break;
 245       }
 246     }
 247
 248   l->reader_terminated = true;
 249   _endthreadex (0); /* calls ExitThread (0) */
 250   abort ();
 251 }
 252
 253 #endif
 254
 255 int
 256 pipe_filter_ii_execute (const char *progname,
 257                         const char *prog_path, const char * const *prog_argv,
 258                         bool null_stderr, bool exit_on_error,
 259                         prepare_write_fn prepare_write,
 260                         done_write_fn done_write,
 261                         prepare_read_fn prepare_read,
 262                         done_read_fn done_read,
 263                         void *private_data)
 264 {
 265   pid_t child;
 266   int fd[2];
 267 #if !((defined _WIN32 && ! defined __CYGWIN__) || defined __KLIBC__)
 268   struct sigaction orig_sigpipe_action;
 269 #endif
 270
 271   /* Open a bidirectional pipe to a subprocess.  */
 272   child = create_pipe_bidi (progname, prog_path, prog_argv,
 273                             NULL, null_stderr, true, exit_on_error,
 274                             fd);
 275   if (child == -1)
 276     return -1;
 277
 278 #if (defined _WIN32 && ! defined __CYGWIN__) || defined __KLIBC__
 279   /* Native Windows API.  */
 280   /* Pipes have a non-blocking mode, see function SetNamedPipeHandleState and
 281      the article "Named Pipe Type, Read, and Wait Modes", but Microsoft's
 282      documentation discourages its use.  So don't use it.
 283      Asynchronous I/O is also not suitable because it notifies the caller only
 284      about completion of the I/O request, not about intermediate progress.
 285      So do the writing and the reading in separate threads.  */
 286   {
 287     struct locals l;
 288     HANDLE handles[2];
 289     #define writer_thread_handle handles[0]
 290     #define reader_thread_handle handles[1]
 291     bool writer_cleaned_up;
 292     bool reader_cleaned_up;
 293
 294     l.prepare_write = prepare_write;
 295     l.done_write = done_write;
 296     l.prepare_read = prepare_read;
 297     l.done_read = done_read;
 298     l.private_data = private_data;
 299     l.fd[0] = fd[0];
 300     l.fd[1] = fd[1];
 301     l.writer_terminated = false;
 302     l.writer_errno = 0;
 303     l.reader_terminated = false;
 304     l.reader_errno = 0;
 305
 306     writer_thread_handle =
 307       (HANDLE) _beginthreadex (NULL, 100000, writer_thread_func, &l, 0, NULL);
 308     reader_thread_handle =
 309       (HANDLE) _beginthreadex (NULL, 100000, reader_thread_func, &l, 0, NULL);
 310     if (writer_thread_handle == NULL || reader_thread_handle == NULL)
 311       {
 312         if (exit_on_error)
 313           error (EXIT_FAILURE, 0, _("creation of threads failed"));
 314         if (reader_thread_handle != NULL)
 315           CloseHandle (reader_thread_handle);
 316         if (writer_thread_handle != NULL)
 317           CloseHandle (writer_thread_handle);
 318         goto fail;
 319       }
 320     writer_cleaned_up = false;
 321     reader_cleaned_up = false;
 322     for (;;)
 323       {
 324         DWORD ret;
 325
 326         /* Here !(writer_cleaned_up && reader_cleaned_up).  */
 327         if (writer_cleaned_up)
 328           ret = WaitForSingleObject (reader_thread_handle, INFINITE);
 329         else if (reader_cleaned_up)
 330           ret = WaitForSingleObject (writer_thread_handle, INFINITE);
 331         else
 332           ret = WaitForMultipleObjects (2, handles, FALSE, INFINITE);
 333         if (!(ret == WAIT_OBJECT_0 + 0 || ret == WAIT_OBJECT_0 + 1))
 334           abort ();
 335
 336         if (l.writer_terminated)
 337           {
 338             /* The writer thread has just terminated.  */
 339             l.writer_terminated = false;
 340             CloseHandle (writer_thread_handle);
 341             if (l.writer_errno)
 342               {
 343                 if (exit_on_error)
 344                   error (EXIT_FAILURE, l.writer_errno,
 345                          _("write to %s subprocess failed"), progname);
 346                 if (!reader_cleaned_up)
 347                   {
 348                     TerminateThread (reader_thread_handle, 1);
 349                     CloseHandle (reader_thread_handle);
 350                   }
 351                 goto fail;
 352               }
 353             /* Tell the child there is nothing more the parent will send.  */
 354             close (fd[1]);
 355             writer_cleaned_up = true;
 356           }
 357         if (l.reader_terminated)
 358           {
 359             /* The reader thread has just terminated.  */
 360             l.reader_terminated = false;
 361             CloseHandle (reader_thread_handle);
 362             if (l.reader_errno)
 363               {
 364                 if (exit_on_error)
 365                   error (EXIT_FAILURE, l.reader_errno,
 366                          _("read from %s subprocess failed"), progname);
 367                 if (!writer_cleaned_up)
 368                   {
 369                     TerminateThread (writer_thread_handle, 1);
 370                     CloseHandle (writer_thread_handle);
 371                   }
 372                 goto fail;
 373               }
 374             reader_cleaned_up = true;
 375           }
 376         if (writer_cleaned_up && reader_cleaned_up)
 377           break;
 378       }
 379   }
 380 #else
 381   /* When we write to the child process and it has just terminated,
 382      we don't want to die from a SIGPIPE signal.  So set the SIGPIPE
 383      handler to SIG_IGN, and handle EPIPE error codes in write().  */
 384   {
 385     struct sigaction sigpipe_action;
 386
 387     sigpipe_action.sa_handler = SIG_IGN;
 388     sigpipe_action.sa_flags = 0;
 389     sigemptyset (&sigpipe_action.sa_mask);
 390     if (sigaction (SIGPIPE, &sigpipe_action, &orig_sigpipe_action) < 0)
 391       abort ();
 392   }
 393
 394   {
 395 # if HAVE_SELECT
 396     fd_set readfds;  /* All bits except fd[0] are always cleared.  */
 397     fd_set writefds; /* All bits except fd[1] are always cleared.  */
 398 # endif
 399     bool done_writing;
 400
 401     /* Enable non-blocking I/O.  This permits the read() and write() calls
 402        to return -1/EAGAIN without blocking; this is important for polling
 403        if HAVE_SELECT is not defined.  It also permits the read() and write()
 404        calls to return after partial reads/writes; this is important if
 405        HAVE_SELECT is defined, because select() only says that some data
 406        can be read or written, not how many.  Without non-blocking I/O,
 407        Linux 2.2.17 and BSD systems prefer to block instead of returning
 408        with partial results.  */
 409     {
 410       int fcntl_flags;
 411
 412       if ((fcntl_flags = fcntl (fd[1], F_GETFL, 0)) < 0
 413           || fcntl (fd[1], F_SETFL, fcntl_flags | O_NONBLOCK) == -1
 414           || (fcntl_flags = fcntl (fd[0], F_GETFL, 0)) < 0
 415           || fcntl (fd[0], F_SETFL, fcntl_flags | O_NONBLOCK) == -1)
 416         {
 417           if (exit_on_error)
 418             error (EXIT_FAILURE, errno,
 419                    _("cannot set up nonblocking I/O to %s subprocess"),
 420                    progname);
 421           goto fail;
 422         }
 423     }
 424
 425 # if HAVE_SELECT
 426     FD_ZERO (&readfds);
 427     FD_ZERO (&writefds);
 428 # endif
 429     done_writing = false;
 430     for (;;)
 431       {
 432 # if HAVE_SELECT
 433         int n, retval;
 434
 435         FD_SET (fd[0], &readfds);
 436         n = fd[0] + 1;
 437         if (!done_writing)
 438           {
 439             FD_SET (fd[1], &writefds);
 440             if (n <= fd[1])
 441               n = fd[1] + 1;
 442           }
 443
 444         /* Do EINTR handling here instead of in pipe-filter-aux.h,
 445            because select() cannot be referred to from an inline
 446            function on AIX 7.1.  */
 447         do
 448           retval = select (n, &readfds, (!done_writing ? &writefds : NULL),
 449                            NULL, NULL);
 450         while (retval < 0 && errno == EINTR);
 451         n = retval;
 452
 453         if (n < 0)
 454           {
 455             if (exit_on_error)
 456               error (EXIT_FAILURE, errno,
 457                      _("communication with %s subprocess failed"), progname);
 458             goto fail;
 459           }
 460         if (!done_writing && FD_ISSET (fd[1], &writefds))
 461           goto try_write;
 462         if (FD_ISSET (fd[0], &readfds))
 463           goto try_read;
 464         /* How could select() return if none of the two descriptors is ready?  */
 465         abort ();
 466 # endif
 467
 468         /* Attempt to write.  */
 469 # if HAVE_SELECT
 470       try_write:
 471 # endif
 472         if (!done_writing)
 473           {
 474             size_t bufsize;
 475             const void *buf = prepare_write (&bufsize, private_data);
 476             if (buf != NULL)
 477               {
 478                 /* Writing to a pipe in non-blocking mode is tricky: The
 479                    write() call may fail with EAGAIN, simply because sufficient
 480                    space is not available in the pipe. See POSIX:2008
 481                    <https://pubs.opengroup.org/onlinepubs/9699919799/functions/write.html>.
 482                    This happens actually on AIX and IRIX, when bufsize >= 8192
 483                    (even though PIPE_BUF and pathconf ("/", _PC_PIPE_BUF) are
 484                    both 32768).  */
 485                 size_t attempt_to_write =
 486                   (bufsize > SSIZE_MAX ? SSIZE_MAX : bufsize);
 487                 for (;;)
 488                   {
 489                     ssize_t nwritten = write (fd[1], buf, attempt_to_write);
 490                     if (nwritten < 0)
 491                       {
 492                         if (errno == EAGAIN)
 493                           {
 494                             attempt_to_write = attempt_to_write / 2;
 495                             if (attempt_to_write == 0)
 496                               break;
 497                           }
 498                         else if (!IS_EAGAIN (errno))
 499                           {
 500                             if (exit_on_error)
 501                               error (EXIT_FAILURE, errno,
 502                                      _("write to %s subprocess failed"),
 503                                      progname);
 504                             goto fail;
 505                           }
 506                       }
 507                     else
 508                       {
 509                         if (nwritten > 0)
 510                           done_write ((void *) buf, nwritten, private_data);
 511                         break;
 512                       }
 513                   }
 514               }
 515             else
 516               {
 517                 /* Tell the child there is nothing more the parent will send.  */
 518                 close (fd[1]);
 519                 done_writing = true;
 520               }
 521           }
 522 # if HAVE_SELECT
 523         continue;
 524 # endif
 525
 526         /* Attempt to read.  */
 527 # if HAVE_SELECT
 528       try_read:
 529 # endif
 530         {
 531           size_t bufsize;
 532           void *buf = prepare_read (&bufsize, private_data);
 533           if (!(buf != NULL && bufsize > 0))
 534             /* prepare_read returned wrong values.  */
 535             abort ();
 536           {
 537             ssize_t nread =
 538               read (fd[0], buf, bufsize > SSIZE_MAX ? SSIZE_MAX : bufsize);
 539             if (nread < 0)
 540               {
 541                 if (!IS_EAGAIN (errno))
 542                   {
 543                     if (exit_on_error)
 544                       error (EXIT_FAILURE, errno,
 545                              _("read from %s subprocess failed"), progname);
 546                     goto fail;
 547                   }
 548               }
 549             else if (nread > 0)
 550               done_read (buf, nread, private_data);
 551             else /* nread == 0 */
 552               {
 553                 if (done_writing)
 554                   break;
 555               }
 556           }
 557         }
 558 # if HAVE_SELECT
 559         continue;
 560 # endif
 561       }
 562   }
 563
 564   /* Restore SIGPIPE signal handler.  */
 565   if (sigaction (SIGPIPE, &orig_sigpipe_action, NULL) < 0)
 566     abort ();
 567 #endif
 568
 569   close (fd[0]);
 570
 571   /* Remove zombie process from process list.  */
 572   {
 573     int exitstatus =
 574       wait_subprocess (child, progname, false, null_stderr,
 575                        true, exit_on_error, NULL);
 576     if (exitstatus != 0 && exit_on_error)
 577       error (EXIT_FAILURE, 0, _("%s subprocess terminated with exit code %d"),
 578              progname, exitstatus);
 579     return exitstatus;
 580   }
 581
 582  fail:
 583   {
 584     int saved_errno = errno;
 585     close (fd[1]);
 586 #if !((defined _WIN32 && ! defined __CYGWIN__) || defined __KLIBC__)
 587     if (sigaction (SIGPIPE, &orig_sigpipe_action, NULL) < 0)
 588       abort ();
 589 #endif
 590     close (fd[0]);
 591     wait_subprocess (child, progname, true, true, true, false, NULL);
 592     errno = saved_errno;
 593     return -1;
 594   }
 595 }