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

This conceptually reverts commit 1fdc161d8f
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 1fdc161d8f seems to have been
do ... while loops, but the 'do' keyword was missing. This total
breakage was missed in commit 260161fc9f
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.

storage/innobase/include/srw_lock.h

@@ -43,11 +43,15 @@ public:
 
 #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 @@ public:
 #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 @@ public:
   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 @@ public:
     }
     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 @@ public:
       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 @@ public:
   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