Merged in WL#5138

include/atomic/x86-gcc.h

@@ -42,15 +42,37 @@
 #endif
 
 #ifndef MY_ATOMIC_NO_XADD
-#define make_atomic_add_body(S)					\
-  asm volatile (LOCK_prefix "; xadd %0, %1;" : "+r" (v) , "+m" (*a))
+#define make_atomic_add_body(S)         make_atomic_add_body ## S
+#define make_atomic_cas_body(S)         make_atomic_cas_body ## S
 #endif
-#define make_atomic_fas_body(S)				\
-  asm volatile ("xchg %0, %1;" : "+q" (v) , "+m" (*a))
-#define make_atomic_cas_body(S)					\
+
+#define make_atomic_add_body32                                  \
+  asm volatile (LOCK_prefix "; xadd %0, %1;" : "+r" (v) , "+m" (*a))
+
+#define make_atomic_cas_body32                                  \
   asm volatile (LOCK_prefix "; cmpxchg %3, %0; setz %2;"	\
                : "+m" (*a), "+a" (*cmp), "=q" (ret): "r" (set))
 
+#define make_atomic_cas_bodyptr make_atomic_cas_body32
+
+#ifndef __x86_64__
+#define make_atomic_add_body64 make_atomic_add_body32
+#define make_atomic_cas_body64 make_atomic_cas_body32
+#else
+#define make_atomic_add_body64                                  \
+  int64 tmp=*a;                                                 \
+  while (!my_atomic_cas64(a, &tmp, tmp+v));                     \
+  v=tmp;
+#define make_atomic_cas_body64                                  \
+  int32 ebx=(set & 0xFFFFFFFF), ecx=(set >> 32);                \
+  asm volatile (LOCK_prefix "; cmpxchg8b %0; setz %2;"         \
+               : "+m" (*a), "+A" (*cmp), "=q" (ret)             \
+               :"b" (ebx), "c" (ecx))
+#endif
+
+#define make_atomic_fas_body(S)                                        \
+  asm volatile ("xchg %0, %1;" : "+r" (v) , "+m" (*a))
+
 #ifdef MY_ATOMIC_MODE_DUMMY
 #define make_atomic_load_body(S)   ret=*a
 #define make_atomic_store_body(S)  *a=v
@@ -66,5 +88,4 @@
 #define make_atomic_store_body(S)				\
   asm volatile ("; xchg %0, %1;" : "+m" (*a), "+r" (v))
 #endif
-
 #endif /* ATOMIC_X86_GCC_INCLUDED */

include/my_atomic.h

@@ -37,7 +37,7 @@
   my_atomic_store#(&var, what)
     store 'what' in *var
 
-  '#' is substituted by a size suffix - 8, 16, 32, or ptr
+  '#' is substituted by a size suffix - 8, 16, 32, 64, or ptr
   (e.g. my_atomic_add8, my_atomic_fas32, my_atomic_casptr).
 
   NOTE This operations are not always atomic, so they always must be
@@ -49,7 +49,7 @@
   On architectures where these operations are really atomic, rwlocks will
   be optimized away.
   8- and 16-bit atomics aren't implemented for windows (see generic-msvc.h),
-  but can be added, if necessary.
+  but can be added, if necessary. 
 */
 
 #ifndef my_atomic_rwlock_init
@@ -129,6 +129,7 @@
 make_transparent_unions(8)
 make_transparent_unions(16)
 make_transparent_unions(32)
+make_transparent_unions(64)
 make_transparent_unions(ptr)
 #undef uintptr
 #undef make_transparent_unions
@@ -140,10 +141,12 @@ make_transparent_unions(ptr)
 #define U_8    int8
 #define U_16   int16
 #define U_32   int32
+#define U_64   int64
 #define U_ptr  intptr
 #define Uv_8   int8
 #define Uv_16  int16
 #define Uv_32  int32
+#define Uv_64  int64
 #define Uv_ptr intptr
 #define U_a    volatile *a
 #define U_cmp  *cmp
@@ -217,6 +220,7 @@ make_atomic_cas(8)
 make_atomic_cas(16)
 #endif
 make_atomic_cas(32)
+make_atomic_cas(64)
 make_atomic_cas(ptr)
 
 #ifdef MY_ATOMIC_HAS_8_16
@@ -224,12 +228,14 @@ make_atomic_add(8)
 make_atomic_add(16)
 #endif
 make_atomic_add(32)
+make_atomic_add(64)
 
 #ifdef MY_ATOMIC_HAS_8_16
 make_atomic_load(8)
 make_atomic_load(16)
 #endif
 make_atomic_load(32)
+make_atomic_load(64)
 make_atomic_load(ptr)
 
 #ifdef MY_ATOMIC_HAS_8_16
