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

.bzrignore

@@ -128,6 +128,7 @@ client/thimble
 client/thread_test
 client/tmp.diff
 client/transaction.h
+client/async_example
 client_debug/*
 client_release/*
 client_test
@@ -1028,6 +1029,7 @@ tests/bug25714
 tests/client_test
 tests/connect_test
 tests/mysql_client_test
+tests/async_queries
 thr_insert_test/*
 thr_test/*
 thread_test

CMakeLists.txt

@@ -349,7 +349,7 @@ ENDIF()
 # RPM installs documentation directly from the source tree
 #
 IF(NOT INSTALL_LAYOUT MATCHES "RPM")
-  INSTALL(FILES COPYING LICENSE.mysql 
+  INSTALL(FILES COPYING COPYING.LESSER LICENSE.mysql 
   DESTINATION ${INSTALL_DOCREADMEDIR} 
   COMPONENT Readme
   OPTIONAL

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)
 

client/async_example.c

+/*
+  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;
+}

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;

dbug/dbug.c

@@ -349,6 +349,27 @@ static CODE_STATE *code_state(void)
   return cs;
 }
 
+void
+dbug_swap_code_state(void **code_state_store)
+{
+  CODE_STATE *cs, **cs_ptr;
+
+  if (!(cs_ptr= (CODE_STATE**) my_thread_var_dbug()))
+    return;
+  cs= *cs_ptr;
+  *cs_ptr= *code_state_store;
+  *code_state_store= cs;
+}
+
+void dbug_free_code_state(void **code_state_store)
+{
+  if (*code_state_store)
+  {
+    free(*code_state_store);
+    *code_state_store= NULL;
+  }
+}
+
 /*
  *      Translate some calls among different systems.
  */

include/my_context.h

