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.

4 files changed, 246 insertions, 2 deletions

diff --git a/client/CMakeLists.txt b/client/CMakeLists.txt
index ecfd3a0f0ce..53c445c21d8 100755
--- a/client/CMakeLists.txt
+++ b/client/CMakeLists.txt
@@ -68,5 +68,8 @@ MYSQL_ADD_EXECUTABLE(mysqlslap mysqlslap.c DESTINATION bin)
 SET_SOURCE_FILES_PROPERTIES(mysqlslap.c PROPERTIES COMPILE_FLAGS "-DTHREADS")
 TARGET_LINK_LIBRARIES(mysqlslap mysqlclient mysys zlib wsock32 dbug)
 
+MYSQL_ADD_EXECUTABLE(async_example async_example.c)
+TARGET_LINK_LIBRARIES(async_example mysqlclient mysys zlib wsock32 dbug)
+
 
 MYSQL_ADD_EXECUTABLE(echo echo.c COMPONENT Test)
diff --git a/client/Makefile.am b/client/Makefile.am
index f3f964ace7b..f4d9c8faf73 100644
--- a/client/Makefile.am
+++ b/client/Makefile.am
@@ -51,6 +51,8 @@ bin_PROGRAMS =			mysql \
 				mysqltest \
 				mysql_upgrade
 
+noinst_PROGRAMS = 		async_example
+
 mysql_SOURCES =			mysql.cc readline.cc sql_string.cc \
 				completion_hash.cc
 mysql_LDADD =			@readline_link@ @TERMCAP_LIB@ \
@@ -96,6 +98,11 @@ mysqltest_LDADD =		$(CXXLDFLAGS) $(CLIENT_THREAD_LIBS) \
 mysql_upgrade_SOURCES=          mysql_upgrade.c \
                                 $(top_srcdir)/mysys/my_getpagesize.c
 
+async_example_SOURCES=		async_example.c
+async_example_LDADD =		$(CXXLDFLAGS) $(CLIENT_THREAD_LIBS) \
+				@CLIENT_EXTRA_LDFLAGS@ \
+				$(LIBMYSQLCLIENT_LA)
+
 # Fix for mit-threads
 DEFS =			-DMYSQL_CLIENT_NO_THREADS \
 			-DDEFAULT_MYSQL_HOME='"$(prefix)"' \