@@ -237,6 +243,7 @@ make_atomic_fas(8)
 make_atomic_fas(16)
 #endif
 make_atomic_fas(32)
+make_atomic_fas(64)
 make_atomic_fas(ptr)
 
 #ifdef MY_ATOMIC_HAS_8_16
@@ -244,6 +251,7 @@ make_atomic_store(8)
 make_atomic_store(16)
 #endif
 make_atomic_store(32)
+make_atomic_store(64)
 make_atomic_store(ptr)
 
 #ifdef _atomic_h_cleanup_
@@ -254,10 +262,12 @@ make_atomic_store(ptr)
 #undef U_8
 #undef U_16
 #undef U_32
+#undef U_64
 #undef U_ptr
 #undef Uv_8
 #undef Uv_16
 #undef Uv_32
+#undef Uv_64
 #undef Uv_ptr
 #undef a
 #undef cmp

include/my_global.h

@@ -866,6 +866,8 @@ typedef SOCKET_SIZE_TYPE size_socket;
 #endif
 #endif /* defined (HAVE_LONG_LONG) && !defined(ULONGLONG_MAX)*/
 
+#define INT_MIN64       (~0x7FFFFFFFFFFFFFFFLL)
+#define INT_MAX64       0x7FFFFFFFFFFFFFFFLL
 #define INT_MIN32       (~0x7FFFFFFFL)
 #define INT_MAX32       0x7FFFFFFFL
 #define UINT_MAX32      0xFFFFFFFFL

sql/event_scheduler.cc

@@ -133,9 +133,8 @@ post_init_event_thread(THD *thd)
   pthread_mutex_lock(&LOCK_thread_count);
   threads.append(thd);
   thread_count++;
-  thread_running++;
+  inc_thread_running();
   pthread_mutex_unlock(&LOCK_thread_count);
-
   return FALSE;
 }
 
@@ -157,7 +156,7 @@ deinit_event_thread(THD *thd)
   DBUG_PRINT("exit", ("Event thread finishing"));
   pthread_mutex_lock(&LOCK_thread_count);
   thread_count--;
-  thread_running--;
+  dec_thread_running();
   delete thd;
   pthread_cond_broadcast(&COND_thread_count);
   pthread_mutex_unlock(&LOCK_thread_count);
@@ -418,7 +417,7 @@ Event_scheduler::start()
     net_end(&new_thd->net);
     pthread_mutex_lock(&LOCK_thread_count);
     thread_count--;
-    thread_running--;
+    dec_thread_running();
     delete new_thd;
     pthread_cond_broadcast(&COND_thread_count);
     pthread_mutex_unlock(&LOCK_thread_count);
@@ -551,7 +550,7 @@ error:
     net_end(&new_thd->net);
     pthread_mutex_lock(&LOCK_thread_count);
     thread_count--;
-    thread_running--;
+    dec_thread_running();
     delete new_thd;
     pthread_cond_broadcast(&COND_thread_count);
     pthread_mutex_unlock(&LOCK_thread_count);

sql/log_event.cc

@@ -3055,10 +3055,7 @@ int Query_log_event::do_apply_event(Relay_log_info const *rli,
       rpl_filter->db_ok(thd->db))
   {
     thd->set_time((time_t)when);
-    thd->set_query((char*)query_arg, q_len_arg);
-    pthread_mutex_lock(&LOCK_thread_count);
-    thd->query_id = next_query_id();
-    pthread_mutex_unlock(&LOCK_thread_count);
+    thd->set_query_and_id((char*)query_arg, q_len_arg, next_query_id());
     thd->variables.pseudo_thread_id= thread_id;		// for temp tables
     DBUG_PRINT("query",("%s", thd->query()));
 
@@ -4580,9 +4577,7 @@ 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);
-    pthread_mutex_lock(&LOCK_thread_count);
-    thd->query_id = next_query_id();
-    pthread_mutex_unlock(&LOCK_thread_count);
+    thd->set_query_id(next_query_id());
     thd->warning_info->opt_clear_warning_info(thd->query_id);
 
     TABLE_LIST tables;
