Merge branch '10.0' into 10.1

6 files changed, 130 insertions, 64 deletions

diff --git a/storage/perfschema/pfs.cc b/storage/perfschema/pfs.cc
index dd61df5f861..ad3faeea618 100644
--- a/storage/perfschema/pfs.cc
+++ b/storage/perfschema/pfs.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2008, 2015, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2008, 2016, 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
@@ -2560,10 +2560,7 @@ start_table_io_wait_v1(PSI_table_locker_state *state,
   if (! pfs_table->m_io_enabled)
     return NULL;
 
-  PFS_thread *pfs_thread= pfs_table->m_thread_owner;
-
-  DBUG_ASSERT(pfs_thread ==
-              my_pthread_getspecific_ptr(PFS_thread*, THR_PFS));
+  PFS_thread *pfs_thread= my_pthread_getspecific_ptr(PFS_thread*, THR_PFS);
 
   register uint flags;
   ulonglong timer_start= 0;
@@ -2666,7 +2663,7 @@ start_table_lock_wait_v1(PSI_table_locker_state *state,
   if (! pfs_table->m_lock_enabled)
     return NULL;
 
-  PFS_thread *pfs_thread= pfs_table->m_thread_owner;
+  PFS_thread *pfs_thread= my_pthread_getspecific_ptr(PFS_thread*, THR_PFS);
 
   PFS_TL_LOCK_TYPE lock_type;
 
@@ -3068,7 +3065,12 @@ start_socket_wait_v1(PSI_socket_locker_state *state,
 
   if (flag_thread_instrumentation)
   {
-    PFS_thread *pfs_thread= pfs_socket->m_thread_owner;
+    /*
+       Do not use pfs_socket->m_thread_owner here,
+       as different threads may use concurrently the same socket,
+       for example during a KILL.
+    */
+    PFS_thread *pfs_thread= my_pthread_getspecific_ptr(PFS_thread*, THR_PFS);
 
     if (unlikely(pfs_thread == NULL))
       return NULL;
@@ -3436,6 +3438,8 @@ static void end_idle_wait_v1(PSI_idle_locker* locker)
       if (flag_events_waits_history_long)
         insert_events_waits_history_long(wait);
       thread->m_events_waits_current--;
+
+      DBUG_ASSERT(wait == thread->m_events_waits_current);
     }
   }
 
@@ -3517,6 +3521,8 @@ static void end_mutex_wait_v1(PSI_mutex_locker* locker, int rc)
       if (flag_events_waits_history_long)
         insert_events_waits_history_long(wait);
       thread->m_events_waits_current--;
+
+      DBUG_ASSERT(wait == thread->m_events_waits_current);
     }
   }
 }
@@ -3596,6 +3602,8 @@ static void end_rwlock_rdwait_v1(PSI_rwlock_locker* locker, int rc)
       if (flag_events_waits_history_long)
         insert_events_waits_history_long(wait);
       thread->m_events_waits_current--;
+
+      DBUG_ASSERT(wait == thread->m_events_waits_current);
     }
   }
 }
@@ -3668,6 +3676,8 @@ static void end_rwlock_wrwait_v1(PSI_rwlock_locker* locker, int rc)
       if (flag_events_waits_history_long)
         insert_events_waits_history_long(wait);
       thread->m_events_waits_current--;
+
+      DBUG_ASSERT(wait == thread->m_events_waits_current);
     }
   }
 }
@@ -3732,6 +3742,8 @@ static void end_cond_wait_v1(PSI_cond_locker* locker, int rc)
       if (flag_events_waits_history_long)
         insert_events_waits_history_long(wait);
       thread->m_events_waits_current--;
+
+      DBUG_ASSERT(wait == thread->m_events_waits_current);
     }
   }
 }
@@ -3826,6 +3838,8 @@ static void end_table_io_wait_v1(PSI_table_locker* locker)
       if (flag_events_waits_history_long)
         insert_events_waits_history_long(wait);
       thread->m_events_waits_current--;
+
+      DBUG_ASSERT(wait == thread->m_events_waits_current);
     }
   }
 
@@ -3895,6 +3909,8 @@ static void end_table_lock_wait_v1(PSI_table_locker* locker)
       if (flag_events_waits_history_long)
         insert_events_waits_history_long(wait);
       thread->m_events_waits_current--;
+
+      DBUG_ASSERT(wait == thread->m_events_waits_current);
     }
   }
 
@@ -4143,6 +4159,8 @@ static void end_file_wait_v1(PSI_file_locker *locker,
       if (flag_events_waits_history_long)
         insert_events_waits_history_long(wait);
       thread->m_events_waits_current--;
+
+      DBUG_ASSERT(wait == thread->m_events_waits_current);
     }
   }
 }
@@ -5070,6 +5088,8 @@ static void end_socket_wait_v1(PSI_socket_locker *locker, size_t byte_count)
     if (flag_events_waits_history_long)
       insert_events_waits_history_long(wait);
     thread->m_events_waits_current--;
+
+    DBUG_ASSERT(wait == thread->m_events_waits_current);
   }
 }
 
