1 files changed, 161 insertions, 113 deletions

diff --git a/sql/sql_connect.cc b/sql/sql_connect.cc
index f6bfe975d51..b2e37758d5b 100644
--- a/sql/sql_connect.cc
+++ b/sql/sql_connect.cc
@@ -1,4 +1,4 @@
-/* Copyright (C) 2007 MySQL AB
+/* Copyright (c) 2007, 2011, Oracle and/or its affiliates. All rights reserved.
 
    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
@@ -11,25 +11,36 @@
 
    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 */
-
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA */
 
 /*
   Functions to autenticate and handle reqests for a connection
 */
 
-#include "mysql_priv.h"
+#include "my_global.h"
+#include "sql_priv.h"
+#ifndef __WIN__
+#include <netdb.h>        // getservbyname, servent
+#endif
+#include "sql_audit.h"
+#include "sql_connect.h"
+#include "probes_mysql.h"
+#include "unireg.h"                    // REQUIRED: for other includes
+#include "sql_parse.h"                          // sql_command_flags,
+                                                // execute_init_command,
+                                                // do_command
+#include "sql_db.h"                             // mysql_change_db
+#include "hostname.h" // inc_host_errors, ip_to_hostname,
+                      // reset_host_errors
+#include "sql_acl.h"  // acl_getroot, NO_ACCESS, SUPER_ACL
+#include "sql_callback.h"
 
 HASH global_user_stats, global_client_stats, global_table_stats;
 HASH global_index_stats;
 /* Protects the above global stats */
-extern pthread_mutex_t LOCK_global_user_client_stats;
-extern pthread_mutex_t LOCK_global_table_stats;
-extern pthread_mutex_t LOCK_global_index_stats;
-
-#ifdef __WIN__
-extern void win_install_sigabrt_handler();
-#endif
+extern mysql_mutex_t LOCK_global_user_client_stats;
+extern mysql_mutex_t LOCK_global_table_stats;
+extern mysql_mutex_t LOCK_global_index_stats;
 
 /*
   Get structure for logging connection data for the current user
@@ -40,7 +51,7 @@ static HASH hash_user_connections;
 
 int get_or_create_user_conn(THD *thd, const char *user,
                             const char *host,
-                            USER_RESOURCES *mqh)
+                            const USER_RESOURCES *mqh)
 {
   int return_val= 0;
   size_t temp_len, user_len;
@@ -53,8 +64,8 @@ int get_or_create_user_conn(THD *thd, const char *user,
 
   user_len= strlen(user);
   temp_len= (strmov(strmov(temp_user, user)+1, host) - temp_user)+1;
-  (void) pthread_mutex_lock(&LOCK_user_conn);
-  if (!(uc = (struct  user_conn *) hash_search(&hash_user_connections,
+  mysql_mutex_lock(&LOCK_user_conn);
+  if (!(uc = (struct  user_conn *) my_hash_search(&hash_user_connections,
 					       (uchar*) temp_user, temp_len)))
   {
     /* First connection for user; Create a user connection object */
