MDEV-26467: Universally implement spin loop

Previously, neither our wrapper of Microsoft Windows SRWLOCK
nor the futex-less implementation SUX_LOCK_GENERIC supported spin loops.

This was suggested by Vladislav Vaintroub.

storage/innobase/include/srw_lock.h

@@ -22,17 +22,24 @@ this program; if not, write to the Free Software Foundation, Inc.,
 
 #ifdef SUX_LOCK_GENERIC
 /** An exclusive-only variant of srw_lock */
-class srw_mutex final
+template<bool spinloop>
+class srw_mutex_impl final
 {
   pthread_mutex_t lock;
+  void wr_wait();
 public:
   void init() { pthread_mutex_init(&lock, nullptr); }
   void destroy() { pthread_mutex_destroy(&lock); }
-  void wr_lock() { pthread_mutex_lock(&lock); }
+  inline void wr_lock();
   void wr_unlock() { pthread_mutex_unlock(&lock); }
   bool wr_lock_try() { return !pthread_mutex_trylock(&lock); }
 };
-typedef srw_mutex srw_spin_mutex;
+
+template<> void srw_mutex_impl<true>::wr_wait();
+template<>
+inline void srw_mutex_impl<false>::wr_lock() { pthread_mutex_lock(&lock); }
+template<>
+inline void srw_mutex_impl<true>::wr_lock() { if (!wr_lock_try()) wr_wait(); }
 #else
 /** Futex-based mutex */
 template<bool spinloop>
@@ -81,15 +88,12 @@ class srw_mutex_impl final
     }
   }
 };
+#endif
 
 typedef srw_mutex_impl<true> srw_spin_mutex;
 typedef srw_mutex_impl<false> srw_mutex;
-#endif
 
-# if defined _WIN32 || defined SUX_LOCK_GENERIC
-# else
-template<bool spinlock> class srw_lock_impl;
-#endif
+template<bool spinloop> class srw_lock_impl;
 
 /** Slim shared-update-exclusive lock with no recursion */
 template<bool spinloop>
@@ -265,45 +269,54 @@ class ssux_lock_impl final
 #endif
 };
 
-#ifdef _WIN32
+#if defined _WIN32 || defined SUX_LOCK_GENERIC
 /** Slim read-write lock */
-class srw_lock_low
+template<bool spinloop>
+class srw_lock_
 {
 # ifdef UNIV_PFS_RWLOCK
-  friend class srw_lock_impl;
+  friend srw_lock_impl<spinloop>;
 # 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); }
-};
-
-typedef srw_lock_low srw_spin_lock_low;
-#elif defined SUX_LOCK_GENERIC
-/** Slim read-write lock */
-class srw_lock_low
-{
-# ifdef UNIV_PFS_RWLOCK
-  friend class srw_lock_impl;
-# endif
+# else
   rw_lock_t lock;
+# endif
+
+  void rd_wait();
+  void wr_wait();
 public:
-  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); }
+  void init() { IF_WIN(,my_rwlock_init(&lock, nullptr)); }
+  void destroy() { IF_WIN(,rwlock_destroy(&lock)); }
+  inline void rd_lock();
+  inline void wr_lock();
+  bool rd_lock_try()
+  { return IF_WIN(TryAcquireSRWLockShared(&lock), !rw_tryrdlock(&lock)); }
+  void rd_unlock()
+  { IF_WIN(ReleaseSRWLockShared(&lock), rw_unlock(&lock)); }
+  bool wr_lock_try()
+  { return IF_WIN(TryAcquireSRWLockExclusive(&lock), !rw_trywrlock(&lock)); }
+  void wr_unlock()
+  { IF_WIN(ReleaseSRWLockExclusive(&lock), rw_unlock(&lock)); }
 };
-typedef srw_lock_low srw_spin_lock_low;
+
+template<> void srw_lock_<true>::rd_wait();
+template<> void srw_lock_<true>::wr_wait();
+
+template<>
+inline void srw_lock_<false>::rd_lock()
+{ IF_WIN(AcquireSRWLockShared(&lock), rw_rdlock(&lock)); }
+template<>
+inline void srw_lock_<false>::wr_lock()
+{ IF_WIN(AcquireSRWLockExclusive(&lock), rw_wrlock(&lock)); }
+
+template<>
+inline void srw_lock_<true>::rd_lock() { if (!rd_lock_try()) rd_wait(); }
+template<>
+inline void srw_lock_<true>::wr_lock() { if (!wr_lock_try()) wr_wait(); }
+
+typedef srw_lock_<false> srw_lock_low;
+typedef srw_lock_<true> srw_spin_lock_low;
 #else
 typedef ssux_lock_impl<false> srw_lock_low;
 typedef ssux_lock_impl<true> srw_spin_lock_low;