@@ -8071,9 +8066,7 @@ 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);
-  thd->query_id= next_query_id();
-  pthread_mutex_unlock(&LOCK_thread_count);
+  thd->set_query_id(next_query_id());
 
   if (!(memory= my_multi_malloc(MYF(MY_WME),
                                 &table_list, (uint) sizeof(RPL_TABLE_LIST),

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,60 @@ 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;
+extern my_atomic_rwlock_t thread_running_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;
+  my_atomic_rwlock_wrlock(&global_query_id_lock);
+  id= my_atomic_add64(&global_query_id, 1);
+  my_atomic_rwlock_wrunlock(&global_query_id_lock);
+  return (id+1);
+}
+
+inline query_id_t get_query_id()
+{
+  query_id_t id;
+  my_atomic_rwlock_wrlock(&global_query_id_lock);
+  id= my_atomic_load64(&global_query_id);
+  my_atomic_rwlock_wrunlock(&global_query_id_lock);
+  return id;
+}
+
+inline int32
+inc_thread_running()
+{
+  int32 num_thread_running;
+  my_atomic_rwlock_wrlock(&thread_running_lock);
+  num_thread_running= my_atomic_add32(&thread_running, 1);
+  my_atomic_rwlock_wrunlock(&thread_running_lock);
+  return (num_thread_running+1);
+}
+
+inline int32
+dec_thread_running()
+{
+  int32 num_thread_running;
+  my_atomic_rwlock_wrlock(&thread_running_lock);
+  num_thread_running= my_atomic_add32(&thread_running, -1);
+  my_atomic_rwlock_wrunlock(&thread_running_lock);
+  return (num_thread_running-1);
+}
+
+inline int32
+get_thread_running()
+{
+  int32 num_thread_running;
+  my_atomic_rwlock_wrlock(&thread_running_lock);
+  num_thread_running= my_atomic_load32(&thread_running);
+  my_atomic_rwlock_wrunlock(&thread_running_lock);
+  return num_thread_running;
+}
 
 /* useful constants */
 extern MYSQL_PLUGIN_IMPORT const key_map key_map_empty;
@@ -1930,7 +1980,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;

sql/mysqld.cc

@@ -527,7 +527,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;
@@ -543,6 +544,8 @@ 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;
+my_atomic_rwlock_t thread_running_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;
@@ -1376,6 +1379,8 @@ 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);
+  my_atomic_rwlock_destroy(&thread_running_lock);
   (void) pthread_mutex_lock(&LOCK_thread_count);
   DBUG_PRINT("quit", ("got thread count lock"));
   ready_to_exit=1;
@@ -7783,6 +7788,8 @@ 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);
+  my_atomic_rwlock_init(&thread_running_lock);
   strmov(server_version, MYSQL_SERVER_VERSION);
   myisam_recover_options_str= sql_mode_str= "OFF";
   myisam_stats_method_str= "nulls_unequal";

sql/sp_head.cc

@@ -1338,7 +1338,7 @@ sp_head::execute(THD *thd)
   /* To avoid wiping out thd->change_list on old_change_list destruction */
   old_change_list.empty();
   thd->lex= old_lex;
-  thd->query_id= old_query_id;
+  thd->set_query_id(old_query_id);
   DBUG_ASSERT(!thd->derived_tables);
   thd->derived_tables= old_derived_tables;
   thd->variables.sql_mode= save_sql_mode;
@@ -2736,9 +2736,7 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
   */
   thd->lex= m_lex;
 
-  pthread_mutex_lock(&LOCK_thread_count);
-  thd->query_id= next_query_id();
-  pthread_mutex_unlock(&LOCK_thread_count);
+  thd->set_query_id(next_query_id());
 
   if (thd->prelocked_mode == NON_PRELOCKED)
   {

sql/sql_class.cc

@@ -3269,6 +3269,26 @@ void THD::set_query(char *query_arg, uint32 query_length_arg)
   pthread_mutex_unlock(&LOCK_thd_data);
 }
 
+/** Assign a new value to thd->query and thd->query_id.  */
+
+void THD::set_query_and_id(char *query_arg, uint32 query_length_arg,
+                           query_id_t new_query_id)
+{
+  pthread_mutex_lock(&LOCK_thd_data);
+  set_query_inner(query_arg, query_length_arg);
+  query_id= new_query_id;
+  pthread_mutex_unlock(&LOCK_thd_data);
+}
+
+/** Assign a new value to thd->query_id.  */
+
+void THD::set_query_id(query_id_t new_query_id)
+{
+  pthread_mutex_lock(&LOCK_thd_data);
+  query_id= new_query_id;
+  pthread_mutex_unlock(&LOCK_thd_data);
+}
+
 
 /**
   Mark transaction to rollback and mark error as fatal to a sub-statement.

sql/sql_class.h

@@ -2314,10 +2314,13 @@ public:
   virtual void set_statement(Statement *stmt);
 
   /**
-    Assign a new value to thd->query.
+    Assign a new value to thd->query and thd->query_id.
     Protected with LOCK_thd_data mutex.
   */
   void set_query(char *query_arg, uint32 query_length_arg);
+  void set_query_and_id(char *query_arg, uint32 query_length_arg,
+                        query_id_t new_query_id);
+  void set_query_id(query_id_t new_query_id);
 private:
   /** The current internal error handler for this thread, or NULL. */
   Internal_error_handler *m_internal_handler;

sql/sql_cursor.cc