+/*
+  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/>.
+*/
+
+/*
+  Simple API for spawning a co-routine, to be used for async libmysqlclient.
+
+  Idea is that by implementing this interface using whatever facilities are
+  available for given platform, we can use the same code for the generic
+  libmysqlclient-async code.
+
+  (This particular implementation uses Posix ucontext swapcontext().)
+*/
+
+#ifdef __WIN__
+#define MY_CONTEXT_USE_WIN32_FIBERS 1
+#elif defined(__GNUC__) && __GNUC__ >= 3 && defined(__x86_64__)
+#define MY_CONTEXT_USE_X86_64_GCC_ASM
+#elif defined(__GNUC__) && __GNUC__ >= 3 && defined(__i386__)
+#define MY_CONTEXT_USE_I386_GCC_ASM
+#else
+#define MY_CONTEXT_USE_UCONTEXT
+#endif
+
+#ifdef MY_CONTEXT_USE_WIN32_FIBERS
+struct my_context {
+  void (*user_func)(void *);
+  void *user_arg;
+  void *app_fiber;
+  void *lib_fiber;
+  int return_value;
+#ifndef DBUG_OFF
+  void *dbug_state;
+#endif
+};
+#endif
+
+
+#ifdef MY_CONTEXT_USE_UCONTEXT
+#include <ucontext.h>
+
+struct my_context {
+  void (*user_func)(void *);
+  void *user_data;
+  void *stack;
+  size_t stack_size;
+  ucontext_t base_context;
+  ucontext_t spawned_context;
+  int active;
+#ifdef HAVE_VALGRIND_VALGRIND_H
+  unsigned int valgrind_stack_id;
+#endif
+#ifndef DBUG_OFF
+  void *dbug_state;
+#endif
+};
+#endif
+
+
+#ifdef MY_CONTEXT_USE_X86_64_GCC_ASM
+#include <stdint.h>
+
+struct my_context {
+  uint64_t save[9];
+  void *stack_top;
+  void *stack_bot;
+#ifdef HAVE_VALGRIND_VALGRIND_H
+  unsigned int valgrind_stack_id;
+#endif
+#ifndef DBUG_OFF
+  void *dbug_state;
+#endif
+};
+#endif
+
+
+#ifdef MY_CONTEXT_USE_I386_GCC_ASM
+#include <stdint.h>
+
+struct my_context {
+  uint64_t save[7];
+  void *stack_top;
+  void *stack_bot;
+#ifdef HAVE_VALGRIND_VALGRIND_H
+  unsigned int valgrind_stack_id;
+#endif
+#ifndef DBUG_OFF
+  void *dbug_state;
+#endif
+};
+#endif
+
+
+/*
+  Initialize an asynchroneous context object.
+  Returns 0 on success, non-zero on failure.
+*/
+extern int my_context_init(struct my_context *c, size_t stack_size);
+
+/* Free an asynchroneous context object, deallocating any resources used. */
+extern void my_context_destroy(struct my_context *c);
+
+/*
+  Spawn an asynchroneous context. The context will run the supplied user
+  function, passing the supplied user data pointer.
+
+  The context must have been initialised with my_context_init() prior to
+  this call.
+
+  The user function may call my_context_yield(), which will cause this
+  function to return 1. Then later my_context_continue() may be called, which
+  will resume the asynchroneous context by returning from the previous
+  my_context_yield() call.
+
+  When the user function returns, this function returns 0.
+
+  In case of error, -1 is returned.
+*/
+extern int my_context_spawn(struct my_context *c, void (*f)(void *), void *d);
+
+/*
+  Suspend an asynchroneous context started with my_context_spawn.
+
+  When my_context_yield() is called, execution immediately returns from the
+  last my_context_spawn() or my_context_continue() call. Then when later
+  my_context_continue() is called, execution resumes by returning from this
+  my_context_yield() call.
+
+  Returns 0 if ok, -1 in case of error.
+*/
+extern int my_context_yield(struct my_context *c);
+
+/*
+  Resume an asynchroneous context. The context was spawned by
+  my_context_spawn(), and later suspended inside my_context_yield().
+
+  The asynchroneous context may be repeatedly suspended with
+  my_context_yield() and resumed with my_context_continue().
+
+  Each time it is suspended, this function returns 1. When the originally
+  spawned user function returns, this function returns 0.
+
+  In case of error, -1 is returned.
+*/
+extern int my_context_continue(struct my_context *c);
+
+
+struct mysql_async_context {
+  /*
+    This is set to the value that should be returned from foo_start() or
+    foo_cont() when a call is suspended.
+  */
+  unsigned int events_to_wait_for;
+  /*
+    It is also set to the event(s) that triggered when a suspended call is
+    resumed, eg. whether we woke up due to connection completed or timeout
+    in mysql_real_connect_cont().
+  */
+  unsigned int events_occured;
+  /*
+    This is set to the result of the whole asynchronous operation when it
+    completes. It uses a union, as different calls have different return
+    types.
+  */
+  union {
+    void *r_ptr;
+    const void *r_const_ptr;
+    int r_int;
+    my_bool r_my_bool;
+  } ret_result;
+  /*
+    The timeout value, for suspended calls that need to wake up on a timeout
+    (eg. mysql_real_connect_start().
+  */
+  unsigned int timeout_value;
+  /*
+    This flag is set when we are executing inside some asynchronous call
+    foo_start() or foo_cont(). It is used to decide whether to use the
+    synchronous or asynchronous version of calls that may block such as
+    recv().
+
+    Note that this flag is not set when a call is suspended, eg. after
+    returning from foo_start() and before re-entering foo_cont().
+  */
+  my_bool active;
+  /*
+    This flag is set when an asynchronous operation is in progress, but
+    suspended. Ie. it is set when foo_start() or foo_cont() returns because
+    the operation needs to block, suspending the operation.
+
+    It is used to give an error (rather than crash) if the application
+    attempts to call some foo_cont() method when no suspended operation foo is
+    in progress.
+  */
+  my_bool suspended;
+  /*
+    If non-NULL, this is a pointer to a callback hook that will be invoked with
+    the user data argument just before the context is suspended, and just after
+    it is resumed.
+  */
+  void (*suspend_resume_hook)(my_bool suspend, void *user_data);
+  void *suspend_resume_hook_user_data;
+  /*
+    This is used to save the execution contexts so that we can suspend an
+    operation and switch back to the application context, to resume the
+    suspended context later when the application re-invokes us with
+    foo_cont().
+  */
+  struct my_context async_context;
+};