@@ -398,17 +411,14 @@ class ssux_lock
 };
 
 /** Slim reader-writer lock with PERFORMANCE_SCHEMA instrumentation */
-# if defined _WIN32 || defined SUX_LOCK_GENERIC
-# else
-template<bool spinlock>
-# endif
+template<bool spinloop>
 class srw_lock_impl
 {
   PSI_rwlock *pfs_psi;
 # if defined _WIN32 || defined SUX_LOCK_GENERIC
-  srw_lock_low lock;
+  srw_lock_<spinloop> lock;
 # else
-  ssux_lock_impl<spinlock> lock;
+  ssux_lock_impl<spinloop> lock;
 # endif
 
   ATTRIBUTE_NOINLINE void psi_rd_lock(const char *file, unsigned line);
@@ -458,12 +468,7 @@ class srw_lock_impl
   bool wr_lock_try() { return lock.wr_lock_try(); }
 };
 
-# if defined _WIN32 || defined SUX_LOCK_GENERIC
-typedef srw_lock_impl srw_lock;
-typedef srw_lock_impl srw_spin_lock;
-# else
 typedef srw_lock_impl<false> srw_lock;
 typedef srw_lock_impl<true> srw_spin_lock;
-# endif
 
 #endif

storage/innobase/sync/srw_lock.cc

@@ -36,6 +36,20 @@ static inline void srw_pause(unsigned delay)
 }
 
 #ifdef SUX_LOCK_GENERIC
+template<> void srw_mutex_impl<true>::wr_wait()
+{
+  const unsigned delay= srw_pause_delay();
+
+  for (auto spin= srv_n_spin_wait_rounds; spin; spin--)
+  {
+    srw_pause(delay);
+    if (wr_lock_try())
+      return;
+  }
+
+  pthread_mutex_lock(&lock);
+}
+
 template<bool spinloop>
 void ssux_lock_impl<spinloop>::init()
 {
@@ -247,7 +261,6 @@ inline void ssux_lock_impl<spinloop>::wait(uint32_t lk)
 { WaitOnAddress(&readers, &lk, 4, INFINITE); }
 template<bool spinloop>
 void ssux_lock_impl<spinloop>::wake() { WakeByAddressSingle(&readers); }
-
 # else
 #  ifdef __linux__
 #   include <linux/futex.h>
@@ -435,17 +448,44 @@ template void ssux_lock_impl<true>::rd_wait();
 template void ssux_lock_impl<false>::rd_wait();
 #endif /* SUX_LOCK_GENERIC */
 
+#if defined _WIN32 || defined SUX_LOCK_GENERIC
+template<> void srw_lock_<true>::rd_wait()
+{
+  const unsigned delay= srw_pause_delay();
+
+  for (auto spin= srv_n_spin_wait_rounds; spin; spin--)
+  {
+    srw_pause(delay);
+    if (rd_lock_try())
+      return;
+  }
+
+  IF_WIN(AcquireSRWLockShared(&lock), rw_rdlock(&lock));
+}
+
+template<> void srw_lock_<true>::wr_wait()
+{
+  const unsigned delay= srw_pause_delay();
+
+  for (auto spin= srv_n_spin_wait_rounds; spin; spin--)
+  {
+    srw_pause(delay);
+    if (wr_lock_try())
+      return;
+  }
+
+  IF_WIN(AcquireSRWLockExclusive(&lock), rw_wrlock(&lock));
+}
+#endif
+
 #ifdef UNIV_PFS_RWLOCK
-# if defined _WIN32 || defined SUX_LOCK_GENERIC
-#  define void_srw_lock void srw_lock_impl
-# else
-#  define void_srw_lock template<bool spinloop> void srw_lock_impl<spinloop>
 template void srw_lock_impl<false>::psi_rd_lock(const char*, unsigned);
 template void srw_lock_impl<false>::psi_wr_lock(const char*, unsigned);
 template void srw_lock_impl<true>::psi_rd_lock(const char*, unsigned);
 template void srw_lock_impl<true>::psi_wr_lock(const char*, unsigned);
-# endif
-void_srw_lock::psi_rd_lock(const char *file, unsigned line)
+
+template<bool spinloop>
+void srw_lock_impl<spinloop>::psi_rd_lock(const char *file, unsigned line)
 {
   PSI_rwlock_locker_state state;
   const bool nowait= lock.rd_lock_try();
@@ -461,7 +501,8 @@ void_srw_lock::psi_rd_lock(const char *file, unsigned line)
     lock.rd_lock();
 }
 
-void_srw_lock::psi_wr_lock(const char *file, unsigned line)
+template<bool spinloop>
+void srw_lock_impl<spinloop>::psi_wr_lock(const char *file, unsigned line)
 {
   PSI_rwlock_locker_state state;
   const bool nowait= lock.wr_lock_try();