Blame tests/t-poll.c

Packit fc043f
/* t-poll.c - Check the poll function
Packit fc043f
 * Copyright (C) 2015 g10 Code GmbH
Packit fc043f
 *
Packit fc043f
 * This file is part of libgpg-error.
Packit fc043f
 *
Packit fc043f
 * libgpg-error is free software; you can redistribute it and/or
Packit fc043f
 * modify it under the terms of the GNU Lesser General Public License
Packit fc043f
 * as published by the Free Software Foundation; either version 2.1 of
Packit fc043f
 * the License, or (at your option) any later version.
Packit fc043f
 *
Packit fc043f
 * libgpg-error is distributed in the hope that it will be useful, but
Packit fc043f
 * WITHOUT ANY WARRANTY; without even the implied warranty of
Packit fc043f
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
Packit fc043f
 * Lesser General Public License for more details.
Packit fc043f
 *
Packit fc043f
 * You should have received a copy of the GNU Lesser General Public
Packit fc043f
 * License along with this program; if not, see <https://www.gnu.org/licenses/>.
Packit fc043f
 */
Packit fc043f
Packit fc043f
/* FIXME: We need much better tests that this very basic one.  */
Packit fc043f
Packit fc043f
#if HAVE_CONFIG_H
Packit fc043f
# include <config.h>
Packit fc043f
#endif
Packit fc043f
Packit fc043f
#include <stdio.h>
Packit fc043f
#include <stdlib.h>
Packit fc043f
#include <string.h>
Packit fc043f
#include <assert.h>
Packit fc043f
#include <sys/types.h>
Packit fc043f
#include <unistd.h>
Packit fc043f
#include <errno.h>
Packit fc043f
#ifdef _WIN32
Packit fc043f
# include <windows.h>
Packit fc043f
# include <time.h>
Packit fc043f
#else
Packit fc043f
# ifdef USE_POSIX_THREADS
Packit fc043f
#  include <pthread.h>
Packit fc043f
# endif
Packit fc043f
#endif
Packit fc043f
Packit fc043f
#define PGM "t-poll"
Packit fc043f
Packit fc043f
#include "t-common.h"
Packit fc043f
Packit fc043f
#ifdef _WIN32
Packit fc043f
# define THREAD_RET_TYPE  DWORD WINAPI
Packit fc043f
# define THREAD_RET_VALUE 0
Packit fc043f
#else
Packit fc043f
# define THREAD_RET_TYPE  void *
Packit fc043f
# define THREAD_RET_VALUE NULL
Packit fc043f
#endif
Packit fc043f
Packit fc043f
Packit fc043f
/* Object to convey data to a thread.  */
Packit fc043f
struct thread_arg
Packit fc043f
{
Packit fc043f
  const char *name;
Packit fc043f
  estream_t stream;
Packit fc043f
  volatile int stop_me;
Packit fc043f
#ifdef USE_POSIX_THREADS
Packit fc043f
  pthread_t thread;
Packit fc043f
#elif _WIN32
Packit fc043f
  HANDLE thread;
Packit fc043f
#endif
Packit fc043f
};
Packit fc043f
Packit fc043f
Packit fc043f
static struct thread_arg peer_stdin;  /* Thread to feed the stdin.  */
Packit fc043f
static struct thread_arg peer_stdout; /* Thread to feed the stdout. */
Packit fc043f
static struct thread_arg peer_stderr; /* Thread to feed the stderr. */
Packit fc043f
Packit fc043f
static estream_t test_stdin;
Packit fc043f
static estream_t test_stdout;
Packit fc043f
static estream_t test_stderr;
Packit fc043f
Packit fc043f
#if defined(_WIN32) || defined(USE_POSIX_THREADS)
Packit fc043f
Packit fc043f
/* This thread feeds data to the given stream.  */
Packit fc043f
static THREAD_RET_TYPE
Packit fc043f
producer_thread (void *argaddr)
Packit fc043f
{
Packit fc043f
  struct thread_arg *arg = argaddr;
Packit fc043f
  int i = 0;
Packit fc043f
Packit fc043f
  (void)arg;
Packit fc043f
Packit fc043f
  while (!arg->stop_me && i++ < 3)
Packit fc043f
    {
Packit fc043f
      show ("thread '%s' about to write\n", arg->name);
Packit fc043f
      es_fprintf (arg->stream, "This is '%s' count=%d\n", arg->name, i);
Packit fc043f
      es_fflush (arg->stream);
Packit fc043f
    }
Packit fc043f
  es_fclose (arg->stream);
Packit fc043f
  return THREAD_RET_VALUE;
Packit fc043f
}
Packit fc043f
Packit fc043f
/* This thread eats data from the given stream.  */
Packit fc043f
static THREAD_RET_TYPE
Packit fc043f
consumer_thread (void *argaddr)
Packit fc043f
{
Packit fc043f
  struct thread_arg *arg = argaddr;
Packit fc043f
  char buf[15];
Packit fc043f
Packit fc043f
  (void)arg;
Packit fc043f
Packit fc043f
  while (!arg->stop_me)
Packit fc043f
    {
Packit fc043f
      show ("thread '%s' ready to read\n", arg->name);
Packit fc043f
      if (!es_fgets (buf, sizeof buf, arg->stream))
Packit fc043f
        {
Packit fc043f
          show ("Thread '%s' received EOF or error\n", arg->name);
Packit fc043f
          break;
Packit fc043f
        }
Packit fc043f
      show ("Thread '%s' got: '%s'\n", arg->name, buf);
Packit fc043f
    }
Packit fc043f
  es_fclose (arg->stream);
Packit fc043f
  return THREAD_RET_VALUE;
Packit fc043f
}
Packit fc043f
Packit fc043f
#endif /*_WIN32 || USE_POSIX_THREADS */
Packit fc043f
Packit fc043f
Packit fc043f
static void
Packit fc043f
launch_thread (THREAD_RET_TYPE (*fnc)(void *), struct thread_arg *th)
Packit fc043f
{
Packit fc043f
  int fd;
Packit fc043f
Packit fc043f
  th->stop_me = 0;
Packit fc043f
  fd = es_fileno (th->stream);
Packit fc043f
#ifdef _WIN32
Packit fc043f
Packit fc043f
  th->thread = CreateThread (NULL, 0, fnc, th, 0, NULL);
Packit fc043f
  if (!th->thread)
Packit fc043f
    die ("creating thread '%s' failed: rc=%d", th->name, (int)GetLastError ());
Packit fc043f
  show ("thread '%s' launched (fd=%d)\n", th->name, fd);
Packit fc043f
Packit fc043f
#elif USE_POSIX_THREADS
Packit fc043f
Packit fc043f
  if (pthread_create (&th->thread, NULL, fnc, th))
Packit fc043f
    die ("creating thread '%s' failed: %s\n", th->name, strerror (errno));
Packit fc043f
  show ("thread '%s' launched (fd=%d)\n", th->name, fd);
Packit fc043f
Packit fc043f
# else /* no thread support */
Packit fc043f
Packit fc043f
  verbose++;
Packit fc043f
  show ("no thread support - skipping test\n", PGM);
Packit fc043f
  verbose--;
Packit fc043f
Packit fc043f
#endif /* no thread support */
Packit fc043f
}
Packit fc043f
Packit fc043f
Packit fc043f
static void
Packit fc043f
join_thread (struct thread_arg *th)
Packit fc043f
{
Packit fc043f
#ifdef _WIN32
Packit fc043f
  int rc;
Packit fc043f
Packit fc043f
  rc = WaitForSingleObject (th->thread, INFINITE);
Packit fc043f
  if (rc == WAIT_OBJECT_0)
Packit fc043f
    show ("thread '%s' has terminated\n", th->name);
Packit fc043f
  else
Packit fc043f
    fail ("waiting for thread '%s' failed: %d", th->name, (int)GetLastError ());
Packit fc043f
  CloseHandle (th->thread);
Packit fc043f
Packit fc043f
#elif USE_POSIX_THREADS
Packit fc043f
Packit fc043f
  pthread_join (th->thread, NULL);
Packit fc043f
  show ("thread '%s' has terminated\n", th->name);
Packit fc043f
Packit fc043f
#endif
Packit fc043f
}
Packit fc043f
Packit fc043f
Packit fc043f
static void
Packit fc043f
create_pipe (estream_t *r_in, estream_t *r_out)
Packit fc043f
{
Packit fc043f
  gpg_error_t err;
Packit fc043f
  int filedes[2];
Packit fc043f
Packit fc043f
#ifdef _WIN32
Packit fc043f
  if (_pipe (filedes, 512, 0) == -1)
Packit fc043f
#else
Packit fc043f
  if (pipe (filedes) == -1)
Packit fc043f
#endif
Packit fc043f
    {
Packit fc043f
      err = gpg_error_from_syserror ();
Packit fc043f
      die ("error creating a pipe: %s\n", gpg_strerror (err));
Packit fc043f
    }
Packit fc043f
Packit fc043f
  show ("created pipe [%d, %d]\n", filedes[0], filedes[1]);
Packit fc043f
Packit fc043f
  *r_in = es_fdopen (filedes[0], "r,pollable");
Packit fc043f
  if (!*r_in)
Packit fc043f
    {
Packit fc043f
      err = gpg_error_from_syserror ();
Packit fc043f
      die ("error creating a stream for a pipe: %s\n", gpg_strerror (err));
Packit fc043f
    }
Packit fc043f
Packit fc043f
  *r_out = es_fdopen (filedes[1], "w,pollable");
Packit fc043f
  if (!*r_out)
Packit fc043f
    {
Packit fc043f
      err = gpg_error_from_syserror ();
Packit fc043f
      die ("error creating a stream for a pipe: %s\n", gpg_strerror (err));
Packit fc043f
    }
Packit fc043f
}
Packit fc043f
Packit fc043f
Packit fc043f
static void
Packit fc043f
test_poll (void)
Packit fc043f
{
Packit fc043f
  int ret;
Packit fc043f
  gpgrt_poll_t fds[3];
Packit fc043f
  char buffer[16];
Packit fc043f
  size_t used, nwritten;
Packit fc043f
  int c;
Packit fc043f
Packit fc043f
  memset (fds, 0, sizeof fds);
Packit fc043f
  fds[0].stream = test_stdin;
Packit fc043f
  fds[0].want_read = 1;
Packit fc043f
  fds[1].stream = test_stdout;
Packit fc043f
  fds[1].want_write = 1;
Packit fc043f
  /* FIXME: We don't use the next stream at all.  */
Packit fc043f
  fds[2].stream = test_stderr;
Packit fc043f
  fds[2].want_write = 1;
Packit fc043f
  fds[2].ignore = 1;
Packit fc043f
Packit fc043f
Packit fc043f
  used = 0;
Packit fc043f
  while (used || !fds[0].ignore)
Packit fc043f
    {
Packit fc043f
      ret = gpgrt_poll (fds, DIM(fds), -1);
Packit fc043f
      if (ret == -1)
Packit fc043f
        {
Packit fc043f
          fail ("gpgrt_poll failed: %s\n", strerror (errno));
Packit fc043f
          continue;
Packit fc043f
        }
Packit fc043f
      if (!ret)
Packit fc043f
        {
Packit fc043f
          fail ("gpgrt_poll unexpectedly timed out\n");
Packit fc043f
          continue;
Packit fc043f
        }
Packit fc043f
Packit fc043f
      show ("gpgrt_poll detected %d events\n", ret);
Packit fc043f
      if (debug)
Packit fc043f
        show ("gpgrt_poll: r=%d"
Packit fc043f
              " 0:%c%c%c%c%c%c%c%c%c%c%c%c"
Packit fc043f
              " 1:%c%c%c%c%c%c%c%c%c%c%c%c"
Packit fc043f
              " 2:%c%c%c%c%c%c%c%c%c%c%c%c"
Packit fc043f
              "\n",
Packit fc043f
              ret,
Packit fc043f
              fds[0].want_read?  'r':'-',
Packit fc043f
              fds[0].want_write? 'w':'-',
Packit fc043f
              fds[0].want_oob?   'o':'-',
Packit fc043f
              fds[0].want_rdhup? 'h':'-',
Packit fc043f
              fds[0].ignore?     '!':'=',
Packit fc043f
              fds[0].got_read?   'r':'-',
Packit fc043f
              fds[0].got_write?  'w':'-',
Packit fc043f
              fds[0].got_oob?    'o':'-',
Packit fc043f
              fds[0].got_rdhup?  'h':'-',
Packit fc043f
              fds[0].got_hup?    'H':' ',
Packit fc043f
              fds[0].got_err?    'e':' ',
Packit fc043f
              fds[0].got_nval?   'n':' ',
Packit fc043f
Packit fc043f
              fds[1].want_read?  'r':'-',
Packit fc043f
              fds[1].want_write? 'w':'-',
Packit fc043f
              fds[1].want_oob?   'o':'-',
Packit fc043f
              fds[1].want_rdhup? 'h':'-',
Packit fc043f
              fds[1].ignore?     '!':'=',
Packit fc043f
              fds[1].got_read?   'r':'-',
Packit fc043f
              fds[1].got_write?  'w':'-',
Packit fc043f
              fds[1].got_oob?    'o':'-',
Packit fc043f
              fds[1].got_rdhup?  'h':'-',
Packit fc043f
              fds[1].got_hup?    'H':' ',
Packit fc043f
              fds[1].got_err?    'e':' ',
Packit fc043f
              fds[1].got_nval?   'n':' ',
Packit fc043f
Packit fc043f
              fds[2].want_read?  'r':'-',
Packit fc043f
              fds[2].want_write? 'w':'-',
Packit fc043f
              fds[2].want_oob?   'o':'-',
Packit fc043f
              fds[2].want_rdhup? 'h':'-',
Packit fc043f
              fds[2].ignore?     '!':'=',
Packit fc043f
              fds[2].got_read?   'r':'-',
Packit fc043f
              fds[2].got_write?  'w':'-',
Packit fc043f
              fds[2].got_oob?    'o':'-',
Packit fc043f
              fds[2].got_rdhup?  'h':'-',
Packit fc043f
              fds[2].got_hup?    'H':' ',
Packit fc043f
              fds[2].got_err?    'e':' ',
Packit fc043f
              fds[2].got_nval?   'n':' '
Packit fc043f
              );
Packit fc043f
      else
Packit fc043f
        show ("gpgrt_poll detected %d events\n", ret);
Packit fc043f
Packit fc043f
      if (fds[0].got_read)
Packit fc043f
        {
Packit fc043f
          /* Read from the producer.  */
Packit fc043f
          for (;;)
Packit fc043f
            {
Packit fc043f
              c = es_fgetc (fds[0].stream);
Packit fc043f
              if (c == EOF)
Packit fc043f
                {
Packit fc043f
                  if (es_feof (fds[0].stream))
Packit fc043f
                    {
Packit fc043f
                      show ("reading '%s': EOF\n", peer_stdin.name);
Packit fc043f
                      fds[0].ignore = 1; /* Not anymore needed.  */
Packit fc043f
                      peer_stdin.stop_me = 1; /* Tell the thread to stop.  */
Packit fc043f
                    }
Packit fc043f
                  else if (es_ferror (fds[0].stream))
Packit fc043f
                    {
Packit fc043f
                      fail ("error reading '%s': %s\n",
Packit fc043f
                            peer_stdin.name, strerror (errno));
Packit fc043f
                      fds[0].ignore = 1;    /* Disable.  */
Packit fc043f
                      peer_stdin.stop_me = 1; /* Tell the thread to stop.  */
Packit fc043f
                    }
Packit fc043f
                  else
Packit fc043f
                    show ("reading '%s': EAGAIN\n", peer_stdin.name);
Packit fc043f
                  break;
Packit fc043f
                }
Packit fc043f
              else
Packit fc043f
                {
Packit fc043f
                  if (used <= sizeof buffer -1)
Packit fc043f
                    buffer[used++] = c;
Packit fc043f
                  if (used == sizeof buffer)
Packit fc043f
                    {
Packit fc043f
                      show ("throttling reading from '%s'\n", peer_stdin.name);
Packit fc043f
                      fds[0].ignore = 1;
Packit fc043f
                      break;
Packit fc043f
                    }
Packit fc043f
                }
Packit fc043f
            }
Packit fc043f
          show ("read from '%s': %zu bytes\n", peer_stdin.name, used);
Packit fc043f
          if (used)
Packit fc043f
            fds[1].ignore = 0; /* Data to send.  */
Packit fc043f
        }
Packit fc043f
      if (fds[1].got_write)
Packit fc043f
        {
Packit fc043f
          if (used)
Packit fc043f
            {
Packit fc043f
              ret = es_write (fds[1].stream, buffer, used, &nwritten);
Packit fc043f
              show ("result for writing to '%s': ret=%d, n=%zu, nwritten=%zu\n",
Packit fc043f
                    peer_stdout.name, ret, used, nwritten);
Packit fc043f
              if (!ret)
Packit fc043f
                {
Packit fc043f
                  assert (nwritten <= used);
Packit fc043f
                  /* Move the remaining data to the front of buffer.  */
Packit fc043f
                  memmove (buffer, buffer + nwritten,
Packit fc043f
                           sizeof buffer - nwritten);
Packit fc043f
                  used -= nwritten;
Packit fc043f
                }
Packit fc043f
              ret = es_fflush (fds[1].stream);
Packit fc043f
              if (ret)
Packit fc043f
                fail ("Flushing for '%s' failed: %s\n",
Packit fc043f
                      peer_stdout.name, strerror (errno));
Packit fc043f
            }
Packit fc043f
          if (!used)
Packit fc043f
            fds[1].ignore = 1; /* No need to send data.  */
Packit fc043f
        }
Packit fc043f
Packit fc043f
      if (used < sizeof buffer / 2 && !peer_stdin.stop_me && fds[0].ignore)
Packit fc043f
        {
Packit fc043f
          show ("accelerate reading from '%s'\n", peer_stdin.name);
Packit fc043f
          fds[0].ignore = 0;
Packit fc043f
        }
Packit fc043f
    }
Packit fc043f
}
Packit fc043f
Packit fc043f
Packit fc043f
int
Packit fc043f
main (int argc, char **argv)
Packit fc043f
{
Packit fc043f
  int last_argc = -1;
Packit fc043f
Packit fc043f
  if (argc)
Packit fc043f
    {
Packit fc043f
      argc--; argv++;
Packit fc043f
    }
Packit fc043f
  while (argc && last_argc != argc )
Packit fc043f
    {
Packit fc043f
      last_argc = argc;
Packit fc043f
      if (!strcmp (*argv, "--help"))
Packit fc043f
        {
Packit fc043f
          puts (
Packit fc043f
"usage: ./t-poll [options]\n"
Packit fc043f
"\n"
Packit fc043f
"Options:\n"
Packit fc043f
"  --verbose      Show what is going on\n"
Packit fc043f
"  --debug        Flyswatter\n"
Packit fc043f
);
Packit fc043f
          exit (0);
Packit fc043f
        }
Packit fc043f
      if (!strcmp (*argv, "--verbose"))
Packit fc043f
        {
Packit fc043f
          verbose = 1;
Packit fc043f
          argc--; argv++;
Packit fc043f
        }
Packit fc043f
      else if (!strcmp (*argv, "--debug"))
Packit fc043f
        {
Packit fc043f
          verbose = debug = 1;
Packit fc043f
          argc--; argv++;
Packit fc043f
        }
Packit fc043f
    }
Packit fc043f
Packit fc043f
  if (!gpg_error_check_version (GPG_ERROR_VERSION))
Packit fc043f
    {
Packit fc043f
      die ("gpg_error_check_version returned an error");
Packit fc043f
      errorcount++;
Packit fc043f
    }
Packit fc043f
Packit fc043f
  peer_stdin.name  = "stdin producer";
Packit fc043f
  create_pipe (&test_stdin, &peer_stdin.stream);
Packit fc043f
  peer_stdout.name = "stdout consumer";
Packit fc043f
  create_pipe (&peer_stdout.stream, &test_stdout);
Packit fc043f
  peer_stderr.name = "stderr consumer";
Packit fc043f
  create_pipe (&peer_stderr.stream, &test_stderr);
Packit fc043f
Packit fc043f
  if (es_set_nonblock (test_stdin, 1))
Packit fc043f
    fail ("error setting test_stdin to nonblock: %s\n", strerror (errno));
Packit fc043f
  if (es_set_nonblock (test_stdout, 1))
Packit fc043f
    fail ("error setting test_stdout to nonblock: %s\n", strerror (errno));
Packit fc043f
  if (es_set_nonblock (test_stderr, 1))
Packit fc043f
    fail ("error setting test_stderr to nonblock: %s\n", strerror (errno));
Packit fc043f
Packit fc043f
  launch_thread (producer_thread, &peer_stdin );
Packit fc043f
  launch_thread (consumer_thread, &peer_stdout);
Packit fc043f
  launch_thread (consumer_thread, &peer_stderr);
Packit fc043f
  test_poll ();
Packit fc043f
  show ("Waiting for threads to terminate...\n");
Packit fc043f
  es_fclose (test_stdin);
Packit fc043f
  es_fclose (test_stdout);
Packit fc043f
  es_fclose (test_stderr);
Packit fc043f
  peer_stdin.stop_me = 1;
Packit fc043f
  peer_stdout.stop_me = 1;
Packit fc043f
  peer_stderr.stop_me = 1;
Packit fc043f
  join_thread (&peer_stdin);
Packit fc043f
  join_thread (&peer_stdout);
Packit fc043f
  join_thread (&peer_stderr);
Packit fc043f
Packit fc043f
  return errorcount ? 1 : 0;
Packit fc043f
}