MWL#192: Non-blocking client API for libmysqlclient.

All client functions that can block on I/O have alternate _start() and
_cont() versions that do not block but return control back to the
application, which can then issue I/O wait in its own fashion and later
call back into the library to continue the operation.

Works behind the scenes by spawning a co-routine/fiber to run the
blocking operation and suspend it while waiting for I/O. This
co-routine/fiber use is invisible to applications.

For i368/x86_64 on GCC, uses very fast assembler co-routine support. On
Windows uses native Win32 Fibers. Falls back to POSIX ucontext on other
platforms. Assembler routines for more platforms are relatively easy to
add by extending mysys/my_context.c, eg. similar to the Lua lcoco
library.

For testing, mysqltest and mysql_client_test are extended with the
option --non-blocking-api. This causes the programs to use the
non-blocking API for database access. mysql-test-run.pl has a similar
option --non-blocking-api that uses this, as well as additional
testcases.

An example program tests/async_queries.c is included that uses the new
non-blocking API with libevent to show how, in a single-threaded
program, to issue many queries in parallel against a database.


client/async_example.c:
  Fix const warning
  ******
  Fix bug with wrong timeout value for poll().
include/Makefile.am:
  Fix missing include for `make dist`
include/mysql.h:
  Add prototypes for all non-blocking API calls.
include/mysql.h.pp:
  Add prototypes for all non-blocking API calls.
mysys/my_context.c:
  Fix type warning for makecontext() function pointer argument.
sql-common/mysql_async.c:
  Fix crashes in the non-blocking API for functions that can take MYSQL argument
  that is NULL.
tests/Makefile.am:
  Add header file to `make dist`
tests/mysql_client_test.c:
  Replace blocking calls with wrappers around the non-blocking calls, used in
  mysql_client_test to test the new non-blocking API.
tests/nonblock-wrappers.h:
  Replace blocking calls with wrappers around the non-blocking calls, used in
  mysql_client_test to test the new non-blocking API.

5 files changed, 1046 insertions, 0 deletions

diff --git a/tests/Makefile.am b/tests/Makefile.am
index 5283edacf25..3610d97135f 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -24,6 +24,7 @@ else
 LIBMYSQLCLIENT_LA =		$(top_builddir)/libmysql/libmysqlclient.la
 endif
 
+noinst_HEADERS = nonblock-wrappers.h
 EXTRA_DIST =		auto_increment.res auto_increment.tst \
 			function.res function.tst lock_test.pl lock_test.res \
 			export.pl big_record.pl \
@@ -36,6 +37,9 @@ EXTRA_DIST =		auto_increment.res auto_increment.tst \
 
 bin_PROGRAMS =		mysql_client_test
 noinst_PROGRAMS =	insert_test select_test thread_test bug25714
+if HAVE_LIBEVENT
+noinst_PROGRAMS +=	async_queries
+endif
 
 INCLUDES =		-I$(top_builddir)/include -I$(top_srcdir)/include \
 			$(openssl_includes)
@@ -52,6 +56,10 @@ select_test_SOURCES=       select_test.c
 insert_test_DEPENDENCIES=	$(LIBRARIES) $(pkglib_LTLIBRARIES)
 select_test_DEPENDENCIES=	$(LIBRARIES) $(pkglib_LTLIBRARIES)
 
+async_queries_SOURCES=	async_queries.c
+async_queries_CFLAGS=	$(AM_CFLAGS) @libevent_includes@
+async_queries_LDADD=	$(LDADD) @libevent_libs@
+
 bug25714_SOURCES=          bug25714.c
 bug25714_DEPENDENCIES=     $(LIBRARIES) $(pkglib_LTLIBRARIES)
 
