storage/perfschema/pfs_instr_class.cc

/* Copyright (c) 2008, 2015, Oracle and/or its affiliates. All rights reserved.
  Copyright (c) 2020, MariaDB Corporation.

  This program is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License, version 2.0,
  as published by the Free Software Foundation.

  This program is also distributed with certain software (including
  but not limited to OpenSSL) that is licensed under separate terms,
  as designated in a particular file or component or in included license
  documentation.  The authors of MySQL hereby grant you an additional
  permission to link the program and your derivative works with the
  separately licensed software that they have included with MySQL.

  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, version 2.0, 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,
  51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA */

/**
  @file storage/perfschema/pfs_instr_class.cc
  Performance schema instruments meta data (implementation).
*/

#include "my_global.h"
#include "my_sys.h"
#include "structs.h"
#include "table.h"
#include "pfs_instr_class.h"
#include "pfs_instr.h"
#include "pfs_global.h"
#include "pfs_timer.h"
#include "pfs_events_waits.h"
#include "pfs_setup_object.h"
#include "pfs_atomic.h"
#include "mysql/psi/mysql_thread.h"
#include "lf.h"

#include <string.h>

/**
  @defgroup Performance_schema_buffers Performance Schema Buffers
  @ingroup Performance_schema_implementation
  @{
*/

/**
  PFS_INSTRUMENT option settings array and associated state variable to
  serialize access during shutdown.
 */
DYNAMIC_ARRAY pfs_instr_config_array;
int pfs_instr_config_state= PFS_INSTR_CONFIG_NOT_INITIALIZED;

static void configure_instr_class(PFS_instr_class *entry);

static void init_instr_class(PFS_instr_class *klass,
                             const char *name,
                             uint name_length,
                             int flags,
                             PFS_class_type class_type);

/**
  Current number of elements in mutex_class_array.
  This global variable is written to during:
  - the performance schema initialization
  - a plugin initialization
*/
static volatile uint32 mutex_class_dirty_count= 0;
static volatile uint32 mutex_class_allocated_count= 0;
static volatile uint32 rwlock_class_dirty_count= 0;
static volatile uint32 rwlock_class_allocated_count= 0;
static volatile uint32 cond_class_dirty_count= 0;
static volatile uint32 cond_class_allocated_count= 0;

/** Size of the mutex class array. @sa mutex_class_array */
ulong mutex_class_max= 0;
/** Number of mutex class lost. @sa mutex_class_array */
ulong mutex_class_lost= 0;
/** Size of the rwlock class array. @sa rwlock_class_array */
ulong rwlock_class_max= 0;
/** Number of rwlock class lost. @sa rwlock_class_array */
ulong rwlock_class_lost= 0;
/** Size of the condition class array. @sa cond_class_array */
ulong cond_class_max= 0;
/** Number of condition class lost. @sa cond_class_array */
ulong cond_class_lost= 0;
/** Size of the thread class array. @sa thread_class_array */
ulong thread_class_max= 0;
/** Number of thread class lost. @sa thread_class_array */
ulong thread_class_lost= 0;
/** Size of the file class array. @sa file_class_array */
ulong file_class_max= 0;
/** Number of file class lost. @sa file_class_array */
ulong file_class_lost= 0;
/** Size of the stage class array. @sa stage_class_array */
ulong stage_class_max= 0;
/** Number of stage class lost. @sa stage_class_array */
ulong stage_class_lost= 0;
/** Size of the statement class array. @sa statement_class_array */
ulong statement_class_max= 0;
/** Number of statement class lost. @sa statement_class_array */
ulong statement_class_lost= 0;
/** Size of the table share array. @sa table_share_array */
ulong table_share_max= 0;
/** Number of table share lost. @sa table_share_array */
ulong table_share_lost= 0;
/** Size of the socket class array. @sa socket_class_array */
ulong socket_class_max= 0;
/** Number of socket class lost. @sa socket_class_array */
ulong socket_class_lost= 0;

PFS_mutex_class *mutex_class_array= NULL;
PFS_rwlock_class *rwlock_class_array= NULL;
PFS_cond_class *cond_class_array= NULL;

/**
  Current number or elements in thread_class_array.
  This global variable is written to during:
  - the performance schema initialization
  - a plugin initialization
*/
static volatile uint32 thread_class_dirty_count= 0;
static volatile uint32 thread_class_allocated_count= 0;

static PFS_thread_class *thread_class_array= NULL;

/**
  Table instance array.
  @sa table_share_max
  @sa table_share_lost
  @sa table_share_hash
*/
PFS_table_share *table_share_array= NULL;

PFS_ALIGNED PFS_single_stat global_idle_stat;
PFS_ALIGNED PFS_table_io_stat global_table_io_stat;
PFS_ALIGNED PFS_table_lock_stat global_table_lock_stat;
PFS_ALIGNED PFS_instr_class global_table_io_class;
PFS_ALIGNED PFS_instr_class global_table_lock_class;
PFS_ALIGNED PFS_instr_class global_idle_class;

/** Class-timer map */
enum_timer_name *class_timers[] =
{&wait_timer,      /* PFS_CLASS_NONE */
 &wait_timer,      /* PFS_CLASS_MUTEX */
 &wait_timer,      /* PFS_CLASS_RWLOCK */
 &wait_timer,      /* PFS_CLASS_COND */
 &wait_timer,      /* PFS_CLASS_FILE */
 &wait_timer,      /* PFS_CLASS_TABLE */
 &stage_timer,     /* PFS_CLASS_STAGE */
 &statement_timer, /* PFS_CLASS_STATEMENT */
 &wait_timer,      /* PFS_CLASS_SOCKET */
 &wait_timer,      /* PFS_CLASS_TABLE_IO */
 &wait_timer,      /* PFS_CLASS_TABLE_LOCK */
 &idle_timer       /* PFS_CLASS_IDLE */
};

