MDEV-37482: Contention on btr_sea::partition::latch

To reduce contention between insert, erase and search, let us mimic
commit b08448d (MDEV-20612).
That is, btr_sea::partition::insert() and btr_sea::partition::erase()
will use a combination of a shared btr_sea::partition::latch and a
tiny page_hash_latch that is pushed down to the btr_sea::hash_table::array.

An exclusive btr_sea::partition::latch will be used in the final part of
btr_search_drop_page_hash_index(), where we must guarantee that all
entries will be removed, as well as in operations that affect an entire
adaptive hash index partition.

btr_sea::hash_chain: Chain of ahi_node hash buckets.

btr_sea::hash_table: A hash table that includes page_hash_latch
interleaved with hash_chain.

page_hash_latch::try_lock(): Attempt to acquire an exclusive latch
without waiting.

btr_search_guess_on_hash(): Acquire also the page_hash_latch in order
to prevent a concurrent modification of the hash bucket chain that our
lookup is traversing.

ha_insert_for_fold(): Remove. Invoke btr_sea::partition::insert() directly.

btr_sea::partition::erase(): Add template<bool ex> for indicating whether
an exclusive or a shared btr_sea::partition::latch is being held.
If the ex=false operation fails to free the memory,
btr_search_update_hash_on_delete() will retry with ex=true.

btr_sea::partition::cleanup_after_erase(): Add an overload for the case
where instead of holding an exclusive latch, we hold a shared latch
along with a page_hash_latch. When not holding an exclusve latch,
we may fail to free the memory, and the caller has to retry with an
exclusive latch.

btr_sea::partition::cleanup_after_erase_start(),
btr_sea::partition::cleanup_after_erase_finish(): Split from
cleanup_after_erase() to reduce the amount of code duplication.

btr_sea::partition::block_mutex: Protect only the linked list of blocks.
The spare block will exclusively be updated via Atomic_relaxed::exchange().

btr_sea::partition::rollback_insert(): Free the spare block in the unlikely
event that the adaptive hash index has been disabled after our invocation
of btr_sea::partition::prepare_insert().

ha_remove_all_nodes_to_page(): Merged to the only caller
btr_search_drop_page_hash_index().

ssux_lock_impl::wr_rd_downgrade(): Downgrade an X latch to S.

srw_lock_debug::wr_rd_downgrade(), srw_lock_impl::wr_rd_downgrade():
Downgrade from exclusive to shared. This operation is unavailable
if _WIN32 or SUX_LOCK_GENERIC is defined.

btr_search_build_page_hash_index(): Downgrade from exclusive to shared
part.latch before starting to insert records into the adaptive hash index.
In multi-batch operation, we preserve the last fr[] value in order to
ensure the correct operation when buf_block_t::LEFT_SIDE is not set.

Author: dr-m

storage/innobase/btr/btr0sea.cc

@@ -119,14 +119,107 @@ struct btr_sea
   @param resize whether buf_pool_t::resize() is the caller */
   ATTRIBUTE_COLD void enable(bool resize= false) noexcept;
 
