summaryrefslogtreecommitdiffstats
path: root/sql/sql_connect.cc
diff options
context:
space:
mode:
Diffstat (limited to 'sql/sql_connect.cc')
-rw-r--r--sql/sql_connect.cc1582
1 files changed, 1582 insertions, 0 deletions
diff --git a/sql/sql_connect.cc b/sql/sql_connect.cc
new file mode 100644
index 00000000..999830ff
--- /dev/null
+++ b/sql/sql_connect.cc
@@ -0,0 +1,1582 @@
+/*
+ Copyright (c) 2007, 2013, Oracle and/or its affiliates.
+ Copyright (c) 2008, 2020, MariaDB
+
+ 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335 USA
+*/
+
+/*
+ Functions to autenticate and handle reqests for a connection
+*/
+
+#include "mariadb.h"
+#include "mysqld.h"
+#include "sql_priv.h"
+#ifndef __WIN__
+#include <netdb.h> // getservbyname, servent
+#endif
+#include "sql_audit.h"
+#include "sql_connect.h"
+#include "thread_cache.h"
+#include "probes_mysql.h"
+#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_callback.h"
+
+#ifdef WITH_WSREP
+#include "wsrep_trans_observer.h" /* wsrep open/close */
+#include "wsrep_mysqld.h"
+#endif /* WITH_WSREP */
+#include "proxy_protocol.h"
+#include <ssl_compat.h>
+
+HASH global_user_stats, global_client_stats, global_table_stats;
+HASH global_index_stats;
+/* Protects the above global stats */
+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;
+extern vio_keepalive_opts opt_vio_keepalive;
+
+/*
+ Get structure for logging connection data for the current user
+*/
+
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+static HASH hash_user_connections;
+
+int get_or_create_user_conn(THD *thd, const char *user,
+ const char *host,
+ const USER_RESOURCES *mqh)
+{
+ int return_val= 0;
+ size_t temp_len, user_len;
+ char temp_user[USER_HOST_BUFF_SIZE];
+ struct user_conn *uc;
+
+ DBUG_ASSERT(user != 0);
+ DBUG_ASSERT(host != 0);
+ DBUG_ASSERT(thd->user_connect == 0);
+
+ user_len= strlen(user);
+ temp_len= (strmov(strmov(temp_user, user)+1, host) - temp_user)+1;
+ 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 */
+ if (!(uc= ((struct user_conn*)
+ my_malloc(key_memory_user_conn,
+ sizeof(struct user_conn) + temp_len+1, MYF(MY_WME)))))
+ {
+ /* MY_WME ensures an error is set in THD. */
+ return_val= 1;
+ goto end;
+ }
+ uc->user=(char*) (uc+1);
+ memcpy(uc->user,temp_user,temp_len+1);
+ uc->host= uc->user + user_len + 1;
+ uc->len= (uint)temp_len;
+ uc->connections= uc->questions= uc->updates= uc->conn_per_hour= 0;
+ uc->reset_utime= thd->thr_create_utime;
+ if (my_hash_insert(&hash_user_connections, (uchar*) uc))
+ {
+ /* The only possible error is out of memory, MY_WME sets an error. */
+ my_free(uc);
+ return_val= 1;
+ goto end;
+ }
+ }
+ uc->user_resources= *mqh;
+ thd->user_connect=uc;
+ uc->connections++;
+end:
+ mysql_mutex_unlock(&LOCK_user_conn);
+ return return_val;
+}
+
+
+/*
+ check if user has already too many connections
+
+ SYNOPSIS
+ check_for_max_user_connections()
+ thd Thread handle
+ uc User connect object
+
+ NOTES
+ If check fails, we decrease user connection count, which means one
+ shouldn't call decrease_user_connections() after this function.
+
+ RETURN
+ 0 ok
+ 1 error
+*/
+
+int check_for_max_user_connections(THD *thd, USER_CONN *uc)
+{
+ int error= 1;
+ Host_errors errors;
+ DBUG_ENTER("check_for_max_user_connections");
+
+ mysql_mutex_lock(&LOCK_user_conn);
+
+ /* Root is not affected by the value of max_user_connections */
+ if (global_system_variables.max_user_connections &&
+ !uc->user_resources.user_conn &&
+ global_system_variables.max_user_connections < uc->connections &&
+ !(thd->security_ctx->master_access & PRIV_IGNORE_MAX_USER_CONNECTIONS))
+ {
+ my_error(ER_TOO_MANY_USER_CONNECTIONS, MYF(0), uc->user);
+ error=1;
+ errors.m_max_user_connection= 1;
+ goto end;
+ }
+ time_out_user_resource_limits(thd, uc);
+ if (uc->user_resources.user_conn &&
+ uc->user_resources.user_conn < uc->connections)
+ {
+ my_error(ER_USER_LIMIT_REACHED, MYF(0), uc->user,
+ "max_user_connections",
+ (long) uc->user_resources.user_conn);
+ error= 1;
+ errors.m_max_user_connection= 1;
+ goto end;
+ }
+ if (uc->user_resources.conn_per_hour &&
+ uc->user_resources.conn_per_hour <= uc->conn_per_hour)
+ {
+ my_error(ER_USER_LIMIT_REACHED, MYF(0), uc->user,
+ "max_connections_per_hour",
+ (long) uc->user_resources.conn_per_hour);
+ error=1;
+ errors.m_max_user_connection_per_hour= 1;
+ goto end;
+ }
+ uc->conn_per_hour++;
+ error= 0;
+
+end:
+ if (unlikely(error))
+ {
+ uc->connections--; // no need for decrease_user_connections() here
+ /*
+ 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);
+ if (unlikely(error))
+ {
+ inc_host_errors(thd->main_security_ctx.ip, &errors);
+ }
+ DBUG_RETURN(error);
+}
+
+
+/*
+ Decrease user connection count
+
+ SYNOPSIS
+ decrease_user_connections()
+ uc User connection object
+
+ NOTES
+ If there is a n user connection object for a connection
+ (which only happens if 'max_user_connections' is defined or
+ if someone has created a resource grant for a user), then
+ the connection count is always incremented on connect.
+
+ The user connect object is not freed if some users has
+ 'max connections per hour' defined as we need to be able to hold
+ count over the lifetime of the connection.
+*/
+
+void decrease_user_connections(USER_CONN *uc)
+{
+ DBUG_ENTER("decrease_user_connections");
+ mysql_mutex_lock(&LOCK_user_conn);
+ DBUG_ASSERT(uc->connections);
+ if (!--uc->connections && !mqh_used)
+ {
+ /* Last connection for user; Delete it */
+ (void) my_hash_delete(&hash_user_connections,(uchar*) uc);
+ }
+ mysql_mutex_unlock(&LOCK_user_conn);
+ DBUG_VOID_RETURN;
+}
+
+
+/*
+ Reset per-hour user resource limits when it has been more than
+ an hour since they were last checked
+
+ SYNOPSIS:
+ time_out_user_resource_limits()
+ thd Thread handler
+ uc User connection details
+
+ NOTE:
+ This assumes that the LOCK_user_conn mutex has been acquired, so it is
+ safe to test and modify members of the USER_CONN structure.
+*/
+
+void time_out_user_resource_limits(THD *thd, USER_CONN *uc)
+{
+ ulonglong check_time= thd->start_utime;
+ DBUG_ENTER("time_out_user_resource_limits");
+
+ /* If more than a hour since last check, reset resource checking */
+ if (check_time - uc->reset_utime >= 3600000000ULL)
+ {
+ uc->questions=0;
+ uc->updates=0;
+ uc->conn_per_hour=0;
+ uc->reset_utime= check_time;
+ }
+
+ DBUG_VOID_RETURN;
+}
+
+/*
+ Check if maximum queries per hour limit has been reached
+ returns 0 if OK.
+*/
+
+bool check_mqh(THD *thd, uint check_command)
+{
+ bool error= 0;
+ USER_CONN *uc=thd->user_connect;
+ DBUG_ENTER("check_mqh");
+ DBUG_ASSERT(uc != 0);
+
+ mysql_mutex_lock(&LOCK_user_conn);
+
+ time_out_user_resource_limits(thd, uc);
+
+ /* Check that we have not done too many questions / hour */
+ if (uc->user_resources.questions &&
+ uc->questions++ >= uc->user_resources.questions)
+ {
+ my_error(ER_USER_LIMIT_REACHED, MYF(0), uc->user, "max_queries_per_hour",
+ (long) uc->user_resources.questions);
+ error=1;
+ goto end;
+ }
+ if (check_command < (uint) SQLCOM_END)
+ {
+ /* Check that we have not done too many updates / hour */
+ if (uc->user_resources.updates &&
+ (sql_command_flags[check_command] & CF_CHANGES_DATA) &&
+ uc->updates++ >= uc->user_resources.updates)
+ {
+ my_error(ER_USER_LIMIT_REACHED, MYF(0), uc->user, "max_updates_per_hour",
+ (long) uc->user_resources.updates);
+ error=1;
+ goto end;
+ }
+ }
+end:
+ 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.
+*/
+
+extern "C" uchar *get_key_conn(user_conn *buff, size_t *length,
+ my_bool not_used __attribute__((unused)))
+{
+ *length= buff->len;
+ return (uchar*) buff->user;
+}
+
+
+extern "C" void free_user(struct user_conn *uc)
+{
+ my_free(uc);
+}
+
+
+void init_max_user_conn(void)
+{
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+ my_hash_init(key_memory_user_conn, &hash_user_connections,
+ system_charset_info, max_connections, 0, 0, (my_hash_get_key)
+ get_key_conn, (my_hash_free_key) free_user, 0);
+#endif
+}
+
+
+void free_max_user_conn(void)
+{
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+ my_hash_free(&hash_user_connections);
+#endif /* NO_EMBEDDED_ACCESS_CHECKS */
+}
+
+
+void reset_mqh(LEX_USER *lu, bool get_them= 0)
+{
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+ mysql_mutex_lock(&LOCK_user_conn);
+ if (lu) // for GRANT
+ {
+ USER_CONN *uc;
+ size_t temp_len=lu->user.length+lu->host.length+2;
+ char temp_user[USER_HOST_BUFF_SIZE];
+
+ 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 *) 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);
+ uc->updates=0;
+ uc->conn_per_hour=0;
+ }
+ }
+ else
+ {
+ /* for FLUSH PRIVILEGES and FLUSH USER_RESOURCES */
+ for (uint idx=0;idx < hash_user_connections.records; 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;
+ uc->updates=0;
+ uc->conn_per_hour=0;
+ }
+ }
+ mysql_mutex_unlock(&LOCK_user_conn);
+#endif /* NO_EMBEDDED_ACCESS_CHECKS */
+}
+
+/*****************************************************************************
+ Handle users statistics
+*****************************************************************************/
+
+/* 'mysql_system_user' is used for when the user is not defined for a THD. */
+static const char mysql_system_user[]= "#mysql_system#";
+
+// Returns 'user' if it's not NULL. Returns 'mysql_system_user' otherwise.
+static const char * get_valid_user_string(const char* user)
+{
+ return user ? user : mysql_system_user;
+}
+
+/*
+ Returns string as 'IP' for the client-side of the connection represented by
+ 'client'. Does not allocate memory. May return "".
+*/
+
+static const char *get_client_host(THD *client)
+{
+ return client->security_ctx->host_or_ip[0] ?
+ client->security_ctx->host_or_ip :
+ client->security_ctx->host ? client->security_ctx->host : "";
+}
+
+extern "C" uchar *get_key_user_stats(USER_STATS *user_stats, size_t *length,
+ my_bool not_used __attribute__((unused)))
+{
+ *length= user_stats->user_name_length;
+ return (uchar*) user_stats->user;
+}
+
+void free_user_stats(USER_STATS* user_stats)
+{
+ my_free(user_stats);
+}
+
+void init_user_stats(USER_STATS *user_stats,
+ const char *user,
+ size_t user_length,
+ const char *priv_user,
+ uint total_connections,
+ uint total_ssl_connections,
+ uint concurrent_connections,
+ time_t connected_time,
+ double busy_time,
+ double cpu_time,
+ ulonglong bytes_received,
+ ulonglong bytes_sent,
+ ulonglong binlog_bytes_written,
+ ha_rows rows_sent,
+ ha_rows rows_read,
+ ha_rows rows_inserted,
+ ha_rows rows_deleted,
+ ha_rows rows_updated,
+ ulonglong select_commands,
+ ulonglong update_commands,
+ ulonglong other_commands,
+ ulonglong commit_trans,
+ ulonglong rollback_trans,
+ ulonglong denied_connections,
+ ulonglong lost_connections,
+ ulonglong max_statement_time_exceeded,
+ ulonglong access_denied_errors,
+ ulonglong empty_queries)
+{
+ DBUG_ENTER("init_user_stats");
+ DBUG_PRINT("enter", ("user: %s priv_user: %s", user, priv_user));
+
+ user_length= MY_MIN(user_length, sizeof(user_stats->user)-1);
+ memcpy(user_stats->user, user, user_length);
+ user_stats->user[user_length]= 0;
+ user_stats->user_name_length= (uint)user_length;
+ strmake_buf(user_stats->priv_user, priv_user);
+
+ user_stats->total_connections= total_connections;
+ user_stats->total_ssl_connections= total_ssl_connections;
+ user_stats->concurrent_connections= concurrent_connections;
+ user_stats->connected_time= connected_time;
+ user_stats->busy_time= busy_time;
+ user_stats->cpu_time= cpu_time;
+ user_stats->bytes_received= bytes_received;
+ user_stats->bytes_sent= bytes_sent;
+ user_stats->binlog_bytes_written= binlog_bytes_written;
+ user_stats->rows_sent= rows_sent;
+ user_stats->rows_read= rows_read;
+ user_stats->rows_inserted= rows_inserted;
+ user_stats->rows_deleted= rows_deleted;
+ user_stats->rows_updated= rows_updated;
+ user_stats->select_commands= select_commands;
+ user_stats->update_commands= update_commands;
+ user_stats->other_commands= other_commands;
+ user_stats->commit_trans= commit_trans;
+ user_stats->rollback_trans= rollback_trans;
+ user_stats->denied_connections= denied_connections;
+ user_stats->lost_connections= lost_connections;
+ user_stats->max_statement_time_exceeded= max_statement_time_exceeded;
+ user_stats->access_denied_errors= access_denied_errors;
+ user_stats->empty_queries= empty_queries;
+ DBUG_VOID_RETURN;
+}
+
+
+void init_global_user_stats(void)
+{
+ my_hash_init(PSI_INSTRUMENT_ME, &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);
+}
+
+void init_global_client_stats(void)
+{
+ my_hash_init(PSI_INSTRUMENT_ME, &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);
+}
+
+extern "C" uchar *get_key_table_stats(TABLE_STATS *table_stats, size_t *length,
+ my_bool not_used __attribute__((unused)))
+{
+ *length= table_stats->table_name_length;
+ return (uchar*) table_stats->table;
+}
+
+extern "C" void free_table_stats(TABLE_STATS* table_stats)
+{
+ my_free(table_stats);
+}
+
+void init_global_table_stats(void)
+{
+ my_hash_init(PSI_INSTRUMENT_ME, &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);
+}
+
+extern "C" uchar *get_key_index_stats(INDEX_STATS *index_stats, size_t *length,
+ my_bool not_used __attribute__((unused)))
+{
+ *length= index_stats->index_name_length;
+ return (uchar*) index_stats->index;
+}
+
+extern "C" void free_index_stats(INDEX_STATS* index_stats)
+{
+ my_free(index_stats);
+}
+
+void init_global_index_stats(void)
+{
+ my_hash_init(PSI_INSTRUMENT_ME, &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);
+}
+
+
+void free_global_user_stats(void)
+{
+ my_hash_free(&global_user_stats);
+}
+
+void free_global_table_stats(void)
+{
+ my_hash_free(&global_table_stats);
+}
+
+void free_global_index_stats(void)
+{
+ my_hash_free(&global_index_stats);
+}
+
+void free_global_client_stats(void)
+{
+ my_hash_free(&global_client_stats);
+}
+
+/*
+ Increments the global stats connection count for an entry from
+ global_client_stats or global_user_stats. Returns 0 on success
+ and 1 on error.
+*/
+
+static bool increment_count_by_name(const char *name, size_t name_length,
+ const char *role_name,
+ HASH *users_or_clients, THD *thd)
+{
+ USER_STATS *user_stats;
+
+ if (!(user_stats= (USER_STATS*) my_hash_search(users_or_clients, (uchar*) name,
+ name_length)))
+ {
+ /* First connection for this user or client */
+ if (!(user_stats= ((USER_STATS*)
+ my_malloc(PSI_INSTRUMENT_ME, sizeof(USER_STATS),
+ MYF(MY_WME | MY_ZEROFILL)))))
+ return TRUE; // Out of memory
+
+ init_user_stats(user_stats, name, name_length, role_name,
+ 0, 0, 0, // connections
+ 0, 0, 0, // time
+ 0, 0, 0, // bytes sent, received and written
+ 0, 0, // rows sent and read
+ 0, 0, 0, // rows inserted, deleted and updated
+ 0, 0, 0, // select, update and other commands
+ 0, 0, // commit and rollback trans
+ thd->status_var.access_denied_errors,
+ 0, // lost connections
+ 0, // max query timeouts
+ 0, // access denied errors
+ 0); // empty queries
+
+ if (my_hash_insert(users_or_clients, (uchar*)user_stats))
+ {
+ my_free(user_stats);
+ return TRUE; // Out of memory
+ }
+ }
+ user_stats->total_connections++;
+ if (thd->net.vio && thd->net.vio->type == VIO_TYPE_SSL)
+ user_stats->total_ssl_connections++;
+ return FALSE;
+}
+
+
+/*
+ Increments the global user and client stats connection count.
+
+ @param use_lock if true, LOCK_global_user_client_stats will be locked
+
+ @retval 0 ok
+ @retval 1 error.
+*/
+
+#ifndef EMBEDDED_LIBRARY
+static bool increment_connection_count(THD* thd, bool use_lock)
+{
+ const char *user_string= get_valid_user_string(thd->main_security_ctx.user);
+ const char *client_string= get_client_host(thd);
+ bool return_value= FALSE;
+
+ if (!thd->userstat_running)
+ return FALSE;
+
+ if (use_lock)
+ mysql_mutex_lock(&LOCK_global_user_client_stats);
+
+ if (increment_count_by_name(user_string, strlen(user_string), user_string,
+ &global_user_stats, thd))
+ {
+ return_value= TRUE;
+ goto end;
+ }
+ if (increment_count_by_name(client_string, strlen(client_string),
+ user_string, &global_client_stats, thd))
+ {
+ return_value= TRUE;
+ goto end;
+ }
+
+end:
+ if (use_lock)
+ mysql_mutex_unlock(&LOCK_global_user_client_stats);
+ return return_value;
+}
+#endif
+
+/*
+ Used to update the global user and client stats
+*/
+
+static void update_global_user_stats_with_user(THD *thd,
+ USER_STATS *user_stats,
+ time_t now)
+{
+ DBUG_ASSERT(thd->userstat_running);
+
+ user_stats->connected_time+= now - thd->last_global_update_time;
+ user_stats->busy_time+= (thd->status_var.busy_time -
+ thd->org_status_var.busy_time);
+ user_stats->cpu_time+= (thd->status_var.cpu_time -
+ thd->org_status_var.cpu_time);
+ /*
+ This is handle specially as bytes_received is incremented BEFORE
+ org_status_var is copied.
+ */
+ user_stats->bytes_received+= (thd->org_status_var.bytes_received-
+ thd->start_bytes_received);
+ user_stats->bytes_sent+= (thd->status_var.bytes_sent -
+ thd->org_status_var.bytes_sent);
+ user_stats->binlog_bytes_written+=
+ (thd->status_var.binlog_bytes_written -
+ thd->org_status_var.binlog_bytes_written);
+ /* We are not counting rows in internal temporary tables here ! */
+ user_stats->rows_read+= (thd->status_var.rows_read -
+ thd->org_status_var.rows_read);
+ user_stats->rows_sent+= (thd->status_var.rows_sent -
+ thd->org_status_var.rows_sent);
+ user_stats->rows_inserted+= (thd->status_var.ha_write_count -
+ thd->org_status_var.ha_write_count);
+ user_stats->rows_deleted+= (thd->status_var.ha_delete_count -
+ thd->org_status_var.ha_delete_count);
+ user_stats->rows_updated+= (thd->status_var.ha_update_count -
+ thd->org_status_var.ha_update_count);
+ user_stats->select_commands+= thd->select_commands;
+ user_stats->update_commands+= thd->update_commands;
+ user_stats->other_commands+= thd->other_commands;
+ user_stats->commit_trans+= (thd->status_var.ha_commit_count -
+ thd->org_status_var.ha_commit_count);
+ user_stats->rollback_trans+= (thd->status_var.ha_rollback_count +
+ thd->status_var.ha_savepoint_rollback_count -
+ thd->org_status_var.ha_rollback_count -
+ thd->org_status_var.
+ ha_savepoint_rollback_count);
+ user_stats->access_denied_errors+=
+ (thd->status_var.access_denied_errors -
+ thd->org_status_var.access_denied_errors);
+ user_stats->empty_queries+= (thd->status_var.empty_queries -
+ thd->org_status_var.empty_queries);
+
+ /* The following can only contain 0 or 1 and then connection ends */
+ user_stats->denied_connections+= thd->status_var.access_denied_errors;
+ user_stats->lost_connections+= thd->status_var.lost_connections;
+ user_stats->max_statement_time_exceeded+= thd->status_var.max_statement_time_exceeded;
+}
+
+
+/* Updates the global stats of a user or client */
+void update_global_user_stats(THD *thd, bool create_user, time_t now)
+{
+ const char *user_string, *client_string;
+ USER_STATS *user_stats;
+ size_t user_string_length, client_string_length;
+ DBUG_ASSERT(thd->userstat_running);
+
+ user_string= get_valid_user_string(thd->main_security_ctx.user);
+ user_string_length= strlen(user_string);
+ client_string= get_client_host(thd);
+ client_string_length= strlen(client_string);
+
+ mysql_mutex_lock(&LOCK_global_user_client_stats);
+
+ // Update by user name
+ if ((user_stats= (USER_STATS*) my_hash_search(&global_user_stats,
+ (uchar*) user_string,
+ user_string_length)))
+ {
+ /* Found user. */
+ update_global_user_stats_with_user(thd, user_stats, now);
+ }
+ else
+ {
+ /* Create the entry */
+ if (create_user)
+ {
+ increment_count_by_name(user_string, user_string_length, user_string,
+ &global_user_stats, thd);
+ }
+ }
+
+ /* Update by client IP */
+ if ((user_stats= (USER_STATS*)my_hash_search(&global_client_stats,
+ (uchar*) client_string,
+ client_string_length)))
+ {
+ // Found by client IP
+ update_global_user_stats_with_user(thd, user_stats, now);
+ }
+ else
+ {
+ // Create the entry
+ if (create_user)
+ {
+ increment_count_by_name(client_string, client_string_length,
+ user_string, &global_client_stats, thd);
+ }
+ }
+ /* Reset variables only used for counting */
+ thd->select_commands= thd->update_commands= thd->other_commands= 0;
+ thd->last_global_update_time= now;
+
+ mysql_mutex_unlock(&LOCK_global_user_client_stats);
+}
+
+
+/**
+ Set thread character set variables from the given ID
+
+ @param thd thread handle
+ @param cs_number character set and collation ID
+
+ @retval 0 OK; character_set_client, collation_connection and
+ character_set_results are set to the new value,
+ or to the default global values.
+
+ @retval 1 error, e.g. the given ID is not supported by parser.
+ Corresponding SQL error is sent.
+*/
+
+bool thd_init_client_charset(THD *thd, uint cs_number)
+{
+ CHARSET_INFO *cs;
+ /*
+ Use server character set and collation if
+ - opt_character_set_client_handshake is not set
+ - client has not specified a character set
+ - client character set doesn't exists in server
+ */
+ if (!opt_character_set_client_handshake ||
+ !(cs= get_charset(cs_number, MYF(0))))
+ {
+ thd->update_charset(global_system_variables.character_set_client,
+ global_system_variables.collation_connection,
+ global_system_variables.character_set_results);
+ }
+ else
+ {
+ if (!is_supported_parser_charset(cs))
+ {
+ /* Disallow non-supported parser character sets: UCS2, UTF16, UTF32 */
+ my_error(ER_WRONG_VALUE_FOR_VAR, MYF(0), "character_set_client",
+ cs->csname);
+ return true;
+ }
+ thd->org_charset= cs;
+ thd->update_charset(cs,cs,cs);
+ }
+ return false;
+}
+
+
+/*
+ Initialize connection threads
+*/
+
+#ifndef EMBEDDED_LIBRARY
+bool init_new_connection_handler_thread()
+{
+ pthread_detach_this_thread();
+ if (my_thread_init())
+ {
+ statistic_increment(aborted_connects,&LOCK_status);
+ statistic_increment(connection_errors_internal, &LOCK_status);
+ return 1;
+ }
+ DBUG_EXECUTE_IF("simulate_failed_connection_1", return(1); );
+ return 0;
+}
+
+/**
+ Set client address during authentication.
+
+ Initializes THD::main_security_ctx and THD::peer_port.
+ Optionally does ip to hostname translation.
+
+ @param thd current THD handle
+ @param addr peer address (can be NULL, if 'ip' is set)
+ @param ip peer address as string (can be NULL if 'addr' is set)
+ @param port peer port
+ @param check_proxy_networks if true, and host is in
+ 'proxy_protocol_networks' list, skip
+ "host not privileged" check
+ @param[out] host_errors - number of connect
+ errors for this host
+
+ @retval 0 ok, 1 error
+*/
+int thd_set_peer_addr(THD *thd,
+ sockaddr_storage *addr,
+ const char *ip,
+ uint port,
+ bool check_proxy_networks,
+ uint *host_errors)
+{
+ *host_errors= 0;
+
+ thd->peer_port= port;
+
+ char ip_string[128];
+ if (!ip)
+ {
+ void *addr_data;
+ if (addr->ss_family == AF_UNIX)
+ {
+ /* local connection */
+ my_free((void *)thd->main_security_ctx.ip);
+ thd->main_security_ctx.host_or_ip= thd->main_security_ctx.host = my_localhost;
+ thd->main_security_ctx.ip= 0;
+ return 0;
+ }
+ else if (addr->ss_family == AF_INET)
+ addr_data= &((struct sockaddr_in *)addr)->sin_addr;
+ else
+ addr_data= &((struct sockaddr_in6 *)addr)->sin6_addr;
+ if (!inet_ntop(addr->ss_family,addr_data, ip_string, sizeof(ip_string)))
+ {
+ DBUG_ASSERT(0);
+ return 1;
+ }
+ ip= ip_string;
+ }
+
+ my_free((void *)thd->main_security_ctx.ip);
+ if (!(thd->main_security_ctx.ip = my_strdup(PSI_INSTRUMENT_ME, ip, MYF(MY_WME))))
+ {
+ /*
+ No error accounting per IP in host_cache,
+ this is treated as a global server OOM error.
+ TODO: remove the need for my_strdup.
+ */
+ statistic_increment(aborted_connects, &LOCK_status);
+ statistic_increment(connection_errors_internal, &LOCK_status);
+ return 1; /* The error is set by my_strdup(). */
+ }
+ thd->main_security_ctx.host_or_ip = thd->main_security_ctx.ip;
+ if (!(specialflag & SPECIAL_NO_RESOLVE))
+ {
+ int rc;
+
+ rc = ip_to_hostname(addr,
+ thd->main_security_ctx.ip,
+ &thd->main_security_ctx.host,
+ host_errors);
+
+ /* Cut very long hostnames to avoid possible overflows */
+ if (thd->main_security_ctx.host)
+ {
+ if (thd->main_security_ctx.host != my_localhost)
+ ((char*)thd->main_security_ctx.host)[MY_MIN(strlen(thd->main_security_ctx.host),
+ HOSTNAME_LENGTH)] = 0;
+ thd->main_security_ctx.host_or_ip = thd->main_security_ctx.host;
+ }
+
+ if (rc == RC_BLOCKED_HOST)
+ {
+ /* HOST_CACHE stats updated by ip_to_hostname(). */
+ my_error(ER_HOST_IS_BLOCKED, MYF(0), thd->main_security_ctx.host_or_ip);
+ return 1;
+ }
+ }
+ DBUG_PRINT("info", ("Host: %s ip: %s",
+ (thd->main_security_ctx.host ?
+ thd->main_security_ctx.host : "unknown host"),
+ (thd->main_security_ctx.ip ?
+ thd->main_security_ctx.ip : "unknown ip")));
+ if ((!check_proxy_networks || !is_proxy_protocol_allowed((struct sockaddr *) addr))
+ && acl_check_host(thd->main_security_ctx.host, thd->main_security_ctx.ip))
+ {
+ /* HOST_CACHE stats updated by acl_check_host(). */
+ my_error(ER_HOST_NOT_PRIVILEGED, MYF(0),
+ thd->main_security_ctx.host_or_ip);
+ return 1;
+ }
+ return 0;
+}
+
+/*
+ Perform handshake, authorize client and update thd ACL variables.
+
+ SYNOPSIS
+ check_connection()
+ thd thread handle
+
+ RETURN
+ 0 success, thd is updated.
+ 1 error
+*/
+
+static int check_connection(THD *thd)
+{
+ uint connect_errors= 0;
+ int auth_rc;
+ NET *net= &thd->net;
+
+ DBUG_PRINT("info",
+ ("New connection received on %s", vio_description(net->vio)));
+
+#ifdef SIGNAL_WITH_VIO_CLOSE
+ thd->set_active_vio(net->vio);
+#endif
+
+ if (!thd->main_security_ctx.host) // If TCP/IP connection
+ {
+ my_bool peer_rc;
+ char ip[NI_MAXHOST];
+ uint16 peer_port;
+
+ peer_rc= vio_peer_addr(net->vio, ip, &peer_port, NI_MAXHOST);
+
+ /*
+ ===========================================================================
+ DEBUG code only (begin)
+ Simulate various output from vio_peer_addr().
+ ===========================================================================
+ */
+
+ DBUG_EXECUTE_IF("vio_peer_addr_error",
+ {
+ peer_rc= 1;
+ }
+ );
+ DBUG_EXECUTE_IF("vio_peer_addr_fake_ipv4",
+ {
+ struct sockaddr *sa= (sockaddr *) &net->vio->remote;
+ sa->sa_family= AF_INET;
+ struct in_addr *ip4= &((struct sockaddr_in *) sa)->sin_addr;
+ /* See RFC 5737, 192.0.2.0/24 is reserved. */
+ const char* fake= "192.0.2.4";
+ inet_pton(AF_INET,fake, ip4);
+ strcpy(ip, fake);
+ peer_rc= 0;
+ }
+ );
+
+#ifdef HAVE_IPV6
+ DBUG_EXECUTE_IF("vio_peer_addr_fake_ipv6",
+ {
+ struct sockaddr_in6 *sa= (sockaddr_in6 *) &net->vio->remote;
+ sa->sin6_family= AF_INET6;
+ struct in6_addr *ip6= & sa->sin6_addr;
+ /* See RFC 3849, ipv6 2001:DB8::/32 is reserved. */
+ const char* fake= "2001:db8::6:6";
+ /* inet_pton(AF_INET6, fake, ip6); not available on Windows XP. */
+ ip6->s6_addr[ 0] = 0x20;
+ ip6->s6_addr[ 1] = 0x01;
+ ip6->s6_addr[ 2] = 0x0d;
+ ip6->s6_addr[ 3] = 0xb8;
+ ip6->s6_addr[ 4] = 0x00;
+ ip6->s6_addr[ 5] = 0x00;
+ ip6->s6_addr[ 6] = 0x00;
+ ip6->s6_addr[ 7] = 0x00;
+ ip6->s6_addr[ 8] = 0x00;
+ ip6->s6_addr[ 9] = 0x00;
+ ip6->s6_addr[10] = 0x00;
+ ip6->s6_addr[11] = 0x00;
+ ip6->s6_addr[12] = 0x00;
+ ip6->s6_addr[13] = 0x06;
+ ip6->s6_addr[14] = 0x00;
+ ip6->s6_addr[15] = 0x06;
+ strcpy(ip, fake);
+ peer_rc= 0;
+ }
+ );
+#endif /* HAVE_IPV6 */
+
+ /*
+ ===========================================================================
+ DEBUG code only (end)
+ ===========================================================================
+ */
+
+ if (peer_rc)
+ {
+ /*
+ Since we can not even get the peer IP address,
+ there is nothing to show in the host_cache,
+ so increment the global status variable for peer address errors.
+ */
+ statistic_increment(connection_errors_peer_addr, &LOCK_status);
+ my_error(ER_BAD_HOST_ERROR, MYF(0));
+ statistic_increment(aborted_connects_preauth, &LOCK_status);
+ return 1;
+ }
+
+ if (thd_set_peer_addr(thd, &net->vio->remote, ip, peer_port,
+ true, &connect_errors))
+ {
+ statistic_increment(aborted_connects_preauth, &LOCK_status);
+ return 1;
+ }
+ }
+ else /* Hostname given means that the connection was on a socket */
+ {
+ DBUG_PRINT("info",("Host: %s", thd->main_security_ctx.host));
+ thd->main_security_ctx.host_or_ip= thd->main_security_ctx.host;
+ thd->main_security_ctx.ip= 0;
+ /* Reset sin_addr */
+ bzero((char*) &net->vio->remote, sizeof(net->vio->remote));
+ }
+ vio_keepalive(net->vio, TRUE);
+ vio_set_keepalive_options(net->vio, &opt_vio_keepalive);
+
+ if (unlikely(thd->packet.alloc(thd->variables.net_buffer_length)))
+ {
+ /*
+ Important note:
+ net_buffer_length is a SESSION variable,
+ so it may be tempting to account OOM conditions per IP in the HOST_CACHE,
+ in case some clients are more demanding than others ...
+ However, this session variable is *not* initialized with a per client
+ value during the initial connection, it is initialized from the
+ GLOBAL net_buffer_length variable from the server.
+ Hence, there is no reason to account on OOM conditions per client IP,
+ we count failures in the global server status instead.
+ */
+ statistic_increment(aborted_connects,&LOCK_status);
+ statistic_increment(connection_errors_internal, &LOCK_status);
+ statistic_increment(aborted_connects_preauth, &LOCK_status);
+ return 1; /* The error is set by alloc(). */
+ }
+
+ auth_rc= acl_authenticate(thd, 0);
+ if (auth_rc == 0 && connect_errors != 0)
+ {
+ /*
+ A client connection from this IP was successful,
+ after some previous failures.
+ Reset the connection error counter.
+ */
+ reset_host_connect_errors(thd->main_security_ctx.ip);
+ }
+
+ return auth_rc;
+}
+
+
+/*
+ Setup thread to be used with the current thread
+
+ SYNOPSIS
+ bool setup_connection_thread_globals()
+ thd Thread/connection handler
+
+ RETURN
+ 0 ok
+ 1 Error (out of memory)
+ In this case we will close the connection and increment status
+*/
+
+void setup_connection_thread_globals(THD *thd)
+{
+ thd->store_globals();
+}
+
+
+/*
+ Autenticate user, with error reporting
+
+ SYNOPSIS
+ login_connection()
+ thd Thread handler
+
+ NOTES
+ Connection is not closed in case of errors
+
+ RETURN
+ 0 ok
+ 1 error
+*/
+
+bool login_connection(THD *thd)
+{
+ NET *net= &thd->net;
+ int error= 0;
+ DBUG_ENTER("login_connection");
+ DBUG_PRINT("info", ("login_connection called by thread %lu",
+ (ulong) thd->thread_id));
+
+ /* Use "connect_timeout" value during connection phase */
+ my_net_set_read_timeout(net, connect_timeout);
+ my_net_set_write_timeout(net, connect_timeout);
+
+ error= check_connection(thd);
+ thd->protocol->end_statement();
+
+ if (unlikely(error))
+ { // Wrong permissions
+#ifdef _WIN32
+ if (vio_type(net->vio) == VIO_TYPE_NAMEDPIPE)
+ my_sleep(1000); /* must wait after eof() */
+#endif
+ statistic_increment(aborted_connects,&LOCK_status);
+ error=1;
+ goto exit;
+ }
+ /* Connect completed, set read/write timeouts back to default */
+ my_net_set_read_timeout(net, thd->variables.net_read_timeout);
+ my_net_set_write_timeout(net, thd->variables.net_write_timeout);
+
+ /* Updates global user connection stats. */
+ if (increment_connection_count(thd, TRUE))
+ {
+ my_error(ER_OUTOFMEMORY, MYF(0), (int) (2*sizeof(USER_STATS)));
+ error= 1;
+ goto exit;
+ }
+
+exit:
+ mysql_audit_notify_connection_connect(thd);
+ DBUG_RETURN(error);
+}
+
+
+/*
+ Close an established connection
+
+ NOTES
+ This mainly updates status variables
+*/
+
+void end_connection(THD *thd)
+{
+ NET *net= &thd->net;
+
+#ifdef WITH_WSREP
+ if (thd->wsrep_cs().state() == wsrep::client_state::s_exec)
+ {
+ /* Error happened after the thread acquired ownership to wsrep
+ client state, but before command was processed. Clean up the
+ state before wsrep_close(). */
+ wsrep_after_command_ignore_result(thd);
+ }
+ wsrep_close(thd);
+#endif /* WITH_WSREP */
+ if (thd->user_connect)
+ {
+ /*
+ We decrease this variable early to make it easy to log again quickly.
+ This code is not critical as we will in any case do this test
+ again in thd->cleanup()
+ */
+ decrease_user_connections(thd->user_connect);
+ /*
+ 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 (unlikely(thd->killed) || (net->error && net->vio != 0))
+ {
+ statistic_increment(aborted_threads,&LOCK_status);
+ status_var_increment(thd->status_var.lost_connections);
+ }
+
+ if (likely(!thd->killed) && (net->error && net->vio != 0))
+ thd->print_aborted_warning(1, thd->get_stmt_da()->is_error()
+ ? thd->get_stmt_da()->message() : ER_THD(thd, ER_UNKNOWN_ERROR));
+}
+
+
+/*
+ Initialize THD to handle queries
+*/
+
+void prepare_new_connection_state(THD* thd)
+{
+ Security_context *sctx= thd->security_ctx;
+
+ if (thd->client_capabilities & CLIENT_COMPRESS)
+ thd->net.compress=1; // Use compression
+
+ /*
+ Much of this is duplicated in create_embedded_thd() for the
+ embedded server library.
+ TODO: refactor this to avoid code duplication there
+ */
+ thd->proc_info= 0;
+ thd->set_command(COM_SLEEP);
+ thd->init_for_queries();
+
+ if (opt_init_connect.length &&
+ !(sctx->master_access & PRIV_IGNORE_INIT_CONNECT))
+ {
+ execute_init_command(thd, &opt_init_connect, &LOCK_sys_init_connect);
+ if (unlikely(thd->is_error()))
+ {
+ Host_errors errors;
+ thd->set_killed(KILL_CONNECTION);
+ thd->print_aborted_warning(0, "init_connect command failed");
+ sql_print_warning("%s", thd->get_stmt_da()->message());
+
+ /*
+ now let client to send its first command,
+ to be able to send the error back
+ */
+ NET *net= &thd->net;
+ thd->lex->current_select= 0;
+ my_net_set_read_timeout(net, thd->variables.net_wait_timeout);
+ thd->clear_error();
+ net_new_transaction(net);
+ ulong packet_length= my_net_read(net);
+ /*
+ If my_net_read() failed, my_error() has been already called,
+ and the main Diagnostics Area contains an error condition.
+ */
+ if (packet_length != packet_error)
+ my_error(ER_NEW_ABORTING_CONNECTION, MYF(0),
+ thd->thread_id,
+ thd->db.str ? thd->db.str : "unconnected",
+ sctx->user ? sctx->user : "unauthenticated",
+ sctx->host_or_ip, "init_connect command failed");
+ thd->server_status&= ~SERVER_STATUS_CLEAR_SET;
+ thd->protocol->end_statement();
+ thd->killed = KILL_CONNECTION;
+ errors.m_init_connect= 1;
+ inc_host_errors(thd->main_security_ctx.ip, &errors);
+ return;
+ }
+
+ thd->proc_info=0;
+ }
+}
+
+
+/*
+ Thread handler for a connection
+
+ SYNOPSIS
+ handle_one_connection()
+ arg Connection object (THD)
+
+ IMPLEMENTATION
+ This function (normally) does the following:
+ - Initialize thread
+ - Initialize THD to be used with this thread
+ - Authenticate user
+ - Execute all queries sent on the connection
+ - Take connection down
+ - End thread / Handle next connection using thread from thread cache
+*/
+
+pthread_handler_t handle_one_connection(void *arg)
+{
+ CONNECT *connect= (CONNECT*) arg;
+
+ mysql_thread_set_psi_id(connect->thread_id);
+
+ if (init_new_connection_handler_thread())
+ connect->close_with_error(0, 0, ER_OUT_OF_RESOURCES);
+ else
+ do_handle_one_connection(connect, true);
+
+ DBUG_PRINT("info", ("killing thread"));
+#if defined(HAVE_OPENSSL) && !defined(EMBEDDED_LIBRARY)
+ ERR_remove_state(0);
+#endif
+ my_thread_end();
+ return 0;
+}
+
+bool thd_prepare_connection(THD *thd)
+{
+ bool rc;
+ lex_start(thd);
+ rc= login_connection(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);
+#ifdef WITH_WSREP
+ thd->wsrep_client_thread= true;
+ wsrep_open(thd);
+#endif /* WITH_WSREP */
+ return FALSE;
+}
+
+bool thd_is_connection_alive(THD *thd)
+{
+ NET *net= &thd->net;
+ if (likely(!net->error &&
+ net->vio != 0 &&
+ thd->killed < KILL_CONNECTION))
+ return TRUE;
+ return FALSE;
+}
+
+
+void do_handle_one_connection(CONNECT *connect, bool put_in_cache)
+{
+ ulonglong thr_create_utime= microsecond_interval_timer();
+ THD *thd;
+ if (!(thd= connect->create_thd(NULL)))
+ {
+ connect->close_and_delete();
+ return;
+ }
+
+ DBUG_EXECUTE_IF("CONNECT_wait",
+ {
+ extern MYSQL_SOCKET unix_sock;
+ DBUG_ASSERT(unix_sock.fd >= 0);
+ while (unix_sock.fd >= 0)
+ my_sleep(1000);
+ });
+
+ /*
+ If a thread was created to handle this connection:
+ increment slow_launch_threads counter if it took more than
+ slow_launch_time seconds to create the thread.
+ */
+
+ if (connect->prior_thr_create_utime)
+ {
+ ulong launch_time= (ulong) (thr_create_utime -
+ connect->prior_thr_create_utime);
+ if (launch_time >= slow_launch_time*1000000L)
+ statistic_increment(slow_launch_threads, &LOCK_status);
+ }
+
+ server_threads.insert(thd); // Make THD visible in show processlist
+
+ delete connect; // must be after server_threads.insert, see close_connections()
+
+ thd->thr_create_utime= thr_create_utime;
+ /* We need to set this because of time_out_user_resource_limits */
+ thd->start_utime= thr_create_utime;
+
+ /*
+ handle_one_connection() is normally the only way a thread would
+ start and would always be on the very high end of the stack ,
+ therefore, the thread stack always starts at the address of the
+ first local variable of handle_one_connection, which is thd. We
+ need to know the start of the stack so that we could check for
+ stack overruns.
+ */
+ thd->thread_stack= (char*) &thd;
+ setup_connection_thread_globals(thd);
+
+ for (;;)
+ {
+ bool create_user= TRUE;
+
+ mysql_socket_set_thread_owner(thd->net.vio->mysql_socket);
+ if (thd_prepare_connection(thd))
+ {
+ create_user= FALSE;
+ goto end_thread;
+ }
+
+ while (thd_is_connection_alive(thd))
+ {
+ if (mysql_audit_release_required(thd))
+ mysql_audit_release(thd);
+ if (do_command(thd))
+ break;
+ }
+ end_connection(thd);
+
+end_thread:
+ close_connection(thd);
+
+ if (thd->userstat_running)
+ update_global_user_stats(thd, create_user, time(NULL));
+
+ unlink_thd(thd);
+ if (IF_WSREP(thd->wsrep_applier, false) || !put_in_cache ||
+ !(connect= thread_cache.park()))
+ break;
+
+ /* Create new instrumentation for the new THD job */
+ PSI_CALL_set_thread(PSI_CALL_new_thread(key_thread_one_connection, thd,
+ thd->thread_id));
+
+ if (!(connect->create_thd(thd)))
+ {
+ /* Out of resources. Free thread to get more resources */
+ connect->close_and_delete();
+ break;
+ }
+ delete connect;
+
+ /*
+ We have to call store_globals to update mysys_var->id and lock_info
+ with the new thread_id
+ */
+ thd->store_globals();
+
+ /* reset abort flag for the thread */
+ thd->mysys_var->abort= 0;
+ thd->thr_create_utime= microsecond_interval_timer();
+ thd->start_utime= thd->thr_create_utime;
+
+ server_threads.insert(thd);
+ }
+ delete thd;
+}
+#endif /* EMBEDDED_LIBRARY */
+
+
+/* Handling of CONNECT objects */
+
+/*
+ Close connection without error and delete the connect object
+ This and close_with_error are only called if we didn't manage to
+ create a new thd object.
+*/
+
+void CONNECT::close_and_delete()
+{
+ DBUG_ENTER("close_and_delete");
+
+#if _WIN32
+ if (vio_type == VIO_TYPE_NAMEDPIPE)
+ CloseHandle(pipe);
+ else
+#endif
+ if (vio_type != VIO_CLOSED)
+ mysql_socket_close(sock);
+ vio_type= VIO_CLOSED;
+
+ --*scheduler->connection_count;
+ statistic_increment(connection_errors_internal, &LOCK_status);
+ statistic_increment(aborted_connects,&LOCK_status);
+
+ delete this;
+ DBUG_VOID_RETURN;
+}
+
+/*
+ Close a connection with a possible error to the end user
+ Alse deletes the connection object, like close_and_delete()
+*/
+
+void CONNECT::close_with_error(uint sql_errno,
+ const char *message, uint close_error)
+{
+ THD *thd= create_thd(NULL);
+ if (thd)
+ {
+ if (sql_errno)
+ thd->protocol->net_send_error(thd, sql_errno, message, NULL);
+ close_connection(thd, close_error);
+ delete thd;
+ set_current_thd(0);
+ }
+ close_and_delete();
+}
+
+
+/* Reuse or create a THD based on a CONNECT object */
+
+THD *CONNECT::create_thd(THD *thd)
+{
+ bool res, thd_reused= thd != 0;
+ Vio *vio;
+ DBUG_ENTER("create_thd");
+
+ DBUG_EXECUTE_IF("simulate_failed_connection_2", DBUG_RETURN(0); );
+
+ if (thd)
+ {
+ /* reuse old thd */
+ thd->reset_for_reuse();
+ /*
+ reset tread_id's, but not thread_dbug_id's as the later isn't allowed
+ to change as there is already structures in thd marked with the old
+ value.
+ */
+ thd->thread_id= thd->variables.pseudo_thread_id= thread_id;
+ }
+ else if (!(thd= new THD(thread_id)))
+ DBUG_RETURN(0);
+
+#if _WIN32
+ if (vio_type == VIO_TYPE_NAMEDPIPE)
+ vio= vio_new_win32pipe(pipe);
+ else
+#endif
+ vio= mysql_socket_vio_new(sock, vio_type, vio_type == VIO_TYPE_SOCKET ?
+ VIO_LOCALHOST : 0);
+ if (!vio)
+ {
+ if (!thd_reused)
+ delete thd;
+ DBUG_RETURN(0);
+ }
+
+ set_current_thd(thd);
+ res= my_net_init(&thd->net, vio, thd, MYF(MY_THREAD_SPECIFIC));
+ vio_type= VIO_CLOSED; // Vio now handled by thd
+
+ if (unlikely(res || thd->is_error()))
+ {
+ if (!thd_reused)
+ delete thd;
+ set_current_thd(0);
+ DBUG_RETURN(0);
+ }
+
+ init_net_server_extension(thd);
+
+ thd->security_ctx->host= thd->net.vio->type == VIO_TYPE_NAMEDPIPE ||
+ thd->net.vio->type == VIO_TYPE_SOCKET ?
+ my_localhost : 0;
+
+ thd->scheduler= scheduler;
+ thd->real_id= pthread_self(); /* Duplicates THD::store_globals() setting. */
+
+ /* Attach PSI instrumentation to the new THD */
+
+ PSI_thread *psi= PSI_CALL_get_thread();
+ PSI_CALL_set_thread_os_id(psi);
+ PSI_CALL_set_thread_THD(psi, thd);
+ PSI_CALL_set_thread_id(psi, thd->thread_id);
+ thd->set_psi(psi);
+
+ DBUG_RETURN(thd);
+}