diff --git a/storage/perfschema/pfs_digest.cc b/storage/perfschema/pfs_digest.cc
index 0a01ee08993..3330c29795f 100644
--- a/storage/perfschema/pfs_digest.cc
+++ b/storage/perfschema/pfs_digest.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2008, 2015, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2008, 2016, 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
@@ -47,7 +47,7 @@ bool flag_statements_digest= true;
   Current index in Stat array where new record is to be inserted.
   index 0 is reserved for "all else" case when entire array is full.
 */
-volatile uint32 digest_index;
+volatile uint32 PFS_ALIGNED digest_monotonic_index;
 bool digest_full= false;
 
 LF_HASH digest_hash;
@@ -65,7 +65,7 @@ int init_digest(const PFS_global_param *param)
   */
   digest_max= param->m_digest_sizing;
   digest_lost= 0;
-  digest_index= 1;
+  PFS_atomic::store_u32(& digest_monotonic_index, 1);
   digest_full= false;
 
   if (digest_max == 0)
@@ -107,6 +107,9 @@ int init_digest(const PFS_global_param *param)
                                                    + index * pfs_max_digest_length, pfs_max_digest_length);
   }
 
+  /* Set record[0] as allocated. */
+  statements_digest_stat_array[0].m_lock.set_allocated();
+
   return 0;
 }
 
@@ -209,9 +212,10 @@ find_or_create_digest(PFS_thread *thread,
     memcpy(hash_key.m_schema_name, schema_name, schema_name_length);
 
   int res;
-  ulong safe_index;
   uint retry_count= 0;
   const uint retry_max= 3;
+  size_t safe_index;
+  size_t attempts= 0;
   PFS_statements_digest_stat **entry;
   PFS_statements_digest_stat *pfs= NULL;
 
@@ -247,55 +251,70 @@ search:
     return & pfs->m_stat;
   }
 
-  safe_index= PFS_atomic::add_u32(& digest_index, 1);
-  if (safe_index >= digest_max)
+  while (++attempts <= digest_max)
   {
-    /* The digest array is now full. */
-    digest_full= true;
-    pfs= &statements_digest_stat_array[0];
-
-    if (pfs->m_first_seen == 0)
-      pfs->m_first_seen= now;
-    pfs->m_last_seen= now;
-    return & pfs->m_stat;
-  }
-
-  /* Add a new record in digest stat array. */
-  pfs= &statements_digest_stat_array[safe_index];
-
-  /* Copy digest hash/LF Hash search key. */
-  memcpy(& pfs->m_digest_key, &hash_key, sizeof(PFS_digest_key));
-
-  /*
-    Copy digest storage to statement_digest_stat_array so that it could be
-    used later to generate digest text.
-  */
-  pfs->m_digest_storage.copy(digest_storage);
-
-  pfs->m_first_seen= now;
-  pfs->m_last_seen= now;
+    safe_index= PFS_atomic::add_u32(& digest_monotonic_index, 1) % digest_max;
+    if (safe_index == 0)
+    {
+      /* Record [0] is reserved. */
+      safe_index= 1;
+    }
 
-  res= lf_hash_insert(&digest_hash, pins, &pfs);
-  if (likely(res == 0))
-  {
-    return & pfs->m_stat;
-  }
+    /* Add a new record in digest stat array. */
+    pfs= &statements_digest_stat_array[safe_index];
 
-  if (res > 0)
-  {
-    /* Duplicate insert by another thread */
-    if (++retry_count > retry_max)
+    if (pfs->m_lock.is_free())
     {
-      /* Avoid infinite loops */
-      digest_lost++;
-      return NULL;
+      if (pfs->m_lock.free_to_dirty())
+      {
+        /* Copy digest hash/LF Hash search key. */
+        memcpy(& pfs->m_digest_key, &hash_key, sizeof(PFS_digest_key));
+
+        /*
+          Copy digest storage to statement_digest_stat_array so that it could be
+          used later to generate digest text.
+        */
+        pfs->m_digest_storage.copy(digest_storage);
+
+        pfs->m_first_seen= now;
+        pfs->m_last_seen= now;
+
+        res= lf_hash_insert(&digest_hash, pins, &pfs);
+        if (likely(res == 0))
+        {
+          pfs->m_lock.dirty_to_allocated();
+          return & pfs->m_stat;
+        }
+
+        pfs->m_lock.dirty_to_free();
+
+        if (res > 0)
+        {
+          /* Duplicate insert by another thread */
+          if (++retry_count > retry_max)
+          {
+            /* Avoid infinite loops */
+            digest_lost++;
+            return NULL;
+          }
+          goto search;
+        }
+
+        /* OOM in lf_hash_insert */
+        digest_lost++;
+        return NULL;
+      }
     }
-    goto search;
   }
 
-  /* OOM in lf_hash_insert */
-  digest_lost++;
-  return NULL;
+  /* The digest array is now full. */
+  digest_full= true;
+  pfs= &statements_digest_stat_array[0];
+
+  if (pfs->m_first_seen == 0)
+    pfs->m_first_seen= now;
+  pfs->m_last_seen= now;
+  return & pfs->m_stat;
 }
 
 void purge_digest(PFS_thread* thread, PFS_digest_key *hash_key)
