Merge MWL#192: Non-blocking client library, into MariaDB 5.5.

3 files changed, 244 insertions, 1 deletions

diff --git a/client/CMakeLists.txt b/client/CMakeLists.txt
index de503e8927e..d29cc902aa3 100644
--- a/client/CMakeLists.txt
+++ b/client/CMakeLists.txt
@@ -73,6 +73,10 @@ IF(WIN32)
   MYSQL_ADD_EXECUTABLE(echo echo.c COMPONENT Junk)
 ENDIF(WIN32)
 
+# async_example is just a code example, do not install it.
+ADD_EXECUTABLE(async_example async_example.c)
+TARGET_LINK_LIBRARIES(async_example mysqlclient)
+
 SET_TARGET_PROPERTIES (mysqlcheck mysqldump mysqlimport mysql_upgrade mysqlshow mysqlslap mysql_plugin 
 PROPERTIES HAS_CXX TRUE)
 
diff --git a/client/async_example.c b/client/async_example.c
new file mode 100644
index 00000000000..ccb60950904
--- /dev/null
+++ b/client/async_example.c
@@ -0,0 +1,216 @@
+/*
+  Copyright 2011 Kristian Nielsen and Monty Program Ab.
+
+  This file 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.
+
+  This library 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 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_OPT_NONBLOCK, 0);
+  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);
+
+  /*
+    mysql_close() sends a COM_QUIT packet, and so in principle could block
+    waiting for the socket to accept the data.
+    In practise, for many applications it will probably be fine to use the
+    blocking mysql_close().
+   */
+  status= mysql_close_start(&mysql);
+  while (status)
+  {
+    status= wait_for_mysql(&mysql, status);
+    status= mysql_close_cont(&mysql, status);
+  }
+}
+
+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 0e79f57e6a3..4cecbe1f819 100644
--- a/client/mysqltest.cc
+++ b/client/mysqltest.cc
@@ -61,6 +61,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
@@ -91,7 +97,7 @@ enum {
   OPT_CURSOR_PROTOCOL, OPT_VIEW_PROTOCOL, OPT_MAX_CONNECT_RETRIES,
   OPT_MAX_CONNECTIONS, OPT_MARK_PROGRESS, OPT_LOG_DIR,
   OPT_TAIL_LINES, OPT_RESULT_FORMAT_VERSION,
-  OPT_MY_CONNECT_TIMEOUT
+  OPT_MY_CONNECT_TIMEOUT, OPT_NON_BLOCKING_API
 };
 
 static int record= 0, opt_sleep= -1;
@@ -353,6 +359,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,
@@ -434,6 +441,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",
@@ -5682,6 +5691,7 @@ void do_connect(struct st_command *command)
     mysql_options(con_slot->mysql, MYSQL_OPT_CONNECT_TIMEOUT,
                   (void *) &opt_connect_timeout);
 
+  mysql_options(con_slot->mysql, MYSQL_OPT_NONBLOCK, 0);
   if (opt_compress || con_compress)
     mysql_options(con_slot->mysql, MYSQL_OPT_COMPRESS, NullS);
   mysql_options(con_slot->mysql, MYSQL_OPT_LOCAL_INFILE, 0);
@@ -6631,6 +6641,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.",
@@ -7953,6 +7967,7 @@ int util_query(MYSQL* org_mysql, const char* query){
 
     /* enable local infile, in non-binary builds often disabled by default */
     mysql_options(mysql, MYSQL_OPT_LOCAL_INFILE, 0);
+    mysql_options(mysql, MYSQL_OPT_NONBLOCK, 0);
     safe_connect(mysql, "util", org_mysql->host, org_mysql->user,
                  org_mysql->passwd, org_mysql->db, org_mysql->port,
                  org_mysql->unix_socket);
@@ -8622,6 +8637,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);
@@ -8705,6 +8721,7 @@ int main(int argc, char **argv)
 
   if (!(con->name = my_strdup("default", MYF(MY_WME))))
     die("Out of memory");
+  mysql_options(con->mysql, MYSQL_OPT_NONBLOCK, 0);
 
   safe_connect(con->mysql, con->name, opt_host, opt_user, opt_pass,
                opt_db, opt_port, unix_sock);
@@ -9061,6 +9078,12 @@ int main(int argc, char **argv)
       case Q_ENABLE_PS_PROTOCOL:
         set_property(command, P_PS, 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;