include/my_dbug.h

@@ -59,6 +59,8 @@ extern  void _db_lock_file_(void);
 extern  void _db_unlock_file_(void);
 extern  FILE *_db_fp_(void);
 extern  void _db_flush_();
+extern void dbug_swap_code_state(void **code_state_store);
+extern void dbug_free_code_state(void **code_state_store);
 extern  const char* _db_get_func_(void);
 
 #define DBUG_ENTER(a) struct _db_stack_frame_ _db_stack_frame_; \
@@ -93,6 +95,8 @@ extern  const char* _db_get_func_(void);
 #define DEBUGGER_OFF                    do { _dbug_on_= 0; } while(0)
 #define DEBUGGER_ON                     do { _dbug_on_= 1; } while(0)
 #define IF_DBUG(A,B)                    A
+#define DBUG_SWAP_CODE_STATE(arg) dbug_swap_code_state(arg)
+#define DBUG_FREE_CODE_STATE(arg) dbug_free_code_state(arg)
 
 #ifndef __WIN__
 #define DBUG_ABORT()                    (_db_flush_(), abort())
@@ -151,6 +155,8 @@ extern void _db_suicide_();
 #define DEBUGGER_OFF                    do { } while(0)
 #define DEBUGGER_ON                     do { } while(0)
 #define IF_DBUG(A,B)                    B
+#define DBUG_SWAP_CODE_STATE(arg)       do { } while(0)
+#define DBUG_FREE_CODE_STATE(arg)       do { } while(0)
 #define DBUG_ABORT()                    do { } while(0)
 #define DBUG_CRASH_ENTER(func)
 #define DBUG_CRASH_RETURN(val)          do { return(val); } while(0)

include/mysql_async.h

+/* Copyright (C) 2012 MariaDB Services and Kristian Nielsen
+
+   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 */
+
+/* Common definitions for MariaDB non-blocking client library. */
+
+#ifndef MYSQL_ASYNC_H
+#define MYSQL_ASYNC_H
+
+extern int my_connect_async(struct mysql_async_context *b, my_socket fd,
+                            const struct sockaddr *name, uint namelen,
+                            uint timeout);
+extern ssize_t my_recv_async(struct mysql_async_context *b, int fd,
+                             unsigned char *buf, size_t size, uint timeout);
+extern ssize_t my_send_async(struct mysql_async_context *b, int fd,
+                             const unsigned char *buf, size_t size,
+                             uint timeout);
+extern my_bool my_poll_read_async(struct mysql_async_context *b,
+                                  uint timeout);
+#ifdef HAVE_OPENSSL
+extern int my_ssl_read_async(struct mysql_async_context *b, SSL *ssl,
+                             void *buf, int size);
+extern int my_ssl_write_async(struct mysql_async_context *b, SSL *ssl,
+                              const void *buf, int size);
+#endif
+
+#endif  /* MYSQL_ASYNC_H */

include/mysql_com.h

@@ -187,6 +187,15 @@ enum enum_server_command
 #define CLIENT_PROGRESS  (1UL << 29)   /* Client support progress indicator */
 
 #define CLIENT_SSL_VERIFY_SERVER_CERT (1UL << 30)