@@ -76,7 +87,7 @@ int get_or_create_user_conn(THD *thd, const char *user,
     if (my_hash_insert(&hash_user_connections, (uchar*) uc))
     {
       /* The only possible error is out of memory, MY_WME sets an error. */
-      my_free((char*) uc,0);
+      my_free(uc);
       return_val= 1;
       goto end;
     }
@@ -84,7 +95,7 @@ int get_or_create_user_conn(THD *thd, const char *user,
   thd->user_connect=uc;
   uc->connections++;
 end:
-  (void) pthread_mutex_unlock(&LOCK_user_conn);
+  mysql_mutex_unlock(&LOCK_user_conn);
   return return_val;
 
 }
@@ -112,9 +123,10 @@ int check_for_max_user_connections(THD *thd, USER_CONN *uc)
   int error= 1;
   DBUG_ENTER("check_for_max_user_connections");
 
-  (void) pthread_mutex_lock(&LOCK_user_conn);
-  if (max_user_connections && !uc->user_resources.user_conn &&
-      max_user_connections < (uint) uc->connections)
+  mysql_mutex_lock(&LOCK_user_conn);
+  if (global_system_variables.max_user_connections &&
+      !uc->user_resources.user_conn &&
+      global_system_variables.max_user_connections < (uint) uc->connections)
   {
     my_error(ER_TOO_MANY_USER_CONNECTIONS, MYF(0), uc->user);
     goto end;
@@ -141,8 +153,16 @@ int check_for_max_user_connections(THD *thd, USER_CONN *uc)
 
 end:
   if (error)
+  {
     uc->connections--; // no need for decrease_user_connections() here
-  (void) pthread_mutex_unlock(&LOCK_user_conn);
+    /*
+      The thread may returned back to the pool and assigned to a user
+      that doesn't have a limit. Ensure the user is not using resources
+      of someone else.
+    */
+    thd->user_connect= NULL;
+  }
+  mysql_mutex_unlock(&LOCK_user_conn);
   DBUG_RETURN(error);
 }
 
@@ -168,14 +188,14 @@ end:
 void decrease_user_connections(USER_CONN *uc)
 {
   DBUG_ENTER("decrease_user_connections");
-  (void) pthread_mutex_lock(&LOCK_user_conn);
+  mysql_mutex_lock(&LOCK_user_conn);
   DBUG_ASSERT(uc->connections);
   if (!--uc->connections && !mqh_used)
   {
     /* Last connection for user; Delete it */
-    (void) hash_delete(&hash_user_connections,(uchar*) uc);
+    (void) my_hash_delete(&hash_user_connections,(uchar*) uc);
   }
-  (void) pthread_mutex_unlock(&LOCK_user_conn);
+  mysql_mutex_unlock(&LOCK_user_conn);
   DBUG_VOID_RETURN;
 }
 
@@ -223,7 +243,7 @@ bool check_mqh(THD *thd, uint check_command)
   DBUG_ENTER("check_mqh");
   DBUG_ASSERT(uc != 0);
 
-  (void) pthread_mutex_lock(&LOCK_user_conn);
+  mysql_mutex_lock(&LOCK_user_conn);
 
   time_out_user_resource_limits(thd, uc);
 
@@ -250,13 +270,12 @@ bool check_mqh(THD *thd, uint check_command)
     }
   }
 end:
-  (void) pthread_mutex_unlock(&LOCK_user_conn);
+  mysql_mutex_unlock(&LOCK_user_conn);
   DBUG_RETURN(error);
 }
 
 #endif /* NO_EMBEDDED_ACCESS_CHECKS */
 
-
 /*
   Check for maximum allowable user connections, if the mysqld server is
   started with corresponding variable that is greater then 0.
@@ -272,17 +291,16 @@ extern "C" uchar *get_key_conn(user_conn *buff, size_t *length,
 
 extern "C" void free_user(struct user_conn *uc)
 {
-  my_free((char*) uc,MYF(0));
+  my_free(uc);
 }
 
 
 void init_max_user_conn(void)
 {
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
-  if (hash_init(&hash_user_connections,system_charset_info,max_connections,
-                0,0,
-                (hash_get_key) get_key_conn, (hash_free_key) free_user,
-                0))
+  if (my_hash_init(&hash_user_connections,system_charset_info,max_connections,
+                 0,0, (my_hash_get_key) get_key_conn,
+                 (my_hash_free_key) free_user, 0))
   {
     sql_print_error("Initializing hash_user_connections failed.");
     exit(1);
@@ -294,7 +312,7 @@ void init_max_user_conn(void)
 void free_max_user_conn(void)
 {
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
-  hash_free(&hash_user_connections);
+  my_hash_free(&hash_user_connections);
 #endif /* NO_EMBEDDED_ACCESS_CHECKS */
 }
 
