MDEV-24142/MDEV-24167 fixup: Split ssux_lock and srw_lock

This conceptually reverts commit 1fdc161
and reintroduces an option for srw_lock to wrap a native implementation.

The srw_lock and srw_lock_low differ from ssux_lock and ssux_lock_low
in that Slim SUX locks support three modes (Shared, Update, eXclusive)
while Slim RW locks support only two (Read, Write).

On Microsoft Windows, the srw_lock will be implemented by SRWLOCK.
On Linux and OpenBSD, it will be implemented by rw_lock and the
futex system call, just like earlier.
On other systems or if SRW_LOCK_DUMMY is defined on anything else
than Microsoft Windows, rw_lock_t will be used.

ssux_lock_low::read_lock(), ssux_lock_low::update_lock(): Correct
the SRW_LOCK_DUMMY implementation to prevent hangs. The intention of
commit 1fdc161 seems to have been
do ... while loops, but the 'do' keyword was missing. This total
breakage was missed in commit 260161f
which did reduce the probability of the hangs.

ssux_lock_low::u_unlock(): In the SRW_LOCK_DUMMY implementation
(based on a mutex and two condition variables), always invoke
writer_wake() in order to ensure that a waiting update_lock()
will be woken up.

ssux_lock_low::writer_wait(), ssux_lock_low::readers_wait():
In the SRW_LOCK_DUMMY implementation, keep waiting for the signal
until the lock word has changed. The "while" had been changed to "if"
in order to avoid hangs.

Author: dr-m

storage/innobase/include/srw_lock.h

@@ -43,11 +43,15 @@ class srw_mutex
 
 #include "rw_lock.h"
 
-/** Slim reader-writer lock with no recursion */
-class srw_lock_low final : private rw_lock
+/** Slim shared-update-exclusive lock with no recursion */
+class ssux_lock_low final : private rw_lock
 {
 #ifdef UNIV_PFS_RWLOCK
+  friend class ssux_lock;
+# if defined SRW_LOCK_DUMMY || defined _WIN32
+# else
   friend class srw_lock;
+# endif
 #endif
 #ifdef SRW_LOCK_DUMMY
   pthread_mutex_t mutex;
@@ -85,14 +89,10 @@ class srw_lock_low final : private rw_lock
 #endif
   bool rd_lock_try() { uint32_t l; return read_trylock(l); }
   bool wr_lock_try() { return write_trylock(); }
-  /** @tparam support_u_lock dummy parameter for UNIV_PFS_RWLOCK */
-  template<bool support_u_lock= false>
   void rd_lock() { uint32_t l; if (!read_trylock(l)) read_lock(l); }
   void u_lock() { uint32_t l; if (!update_trylock(l)) update_lock(l); }
   bool u_lock_try() { uint32_t l; return update_trylock(l); }
   void u_wr_upgrade() { if (!upgrade_trylock()) write_lock(true); }
-  /** @tparam support_u_lock dummy parameter for UNIV_PFS_RWLOCK */
-  template<bool support_u_lock= false>
   void wr_lock() { if (!write_trylock()) write_lock(false); }
   void rd_unlock();
   void u_unlock();
@@ -101,25 +101,58 @@ class srw_lock_low final : private rw_lock
   bool is_waiting() const { return value() & WRITER_WAITING; }
 };
 