/**
  Hash index for instrumented table shares.
  This index is searched by table fully qualified name (@c PFS_table_share_key),
  and points to instrumented table shares (@c PFS_table_share).
  @sa table_share_array
  @sa PFS_table_share_key
  @sa PFS_table_share
  @sa table_share_hash_get_key
  @sa get_table_share_hash_pins
*/
LF_HASH table_share_hash;
/** True if table_share_hash is initialized. */
static bool table_share_hash_inited= false;

static volatile uint32 file_class_dirty_count= 0;
static volatile uint32 file_class_allocated_count= 0;

PFS_file_class *file_class_array= NULL;

static volatile uint32 stage_class_dirty_count= 0;
static volatile uint32 stage_class_allocated_count= 0;

static PFS_stage_class *stage_class_array= NULL;

static volatile uint32 statement_class_dirty_count= 0;
static volatile uint32 statement_class_allocated_count= 0;

static PFS_statement_class *statement_class_array= NULL;

static volatile uint32 socket_class_dirty_count= 0;
static volatile uint32 socket_class_allocated_count= 0;

static PFS_socket_class *socket_class_array= NULL;

uint mutex_class_start= 0;
uint rwlock_class_start= 0;
uint cond_class_start= 0;
uint file_class_start= 0;
uint wait_class_max= 0;
uint socket_class_start= 0;

void init_event_name_sizing(const PFS_global_param *param)
{
  mutex_class_start= 3; /* global table io, table lock, idle */
  rwlock_class_start= mutex_class_start + param->m_mutex_class_sizing;
  cond_class_start= rwlock_class_start + param->m_rwlock_class_sizing;
  file_class_start= cond_class_start + param->m_cond_class_sizing;
  socket_class_start= file_class_start + param->m_file_class_sizing;
  wait_class_max= socket_class_start + param->m_socket_class_sizing;
}

void register_global_classes()
{
  /* Table IO class */
  init_instr_class(&global_table_io_class, "wait/io/table/sql/handler", 25,
                   0, PFS_CLASS_TABLE_IO);
  global_table_io_class.m_event_name_index= GLOBAL_TABLE_IO_EVENT_INDEX;
  configure_instr_class(&global_table_io_class);

  /* Table lock class */
  init_instr_class(&global_table_lock_class, "wait/lock/table/sql/handler", 27,
                   0, PFS_CLASS_TABLE_LOCK);
  global_table_lock_class.m_event_name_index= GLOBAL_TABLE_LOCK_EVENT_INDEX;
  configure_instr_class(&global_table_lock_class);

  /* Idle class */
  init_instr_class(&global_idle_class, "idle", 4,
                   0, PFS_CLASS_IDLE);
  global_idle_class.m_event_name_index= GLOBAL_IDLE_EVENT_INDEX;
  configure_instr_class(&global_idle_class);
}

/**
  Initialize the instrument synch class buffers.
  @param mutex_class_sizing           max number of mutex class
  @param rwlock_class_sizing          max number of rwlock class
  @param cond_class_sizing            max number of condition class
  @return 0 on success
*/
int init_sync_class(uint mutex_class_sizing,
                    uint rwlock_class_sizing,
                    uint cond_class_sizing)
{
  mutex_class_dirty_count= mutex_class_allocated_count= 0;
  rwlock_class_dirty_count= rwlock_class_allocated_count= 0;
  cond_class_dirty_count= cond_class_allocated_count= 0;
  mutex_class_max= mutex_class_sizing;
  rwlock_class_max= rwlock_class_sizing;
  cond_class_max= cond_class_sizing;
  mutex_class_lost= rwlock_class_lost= cond_class_lost= 0;

  mutex_class_array= NULL;
  rwlock_class_array= NULL;
  cond_class_array= NULL;

  if (mutex_class_max > 0)
  {
    mutex_class_array= PFS_MALLOC_ARRAY(mutex_class_max, sizeof(PFS_mutex_class),
                                        PFS_mutex_class, MYF(MY_ZEROFILL));
    if (unlikely(mutex_class_array == NULL))
      return 1;
  }

  if (rwlock_class_max > 0)
  {
    rwlock_class_array= PFS_MALLOC_ARRAY(rwlock_class_max, sizeof(PFS_rwlock_class),
                                         PFS_rwlock_class, MYF(MY_ZEROFILL));
    if (unlikely(rwlock_class_array == NULL))
      return 1;
  }

  if (cond_class_max > 0)
  {
    cond_class_array= PFS_MALLOC_ARRAY(cond_class_max, sizeof(PFS_cond_class),
                                       PFS_cond_class, MYF(MY_ZEROFILL));
    if (unlikely(cond_class_array == NULL))
      return 1;
  }

  return 0;
}

/** Cleanup the instrument synch class buffers. */
void cleanup_sync_class(void)
{
  pfs_free(mutex_class_array);
  mutex_class_array= NULL;
  mutex_class_dirty_count= mutex_class_allocated_count= mutex_class_max= 0;
  pfs_free(rwlock_class_array);
  rwlock_class_array= NULL;
  rwlock_class_dirty_count= rwlock_class_allocated_count= rwlock_class_max= 0;
  pfs_free(cond_class_array);
  cond_class_array= NULL;
  cond_class_dirty_count= cond_class_allocated_count= cond_class_max= 0;
}

/**
  Initialize the thread class buffer.
  @param thread_class_sizing          max number of thread class
  @return 0 on success
*/
int init_thread_class(uint thread_class_sizing)
{
  int result= 0;
  thread_class_dirty_count= thread_class_allocated_count= 0;
  thread_class_max= thread_class_sizing;
  thread_class_lost= 0;

  if (thread_class_max > 0)
  {
    thread_class_array= PFS_MALLOC_ARRAY(thread_class_max, sizeof(PFS_thread_class),
                                         PFS_thread_class, MYF(MY_ZEROFILL));
    if (unlikely(thread_class_array == NULL))
      result= 1;
  }
  else
    thread_class_array= NULL;

  return result;
}