@@ -438,7 +438,7 @@ Sensitive_cursor::fetch(ulong num_rows)
   thd->derived_tables= derived_tables;
   thd->open_tables= open_tables;
   thd->lock= lock;
-  thd->query_id= query_id;
+  thd->set_query_id(query_id);
   thd->change_list= change_list;
   /* save references to memory allocated during fetch */
   thd->set_n_backup_active_arena(this, &backup_arena);

sql/sql_parse.cc

@@ -484,7 +484,7 @@ static void handle_bootstrap_impl(THD *thd)
     query= (char *) thd->memdup_w_gap(buff, length + 1,
                                       thd->db_length + 1 +
                                       QUERY_CACHE_FLAGS_SIZE);
-    thd->set_query(query, length);
+    thd->set_query_and_id(query, length, next_query_id());
     DBUG_PRINT("query",("%-.4096s",thd->query()));
 #if defined(ENABLED_PROFILING)
     thd->profiling.start_new_query();
@@ -495,7 +495,6 @@ 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.
     */
-    thd->query_id=next_query_id();
     thd->set_time();
     mysql_parse(thd, thd->query(), length, & found_semicolon);
     close_thread_tables(thd);			// Free tables
@@ -983,29 +982,29 @@ 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();
-  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();
+  {
+    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->set_query_id(query_id);
   }
-
-  thread_running++;
+  inc_thread_running();
   /* TODO: set thd->lex->sql_command to SQLCOM_END here */
-  pthread_mutex_unlock(&LOCK_thread_count);
 
   /**
     Clear the set of flags that are expected to be cleared at the
@@ -1267,16 +1266,13 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
                         thd->security_ctx->priv_user,
                         (char *) thd->security_ctx->host_or_ip);
 
-      thd->set_query(beginning_of_next_stmt, length);
-      pthread_mutex_lock(&LOCK_thread_count);
+      thd->set_query_and_id(beginning_of_next_stmt, length, next_query_id());
       /*
         Count each statement from the client.
       */
       statistic_increment(thd->status_var.questions, &LOCK_status);
-      thd->query_id= next_query_id();
       thd->set_time(); /* Reset the query start time. */
       /* TODO: set thd->lex->sql_command to SQLCOM_END here */
-      pthread_mutex_unlock(&LOCK_thread_count);
       mysql_parse(thd, beginning_of_next_stmt, length, &end_of_stmt);
     }
 
@@ -1590,9 +1586,7 @@ 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;
-  pthread_mutex_lock(&LOCK_thread_count); // For process list
-  thread_running--;
-  pthread_mutex_unlock(&LOCK_thread_count);
+  dec_thread_running();
   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));

unittest/mysys/my_atomic-t.c

@@ -48,6 +48,34 @@ pthread_handler_t test_atomic_add(void *arg)
   return 0;
 }
 
+volatile int64 a64;
+/* add and sub a random number in a loop. Must get 0 at the end */
+pthread_handler_t test_atomic_add64(void *arg)
+{
+  int    m= (*(int *)arg)/2;
+  GCC_BUG_WORKAROUND int64 x;
+  for (x= ((int64)(intptr)(&m)); m ; m--)
+  {
+    x= (x*m+0xfdecba987654321LL) & INT_MAX64;
+    my_atomic_rwlock_wrlock(&rwl);
+    my_atomic_add64(&a64, x);
+    my_atomic_rwlock_wrunlock(&rwl);
+
+    my_atomic_rwlock_wrlock(&rwl);
+    my_atomic_add64(&a64, -x);
+    my_atomic_rwlock_wrunlock(&rwl);
+  }
+  pthread_mutex_lock(&mutex);
+  if (!--running_threads)
+  {
+    bad= (a64 != 0);
+    pthread_cond_signal(&cond);
+  }
+  pthread_mutex_unlock(&mutex);
+  return 0;
+}
+
+
 /*
   1. generate thread number 0..N-1 from b32
   2. add it to bad
@@ -128,7 +156,7 @@ pthread_handler_t test_atomic_cas(void *arg)
 
 void do_tests()
 {
-  plan(4);
+  plan(6);
 
   bad= my_atomic_initialize();
   ok(!bad, "my_atomic_initialize() returned %d", bad);
@@ -142,5 +170,14 @@ void do_tests()
   b32= c32= 0;
   test_concurrently("my_atomic_cas32", test_atomic_cas, THREADS, CYCLES);
 
+  {
+    int64 b=0x1000200030004000LL;
+    a64=0;
+    my_atomic_add64(&a64, b);
+    ok(a64==b, "add64");
+  }
+  a64=0;
+  test_concurrently("my_atomic_add64", test_atomic_add64, THREADS, CYCLES);
+
   my_atomic_rwlock_destroy(&rwl);
 }