+#if defined SRW_LOCK_DUMMY || defined _WIN32
+/** Slim read-write lock */
+class srw_lock_low
+{
+# ifdef UNIV_PFS_RWLOCK
+  friend class srw_lock;
+# endif
+# ifdef _WIN32
+  SRWLOCK lock;
+public:
+  void init() {}
+  void destroy() {}
+  void rd_lock() { AcquireSRWLockShared(&lock); }
+  bool rd_lock_try() { return TryAcquireSRWLockShared(&lock); }
+  void rd_unlock() { ReleaseSRWLockShared(&lock); }
+  void wr_lock() { AcquireSRWLockExclusive(&lock); }
+  bool wr_lock_try() { return TryAcquireSRWLockExclusive(&lock); }
+  void wr_unlock() { ReleaseSRWLockExclusive(&lock); }
+# else
+  rw_lock_t lock;
+  void init() { my_rwlock_init(&lock, nullptr); }
+  void destroy() { rwlock_destroy(&lock); }
+  void rd_lock() { rw_rdlock(&lock); }
+  bool rd_lock_try() { return rw_tryrdlock(&lock); }
+  void rd_unlock() { rw_unlock(&lock); }
+  void wr_lock() { rw_wrlock(&lock); }
+  bool wr_lock_try() { return rw_trywrlock(&lock); }
+  void wr_unlock() { rw_unlock(&lock); }
+# endif
+};
+#else
+typedef ssux_lock_low srw_lock_low;
+#endif
+
 #ifndef UNIV_PFS_RWLOCK
 # define SRW_LOCK_INIT(key) init()
 # define SRW_LOCK_ARGS(file, line) /* nothing */
 # define SRW_LOCK_CALL /* nothing */
 typedef srw_lock_low srw_lock;
+typedef ssux_lock_low ssux_lock;
 #else
 # define SRW_LOCK_INIT(key) init(key)
 # define SRW_LOCK_ARGS(file, line) file, line
 # define SRW_LOCK_CALL __FILE__, __LINE__
 
-/** Slim reader-writer lock with PERFORMANCE_SCHEMA instrumentation */
-class srw_lock
+/** Slim shared-update-exclusive lock with PERFORMANCE_SCHEMA instrumentation */
+class ssux_lock
 {
   PSI_rwlock *pfs_psi;
-  srw_lock_low lock;
+  ssux_lock_low lock;
 
-  template<bool support_u_lock>
   ATTRIBUTE_NOINLINE void psi_rd_lock(const char *file, unsigned line);
-  template<bool support_u_lock>
   ATTRIBUTE_NOINLINE void psi_wr_lock(const char *file, unsigned line);
   ATTRIBUTE_NOINLINE void psi_u_lock(const char *file, unsigned line);
   ATTRIBUTE_NOINLINE void psi_u_wr_upgrade(const char *file, unsigned line);
@@ -138,11 +171,10 @@ class srw_lock
     }
     lock.destroy();
   }
-  template<bool support_u_lock= false>
   void rd_lock(const char *file, unsigned line)
   {
     if (psi_likely(pfs_psi != nullptr))
-      psi_rd_lock<support_u_lock>(file, line);
+      psi_rd_lock(file, line);
     else
       lock.rd_lock();
   }
@@ -165,11 +197,10 @@ class srw_lock
       PSI_RWLOCK_CALL(unlock_rwlock)(pfs_psi);
     lock.u_unlock();
   }
-  template<bool support_u_lock= false>
   void wr_lock(const char *file, unsigned line)
   {
     if (psi_likely(pfs_psi != nullptr))
-      psi_wr_lock<support_u_lock>(file, line);
+      psi_wr_lock(file, line);
     else
       lock.wr_lock();
   }
@@ -191,4 +222,57 @@ class srw_lock
   bool wr_lock_try() { return lock.wr_lock_try(); }
   bool is_waiting() const { return lock.is_waiting(); }
 };
