WL#5138 merged to mysql-next-mr

6 files changed, 97 insertions, 44 deletions

diff --git a/sql/event_scheduler.cc b/sql/event_scheduler.cc
index ea20270b457..880bccdd67e 100644
--- a/sql/event_scheduler.cc
+++ b/sql/event_scheduler.cc
@@ -133,9 +133,10 @@ post_init_event_thread(THD *thd)
   pthread_mutex_lock(&LOCK_thread_count);
   threads.append(thd);
   thread_count++;
-  thread_running++;
   pthread_mutex_unlock(&LOCK_thread_count);
-
+  my_atomic_rwlock_wrlock(&global_query_id_lock);
+  inc_thread_running();
+  my_atomic_rwlock_wrunlock(&global_query_id_lock);
   return FALSE;
 }
 
@@ -157,10 +158,12 @@ deinit_event_thread(THD *thd)
   DBUG_PRINT("exit", ("Event thread finishing"));
   pthread_mutex_lock(&LOCK_thread_count);
   thread_count--;
-  thread_running--;
   delete thd;
   pthread_cond_broadcast(&COND_thread_count);
   pthread_mutex_unlock(&LOCK_thread_count);
+  my_atomic_rwlock_wrlock(&global_query_id_lock);
+  dec_thread_running();
+  my_atomic_rwlock_wrunlock(&global_query_id_lock);
 }
 
 
@@ -418,10 +421,12 @@ Event_scheduler::start()
     net_end(&new_thd->net);
     pthread_mutex_lock(&LOCK_thread_count);
     thread_count--;
-    thread_running--;
     delete new_thd;
     pthread_cond_broadcast(&COND_thread_count);
     pthread_mutex_unlock(&LOCK_thread_count);
+    my_atomic_rwlock_wrlock(&global_query_id_lock);
+    dec_thread_running();
+    my_atomic_rwlock_wrunlock(&global_query_id_lock);
   }
 end:
   UNLOCK_DATA();
@@ -551,10 +556,12 @@ error:
     net_end(&new_thd->net);
     pthread_mutex_lock(&LOCK_thread_count);
     thread_count--;
-    thread_running--;
     delete new_thd;
     pthread_cond_broadcast(&COND_thread_count);
     pthread_mutex_unlock(&LOCK_thread_count);