diff --git a/client/async_example.c b/client/async_example.c
new file mode 100644
index 00000000000..de9d455171c
--- /dev/null
+++ b/client/async_example.c
@@ -0,0 +1,207 @@
+/*
+  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/>.
+*/
+
+
+#ifndef __WIN__
+#include <poll.h>
+#else
+#include <WinSock2.h>
+#endif
+
+#include <stdlib.h>
+#include <stdio.h>
+#include <mysql.h>
+
+#define SL(s) (s), sizeof(s)
+
+static const char *my_groups[]= { "client", NULL };
+
+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)
+  {
+    /*
+      In a real event framework, we should handle errors and re-try the select.
+    */
+    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;
+  res= poll(&pfd, 1, timeout);
+  if (res == 0)
+    return MYSQL_WAIT_TIMEOUT;
+  else if (res < 0)
+  {
+    /*
+      In a real event framework, we should handle EINTR and re-try the poll.
+    */
+    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
+}
+
+static void
+fatal(MYSQL *mysql, const char *msg)
+{
+  fprintf(stderr, "%s: %s\n", msg, mysql_error(mysql));
+  exit(1);
+}
+
+static void
+doit(const char *host, const char *user, const char *password)
+{
+  int err;
+  MYSQL mysql, *ret;
+  MYSQL_RES *res;
+  MYSQL_ROW row;
+  int status;
+
+  mysql_init(&mysql);
+  mysql_options(&mysql, MYSQL_READ_DEFAULT_GROUP, "myapp");
+
+  /* Returns 0 when done, else flag for what to wait for when need to block. */
+  status= mysql_real_connect_start(&ret, &mysql, host, user, password, NULL,
+                                   0, NULL, 0);
+  while (status)
+  {
+    status= wait_for_mysql(&mysql, status);
+    status= mysql_real_connect_cont(&ret, &mysql, status);
+  }
+
+  if (!ret)
+    fatal(&mysql, "Failed to mysql_real_connect()");
+
+  status= mysql_real_query_start(&err, &mysql, SL("SHOW STATUS"));
+  while (status)
+  {
+    status= wait_for_mysql(&mysql, status);
+    status= mysql_real_query_cont(&err, &mysql, status);
+  }
+  if (err)
+    fatal(&mysql, "mysql_real_query() returns error");
+
+  /* This method cannot block. */
+  res= mysql_use_result(&mysql);
+  if (!res)
+    fatal(&mysql, "mysql_use_result() returns error");
+
+  for (;;)
+  {
+    status= mysql_fetch_row_start(&row, res);
+    while (status)
+    {
+      status= wait_for_mysql(&mysql, status);
+      status= mysql_fetch_row_cont(&row, res, status);
+    }
+    if (!row)
+      break;
+    printf("%s: %s\n", row[0], row[1]);
+  }
+  if (mysql_errno(&mysql))
+    fatal(&mysql, "Got error while retrieving rows");
+  mysql_free_result(res);
+
+  /* I suppose this must be non-blocking too. */
+  mysql_close(&mysql);
+}
+
+int
+main(int argc, char *argv[])
+{
+  int err;
+
+  if (argc != 4)
+  {
+    fprintf(stderr, "Usage: %s <host> <user> <password>\n", argv[0]);
+    exit(1);
+  }
+
+  err= mysql_library_init(argc, argv, (char **)my_groups);
+  if (err)
+  {
+    fprintf(stderr, "Fatal: mysql_library_init() returns error: %d\n", err);
+    exit(1);
+  }
+
+  doit(argv[1], argv[2], argv[3]);
+
+  mysql_library_end();
+
+  return 0;
+}
diff --git a/client/mysqltest.cc b/client/mysqltest.cc
index 772dd69c089..7aeaa48519d 100644
--- a/client/mysqltest.cc
+++ b/client/mysqltest.cc
@@ -60,6 +60,12 @@
 #define SIGNAL_FMT "signal %d"
 #endif
 
+static my_bool non_blocking_api_enabled= 0;
+#if !defined(EMBEDDED_LIBRARY)
+#define WRAP_NONBLOCK_ENABLED non_blocking_api_enabled
+#include "../tests/nonblock-wrappers.h"
+#endif
+
 /* Use cygwin for --exec and --system before 5.0 */
 #if MYSQL_VERSION_ID < 50000
 #define USE_CYGWIN
@@ -84,7 +90,7 @@ enum {
   OPT_PS_PROTOCOL, OPT_SP_PROTOCOL, OPT_CURSOR_PROTOCOL, OPT_VIEW_PROTOCOL,
   OPT_MAX_CONNECT_RETRIES, OPT_MAX_CONNECTIONS,
   OPT_MARK_PROGRESS, OPT_LOG_DIR, OPT_TAIL_LINES,
-  OPT_GLOBAL_SUBST, OPT_MY_CONNECT_TIMEOUT
+  OPT_GLOBAL_SUBST, OPT_MY_CONNECT_TIMEOUT, OPT_NON_BLOCKING_API
 };
 
 static int record= 0, opt_sleep= -1;
@@ -305,6 +311,7 @@ enum enum_commands {
   Q_LOWERCASE,
   Q_START_TIMER, Q_END_TIMER,
   Q_CHARACTER_SET, Q_DISABLE_PS_PROTOCOL, Q_ENABLE_PS_PROTOCOL,
+  Q_ENABLE_NON_BLOCKING_API, Q_DISABLE_NON_BLOCKING_API,
   Q_DISABLE_RECONNECT, Q_ENABLE_RECONNECT,
   Q_IF,
   Q_DISABLE_PARSING, Q_ENABLE_PARSING,
@@ -386,6 +393,8 @@ const char *command_names[]=
   "character_set",
   "disable_ps_protocol",
   "enable_ps_protocol",
+  "enable_non_blocking_api",
+  "disable_non_blocking_api",
   "disable_reconnect",
   "enable_reconnect",
   "if",
@@ -5235,7 +5244,10 @@ void do_connect(struct st_command *command)
   int con_port= opt_port;
   char *con_options;
   my_bool con_ssl= 0, con_compress= 0;
-  my_bool con_pipe= 0, con_shm= 0;
+  my_bool con_pipe= 0;
+#ifdef HAVE_SMEM
+  my_bool con_shm= 0;
+#endif
   struct st_connection* con_slot;
 
   static DYNAMIC_STRING ds_connection_name;
@@ -5324,7 +5336,11 @@ void do_connect(struct st_command *command)
     else if (length == 4 && !strncmp(con_options, "PIPE", 4))
       con_pipe= 1;
     else if (length == 3 && !strncmp(con_options, "SHM", 3))
+#ifdef HAVE_SMEM
       con_shm= 1;
+#else
+    { }
+#endif
     else
       die("Illegal option to connect: %.*s", 
           (int) (end - con_options), con_options);
@@ -6146,6 +6162,10 @@ static struct my_option my_long_options[] =
    "Use prepared-statement protocol for communication.",
    &ps_protocol, &ps_protocol, 0,
    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"non-blocking-api", OPT_NON_BLOCKING_API,
+   "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},
   {"quiet", 's', "Suppress all normal output.", &silent,
    &silent, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"record", 'r', "Record output of test_file into result file.",
@@ -8130,6 +8150,7 @@ int main(int argc, char **argv)
   next_con= connections + 1;
   
   var_set_int("$PS_PROTOCOL", ps_protocol);
+  var_set_int("$NON_BLOCKING_API", non_blocking_api_enabled);
   var_set_int("$SP_PROTOCOL", sp_protocol);
   var_set_int("$VIEW_PROTOCOL", view_protocol);
   var_set_int("$CURSOR_PROTOCOL", cursor_protocol);
@@ -8538,6 +8559,12 @@ int main(int argc, char **argv)
       case Q_ENABLE_PS_PROTOCOL:
         ps_protocol_enabled= ps_protocol;
         break;
+      case Q_DISABLE_NON_BLOCKING_API:
+        non_blocking_api_enabled= 0;
+        break;
+      case Q_ENABLE_NON_BLOCKING_API:
+        non_blocking_api_enabled= 1;
+        break;
       case Q_DISABLE_RECONNECT:
         set_reconnect(cur_con->mysql, 0);
         break;