MDEV-24642 Assertion r->emplace... failed in sux_lock::s_lock_register()

In commit 03ca649 (MDEV-24142)
we replaced a debug data structure that holds information about
S-latch holders with a std::set, which does not allow duplicates.

The assertion failed in btr_search_guess_on_hash() in an
s_lock_try() operation.

The reason why recursive S-latch requests are not normally allowed
is that if some other thread has enqueued a waiting X-lock, then
further S-latch requests will block until the exclusive lock has been
granted and released. If a thread were already holding one S-latch
while waiting for the X-latch to be granted and released by another
thread, the two threads would deadlock.

However, the nonblocking s_lock_try() is perfectly fine;
it will immediately return failure in case of conflict.

sux_lock::readers: Use std::unordered_multiset instead of std::set.

sux_lock::s_lock_register(): Allow 'duplicate' requests. Blocking-mode
latch acquisitions are already covered by !have_s() assertions.

sux_lock::s_unlock(): Erase only one element from readers.

buf_page_try_get(): Revert to s_lock_try(). It had been previously
changed to the more intrusive u_lock_try() in response to the
debug check failing.

Author: dr-m

storage/innobase/buf/buf0buf.cc

@@ -3434,12 +3434,12 @@ buf_page_optimistic_get(
 	return(TRUE);
 }
 
-/** Try to U-latch a page.
+/** Try to S-latch a page.
 Suitable for using when holding the lock_sys latches (as it avoids deadlock).
 @param[in]	page_id	page identifier
 @param[in,out]	mtr	mini-transaction
 @return the block
-@retval nullptr if an U-latch cannot be granted immediately */
+@retval nullptr if an S-latch cannot be granted immediately */
 buf_block_t *buf_page_try_get(const page_id_t page_id, mtr_t *mtr)
 {
   ut_ad(mtr);
@@ -3461,16 +3461,13 @@ buf_block_t *buf_page_try_get(const page_id_t page_id, mtr_t *mtr)
   buf_block_buf_fix_inc(block);
   hash_lock->read_unlock();
 
-  /* We will always try to acquire an U latch.
-  In lock_rec_print() we may already be holding an S latch on the page,
-  and recursive S latch acquisition is not allowed. */
-  if (!block->lock.u_lock_try(false))
+  if (!block->lock.s_lock_try())
   {
     buf_block_buf_fix_dec(block);
     return nullptr;
   }
 
-  mtr_memo_push(mtr, block, MTR_MEMO_PAGE_SX_FIX);
+  mtr_memo_push(mtr, block, MTR_MEMO_PAGE_S_FIX);
 
 #ifdef UNIV_DEBUG
   if (!(++buf_dbg_counter % 5771)) buf_pool.validate();

storage/innobase/include/buf0buf.h

@@ -232,12 +232,12 @@ buf_page_optimistic_get(
 	ib_uint64_t	modify_clock,/*!< in: modify clock value */
 	mtr_t*		mtr);	/*!< in: mini-transaction */
 
-/** Try to U-latch a page.
+/** Try to S-latch a page.
 Suitable for using when holding the lock_sys latches (as it avoids deadlock).
 @param[in]	page_id	page identifier
 @param[in,out]	mtr	mini-transaction
 @return the block
-@retval nullptr if an U-latch cannot be granted immediately */
+@retval nullptr if an S-latch cannot be granted immediately */
 buf_block_t *buf_page_try_get(const page_id_t page_id, mtr_t *mtr);
 
 /** Get read access to a compressed page (usually of type

storage/innobase/include/sux_lock.h

@@ -21,7 +21,7 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include "my_atomic_wrapper.h"
 #include "os0thread.h"
 #ifdef UNIV_DEBUG
-# include <set>
+# include <unordered_set>
 #endif
 
 /** A "fat" rw-lock that supports
@@ -48,7 +48,7 @@ class sux_lock final
   /** Protects readers */
   mutable srw_mutex readers_lock;
   /** Threads that hold the lock in shared mode */
-  std::atomic<std::set<os_thread_id_t>*> readers;
+  std::atomic<std::unordered_multiset<os_thread_id_t>*> readers;
 #endif
 
   /** The multiplier in recursive for X locks */
@@ -130,14 +130,15 @@ class sux_lock final
   /** Register the current thread as a holder of a shared lock */
   void s_lock_register()
   {
+    const os_thread_id_t id= os_thread_get_curr_id();
     readers_lock.wr_lock();
     auto r= readers.load(std::memory_order_relaxed);
     if (!r)
     {
-      r= new std::set<os_thread_id_t>();
+      r= new std::unordered_multiset<os_thread_id_t>();
       readers.store(r, std::memory_order_relaxed);
     }
-    ut_ad(r->emplace(os_thread_get_curr_id()).second);
+    r->emplace(id);
     readers_lock.wr_unlock();
   }
 #endif
@@ -218,10 +219,13 @@ class sux_lock final
   void s_unlock()
   {
 #ifdef UNIV_DEBUG
+    const os_thread_id_t id= os_thread_get_curr_id();
     auto r= readers.load(std::memory_order_relaxed);
     ut_ad(r);
     readers_lock.wr_lock();
-    ut_ad(r->erase(os_thread_get_curr_id()) == 1);
+    auto i= r->find(id);
+    ut_ad(i != r->end());
+    r->erase(i);
     readers_lock.wr_unlock();
 #endif
     lock.rd_unlock();