MDEV-31048 PERFORMANCE_SCHEMA lakcs InnoDB read_slots and write_slots

tpool::cache::m_mtx: Add PERFORMANCE_SCHEMA instrumentation
(wait/synch/mutex/innodb/tpool_cache_mutex). This covers the
InnoDB read_slots and write_slots for asynchronous data page I/O.

3 files changed, 54 insertions, 24 deletions

diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 04a857ecd0c..f102789d7ab 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -542,6 +542,7 @@ mysql_pfs_key_t	trx_pool_manager_mutex_key;
 mysql_pfs_key_t	lock_wait_mutex_key;
 mysql_pfs_key_t	trx_sys_mutex_key;
 mysql_pfs_key_t	srv_threads_mutex_key;
+mysql_pfs_key_t	tpool_cache_mutex_key;
 
 /* all_innodb_mutexes array contains mutexes that are
 performance schema instrumented if "UNIV_PFS_MUTEX"
@@ -577,6 +578,7 @@ static PSI_mutex_info all_innodb_mutexes[] = {
 	PSI_KEY(rtr_match_mutex),
 	PSI_KEY(rtr_path_mutex),
 	PSI_KEY(trx_sys_mutex),
+	PSI_KEY(tpool_cache_mutex),
 };
 # endif /* UNIV_PFS_MUTEX */
 
diff --git a/storage/innobase/os/os0file.cc b/storage/innobase/os/os0file.cc
index 12561877c0c..d366c784b96 100644
--- a/storage/innobase/os/os0file.cc
+++ b/storage/innobase/os/os0file.cc
@@ -132,7 +132,7 @@ public:
 		wait();
 	}
 
-	std::mutex& mutex()
+	mysql_mutex_t& mutex()
 	{
 		return m_cache.mutex();
 	}
@@ -3668,8 +3668,10 @@ void os_aio_wait_until_no_pending_writes()
 /** @return number of pending reads */
 size_t os_aio_pending_reads()
 {
-  std::unique_lock<std::mutex> lk(read_slots->mutex());
-  return read_slots->pending_io_count();
+  mysql_mutex_lock(&read_slots->mutex());
+  size_t pending= read_slots->pending_io_count();
+  mysql_mutex_unlock(&read_slots->mutex());
+  return pending;
 }
 
 /** @return approximate number of pending reads */
@@ -3681,8 +3683,10 @@ size_t os_aio_pending_reads_approx()
 /** @return number of pending writes */
 size_t os_aio_pending_writes()
 {
-  std::unique_lock<std::mutex> lk(write_slots->mutex());
-  return write_slots->pending_io_count();
+  mysql_mutex_lock(&write_slots->mutex());
+  size_t pending= write_slots->pending_io_count();
+  mysql_mutex_unlock(&write_slots->mutex());
+  return pending;
 }
 
 /** Wait until all pending asynchronous reads have completed. */
diff --git a/tpool/tpool_structs.h b/tpool/tpool_structs.h
index b6ca3f54016..550a92d6e58 100644
--- a/tpool/tpool_structs.h
+++ b/tpool/tpool_structs.h
@@ -14,14 +14,13 @@ along with this program; if not, write to the Free Software
 Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02111 - 1301 USA*/
 
 #pragma once
+#include <my_global.h>
+#include <my_pthread.h>
 #include <vector>
 #include <stack>
-#include <mutex>
-#include <condition_variable>
 #include <assert.h>
 #include <algorithm>
 
-
 /* Suppress TSAN warnings, that we believe are not critical. */
 #if defined(__has_feature)
 #define TPOOL_HAS_FEATURE(...) __has_feature(__VA_ARGS__)
@@ -37,6 +36,11 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02111 - 1301 USA*/
 #define TPOOL_SUPPRESS_TSAN
 #endif
 