/** Cleanup the thread class buffers. */
void cleanup_thread_class(void)
{
  pfs_free(thread_class_array);
  thread_class_array= NULL;
  thread_class_dirty_count= thread_class_allocated_count= 0;
  thread_class_max= 0;
}

/**
  Initialize the table share buffer.
  @param table_share_sizing           max number of table share
  @return 0 on success
*/
int init_table_share(uint table_share_sizing)
{
  int result= 0;
  table_share_max= table_share_sizing;
  table_share_lost= 0;

  if (table_share_max > 0)
  {
    table_share_array= PFS_MALLOC_ARRAY(table_share_max, sizeof(PFS_table_share),
                                        PFS_table_share, MYF(MY_ZEROFILL));
    if (unlikely(table_share_array == NULL))
      result= 1;
  }
  else
    table_share_array= NULL;

  return result;
}

/** Cleanup the table share buffers. */
void cleanup_table_share(void)
{
  pfs_free(table_share_array);
  table_share_array= NULL;
  table_share_max= 0;
}

C_MODE_START
/** get_key function for @c table_share_hash. */
static uchar *table_share_hash_get_key(const uchar *entry, size_t *length,
                                       my_bool)
{
  const PFS_table_share * const *typed_entry;
  const PFS_table_share *share;
  const void *result;
  typed_entry= reinterpret_cast<const PFS_table_share* const *> (entry);
  DBUG_ASSERT(typed_entry != NULL);
  share= *typed_entry;
  DBUG_ASSERT(share != NULL);
  *length= share->m_key.m_key_length;
  result= &share->m_key.m_hash_key[0];
  return const_cast<uchar*> (reinterpret_cast<const uchar*> (result));
}
C_MODE_END

/** Initialize the table share hash table. */
int init_table_share_hash(void)
{
  if ((! table_share_hash_inited) && (table_share_max > 0))
  {
    lf_hash_init(&table_share_hash, sizeof(PFS_table_share*), LF_HASH_UNIQUE,
                 0, 0, table_share_hash_get_key, &my_charset_bin);
    /* table_share_hash.size= table_share_max; */
    table_share_hash_inited= true;
  }
  return 0;
}

/** Cleanup the table share hash table. */
void cleanup_table_share_hash(void)
{
  if (table_share_hash_inited)
  {
    lf_hash_destroy(&table_share_hash);
    table_share_hash_inited= false;
  }
}

/**
  Get the hash pins for @sa table_share_hash.
  @param thread The running thread.
  @returns The LF_HASH pins for the thread.
*/
LF_PINS* get_table_share_hash_pins(PFS_thread *thread)
{
  if (unlikely(thread->m_table_share_hash_pins == NULL))
  {
    if (! table_share_hash_inited)
      return NULL;
    thread->m_table_share_hash_pins= lf_hash_get_pins(&table_share_hash);
  }
  return thread->m_table_share_hash_pins;
}

/**
  Set a table share hash key.
  @param [out] key The key to populate.
  @param temporary True for TEMPORARY TABLE.
  @param schema_name The table schema name.
  @param schema_name_length The table schema name length.
  @param table_name The table name.
  @param table_name_length The table name length.
*/
static void set_table_share_key(PFS_table_share_key *key,
                                bool temporary,
                                const char *schema_name, uint schema_name_length,
                                const char *table_name, uint table_name_length)
{
  DBUG_ASSERT(schema_name_length <= NAME_LEN);
  DBUG_ASSERT(table_name_length <= NAME_LEN);
  char *saved_schema_name;
  char *saved_table_name;

  char *ptr= &key->m_hash_key[0];
  ptr[0]= (temporary ? OBJECT_TYPE_TEMPORARY_TABLE : OBJECT_TYPE_TABLE);
  ptr++;
  saved_schema_name= ptr;
  memcpy(ptr, schema_name, schema_name_length);
  ptr+= schema_name_length;
  ptr[0]= 0;
  ptr++;
  saved_table_name= ptr;
  memcpy(ptr, table_name, table_name_length);
  ptr+= table_name_length;
  ptr[0]= 0;
  ptr++;
  key->m_key_length= (uint)(ptr - &key->m_hash_key[0]);

  if (lower_case_table_names)
  {
    my_casedn_str(files_charset_info, saved_schema_name);
    my_casedn_str(files_charset_info, saved_table_name);
  }
}

void PFS_table_share::refresh_setup_object_flags(PFS_thread *thread)
{
  lookup_setup_object(thread,
                      OBJECT_TYPE_TABLE,
                      m_schema_name, m_schema_name_length,
                      m_table_name, m_table_name_length,
                      &m_enabled, &m_timed);
}

/**
  Initialize the file class buffer.
  @param file_class_sizing            max number of file class
  @return 0 on success
*/
int init_file_class(uint file_class_sizing)
{
  int result= 0;
  file_class_dirty_count= file_class_allocated_count= 0;
  file_class_max= file_class_sizing;
  file_class_lost= 0;

  if (file_class_max > 0)
  {
    file_class_array= PFS_MALLOC_ARRAY(file_class_max, sizeof(PFS_file_class),
                                       PFS_file_class, MYF(MY_ZEROFILL));
    if (unlikely(file_class_array == NULL))
      return 1;
  }
  else
    file_class_array= NULL;

  return result;
}

/** Cleanup the file class buffers. */
void cleanup_file_class(void)
{
  pfs_free(file_class_array);
  file_class_array= NULL;
  file_class_dirty_count= file_class_allocated_count= 0;
  file_class_max= 0;
}

/**
  Initialize the stage class buffer.
  @param stage_class_sizing            max number of stage class
  @return 0 on success
*/
int init_stage_class(uint stage_class_sizing)
{
  int result= 0;
  stage_class_dirty_count= stage_class_allocated_count= 0;
  stage_class_max= stage_class_sizing;
  stage_class_lost= 0;

  if (stage_class_max > 0)
  {
    stage_class_array= PFS_MALLOC_ARRAY(stage_class_max, sizeof(PFS_stage_class),
                                        PFS_stage_class, MYF(MY_ZEROFILL));
    if (unlikely(stage_class_array == NULL))
      return 1;
  }
  else
    stage_class_array= NULL;

  return result;
}