@@ -302,7 +320,7 @@ void free_max_user_conn(void)
 void reset_mqh(LEX_USER *lu, bool get_them= 0)
 {
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
-  (void) pthread_mutex_lock(&LOCK_user_conn);
+  mysql_mutex_lock(&LOCK_user_conn);
   if (lu)  // for GRANT
   {
     USER_CONN *uc;
@@ -312,8 +330,9 @@ void reset_mqh(LEX_USER *lu, bool get_them= 0)
     memcpy(temp_user,lu->user.str,lu->user.length);
     memcpy(temp_user+lu->user.length+1,lu->host.str,lu->host.length);
     temp_user[lu->user.length]='\0'; temp_user[temp_len-1]=0;
-    if ((uc = (struct  user_conn *) hash_search(&hash_user_connections,
-						(uchar*) temp_user, temp_len)))
+    if ((uc = (struct  user_conn *) my_hash_search(&hash_user_connections,
+                                                   (uchar*) temp_user,
+                                                   temp_len)))
     {
       uc->questions=0;
       get_mqh(temp_user,&temp_user[lu->user.length+1],uc);
@@ -326,8 +345,8 @@ void reset_mqh(LEX_USER *lu, bool get_them= 0)
     /* for FLUSH PRIVILEGES and FLUSH USER_RESOURCES */
     for (uint idx=0;idx < hash_user_connections.records; idx++)
     {
-      USER_CONN *uc=(struct user_conn *) hash_element(&hash_user_connections,
-						      idx);
+      USER_CONN *uc=(struct user_conn *)
+        my_hash_element(&hash_user_connections, idx);
       if (get_them)
 	get_mqh(uc->user,uc->host,uc);
       uc->questions=0;
@@ -335,7 +354,7 @@ void reset_mqh(LEX_USER *lu, bool get_them= 0)
       uc->conn_per_hour=0;
     }
   }
-  (void) pthread_mutex_unlock(&LOCK_user_conn);
+  mysql_mutex_unlock(&LOCK_user_conn);
 #endif /* NO_EMBEDDED_ACCESS_CHECKS */
 }
 
@@ -373,7 +392,7 @@ extern "C" uchar *get_key_user_stats(USER_STATS *user_stats, size_t *length,
 
 void free_user_stats(USER_STATS* user_stats)
 {
-  my_free(user_stats, MYF(0));
+  my_free(user_stats);
 }
 
 void init_user_stats(USER_STATS *user_stats,
@@ -436,7 +455,7 @@ void init_user_stats(USER_STATS *user_stats,
 }
 
 
-#ifdef COMPLEAT_PATCH_NOT_ADDED_YET
+#ifdef COMPLETE_PATCH_NOT_ADDED_YET
 
 void add_user_stats(USER_STATS *user_stats,
                     uint total_connections,
@@ -490,9 +509,9 @@ void add_user_stats(USER_STATS *user_stats,
 
 void init_global_user_stats(void)
 {
-  if (hash_init(&global_user_stats, system_charset_info, max_connections,
-                0, 0, (hash_get_key) get_key_user_stats,
-                (hash_free_key)free_user_stats, 0))
+  if (my_hash_init(&global_user_stats, system_charset_info, max_connections,
+                0, 0, (my_hash_get_key) get_key_user_stats,
+                (my_hash_free_key)free_user_stats, 0))
   {
     sql_print_error("Initializing global_user_stats failed.");
     exit(1);
@@ -501,9 +520,9 @@ void init_global_user_stats(void)
 
 void init_global_client_stats(void)
 {
-  if (hash_init(&global_client_stats, system_charset_info, max_connections,
-                0, 0, (hash_get_key) get_key_user_stats,
-                (hash_free_key)free_user_stats, 0))
+  if (my_hash_init(&global_client_stats, system_charset_info, max_connections,
+                0, 0, (my_hash_get_key) get_key_user_stats,
+                (my_hash_free_key)free_user_stats, 0))
   {
     sql_print_error("Initializing global_client_stats failed.");
     exit(1);
@@ -519,14 +538,14 @@ extern "C" uchar *get_key_table_stats(TABLE_STATS *table_stats, size_t *length,
 
 extern "C" void free_table_stats(TABLE_STATS* table_stats)
 {
-  my_free(table_stats, MYF(0));
+  my_free(table_stats);
 }
 
 void init_global_table_stats(void)
 {
-  if (hash_init(&global_table_stats, system_charset_info, max_connections,
-                0, 0, (hash_get_key) get_key_table_stats,
-                (hash_free_key)free_table_stats, 0)) {
+  if (my_hash_init(&global_table_stats, system_charset_info, max_connections,
+                0, 0, (my_hash_get_key) get_key_table_stats,
+                (my_hash_free_key)free_table_stats, 0)) {
     sql_print_error("Initializing global_table_stats failed.");
     exit(1);
   }
@@ -541,14 +560,14 @@ extern "C" uchar *get_key_index_stats(INDEX_STATS *index_stats, size_t *length,
 
 extern "C" void free_index_stats(INDEX_STATS* index_stats)
 {
-  my_free(index_stats, MYF(0));
+  my_free(index_stats);
 }
 
 void init_global_index_stats(void)
 {
-  if (hash_init(&global_index_stats, system_charset_info, max_connections,
-                0, 0, (hash_get_key) get_key_index_stats,
-                (hash_free_key)free_index_stats, 0))
+  if (my_hash_init(&global_index_stats, system_charset_info, max_connections,
+                0, 0, (my_hash_get_key) get_key_index_stats,
+                (my_hash_free_key)free_index_stats, 0))
   {
     sql_print_error("Initializing global_index_stats failed.");
     exit(1);
@@ -558,22 +577,22 @@ void init_global_index_stats(void)
 
 void free_global_user_stats(void)
 {
-  hash_free(&global_user_stats);
+  my_hash_free(&global_user_stats);
 }
 
 void free_global_table_stats(void)
 {
-  hash_free(&global_table_stats);
+  my_hash_free(&global_table_stats);
 }
 
 void free_global_index_stats(void)
 {
-  hash_free(&global_index_stats);
+  my_hash_free(&global_index_stats);
 }
 
 void free_global_client_stats(void)
 {
-  hash_free(&global_client_stats);
+  my_hash_free(&global_client_stats);
 }
 
 /*
@@ -588,7 +607,7 @@ static bool increment_count_by_name(const char *name, size_t name_length,
 {
   USER_STATS *user_stats;
 
-  if (!(user_stats= (USER_STATS*) hash_search(users_or_clients, (uchar*) name,
+  if (!(user_stats= (USER_STATS*) my_hash_search(users_or_clients, (uchar*) name,
                                               name_length)))
   {
     /* First connection for this user or client */
@@ -612,7 +631,7 @@ static bool increment_count_by_name(const char *name, size_t name_length,
 
     if (my_hash_insert(users_or_clients, (uchar*)user_stats))
     {
-      my_free(user_stats, 0);
+      my_free(user_stats);
       return TRUE;                              // Out of memory
     }
   }
@@ -641,7 +660,7 @@ static bool increment_connection_count(THD* thd, bool use_lock)
     return FALSE;
 
   if (use_lock)
-    pthread_mutex_lock(&LOCK_global_user_client_stats);
+    mysql_mutex_lock(&LOCK_global_user_client_stats);
 
   if (increment_count_by_name(user_string, strlen(user_string), user_string,
                               &global_user_stats, thd))
@@ -658,7 +677,7 @@ static bool increment_connection_count(THD* thd, bool use_lock)
 
 end:
   if (use_lock)
-    pthread_mutex_unlock(&LOCK_global_user_client_stats);
+    mysql_mutex_unlock(&LOCK_global_user_client_stats);
   return return_value;
 }
 #endif
@@ -735,10 +754,10 @@ void update_global_user_stats(THD *thd, bool create_user, time_t now)
   client_string= get_client_host(thd);
   client_string_length= strlen(client_string);
 
-  pthread_mutex_lock(&LOCK_global_user_client_stats);
+  mysql_mutex_lock(&LOCK_global_user_client_stats);
 
   // Update by user name
-  if ((user_stats= (USER_STATS*) hash_search(&global_user_stats,
+  if ((user_stats= (USER_STATS*) my_hash_search(&global_user_stats,
                                              (uchar*) user_string,
                                              user_string_length)))
   {
@@ -756,7 +775,7 @@ void update_global_user_stats(THD *thd, bool create_user, time_t now)
   }
 
   /* Update by client IP */
-  if ((user_stats= (USER_STATS*)hash_search(&global_client_stats,
+  if ((user_stats= (USER_STATS*)my_hash_search(&global_client_stats,
                                             (uchar*) client_string,
                                             client_string_length)))
   {
@@ -776,7 +795,7 @@ void update_global_user_stats(THD *thd, bool create_user, time_t now)
   thd->select_commands= thd->update_commands= thd->other_commands= 0;
   thd->last_global_update_time= now;
 
-  pthread_mutex_unlock(&LOCK_global_user_client_stats);
+  mysql_mutex_unlock(&LOCK_global_user_client_stats);
 }
 
 
@@ -825,7 +844,7 @@ bool thd_init_client_charset(THD *thd, uint cs_number)
       my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), "character_set_client",
                cs->csname);
       return true;
-    }
+    }    
     thd->variables.character_set_results=
       thd->variables.collation_connection= 
       thd->variables.character_set_client= cs;
@@ -841,16 +860,12 @@ bool thd_init_client_charset(THD *thd, uint cs_number)
 bool init_new_connection_handler_thread()
 {
   pthread_detach_this_thread();
-#if defined(__WIN__)
-  win_install_sigabrt_handler();
-#else
-  /* Win32 calls this in pthread_create */
   if (my_thread_init())
     return 1;
-#endif /* __WIN__ */
   return 0;
 }
 
+#ifndef EMBEDDED_LIBRARY
 /*
   Perform handshake, authorize client and update thd ACL variables.
 
@@ -863,7 +878,6 @@ bool init_new_connection_handler_thread()
      1  error
 */
 
-#ifndef EMBEDDED_LIBRARY
 static int check_connection(THD *thd)
 {
   uint connect_errors= 0;
@@ -877,22 +891,25 @@ static int check_connection(THD *thd)
 
   if (!thd->main_security_ctx.host)         // If TCP/IP connection
   {
-    char ip[30];
+    char ip[NI_MAXHOST];
 
-    if (vio_peer_addr(net->vio, ip, &thd->peer_port))
+    if (vio_peer_addr(net->vio, ip, &thd->peer_port, NI_MAXHOST))
     {
-      my_error(ER_BAD_HOST_ERROR, MYF(0), thd->main_security_ctx.host_or_ip);
+      my_error(ER_BAD_HOST_ERROR, MYF(0));
       return 1;
     }
     if (!(thd->main_security_ctx.ip= my_strdup(ip,MYF(MY_WME))))
       return 1; /* The error is set by my_strdup(). */
     thd->main_security_ctx.host_or_ip= thd->main_security_ctx.ip;
-    vio_in_addr(net->vio,&thd->remote.sin_addr);
     if (!(specialflag & SPECIAL_NO_RESOLVE))
     {
-      vio_in_addr(net->vio,&thd->remote.sin_addr);
-      thd->main_security_ctx.host=
-        ip_to_hostname(&thd->remote.sin_addr, &connect_errors);
+      if (ip_to_hostname(&net->vio->remote, thd->main_security_ctx.ip,
+                         &thd->main_security_ctx.host, &connect_errors))
+      {
+        my_error(ER_BAD_HOST_ERROR, MYF(0));
+        return 1;
+      }
+
       /* Cut very long hostnames to avoid possible overflows */
       if (thd->main_security_ctx.host)
       {
@@ -925,7 +942,7 @@ static int check_connection(THD *thd)
     thd->main_security_ctx.host_or_ip= thd->main_security_ctx.host;
     thd->main_security_ctx.ip= 0;
     /* Reset sin_addr */
-    bzero((char*) &thd->remote, sizeof(thd->remote));
+    bzero((char*) &net->vio->remote, sizeof(net->vio->remote));
   }
   vio_keepalive(net->vio, TRUE);
   
@@ -953,9 +970,9 @@ bool setup_connection_thread_globals(THD *thd)
 {
   if (thd->store_globals())
   {
-    close_connection(thd, ER_OUT_OF_RESOURCES, 1);
+    close_connection(thd, ER_OUT_OF_RESOURCES);
     statistic_increment(aborted_connects,&LOCK_status);
-    thd->scheduler->end_thread(thd, 0);
+    MYSQL_CALLBACK(thd->scheduler, end_thread, (thd, 0));
     return 1;                                   // Error
   }
   return 0;
@@ -990,11 +1007,11 @@ bool login_connection(THD *thd)
   my_net_set_write_timeout(net, connect_timeout);
 
   error= check_connection(thd);
-  net_end_statement(thd);
+  thd->protocol->end_statement();
 
   if (error)
   {						// Wrong permissions
-#ifdef __NT__
+#ifdef _WIN32
     if (vio_type(net->vio) == VIO_TYPE_NAMEDPIPE)
       my_sleep(1000);				/* must wait after eof() */
 #endif
@@ -1008,7 +1025,7 @@ bool login_connection(THD *thd)
   /*  Updates global user connection stats. */
   if (increment_connection_count(thd, TRUE))
   {
-    net_send_error(thd, ER_OUTOFMEMORY);  // Out of memory
+    my_error(ER_OUTOFMEMORY, MYF(0), 2*sizeof(USER_STATS));
     DBUG_RETURN(1);
   }
 
@@ -1036,7 +1053,12 @@ void end_connection(THD *thd)
       again in thd->cleanup()
     */
     decrease_user_connections(thd->user_connect);
-    thd->user_connect= 0;
+    /*
+      The thread may returned back to the pool and assigned to a user
+      that doesn't have a limit. Ensure the user is not using resources
+      of someone else.
+    */
+    thd->user_connect= NULL;
   }
 
   if (thd->killed || (net->error && net->vio != 0))
@@ -1055,7 +1077,7 @@ void end_connection(THD *thd)
                         thd->thread_id,(thd->db ? thd->db : "unconnected"),
                         sctx->user ? sctx->user : "unauthenticated",
                         sctx->host_or_ip,
-                        (thd->main_da.is_error() ? thd->main_da.message() :
+                        (thd->stmt_da->is_error() ? thd->stmt_da->message() :
                          ER(ER_UNKNOWN_ERROR)));
     }
   }
@@ -1070,12 +1092,6 @@ void prepare_new_connection_state(THD* thd)
 {
   Security_context *sctx= thd->security_ctx;
 
-#ifdef __NETWARE__
-  netware_reg_user(sctx->ip, sctx->user, "MySQL");
-#endif
-
-  if (thd->variables.max_join_size == HA_POS_ERROR)
-    thd->options |= OPTION_BIG_SELECTS;
   if (thd->client_capabilities & CLIENT_COMPRESS)
     thd->net.compress=1;				// Use compression
 
@@ -1084,15 +1100,14 @@ void prepare_new_connection_state(THD* thd)
     embedded server library.
     TODO: refactor this to avoid code duplication there
   */
-  thd->version= refresh_version;
   thd->proc_info= 0;
   thd->command= COM_SLEEP;
   thd->set_time();
   thd->init_for_queries();
 
-  if (sys_init_connect.value_length && !(sctx->master_access & SUPER_ACL))
+  if (opt_init_connect.length && !(sctx->master_access & SUPER_ACL))
   {
-    execute_init_command(thd, &sys_init_connect, &LOCK_sys_init_connect);
+    execute_init_command(thd, &opt_init_connect, &LOCK_sys_init_connect);
     if (thd->is_error())
     {
       thd->killed= THD::KILL_CONNECTION;
@@ -1100,7 +1115,7 @@ void prepare_new_connection_state(THD* thd)
                         thd->thread_id,(thd->db ? thd->db : "unconnected"),
                         sctx->user ? sctx->user : "unauthenticated",
                         sctx->host_or_ip, "init_connect command failed");
-      sql_print_warning("%s", thd->main_da.message());
+      sql_print_warning("%s", thd->stmt_da->message());
     }
     thd->proc_info=0;
     thd->set_time();
@@ -1130,14 +1145,50 @@ pthread_handler_t handle_one_connection(void *arg)
 {
   THD *thd= (THD*) arg;
 
+  mysql_thread_set_psi_id(thd->thread_id);
+
+  do_handle_one_connection(thd);
+  return 0;
+}
+
+bool thd_prepare_connection(THD *thd)
+{
+  bool rc;
+  lex_start(thd);
+  rc= login_connection(thd);
+  MYSQL_AUDIT_NOTIFY_CONNECTION_CONNECT(thd);
+  if (rc)
+    return rc;
+
+  MYSQL_CONNECTION_START(thd->thread_id, &thd->security_ctx->priv_user[0],
+                         (char *) thd->security_ctx->host_or_ip);
+
+  prepare_new_connection_state(thd);
+  return FALSE;
+}
+
+bool thd_is_connection_alive(THD *thd)
+{
+  NET *net= &thd->net;
+  if (!net->error &&
+      net->vio != 0 &&
+      !(thd->killed == THD::KILL_CONNECTION))
+    return TRUE;
+  return FALSE;
+}
+
+void do_handle_one_connection(THD *thd_arg)
+{
+  THD *thd= thd_arg;
+
   thd->thr_create_utime= microsecond_interval_timer();
 
-  if (thread_scheduler.init_new_connection_thread())
+  if (MYSQL_CALLBACK_ELSE(thread_scheduler, init_new_connection_thread, (), 0))
   {
-    close_connection(thd, ER_OUT_OF_RESOURCES, 1);
+    close_connection(thd, ER_OUT_OF_RESOURCES);
     statistic_increment(aborted_connects,&LOCK_status);
-    thd->scheduler->end_thread(thd,0);
-    return 0;
+    MYSQL_CALLBACK(thd->scheduler, end_thread, (thd, 0));
+    return;
   }
 
   /*
@@ -1164,37 +1215,34 @@ pthread_handler_t handle_one_connection(void *arg)
   */
   thd->thread_stack= (char*) &thd;
   if (setup_connection_thread_globals(thd))
-    return 0;
+    return;
 
   for (;;)
   {
-    NET *net= &thd->net;
     bool create_user= TRUE;
 
-    lex_start(thd);
-    if (login_connection(thd))
+    if (thd_prepare_connection(thd))
     {
       create_user= FALSE;
       goto end_thread;
     }      
 
-    prepare_new_connection_state(thd);
-
-    while (!net->error && net->vio != 0 &&
-           !(thd->killed == THD::KILL_CONNECTION))
+    while (thd_is_connection_alive(thd))
     {
+      mysql_audit_release(thd);
       if (do_command(thd))
 	break;
     }
     end_connection(thd);
    
 end_thread:
-    close_connection(thd, 0, 1);
+    close_connection(thd);
+
     if (thd->userstat_running)
       update_global_user_stats(thd, create_user, time(NULL));
 
-    if (thd->scheduler->end_thread(thd,1))
-      return 0;                                 // Probably no-threads
+    if (MYSQL_CALLBACK_ELSE(thd->scheduler, end_thread, (thd, 1), 0))
+      return;                                 // Probably no-threads
 
     /*
       If end_thread() returns, this thread has been schedule to