+
+/** Slim reader-writer lock with PERFORMANCE_SCHEMA instrumentation */
+class srw_lock
+{
+  PSI_rwlock *pfs_psi;
+  srw_lock_low lock;
+
+  ATTRIBUTE_NOINLINE void psi_rd_lock(const char *file, unsigned line);
+  ATTRIBUTE_NOINLINE void psi_wr_lock(const char *file, unsigned line);
+public:
+  void init(mysql_pfs_key_t key)
+  {
+    pfs_psi= PSI_RWLOCK_CALL(init_rwlock)(key, this);
+    lock.init();
+  }
+  void destroy()
+  {
+    if (psi_likely(pfs_psi != nullptr))
+    {
+      PSI_RWLOCK_CALL(destroy_rwlock)(pfs_psi);
+      pfs_psi= nullptr;
+    }
+    lock.destroy();
+  }
+  void rd_lock(const char *file, unsigned line)
+  {
+    if (psi_likely(pfs_psi != nullptr))
+      psi_rd_lock(file, line);
+    else
+      lock.rd_lock();
+  }
+  void rd_unlock()
+  {
+    if (psi_likely(pfs_psi != nullptr))
+      PSI_RWLOCK_CALL(unlock_rwlock)(pfs_psi);
+    lock.rd_unlock();
+  }
+  void wr_lock(const char *file, unsigned line)
+  {
+    if (psi_likely(pfs_psi != nullptr))
+      psi_wr_lock(file, line);
+    else
+      lock.wr_lock();
+  }
+  void wr_unlock()
+  {
+    if (psi_likely(pfs_psi != nullptr))
+      PSI_RWLOCK_CALL(unlock_rwlock)(pfs_psi);
+    lock.wr_unlock();
+  }
+  bool rd_lock_try() { return lock.rd_lock_try(); }
+  bool wr_lock_try() { return lock.wr_lock_try(); }
+};
 #endif

storage/innobase/include/sux_lock.h

@@ -27,7 +27,7 @@ this program; if not, write to the Free Software Foundation, Inc.,
 /** A "fat" rw-lock that supports
 S (shared), U (update, or shared-exclusive), and X (exclusive) modes
 as well as recursive U and X latch acquisition
-@tparam srw srw_lock_low or srw_lock */
+@tparam srw ssux_lock_low or ssux_lock */
 template<typename srw>
 class sux_lock final
 {
@@ -259,19 +259,19 @@ class sux_lock final
 };
 
 /** needed for dict_index_t::clone() */
-template<> inline void sux_lock<srw_lock>::operator=(const sux_lock&)
+template<> inline void sux_lock<ssux_lock>::operator=(const sux_lock&)
 {
   memset((void*) this, 0, sizeof *this);
 }
 
-typedef sux_lock<srw_lock_low> block_lock;
+typedef sux_lock<ssux_lock_low> block_lock;
 
 #ifndef UNIV_PFS_RWLOCK
 typedef block_lock index_lock;
 #else
-typedef sux_lock<srw_lock> index_lock;
+typedef sux_lock<ssux_lock> index_lock;
 
-template<> inline void sux_lock<srw_lock_low>::init()
+template<> inline void sux_lock<ssux_lock_low>::init()
 {
   lock.init();
   ut_ad(!writer.load(std::memory_order_relaxed));
@@ -281,16 +281,16 @@ template<> inline void sux_lock<srw_lock_low>::init()
 }
 
 template<>
-inline void sux_lock<srw_lock>::s_lock(const char *file, unsigned line)
+inline void sux_lock<ssux_lock>::s_lock(const char *file, unsigned line)
 {
   ut_ad(!have_x());
   ut_ad(!have_s());
-  lock.template rd_lock<true>(file, line);
+  lock.rd_lock(file, line);
   ut_d(s_lock_register());
 }
 
 template<>
-inline void sux_lock<srw_lock>::u_lock(const char *file, unsigned line)
+inline void sux_lock<ssux_lock>::u_lock(const char *file, unsigned line)
 {
   os_thread_id_t id= os_thread_get_curr_id();
   if (writer.load(std::memory_order_relaxed) == id)
@@ -305,22 +305,22 @@ inline void sux_lock<srw_lock>::u_lock(const char *file, unsigned line)
 }
 
 template<>
