estream: Add gpgrt_set_nonblock and gpgrt_poll.

* configure.ac (AC_CHECK_HEADERS): Add sys/select.h and sys/time.h.
* src/estream.c: Include both header if available.
(COOKIE_IOCTL_NONBLOCK): New.
(struct estream_cookie_fd): Add field nonblock.
(func_fd_create): Set nonblock from MODEFLAGS.
(es_func_fd_ioctl): New.
(parse_mode): Add modeflag "nonblock".
(es_fill): Map EWOULDBLOCK to EAGAIN.  Do not set error indicator for
EAGAIN.
(es_flush, es_seek, es_write_nbf): Map EWOULDBLOCK to EAGAIN.
(do_fdopen): Call COOKIE_IOCTL_NONBLOCK.
(_gpgrt_set_nonblock): New.
(_gpgrt_get_nonblock): New.
(_gpgrt_poll): New.

* src/gpg-error.h.in (struct _gpgrt_poll_s): New.
(gpgrt_poll_t, es_poll_t): New.
(es_set_nonblock, es_get_nonblock, es_poll): New.

* src/gpg-error.vers, src/gpg-error.def.in: Add gpgrt_set_nonblock,
gpgrt_get_nonblock, and gpgrt_poll.
* src/visibility.c (gpgrt_set_nonblock, gpgrt_get_nonblock): New.
(gpgrt_poll): New.

* tests/t-common.h (DIM): New.
* tests/t-poll.c: New.
* tests/Makefile.am (TESTS): Add t-poll.
(t_poll_LDADD): New.
--

The poll interface uses select(2) internally because that is more
portable than poll(2).

Signed-off-by: Werner Koch <wk@gnupg.org>

3 files changed, 388 insertions, 1 deletions

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