MDEV-20612: Enable concurrent lock_release()

lock_release_try(): Try to release locks while only holding
shared lock_sys.latch.

lock_release(): If 5 attempts of lock_release_try() fail,
proceed to acquire exclusive lock_sys.latch.

storage/innobase/include/dict0mem.h

@@ -1972,6 +1972,14 @@ struct dict_table_t {
     lock_mutex.wr_lock();
     ut_ad(!lock_mutex_owner.exchange(os_thread_get_curr_id()));
   }
+  /** Try to acquire lock_mutex */
+  bool lock_mutex_trylock()
+  {
+    ut_ad(!lock_mutex_is_owner());
+    bool acquired= lock_mutex.wr_lock_try();
+    ut_ad(!acquired || !lock_mutex_owner.exchange(os_thread_get_curr_id()));
+    return acquired;
+  }
   /** Release lock_mutex */
   void lock_mutex_unlock()
   {

storage/innobase/include/lock0lock.h

@@ -581,8 +581,10 @@ class lock_sys_t
   {
     /** Wait for an exclusive lock */
     void wait();
+    /** Try to acquire a lock */
+    bool try_acquire() { return write_trylock(); }
     /** Acquire a lock */
-    void acquire() { if (!write_trylock()) wait(); }
+    void acquire() { if (!try_acquire()) wait(); }
     /** Release a lock */
     void release();
 #else
@@ -590,6 +592,8 @@ class lock_sys_t
   private:
     srw_lock_low lock;
   public:
+    /** Try to acquire a lock */
+    bool try_acquire() { return lock.wr_lock_try(); }
     /** Acquire a lock */
     void acquire() { lock.wr_lock(); }
     /** Release a lock */

storage/innobase/lock/lock0lock.cc

@@ -3939,12 +3939,10 @@ released:
 }
 
 /** Release the explicit locks of a committing transaction,
-and release possible other transactions waiting because of these locks. */
-void lock_release(trx_t *trx)
+and release possible other transactions waiting because of these locks.
+@return whether the operation succeeded */
+static bool lock_release_try(trx_t *trx)
 {
-  ulint count= 0;
-
-  ut_ad(!trx->mutex_is_owner());
   /* At this point, trx->lock.trx_locks cannot be modified by other
   threads, because our transaction has been committed.
   See the checks and assertions in lock_rec_create_low() and
@@ -3956,11 +3954,82 @@ void lock_release(trx_t *trx)
   DBUG_ASSERT(trx->state == TRX_STATE_COMMITTED_IN_MEMORY);
   DBUG_ASSERT(!trx->is_referenced());
 
-  LockMutexGuard g{SRW_LOCK_CALL};
+  bool all_released= true;
+restart:
+  ulint count= 1000;
+  lock_sys.rd_lock(SRW_LOCK_CALL);
   trx->mutex_lock();
 
-  for (lock_t *lock= UT_LIST_GET_LAST(trx->lock.trx_locks); lock;
-       lock= UT_LIST_GET_LAST(trx->lock.trx_locks))
+  /* Note: Anywhere else, trx->mutex is not held while acquiring
+  a lock table latch, but here we are following the opposite order.
+  To avoid deadlocks, we only try to acquire the lock table latches
+  but not keep waiting for them. */
+
+  for (lock_t *lock= UT_LIST_GET_LAST(trx->lock.trx_locks); lock; )
+  {
+    ut_ad(lock->trx == trx);
+    lock_t *prev= UT_LIST_GET_PREV(trx_locks, lock);
+    if (!lock->is_table())
+    {
+      ut_ad(!lock->index->table->is_temporary());
+      ut_ad(lock->mode() != LOCK_X ||
+            lock->index->table->id >= DICT_HDR_FIRST_ID ||
+            trx->dict_operation);
+      auto &lock_hash= lock_sys.hash_get(lock->type_mode);
+      auto latch= lock_hash.lock_get(lock->un_member.rec_lock.page_id.fold());
+      if (!latch->try_acquire())
+        all_released= false;
+      else
+      {
+        lock_rec_dequeue_from_page(lock, false);
+        latch->release();
+      }
+    }
+    else
+    {
+      dict_table_t *table= lock->un_member.tab_lock.table;
+      ut_ad(!table->is_temporary());
+      ut_ad(table->id >= DICT_HDR_FIRST_ID ||
+            (lock->mode() != LOCK_IX && lock->mode() != LOCK_X) ||
+            trx->dict_operation);
+      if (!table->lock_mutex_trylock())
+        all_released= false;
+      else
+      {
+        lock_table_dequeue(lock, false);
+        table->lock_mutex_unlock();
+      }
+    }
+
+    lock= all_released ? UT_LIST_GET_LAST(trx->lock.trx_locks) : prev;
+    if (!--count)
+      break;
+  }
+
+  lock_sys.rd_unlock();
+  trx->mutex_unlock();
+  if (all_released && !count)
+    goto restart;
+  return all_released;
+}
+
+/** Release the explicit locks of a committing transaction,
+and release possible other transactions waiting because of these locks. */
+void lock_release(trx_t *trx)
+{
+  ulint count;
+
+  for (count= 5; count--; )
+    if (lock_release_try(trx))
+      goto released;
+
+  /* Fall back to acquiring lock_sys.latch in exclusive mode */
+restart:
+  count= 1000;
+  lock_sys.wr_lock(SRW_LOCK_CALL);
+  trx->mutex_lock();
+
+  while (lock_t *lock= UT_LIST_GET_LAST(trx->lock.trx_locks))
   {
     ut_ad(lock->trx == trx);
     if (!lock->is_table())
@@ -3981,20 +4050,16 @@ void lock_release(trx_t *trx)
       lock_table_dequeue(lock, false);
     }
 
-    if (count == 1000)
-    {
-      /* Release the latch for a while, so that we do not monopolize it */
-      lock_sys.wr_unlock();
-      trx->mutex_unlock();
-      count= 0;
-      lock_sys.wr_lock(SRW_LOCK_CALL);
-      trx->mutex_lock();
-    }
-
-    ++count;
+    if (!--count)
+      break;
   }
 
+  lock_sys.wr_unlock();
   trx->mutex_unlock();
+  if (!count)
+    goto restart;
+
+released:
   trx->lock.was_chosen_as_deadlock_victim= false;
   trx->lock.n_rec_locks= 0;
 }