+#ifdef HAVE_PSI_INTERFACE
+typedef unsigned int mysql_pfs_key_t;
+extern mysql_pfs_key_t tpool_cache_mutex_key;
+#endif
+
 namespace tpool
 {
 
@@ -55,13 +59,13 @@ namespace tpool
 template<typename T> class cache
 {
   /** Protects updates of m_pos and m_cache members */
-  std::mutex m_mtx;
+  mysql_mutex_t m_mtx;
 
   /**
     Notify waiting threads about "cache full" or "cache not empty" conditions
     @see get() and wait()
   */
-  std::condition_variable m_cv;
+  pthread_cond_t m_cv;
 
   /** Cached items vector.Does not change after construction */
   std::vector<T> m_base;
@@ -108,13 +112,22 @@ public:
   Constructor
   @param size - maximum number of items in cache
   */
-  cache(size_t size) : m_mtx(), m_cv(), m_base(size), m_cache(size),
+  cache(size_t size) : m_base(size), m_cache(size),
     m_waiters(), m_pos(0)
   {
+    mysql_mutex_init(tpool_cache_mutex_key, &m_mtx, nullptr);
+    pthread_cond_init(&m_cv, nullptr);
+
     for(size_t i= 0 ; i < size; i++)
       m_cache[i]= &m_base[i];
   }
 
+  ~cache()
+  {
+    mysql_mutex_destroy(&m_mtx);
+    pthread_cond_destroy(&m_cv);
+ }
+
   /**
    Retrieve an item from cache. Waits for free item, if cache is
    currently empty.
@@ -122,16 +135,17 @@ public:
   */
   T* get()
   {
-    std::unique_lock<std::mutex> lk(m_mtx);
-    while(is_empty())
-      m_cv.wait(lk);
+    mysql_mutex_lock(&m_mtx);
+    while (is_empty())
+      my_cond_wait(&m_cv, &m_mtx.m_mutex);
     assert(m_pos < capacity());
     //  return last element
-    return m_cache[m_pos++];
+    T *t= m_cache[m_pos++];
+    mysql_mutex_unlock(&m_mtx);
+    return t;
   }
 
-
-  std::mutex &mutex() { return m_mtx; }
+  mysql_mutex_t &mutex() { return m_mtx; }
 
   /**
    Put back an element to cache.
@@ -139,7 +153,7 @@ public:
   */
   void put(T *ele)
   {
-    std::unique_lock<std::mutex> lk(m_mtx);
+    mysql_mutex_lock(&m_mtx);
     assert(!is_full());
     // put element to the logical end of the array
     m_cache[--m_pos] = ele;
@@ -147,7 +161,8 @@ public:
     /* Notify waiters  when the cache becomes
      not empty, or when it becomes full */
     if (m_pos == 1 || (m_waiters && is_full()))
-      m_cv.notify_all();
+      pthread_cond_broadcast(&m_cv);
+    mysql_mutex_unlock(&m_mtx);
   }
 
   /** Check if pointer represents cached element */
@@ -157,16 +172,25 @@ public:
     return ele >= &m_base[0] && ele <= &m_base[capacity() - 1];
   }
 
-  /** Wait until cache is full.*/
-  void wait()
+  /** Wait until cache is full
+  @param m cache mutex (locked) */
+  void wait(mysql_mutex_t &m)
   {
-    std::unique_lock<std::mutex> lk(m_mtx);
+    mysql_mutex_assert_owner(&m);
     m_waiters++;
-    while(!is_full())
-      m_cv.wait(lk);
+    while (!is_full())
+      my_cond_wait(&m_cv, &m.m_mutex);
     m_waiters--;
   }
 
+  /* Wait until cache is full.*/
+  void wait()
+  {
+    mysql_mutex_lock(&m_mtx);
+    wait(m_mtx);
+    mysql_mutex_unlock(&m_mtx);
+  }
+
   /**
    @return approximate number of "borrowed" items.
    A "dirty" read, not used in any critical functionality.