/** Cleanup the stage class buffers. */
void cleanup_stage_class(void)
{
  pfs_free(stage_class_array);
  stage_class_array= NULL;
  stage_class_dirty_count= stage_class_allocated_count= 0;
  stage_class_max= 0;
}

/**
  Initialize the statement class buffer.
  @param statement_class_sizing            max number of statement class
  @return 0 on success
*/
int init_statement_class(uint statement_class_sizing)
{
  int result= 0;
  statement_class_dirty_count= statement_class_allocated_count= 0;
  statement_class_max= statement_class_sizing;
  statement_class_lost= 0;

  if (statement_class_max > 0)
  {
    statement_class_array= PFS_MALLOC_ARRAY(statement_class_max, sizeof(PFS_statement_class),
                                            PFS_statement_class, MYF(MY_ZEROFILL));
    if (unlikely(statement_class_array == NULL))
      return 1;
  }
  else
    statement_class_array= NULL;

  return result;
}

/** Cleanup the statement class buffers. */
void cleanup_statement_class(void)
{
  pfs_free(statement_class_array);
  statement_class_array= NULL;
  statement_class_dirty_count= statement_class_allocated_count= 0;
  statement_class_max= 0;
}

/**
  Initialize the socket class buffer.
  @param socket_class_sizing            max number of socket class
  @return 0 on success
*/
int init_socket_class(uint socket_class_sizing)
{
  int result= 0;
  socket_class_dirty_count= socket_class_allocated_count= 0;
  socket_class_max= socket_class_sizing;
  socket_class_lost= 0;

  if (socket_class_max > 0)
  {
    socket_class_array= PFS_MALLOC_ARRAY(socket_class_max, sizeof(PFS_socket_class),
                                         PFS_socket_class, MYF(MY_ZEROFILL));
    if (unlikely(socket_class_array == NULL))
      return 1;
  }
  else
    socket_class_array= NULL;

  return result;
}

/** Cleanup the socket class buffers. */
void cleanup_socket_class(void)
{
  pfs_free(socket_class_array);
  socket_class_array= NULL;
  socket_class_dirty_count= socket_class_allocated_count= 0;
  socket_class_max= 0;
}

static void init_instr_class(PFS_instr_class *klass,
                             const char *name,
                             uint name_length,
                             int flags,
                             PFS_class_type class_type)
{
  DBUG_ASSERT(name_length <= PFS_MAX_INFO_NAME_LENGTH);
  memset(klass, 0, sizeof(PFS_instr_class));
  strncpy(klass->m_name, name, name_length);
  klass->m_name[PFS_MAX_INFO_NAME_LENGTH - 1]= '\0';
  klass->m_name_length= name_length;
  klass->m_flags= flags;
  klass->m_enabled= true;
  klass->m_timed= true;
  klass->m_type= class_type;
  klass->m_timer= class_timers[class_type];
}

/**
  Set user-defined configuration values for an instrument.
*/
static void configure_instr_class(PFS_instr_class *entry)
{
  uint match_length= 0; /* length of matching pattern */

  for (uint i= 0; i < pfs_instr_config_array.elements; i++)
  {
    PFS_instr_config* e;
    get_dynamic(&pfs_instr_config_array, (uchar*)&e, i);

    /**
      Compare class name to all configuration entries. In case of multiple
      matches, the longer specification wins. For example, the pattern
      'ABC/DEF/GHI=ON' has precedence over 'ABC/DEF/%=OFF' regardless of
      position within the configuration file or command line.

      Consecutive wildcards affect the count.
    */
    if (!my_wildcmp(&my_charset_latin1,
                    entry->m_name, entry->m_name+entry->m_name_length,
                    e->m_name, e->m_name+e->m_name_length,
                    '\\', '?','%'))
    {
        if (e->m_name_length >= match_length)
        {
           entry->m_enabled= e->m_enabled;
           entry->m_timed= e->m_timed;
           match_length= MY_MAX(e->m_name_length, match_length);
        }
    }
  }
}

#define REGISTER_CLASS_BODY_PART(INDEX, ARRAY, MAX, NAME, NAME_LENGTH) \
  for (INDEX= 0; INDEX < MAX; INDEX++)                                 \
  {                                                                    \
    entry= &ARRAY[INDEX];                                              \
    if ((entry->m_name_length == NAME_LENGTH) &&                       \
        (strncmp(entry->m_name, NAME, NAME_LENGTH) == 0))              \
    {                                                                  \
      DBUG_ASSERT(entry->m_flags == flags);                            \
      return (INDEX + 1);                                              \
    }                                                                  \
  }

