Bug#22320066 EVENTS_STATEMENTS_SUMMARY_BY_DIGEST: NO UNIQUE ROWS BY

DIGEST/SCHEMA_NAME

Fix for 5.6 (backported from 5.7 and 8.0)

Before this fix, table events_statements_summary_by_digest
exposed many rows for the same statement digest and schema,
breaking the expected uniqueness of digests.

The root cause is in function find_or_create_digest(),
which does not handle duplicate inserts in the LF_HASH properly.

When two different sessions execute
the same statement for the first time,
each session creates an entry for the statement digest.

In this case, the index is maintained properly with only one entry,
but the table data itself still contained duplicated rows, orphan.

The fix is to:
- use a pfs_lock for a statement digest record
- free the duplicate record when duplication is detected in the unique index
- loop in the entire buffer to find an empty record,
  so that duplicate entries do not create holes and do not cause leaks
- honor the pfs_lock when exposing the data.

With this fix, the allocation pattern is similar to other instrumentations,
like the mutex instances for example.

Tested manually with debug code added to expose the race conditions.
Not testable by MTR scripts.

Author: marcalff

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
@@ -45,7 +45,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;
@@ -63,7 +63,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)
@@ -105,6 +105,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;
 }
 
@@ -207,9 +210,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;
 
@@ -245,55 +249,70 @@ find_or_create_digest(PFS_thread *thread,
     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)
@@ -320,10 +339,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)
@@ -351,11 +372,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;
 }
 

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;
 

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
@@ -155,6 +155,18 @@ struct pfs_lock
     PFS_atomic::store_u32(&m_version_state, new_val);
   }
 
+  /**
+    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.

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
@@ -238,11 +238,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;
+      }
     }
   }
 
@@ -260,10 +263,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;