+  /** Hash cell chain in hash_table */
+  struct hash_chain
+  {
+    /** pointer to the first block */
+    ahi_node *first;
+
+    /** Find an element.
+    @param u   unary predicate
+    @return the first matching element
+    @retval nullptr if not found */
+    template<typename UnaryPred>
+    inline ahi_node *find(UnaryPred u) const noexcept;
+
+    /** Search for a pointer to an element.
+    @param u   unary predicate
+    @return pointer to the first matching element,
+    or to the last element in the chain */
+    template<typename UnaryPred>
+    inline ahi_node **search(UnaryPred u) noexcept;
+  };
+
+  /** Hash table with singly-linked overflow lists.
+  Based on @see buf_pool_t::page_hash_table */
+  struct hash_table
+  {
+    static_assert(CPU_LEVEL1_DCACHE_LINESIZE >= 64, "less than 64 bytes");
+    static_assert(!(CPU_LEVEL1_DCACHE_LINESIZE & 63),
+      "not a multiple of 64 bytes");
+
+    /** Number of array[] elements per page_hash_latch.
+    Must be one less than a power of 2. */
+#if 0
+    static constexpr size_t ELEMENTS_PER_LATCH= 64 / sizeof(void*) - 1;
+
+    /** Extra padding. FIXME: Is this ever useful to be nonzero?
+    Long time ago, some testing on an ARMv8 implementation seemed
+    to suggest so, but this has not been validated recently. */
+    static constexpr size_t EMPTY_SLOTS_PER_LATCH=
+      ((CPU_LEVEL1_DCACHE_LINESIZE / 64) - 1) * (64 / sizeof(void*));
+#else
+    static constexpr size_t ELEMENTS_PER_LATCH=
+      CPU_LEVEL1_DCACHE_LINESIZE / sizeof(void*) - 1;
+    static constexpr size_t EMPTY_SLOTS_PER_LATCH= 0;
+#endif
+
+    /** number of payload elements in array[] */
+    Atomic_relaxed<ulint> n_cells;
+    /** the hash table, with pad(n_cells) elements, aligned to L1 cache size */
+    hash_chain *array;
+
+    /** Create the hash table.
+    @param n  the lower bound of n_cells */
+    inline void create(ulint n) noexcept;
+
+    /** Free the hash table. */
+    void free() noexcept { aligned_free(array); array= nullptr; }
+
+    /** @return the index of an array element */
+    ulint calc_hash(ulint fold) const noexcept
+    { return calc_hash(fold, n_cells); }
+    /** @return raw array index converted to padded index */
+    static ulint pad(ulint h) noexcept
+    {
+      ulint latches= h / ELEMENTS_PER_LATCH;
+      ulint empty_slots= latches * EMPTY_SLOTS_PER_LATCH;
+      return 1 + latches + empty_slots + h;
+    }
+  private:
+    /** @return the index of an array element */
+    static ulint calc_hash(ulint fold, ulint n_cells) noexcept
+    {
+      return pad(fold % n_cells);
+    }
+  public:
+    /** @return the latch covering a hash table chain */
+    static page_hash_latch &lock_get(hash_chain &chain) noexcept
+    {
+      static_assert(!((ELEMENTS_PER_LATCH + 1) & ELEMENTS_PER_LATCH),
+                    "must be one less than a power of 2");
+      const size_t addr= reinterpret_cast<size_t>(&chain);
+      ut_ad(addr & (ELEMENTS_PER_LATCH * sizeof chain));
+      return *reinterpret_cast<page_hash_latch*>
+        (addr & ~(ELEMENTS_PER_LATCH * sizeof chain));
+    }
+
+    /** Get a hash table slot. */
+    hash_chain &cell_get(ulint fold) const
+    { return array[calc_hash(fold, n_cells)]; }
+  };
+
   /** Partition of the hash table */
   struct partition
   {
-    /** latch protecting table */
-    alignas(CPU_LEVEL1_DCACHE_LINESIZE) srw_spin_lock latch;
+    /** latch protecting table: either an exclusive latch, or
+    a shared latch combined with lock_get() */
+    alignas(CPU_LEVEL1_DCACHE_LINESIZE)
+    IF_DBUG(srw_lock_debug,srw_spin_lock) latch;
     /** map of CRC-32C of rec prefix to rec_t* in buf_page_t::frame */
-    hash_table_t table;
-    /** latch protecting blocks, spare; may be acquired while holding latch */
+    hash_table table;
+    /** protects blocks; acquired while holding latch
+    and possibly table.lock_get() */
     srw_mutex blocks_mutex;
     /** allocated blocks */
     UT_LIST_BASE_NODE_T(buf_page_t) blocks;
@@ -141,15 +234,50 @@ struct btr_sea
 
     inline void free() noexcept;
 
+    /** @return the number of allocated buffer pool blocks */
+    TPOOL_SUPPRESS_TSAN size_t get_blocks() const noexcept
+    { return UT_LIST_GET_LEN(blocks) + !!spare; }
+
     /** Ensure that there is a spare block for a future insert() */
     void prepare_insert() noexcept;
 
-    /** Clean up after erasing an AHI node
-    @param erase   node being erased
+    /** Undo prepare_insert() in case !btr_search.enabled */
+    void rollback_insert() noexcept;
+
+  private:
+    /** Start cleanup_after_erase()
+    @return the last allocated element */
+    inline ahi_node *cleanup_after_erase_start() noexcept;
+    /** Finish cleanup_after_erase().
+    We reduce the allocated size in UT_LIST_GET_LAST(blocks)->free_offset.
+    If that size reaches 0, the last block will be removed from blocks,
+    and a block may have to be freed by our caller.
+    @return buffer block to be freed
+    @retval nullptr if no buffer block was freed */
+    buf_block_t *cleanup_after_erase_finish() noexcept;
+  public:
+    __attribute__((nonnull))
+    /** Clean up after erasing an AHI node, while the caller is
+    holding an exclusive latch. Unless "erase" is the last allocated
+    element, we will swap it with the last allocated element.
+    Finally, we return via cleanup_after_erase_finish().
+    @param erase node being erased
     @return buffer block to be freed
     @retval nullptr if no buffer block was freed */
     buf_block_t *cleanup_after_erase(ahi_node *erase) noexcept;
 
+    __attribute__((nonnull))
+    /** Clean up after erasing an AHI node. This is similar to
+    cleanup_after_erase(ahi_node*), except that the operation may fail.
+    @param erase   node being erased
+    @param l       the latch held together with shared latch
+    @return buffer block to be freed
+    @retval nullptr if no buffer block was freed
+    @retval -1     if we fail to shrink the allocation and erasing
+                   needs to be retried while holding an exclusive latch */
+    buf_block_t *cleanup_after_erase(ahi_node *erase, page_hash_latch *l)
+      noexcept;
+
     __attribute__((nonnull))
 # if defined UNIV_AHI_DEBUG || defined UNIV_DEBUG
     /** Insert or replace an entry into the hash table.
@@ -164,11 +292,23 @@ struct btr_sea
     void insert(uint32_t fold, const rec_t *rec) noexcept;
 # endif
 
-    /** Delete a pointer to a record if it exists.
-    @param fold  CRC-32C of rec prefix
+    /** erase() return value */
+    enum erase_status{
+      /** must retry with exclusive latch */
+      ERASE_RETRY= -1,
+      /** the pointer to the record was erased */
+      ERASED= 0,
+      /** nothing was erased */
+      NOT_ERASED= 1
+    };
+
+    /** Delete a pointer to a record if it exists, and release the latch.
+    @tparam ex   true=holding exclusive latch, false=shared latch
+    @param cell  hash table cell that may contain the CRC-32C of rec prefix
     @param rec   B-tree leaf page record
-    @return whether a record existed and was removed */
-    inline bool erase(uint32_t fold, const rec_t *rec) noexcept;
+    @return status */
+    template<bool ex>
+    erase_status erase(hash_chain &cell, const rec_t *rec) noexcept;
   };
 
   /** innodb_adaptive_hash_index_parts */