+/*
+  It used to be that if mysql_real_connect() failed, it would delete any
+  options set by the client, unless the CLIENT_REMEMBER_OPTIONS flag was
+  given.
+  That behaviour does not appear very useful, and it seems unlikely that
+  any applications would actually depend on this. So from MariaDB 5.5 we
+  always preserve any options set in case of failed connect, and this
+  option is effectively always set.
+*/
 #define CLIENT_REMEMBER_OPTIONS (1UL << 31)
 
 #ifdef HAVE_COMPRESS

include/sql_common.h

@@ -26,6 +26,9 @@ extern const char	*unknown_sqlstate;
 extern const char	*cant_connect_sqlstate;
 extern const char	*not_error_sqlstate;
 
+
+struct mysql_async_context;
+
 struct st_mysql_options_extention {
   char *plugin_dir;
   char *default_auth;
@@ -35,6 +38,7 @@ struct st_mysql_options_extention {
                           double progress,
                           const char *proc_info,
                           uint proc_info_length);
+  struct mysql_async_context *async_context;
 };
 
 typedef struct st_mysql_methods
@@ -108,6 +112,10 @@ void mysql_client_plugin_deinit();
 struct st_mysql_client_plugin;
 extern struct st_mysql_client_plugin *mysql_client_builtins[];
 
+/* Non-blocking client API. */
+void my_context_install_suspend_resume_hook(struct mysql_async_context *b,
+                                            void (*)(my_bool, void *), void *);
+
 #ifdef	__cplusplus
 }
 #endif

include/violite.h

@@ -202,6 +202,8 @@ struct st_vio
   char                  *read_pos;      /* start of unfetched data in the
                                            read buffer */
   char                  *read_end;      /* end of unfetched data */
+  struct mysql_async_context *async_context; /* For non-blocking API */
+  uint read_timeout, write_timeout;
   /* function pointers. They are similar for socket/SSL/whatever */
   void    (*viodelete)(Vio*);
   int     (*vioerrno)(Vio*);

libmysql/CMakeLists.txt