/**
  Register a mutex instrumentation metadata.
  @param name                         the instrumented name
  @param name_length                  length in bytes of name
  @param flags                        the instrumentation flags
  @return a mutex instrumentation key
*/
PFS_sync_key register_mutex_class(const char *name, uint name_length,
                                  int flags)
{
  uint32 index;
  PFS_mutex_class *entry;

  /*
    This is a full array scan, which is not optimal.
    This is acceptable since this code is only used at startup,
    or when a plugin is loaded.
  */
  REGISTER_CLASS_BODY_PART(index, mutex_class_array, mutex_class_max,
                           name, name_length)
  /*
    Note that:
    mutex_class_dirty_count is incremented *before* an entry is added
    mutex_class_allocated_count is incremented *after* an entry is added
  */
  index= PFS_atomic::add_u32(&mutex_class_dirty_count, 1);

  if (index < mutex_class_max)
  {
    /*
      The instrument was not found (from a possible previous
      load / unload of a plugin), allocate it.
      This code is safe when 2 threads execute in parallel
      for different mutex classes:
      - thread 1 registering class A
      - thread 2 registering class B
      will not collide in the same mutex_class_array[index] entry.
      This code does not protect against 2 threads registering
      in parallel the same class:
      - thread 1 registering class A
      - thread 2 registering class A
      could lead to a duplicate class A entry.
      This is ok, since this case can not happen in the caller:
      - classes names are derived from a plugin name
        ('wait/synch/mutex/<plugin>/xxx')
      - 2 threads can not register concurrently the same plugin
        in INSTALL PLUGIN.
    */
    entry= &mutex_class_array[index];
    init_instr_class(entry, name, name_length, flags, PFS_CLASS_MUTEX);
    entry->m_mutex_stat.reset();
    entry->m_event_name_index= mutex_class_start + index;
    entry->m_singleton= NULL;
    entry->m_enabled= false; /* disabled by default */
    entry->m_timed= false;

    /* Set user-defined configuration options for this instrument */
    configure_instr_class(entry);

    /*
      Now that this entry is populated, advertise it

      Technically, there is a small race condition here:
      T0:
      mutex_class_dirty_count= 10
      mutex_class_allocated_count= 10
      T1: Thread A increment mutex_class_dirty_count to 11
      T2: Thread B increment mutex_class_dirty_count to 12
      T3: Thread A populate entry 11
      T4: Thread B populate entry 12
      T5: Thread B increment mutex_class_allocated_count to 11,
          advertise thread A incomplete record 11,
          but does not advertise thread B complete record 12
      T6: Thread A increment mutex_class_allocated_count to 12
      This has no impact, and is acceptable.
      A reader will not see record 12 for a short time.
      A reader will see an incomplete record 11 for a short time,
      which is ok: the mutex name / statistics will be temporarily
      empty/NULL/zero, but this won't cause a crash
      (mutex_class_array is initialized with MY_ZEROFILL).
    */
    PFS_atomic::add_u32(&mutex_class_allocated_count, 1);
    return (index + 1);
  }

  /*
    Out of space, report to SHOW STATUS that
    the allocated memory was too small.
  */
  mutex_class_lost++;
  return 0;
}

/**
  Register a rwlock instrumentation metadata.
  @param name                         the instrumented name
  @param name_length                  length in bytes of name
  @param flags                        the instrumentation flags
  @return a rwlock instrumentation key
*/
PFS_sync_key register_rwlock_class(const char *name, uint name_length,
                                   int flags)
{
  /* See comments in register_mutex_class */
  uint32 index;
  PFS_rwlock_class *entry;

  REGISTER_CLASS_BODY_PART(index, rwlock_class_array, rwlock_class_max,
                           name, name_length)

  index= PFS_atomic::add_u32(&rwlock_class_dirty_count, 1);

  if (index < rwlock_class_max)
  {
    entry= &rwlock_class_array[index];
    init_instr_class(entry, name, name_length, flags, PFS_CLASS_RWLOCK);
    entry->m_rwlock_stat.reset();
    entry->m_event_name_index= rwlock_class_start + index;
    entry->m_singleton= NULL;
    entry->m_enabled= false; /* disabled by default */
    entry->m_timed= false;
    /* Set user-defined configuration options for this instrument */
    configure_instr_class(entry);
    PFS_atomic::add_u32(&rwlock_class_allocated_count, 1);
    return (index + 1);
  }

  rwlock_class_lost++;
  return 0;
}

/**
  Register a condition instrumentation metadata.
  @param name                         the instrumented name
  @param name_length                  length in bytes of name
  @param flags                        the instrumentation flags
  @return a condition instrumentation key
*/
PFS_sync_key register_cond_class(const char *name, uint name_length,
                                 int flags)
{
  /* See comments in register_mutex_class */
  uint32 index;
  PFS_cond_class *entry;

  REGISTER_CLASS_BODY_PART(index, cond_class_array, cond_class_max,
                           name, name_length)

  index= PFS_atomic::add_u32(&cond_class_dirty_count, 1);

  if (index < cond_class_max)
  {
    entry= &cond_class_array[index];
    init_instr_class(entry, name, name_length, flags, PFS_CLASS_COND);
    entry->m_event_name_index= cond_class_start + index;
    entry->m_singleton= NULL;
    entry->m_enabled= false; /* disabled by default */
    entry->m_timed= false;
    /* Set user-defined configuration options for this instrument */
    configure_instr_class(entry);
    PFS_atomic::add_u32(&cond_class_allocated_count, 1);
    return (index + 1);
  }

  cond_class_lost++;
  return 0;
}

#define FIND_CLASS_BODY(KEY, COUNT, ARRAY) \
  if ((KEY == 0) || (KEY > COUNT))         \
    return NULL;                           \
  return &ARRAY[KEY - 1]

/**
  Find a mutex instrumentation class by key.
  @param key                          the instrument key
  @return the instrument class, or NULL
*/
PFS_mutex_class *find_mutex_class(PFS_sync_key key)
{
  FIND_CLASS_BODY(key, mutex_class_allocated_count, mutex_class_array);
}

PFS_mutex_class *sanitize_mutex_class(PFS_mutex_class *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_mutex_class, mutex_class_array, mutex_class_max, unsafe);
}

/**
  Find a rwlock instrumentation class by key.
  @param key                          the instrument key
  @return the instrument class, or NULL
*/
PFS_rwlock_class *find_rwlock_class(PFS_sync_key key)
{
  FIND_CLASS_BODY(key, rwlock_class_allocated_count, rwlock_class_array);
}

PFS_rwlock_class *sanitize_rwlock_class(PFS_rwlock_class *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_rwlock_class, rwlock_class_array, rwlock_class_max, unsafe);
}

/**
  Find a condition instrumentation class by key.
  @param key                          the instrument key
  @return the instrument class, or NULL
*/
PFS_cond_class *find_cond_class(PFS_sync_key key)
{
  FIND_CLASS_BODY(key, cond_class_allocated_count, cond_class_array);
}