+    my_atomic_rwlock_wrlock(&global_query_id_lock);
+    dec_thread_running();
+    my_atomic_rwlock_wrunlock(&global_query_id_lock);
   }
   delete event_name;
   DBUG_RETURN(TRUE);
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 3a54717a45f..31908a818be 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3056,9 +3056,9 @@ int Query_log_event::do_apply_event(Relay_log_info const *rli,
   {
     thd->set_time((time_t)when);
     thd->set_query((char*)query_arg, q_len_arg);
-    VOID(pthread_mutex_lock(&LOCK_thread_count));
+    my_atomic_rwlock_wrlock(&global_query_id_lock);
     thd->query_id = next_query_id();
-    VOID(pthread_mutex_unlock(&LOCK_thread_count));
+    my_atomic_rwlock_wrunlock(&global_query_id_lock);
     thd->variables.pseudo_thread_id= thread_id;		// for temp tables
     DBUG_PRINT("query",("%s", thd->query()));
 
@@ -4581,9 +4581,9 @@ int Load_log_event::do_apply_event(NET* net, Relay_log_info const *rli,
   if (rpl_filter->db_ok(thd->db))
   {
     thd->set_time((time_t)when);
-    VOID(pthread_mutex_lock(&LOCK_thread_count));
+    my_atomic_rwlock_wrlock(&global_query_id_lock);
     thd->query_id = next_query_id();
-    VOID(pthread_mutex_unlock(&LOCK_thread_count));
+    my_atomic_rwlock_wrunlock(&global_query_id_lock);
     thd->warning_info->opt_clear_warning_info(thd->query_id);
 
     TABLE_LIST tables;
@@ -8072,9 +8072,9 @@ int Table_map_log_event::do_apply_event(Relay_log_info const *rli)
   DBUG_ASSERT(rli->sql_thd == thd);
 
   /* Step the query id to mark what columns that are actually used. */
-  pthread_mutex_lock(&LOCK_thread_count);
+  my_atomic_rwlock_wrlock(&global_query_id_lock);
   thd->query_id= next_query_id();
-  pthread_mutex_unlock(&LOCK_thread_count);
+  my_atomic_rwlock_wrunlock(&global_query_id_lock);
 
   if (!(memory= my_multi_malloc(MYF(MY_WME),
                                 &table_list, (uint) sizeof(RPL_TABLE_LIST),
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 6d97bfe3f16..0caf3197fb8 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -53,6 +53,7 @@
 #include "sql_array.h"
 #include "sql_plugin.h"
 #include "scheduler.h"
+#include <my_atomic.h>
 
 class Parser_state;
 
@@ -85,11 +86,49 @@ typedef ulong nesting_map;  /* Used for flags of nesting constructs */
 typedef ulonglong nested_join_map;
 
 /* query_id */
-typedef ulonglong query_id_t;
+typedef int64 query_id_t;
 extern query_id_t global_query_id;
+extern int32 thread_running;
+extern my_atomic_rwlock_t global_query_id_lock;
 
 /* increment query_id and return it.  */
-inline query_id_t next_query_id() { return global_query_id++; }
+inline query_id_t next_query_id()
+{
+  query_id_t id;
+  id= my_atomic_add64(&global_query_id, 1);
+  return (id+1);
+}
+
+inline query_id_t get_query_id()
+{
+  query_id_t id;
+  id= my_atomic_load64(&global_query_id);
+  return id;
+}
+
+inline int32
+inc_thread_running()
+{
+  int32 num_thread_running;
+  num_thread_running= my_atomic_add32(&thread_running, 1);
+  return (num_thread_running+1);
+}
+
+inline int32
+dec_thread_running()
+{
+  int32 num_thread_running;
+  num_thread_running= my_atomic_add32(&thread_running, -1);
+  return (num_thread_running-1);
+}
+
+inline int32
+get_thread_running()
+{
+  int32 num_thread_running;
+  num_thread_running= my_atomic_load32(&thread_running);
+  return num_thread_running;
+}
 
 /* useful constants */
 extern MYSQL_PLUGIN_IMPORT const key_map key_map_empty;
@@ -1940,7 +1979,7 @@ extern bool opt_ignore_builtin_innodb;
 extern my_bool opt_character_set_client_handshake;
 extern bool volatile abort_loop, shutdown_in_progress;
 extern bool in_bootstrap;
-extern uint volatile thread_count, thread_running, global_read_lock;
+extern uint volatile thread_count, global_read_lock;
 extern uint connection_count;
 extern my_bool opt_sql_bin_update, opt_safe_user_create, opt_no_mix_types;
 extern my_bool opt_safe_show_db, opt_local_infile, opt_myisam_use_mmap;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index c1b8b62c470..ddc6d53e019 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -531,7 +531,8 @@ uint mysqld_port_timeout;
 uint delay_key_write_options, protocol_version;
 uint lower_case_table_names;
 uint tc_heuristic_recover= 0;
-uint volatile thread_count, thread_running;
+uint volatile thread_count;
+int32 thread_running;
 ulonglong thd_startup_options;
 ulong back_log, connect_timeout, concurrency, server_id;
 ulong table_cache_size, table_def_size;
@@ -547,6 +548,7 @@ ulonglong  max_binlog_cache_size=0;
 ulong query_cache_size=0;
 ulong refresh_version;  /* Increments on each reload */
 query_id_t global_query_id;
+my_atomic_rwlock_t global_query_id_lock;
 ulong aborted_threads, aborted_connects;
 ulong delayed_insert_timeout, delayed_insert_limit, delayed_queue_size;
 ulong delayed_insert_threads, delayed_insert_writes, delayed_rows_in_use;
@@ -1380,6 +1382,7 @@ void clean_up(bool print_message)
   DBUG_PRINT("quit", ("Error messages freed"));
   /* Tell main we are ready */
   logger.cleanup_end();
+  my_atomic_rwlock_destroy(&global_query_id_lock);
   (void) pthread_mutex_lock(&LOCK_thread_count);
   DBUG_PRINT("quit", ("got thread count lock"));
   ready_to_exit=1;
@@ -7795,6 +7798,7 @@ static int mysql_init_variables(void)
   what_to_log= ~ (1L << (uint) COM_TIME);
   refresh_version= 1L;	/* Increments on each reload */
   global_query_id= thread_id= 1L;
+  my_atomic_rwlock_init(&global_query_id_lock);
   strmov(server_version, MYSQL_SERVER_VERSION);
   myisam_recover_options_str= sql_mode_str= "OFF";
   myisam_stats_method_str= "nulls_unequal";
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 1fb8fd257df..d0453c08a00 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2736,9 +2736,9 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
   */
   thd->lex= m_lex;
 
-  VOID(pthread_mutex_lock(&LOCK_thread_count));
+  my_atomic_rwlock_wrlock(&global_query_id_lock);
   thd->query_id= next_query_id();
-  VOID(pthread_mutex_unlock(&LOCK_thread_count));
+  my_atomic_rwlock_wrunlock(&global_query_id_lock);
 
   if (thd->prelocked_mode == NON_PRELOCKED)
   {
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 99fb08abcca..7b5df421785 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -495,7 +495,9 @@ static void handle_bootstrap_impl(THD *thd)
       We don't need to obtain LOCK_thread_count here because in bootstrap
       mode we have only one thread.
     */
+    my_atomic_rwlock_wrlock(&global_query_id_lock);
     thd->query_id=next_query_id();
+    my_atomic_rwlock_wrunlock(&global_query_id_lock);
     thd->set_time();
     mysql_parse(thd, thd->query(), length, & found_semicolon);
     close_thread_tables(thd);			// Free tables
@@ -989,29 +991,30 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
   thd->enable_slow_log= TRUE;
   thd->lex->sql_command= SQLCOM_END; /* to avoid confusing VIEW detectors */
   thd->set_time();
-  VOID(pthread_mutex_lock(&LOCK_thread_count));
-  thd->query_id= global_query_id;
-
-  switch( command ) {
-  /* Ignore these statements. */
-  case COM_STATISTICS:
-  case COM_PING:
-    break;
-  /* Only increase id on these statements but don't count them. */
-  case COM_STMT_PREPARE: 
-  case COM_STMT_CLOSE:
-  case COM_STMT_RESET:
-    next_query_id();
-    break;
-  /* Increase id and count all other statements. */
-  default:
-    statistic_increment(thd->status_var.questions, &LOCK_status);
-    next_query_id();
+  my_atomic_rwlock_wrlock(&global_query_id_lock);
+  {
+    query_id_t query_id;
+    switch( command ) {
+    /* Ignore these statements. */
+    case COM_STATISTICS:
+    case COM_PING:
+      query_id= get_query_id();
+      break;
+    /* Only increase id on these statements but don't count them. */
+    case COM_STMT_PREPARE: 
+    case COM_STMT_CLOSE:
+    case COM_STMT_RESET:
+      query_id= next_query_id() - 1;
+      break;
+    /* Increase id and count all other statements. */
+    default:
+      statistic_increment(thd->status_var.questions, &LOCK_status);
+      query_id= next_query_id() - 1;
+    }
+    thd->query_id= query_id;
   }
-
-  thread_running++;
-  /* TODO: set thd->lex->sql_command to SQLCOM_END here */
-  VOID(pthread_mutex_unlock(&LOCK_thread_count));
+  inc_thread_running();
+  my_atomic_rwlock_wrunlock(&global_query_id_lock);
 
   /**
     Clear the set of flags that are expected to be cleared at the
@@ -1277,15 +1280,15 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
                         (char *) thd->security_ctx->host_or_ip);
 
       thd->set_query(beginning_of_next_stmt, length);
-      VOID(pthread_mutex_lock(&LOCK_thread_count));
       /*
         Count each statement from the client.
       */
       statistic_increment(thd->status_var.questions, &LOCK_status);
+      my_atomic_rwlock_wrlock(&global_query_id_lock);
       thd->query_id= next_query_id();
+      my_atomic_rwlock_wrunlock(&global_query_id_lock);
       thd->set_time(); /* Reset the query start time. */
       /* TODO: set thd->lex->sql_command to SQLCOM_END here */
-      VOID(pthread_mutex_unlock(&LOCK_thread_count));
       mysql_parse(thd, beginning_of_next_stmt, length, &end_of_stmt);
     }
 
@@ -1601,9 +1604,9 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
   thd_proc_info(thd, "cleaning up");
   thd->set_query(NULL, 0);
   thd->command=COM_SLEEP;
-  VOID(pthread_mutex_lock(&LOCK_thread_count)); // For process list
-  thread_running--;
-  VOID(pthread_mutex_unlock(&LOCK_thread_count));
+  my_atomic_rwlock_wrlock(&global_query_id_lock);
+  dec_thread_running();
+  my_atomic_rwlock_wrunlock(&global_query_id_lock);
   thd_proc_info(thd, 0);
   thd->packet.shrink(thd->variables.net_buffer_length);	// Reclaim some memory
   free_root(thd->mem_root,MYF(MY_KEEP_PREALLOC));