@@ -322,10 +341,12 @@ void purge_digest(PFS_thread* thread, PFS_digest_key *hash_key)
 
 void PFS_statements_digest_stat::reset_data(unsigned char *token_array, uint length)
 {
+  m_lock.set_dirty();
   m_digest_storage.reset(token_array, length);
   m_stat.reset();
   m_first_seen= 0;
   m_last_seen= 0;
+  m_lock.dirty_to_free();
 }
 
 void PFS_statements_digest_stat::reset_index(PFS_thread *thread)
@@ -353,11 +374,14 @@ void reset_esms_by_digest()
     statements_digest_stat_array[index].reset_data(statements_digest_token_array + index * pfs_max_digest_length, pfs_max_digest_length);
   }
 
+  /* Mark record[0] as allocated again. */
+  statements_digest_stat_array[0].m_lock.set_allocated();
+
   /*
     Reset index which indicates where the next calculated digest information
     to be inserted in statements_digest_stat_array.
   */
-  digest_index= 1;
+  PFS_atomic::store_u32(& digest_monotonic_index, 1);
   digest_full= false;
 }
 
diff --git a/storage/perfschema/pfs_digest.h b/storage/perfschema/pfs_digest.h
index 76d6c33d984..429a9f4250a 100644
--- a/storage/perfschema/pfs_digest.h
+++ b/storage/perfschema/pfs_digest.h
@@ -1,4 +1,4 @@
-/* Copyright (c) 2011, 2015, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2011, 2016, 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
@@ -44,6 +44,9 @@ struct PFS_digest_key
 /** A statement digest stat record. */
 struct PFS_ALIGNED PFS_statements_digest_stat
 {
+  /** Internal lock. */
+  pfs_lock m_lock;
+
   /** Digest Schema + MD5 Hash. */
   PFS_digest_key m_digest_key;
 
diff --git a/storage/perfschema/pfs_lock.h b/storage/perfschema/pfs_lock.h
index 09efecd1c5f..be84d0f7fb4 100644
--- a/storage/perfschema/pfs_lock.h
+++ b/storage/perfschema/pfs_lock.h
@@ -1,4 +1,4 @@
-/* Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2009, 2016, 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
@@ -156,6 +156,18 @@ struct pfs_lock
   }
 
   /**
+    Initialize a lock to dirty.
+  */
+  void set_dirty(void)
+  {
+    /* Do not set the version to 0, read the previous value. */
+    uint32 copy= PFS_atomic::load_u32(&m_version_state);
+    /* Increment the version, set the DIRTY state */
+    uint32 new_val= (copy & VERSION_MASK) + VERSION_INC + PFS_LOCK_DIRTY;
+    PFS_atomic::store_u32(&m_version_state, new_val);
+  }
+
+  /**
     Execute a dirty to free transition.
     This transition should be executed by the writer that owns the record.
   */
diff --git a/storage/perfschema/table_esms_by_digest.cc b/storage/perfschema/table_esms_by_digest.cc
index 88860a77c5d..6c2b173d000 100644
--- a/storage/perfschema/table_esms_by_digest.cc
+++ b/storage/perfschema/table_esms_by_digest.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2010, 2012, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2010, 2016, 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
@@ -113,11 +113,14 @@ int table_esms_by_digest::rnd_next(void)
        m_pos.next())
   {
     digest_stat= &statements_digest_stat_array[m_pos.m_index];
-    if (digest_stat->m_first_seen != 0)
+    if (digest_stat->m_lock.is_populated())
     {
-      make_row(digest_stat);
-      m_next_pos.set_after(&m_pos);
-      return 0;
+      if (digest_stat->m_first_seen != 0)
+      {
+        make_row(digest_stat);
+        m_next_pos.set_after(&m_pos);
+        return 0;
+      }
     }
   }
 
@@ -135,10 +138,13 @@ table_esms_by_digest::rnd_pos(const void *pos)
   set_position(pos);
   digest_stat= &statements_digest_stat_array[m_pos.m_index];
 
-  if (digest_stat->m_first_seen != 0)
+  if (digest_stat->m_lock.is_populated())
   {
-    make_row(digest_stat);
-    return 0;
+    if (digest_stat->m_first_seen != 0)
+    {
+      make_row(digest_stat);
+      return 0;
+    }
   }
 
   return HA_ERR_RECORD_DELETED;
diff --git a/storage/perfschema/unittest/pfs-t.cc b/storage/perfschema/unittest/pfs-t.cc
index 6121fac098f..f76b1aa2e75 100644
--- a/storage/perfschema/unittest/pfs-t.cc
+++ b/storage/perfschema/unittest/pfs-t.cc
@@ -1316,6 +1316,7 @@ void test_locker_disabled()
   /* Pretend the socket does not have a thread owner */
   /* ---------------------------------------------- */
 
+  psi->delete_current_thread();
   socket_class_A->m_enabled= true;
   socket_A1= psi->init_socket(socket_key_A, NULL, NULL, 0);
   ok(socket_A1 != NULL, "instrumented");