PFS_cond_class *sanitize_cond_class(PFS_cond_class *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_cond_class, cond_class_array, cond_class_max, unsafe);
}

/**
  Register a thread instrumentation metadata.
  @param name                         the instrumented name
  @param name_length                  length in bytes of name
  @param flags                        the instrumentation flags
  @return a thread instrumentation key
*/
PFS_thread_key register_thread_class(const char *name, uint name_length,
                                     int flags)
{
  /* See comments in register_mutex_class */
  uint32 index;
  PFS_thread_class *entry;

  for (index= 0; index < thread_class_max; index++)
  {
    entry= &thread_class_array[index];

    if ((entry->m_name_length == name_length) &&
        (strncmp(entry->m_name, name, name_length) == 0))
      return (index + 1);
  }

  index= PFS_atomic::add_u32(&thread_class_dirty_count, 1);

  if (index < thread_class_max)
  {
    entry= &thread_class_array[index];
    DBUG_ASSERT(name_length <= PFS_MAX_INFO_NAME_LENGTH);
    strncpy(entry->m_name, name, name_length);
    entry->m_name_length= name_length;
    entry->m_enabled= true;
    PFS_atomic::add_u32(&thread_class_allocated_count, 1);
    return (index + 1);
  }

  thread_class_lost++;
  return 0;
}

/**
  Find a thread instrumentation class by key.
  @param key                          the instrument key
  @return the instrument class, or NULL
*/
PFS_thread_class *find_thread_class(PFS_sync_key key)
{
  FIND_CLASS_BODY(key, thread_class_allocated_count, thread_class_array);
}

PFS_thread_class *sanitize_thread_class(PFS_thread_class *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_thread_class, thread_class_array, thread_class_max, unsafe);
}

/**
  Register a file instrumentation metadata.
  @param name                         the instrumented name
  @param name_length                  length in bytes of name
  @param flags                        the instrumentation flags
  @return a file instrumentation key
*/
PFS_file_key register_file_class(const char *name, uint name_length,
                                 int flags)
{
  /* See comments in register_mutex_class */
  uint32 index;
  PFS_file_class *entry;

  REGISTER_CLASS_BODY_PART(index, file_class_array, file_class_max,
                           name, name_length)

  index= PFS_atomic::add_u32(&file_class_dirty_count, 1);

  if (index < file_class_max)
  {
    entry= &file_class_array[index];
    init_instr_class(entry, name, name_length, flags, PFS_CLASS_FILE);
    entry->m_event_name_index= file_class_start + index;
    entry->m_singleton= NULL;
    entry->m_enabled= true; /* enabled by default */
    entry->m_timed= true;
    /* Set user-defined configuration options for this instrument */
    configure_instr_class(entry);
    PFS_atomic::add_u32(&file_class_allocated_count, 1);
    return (index + 1);
  }

  file_class_lost++;
  return 0;
}

/**
  Register a stage instrumentation metadata.
  @param name                         the instrumented name
  @param prefix_length                length in bytes of the name prefix
  @param name_length                  length in bytes of name
  @param flags                        the instrumentation flags
  @return a stage instrumentation key
*/
PFS_stage_key register_stage_class(const char *name,
                                   uint prefix_length,
                                   uint name_length,
                                   int flags)
{
  /* See comments in register_mutex_class */
  uint32 index;
  PFS_stage_class *entry;

  REGISTER_CLASS_BODY_PART(index, stage_class_array, stage_class_max,
                           name, name_length)

  index= PFS_atomic::add_u32(&stage_class_dirty_count, 1);

  if (index < stage_class_max)
  {
    entry= &stage_class_array[index];
    init_instr_class(entry, name, name_length, flags, PFS_CLASS_STAGE);
    entry->m_prefix_length= prefix_length;
    entry->m_event_name_index= index;
    entry->m_enabled= false; /* disabled by default */
    entry->m_timed= false;
    /* Set user-defined configuration options for this instrument */
    configure_instr_class(entry);
    PFS_atomic::add_u32(&stage_class_allocated_count, 1);

    return (index + 1);
  }

  stage_class_lost++;
  return 0;
}

/**
  Register a statement instrumentation metadata.
  @param name                         the instrumented name
  @param name_length                  length in bytes of name
  @param flags                        the instrumentation flags
  @return a statement instrumentation key
*/
PFS_statement_key register_statement_class(const char *name, uint name_length,
                                           int flags)
{
  /* See comments in register_mutex_class */
  uint32 index;
  PFS_statement_class *entry;

  REGISTER_CLASS_BODY_PART(index, statement_class_array, statement_class_max,
                           name, name_length)

  index= PFS_atomic::add_u32(&statement_class_dirty_count, 1);

  if (index < statement_class_max)
  {
    entry= &statement_class_array[index];
    init_instr_class(entry, name, name_length, flags, PFS_CLASS_STATEMENT);
    entry->m_event_name_index= index;
    entry->m_enabled= true; /* enabled by default */
    entry->m_timed= true;
    /* Set user-defined configuration options for this instrument */
    configure_instr_class(entry);
    PFS_atomic::add_u32(&statement_class_allocated_count, 1);

    return (index + 1);
  }

  statement_class_lost++;
  return 0;
}

/**
  Find a file instrumentation class by key.
  @param key                          the instrument key
  @return the instrument class, or NULL
*/
PFS_file_class *find_file_class(PFS_file_key key)
{
  FIND_CLASS_BODY(key, file_class_allocated_count, file_class_array);
}

PFS_file_class *sanitize_file_class(PFS_file_class *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_file_class, file_class_array, file_class_max, unsafe);
}

/**
  Find a stage instrumentation class by key.
  @param key                          the instrument key
  @return the instrument class, or NULL
*/
PFS_stage_class *find_stage_class(PFS_stage_key key)
{
  FIND_CLASS_BODY(key, stage_class_allocated_count, stage_class_array);
}

PFS_stage_class *sanitize_stage_class(PFS_stage_class *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_stage_class, stage_class_array, stage_class_max, unsafe);
}