diff --git a/tests/async_queries.c b/tests/async_queries.c
new file mode 100644
index 00000000000..677208f11f5
--- /dev/null
+++ b/tests/async_queries.c
@@ -0,0 +1,435 @@
+/*
+  Copyright 2011 Kristian Nielsen and Monty Program Ab.
+
+  Experiments with non-blocking libmysql.
+
+  This is free software: you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation, either version 2 of the License, or
+  (at your option) any later version.
+
+  This 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 General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+
+/*
+  Run a set of queries in parallel against a server using the non-blocking
+  API, and compare to running same queries with the normal blocking API.
+*/
+
+#include <sys/time.h>
+#include <event.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+
+#include <my_global.h>
+#include <my_sys.h>
+#include <mysql.h>
+#include <my_getopt.h>
+
+
+#define SL(s) (s), sizeof(s)
+static const char *my_groups[]= { "client", NULL };
+
+/* Maintaining a list of queries to run. */
+struct query_entry {
+  struct query_entry *next;
+  char *query;
+  int index;
+};
+static struct query_entry *query_list;
+static struct query_entry **tail_ptr= &query_list;
+static int query_counter= 0;
+
+
+/* State kept for each connection. */
+struct state_data {
+  int ST;                                    /* State machine current state */
+  struct event ev_mysql;
+  MYSQL mysql;
+  MYSQL_RES *result;
+  MYSQL *ret;
+  int err;
+  MYSQL_ROW row;
+  struct query_entry *query_element;
+  int index;
+};
+
+
+static const char *opt_db= NULL;
+static const char *opt_user= NULL;
+static const char *opt_password= NULL;
+static int tty_password= 0;
+static const char *opt_host= NULL;
+static const char *opt_socket= NULL;
+static unsigned int opt_port= 0;
+static unsigned int opt_connections= 5;
+static const char *opt_query_file= NULL;
+
+static struct my_option options[] =
+{
+  {"database", 'D', "Database to use", &opt_db, &opt_db,
+   0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  {"help", '?', "Display this help and exit", 0, 0, 0, GET_NO_ARG, NO_ARG, 0,
+   0, 0, 0, 0, 0},
+  {"host", 'h', "Connect to host", &opt_host, &opt_host,
+   0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  {"password", 'p',
+   "Password to use when connecting to server. If password is not given it's asked from the tty.",
+   0, 0, 0, GET_STR, OPT_ARG, 0, 0, 0, 0, 0, 0},
+  {"port", 'P', "Port number to use for connection.",
+   &opt_port, &opt_port, 0, GET_UINT, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  {"socket", 'S', "Socket file to use for connection",
+   &opt_socket, &opt_socket, 0, GET_STR,
+   REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  {"user", 'u', "User for login if not current user", &opt_user,
+   &opt_user, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  {"connections", 'n', "Number of simultaneous connections/queries.",
+   &opt_connections, &opt_connections, 0, GET_UINT, REQUIRED_ARG,
+   0, 0, 0, 0, 0, 0},
+  {"queryfile", 'q', "Name of file containing extra queries to run",
+   &opt_query_file, &opt_query_file, 0, GET_STR, REQUIRED_ARG,
+   0, 0, 0, 0, 0, 0},
+  { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
+};
+
+static void
+fatal(struct state_data *sd, const char *msg)
+{
+  fprintf(stderr, "%s: %s\n", msg, (sd ? mysql_error(&sd->mysql) : ""));
+  exit(1);
+}
+
+
+static void state_machine_handler(int fd, short event, void *arg);
+
+static void
+next_event(int new_st, int status, struct state_data *sd)
+{
+  short wait_event= 0;
+  struct timeval tv, *ptv;
+  int fd;
+
+  if (status & MYSQL_WAIT_READ)
+    wait_event|= EV_READ;
+  if (status & MYSQL_WAIT_WRITE)
+    wait_event|= EV_WRITE;
+  if (wait_event)
+    fd= mysql_get_socket(&sd->mysql);
+  else
+    fd= -1;
+  if (status & MYSQL_WAIT_TIMEOUT)
+  {
+    tv.tv_sec= mysql_get_timeout_value(&sd->mysql);
+    tv.tv_usec= 0;
+    ptv= &tv;
+  }
+  else
+    ptv= NULL;
+  event_set(&sd->ev_mysql, fd, wait_event, state_machine_handler, sd);
+  event_add(&sd->ev_mysql, ptv);
+  sd->ST= new_st;
+}
+
+static int
+mysql_status(short event)
+{
+  int status= 0;
+  if (event & EV_READ)
+    status|= MYSQL_WAIT_READ;
+  if (event & EV_WRITE)
+    status|= MYSQL_WAIT_WRITE;
+  if (event & EV_TIMEOUT)
+    status|= MYSQL_WAIT_TIMEOUT;
+  return status;
+}
+
+
+static int num_active_connections;
+
+/* Shortcut for going to new state immediately without waiting. */
+#define NEXT_IMMEDIATE(sd_, new_st) do { sd_->ST= new_st; goto again; } while (0)
+
+static void
+state_machine_handler(int fd __attribute__((unused)), short event, void *arg)
+{
+  struct state_data *sd= arg;
+  int status;
+
+again:
+  switch(sd->ST)
+  {
+  case 0:
+    /* Initial state, start making the connection. */
+    status= mysql_real_connect_start(&sd->ret, &sd->mysql, opt_host, opt_user, opt_password, opt_db, opt_port, opt_socket, 0);
+    if (status)
+      /* Wait for connect to complete. */
+      next_event(1, status, sd);
+    else
+      NEXT_IMMEDIATE(sd, 9);
+    break;
+
+  case 1:
+    status= mysql_real_connect_cont(&sd->ret, &sd->mysql, mysql_status(event));
+    if (status)
+      next_event(1, status, sd);
+    else
+      NEXT_IMMEDIATE(sd, 9);
+    break;
+
+  case 9:
+    if (!sd->ret)
+      fatal(sd, "Failed to mysql_real_connect()");
+    NEXT_IMMEDIATE(sd, 10);
+    break;
+
+  case 10:
+    /* Now run the next query. */
+    sd->query_element= query_list;
+    if (!sd->query_element)
+    {
+      /* No more queries, end the connection. */
+      NEXT_IMMEDIATE(sd, 40);
+    }
+    query_list= query_list->next;
+
+    sd->index= sd->query_element->index;
+    printf("%d ! %s\n", sd->index, sd->query_element->query);
+    status= mysql_real_query_start(&sd->err, &sd->mysql, sd->query_element->query,
+                                   strlen(sd->query_element->query));
+    if (status)
+      next_event(11, status, sd);
+    else
+      NEXT_IMMEDIATE(sd, 20);
+    break;
+
+  case 11:
+    status= mysql_real_query_cont(&sd->err, &sd->mysql, mysql_status(event));
+    if (status)
+      next_event(11, status, sd);
+    else
+      NEXT_IMMEDIATE(sd, 20);
+    break;
+
+  case 20:
+    free(sd->query_element->query);
+    free(sd->query_element);
+    if (sd->err)
+    {
+      printf("%d | Error: %s\n", sd->index, mysql_error(&sd->mysql));
+      NEXT_IMMEDIATE(sd, 10);
+    }
+    else
+    {
+      sd->result= mysql_use_result(&sd->mysql);
+      if (!sd->result)
+        fatal(sd, "mysql_use_result() returns error");
+      NEXT_IMMEDIATE(sd, 30);
+    }
+    break;
+
+  case 30:
+    status= mysql_fetch_row_start(&sd->row, sd->result);
+    if (status)
+      next_event(31, status, sd);
+    else
+      NEXT_IMMEDIATE(sd, 39);
+    break;
+
+  case 31:
+    status= mysql_fetch_row_cont(&sd->row, sd->result, mysql_status(event));
+    if (status)
+      next_event(31, status, sd);
+    else
+      NEXT_IMMEDIATE(sd, 39);
+    break;
+
+  case 39:
+    if (sd->row)
+    {
+      /* Got a row. */
+      unsigned int i;
+      printf("%d - ", sd->index);
+      for (i= 0; i < mysql_num_fields(sd->result); i++)
+        printf("%s%s", (i ? "\t" : ""), (sd->row[i] ? sd->row[i] : "(null)"));
+      printf ("\n");
+      NEXT_IMMEDIATE(sd, 30);
+    }
+    else
+    {
+      if (mysql_errno(&sd->mysql))
+      {
+        /* An error occured. */
+        printf("%d | Error: %s\n", sd->index, mysql_error(&sd->mysql));
+      }
+      else
+      {
+        /* EOF. */
+        printf("%d | EOF\n", sd->index);
+      }
+      mysql_free_result(sd->result);
+      NEXT_IMMEDIATE(sd, 10);
+    }
+    break;
+
+  case 40:
+    status= mysql_close_start(&sd->mysql);
+    if (status)
+      next_event(41, status, sd);
+    else
+      NEXT_IMMEDIATE(sd, 50);
+    break;
+
+  case 41:
+    status= mysql_close_cont(&sd->mysql, mysql_status(event));
+    if (status)
+      next_event(41, status, sd);
+    else
+      NEXT_IMMEDIATE(sd, 50);
+    break;
+
+  case 50:
+    /* We are done! */
+    num_active_connections--;
+    if (num_active_connections == 0)
+      event_loopbreak();
+    break;
+
+  default:
+    abort();
+  }
+}
+
+
+void
+add_query(const char *q)
+{
+  struct query_entry *e;
+  char *q2;
+  size_t len;
+
+  e= malloc(sizeof(*e));
+  q2= strdup(q);
+  if (!e || !q2)
+    fatal(NULL, "Out of memory");
+
+  /* Remove any trailing newline. */
+  len= strlen(q2);
+  if (q2[len] == '\n')
+    q2[len--]= '\0';
+  if (q2[len] == '\r')
+    q2[len--]= '\0';
+
+  e->next= NULL;
+  e->query= q2;
+  e->index= query_counter++;
+  *tail_ptr= e;
+  tail_ptr= &e->next;
+}
+
+
+static my_bool
+handle_option(int optid, const struct my_option *opt __attribute__((unused)),
+              char *arg)
+{
+  switch (optid)
+  {
+  case '?':
+    printf("Usage: async_queries [OPTIONS] query ...\n");
+    my_print_help(options);
+    my_print_variables(options);
+    exit(0);
+    break;
+
+  case 'p':
+    if (arg)
+      opt_password= arg;
+    else
+      tty_password= 1;
+    break;
+  }
+
+  return 0;
+}
+
+
+int
+main(int argc, char *argv[])
+{
+  struct state_data *sds;
+  unsigned int i;
+  int err;
+  struct event_base *libevent_base;
+
+  err= handle_options(&argc, &argv, options, handle_option);
+  if (err)
+    exit(err);
+  if (tty_password)
+    opt_password= get_tty_password(NullS);
+
+  if (opt_query_file)
+  {
+    FILE *f= fopen(opt_query_file, "r");
+    char buf[65536];
+    if (!f)
+      fatal(NULL, "Cannot open query file");
+    while (!feof(f))
+    {
+      if (!fgets(buf, sizeof(buf), f))
+        break;
+      add_query(buf);
+    }
+    fclose(f);
+  }
+  /* Add extra queries directly on command line. */
+  while (argc > 0)
+  {
+    --argc;
+    add_query(*argv++);
+  }
+
+  sds= malloc(opt_connections * sizeof(*sds));
+  if (!sds)
+    fatal(NULL, "Out of memory");
+
+  libevent_base= event_init();
+
+  err= mysql_library_init(argc, argv, (char **)my_groups);
+  if (err)
+  {
+    fprintf(stderr, "Fatal: mysql_library_init() returns error: %d\n", err);
+    exit(1);
+  }
+
+  num_active_connections= 0;
+  for (i= 0; i < opt_connections; i++)
+  {
+    mysql_init(&sds[i].mysql);
+    mysql_options(&sds[i].mysql, MYSQL_READ_DEFAULT_GROUP, "async_queries");
+
+    /*
+      We put the initial connect call in the first state 0 of the state machine
+      and run that manually, just to have everything in one place.
+    */
+    sds[i].ST= 0;
+    num_active_connections++;
+    state_machine_handler(-1, -1, &sds[i]);
+  }
+
+  event_dispatch();
+
+  free(sds);
+
+  mysql_library_end();
+
+  event_base_free(libevent_base);
+
+  return 0;
+}
diff --git a/tests/check_async_queries.pl b/tests/check_async_queries.pl
new file mode 100644
index 00000000000..b599bc334d3
--- /dev/null
+++ b/tests/check_async_queries.pl
@@ -0,0 +1,73 @@
+#! /usr/bin/perl
+
+# Read the output of async_queries.c. Run the queries again serially, using
+# the normal (not asynchronous) API. Compare the two results for correctness.
+
+use strict;
+use warnings;
+
+use DBI;
+
+my $D= [];
+
+die "Usage: $0 <host> <user> <password> <database>\n"
+    unless @ARGV == 4;
+
+my $dbh= DBI->connect("DBI:mysql:database=$ARGV[3];host=$ARGV[0]",
+                      $ARGV[1], $ARGV[2],
+                      { RaiseError => 1, PrintError => 0 });
+
+while (<STDIN>) {
+  chomp;
+  if (/^([0-9]+) ! (.*);$/) {
+    my ($index, $query)= ($1, $2);
+    $D->[$index]= { QUERY => $query, OUTPUT => [] };
+  } elsif (/^([0-9]+) - (.*)$/) {
+    my ($index, $data)= ($1, $2);
+    push @{$D->[$index]{OUTPUT}}, $data;
+  } elsif (/^([0-9]+) \| Error: (.*)$/) {
+    my ($index, $errmsg)= ($1, $2);
+    my $rows;
+    my $res= eval {
+      my $stm= $dbh->prepare($D->[$index]{QUERY});
+      $stm->execute();
+      $rows= $stm->fetchall_arrayref();
+      1;
+    };
+    if ($res) {
+      die "Query $index succeeded, but should have failed with error.\nquery=$D->[$index]{QUERY}\nerror=$errmsg\n";
+    }
+    my $errmsg2= $@;
+    if ($errmsg2 =~ /^DBD::.*failed: (.*) at .*$/s) {
+      $errmsg2= $1;
+    } else {
+      die "Unexpected DBD error message format: '$errmsg2'\n";
+    }
+    if ($errmsg2 ne $errmsg) {
+      die "Query $index failed with different error message\nquery=$D->[$index]{QUERY}\nerror1=$errmsg\nerror2=$errmsg2\n";
+    }
+    print "OK $index\n";
+    delete $D->[$index];
+  } elsif (/^([0-9]+) \| EOF$/) {
+    my $index= $1;
+    my $rows;
+    my $res= eval {
+      my $stm= $dbh->prepare($D->[$index]{QUERY});
+      $stm->execute();
+      $rows= $stm->fetchall_arrayref();
+      1;
+    };
+    if (!$res) {
+      die "Query $index failed, but should have succeeded.\nquery=$D->[$index]{QUERY}\nerror=$@\n";
+    }
+    my $result_string= join("\n", sort @{$D->[$index]{OUTPUT}});
+    my $result_string2= join("\n", sort(map(join("\t", map((defined($_) ? $_ : "(null)"), @$_)), @$rows)));
+    if ($result_string ne $result_string2) {
+      die "Query $index result difference.\nquery=$D->[$index]{QUERY}\noutput1=\n$$result_string\noutput2=\n$result_string2\n";
+    }
+    delete $D->[$index];
+  } else {
+    die "Unexpected line: '$_'\n";
+  }
+}
+$dbh->disconnect();
diff --git a/tests/mysql_client_test.c b/tests/mysql_client_test.c
index c6efdca60f6..1ca70fd972a 100644
--- a/tests/mysql_client_test.c
+++ b/tests/mysql_client_test.c
@@ -36,6 +36,18 @@
 #include <my_handler.h>
 #include <sql_common.h>
 
+/*
+  If non_blocking_api_enabled is true, we will re-define all the blocking
+  API functions as wrappers that call the corresponding non-blocking API
+  and use poll()/select() to wait for them to complete. This way we can get
+  a good coverage testing of the non-blocking API as well.
+*/
+static my_bool non_blocking_api_enabled= 0;
+#if !defined(EMBEDDED_LIBRARY)
+#define WRAP_NONBLOCK_ENABLED non_blocking_api_enabled
+#include "nonblock-wrappers.h"
+#endif
+
 #define VER "2.1"
 #define MAX_TEST_QUERY_LENGTH 300 /* MAX QUERY BUFFER LENGTH */
 #define MAX_KEY MAX_INDEXES
@@ -18614,6 +18626,10 @@ static struct my_option client_test_long_options[] =
 #endif
   {"vardir", 'v', "Data dir for tests.", (char**) &opt_vardir,
    (char**) &opt_vardir, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  {"non-blocking-api", 'n',
+   "Use the non-blocking client API for communication.",
+   &non_blocking_api_enabled, &non_blocking_api_enabled, 0,
+   GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"getopt-ll-test", 'g', "Option for testing bug in getopt library",
    &opt_getopt_ll_test, &opt_getopt_ll_test, 0,
    GET_LL, REQUIRED_ARG, 0, 0, LONGLONG_MAX, 0, 0, 0},
diff --git a/tests/nonblock-wrappers.h b/tests/nonblock-wrappers.h
new file mode 100644
index 00000000000..a7346ba641c
--- /dev/null
+++ b/tests/nonblock-wrappers.h
@@ -0,0 +1,514 @@
+/* Copyright (c) 2011 Monty Program Ab
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program 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 General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+/*
+  Wrappers that re-implement the normal blocking libmysql API calls in terms
+  of the non-blocking API calls and explicit waiting.
+
+  Used to test the non-blocking calls using mysql_client_test.
+*/
+
+#ifndef __WIN__
+#include <poll.h>
+#else
+#include <WinSock2.h>
+#endif
+
+/*
+  Run the appropriate poll() syscall to wait for the event that libmysql
+  requested. Return which event(s) occured.
+*/
+static int
+wait_for_mysql(MYSQL *mysql, int status)
+{
+#ifdef __WIN__
+  fd_set rs, ws, es;
+  int res;
+  struct timeval tv, *timeout;
+  my_socket s= mysql_get_socket(mysql);
+  FD_ZERO(&rs);
+  FD_ZERO(&ws);
+  FD_ZERO(&es);
+  if (status & MYSQL_WAIT_READ)
+    FD_SET(s, &rs);
+  if (status & MYSQL_WAIT_WRITE)
+    FD_SET(s, &ws);
+  if (status & MYSQL_WAIT_EXCEPT)
+    FD_SET(s, &es);
+  if (status & MYSQL_WAIT_TIMEOUT)
+  {
+    tv.tv_sec= mysql_get_timeout_value(mysql);
+    tv.tv_usec= 0;
+    timeout= &tv;
+  }
+  else
+    timeout= NULL;
+  res= select(1, &rs, &ws, &es, timeout);
+  if (res == 0)
+    return MYSQL_WAIT_TIMEOUT;
+  else if (res == SOCKET_ERROR)
+    return MYSQL_WAIT_TIMEOUT;
+  else
+  {
+    int status= 0;
+    if (FD_ISSET(s, &rs))
+      status|= MYSQL_WAIT_READ;
+    if (FD_ISSET(s, &ws))
+      status|= MYSQL_WAIT_WRITE;
+    if (FD_ISSET(s, &es))
+      status|= MYSQL_WAIT_EXCEPT;
+    return status;
+  }
+#else
+  struct pollfd pfd;
+  int timeout;
+  int res;
+
+  pfd.fd= mysql_get_socket(mysql);
+  pfd.events=
+    (status & MYSQL_WAIT_READ ? POLLIN : 0) |
+    (status & MYSQL_WAIT_WRITE ? POLLOUT : 0) |
+    (status & MYSQL_WAIT_EXCEPT ? POLLPRI : 0);
+  if (status & MYSQL_WAIT_TIMEOUT)
+    timeout= 1000*mysql_get_timeout_value(mysql);
+  else
+    timeout= -1;
+  do {
+    res= poll(&pfd, 1, timeout);
+    /*
+      In a real event framework, we should re-compute the timeout on getting
+      EINTR to account for the time elapsed before the interruption.
+    */
+  } while (res < 0 && errno == EINTR);
+  if (res == 0)
+    return MYSQL_WAIT_TIMEOUT;
+  else if (res < 0)
+    return MYSQL_WAIT_TIMEOUT;
+  else
+  {
+    int status= 0;
+    if (pfd.revents & POLLIN)
+      status|= MYSQL_WAIT_READ;
+    if (pfd.revents & POLLOUT)
+      status|= MYSQL_WAIT_WRITE;
+    if (pfd.revents & POLLPRI)
+      status|= MYSQL_WAIT_EXCEPT;
+    return status;
+  }
+#endif
+}
+
+
+/*
+  If WRAP_NONBLOCK_ENABLED is defined, it is a variable that can be used to
+  enable or disable the use of non-blocking API wrappers. If true the
+  non-blocking API will be used, if false the normal blocking API will be
+  called directly.
+*/
+#ifdef WRAP_NONBLOCK_ENABLED
+#define USE_BLOCKING(name__, invoke_blocking__) \
+  if (!(WRAP_NONBLOCK_ENABLED)) return name__ invoke_blocking__;
+#define USE_BLOCKING_VOID_RETURN(name__, invoke__) \
+  if (!(WRAP_NONBLOCK_ENABLED)) { name__ invoke__; return; }
+#else
+#define USE_BLOCKING(name__, invoke_blocking__)
+#define USE_BLOCKING_VOID_RETURN(name__, invoke__)
+#endif
+
+/*
+  I would preferably have declared the wrappers static.
+  However, if we do so, compilers will warn about definitions not used, and
+  with -Werror this breaks compilation :-(
+*/
+#define MK_WRAPPER(ret_type__, name__, decl__, invoke__, invoke_blocking__, cont_arg__, mysql__) \
+ret_type__ wrap_ ## name__ decl__                                             \
+{                                                                             \
+  ret_type__ res;                                                             \
+  int status;                                                                 \
+  USE_BLOCKING(name__, invoke_blocking__)                                     \
+  status= name__ ## _start invoke__;                                          \
+  while (status)                                                              \
+  {                                                                           \
+    status= wait_for_mysql(mysql__, status);                                  \
+    status= name__ ## _cont(&res, cont_arg__, status);                        \
+  }                                                                           \
+  return res;                                                                 \
+}
+
+#define MK_WRAPPER_VOID_RETURN(name__, decl__, invoke__, cont_arg__, mysql__) \
+void wrap_ ## name__ decl__                                                   \
+{                                                                             \
+  int status;                                                                 \
+  USE_BLOCKING_VOID_RETURN(name__, invoke__)                                  \
+  status= name__ ## _start invoke__;                                          \
+  while (status)                                                              \
+  {                                                                           \
+    status= wait_for_mysql(mysql__, status);                                  \
+    status= name__ ## _cont(cont_arg__, status);                              \
+  }                                                                           \
+}
+
+MK_WRAPPER(
+  MYSQL *,
+  mysql_real_connect,
+  (MYSQL *mysql, const char *host, const char *user, const char *passwd, const char *db, unsigned int port, const char *unix_socket, unsigned long clientflag),
+  (&res, mysql, host, user, passwd, db, port, unix_socket, clientflag),
+  (mysql, host, user, passwd, db, port, unix_socket, clientflag),
+  mysql,
+  mysql)
+
+
+MK_WRAPPER(
+  int,
+  mysql_real_query,
+  (MYSQL *mysql, const char *stmt_str, unsigned long length),
+  (&res, mysql, stmt_str, length),
+  (mysql, stmt_str, length),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  MYSQL_ROW,
+  mysql_fetch_row,
+  (MYSQL_RES *result),
+  (&res, result),
+  (result),
+  result,
+  result->handle)
+
+MK_WRAPPER(
+  int,
+  mysql_set_character_set,
+  (MYSQL *mysql, const char *csname),
+  (&res, mysql, csname),
+  (mysql, csname),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_select_db,
+  (MYSQL *mysql, const char *db),
+  (&res, mysql, db),
+  (mysql, db),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_send_query,
+  (MYSQL *mysql, const char *q, unsigned long length),
+  (&res, mysql, q, length),
+  (mysql, q, length),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  MYSQL_RES *,
+  mysql_store_result,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+MK_WRAPPER_VOID_RETURN(
+  mysql_free_result,
+  (MYSQL_RES *result),
+  (result),
+  result,
+  result->handle)
+
+MK_WRAPPER_VOID_RETURN(
+  mysql_close,
+  (MYSQL *sock),
+  (sock),
+  sock,
+  sock)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_change_user,
+  (MYSQL *mysql, const char *user, const char *passwd, const char *db),
+  (&res, mysql, user, passwd, db),
+  (mysql, user, passwd, db),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_query,
+  (MYSQL *mysql, const char *q),
+  (&res, mysql, q),
+  (mysql, q),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_shutdown,
+  (MYSQL *mysql, enum mysql_enum_shutdown_level shutdown_level),
+  (&res, mysql, shutdown_level),
+  (mysql, shutdown_level),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_dump_debug_info,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_refresh,
+  (MYSQL *mysql, unsigned int refresh_options),
+  (&res, mysql, refresh_options),
+  (mysql, refresh_options),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_kill,
+  (MYSQL *mysql, unsigned long pid),
+  (&res, mysql, pid),
+  (mysql, pid),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_set_server_option,
+  (MYSQL *mysql, enum enum_mysql_set_option option),
+  (&res, mysql, option),
+  (mysql, option),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_ping,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  const char *,
+  mysql_stat,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  MYSQL_RES *,
+  mysql_list_dbs,
+  (MYSQL *mysql, const char *wild),
+  (&res, mysql, wild),
+  (mysql, wild),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  MYSQL_RES *,
+  mysql_list_tables,
+  (MYSQL *mysql, const char *wild),
+  (&res, mysql, wild),
+  (mysql, wild),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  MYSQL_RES *,
+  mysql_list_processes,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  MYSQL_RES *,
+  mysql_list_fields,
+  (MYSQL *mysql, const char *table, const char *wild),
+  (&res, mysql, table, wild),
+  (mysql, table, wild),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_read_query_result,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_stmt_prepare,
+  (MYSQL_STMT *stmt, const char *query, unsigned long length),
+  (&res, stmt, query, length),
+  (stmt, query, length),
+  stmt,
+  stmt->mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_stmt_execute,
+  (MYSQL_STMT *stmt),
+  (&res, stmt),
+  (stmt),
+  stmt,
+  stmt->mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_stmt_fetch,
+  (MYSQL_STMT *stmt),
+  (&res, stmt),
+  (stmt),
+  stmt,
+  stmt->mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_stmt_store_result,
+  (MYSQL_STMT *stmt),
+  (&res, stmt),
+  (stmt),
+  stmt,
+  stmt->mysql)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_stmt_close,
+  (MYSQL_STMT *stmt),
+  (&res, stmt),
+  (stmt),
+  stmt,
+  stmt->mysql)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_stmt_reset,
+  (MYSQL_STMT *stmt),
+  (&res, stmt),
+  (stmt),
+  stmt,
+  stmt->mysql)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_stmt_free_result,
+  (MYSQL_STMT *stmt),
+  (&res, stmt),
+  (stmt),
+  stmt,
+  stmt->mysql)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_stmt_send_long_data,
+  (MYSQL_STMT *stmt, unsigned int param_number, const char *data, unsigned long length),
+  (&res, stmt, param_number, data, length),
+  (stmt, param_number, data, length),
+  stmt,
+  stmt->mysql)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_commit,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_rollback,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  my_bool,
+  mysql_autocommit,
+  (MYSQL *mysql, my_bool auto_mode),
+  (&res, mysql, auto_mode),
+  (mysql, auto_mode),
+  mysql,
+  mysql)
+
+MK_WRAPPER(
+  int,
+  mysql_next_result,
+  (MYSQL *mysql),
+  (&res, mysql),
+  (mysql),
+  mysql,
+  mysql)
+
+#undef USE_BLOCKING
+#undef MK_WRAPPER
+#undef MK_WRAPPER_VOID_RETURN
+
+
+#define mysql_real_connect wrap_mysql_real_connect
+#define mysql_real_query wrap_mysql_real_query
+#define mysql_fetch_row wrap_mysql_fetch_row
+#define mysql_set_character_set wrap_mysql_set_character_set
+#define mysql_select_db wrap_mysql_select_db
+#define mysql_send_query wrap_mysql_send_query
+#define mysql_store_result wrap_mysql_store_result
+#define mysql_free_result wrap_mysql_free_result
+#define mysql_close wrap_mysql_close
+#define mysql_change_user wrap_mysql_change_user
+#define mysql_query wrap_mysql_query
+#define mysql_shutdown wrap_mysql_shutdown
+#define mysql_dump_debug_info wrap_mysql_dump_debug_info
+#define mysql_refresh wrap_mysql_refresh
+#define mysql_kill wrap_mysql_kill
+#define mysql_set_server_option wrap_mysql_set_server_option
+#define mysql_ping wrap_mysql_ping
+#define mysql_stat wrap_mysql_stat
+#define mysql_list_dbs wrap_mysql_list_dbs
+#define mysql_list_tables wrap_mysql_list_tables
+#define mysql_list_processes wrap_mysql_list_processes
+#define mysql_list_fields wrap_mysql_list_fields
+#define mysql_read_query_result wrap_mysql_read_query_result
+#define mysql_stmt_prepare wrap_mysql_stmt_prepare
+#define mysql_stmt_execute wrap_mysql_stmt_execute
+#define mysql_stmt_fetch wrap_mysql_stmt_fetch
+#define mysql_stmt_store_result wrap_mysql_stmt_store_result
+#define mysql_stmt_close wrap_mysql_stmt_close
+#define mysql_stmt_reset wrap_mysql_stmt_reset
+#define mysql_stmt_free_result wrap_mysql_stmt_free_result
+#define mysql_stmt_send_long_data wrap_mysql_stmt_send_long_data
+#define mysql_commit wrap_mysql_commit
+#define mysql_rollback wrap_mysql_rollback
+#define mysql_autocommit wrap_mysql_autocommit
+#define mysql_next_result wrap_mysql_next_result