storage/innobase/include/btr0sea.h

@@ -473,7 +473,8 @@ class buf_page_t
     protected by buf_pool.page_hash.lock_get() */
     buf_page_t *hash;
     /** for state()==MEMORY that are part of recv_sys.pages and
-    protected by recv_sys.mutex */
+    protected by recv_sys.mutex, or part of btr_sea::partition::table
+    and protected by btr_sea::partition::blocks_mutex */
     struct {
       /** number of recv_sys.pages entries stored in the block */
       uint16_t used_records;
@@ -868,7 +869,7 @@ struct buf_block_t{
   Atomic_relaxed<uint16_t> n_hash_helps;
 # if defined UNIV_AHI_DEBUG || defined UNIV_DEBUG
   /** number of pointers from the btr_sea::partition::table;
-  !n_pointers == !index */
+  !index implies n_pointers == 0 */
   Atomic_counter<uint16_t> n_pointers;
 #  define assert_block_ahi_empty(block) ut_a(!(block)->n_pointers)
 #  define assert_block_ahi_valid(b) ut_a((b)->index || !(b)->n_pointers)
@@ -878,7 +879,7 @@ struct buf_block_t{
 # endif /* UNIV_AHI_DEBUG || UNIV_DEBUG */
   /** index for which the adaptive hash index has been created,
   or nullptr if the page does not exist in the index.
-  Protected by btr_sea::partition::latch. */
+  May be modified while holding exclusive btr_sea::partition::latch. */
   Atomic_relaxed<dict_index_t*> index;
   /* @} */
 #else /* BTR_CUR_HASH_ADAPT */

storage/innobase/include/buf0buf.h

@@ -192,6 +192,8 @@ class page_hash_latch : private rw_lock
   inline void lock_shared() noexcept;
   /** Acquire an exclusive lock */
   inline void lock() noexcept;
+  /** @return whether an exclusive lock was acquired without waiting */
+  bool try_lock() noexcept { return write_trylock(); }
 
   /** @return whether an exclusive lock is being held by any thread */
   bool is_write_locked() const noexcept { return rw_lock::is_write_locked(); }
@@ -215,6 +217,7 @@ class page_hash_latch
   void lock_shared() noexcept { lk.rd_lock(); }
   void unlock_shared() noexcept { lk.rd_unlock(); }
   void lock() noexcept { lk.wr_lock(); }
+  bool try_lock() noexcept { return lk.wr_lock_try(); }
   void unlock() noexcept { lk.wr_unlock(); }
   bool is_write_locked() const noexcept { return lk.is_write_locked(); }
   bool is_locked() const noexcept { return lk.is_locked(); }
@@ -229,6 +232,7 @@ class page_hash_latch
   void lock_shared() noexcept { lock(); }
   void unlock_shared() noexcept { unlock(); }
   void lock() noexcept { lk.wr_lock(); }
+  bool try_lock() noexcept { return lk.wr_lock_try(); }
   void unlock() noexcept { lk.wr_unlock(); }
   bool is_locked() const noexcept { return lk.is_locked(); }
   bool is_write_locked() const noexcept { return is_locked(); }

storage/innobase/include/buf0types.h

@@ -34,6 +34,16 @@ this program; if not, write to the Free Software Foundation, Inc.,
 # define SUX_LOCK_GENERIC /* Use dummy implementation for debugging purposes */
 #endif
 
+#ifndef UNIV_PFS_RWLOCK
+# define SRW_LOCK_INIT(key) init()
+# define SRW_LOCK_ARGS(file, line) /* nothing */
+# define SRW_LOCK_CALL /* nothing */
+#else
+# define SRW_LOCK_INIT(key) init(key)
+# define SRW_LOCK_ARGS(file, line) file, line
+# define SRW_LOCK_CALL __FILE__, __LINE__
+#endif
+
 /** An exclusive-only variant of srw_lock */
 template<bool spinloop>
 class pthread_mutex_wrapper final
@@ -310,6 +320,7 @@ class ssux_lock_impl
     ut_ad(lk < WRITER);
     u_unlock();
   }
+  void wr_rd_downgrade() noexcept { wr_u_downgrade(); u_rd_downgrade(); }
 
   void rd_unlock() noexcept
   {
@@ -415,16 +426,9 @@ typedef ssux_lock_impl<true> srw_spin_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 srw_spin_lock_low srw_spin_lock;
 #else
-# define SRW_LOCK_INIT(key) init(key)
-# define SRW_LOCK_ARGS(file, line) file, line
-# define SRW_LOCK_CALL __FILE__, __LINE__
-
 /** Slim shared-update-exclusive lock with PERFORMANCE_SCHEMA instrumentation */
 class ssux_lock
 {
@@ -556,6 +560,23 @@ class srw_lock_impl
       PSI_RWLOCK_CALL(unlock_rwlock)(pfs_psi);
     lock.wr_unlock();
   }
+# if defined _WIN32 || defined SUX_LOCK_GENERIC
+# else
+  void wr_rd_downgrade(const char *file, unsigned line) noexcept
+  {
+    if (psi_likely(pfs_psi != nullptr))
+    {
+      PSI_RWLOCK_CALL(unlock_rwlock)(pfs_psi);
+      PSI_rwlock_locker_state state;
+      if (PSI_rwlock_locker *locker=
+          PSI_RWLOCK_CALL(start_rwlock_rdwait)
+          (&state, pfs_psi, PSI_RWLOCK_READLOCK, file, line))
+        PSI_RWLOCK_CALL(end_rwlock_rdwait)(locker, 0);
+    }
+
+    lock.wr_rd_downgrade();
+  }
+#endif
   bool rd_lock_try() noexcept { return lock.rd_lock_try(); }
   bool wr_lock_try() noexcept { return lock.wr_lock_try(); }
   void lock_shared() noexcept { return rd_lock(SRW_LOCK_CALL); }
@@ -595,20 +616,25 @@ class srw_lock_debug : private srw_lock
   void SRW_LOCK_INIT(mysql_pfs_key_t key) noexcept;
   void destroy() noexcept;
 
-#ifndef SUX_LOCK_GENERIC
+# ifndef SUX_LOCK_GENERIC
   /** @return whether any lock may be held by any thread */
   bool is_locked_or_waiting() const noexcept
   { return srw_lock::is_locked_or_waiting(); }
   /** @return whether an exclusive lock may be held by any thread */
   bool is_write_locked() const noexcept { return srw_lock::is_write_locked(); }
-#endif
+# endif
 
   /** Acquire an exclusive lock */
   void wr_lock(SRW_LOCK_ARGS(const char *file, unsigned line)) noexcept;
   /** @return whether an exclusive lock was acquired */
   bool wr_lock_try() noexcept;
   /** Release after wr_lock() */
   void wr_unlock() noexcept;
+# if defined _WIN32 || defined SUX_LOCK_GENERIC
+# else
+  /** Downgrade wr_lock() to rd_lock() */
+  void wr_rd_downgrade(SRW_LOCK_ARGS(const char*,unsigned)) noexcept;
+# endif
   /** Acquire a shared lock */
   void rd_lock(SRW_LOCK_ARGS(const char *file, unsigned line)) noexcept;
   /** @return whether a shared lock was acquired */

storage/innobase/include/srw_lock.h

@@ -748,7 +748,8 @@ srv_printf_innodb_monitor(
 			part.blocks_mutex.wr_lock();
 			fprintf(file, "Hash table size " ULINTPF
 				", node heap has " ULINTPF " buffer(s)\n",
-				part.table.n_cells, part.blocks.count + !!part.spare);
+				size_t{part.table.n_cells},
+				part.blocks.count + !!part.spare);
 			part.blocks_mutex.wr_unlock();
 		}
 
@@ -830,10 +831,7 @@ srv_export_innodb_status(void)
 
 	ulint mem_adaptive_hash = 0;
 	for (ulong i = 0; i < btr_search.n_parts; i++) {
-		btr_sea::partition& part= btr_search.parts[i];
-		part.blocks_mutex.wr_lock();
-		mem_adaptive_hash += part.blocks.count + !!part.spare;
-		part.blocks_mutex.wr_unlock();
+		mem_adaptive_hash += btr_search.parts[i].get_blocks();
 	}
 	mem_adaptive_hash <<= srv_page_size_shift;
 	btr_search.parts[0].latch.rd_lock(SRW_LOCK_CALL);

storage/innobase/srv/srv0srv.cc

@@ -669,6 +669,18 @@ void srw_lock_debug::wr_unlock() noexcept
   srw_lock::wr_unlock();
 }
 
+# if defined _WIN32 || defined SUX_LOCK_GENERIC
+# else
+void srw_lock_debug::wr_rd_downgrade
+(SRW_LOCK_ARGS(const char *file, unsigned line)) noexcept
+{
+  ut_ad(have_wr());
+  writer.store(0, std::memory_order_relaxed);
+  readers_register();
+  srw_lock::wr_rd_downgrade(SRW_LOCK_ARGS(file, line));
+}
+# endif
+
 void srw_lock_debug::readers_register() noexcept
 {
   readers_lock.wr_lock();