@@ -139,6 +139,7 @@ SET(CLIENT_SOURCES
   libmysql.c
   errmsg.c
   ../sql-common/client.c 
+  ../sql-common/mysql_async.c
   ../sql-common/my_time.c 
   ../sql-common/client_plugin.c 
   ../sql/net_serv.cc

libmysqld/CMakeLists.txt

@@ -41,7 +41,7 @@ SET(SQL_EMBEDDED_SOURCES emb_qcache.cc libmysqld.c lib_sql.cc
            ../libmysql/libmysql.c ../libmysql/errmsg.c ../client/get_password.c
            ../sql-common/client.c ../sql-common/my_time.c 
            ../sql-common/my_user.c ../sql-common/pack.c
-           ../sql-common/client_plugin.c
+           ../sql-common/client_plugin.c ../sql-common/mysql_async.c
            ../sql/password.c ../sql/discover.cc ../sql/derror.cc 
            ../sql/field.cc ../sql/field_conv.cc
            ../sql/filesort.cc ../sql/gstream.cc ../sql/slave.cc

mysql-test/mysql-test-run.pl

@@ -212,6 +212,7 @@ my $opt_ps_protocol;
 my $opt_sp_protocol;
 my $opt_cursor_protocol;
 my $opt_view_protocol;
+my $opt_non_blocking_api;
 
 our $opt_debug;
 my $debug_d= "d,*";
@@ -1123,6 +1124,7 @@ sub command_line_setup {
              'sp-protocol'              => \$opt_sp_protocol,
              'view-protocol'            => \$opt_view_protocol,
              'cursor-protocol'          => \$opt_cursor_protocol,
+             'non-blocking-api'         => \$opt_non_blocking_api,
              'ssl|with-openssl'         => \$opt_ssl,
              'skip-ssl'                 => \$opt_skip_ssl,
              'compress'                 => \$opt_compress,
@@ -5843,6 +5845,11 @@ sub start_mysqltest ($) {
     mtr_add_arg($args, "--cursor-protocol");
   }
 
+  if ( $opt_non_blocking_api )
+  {
+    mtr_add_arg($args, "--non-blocking-api");
+  }
+
   if ( $opt_strace_client )
   {
     $exe=  $opt_strace_client || "strace";
@@ -6303,6 +6310,7 @@ Options to control what engine/variation to run
                         (implies --ps-protocol)
   view-protocol         Create a view to execute all non updating queries
   sp-protocol           Create a stored procedure to execute all queries
+  non-blocking-api      Use the non-blocking client API
   compress              Use the compressed protocol between client and server
   ssl                   Use ssl protocol between client and server
   skip-ssl              Dont start server with support for ssl connections

mysql-test/r/mysql_client_test_nonblock.result

+SET @old_general_log= @@global.general_log;
+SET @old_slow_query_log= @@global.slow_query_log;
+ok
+SET @@global.general_log= @old_general_log;
+SET @@global.slow_query_log= @old_slow_query_log;

mysql-test/r/non_blocking_api.result

+drop table if exists t1;
+CREATE TABLE t1 (a INT PRIMARY KEY);
+INSERT INTO t1 VALUES (1);
+SELECT * FROM t1;
+a
+1
+SELECT * FROM t1;
+a
+1
+DROP TABLE t1;

mysql-test/t/mysql_client_test_nonblock-master.opt

+--log=$MYSQLTEST_VARDIR/log/master.log --log-output=FILE,TABLE

mysql-test/t/mysql_client_test_nonblock.test

+# This runs the mysql_client_test using the non-blocking API.
+
+# The non-blocking API is not supported in the embedded server.
+-- source include/not_embedded.inc
+
+SET @old_general_log= @@global.general_log;
+SET @old_slow_query_log= @@global.slow_query_log;
+
+# We run with different binaries for normal and --embedded-server
+#
+# If this test fails with "command "$MYSQL_CLIENT_TEST" failed",
+# you should either run mysql_client_test separartely against a running
+# server or run mysql-test-run --debug mysql_client_test and check
+# var/log/mysql_client_test.trace
+
+--exec echo "$MYSQL_CLIENT_TEST --non-blocking-api" > $MYSQLTEST_VARDIR/log/mysql_client_test.out.log 2>&1
+--exec $MYSQL_CLIENT_TEST --non-blocking-api --getopt-ll-test=25600M >> $MYSQLTEST_VARDIR/log/mysql_client_test.out.log 2>&1
+
+# End of 4.1 tests
+echo ok;
+
+SET @@global.general_log= @old_general_log;
+SET @@global.slow_query_log= @old_slow_query_log;

mysql-test/t/named_pipe.test

@@ -2,6 +2,8 @@
 # in order  to optimize things we skip this test on all
 # other platforms
 --source include/windows.inc
+# Named pipe does not support the non-blocking API.
+--disable_non_blocking_api
 
 # thread pool causes different results
 -- source include/not_threadpool.inc

mysql-test/t/non_blocking_api.test

+# Test mixing the use of blocking and non-blocking API in a single connection.
+
+--disable_warnings
+drop table if exists t1;
+--enable_warnings
+
+--enable_non_blocking_api
+connect (con_nonblock,localhost,root,,test);
+--disable_non_blocking_api
+connect (con_normal,localhost,root,,test);
+
+connection con_nonblock;
+CREATE TABLE t1 (a INT PRIMARY KEY);
+--enable_non_blocking_api
+INSERT INTO t1 VALUES (1);
+--disable_non_blocking_api
+SELECT * FROM t1;
+--enable_non_blocking_api
+SELECT * FROM t1;
+
+connection con_normal;
+DROP TABLE t1;

mysys/CMakeLists.txt

@@ -37,7 +37,7 @@ SET(MYSYS_SOURCES  array.c charset-def.c charset.c checksum.c default.c
                 safemalloc.c my_new.cc
 				my_atomic.c my_getncpus.c my_safehash.c my_chmod.c my_rnd.c
                                 my_uuid.c wqueue.c waiting_threads.c ma_dyncol.c
-				my_rdtsc.c)
+				my_rdtsc.c my_context.c)
 
 IF (WIN32)
  SET (MYSYS_SOURCES ${MYSYS_SOURCES} my_winthread.c my_wincond.c my_winerr.c my_winfile.c my_windac.c my_conio.c)

sql/CMakeLists.txt

@@ -82,7 +82,7 @@ SET (SQL_SOURCE
                create_options.cc multi_range_read.cc
                opt_index_cond_pushdown.cc opt_subselect.cc
                opt_table_elimination.cc sql_expression_cache.cc
-               gcalc_slicescan.cc gcalc_tools.cc
+               gcalc_slicescan.cc gcalc_tools.cc ../sql-common/mysql_async.c
 
                ${GEN_SOURCES}
                ${MYSYS_LIBWRAP_SOURCE})

tests/CMakeLists.txt

@@ -28,3 +28,10 @@ IF(WITH_UNIT_TESTS)
 ENDIF()
 
 INSTALL(TARGETS mysql_client_test DESTINATION ${INSTALL_BINDIR} COMPONENT Test)
+
+CHECK_INCLUDE_FILE(event.h HAVE_EVENT_H)
+FIND_LIBRARY(EVENT_LIBRARY event)
+IF(HAVE_EVENT_H AND EVENT_LIBRARY)
+  ADD_EXECUTABLE(async_queries async_queries.c)
+  TARGET_LINK_LIBRARIES(async_queries mysqlclient ${EVENT_LIBRARY})
+ENDIF()

tests/check_async_queries.pl

+#! /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();

tests/mysql_client_test.c

@@ -38,6 +38,18 @@
 #include <sql_common.h>
 #include <mysql/client_plugin.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
@@ -251,6 +263,8 @@ static MYSQL *mysql_client_init(MYSQL* con)
   if (res && shared_memory_base_name)
     mysql_options(res, MYSQL_SHARED_MEMORY_BASE_NAME, shared_memory_base_name);
 #endif
+  if (res && non_blocking_api_enabled)
+    mysql_options(res, MYSQL_OPT_NONBLOCK, 0);
   if (opt_plugin_dir && *opt_plugin_dir)
     mysql_options(res, MYSQL_PLUGIN_DIR, opt_plugin_dir);
 
@@ -19937,6 +19951,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},

vio/viossl.c

@@ -21,6 +21,8 @@
 */
 
 #include "vio_priv.h"
+#include "my_context.h"
+#include <mysql_async.h>
 
 #ifdef HAVE_OPENSSL
 
@@ -66,6 +68,9 @@ size_t vio_ssl_read(Vio *vio, uchar* buf, size_t size)
   DBUG_PRINT("enter", ("sd: %d  buf: 0x%lx  size: %u  ssl: 0x%lx",
 		       vio->sd, (long) buf, (uint) size, (long) vio->ssl_arg));
 
+  if (vio->async_context && vio->async_context->active)
+    r= my_ssl_read_async(vio->async_context, (SSL *)vio->ssl_arg, buf, size);
+  else
     r= SSL_read((SSL*) vio->ssl_arg, buf, size);
 #ifndef DBUG_OFF
   if (r == (size_t) -1)
@@ -83,6 +88,9 @@ size_t vio_ssl_write(Vio *vio, const uchar* buf, size_t size)
   DBUG_PRINT("enter", ("sd: %d  buf: 0x%lx  size: %u", vio->sd,
                        (long) buf, (uint) size));
 
+  if (vio->async_context && vio->async_context->active)
+    r= my_ssl_write_async(vio->async_context, (SSL *)vio->ssl_arg, buf, size);
+  else
     r= SSL_write((SSL*) vio->ssl_arg, buf, size);
 #ifndef DBUG_OFF
   if (r == (size_t) -1)