/**
  Find a statement instrumentation class by key.
  @param key                          the instrument key
  @return the instrument class, or NULL
*/
PFS_statement_class *find_statement_class(PFS_stage_key key)
{
  FIND_CLASS_BODY(key, statement_class_allocated_count, statement_class_array);
}

PFS_statement_class *sanitize_statement_class(PFS_statement_class *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_statement_class, statement_class_array, statement_class_max, unsafe);
}

/**
  Register a socket instrumentation metadata.
  @param name                         the instrumented name
  @param name_length                  length in bytes of name
  @param flags                        the instrumentation flags
  @return a socket instrumentation key
*/
PFS_socket_key register_socket_class(const char *name, uint name_length,
                                     int flags)
{
  /* See comments in register_mutex_class */
  uint32 index;
  PFS_socket_class *entry;

  REGISTER_CLASS_BODY_PART(index, socket_class_array, socket_class_max,
                           name, name_length)

  index= PFS_atomic::add_u32(&socket_class_dirty_count, 1);

  if (index < socket_class_max)
  {
    entry= &socket_class_array[index];
    init_instr_class(entry, name, name_length, flags, PFS_CLASS_SOCKET);
    entry->m_event_name_index= socket_class_start + index;
    entry->m_singleton= NULL;
    entry->m_enabled= false; /* disabled by default */
    entry->m_timed= false;
    /* Set user-defined configuration options for this instrument */
    configure_instr_class(entry);
    PFS_atomic::add_u32(&socket_class_allocated_count, 1);
    return (index + 1);
  }

  socket_class_lost++;
  return 0;
}

/**
  Find a socket instrumentation class by key.
  @param key                          the instrument key
  @return the instrument class, or NULL
*/
PFS_socket_class *find_socket_class(PFS_socket_key key)
{
  FIND_CLASS_BODY(key, socket_class_allocated_count, socket_class_array);
}

PFS_socket_class *sanitize_socket_class(PFS_socket_class *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_socket_class, socket_class_array, socket_class_max, unsafe);
}

PFS_instr_class *find_table_class(uint index)
{
  if (index == 1)
    return & global_table_io_class;
  if (index == 2)
    return & global_table_lock_class;
  return NULL;
}

PFS_instr_class *sanitize_table_class(PFS_instr_class *unsafe)
{
  if (likely((& global_table_io_class == unsafe) ||
             (& global_table_lock_class == unsafe)))
    return unsafe;
  return NULL;
}

PFS_instr_class *find_idle_class(uint index)
{
  if (index == 1)
    return & global_idle_class;
  return NULL;
}

PFS_instr_class *sanitize_idle_class(PFS_instr_class *unsafe)
{
  if (likely(& global_idle_class == unsafe))
    return unsafe;
  return NULL;
}

static void set_keys(PFS_table_share *pfs, const TABLE_SHARE *share)
{
  uint len;
  KEY *key_info= share->key_info;
  PFS_table_key *pfs_key= pfs->m_keys;
  PFS_table_key *pfs_key_last= pfs->m_keys + share->keys;
  pfs->m_key_count= share->keys;

  for ( ; pfs_key < pfs_key_last; pfs_key++, key_info++)
  {
    len= (uint)key_info->name.length;
    memcpy(pfs_key->m_name, key_info->name.str, len);
    pfs_key->m_name_length= len;
  }

  pfs_key_last= pfs->m_keys + MAX_INDEXES;
  for ( ; pfs_key < pfs_key_last; pfs_key++)
    pfs_key->m_name_length= 0;
}

static int compare_keys(PFS_table_share *pfs, const TABLE_SHARE *share)
{
  uint len;
  KEY *key_info= share->key_info;
  PFS_table_key *pfs_key= pfs->m_keys;
  PFS_table_key *pfs_key_last= pfs->m_keys + share->keys;

  if (pfs->m_key_count != share->keys)
    return 1;

  for ( ; pfs_key < pfs_key_last; pfs_key++, key_info++)
  {
    len= (uint)key_info->name.length;
    if (len != pfs_key->m_name_length)
      return 1;

    if (memcmp(pfs_key->m_name, key_info->name.str, len) != 0)
      return 1;
  }

  return 0;
}