-inline void sux_lock<srw_lock>::x_lock(const char *file, unsigned line)
+inline void sux_lock<ssux_lock>::x_lock(const char *file, unsigned line)
 {
   os_thread_id_t id= os_thread_get_curr_id();
   if (writer.load(std::memory_order_relaxed) == id)
     writer_recurse<false>();
   else
   {
-    lock.template wr_lock<true>(file, line);
+    lock.wr_lock(file, line);
     ut_ad(!recursive);
     recursive= RECURSIVE_X;
     set_first_owner(id);
   }
 }
 
 template<>
-inline void sux_lock<srw_lock>::u_x_upgrade(const char *file, unsigned line)
+inline void sux_lock<ssux_lock>::u_x_upgrade(const char *file, unsigned line)
 {
   ut_ad(have_u_not_x());
   lock.u_wr_upgrade(file, line);
@@ -329,16 +329,16 @@ inline void sux_lock<srw_lock>::u_x_upgrade(const char *file, unsigned line)
 #endif
 
 template<>
-inline void sux_lock<srw_lock_low>::s_lock()
+inline void sux_lock<ssux_lock_low>::s_lock()
 {
   ut_ad(!have_x());
   ut_ad(!have_s());
-  lock.template rd_lock<true>();
+  lock.rd_lock();
   ut_d(s_lock_register());
 }
 
 template<>
-inline void sux_lock<srw_lock_low>::u_lock()
+inline void sux_lock<ssux_lock_low>::u_lock()
 {
   os_thread_id_t id= os_thread_get_curr_id();
   if (writer.load(std::memory_order_relaxed) == id)
@@ -353,7 +353,7 @@ inline void sux_lock<srw_lock_low>::u_lock()
 }
 
 template<>
-inline void sux_lock<srw_lock_low>::x_lock(bool for_io)
+inline void sux_lock<ssux_lock_low>::x_lock(bool for_io)
 {
   os_thread_id_t id= os_thread_get_curr_id();
   if (writer.load(std::memory_order_relaxed) == id)
@@ -363,22 +363,22 @@ inline void sux_lock<srw_lock_low>::x_lock(bool for_io)
   }
   else
   {
-    lock.template wr_lock<true>();
+    lock.wr_lock();
     ut_ad(!recursive);
     recursive= RECURSIVE_X;
     set_first_owner(for_io ? FOR_IO : id);
   }
 }
 
 template<>
-inline void sux_lock<srw_lock_low>::u_x_upgrade()
+inline void sux_lock<ssux_lock_low>::u_x_upgrade()
 {
   ut_ad(have_u_not_x());
   lock.u_wr_upgrade();
   recursive/= RECURSIVE_U;
 }
 
-template<> inline bool sux_lock<srw_lock_low>::x_lock_upgraded()
+template<> inline bool sux_lock<ssux_lock_low>::x_lock_upgraded()
 {
   os_thread_id_t id= os_thread_get_curr_id();
   if (writer.load(std::memory_order_relaxed) == id)
@@ -397,7 +397,7 @@ template<> inline bool sux_lock<srw_lock_low>::x_lock_upgraded()
   }
   else
   {
-    lock.template wr_lock<true>();
+    lock.wr_lock();
     ut_ad(!recursive);
     recursive= RECURSIVE_X;
     set_first_owner(id);
@@ -406,7 +406,7 @@ template<> inline bool sux_lock<srw_lock_low>::x_lock_upgraded()
 }
 
 template<>
-inline bool sux_lock<srw_lock_low>::u_lock_try(bool for_io)
+inline bool sux_lock<ssux_lock_low>::u_lock_try(bool for_io)
 {
   os_thread_id_t id= os_thread_get_curr_id();
   if (writer.load(std::memory_order_relaxed) == id)
@@ -427,7 +427,7 @@ inline bool sux_lock<srw_lock_low>::u_lock_try(bool for_io)
 }
 
 template<>
-inline bool sux_lock<srw_lock_low>::x_lock_try()
+inline bool sux_lock<ssux_lock_low>::x_lock_try()
 {
   os_thread_id_t id= os_thread_get_curr_id();
   if (writer.load(std::memory_order_relaxed) == id)