/**
  Find or create a table share instrumentation.
  @param thread                       the executing instrumented thread
  @param temporary                    true for TEMPORARY TABLE
  @param share                        table share
  @return a table share, or NULL
*/
PFS_table_share* find_or_create_table_share(PFS_thread *thread,
                                            bool temporary,
                                            const TABLE_SHARE *share)
{
  /* See comments in register_mutex_class */
  PFS_table_share_key key;

  LF_PINS *pins= get_table_share_hash_pins(thread);
  if (unlikely(pins == NULL))
  {
    table_share_lost++;
    return NULL;
  }

  const char *schema_name= share->db.str;
  uint schema_name_length= (uint)share->db.length;
  const char *table_name= share->table_name.str;
  uint table_name_length= (uint)share->table_name.length;

  set_table_share_key(&key, temporary,
                      schema_name, schema_name_length,
                      table_name, table_name_length);

  PFS_table_share **entry;
  uint retry_count= 0;
  const uint retry_max= 3;
  bool enabled= true;
  bool timed= true;
  static uint PFS_ALIGNED table_share_monotonic_index= 0;
  uint index;
  uint attempts= 0;
  PFS_table_share *pfs;

search:
  entry= reinterpret_cast<PFS_table_share**>
    (lf_hash_search(&table_share_hash, pins,
                    key.m_hash_key, key.m_key_length));
  if (entry && (entry != MY_ERRPTR))
  {
    pfs= *entry;
    pfs->inc_refcount() ;
    if (compare_keys(pfs, share) != 0)
    {
      set_keys(pfs, share);
      /* FIXME: aggregate to table_share sink ? */
      pfs->m_table_stat.fast_reset();
    }
    lf_hash_search_unpin(pins);
    return pfs;
  }

  lf_hash_search_unpin(pins);

  if (retry_count == 0)
  {
    lookup_setup_object(thread,
                        OBJECT_TYPE_TABLE,
                        schema_name, schema_name_length,
                        table_name, table_name_length,
                        &enabled, &timed);
    /*
      Even when enabled is false, a record is added in the dictionary:
      - It makes enabling a table already in the table cache possible,
      - It improves performances for the next time a TABLE_SHARE is reloaded
        in the table cache.
    */
  }

  while (++attempts <= table_share_max)
  {
    /* See create_mutex() */
    index= PFS_atomic::add_u32(& table_share_monotonic_index, 1) % table_share_max;
    pfs= table_share_array + index;

    if (pfs->m_lock.is_free())
    {
      if (pfs->m_lock.free_to_dirty())
      {
        pfs->m_key= key;
        pfs->m_schema_name= &pfs->m_key.m_hash_key[1];
        pfs->m_schema_name_length= schema_name_length;
        pfs->m_table_name= &pfs->m_key.m_hash_key[schema_name_length + 2];
        pfs->m_table_name_length= table_name_length;
        pfs->m_enabled= enabled;
        pfs->m_timed= timed;
        pfs->init_refcount();
        pfs->m_table_stat.fast_reset();
        set_keys(pfs, share);

        int res;
        res= lf_hash_insert(&table_share_hash, pins, &pfs);
        if (likely(res == 0))
        {
          pfs->m_lock.dirty_to_allocated();
          return pfs;
        }

        pfs->m_lock.dirty_to_free();

        if (res > 0)
        {
          /* Duplicate insert by another thread */
          if (++retry_count > retry_max)
          {
            /* Avoid infinite loops */
            table_share_lost++;
            return NULL;
          }
          goto search;
        }

        /* OOM in lf_hash_insert */
        table_share_lost++;
        return NULL;
      }
    }
  }

  table_share_lost++;
  return NULL;
}

void PFS_table_share::aggregate_io(void)
{
  uint safe_key_count= sanitize_index_count(m_key_count);
  PFS_table_io_stat *from_stat;
  PFS_table_io_stat *from_stat_last;
  PFS_table_io_stat sum_io;

  /* Aggregate stats for each index, if any */
  from_stat= & m_table_stat.m_index_stat[0];
  from_stat_last= from_stat + safe_key_count;
  for ( ; from_stat < from_stat_last ; from_stat++)
    sum_io.aggregate(from_stat);

  /* Aggregate stats for the table */
  sum_io.aggregate(& m_table_stat.m_index_stat[MAX_INDEXES]);

  /* Add this table stats to the global sink. */
  global_table_io_stat.aggregate(& sum_io);
  m_table_stat.fast_reset_io();
}

void PFS_table_share::aggregate_lock(void)
{
  global_table_lock_stat.aggregate(& m_table_stat.m_lock_stat);
  m_table_stat.fast_reset_lock();
}

void release_table_share(PFS_table_share *pfs)
{
  DBUG_ASSERT(pfs->get_refcount() > 0);
  pfs->dec_refcount();
}

/**
  Drop the instrumented table share associated with a table.
  @param thread The running thread
  @param temporary True for TEMPORARY TABLE
  @param schema_name The table schema name
  @param schema_name_length The table schema name length
  @param table_name The table name
  @param table_name_length The table name length
*/
void drop_table_share(PFS_thread *thread,
                      bool temporary,
                      const char *schema_name, uint schema_name_length,
                      const char *table_name, uint table_name_length)
{
  PFS_table_share_key key;
  LF_PINS* pins= get_table_share_hash_pins(thread);
  if (unlikely(pins == NULL))
    return;
  set_table_share_key(&key, temporary, schema_name, schema_name_length,
                      table_name, table_name_length);
  PFS_table_share **entry;
  entry= reinterpret_cast<PFS_table_share**>
    (lf_hash_search(&table_share_hash, pins,
                    key.m_hash_key, key.m_key_length));
  if (entry && (entry != MY_ERRPTR))
  {
    PFS_table_share *pfs= *entry;
    lf_hash_delete(&table_share_hash, pins,
                   pfs->m_key.m_hash_key, pfs->m_key.m_key_length);
    pfs->m_lock.allocated_to_free();
  }

  lf_hash_search_unpin(pins);
}

/**
  Sanitize an unsafe table_share pointer.
  @param unsafe The possibly corrupt pointer.
  @return A valid table_safe_pointer, or NULL.
*/
PFS_table_share *sanitize_table_share(PFS_table_share *unsafe)
{
  SANITIZE_ARRAY_BODY(PFS_table_share, table_share_array, table_share_max, unsafe);
}

/** Reset the wait statistics per instrument class. */
void reset_events_waits_by_class()
{
  reset_file_class_io();
  reset_socket_class_io();
  global_idle_stat.reset();
  global_table_io_stat.reset();
  global_table_lock_stat.reset();
}

/** Reset the io statistics per file class. */
void reset_file_class_io(void)
{
  PFS_file_class *pfs= file_class_array;
  PFS_file_class *pfs_last= file_class_array + file_class_max;

  for ( ; pfs < pfs_last; pfs++)
    pfs->m_file_stat.m_io_stat.reset();
}

/** Reset the io statistics per socket class. */
void reset_socket_class_io(void)
{
  PFS_socket_class *pfs= socket_class_array;
  PFS_socket_class *pfs_last= socket_class_array + socket_class_max;

  for ( ; pfs < pfs_last; pfs++)
    pfs->m_socket_stat.m_io_stat.reset();
}

void update_table_share_derived_flags(PFS_thread *thread)
{
  PFS_table_share *pfs= table_share_array;
  PFS_table_share *pfs_last= table_share_array + table_share_max;

  for ( ; pfs < pfs_last; pfs++)
  {
    if (pfs->m_lock.is_populated())
      pfs->refresh_setup_object_flags(thread);
  }
}

/** @} */