MDEV-24258 Merge dict_sys.mutex into dict_sys.latch

In the parent commit, dict_sys.latch could theoretically have been
replaced with a mutex. But, we can do better and merge dict_sys.mutex
into dict_sys.latch. Generally, every occurrence of dict_sys.mutex_lock()
will be replaced with dict_sys.lock().

The PERFORMANCE_SCHEMA instrumentation for dict_sys_mutex
will be removed along with dict_sys.mutex. The dict_sys.latch
will remain instrumented as dict_operation_lock.

Some use of dict_sys.lock() will be replaced with dict_sys.freeze(),
which we will reintroduce for the new shared mode. Most notably,
concurrent table lookups are possible as long as the tables are present
in the dict_sys cache. In particular, this will allow more concurrency
among InnoDB purge workers.

Because dict_sys.mutex will no longer 'throttle' the threads that purge
InnoDB transaction history, a performance degradation may be observed
unless innodb_purge_threads=1.

The table cache eviction policy will become FIFO-like,
similar to what happened to fil_system.LRU
in commit 45ed9dd957eebc7fc84feb2509f4aa6baa908a95.
The name of the list dict_sys.table_LRU will become somewhat misleading;
that list contains tables that may be evicted, even though the
eviction policy no longer is least-recently-used but first-in-first-out.
(Note: Tables can never be evicted as long as locks exist on them or
the tables are in use by some thread.)

As demonstrated by the test perfschema.sxlock_func, there
will be less contention on dict_sys.latch, because some previous
use of exclusive latches will be replaced with shared latches.

fts_parse_sql_no_dict_lock(): Replaced with pars_sql().

fts_get_table_name_prefix(): Merged to fts_optimize_create().

dict_stats_update_transient_for_index(): Deduplicated some code.

ha_innobase::info_low(), dict_stats_stop_bg(): Use a combination
of dict_sys.latch and table->stats_mutex_lock() to cover the
changes of BG_STAT_SHOULD_QUIT, because the flag is being read
in dict_stats_update_persistent() while not holding dict_sys.latch.

row_discard_tablespace_for_mysql(): Protect stats_bg_flag by
exclusive dict_sys.latch, like most other code does.

row_quiesce_table_has_fts_index(): Remove unnecessary mutex
acquisition. FLUSH TABLES...FOR EXPORT is protected by MDL.

row_import::set_root_by_heuristic(): Remove unnecessary mutex
acquisition. ALTER TABLE...IMPORT TABLESPACE is protected by MDL.

row_ins_sec_index_entry_low(): Replace a call
to dict_set_corrupted_index_cache_only(). Reads of index->type
were not really protected by dict_sys.mutex, and writes
(flagging an index corrupted) should be extremely rare.

dict_stats_process_entry_from_defrag_pool(): Only freeze the dictionary,
do not lock it exclusively.

dict_stats_wait_bg_to_stop_using_table(), DICT_BG_YIELD: Remove trx.
We can simply invoke dict_sys.unlock() and dict_sys.lock() directly.

dict_acquire_mdl_shared()<trylock=false>: Assert that dict_sys.latch is
only held in shared more, not exclusive mode. Only acquire it in
exclusive mode if the table needs to be loaded to the cache.

dict_sys_t::acquire(): Remove. Relocating elements in dict_sys.table_LRU
would require holding an exclusive latch, which we want to avoid
for performance reasons.

dict_sys_t::allow_eviction(): Add the table first to dict_sys.table_LRU,
to compensate for the removal of dict_sys_t::acquire(). This function
is only invoked by INFORMATION_SCHEMA.INNODB_SYS_TABLESTATS.

dict_table_open_on_id(), dict_table_open_on_name(): If dict_locked=false,
try to acquire dict_sys.latch in shared mode. Only acquire the latch in
exclusive mode if the table is not found in the cache.

Reviewed by: Thirunarayanan Balathandayuthapani

46 files changed, 653 insertions, 728 deletions

diff --git a/mysql-test/suite/perfschema/r/sxlock_func.result b/mysql-test/suite/perfschema/r/sxlock_func.result
index 943f2c9301b..8f4c2fdcece 100644
--- a/mysql-test/suite/perfschema/r/sxlock_func.result
+++ b/mysql-test/suite/perfschema/r/sxlock_func.result
@@ -35,10 +35,10 @@ SELECT DISTINCT event_name FROM performance_schema.events_waits_history_long
 WHERE event_name LIKE 'wait/synch/rwlock/innodb/%'
 AND event_name NOT IN
 ('wait/synch/rwlock/innodb/btr_search_latch',
+'wait/synch/rwlock/innodb/dict_operation_lock',
 'wait/synch/rwlock/innodb/trx_purge_latch')
 ORDER BY event_name;
 event_name
-wait/synch/rwlock/innodb/dict_operation_lock
 wait/synch/rwlock/innodb/fil_space_latch
 wait/synch/rwlock/innodb/lock_latch
 SELECT event_name FROM performance_schema.events_waits_history_long
diff --git a/mysql-test/suite/perfschema/t/sxlock_func.test b/mysql-test/suite/perfschema/t/sxlock_func.test
index b8852c51ccb..c628158c9ea 100644
--- a/mysql-test/suite/perfschema/t/sxlock_func.test
+++ b/mysql-test/suite/perfschema/t/sxlock_func.test
@@ -42,11 +42,13 @@ commit;
 drop table t1;
 
 # Make sure some rw-lock operations have been executed
+# (there will only occasionally be waits on some rw-locks)
 
 SELECT DISTINCT event_name FROM performance_schema.events_waits_history_long
 WHERE event_name LIKE 'wait/synch/rwlock/innodb/%'
 AND event_name NOT IN
 ('wait/synch/rwlock/innodb/btr_search_latch',
+ 'wait/synch/rwlock/innodb/dict_operation_lock',
  'wait/synch/rwlock/innodb/trx_purge_latch')
 ORDER BY event_name;
 
diff --git a/storage/innobase/btr/btr0sea.cc b/storage/innobase/btr/btr0sea.cc
index a5e22fa0c2a..6fb4a01951a 100644
--- a/storage/innobase/btr/btr0sea.cc
+++ b/storage/innobase/btr/btr0sea.cc
@@ -230,12 +230,12 @@ void btr_search_disable()
 {
 	dict_table_t*	table;
 
-	dict_sys.mutex_lock();
+	dict_sys.freeze(SRW_LOCK_CALL);
 
 	btr_search_x_lock_all();
 
 	if (!btr_search_enabled) {
-		dict_sys.mutex_unlock();
+		dict_sys.unfreeze();
 		btr_search_x_unlock_all();
 		return;
 	}
@@ -256,7 +256,7 @@ void btr_search_disable()
 		btr_search_disable_ref_count(table);
 	}
 
-	dict_sys.mutex_unlock();
+	dict_sys.unfreeze();
 
 	/* Set all block->index = NULL. */
 	buf_pool.clear_hash_index();
@@ -1263,7 +1263,7 @@ retry:
 	ut_ad(page_is_leaf(block->frame));
 
 	/* We must not dereference block->index here, because it could be freed
-	if (index->table->n_ref_count == 0).
+	if (!index->table->get_ref_count() && !dict_sys.frozen()).
 	Determine the ahi_slot based on the block contents. */
 
 	const index_id_t	index_id
@@ -1436,10 +1436,8 @@ void btr_search_drop_page_hash_when_freed(const page_id_t page_id)
 			/* In all our callers, the table handle should
 			be open, or we should be in the process of
 			dropping the table (preventing eviction). */
-#ifdef SAFE_MUTEX
 			DBUG_ASSERT(block->index->table->get_ref_count()
-				    || dict_sys.mutex_is_locked());
-#endif /* SAFE_MUTEX */
+				    || dict_sys.locked());
 			btr_search_drop_page_hash_index(block);
 		}
 	}
diff --git a/storage/innobase/dict/dict0boot.cc b/storage/innobase/dict/dict0boot.cc
index 2792f71376d..e6a46699f02 100644
--- a/storage/innobase/dict/dict0boot.cc
+++ b/storage/innobase/dict/dict0boot.cc
@@ -225,7 +225,7 @@ dict_boot(void)
 
 	heap = mem_heap_create(450);
 
-	dict_sys.mutex_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	/* Get the dictionary header */
 	const byte* dict_hdr = &dict_hdr_get(&mtr)->frame[DICT_HDR];
@@ -413,10 +413,10 @@ dict_boot(void)
 		dict_load_sys_table(dict_sys.sys_columns);
 		dict_load_sys_table(dict_sys.sys_indexes);
 		dict_load_sys_table(dict_sys.sys_fields);
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 		dict_sys.load_sys_tables();
 	} else {
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 
 	return(err);
diff --git a/storage/innobase/dict/dict0crea.cc b/storage/innobase/dict/dict0crea.cc
index 70d9e2bb022..c293ab302a3 100644
--- a/storage/innobase/dict/dict0crea.cc
+++ b/storage/innobase/dict/dict0crea.cc
@@ -343,7 +343,7 @@ dict_build_table_def_step(
 	que_thr_t*	thr,	/*!< in: query thread */
 	tab_node_t*	node)	/*!< in: table create node */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	dict_table_t*	table = node->table;
 	ut_ad(!table->is_temporary());
 	ut_ad(!table->space);
@@ -403,7 +403,7 @@ dict_build_v_col_def_step(
 Based on an index object, this function builds the entry to be inserted
 in the SYS_INDEXES system table.
 @return the tuple which should be inserted */
-static
+static MY_ATTRIBUTE((nonnull, warn_unused_result))
 dtuple_t*
 dict_create_sys_indexes_tuple(
 /*==========================*/
@@ -416,7 +416,7 @@ dict_create_sys_indexes_tuple(
 	dfield_t*	dfield;
 	byte*		ptr;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_ad(index);
 	ut_ad(index->table->space || !UT_LIST_GET_LEN(index->table->indexes)
 	      || index->table->file_unreadable);
@@ -646,7 +646,7 @@ dict_build_index_def_step(
 	dtuple_t*	row;
 	trx_t*		trx;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	trx = thr_get_trx(thr);
 
@@ -691,7 +691,7 @@ dict_build_index_def(
 	dict_index_t*		index,	/*!< in/out: index */
 	trx_t*			trx)	/*!< in/out: InnoDB transaction handle */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	ut_ad((UT_LIST_GET_LEN(table->indexes) > 0)
 	      || dict_index_is_clust(index));
@@ -734,7 +734,7 @@ dict_create_index_tree_step(
 	dict_index_t*	index;
 	dtuple_t*	search_tuple;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	index = node->index;
 
@@ -803,7 +803,7 @@ dict_create_index_tree_in_mem(
 {
 	mtr_t		mtr;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_ad(!(index->type & DICT_FTS));
 
 	mtr_start(&mtr);
@@ -833,7 +833,7 @@ uint32_t dict_drop_index_tree(btr_pcur_t *pcur, trx_t *trx, mtr_t *mtr)
 {
   rec_t *rec= btr_pcur_get_rec(pcur);
 
-  ut_d(if (trx) dict_sys.assert_locked());
+  ut_ad(!trx || dict_sys.locked());
   ut_ad(!dict_table_is_comp(dict_sys.sys_indexes));
   btr_pcur_store_position(pcur, mtr);
 
@@ -995,7 +995,7 @@ dict_create_table_step(
 	trx_t*		trx;
 
 	ut_ad(thr);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	trx = thr_get_trx(thr);
 
@@ -1171,7 +1171,7 @@ dict_create_index_step(
 	trx_t*		trx;
 
 	ut_ad(thr);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	trx = thr_get_trx(thr);
 
@@ -1319,7 +1319,7 @@ bool dict_sys_t::load_sys_tables()
 {
   ut_ad(!srv_any_background_activity());
   bool mismatch= false;
-  mutex_lock();
+  lock(SRW_LOCK_CALL);
   if (!(sys_foreign= load_table(SYS_TABLE[SYS_FOREIGN],
                                 DICT_ERR_IGNORE_FK_NOKEY)));
   else if (UT_LIST_GET_LEN(sys_foreign->indexes) == 3 &&
@@ -1354,7 +1354,7 @@ bool dict_sys_t::load_sys_tables()
     mismatch= true;
     ib::error() << "Invalid definition of SYS_VIRTUAL";
   }
-  mutex_unlock();
+  unlock();
   return mismatch;
 }
 
@@ -1453,13 +1453,13 @@ err_exit:
   trx->free();
   srv_file_per_table= srv_file_per_table_backup;
 
-  mutex_lock();
+  lock(SRW_LOCK_CALL);
   if (sys_foreign);
   else if (!(sys_foreign= load_table(SYS_TABLE[SYS_FOREIGN])))
   {
     tablename= SYS_TABLE[SYS_FOREIGN].data();
 load_fail:
-    mutex_unlock();
+    unlock();
     ib::error() << "Failed to CREATE TABLE " << tablename;
     return DB_TABLE_NOT_FOUND;
   }
@@ -1484,7 +1484,7 @@ load_fail:
   else
     prevent_eviction(sys_virtual);
 
-  mutex_unlock();
+  unlock();
   return DB_SUCCESS;
 }
 
@@ -1877,7 +1877,7 @@ dict_create_add_foreigns_to_dictionary(
 	const dict_table_t*	table,
 	trx_t*			trx)
 {
-  dict_sys.assert_locked();
+  ut_ad(dict_sys.locked());
 
   if (!dict_sys.sys_foreign)
   {
diff --git a/storage/innobase/dict/dict0defrag_bg.cc b/storage/innobase/dict/dict0defrag_bg.cc
index 4be6a28882e..44cc4289a67 100644
--- a/storage/innobase/dict/dict0defrag_bg.cc
+++ b/storage/innobase/dict/dict0defrag_bg.cc
@@ -145,7 +145,7 @@ dict_stats_defrag_pool_del(
 {
 	ut_a((table && !index) || (!table && index));
 	ut_ad(!srv_read_only_mode);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 
 	mysql_mutex_lock(&defrag_pool_mutex);
 
@@ -170,45 +170,34 @@ dict_stats_defrag_pool_del(
 /*****************************************************************//**
 Get the first index that has been added for updating persistent defrag
 stats and eventually save its stats. */
-static
-void
-dict_stats_process_entry_from_defrag_pool()
+static void dict_stats_process_entry_from_defrag_pool()
 {
-	table_id_t	table_id;
-	index_id_t	index_id;
-
-	ut_ad(!srv_read_only_mode);
-
-	/* pop the first index from the auto defrag pool */
-	if (!dict_stats_defrag_pool_get(&table_id, &index_id)) {
-		/* no index in defrag pool */
-		return;
-	}
-
-	dict_table_t*	table;
-
-	dict_sys.mutex_lock();
-
-	/* If the table is no longer cached, we've already lost the in
-	memory stats so there's nothing really to write to disk. */
-	table = dict_table_open_on_id(table_id, TRUE,
-				      DICT_TABLE_OP_OPEN_ONLY_IF_CACHED);
-
-	dict_index_t* index = table && !table->corrupted
-		? dict_table_find_index_on_id(table, index_id)
-		: NULL;
-
-	if (!index || index->is_corrupted()) {
-		if (table) {
-			dict_table_close(table, TRUE, FALSE);
-		}
-		dict_sys.mutex_unlock();
-		return;
-	}
-
-	dict_sys.mutex_unlock();
-	dict_stats_save_defrag_stats(index);
-	dict_table_close(table, FALSE, FALSE);
+  table_id_t	table_id;
+  index_id_t	index_id;
+
+  ut_ad(!srv_read_only_mode);
+
+  /* pop the first index from the auto defrag pool */
+  if (!dict_stats_defrag_pool_get(&table_id, &index_id))
+    /* no index in defrag pool */
+    return;
+
+  dict_sys.freeze(SRW_LOCK_CALL);
+
+  /* If the table is no longer cached, we've already lost the in
+  memory stats so there's nothing really to write to disk. */
+  dict_table_t *table= dict_sys.find_table(table_id);
+  dict_index_t *index= table && table->corrupted
+    ? nullptr : dict_table_find_index_on_id(table, index_id);
+  const bool save= index && !index->is_corrupted();
+  if (save)
+    table->acquire();
+  dict_sys.unfreeze();
+  if (save)
+  {
+    dict_stats_save_defrag_stats(index);
+    table->release();
+  }
 }
 
 /*****************************************************************//**
@@ -237,7 +226,7 @@ dict_stats_save_defrag_summary(
 		return DB_SUCCESS;
 	}
 
-	dict_sys_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	ret = dict_stats_save_index_stat(index, time(NULL), "n_pages_freed",
 					 index->stat_defrag_n_pages_freed,
@@ -246,7 +235,7 @@ dict_stats_save_defrag_summary(
 					 " last defragmentation run.",
 					 NULL);
 
-	dict_sys_unlock();
+	dict_sys.unlock();
 
 	return (ret);
 }
@@ -331,7 +320,7 @@ dict_stats_save_defrag_stats(
 		return DB_SUCCESS;
 	}
 
-	dict_sys_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 	ret = dict_stats_save_index_stat(index, now, "n_page_split",
 					 index->stat_defrag_n_page_split,
 					 NULL,
@@ -361,6 +350,6 @@ dict_stats_save_defrag_stats(
 		NULL);
 
 end:
-	dict_sys_unlock();
+	dict_sys.unlock();
 	return ret;
 }
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 2ab5f9a07c9..400cda111fe 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -86,7 +86,7 @@ const span<const char> dict_sys_t::SYS_TABLE[]=
 
 /** Diagnostic message for exceeding the mutex_lock_wait() timeout */
 const char dict_sys_t::fatal_msg[]=
-  "innodb_fatal_semaphore_wait_threshold was exceeded for dict_sys.mutex. "
+  "innodb_fatal_semaphore_wait_threshold was exceeded for dict_sys.latch. "
   "Please refer to "
   "https://mariadb.com/kb/en/how-to-produce-a-full-stack-trace-for-mysqld/";
 
@@ -272,11 +272,10 @@ dict_table_try_drop_aborted(
 /**********************************************************************//**
 When opening a table,
 try to drop any indexes after an aborted index creation.
-Release the dict_sys.mutex. */
+Invoke dict_sys.unlock(). */
 static
 void
-dict_table_try_drop_aborted_and_mutex_exit(
-/*=======================================*/
+dict_table_try_drop_aborted_and_unlock(
 	dict_table_t*	table,		/*!< in: table (may be NULL) */
 	ibool		try_drop)	/*!< in: FALSE if should try to
 					drop indexes whose online creation
@@ -292,11 +291,11 @@ dict_table_try_drop_aborted_and_mutex_exit(
 		was aborted. */
 		table_id_t	table_id = table->id;
 
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 
 		dict_table_try_drop_aborted(table, table_id, 1);
 	} else {
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 }
 
@@ -317,10 +316,10 @@ dict_table_close(
 	MDL_ticket*	mdl)
 {
 	if (!dict_locked) {
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_a(table->get_ref_count() > 0);
 
 	const bool last_handle = table->release();
@@ -330,8 +329,9 @@ dict_table_close(
 	if they have been manually modified. We reset table->stat_initialized
 	only if table reference count is 0 because we do not want too frequent
 	stats re-reads (e.g. in other cases than FLUSH TABLE). */
-	if (last_handle && strchr(table->name.m_name, '/') != NULL
-	    && dict_stats_is_persistent_enabled(table)) {
+	if (last_handle
+	    && dict_stats_is_persistent_enabled(table)
+	    && strchr(table->name.m_name, '/')) {
 
 		table->stats_mutex_lock();
 		dict_stats_deinit(table);
@@ -347,7 +347,7 @@ dict_table_close(
 			&& table->drop_aborted
 			&& dict_table_get_first_index(table);
 
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 
 		/* dict_table_try_drop_aborted() can generate undo logs.
 		So it should be avoided after shutdown of background
@@ -691,7 +691,7 @@ dict_index_get_nth_field_pos(
 }
 
 /** Parse the table file name into table name and database name.
-@tparam		dict_locked	whether dict_sys.mutex is being held
+@tparam		dict_locked	whether dict_sys.lock() was called
 @param[in,out]	db_name		database name buffer
 @param[in,out]	tbl_name	table name buffer
 @param[out]	db_name_len	database name length
@@ -706,8 +706,8 @@ bool dict_table_t::parse_name(char (&db_name)[NAME_LEN + 1],
   char tbl_buf[MAX_TABLE_NAME_LEN + 1];
 
   if (!dict_locked)
-    dict_sys.mutex_lock(); /* protect against renaming */
-  dict_sys.assert_locked();
+    dict_sys.freeze(SRW_LOCK_CALL); /* protect against renaming */
+  ut_ad(dict_sys.frozen());
   const size_t db_len= name.dblen();
   ut_ad(db_len <= MAX_DATABASE_NAME_LEN);
 
@@ -726,7 +726,7 @@ bool dict_table_t::parse_name(char (&db_name)[NAME_LEN + 1],
   tbl_buf[tbl_len]= 0;
 
   if (!dict_locked)
-    dict_sys.mutex_unlock();
+    dict_sys.unfreeze();
 
   *db_name_len= filename_to_tablename(db_buf, db_name,
                                       MAX_DATABASE_NAME_LEN + 1, true);
@@ -766,13 +766,14 @@ dict_acquire_mdl_shared(dict_table_t *table,
 
   if (trylock)
   {
-    dict_sys.mutex_lock();
+    dict_sys.freeze(SRW_LOCK_CALL);
     db_len= dict_get_db_name_len(table->name.m_name);
-    dict_sys.mutex_unlock();
+    dict_sys.unfreeze();
   }
   else
   {
-    dict_sys.assert_locked();
+    ut_ad(dict_sys.frozen());
+    ut_ad(!dict_sys.locked());
     db_len= dict_get_db_name_len(table->name.m_name);
   }
 
@@ -795,7 +796,6 @@ dict_acquire_mdl_shared(dict_table_t *table,
 retry:
   if (!unaccessible && (!table->is_readable() || table->corrupted))
   {
-is_unaccessible:
     if (*mdl)
     {
       mdl_context->release_lock(*mdl);
@@ -811,10 +811,12 @@ is_unaccessible:
     return nullptr;
 
   if (!trylock)
-    dict_sys.mutex_unlock();
+    dict_sys.unfreeze();
+
   {
     MDL_request request;
-    MDL_REQUEST_INIT(&request,MDL_key::TABLE, db_buf, tbl_buf, MDL_SHARED, MDL_EXPLICIT);
+    MDL_REQUEST_INIT(&request,MDL_key::TABLE, db_buf, tbl_buf, MDL_SHARED,
+                     MDL_EXPLICIT);
     if (trylock
         ? mdl_context->try_acquire_lock(&request)
         : mdl_context->acquire_lock(&request,
@@ -828,19 +830,37 @@ is_unaccessible:
         return nullptr;
     }
     else
+    {
       *mdl= request.ticket;
+      if (trylock && !*mdl)
+        return nullptr;
+    }
   }
 
-  if (!trylock)
-    dict_sys.mutex_lock();
-  else if (!*mdl)
-    return nullptr;
-
-  table= dict_table_open_on_id(table_id, !trylock, table_op);
+  dict_sys.freeze(SRW_LOCK_CALL);
+  table= dict_sys.find_table(table_id);
+  if (table)
+    table->acquire();
+  if (!table && table_op != DICT_TABLE_OP_OPEN_ONLY_IF_CACHED)
+  {
+    dict_sys.unfreeze();
+    dict_sys.lock(SRW_LOCK_CALL);
+    table= dict_load_table_on_id(table_id,
+                                 table_op == DICT_TABLE_OP_LOAD_TABLESPACE
+                                 ? DICT_ERR_IGNORE_RECOVER_LOCK
+                                 : DICT_ERR_IGNORE_FK_NOKEY);
+    if (table)
+      table->acquire();
+    dict_sys.unlock();
+    dict_sys.freeze(SRW_LOCK_CALL);
+  }
 
-  if (!table)
+  if (!table || !table->is_accessible())
   {
-    /* The table was dropped. */
+    table= nullptr;
+return_without_mdl:
+    if (trylock)
+      dict_sys.unfreeze();
     if (*mdl)
     {
       mdl_context->release_lock(*mdl);
@@ -849,21 +869,13 @@ is_unaccessible:
     return nullptr;
   }
 
-  if (!table->is_accessible())
-    goto is_unaccessible;
-
   size_t db1_len, tbl1_len;
 
   if (!table->parse_name<!trylock>(db_buf1, tbl_buf1, &db1_len, &tbl1_len))
   {
     /* The table was renamed to #sql prefix.
     Release MDL (if any) for the old name and return. */
-    if (*mdl)
-    {
-      mdl_context->release_lock(*mdl);
-      *mdl= nullptr;
-    }
-    return table;
+    goto return_without_mdl;
   }
 
   if (*mdl)
@@ -871,7 +883,11 @@ is_unaccessible:
     if (db_len == db1_len && tbl_len == tbl1_len &&
         !memcmp(db_buf, db_buf1, db_len) &&
         !memcmp(tbl_buf, tbl_buf1, tbl_len))
+    {
+      if (trylock)
+        dict_sys.unfreeze();
       return table;
+    }
 
     /* The table was renamed. Release MDL for the old name and
     try to acquire MDL for the new name. */
@@ -887,8 +903,10 @@ is_unaccessible:
   goto retry;
 }
 
-template dict_table_t*
-dict_acquire_mdl_shared<true>(dict_table_t*,THD*,MDL_ticket**,dict_table_op_t);
+template dict_table_t* dict_acquire_mdl_shared<false>
+(dict_table_t*,THD*,MDL_ticket**,dict_table_op_t);
+template dict_table_t* dict_acquire_mdl_shared<true>
+(dict_table_t*,THD*,MDL_ticket**,dict_table_op_t);
 
 /** Look up a table by numeric identifier.
 @param[in]      table_id        table identifier
@@ -902,34 +920,47 @@ dict_table_open_on_id(table_id_t table_id, bool dict_locked,
                       dict_table_op_t table_op, THD *thd,
                       MDL_ticket **mdl)
 {
-	if (!dict_locked) {
-		dict_sys.mutex_lock();
-	}
-
-	dict_sys.assert_locked();
-
-	dict_table_t* table = dict_table_open_on_id_low(
-		table_id,
-		table_op == DICT_TABLE_OP_LOAD_TABLESPACE
-		? DICT_ERR_IGNORE_RECOVER_LOCK
-		: DICT_ERR_IGNORE_FK_NOKEY,
-		table_op == DICT_TABLE_OP_OPEN_ONLY_IF_CACHED);
+  if (!dict_locked)
+    dict_sys.freeze(SRW_LOCK_CALL);
 
-	if (table) {
-		dict_sys.acquire(table);
-	}
+  dict_table_t *table= dict_sys.find_table(table_id);
 
-	if (!dict_locked) {
-		if (thd) {
-			table = dict_acquire_mdl_shared<false>(
-				table, thd, mdl, table_op);
-		}
+  if (table)
+  {
+    table->acquire();
+    if (thd && !dict_locked)
+      table= dict_acquire_mdl_shared<false>(table, thd, mdl, table_op);
+  }
+  else if (table_op != DICT_TABLE_OP_OPEN_ONLY_IF_CACHED)
+  {
+    if (!dict_locked)
+    {
+      dict_sys.unfreeze();
+      dict_sys.lock(SRW_LOCK_CALL);
+    }
+    table= dict_load_table_on_id(table_id,
+                                 table_op == DICT_TABLE_OP_LOAD_TABLESPACE
+                                 ? DICT_ERR_IGNORE_RECOVER_LOCK
+                                 : DICT_ERR_IGNORE_FK_NOKEY);
+    if (table)
+      table->acquire();
+    if (!dict_locked)
+    {
+      dict_sys.unlock();
+      if (table && thd)
+      {
+        dict_sys.freeze(SRW_LOCK_CALL);
+        table= dict_acquire_mdl_shared<false>(table, thd, mdl, table_op);
+        dict_sys.unfreeze();
+      }
+      return table;
+    }
+  }
 
-		dict_table_try_drop_aborted_and_mutex_exit(
-			table, table_op == DICT_TABLE_OP_DROP_ORPHAN);
-	}
+  if (!dict_locked)
+    dict_sys.unfreeze();
 
-	return table;
+  return table;
 }
 
 /********************************************************************//**
@@ -991,8 +1022,6 @@ void dict_sys_t::create()
   UT_LIST_INIT(table_LRU, &dict_table_t::table_LRU);
   UT_LIST_INIT(table_non_LRU, &dict_table_t::table_LRU);
 
-  mysql_mutex_init(dict_sys_mutex_key, &mutex, nullptr);
-
   const ulint hash_size = buf_pool_get_curr_size()
     / (DICT_POOL_PER_TABLE_HASH * UNIV_WORD_SIZE);
 
@@ -1013,27 +1042,17 @@ void dict_sys_t::create()
 }
 
 
-/** Acquire a reference to a cached table. */
-inline void dict_sys_t::acquire(dict_table_t *table)
-{
-  ut_ad(dict_sys.find(table));
-  if (table->can_be_evicted)
-  {
-    UT_LIST_REMOVE(table_LRU, table);
-    UT_LIST_ADD_FIRST(table_LRU, table);
-  }
-
-  table->acquire();
-}
-
-void dict_sys_t::mutex_lock_wait()
+void dict_sys_t::lock_wait(SRW_LOCK_ARGS(const char *file, unsigned line))
 {
   ulonglong now= my_hrtime_coarse().val, old= 0;
-  if (mutex_wait_start.compare_exchange_strong
+  if (latch_ex_wait_start.compare_exchange_strong
       (old, now, std::memory_order_relaxed, std::memory_order_relaxed))
   {
-    mysql_mutex_lock(&mutex);
-    mutex_wait_start.store(0, std::memory_order_relaxed);
+    latch.wr_lock(SRW_LOCK_ARGS(file, line));
+    latch_ex_wait_start.store(0, std::memory_order_relaxed);
+    ut_ad(!latch_readers);
+    ut_ad(!latch_ex);
+    ut_d(latch_ex= true);
     return;
   }
 
@@ -1047,32 +1066,36 @@ void dict_sys_t::mutex_lock_wait()
 
   if (waited > threshold / 4)
     ib::warn() << "A long wait (" << waited
-               << " seconds) was observed for dict_sys.mutex";
-  mysql_mutex_lock(&mutex);
+               << " seconds) was observed for dict_sys.latch";
+  latch.wr_lock(SRW_LOCK_ARGS(file, line));
+  ut_ad(!latch_readers);
+  ut_ad(!latch_ex);
+  ut_d(latch_ex= true);
 }
 
-#ifdef HAVE_PSI_MUTEX_INTERFACE
-/** Acquire the mutex */
-void dict_sys_t::mutex_lock()
+#ifdef UNIV_PFS_RWLOCK
+ATTRIBUTE_NOINLINE void dict_sys_t::unlock()
 {
-  if (mysql_mutex_trylock(&mutex))
-    mutex_lock_wait();
+  ut_ad(latch_ex);
+  ut_ad(!latch_readers);
+  ut_d(latch_ex= false);
+  latch.wr_unlock();
 }
 
-/** Release the mutex */
-void dict_sys_t::mutex_unlock() { mysql_mutex_unlock(&mutex); }
-#endif
+ATTRIBUTE_NOINLINE void dict_sys_t::freeze(const char *file, unsigned line)
+{
+  latch.rd_lock(file, line);
+  ut_ad(!latch_ex);
+  ut_d(latch_readers++);
+}
 
-/** Lock the data dictionary cache. */
-void dict_sys_t::lock(SRW_LOCK_ARGS(const char *file, unsigned line))
+ATTRIBUTE_NOINLINE void dict_sys_t::unfreeze()
 {
-  ut_ad(this == &dict_sys);
-  ut_ad(is_initialised());
-  latch.wr_lock(SRW_LOCK_ARGS(file, line));
   ut_ad(!latch_ex);
-  ut_d(latch_ex= true);
-  mutex_lock();
+  ut_ad(latch_readers--);
+  latch.rd_unlock();
 }
+#endif /* UNIV_PFS_RWLOCK */
 
 /**********************************************************************//**
 Returns a table object and increment its open handle count.
@@ -1092,58 +1115,58 @@ dict_table_open_on_name(
 			ignore_err)	/*!< in: error to be ignored when
 					loading a table definition */
 {
-	dict_table_t*	table;
-	DBUG_ENTER("dict_table_open_on_name");
-	DBUG_PRINT("dict_table_open_on_name", ("table: '%s'", table_name));
-
-	if (!dict_locked) {
-		dict_sys.mutex_lock();
-	}
-
-	ut_ad(table_name);
-	table = dict_sys.load_table({table_name, strlen(table_name)},
-				    ignore_err);
-
-	if (table) {
-		ut_ad(table->cached);
-
-		/* If table is encrypted or corrupted */
-		if (!(ignore_err & ~DICT_ERR_IGNORE_FK_NOKEY)
-		    && !table->is_readable()) {
-			/* Make life easy for drop table. */
-			dict_sys.prevent_eviction(table);
-
-			if (table->corrupted) {
-
-				ib::error() << "Table " << table->name
-					<< " is corrupted. Please "
-					"drop the table and recreate.";
-				if (!dict_locked) {
-					dict_sys.mutex_unlock();
-				}
-
-				DBUG_RETURN(NULL);
-			}
+  dict_table_t *table;
+  DBUG_ENTER("dict_table_open_on_name");
+  DBUG_PRINT("dict_table_open_on_name", ("table: '%s'", table_name));
 
-			dict_sys.acquire(table);
+  const span<const char> name{table_name, strlen(table_name)};
 
-			if (!dict_locked) {
-				dict_sys.mutex_unlock();
-			}
+  if (!dict_locked)
+  {
+    dict_sys.freeze(SRW_LOCK_CALL);
+    table= dict_sys.find_table(name);
+    if (table)
+    {
+      ut_ad(table->cached);
+      if (!(ignore_err & ~DICT_ERR_IGNORE_FK_NOKEY) &&
+          !table->is_readable() && table->corrupted)
+      {
+        ib::error() << "Table " << table->name
+                    << " is corrupted. Please drop the table and recreate.";
+        dict_sys.unfreeze();
+        DBUG_RETURN(nullptr);
+      }
+      table->acquire();
+      dict_sys.unfreeze();
+      DBUG_RETURN(table);
+    }
+    dict_sys.unfreeze();
+    dict_sys.lock(SRW_LOCK_CALL);
+  }
 
-			DBUG_RETURN(table);
-		}
+  table= dict_sys.load_table(name, ignore_err);
 
-		dict_sys.acquire(table);
-	}
+  if (table)
+  {
+    ut_ad(table->cached);
+    if (!(ignore_err & ~DICT_ERR_IGNORE_FK_NOKEY) &&
+        !table->is_readable() && table->corrupted)
+    {
+      ib::error() << "Table " << table->name
+                  << " is corrupted. Please drop the table and recreate.";
+      if (!dict_locked)
+        dict_sys.unlock();
+      DBUG_RETURN(nullptr);
+    }
 
-	ut_ad(dict_lru_validate());
+    table->acquire();
+  }
 
-	if (!dict_locked) {
-		dict_table_try_drop_aborted_and_mutex_exit(table, try_drop);
-	}
+  ut_ad(dict_lru_validate());
+  if (!dict_locked)
+    dict_table_try_drop_aborted_and_unlock(table, try_drop);
 
-	DBUG_RETURN(table);
+  DBUG_RETURN(table);
 }
 
 /**********************************************************************//**
@@ -1252,7 +1275,7 @@ inline void dict_sys_t::add(dict_table_t* table)
 @return whether the table can be evicted */
 static bool dict_table_can_be_evicted(dict_table_t *table)
 {
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 	ut_a(table->can_be_evicted);
 	ut_a(table->foreign_set.empty());
 	ut_a(table->referenced_set.empty());
@@ -1423,8 +1446,8 @@ dict_table_find_index_on_id(
 }
 
 /**********************************************************************//**
-Looks for an index with the given id. NOTE that we do not reserve
-the dictionary mutex: this function is for emergency purposes like
+Looks for an index with the given id. NOTE that we do not acquire
+dict_sys.latch: this function is for emergency purposes like
 printing info of a corrupt database page!
 @return index or NULL if not found in cache */
 dict_index_t*
@@ -1540,7 +1563,7 @@ dict_table_rename_in_cache(
 	ulint		fold;
 	char		old_name[MAX_FULL_NAME_LEN + 1];
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	/* store the old/current name to an automatic variable */
 	const size_t old_name_len = strlen(table->name.m_name);
@@ -1864,7 +1887,7 @@ dict_table_change_id_in_cache(
 	dict_table_t*	table,	/*!< in/out: table object already in cache */
 	table_id_t	new_id)	/*!< in: new id to set */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_ad(table->magic_n == DICT_TABLE_MAGIC_N);
 	ut_ad(!table->is_temporary());
 
@@ -1943,7 +1966,7 @@ void dict_sys_t::remove(dict_table_t* table, bool lru, bool keep)
 		and free the index pages. */
 		trx_t* trx = trx_create();
 
-		ut_d(dict_sys.assert_locked());
+		ut_ad(dict_sys.locked());
 		/* Mimic row_mysql_lock_data_dictionary(). */
 		trx->dict_operation_lock_mode = RW_X_LATCH;
 
@@ -2033,7 +2056,7 @@ dict_index_add_to_cache(
 	ulint		n_ord;
 	ulint		i;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_ad(index->n_def == index->n_fields);
 	ut_ad(index->magic_n == DICT_INDEX_MAGIC_N);
 	ut_ad(!dict_index_is_online_ddl(index));
@@ -2161,7 +2184,7 @@ dict_index_remove_from_cache_low(
 	ut_ad(table && index);
 	ut_ad(table->magic_n == DICT_TABLE_MAGIC_N);
 	ut_ad(index->magic_n == DICT_INDEX_MAGIC_N);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_ad(table->id);
 #ifdef BTR_CUR_HASH_ADAPT
 	ut_ad(!index->freed());
@@ -2243,7 +2266,7 @@ dict_index_find_cols(
 
 	const dict_table_t* table = index->table;
 	ut_ad(table->magic_n == DICT_TABLE_MAGIC_N);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	for (ulint i = 0; i < index->n_fields; i++) {
 		ulint		j;
@@ -2512,7 +2535,7 @@ dict_index_build_internal_clust(
 	ut_ad(index->is_primary());
 	ut_ad(!index->has_virtual());
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	/* Create a new index object with certainly enough fields */
 	new_index = dict_mem_index_create(index->table, index->name,
@@ -2667,7 +2690,7 @@ dict_index_build_internal_non_clust(
 	ut_ad(table && index);
 	ut_ad(!dict_index_is_clust(index));
 	ut_ad(!dict_index_is_ibuf(index));
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	/* The clustered index should be the first in the list of indexes */
 	clust_index = UT_LIST_GET_FIRST(table->indexes);
@@ -2761,7 +2784,7 @@ dict_index_build_internal_fts(
 	dict_index_t*	new_index;
 
 	ut_ad(index->type == DICT_FTS);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	/* Create a new index */
 	new_index = dict_mem_index_create(index->table, index->name,
@@ -2813,7 +2836,7 @@ dict_foreign_remove_from_cache(
 /*===========================*/
 	dict_foreign_t*	foreign)	/*!< in, own: foreign constraint */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_a(foreign);
 
 	if (foreign->referenced_table != NULL) {
@@ -2838,7 +2861,7 @@ dict_foreign_find(
 	dict_table_t*	table,		/*!< in: table object */
 	dict_foreign_t*	foreign)	/*!< in: foreign constraint */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 
 	ut_ad(dict_foreign_set_validate(table->foreign_set));
 	ut_ad(dict_foreign_set_validate(table->referenced_set));
@@ -2892,7 +2915,7 @@ dict_foreign_find_index(
 					/*!< out: index where error
 					happened */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 
 	if (error) {
 		*error = FK_INDEX_NOT_FOUND;
@@ -2990,7 +3013,7 @@ dict_foreign_add_to_cache(
 	DBUG_ENTER("dict_foreign_add_to_cache");
 	DBUG_PRINT("dict_foreign_add_to_cache", ("id: %s", foreign->id));
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	for_table = dict_sys.find_table(
 		{foreign->foreign_table_name_lookup,
@@ -3637,7 +3660,7 @@ dict_foreign_parse_drop_constraints(
 
 	ptr = str;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 loop:
 	ptr = dict_scan_to(ptr, "DROP");
 
@@ -3739,14 +3762,14 @@ syntax_error:
 
 /**********************************************************************//**
 Returns an index object if it is found in the dictionary cache.
-Assumes that dict_sys.mutex is already being held.
+Assumes that dict_sys.latch is already being held.
 @return index, NULL if not found */
 dict_index_t*
 dict_index_get_if_in_cache_low(
 /*===========================*/
 	index_id_t	index_id)	/*!< in: index id */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 
 	return(dict_index_find_on_id_low(index_id));
 }
@@ -3766,11 +3789,11 @@ dict_index_get_if_in_cache(
 		return(NULL);
 	}
 
-	dict_sys.mutex_lock();
+	dict_sys.freeze(SRW_LOCK_CALL);
 
 	index = dict_index_get_if_in_cache_low(index_id);
 
-	dict_sys.mutex_unlock();
+	dict_sys.unfreeze();
 
 	return(index);
 }
@@ -4056,7 +4079,7 @@ dict_print_info_on_foreign_keys(
 	dict_foreign_t*	foreign;
 	std::string 	str;
 
-	dict_sys.mutex_lock();
+	dict_sys.freeze(SRW_LOCK_CALL);
 
 	for (dict_foreign_set::iterator it = table->foreign_set.begin();
 	     it != table->foreign_set.end();
@@ -4123,7 +4146,7 @@ dict_print_info_on_foreign_keys(
 		}
 	}
 
-	dict_sys.mutex_unlock();
+	dict_sys.unfreeze();
 	return str;
 }
 
@@ -4221,7 +4244,7 @@ dict_set_corrupted(
 		row_mysql_lock_data_dictionary(trx);
 	}
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_ad(!dict_table_is_comp(dict_sys.sys_tables));
 	ut_ad(!dict_table_is_comp(dict_sys.sys_indexes));
 
@@ -4309,7 +4332,7 @@ dict_set_corrupted_index_cache_only(
 {
 	ut_ad(index != NULL);
 	ut_ad(index->table != NULL);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_ad(!dict_table_is_comp(dict_sys.sys_tables));
 	ut_ad(!dict_table_is_comp(dict_sys.sys_indexes));
 
@@ -4343,13 +4366,13 @@ dict_index_set_merge_threshold(
 	ut_ad(!dict_table_is_comp(dict_sys.sys_tables));
 	ut_ad(!dict_table_is_comp(dict_sys.sys_indexes));
 
-	dict_sys_lock();
-
 	heap = mem_heap_create(sizeof(dtuple_t) + 2 * (sizeof(dfield_t)
 			       + sizeof(que_fork_t) + sizeof(upd_node_t)
 			       + sizeof(upd_t) + 12));
 
-	mtr_start(&mtr);
+	mtr.start();
+
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	sys_index = UT_LIST_GET_FIRST(dict_sys.sys_indexes->indexes);
 
@@ -4387,7 +4410,7 @@ dict_index_set_merge_threshold(
 	mtr_commit(&mtr);
 	mem_heap_free(heap);
 
-	dict_sys_unlock();
+	dict_sys.unlock();
 }
 
 #ifdef UNIV_DEBUG
@@ -4419,14 +4442,14 @@ void
 dict_set_merge_threshold_all_debug(
 	uint	merge_threshold_all)
 {
-	dict_sys.mutex_lock();
+	dict_sys.freeze(SRW_LOCK_CALL);
 
 	dict_set_merge_threshold_list_debug(
 		&dict_sys.table_LRU, merge_threshold_all);
 	dict_set_merge_threshold_list_debug(
 		&dict_sys.table_non_LRU, merge_threshold_all);
 
-	dict_sys.mutex_unlock();
+	dict_sys.unfreeze();
 }
 
 #endif /* UNIV_DEBUG */
@@ -4544,7 +4567,7 @@ dict_table_check_for_dup_indexes(
 	const dict_index_t*	index1;
 	const dict_index_t*	index2;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 
 	/* The primary index _must_ exist */
 	ut_a(UT_LIST_GET_LEN(table->indexes) > 0);
@@ -4656,7 +4679,7 @@ void dict_sys_t::resize()
 {
   ut_ad(this == &dict_sys);
   ut_ad(is_initialised());
-  mutex_lock();
+  lock(SRW_LOCK_CALL);
 
   /* all table entries are in table_LRU and table_non_LRU lists */
   table_hash.free();
@@ -4694,7 +4717,7 @@ void dict_sys_t::resize()
     HASH_INSERT(dict_table_t, id_hash, id_hash, id_fold, table);
   }
 
-  mutex_unlock();
+  unlock();
 }
 
 /** Close the data dictionary cache on shutdown. */
@@ -4703,10 +4726,10 @@ void dict_sys_t::close()
   ut_ad(this == &dict_sys);
   if (!is_initialised()) return;
 
-  mutex_lock();
+  lock(SRW_LOCK_CALL);
 
-  /* Free the hash elements. We don't remove them from the table
-  because we are going to destroy the table anyway. */
+  /* Free the hash elements. We don't remove them from table_hash
+  because we are invoking table_hash.free() below. */
   for (ulint i= table_hash.n_cells; i--; )
     while (dict_table_t *table= static_cast<dict_table_t*>
            (HASH_GET_FIRST(&table_hash, i)))
@@ -4721,8 +4744,7 @@ void dict_sys_t::close()
   /* No temporary tables should exist at this point. */
   temp_id_hash.free();
 
-  mutex_unlock();
-  mysql_mutex_destroy(&mutex);
+  unlock();
   latch.destroy();
 
   mysql_mutex_destroy(&dict_foreign_err_mutex);
@@ -4747,7 +4769,7 @@ dict_lru_validate(void)
 {
 	dict_table_t*	table;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 
 	for (table = UT_LIST_GET_FIRST(dict_sys.table_LRU);
 	     table != NULL;
diff --git a/storage/innobase/dict/dict0load.cc b/storage/innobase/dict/dict0load.cc
index db11f4ff003..74275b77bb0 100644
--- a/storage/innobase/dict/dict0load.cc
+++ b/storage/innobase/dict/dict0load.cc
@@ -509,7 +509,7 @@ dict_sys_tables_rec_check(
 	const byte*	field;
 	ulint		len;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	if (rec_get_deleted_flag(rec, 0)) {
 		return("delete-marked record in SYS_TABLES");
@@ -828,7 +828,7 @@ static ulint dict_check_sys_tables()
 
 	DBUG_ENTER("dict_check_sys_tables");
 
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	mtr_start(&mtr);
 
@@ -934,7 +934,7 @@ void dict_check_tablespaces_and_store_max_id()
 
 	DBUG_ENTER("dict_check_tablespaces_and_store_max_id");
 
-	dict_sys_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	/* Initialize the max space_id from sys header */
 	mtr.start();
@@ -949,7 +949,7 @@ void dict_check_tablespaces_and_store_max_id()
 	max_space_id = dict_check_sys_tables();
 	fil_set_max_space_id_if_bigger(max_space_id);
 
-	dict_sys_unlock();
+	dict_sys.unlock();
 
 	DBUG_VOID_RETURN;
 }
@@ -1237,7 +1237,7 @@ dict_load_columns(
 	mtr_t		mtr;
 	ulint		n_skipped = 0;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	mtr_start(&mtr);
 
@@ -1351,7 +1351,7 @@ dict_load_virtual_one_col(
 	mtr_t		mtr;
 	ulint		skipped = 0;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	if (v_col->num_base == 0) {
 		return;
@@ -1581,7 +1581,7 @@ dict_load_fields(
 	mtr_t		mtr;
 	dberr_t		error;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	mtr_start(&mtr);
 
@@ -1811,7 +1811,7 @@ dict_load_indexes(
 	mtr_t		mtr;
 	dberr_t		error = DB_SUCCESS;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	mtr_start(&mtr);
 
@@ -2122,7 +2122,7 @@ dict_save_data_dir_path(
 	dict_table_t*	table,		/*!< in/out: table */
 	const char*	filepath)	/*!< in: filepath of tablespace */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 	ut_a(DICT_TF_HAS_DATA_DIR(table->flags));
 
 	ut_a(!table->data_dir_path);
@@ -2145,20 +2145,17 @@ dict_save_data_dir_path(
 	}
 }
 
-/** Make sure the data_dir_path is saved in dict_table_t if needed.
-@param[in]	table		Table object
-@param[in]	dict_mutex_own	true if dict_sys.mutex is owned already */
-void
-dict_get_and_save_data_dir_path(
-	dict_table_t*	table,
-	bool		dict_mutex_own)
+/** Make sure the data_file_name is saved in dict_table_t if needed.
+@param[in,out]	table		Table object
+@param[in]	dict_locked	dict_sys.frozen() */
+void dict_get_and_save_data_dir_path(dict_table_t* table, bool dict_locked)
 {
 	ut_ad(!table->is_temporary());
 	ut_ad(!table->space || table->space->id == table->space_id);
 
 	if (!table->data_dir_path && table->space_id && table->space) {
-		if (!dict_mutex_own) {
-			dict_sys.mutex_lock();
+		if (!dict_locked) {
+			dict_sys.freeze(SRW_LOCK_CALL);
 		}
 
 		table->flags |= 1 << DICT_TF_POS_DATA_DIR
@@ -2176,8 +2173,8 @@ dict_get_and_save_data_dir_path(
 				& ((1U << DICT_TF_BITS) - 1);
 		}
 
-		if (!dict_mutex_own) {
-			dict_sys.mutex_unlock();
+		if (!dict_locked) {
+			dict_sys.unfreeze();
 		}
 	}
 }
@@ -2288,7 +2285,7 @@ static dict_table_t *dict_load_table_one(const span<const char> &name,
 	DBUG_PRINT("dict_load_table_one",
 		   ("table: %.*s", name.size(), name.data()));
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	heap = mem_heap_create(32000);
 
@@ -2544,10 +2541,10 @@ dict_load_table_on_id(
 	ulint		len;
 	mtr_t		mtr;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	/* NOTE that the operation of this function is protected by
-	the dictionary mutex, and therefore no deadlocks can occur
+	dict_sys.latch, and therefore no deadlocks can occur
 	with other dictionary operations. */
 
 	mtr.start();
@@ -2627,7 +2624,7 @@ dict_load_sys_table(
 {
 	mem_heap_t*	heap;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	heap = mem_heap_create(1000);
 
@@ -2662,7 +2659,7 @@ dict_load_foreign_cols(
 	mtr_t		mtr;
 	size_t		id_len;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	id_len = strlen(foreign->id);
 
@@ -2804,7 +2801,7 @@ dict_load_foreign(
 	DBUG_PRINT("dict_load_foreign",
 		   ("id: '%s', check_recursive: %d", id, check_recursive));
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	id_len = strlen(id);
 
@@ -2977,7 +2974,7 @@ dict_load_foreigns(
 
 	DBUG_ENTER("dict_load_foreigns");
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	if (!dict_sys.sys_foreign || !dict_sys.sys_foreign_cols) {
 		if (ignore_err & DICT_ERR_IGNORE_FK_NOKEY) {
diff --git a/storage/innobase/dict/dict0stats.cc b/storage/innobase/dict/dict0stats.cc
index eb2f06ab31e..ee72d7db3cd 100644
--- a/storage/innobase/dict/dict0stats.cc
+++ b/storage/innobase/dict/dict0stats.cc
@@ -326,7 +326,6 @@ static bool innodb_index_stats_not_found_reported;
 Checks whether a table exists and whether it has the given structure.
 The table must have the same number of columns with the same names and
 types. The order of the columns does not matter.
-The caller must own the dictionary mutex.
 dict_table_schema_check() @{
 @return DB_SUCCESS if the table exists and contains the necessary columns */
 static
@@ -478,21 +477,16 @@ dict_table_schema_check(
 Checks whether the persistent statistics storage exists and that all
 tables have the proper structure.
 @return true if exists and all tables are ok */
-static
-bool
-dict_stats_persistent_storage_check(
-/*================================*/
-	bool	caller_has_dict_sys_mutex)	/*!< in: true if the caller
-						owns dict_sys.mutex */
+static bool dict_stats_persistent_storage_check(bool dict_already_locked)
 {
 	char		errstr[512];
 	dberr_t		ret;
 
-	if (!caller_has_dict_sys_mutex) {
-		dict_sys.mutex_lock();
+	if (!dict_already_locked) {
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	/* first check table_stats */
 	ret = dict_table_schema_check(&table_stats_schema, errstr,
@@ -503,8 +497,8 @@ dict_stats_persistent_storage_check(
 					      sizeof(errstr));
 	}
 
-	if (!caller_has_dict_sys_mutex) {
-		dict_sys.mutex_unlock();
+	if (!dict_already_locked) {
+		dict_sys.unlock();
 	}
 
 	if (ret != DB_SUCCESS && ret != DB_STATS_DO_NOT_EXIST) {
@@ -530,7 +524,7 @@ only in the case of error, but not freed.
 static
 dberr_t dict_stats_exec_sql(pars_info_t *pinfo, const char* sql, trx_t *trx)
 {
-  ut_d(dict_sys.assert_locked());
+  ut_ad(dict_sys.locked());
 
   if (!dict_stats_persistent_storage_check(true))
   {
@@ -1014,7 +1008,7 @@ dict_table_t*
 dict_stats_snapshot_create(
 	dict_table_t*	table)
 {
-	dict_sys.mutex_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	dict_stats_assert_initialized(table);
 
@@ -1035,7 +1029,7 @@ dict_stats_snapshot_create(
 	t->stats_sample_pages = table->stats_sample_pages;
 	t->stats_bg_flag = table->stats_bg_flag;
 
-	dict_sys.mutex_unlock();
+	dict_sys.unlock();
 
 	return(t);
 }
@@ -1074,14 +1068,13 @@ dict_stats_update_transient_for_index(
 		Initialize some bogus index cardinality
 		statistics, so that the data can be queried in
 		various means, also via secondary indexes. */
+dummy_empty:
 		index->table->stats_mutex_lock();
 		dict_stats_empty_index(index, false);
 		index->table->stats_mutex_unlock();
 #if defined UNIV_DEBUG || defined UNIV_IBUF_DEBUG
 	} else if (ibuf_debug && !dict_index_is_clust(index)) {
-		index->table->stats_mutex_lock();
-		dict_stats_empty_index(index, false);
-		index->table->stats_mutex_unlock();
+		goto dummy_empty;
 #endif /* UNIV_DEBUG || UNIV_IBUF_DEBUG */
 	} else {
 		mtr_t	mtr;
@@ -1102,10 +1095,7 @@ dict_stats_update_transient_for_index(
 
 		switch (size) {
 		case ULINT_UNDEFINED:
-			index->table->stats_mutex_lock();
-			dict_stats_empty_index(index, false);
-			index->table->stats_mutex_unlock();
-			return;
+			goto dummy_empty;
 		case 0:
 			/* The root node of the tree is a leaf */
 			size = 1;
@@ -2586,7 +2576,7 @@ dict_stats_save_index_stat(
 	char		db_utf8[MAX_DB_UTF8_LEN];
 	char		table_utf8[MAX_TABLE_UTF8_LEN];
 
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	dict_fs2utf8(index->table->name.m_name, db_utf8, sizeof(db_utf8),
 		     table_utf8, sizeof(table_utf8));
@@ -2722,7 +2712,7 @@ dict_stats_save(
 	trx_t*	trx = trx_create();
 	trx_start_internal(trx);
 	trx->dict_operation_lock_mode = RW_X_LATCH;
-	dict_sys_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	pinfo = pars_info_create();
 
@@ -2764,7 +2754,7 @@ rollback_and_exit:
 		trx->rollback();
 free_and_exit:
 		trx->dict_operation_lock_mode = 0;
-		dict_sys_unlock();
+		dict_sys.unlock();
 		trx->free();
 		dict_stats_snapshot_free(table);
 		return ret;
@@ -3634,7 +3624,7 @@ dberr_t dict_stats_delete_from_table_stats(const char *database_name,
 {
 	pars_info_t*	pinfo;
 
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	pinfo = pars_info_create();
 
@@ -3661,7 +3651,7 @@ dberr_t dict_stats_delete_from_index_stats(const char *database_name,
 {
 	pars_info_t*	pinfo;
 
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	pinfo = pars_info_create();
 
@@ -3690,7 +3680,7 @@ dberr_t dict_stats_delete_from_index_stats(const char *database_name,
 {
 	pars_info_t*	pinfo;
 
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	pinfo = pars_info_create();
 
diff --git a/storage/innobase/dict/dict0stats_bg.cc b/storage/innobase/dict/dict0stats_bg.cc
index dbaff373ac4..209a2f33ba9 100644
--- a/storage/innobase/dict/dict0stats_bg.cc
+++ b/storage/innobase/dict/dict0stats_bg.cc
@@ -146,8 +146,6 @@ schedule new estimates for table and index statistics to be calculated.
 void dict_stats_update_if_needed_func(dict_table_t *table)
 #endif
 {
-	dict_sys.assert_not_locked();
-
 	if (UNIV_UNLIKELY(!table->stat_initialized)) {
 		/* The table may have been evicted from dict_sys
 		and reloaded internally by InnoDB for FOREIGN KEY
@@ -258,7 +256,7 @@ dict_stats_recalc_pool_del(
 	const dict_table_t*	table)	/*!< in: table to remove */
 {
 	ut_ad(!srv_read_only_mode);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 
 	mysql_mutex_lock(&recalc_pool_mutex);
 
@@ -280,23 +278,13 @@ dict_stats_recalc_pool_del(
 
 /*****************************************************************//**
 Wait until background stats thread has stopped using the specified table.
-The caller must have locked the data dictionary using
-row_mysql_lock_data_dictionary() and this function may unlock it temporarily
-and restore the lock before it exits.
 The background stats thread is guaranteed not to start using the specified
-table after this function returns and before the caller unlocks the data
-dictionary because it sets the BG_STAT_IN_PROGRESS bit in table->stats_bg_flag
-under dict_sys.mutex. */
-void
-dict_stats_wait_bg_to_stop_using_table(
-/*===================================*/
-	dict_table_t*	table,	/*!< in/out: table */
-	trx_t*		trx)	/*!< in/out: transaction to use for
-				unlocking/locking the data dict */
+table after this function returns and before the caller releases
+dict_sys.latch. */
+void dict_stats_wait_bg_to_stop_using_table(dict_table_t *table)
 {
-	while (!dict_stats_stop_bg(table)) {
-		DICT_BG_YIELD(trx);
-	}
+  while (!dict_stats_stop_bg(table))
+    DICT_BG_YIELD;
 }
 
 /*****************************************************************//**
@@ -347,7 +335,7 @@ next_table_id:
 
 	dict_table_t*	table;
 
-	dict_sys.mutex_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	table = dict_table_open_on_id(table_id, TRUE, DICT_TABLE_OP_NORMAL);
 
@@ -362,13 +350,13 @@ next_table_id:
 	if (!table->is_accessible()) {
 		dict_table_close(table, TRUE, FALSE);
 no_table:
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 		goto next_table_id;
 	}
 
 	table->stats_bg_flag |= BG_STAT_IN_PROGRESS;
 
-	dict_sys.mutex_unlock();
+	dict_sys.unlock();
 
 	/* time() could be expensive, the current function
 	is called once every time a table has been changed more than 10% and
@@ -393,13 +381,13 @@ no_table:
 		ret = true;
 	}
 
-	dict_sys.mutex_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	table->stats_bg_flag = BG_STAT_NONE;
 
 	dict_table_close(table, TRUE, FALSE);
 
-	dict_sys.mutex_unlock();
+	dict_sys.unlock();
 	return ret;
 }
 
diff --git a/storage/innobase/dict/drop.cc b/storage/innobase/dict/drop.cc
index b2aab6992bb..bb0af50c4c2 100644
--- a/storage/innobase/dict/drop.cc
+++ b/storage/innobase/dict/drop.cc
@@ -78,7 +78,7 @@ before transaction commit and must be rolled back explicitly are as follows:
 @return error code */
 dberr_t trx_t::drop_table_foreign(const table_name_t &name)
 {
-  ut_d(dict_sys.assert_locked());
+  ut_ad(dict_sys.locked());
   ut_ad(state == TRX_STATE_ACTIVE);
   ut_ad(dict_operation);
   ut_ad(dict_operation_lock_mode == RW_X_LATCH);
@@ -116,7 +116,7 @@ dberr_t trx_t::drop_table_foreign(const table_name_t &name)
 @return error code */
 dberr_t trx_t::drop_table_statistics(const table_name_t &name)
 {
-  ut_d(dict_sys.assert_locked());
+  ut_ad(dict_sys.locked());
   ut_ad(dict_operation_lock_mode == RW_X_LATCH);
 
   if (strstr(name.m_name, "/" TEMP_FILE_PREFIX_INNODB) ||
@@ -143,7 +143,7 @@ dberr_t trx_t::drop_table_statistics(const table_name_t &name)
 @return error code */
 dberr_t trx_t::drop_table(const dict_table_t &table)
 {
-  ut_d(dict_sys.assert_locked());
+  ut_ad(dict_sys.locked());
   ut_ad(state == TRX_STATE_ACTIVE);
   ut_ad(dict_operation);
   ut_ad(dict_operation_lock_mode == RW_X_LATCH);
@@ -235,7 +235,7 @@ void trx_t::commit(std::vector<pfs_os_file_t> &deleted)
   commit_persist();
   if (dict_operation)
   {
-    dict_sys.assert_locked();
+    ut_ad(dict_sys.locked());
     for (const auto &p : mod_tables)
     {
       if (p.second.is_dropped())
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 0d4aa4901f8..94f9df63149 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -2098,7 +2098,7 @@ to the .err log. This function is used to open a tablespace when we start
 mysqld after the dictionary has been booted, and also in IMPORT TABLESPACE.
 
 NOTE that we assume this operation is used either at the database startup
-or under the protection of the dictionary mutex, so that two users cannot
+or under the protection of dict_sys.latch, so that two users cannot
 race here. This operation does not leave the file associated with the
 tablespace open, but closes it after we have looked at the space id in it.
 
diff --git a/storage/innobase/fts/fts0fts.cc b/storage/innobase/fts/fts0fts.cc
index 7b6ca048271..40a8c270a25 100644
--- a/storage/innobase/fts/fts0fts.cc
+++ b/storage/innobase/fts/fts0fts.cc
@@ -159,6 +159,7 @@ const  fts_index_selector_t fts_index_selector[] = {
 
 /** Default config values for FTS indexes on a table. */
 static const char* fts_config_table_insert_values_sql =
+	"PROCEDURE P() IS\n"
 	"BEGIN\n"
 	"\n"
 	"INSERT INTO $config_table VALUES('"
@@ -174,7 +175,8 @@ static const char* fts_config_table_insert_values_sql =
 		FTS_TOTAL_DELETED_COUNT "', '0');\n"
 	"" /* Note: 0 == FTS_TABLE_STATE_RUNNING */
 	"INSERT INTO $config_table VALUES ('"
-		FTS_TABLE_STATE "', '0');\n";
+		FTS_TABLE_STATE "', '0');\n"
+	"END;\n";
 
 /** FTS tokenize parmameter for plugin parser */
 struct fts_tokenize_param_t {
@@ -185,9 +187,9 @@ struct fts_tokenize_param_t {
 /** Free a query graph */
 void fts_que_graph_free(que_t *graph)
 {
-  dict_sys.mutex_lock();
+  dict_sys.lock(SRW_LOCK_CALL);
   que_graph_free(graph);
-  dict_sys.mutex_unlock();
+  dict_sys.unlock();
 }
 
 /** Run SYNC on the table, i.e., write out data from the cache to the
@@ -455,7 +457,7 @@ fts_load_user_stopword(
 	fts_stopword_t*	stopword_info)		/*!< in: Stopword info */
 {
 	if (!fts->dict_locked) {
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
 	/* Validate the user table existence in the right format */
@@ -464,7 +466,7 @@ fts_load_user_stopword(
 	if (!stopword_info->charset) {
 cleanup:
 		if (!fts->dict_locked) {
-			dict_sys.mutex_unlock();
+			dict_sys.unlock();
 		}
 
 		return ret;
@@ -489,8 +491,9 @@ cleanup:
 	pars_info_bind_function(info, "my_func", fts_read_stopword,
 				stopword_info);
 
-	que_t* graph = fts_parse_sql_no_dict_lock(
+	que_t* graph = pars_sql(
 		info,
+		"PROCEDURE P() IS\n"
 		"DECLARE FUNCTION my_func;\n"
 		"DECLARE CURSOR c IS"
 		" SELECT value"
@@ -504,7 +507,8 @@ cleanup:
 		"    EXIT;\n"
 		"  END IF;\n"
 		"END LOOP;\n"
-		"CLOSE c;");
+		"CLOSE c;"
+		"END;\n");
 
 	for (;;) {
 		dberr_t error = fts_eval_sql(trx, graph);
@@ -882,7 +886,7 @@ fts_drop_index(
 }
 
 /****************************************************************//**
-Free the query graph but check whether dict_sys.mutex is already
+Free the query graph but check whether dict_sys.latch is already
 held */
 void
 fts_que_graph_free_check_lock(
@@ -904,15 +908,15 @@ fts_que_graph_free_check_lock(
 	}
 
 	if (!has_dict) {
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	que_graph_free(graph);
 
 	if (!has_dict) {
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 }
 
@@ -1551,19 +1555,19 @@ static dberr_t fts_lock_table(trx_t *trx, const char *table_name)
   {
     dberr_t err= lock_table_for_trx(table, trx, LOCK_X);
     /* Wait for purge threads to stop using the table. */
-    dict_sys.mutex_lock();
+    dict_sys.freeze(SRW_LOCK_CALL);
     for (uint n= 15; table->get_ref_count() > 1; )
     {
-      dict_sys.mutex_unlock();
+      dict_sys.unfreeze();
       if (!--n)
       {
         err= DB_LOCK_WAIT_TIMEOUT;
         goto fail;
       }
       std::this_thread::sleep_for(std::chrono::milliseconds(50));
-      dict_sys.mutex_lock();
+      dict_sys.freeze(SRW_LOCK_CALL);
     }
-    dict_sys.mutex_unlock();
+    dict_sys.unfreeze();
 fail:
     table->release();
     return err;
@@ -1905,7 +1909,7 @@ fts_create_common_tables(
 	fts_get_table_name(&fts_table, fts_name, true);
 	pars_info_bind_id(info, "config_table", fts_name);
 
-	graph = fts_parse_sql_no_dict_lock(
+	graph = pars_sql(
 		info, fts_config_table_insert_values_sql);
 
 	error = fts_eval_sql(trx, graph);
@@ -2049,8 +2053,7 @@ fts_create_index_tables(trx_t* trx, const dict_index_t* index, table_id_t id)
 		dict_table_t*	new_table;
 
 		/* Create the FTS auxiliary tables that are specific
-		to an FTS index. We need to preserve the table_id %s
-		which fts_parse_sql_no_dict_lock() will fill in for us. */
+		to an FTS index. */
 		fts_table.suffix = fts_get_suffix(i);
 
 		new_table = fts_create_one_index_table(
@@ -6038,8 +6041,6 @@ fts_init_index(
 	fts_cache_t*    cache = table->fts->cache;
 	bool		need_init = false;
 
-	dict_sys.assert_not_locked();
-
 	/* First check cache->get_docs is initialized */
 	if (!has_cache_lock) {
 		mysql_mutex_lock(&cache->lock);
@@ -6103,9 +6104,9 @@ func_exit:
 	}
 
 	if (need_init) {
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 		/* Register the table with the optimize thread. */
 		fts_optimize_add_table(table);
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 }
diff --git a/storage/innobase/fts/fts0opt.cc b/storage/innobase/fts/fts0opt.cc
index 64179947832..2fcbf9b28e3 100644
--- a/storage/innobase/fts/fts0opt.cc
+++ b/storage/innobase/fts/fts0opt.cc
@@ -1603,8 +1603,21 @@ fts_optimize_create(
 	optim->fts_index_table.table = table;
 
 	/* The common prefix for all this parent table's aux tables. */
-	optim->name_prefix = fts_get_table_name_prefix(
-		&optim->fts_common_table);
+	char table_id[FTS_AUX_MIN_TABLE_ID_LENGTH];
+	const size_t table_id_len = 1
+		+ size_t(fts_get_table_id(&optim->fts_common_table, table_id));
+	dict_sys.freeze(SRW_LOCK_CALL);
+	/* Include the separator as well. */
+	const size_t dbname_len = table->name.dblen() + 1;
+	ut_ad(dbname_len > 1);
+	const size_t prefix_name_len = dbname_len + 4 + table_id_len;
+	char* prefix_name = static_cast<char*>(
+		ut_malloc_nokey(prefix_name_len));
+	memcpy(prefix_name, table->name.m_name, dbname_len);
+	dict_sys.unfreeze();
+	memcpy(prefix_name + dbname_len, "FTS_", 4);
+	memcpy(prefix_name + dbname_len + 4, table_id, table_id_len);
+	optim->name_prefix =prefix_name;
 
 	return(optim);
 }
@@ -2589,7 +2602,6 @@ fts_optimize_remove_table(
 
   if (table->fts->in_queue)
   {
-    dict_sys.assert_not_locked();
     fts_msg_t *msg= fts_optimize_create_msg(FTS_MSG_DEL_TABLE, nullptr);
     pthread_cond_t cond;
     pthread_cond_init(&cond, nullptr);
@@ -2937,7 +2949,7 @@ fts_optimize_init(void)
 	/* Add fts tables to fts_slots which could be skipped
 	during dict_load_table_one() because fts_optimize_thread
 	wasn't even started. */
-	dict_sys.mutex_lock();
+	dict_sys.freeze(SRW_LOCK_CALL);
 	for (dict_table_t* table = UT_LIST_GET_FIRST(dict_sys.table_LRU);
 	     table != NULL;
 	     table = UT_LIST_GET_NEXT(table_LRU, table)) {
@@ -2952,7 +2964,7 @@ fts_optimize_init(void)
 		fts_optimize_new_table(table);
 		table->fts->in_queue = true;
 	}
-	dict_sys.mutex_unlock();
+	dict_sys.unfreeze();
 
 	pthread_cond_init(&fts_opt_shutdown_cond, nullptr);
 	last_check_sync_time = time(NULL);
@@ -2966,13 +2978,13 @@ fts_optimize_shutdown()
 
 	/* If there is an ongoing activity on dictionary, such as
 	srv_master_evict_from_table_cache(), wait for it */
-	dict_sys.mutex_lock();
+	dict_sys.freeze(SRW_LOCK_CALL);
 	mysql_mutex_lock(&fts_optimize_wq->mutex);
 	/* Tells FTS optimizer system that we are exiting from
 	optimizer thread, message send their after will not be
 	processed */
 	fts_opt_start_shutdown = true;
-	dict_sys.mutex_unlock();
+	dict_sys.unfreeze();
 
 	/* We tell the OPTIMIZE thread to switch to state done, we
 	can't delete the work queue here because the add thread needs
diff --git a/storage/innobase/fts/fts0sql.cc b/storage/innobase/fts/fts0sql.cc
index 2d41776eb12..1970f6f584f 100644
--- a/storage/innobase/fts/fts0sql.cc
+++ b/storage/innobase/fts/fts0sql.cc
@@ -86,44 +86,21 @@ fts_get_table_id(
 
 /** Construct the name of an internal FTS table for the given table.
 @param[in]	fts_table	metadata on fulltext-indexed table
-@param[in]	dict_locked	whether dict_sys.mutex is being held
-@return	the prefix, must be freed with ut_free() */
-char* fts_get_table_name_prefix(const fts_table_t* fts_table)
-{
-	char		table_id[FTS_AUX_MIN_TABLE_ID_LENGTH];
-	const size_t table_id_len = size_t(fts_get_table_id(fts_table,
-							    table_id)) + 1;
-	dict_sys.mutex_lock();
-	/* Include the separator as well. */
-	const size_t dbname_len = fts_table->table->name.dblen() + 1;
-	ut_ad(dbname_len > 1);
-	const size_t prefix_name_len = dbname_len + 4 + table_id_len;
-	char* prefix_name = static_cast<char*>(
-		ut_malloc_nokey(prefix_name_len));
-	memcpy(prefix_name, fts_table->table->name.m_name, dbname_len);
-	dict_sys.mutex_unlock();
-	memcpy(prefix_name + dbname_len, "FTS_", 4);
-	memcpy(prefix_name + dbname_len + 4, table_id, table_id_len);
-	return prefix_name;
-}
-
-/** Construct the name of an internal FTS table for the given table.
-@param[in]	fts_table	metadata on fulltext-indexed table
 @param[out]	table_name	a name up to MAX_FULL_NAME_LEN
-@param[in]	dict_locked	whether dict_sys.mutex is being held */
+@param[in]	dict_locked	whether dict_sys.latch is being held */
 void fts_get_table_name(const fts_table_t* fts_table, char* table_name,
 			bool dict_locked)
 {
 	if (!dict_locked) {
-		dict_sys.mutex_lock();
+		dict_sys.freeze(SRW_LOCK_CALL);
 	}
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.frozen());
 	/* Include the separator as well. */
 	const size_t dbname_len = fts_table->table->name.dblen() + 1;
 	ut_ad(dbname_len > 1);
 	memcpy(table_name, fts_table->table->name.m_name, dbname_len);
 	if (!dict_locked) {
-		dict_sys.mutex_unlock();
+		dict_sys.unfreeze();
 	}
 	memcpy(table_name += dbname_len, "FTS_", 4);
 	table_name += 4;
@@ -153,14 +130,14 @@ fts_parse_sql(
 
 	if (!dict_locked) {
 		/* The InnoDB SQL parser is not re-entrant. */
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
 	graph = pars_sql(info, str);
 	ut_a(graph);
 
 	if (!dict_locked) {
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 
 	ut_free(str);
@@ -169,30 +146,6 @@ fts_parse_sql(
 }
 
 /******************************************************************//**
-Parse an SQL string.
-@return query graph */
-que_t*
-fts_parse_sql_no_dict_lock(
-/*=======================*/
-	pars_info_t*	info,		/*!< in: info struct, or NULL */
-	const char*	sql)		/*!< in: SQL string to evaluate */
-{
-	char*		str;
-	que_t*		graph;
-
-	dict_sys.assert_locked();
-
-	str = ut_str3cat(fts_sql_begin, sql, fts_sql_end);
-
-	graph = pars_sql(info, str);
-	ut_a(graph);
-
-	ut_free(str);
-
-	return(graph);
-}
-
-/******************************************************************//**
 Evaluate an SQL query graph.
 @return DB_SUCCESS or error code */
 dberr_t
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 1485e8aa0df..0a57b847d14 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -503,7 +503,6 @@ static mysql_pfs_key_t	pending_checkpoint_mutex_key;
 # ifdef UNIV_PFS_MUTEX
 mysql_pfs_key_t	buf_pool_mutex_key;
 mysql_pfs_key_t	dict_foreign_err_mutex_key;
-mysql_pfs_key_t	dict_sys_mutex_key;
 mysql_pfs_key_t	fil_system_mutex_key;
 mysql_pfs_key_t	flush_list_mutex_key;
 mysql_pfs_key_t	fts_cache_mutex_key;
@@ -545,7 +544,6 @@ static PSI_mutex_info all_innodb_mutexes[] = {
 	PSI_KEY(pending_checkpoint_mutex),
 	PSI_KEY(buf_pool_mutex),
 	PSI_KEY(dict_foreign_err_mutex),
-	PSI_KEY(dict_sys_mutex),
 	PSI_KEY(recalc_pool_mutex),
 	PSI_KEY(fil_system_mutex),
 	PSI_KEY(flush_list_mutex),
@@ -5497,7 +5495,7 @@ is done when the table first opened.
 @param[in,out]	s_templ		InnoDB template structure
 @param[in]	add_v		new virtual columns added along with
 				add index call
-@param[in]	locked		true if dict_sys mutex is held */
+@param[in]	locked		true if dict_sys.latch is held */
 void
 innobase_build_v_templ(
 	const TABLE*		table,
@@ -5520,12 +5518,18 @@ innobase_build_v_templ(
 	ut_ad(n_v_col > 0);
 
 	if (!locked) {
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
+#if 0
+	/* This does not (need to) hold for ctx->new_table in
+	alter_rebuild_apply_log() */
+	ut_ad(dict_sys.locked());
+#endif
+
 	if (s_templ->vtempl) {
 		if (!locked) {
-			dict_sys.mutex_unlock();
+			dict_sys.unlock();
 		}
 		DBUG_VOID_RETURN;
 	}
@@ -5631,7 +5635,7 @@ innobase_build_v_templ(
 	}
 
 	if (!locked) {
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 
 	s_templ->db_name = table->s->db.str;
@@ -5919,7 +5923,7 @@ ha_innobase::open(const char* name, int, uint)
 	key_used_on_scan = m_primary_key;
 
 	if (ib_table->n_v_cols) {
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 		if (ib_table->vc_templ == NULL) {
 			ib_table->vc_templ = UT_NEW_NOKEY(dict_vcol_templ_t());
 			innobase_build_v_templ(
@@ -5927,7 +5931,7 @@ ha_innobase::open(const char* name, int, uint)
 				true);
 		}
 
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 
 	if (!check_index_consistency(table, ib_table)) {
@@ -9853,7 +9857,7 @@ wsrep_append_foreign_key(
 		foreign->referenced_table : foreign->foreign_table)) {
 		WSREP_DEBUG("pulling %s table into cache",
 			    (referenced) ? "referenced" : "foreign");
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 
 		if (referenced) {
 			foreign->referenced_table =
@@ -9888,7 +9892,7 @@ wsrep_append_foreign_key(
 						TRUE, FALSE);
 			}
 		}
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 
 	if ( !((referenced) ?
@@ -13016,9 +13020,9 @@ create_table_info_t::create_table_update_dict()
 			DBUG_RETURN(-1);
 		}
 
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 		fts_optimize_add_table(innobase_table);
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 	}
 
 	if (const Field* ai = m_form->found_next_number_field) {
@@ -13277,12 +13281,12 @@ ha_innobase::discard_or_import_tablespace(
 	btr_cur_instant_init(). */
 	table_id_t id = m_prebuilt->table->id;
 	ut_ad(id);
-	dict_sys.mutex_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 	dict_table_close(m_prebuilt->table, TRUE, FALSE);
 	dict_sys.remove(m_prebuilt->table);
 	m_prebuilt->table = dict_table_open_on_id(id, TRUE,
 						  DICT_TABLE_OP_NORMAL);
-	dict_sys.mutex_unlock();
+	dict_sys.unlock();
 	if (!m_prebuilt->table) {
 		err = DB_TABLE_NOT_FOUND;
 	} else {
@@ -13673,7 +13677,7 @@ int ha_innobase::truncate()
 	}
 
 	row_mysql_lock_data_dictionary(trx);
-	dict_stats_wait_bg_to_stop_using_table(ib_table, trx);
+	dict_stats_wait_bg_to_stop_using_table(ib_table);
 	/* Wait for purge threads to stop using the table. */
 	for (uint n = 15; ib_table->get_ref_count() > 1; ) {
 		if (!--n) {
@@ -14303,8 +14307,6 @@ ha_innobase::info_low(
 
 	DEBUG_SYNC_C("ha_innobase_info_low");
 
-	dict_sys.assert_not_locked();
-
 	/* If we are forcing recovery at a high level, we will suppress
 	statistics calculation on tables, because that may crash the
 	server if an index is badly corrupted. */
@@ -14335,13 +14337,11 @@ ha_innobase::info_low(
 			if (dict_stats_is_persistent_enabled(ib_table)) {
 
 				if (is_analyze) {
-					row_mysql_lock_data_dictionary(
-						m_prebuilt->trx);
+					dict_sys.lock(SRW_LOCK_CALL);
 					dict_stats_recalc_pool_del(ib_table);
 					dict_stats_wait_bg_to_stop_using_table(
-						ib_table, m_prebuilt->trx);
-					row_mysql_unlock_data_dictionary(
-						m_prebuilt->trx);
+						ib_table);
+					dict_sys.unlock();
 					opt = DICT_STATS_RECALC_PERSISTENT;
 				} else {
 					/* This is e.g. 'SHOW INDEXES', fetch
@@ -14355,10 +14355,12 @@ ha_innobase::info_low(
 			ret = dict_stats_update(ib_table, opt);
 
 			if (opt == DICT_STATS_RECALC_PERSISTENT) {
-				dict_sys.mutex_lock();
+				dict_sys.freeze(SRW_LOCK_CALL);
+				ib_table->stats_mutex_lock();
 				ib_table->stats_bg_flag
 					&= byte(~BG_STAT_SHOULD_QUIT);
-				dict_sys.mutex_unlock();
+				ib_table->stats_mutex_unlock();
+				dict_sys.unfreeze();
 			}
 
 			if (ret != DB_SUCCESS) {
@@ -15160,10 +15162,7 @@ get_foreign_key_info(
 	/* Load referenced table to update FK referenced key name. */
 	if (foreign->referenced_table == NULL) {
 
-		dict_table_t*	ref_table;
-
-		dict_sys.assert_locked();
-		ref_table = dict_table_open_on_name(
+		dict_table_t*	ref_table = dict_table_open_on_name(
 			foreign->referenced_table_name_lookup,
 			TRUE, FALSE, DICT_ERR_IGNORE_NONE);
 
@@ -15217,7 +15216,7 @@ ha_innobase::get_foreign_key_list(
 
 	m_prebuilt->trx->op_info = "getting list of foreign keys";
 
-	dict_sys.mutex_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	for (dict_foreign_set::iterator it
 		= m_prebuilt->table->foreign_set.begin();
@@ -15234,7 +15233,7 @@ ha_innobase::get_foreign_key_list(
 		}
 	}
 
-	dict_sys.mutex_unlock();
+	dict_sys.unlock();
 
 	m_prebuilt->trx->op_info = "";
 
@@ -15255,7 +15254,7 @@ ha_innobase::get_parent_foreign_key_list(
 
 	m_prebuilt->trx->op_info = "getting list of referencing foreign keys";
 
-	dict_sys.mutex_lock();
+	dict_sys.freeze(SRW_LOCK_CALL);
 
 	for (dict_foreign_set::iterator it
 		= m_prebuilt->table->referenced_set.begin();
@@ -15272,7 +15271,7 @@ ha_innobase::get_parent_foreign_key_list(
 		}
 	}
 
-	dict_sys.mutex_unlock();
+	dict_sys.unfreeze();
 
 	m_prebuilt->trx->op_info = "";
 
@@ -19803,9 +19802,8 @@ TABLE* innobase_init_vc_templ(dict_table_t* table)
 		DBUG_RETURN(NULL);
 	}
 
-	dict_sys.mutex_lock();
-	innobase_build_v_templ(mysql_table, table, table->vc_templ, NULL, true);
-	dict_sys.mutex_unlock();
+	innobase_build_v_templ(mysql_table, table, table->vc_templ, NULL,
+			       false);
 	DBUG_RETURN(mysql_table);
 }
 
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 97d56671fe9..e7b01a4f72b 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -486,7 +486,7 @@ inline bool dict_table_t::instant_column(const dict_table_t& table,
 	DBUG_ASSERT(table.n_cols + table.n_dropped() >= n_cols + n_dropped());
 	DBUG_ASSERT(!table.persistent_autoinc
 		    || persistent_autoinc == table.persistent_autoinc);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	{
 		const char* end = table.col_names;
@@ -731,7 +731,7 @@ inline void dict_table_t::rollback_instant(
 	const char*	old_v_col_names,
 	const ulint*	col_map)
 {
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	if (cols == old_cols) {
 		/* Alter fails before instant operation happens.
@@ -2639,7 +2639,7 @@ innobase_init_foreign(
 	ulint		referenced_num_field)	/*!< in: number of referenced
 						columns */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
         if (constraint_name) {
                 ulint   db_len;
@@ -3043,7 +3043,7 @@ innobase_get_foreign_key_info(
 
 		add_fk[num_fk] = dict_mem_foreign_create();
 
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 
 		referenced_table_name = dict_get_referenced_table(
 			table->name.m_name,
@@ -3059,11 +3059,9 @@ innobase_get_foreign_key_info(
 				referenced_table = NULL;);
 
 		if (!referenced_table && trx->check_foreigns) {
-			dict_sys.mutex_unlock();
 			my_error(ER_FK_CANNOT_OPEN_PARENT,
 				 MYF(0), fk_key->ref_table.str);
-
-			goto err_exit;
+			goto err_exit_unlock;
 		}
 
 		if (fk_key->ref_columns.elements > 0) {
@@ -3092,12 +3090,11 @@ innobase_get_foreign_key_info(
 				/* Check whether there exist such
 				index in the the index create clause */
 				if (!referenced_index) {
-					dict_sys.mutex_unlock();
 					my_error(ER_FK_NO_INDEX_PARENT, MYF(0),
 						 fk_key->name.str
 						 ? fk_key->name.str : "",
 						 fk_key->ref_table.str);
-					goto err_exit;
+					goto err_exit_unlock;
 				}
 			} else {
 				ut_a(!trx->check_foreigns);
@@ -3107,10 +3104,9 @@ innobase_get_foreign_key_info(
 		} else {
 			/* Not possible to add a foreign key without a
 			referenced column */
-			dict_sys.mutex_unlock();
 			my_error(ER_CANNOT_ADD_FOREIGN, MYF(0),
 				 fk_key->ref_table.str);
-			goto err_exit;
+			goto err_exit_unlock;
 		}
 
 		if (!innobase_init_foreign(
@@ -3119,15 +3115,14 @@ innobase_get_foreign_key_info(
 			    num_col, referenced_table_name,
 			    referenced_table, referenced_index,
 			    referenced_column_names, referenced_num_col)) {
-			dict_sys.mutex_unlock();
 			my_error(
 				ER_DUP_CONSTRAINT_NAME,
 				MYF(0),
                                 "FOREIGN KEY", add_fk[num_fk]->id);
-			goto err_exit;
+			goto err_exit_unlock;
 		}
 
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 
 		correct_option = innobase_set_foreign_key_option(
 			add_fk[num_fk], fk_key);
@@ -3158,6 +3153,8 @@ innobase_get_foreign_key_info(
 	*n_add_fk = num_fk;
 
 	DBUG_RETURN(true);
+err_exit_unlock:
+	dict_sys.unlock();
 err_exit:
 	for (ulint i = 0; i <= num_fk; i++) {
 		if (add_fk[i]) {
@@ -4073,7 +4070,7 @@ online_retry_drop_indexes_low(
 	dict_table_t*	table,	/*!< in/out: table */
 	trx_t*		trx)	/*!< in/out: transaction */
 {
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 	ut_ad(trx->dict_operation);
 
@@ -4129,9 +4126,9 @@ online_retry_drop_indexes(
 		trx->free();
 	}
 
-	ut_d(dict_sys.mutex_lock());
+	ut_d(dict_sys.freeze(SRW_LOCK_CALL));
 	ut_d(dict_table_check_for_dup_indexes(table, CHECK_ALL_COMPLETE));
-	ut_d(dict_sys.mutex_unlock());
+	ut_d(dict_sys.unfreeze());
 	ut_ad(!table->drop_aborted);
 }
 
@@ -4212,7 +4209,7 @@ innobase_check_foreigns_low(
 	bool			drop)
 {
 	dict_foreign_t*	foreign;
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	/* Check if any FOREIGN KEY constraints are defined on this
 	column. */
@@ -4811,7 +4808,7 @@ innobase_update_gis_column_type(
 
 	DBUG_ASSERT(trx->dict_operation);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	info = pars_info_create();
 
@@ -6327,7 +6324,7 @@ acquire_lock:
 	at least until checking ut_ad(user_table->n_ref_count == 1) below.
 	XXX what may happen if bg stats opens the table after we
 	have unlocked data dictionary below? */
-	dict_stats_wait_bg_to_stop_using_table(user_table, ctx->trx);
+	dict_stats_wait_bg_to_stop_using_table(user_table);
 	ut_d(stats_wait = true);
 
 	online_retry_drop_indexes_low(ctx->new_table, ctx->trx);
@@ -7056,7 +7053,7 @@ error_handling_drop_uncached:
 	if (fts_index) {
 		ut_ad(ctx->trx->dict_operation);
 		ut_ad(ctx->trx->dict_operation_lock_mode == RW_X_LATCH);
-		ut_d(dict_sys.assert_locked());
+		ut_ad(dict_sys.locked());
 
 		DICT_TF2_FLAG_SET(ctx->new_table, DICT_TF2_FTS);
 		if (ctx->need_rebuild()) {
@@ -7342,7 +7339,7 @@ rename_index_try(
 	trx_t*			trx)
 {
 	DBUG_ENTER("rename_index_try");
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 
 	pars_info_t*	pinfo;
@@ -7400,7 +7397,7 @@ void
 innobase_rename_index_cache(dict_index_t* index, const char* new_name)
 {
 	DBUG_ENTER("innobase_rename_index_cache");
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	size_t	old_name_len = strlen(index->name);
 	size_t	new_name_len = strlen(new_name);
@@ -7559,10 +7556,10 @@ ha_innobase::prepare_inplace_alter_table(
 	}
 #endif /* UNIV_DEBUG */
 
-	ut_d(dict_sys.mutex_lock());
+	ut_d(dict_sys.freeze(SRW_LOCK_CALL));
 	ut_d(dict_table_check_for_dup_indexes(
 		     m_prebuilt->table, CHECK_ABORTED_OK));
-	ut_d(dict_sys.mutex_unlock());
+	ut_d(dict_sys.unfreeze());
 
 	if (!(ha_alter_info->handler_flags & ~INNOBASE_INPLACE_IGNORE)) {
 		/* Nothing to do */
@@ -8507,10 +8504,10 @@ oom:
 		KEY*	dup_key;
 	all_done:
 	case DB_SUCCESS:
-		ut_d(dict_sys.mutex_lock());
+		ut_d(dict_sys.freeze(SRW_LOCK_CALL));
 		ut_d(dict_table_check_for_dup_indexes(
 			     m_prebuilt->table, CHECK_PARTIAL_OK));
-		ut_d(dict_sys.mutex_unlock());
+		ut_d(dict_sys.unfreeze());
 		/* prebuilt->table->n_ref_count can be anything here,
 		given that we hold at most a shared lock on the table. */
 		goto ok_exit;
@@ -8571,7 +8568,7 @@ innobase_online_rebuild_log_free(
 	dict_table_t*	table)
 {
 	dict_index_t* clust_index = dict_table_get_first_index(table);
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 	clust_index->lock.x_lock(SRW_LOCK_CALL);
 
 	if (clust_index->online_log) {
@@ -8776,10 +8773,10 @@ inline bool rollback_inplace_alter_table(Alter_inplace_info *ha_alter_info,
       purge_sys.resume_FTS();
     if (ctx->old_table->fts)
     {
-      dict_sys.mutex_lock();
+      dict_sys.lock(SRW_LOCK_CALL);
       ut_ad(fts_check_cached_index(ctx->old_table));
       fts_optimize_add_table(ctx->old_table);
-      dict_sys.mutex_unlock();
+      dict_sys.unlock();
     }
     goto free_and_exit;
   }
@@ -8850,7 +8847,7 @@ innobase_drop_foreign_try(
 
 	DBUG_ASSERT(trx->dict_operation);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	/* Drop the constraint from the data dictionary. */
 	static const char sql[] =
@@ -8906,7 +8903,7 @@ innobase_rename_column_try(
 
 	DBUG_ASSERT(trx->dict_operation);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	if (ctx.need_rebuild()) {
 		goto rename_foreign;
@@ -9224,7 +9221,7 @@ innobase_rename_or_enlarge_column_try(
 
 	DBUG_ASSERT(trx->dict_operation);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	ulint n_base;
 
@@ -9650,7 +9647,7 @@ innobase_update_foreign_cache(
 
 	DBUG_ENTER("innobase_update_foreign_cache");
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	user_table = ctx->old_table;
 
@@ -10958,7 +10955,7 @@ lock_fail:
 			break;
 		}
 
-		DICT_BG_YIELD(trx);
+		DICT_BG_YIELD;
 	}
 
 	/* Apply the changes to the data dictionary tables, for all
diff --git a/storage/innobase/handler/i_s.cc b/storage/innobase/handler/i_s.cc
index a84d8266b09..ae9aa95339e 100644
--- a/storage/innobase/handler/i_s.cc
+++ b/storage/innobase/handler/i_s.cc
@@ -1335,12 +1335,12 @@ i_s_cmp_per_index_fill_low(
 	RETURN_IF_INNODB_NOT_STARTED(tables->schema_table_name.str);
 
 	/* Create a snapshot of the stats so we do not bump into lock
-	order violations with dict_sys.mutex below. */
+	order violations with dict_sys.latch below. */
 	mysql_mutex_lock(&page_zip_stat_per_index_mutex);
 	page_zip_stat_per_index_t		snap (page_zip_stat_per_index);
 	mysql_mutex_unlock(&page_zip_stat_per_index_mutex);
 
-	dict_sys.mutex_lock();
+	dict_sys.freeze(SRW_LOCK_CALL);
 
 	page_zip_stat_per_index_t::iterator	iter;
 	ulint					i;
@@ -1393,18 +1393,18 @@ i_s_cmp_per_index_fill_low(
 			status = 1;
 			break;
 		}
-		/* Release and reacquire the dict mutex to allow other
+		/* Release and reacquire the dict_sys.latch to allow other
 		threads to proceed. This could eventually result in the
 		contents of INFORMATION_SCHEMA.innodb_cmp_per_index being
 		inconsistent, but it is an acceptable compromise. */
 		if (i == 1000) {
-			dict_sys.mutex_unlock();
+			dict_sys.unfreeze();
 			i = 0;
-			dict_sys.mutex_lock();
+			dict_sys.freeze(SRW_LOCK_CALL);
 		}
 	}
 
-	dict_sys.mutex_unlock();
+	dict_sys.unfreeze();
 
 	if (reset) {
 		page_zip_reset_stat_per_index();
@@ -3973,7 +3973,7 @@ i_s_innodb_buffer_page_fill(
 		if (page_info->page_type == I_S_PAGE_TYPE_INDEX) {
 			bool ret = false;
 
-			dict_sys.mutex_lock();
+			dict_sys.freeze(SRW_LOCK_CALL);
 
 			const dict_index_t* index =
 				dict_index_get_if_in_cache_low(
@@ -3998,7 +3998,7 @@ i_s_innodb_buffer_page_fill(
 						system_charset_info);
 			}
 
-			dict_sys.mutex_unlock();
+			dict_sys.unfreeze();
 
 			OK(ret);
 
@@ -4476,7 +4476,7 @@ i_s_innodb_buf_page_lru_fill(
 		if (page_info->page_type == I_S_PAGE_TYPE_INDEX) {
 			bool ret = false;
 
-			dict_sys.mutex_lock();
+			dict_sys.freeze(SRW_LOCK_CALL);
 
 			const dict_index_t* index =
 				dict_index_get_if_in_cache_low(
@@ -4501,7 +4501,7 @@ i_s_innodb_buf_page_lru_fill(
 						system_charset_info);
 			}
 
-			dict_sys.mutex_unlock();
+			dict_sys.unfreeze();
 
 			OK(ret);
 
@@ -4854,8 +4854,8 @@ i_s_sys_tables_fill_table(
 		DBUG_RETURN(0);
 	}
 
-	dict_sys.mutex_lock();
-	mtr_start(&mtr);
+	mtr.start();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	for (const rec_t *rec = dict_startscan_system(&pcur, &mtr,
 						      dict_sys.sys_tables);
@@ -4871,7 +4871,7 @@ i_s_sys_tables_fill_table(
 		information from SYS_TABLES row */
 		err_msg = i_s_sys_tables_rec(pcur, rec, &table_rec);
 		mtr.commit();
-		dict_sys.mutex_unlock();
+		dict_sys.unlock();
 
 		if (!err_msg) {
 			i_s_dict_fill_sys_tables(thd, table_rec,
@@ -4887,12 +4887,12 @@ i_s_sys_tables_fill_table(
 		}
 
 		/* Get the next record */
-		dict_sys.mutex_lock();
 		mtr.start();
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
 	mtr.commit();
-	dict_sys.mutex_unlock();
+	dict_sys.unlock();
 
 	DBUG_RETURN(0);
 }
@@ -5129,7 +5129,6 @@ i_s_sys_tables_fill_table_stats(
 		/* Get the next record */
 		mtr.start();
 		dict_sys.lock(SRW_LOCK_CALL);
-
 		if (table_rec) {
 			dict_sys.allow_eviction(table_rec);
 		}
@@ -5331,7 +5330,7 @@ i_s_sys_indexes_fill_table(
 	}
 
 	heap = mem_heap_create(1000);
-	dict_sys.mutex_lock();
+	dict_sys.lock(SRW_LOCK_CALL);
 	mtr_start(&mtr);
 
 	/* Start scan the SYS_INDEXES table */
@@ -5352,8 +5351,8 @@ i_s_sys_indexes_fill_table(
 			rec, DICT_FLD__SYS_INDEXES__SPACE, &space_id);
 		space_id = space_id == 4 ? mach_read_from_4(field)
 			: ULINT_UNDEFINED;
-		mtr_commit(&mtr);
-		dict_sys.mutex_unlock();
+		mtr.commit();
+		dict_sys.unlock();
 
 		if (!err_msg) {
 			if (int err = i_s_dict_fill_sys_indexes(
@@ -5371,13 +5370,13 @@ i_s_sys_indexes_fill_table(
 		mem_heap_empty(heap);
 
 		/* Get the next record */
-		dict_sys.mutex_lock();
-		mtr_start(&mtr);
+		mtr.start();
+		dict_sys.lock(SRW_LOCK_CALL);
 		rec = dict_getnext_system(&pcur, &mtr);
 	}
 
-	mtr_commit(&mtr);
-	dict_sys.mutex_unlock();
+	mtr.commit();
+	dict_sys.unlock();
 	mem_heap_free(heap);
 
 	DBUG_RETURN(0);
@@ -5550,8 +5549,8 @@ i_s_sys_columns_fill_table(
 	}
 
 	heap = mem_heap_create(1000);
-	dict_sys.mutex_lock();
-	mtr_start(&mtr);
+	mtr.start();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	rec = dict_startscan_system(&pcur, &mtr, dict_sys.sys_columns);
 
@@ -5567,8 +5566,8 @@ i_s_sys_columns_fill_table(
 						       &table_id, &col_name,
 						       &nth_v_col);
 
-		mtr_commit(&mtr);
-		dict_sys.mutex_unlock();
+		mtr.commit();
+		dict_sys.unlock();
 
 		if (!err_msg) {
 			i_s_dict_fill_sys_columns(thd, table_id, col_name,
@@ -5583,17 +5582,18 @@ i_s_sys_columns_fill_table(
 		mem_heap_empty(heap);
 
 		/* Get the next record */
-		dict_sys.mutex_lock();
-		mtr_start(&mtr);
+		mtr.start();
+		dict_sys.lock(SRW_LOCK_CALL);
 		rec = dict_getnext_system(&pcur, &mtr);
 	}
 
-	mtr_commit(&mtr);
-	dict_sys.mutex_unlock();
+	mtr.commit();
+	dict_sys.unlock();
 	mem_heap_free(heap);
 
 	DBUG_RETURN(0);
 }
+
 /*******************************************************************//**
 Bind the dynamic table INFORMATION_SCHEMA.innodb_sys_columns
 @return 0 on success */
@@ -5743,8 +5743,8 @@ i_s_sys_virtual_fill_table(
 		DBUG_RETURN(0);
 	}
 
-	dict_sys.mutex_lock();
-	mtr_start(&mtr);
+	mtr.start();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	rec = dict_startscan_system(&pcur, &mtr, dict_sys.sys_virtual);
 
@@ -5758,8 +5758,8 @@ i_s_sys_virtual_fill_table(
 						       &table_id, &pos,
 						       &base_pos);
 
-		mtr_commit(&mtr);
-		dict_sys.mutex_unlock();
+		mtr.commit();
+		dict_sys.unlock();
 
 		if (!err_msg) {
 			i_s_dict_fill_sys_virtual(thd, table_id, pos, base_pos,
@@ -5771,13 +5771,13 @@ i_s_sys_virtual_fill_table(
 		}
 
 		/* Get the next record */
-		dict_sys.mutex_lock();
-		mtr_start(&mtr);
+		mtr.start();
+		dict_sys.lock(SRW_LOCK_CALL);
 		rec = dict_getnext_system(&pcur, &mtr);
 	}
 
-	mtr_commit(&mtr);
-	dict_sys.mutex_unlock();
+	mtr.commit();
+	dict_sys.unlock();
 
 	DBUG_RETURN(0);
 }
@@ -5929,13 +5929,13 @@ i_s_sys_fields_fill_table(
 	}
 
 	heap = mem_heap_create(1000);
-	dict_sys.mutex_lock();
-	mtr_start(&mtr);
+	mtr.start();
 
 	/* will save last index id so that we know whether we move to
 	the next index. This is used to calculate prefix length */
 	last_id = 0;
 
+	dict_sys.lock(SRW_LOCK_CALL);
 	rec = dict_startscan_system(&pcur, &mtr, dict_sys.sys_fields);
 
 	while (rec) {
@@ -5949,8 +5949,8 @@ i_s_sys_fields_fill_table(
 		err_msg = dict_process_sys_fields_rec(heap, rec, &field_rec,
 						      &pos, &index_id, last_id);
 
-		mtr_commit(&mtr);
-		dict_sys.mutex_unlock();
+		mtr.commit();
+		dict_sys.unlock();
 
 		if (!err_msg) {
 			i_s_dict_fill_sys_fields(thd, index_id, &field_rec,
@@ -5965,13 +5965,13 @@ i_s_sys_fields_fill_table(
 		mem_heap_empty(heap);
 
 		/* Get the next record */
-		dict_sys.mutex_lock();
-		mtr_start(&mtr);
+		mtr.start();
+		dict_sys.lock(SRW_LOCK_CALL);
 		rec = dict_getnext_system(&pcur, &mtr);
 	}
 
-	mtr_commit(&mtr);
-	dict_sys.mutex_unlock();
+	mtr.commit();
+	dict_sys.unlock();
 	mem_heap_free(heap);
 
 	DBUG_RETURN(0);
@@ -6132,8 +6132,8 @@ i_s_sys_foreign_fill_table(
 	}
 
 	heap = mem_heap_create(1000);
-	dict_sys.mutex_lock();
-	mtr_start(&mtr);
+	mtr.start();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	rec = dict_startscan_system(&pcur, &mtr, dict_sys.sys_foreign);
 
@@ -6145,8 +6145,8 @@ i_s_sys_foreign_fill_table(
 		a SYS_FOREIGN row */
 		err_msg = dict_process_sys_foreign_rec(heap, rec, &foreign_rec);
 
-		mtr_commit(&mtr);
-		dict_sys.mutex_unlock();
+		mtr.commit();
+		dict_sys.unlock();
 
 		if (!err_msg) {
 			i_s_dict_fill_sys_foreign(thd, &foreign_rec,
@@ -6160,13 +6160,13 @@ i_s_sys_foreign_fill_table(
 		mem_heap_empty(heap);
 
 		/* Get the next record */
-		mtr_start(&mtr);
-		dict_sys.mutex_lock();
+		mtr.start();
+		dict_sys.lock(SRW_LOCK_CALL);
 		rec = dict_getnext_system(&pcur, &mtr);
 	}
 
-	mtr_commit(&mtr);
-	dict_sys.mutex_unlock();
+	mtr.commit();
+	dict_sys.unlock();
 	mem_heap_free(heap);
 
 	DBUG_RETURN(0);
@@ -6325,8 +6325,8 @@ i_s_sys_foreign_cols_fill_table(
 	}
 
 	heap = mem_heap_create(1000);
-	dict_sys.mutex_lock();
-	mtr_start(&mtr);
+	mtr.start();
+	dict_sys.lock(SRW_LOCK_CALL);
 
 	rec = dict_startscan_system(&pcur, &mtr, dict_sys.sys_foreign_cols);
 
@@ -6341,8 +6341,8 @@ i_s_sys_foreign_cols_fill_table(
 		err_msg = dict_process_sys_foreign_col_rec(
 			heap, rec, &name, &for_col_name, &ref_col_name, &pos);
 
-		mtr_commit(&mtr);
-		dict_sys.mutex_unlock();
+		mtr.commit();
+		dict_sys.unlock();
 
 		if (!err_msg) {
 			i_s_dict_fill_sys_foreign_cols(
@@ -6357,13 +6357,13 @@ i_s_sys_foreign_cols_fill_table(
 		mem_heap_empty(heap);
 
 		/* Get the next record */
-		dict_sys.mutex_lock();
-		mtr_start(&mtr);
+		mtr.start();
+		dict_sys.lock(SRW_LOCK_CALL);
 		rec = dict_getnext_system(&pcur, &mtr);
 	}
 
-	mtr_commit(&mtr);
-	dict_sys.mutex_unlock();
+	mtr.commit();
+	dict_sys.unlock();
 	mem_heap_free(heap);
 
 	DBUG_RETURN(0);
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index 6f5d2377b07..1842650f1f6 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -934,8 +934,8 @@ dict_table_copy_types(
 	const dict_table_t*	table)	/*!< in: table */
 	MY_ATTRIBUTE((nonnull));
 /**********************************************************************//**
-Looks for an index with the given id. NOTE that we do not reserve
-the dictionary mutex: this function is for emergency purposes like
+Looks for an index with the given id. NOTE that we do not acquire
+dict_sys.latch: this function is for emergency purposes like
 printing info of a corrupt database page!
 @return index or NULL if not found from cache */
 dict_index_t*
@@ -1146,7 +1146,6 @@ dict_field_get_col(
 
 /**********************************************************************//**
 Returns an index object if it is found in the dictionary cache.
-Assumes that dict_sys.mutex is already being held.
 @return index, NULL if not found */
 dict_index_t*
 dict_index_get_if_in_cache_low(
@@ -1352,13 +1351,12 @@ extern mysql_mutex_t dict_foreign_err_mutex;
 /** InnoDB data dictionary cache */
 class dict_sys_t
 {
-  /** The my_hrtime_coarse().val of the oldest mutex_lock_wait() start, or 0 */
-  std::atomic<ulonglong> mutex_wait_start;
+  /** The my_hrtime_coarse().val of the oldest lock_wait() start, or 0 */
+  std::atomic<ulonglong> latch_ex_wait_start;
 
   /** @brief the data dictionary rw-latch protecting dict_sys
 
-  Table create, drop, etc. reserve this in X-mode (along with
-  acquiring dict_sys.mutex); implicit or
+  Table create, drop, etc. reserve this in X-mode; implicit or
   backround operations that are not fully covered by MDL
   (rollback, foreign key checks) reserve this in S-mode.
 
@@ -1368,11 +1366,9 @@ class dict_sys_t
 #ifdef UNIV_DEBUG
   /** whether latch is being held in exclusive mode (by any thread) */
   bool latch_ex;
+  /** number of S-latch holders */
+  Atomic_counter<uint32_t> latch_readers;
 #endif
-  /** Mutex protecting dict_sys. Whenever latch is acquired
-  exclusively, also the mutex will be acquired.
-  FIXME: merge the mutex and the latch, once MDEV-23484 has been fixed */
-  mysql_mutex_t mutex;
 public:
   /** Indexes of SYS_TABLE[] */
   enum
@@ -1429,7 +1425,7 @@ private:
   /** The synchronization interval of row_id */
   static constexpr size_t ROW_ID_WRITE_MARGIN= 256;
 public:
-  /** Diagnostic message for exceeding the mutex_lock_wait() timeout */
+  /** Diagnostic message for exceeding the lock_wait() timeout */
   static const char fatal_msg[];
 
   /** @return A new value for GEN_CLUST_INDEX(DB_ROW_ID) */
@@ -1457,7 +1453,7 @@ public:
   (should only happen during the rollback of CREATE...SELECT) */
   dict_table_t *acquire_temporary_table(table_id_t id)
   {
-    mysql_mutex_assert_owner(&mutex);
+    ut_ad(frozen());
     dict_table_t *table;
     ulint fold = ut_fold_ull(id);
     HASH_SEARCH(id_hash, &temp_id_hash, fold, dict_table_t*, table,
@@ -1477,7 +1473,7 @@ public:
   @retval nullptr if not cached */
   dict_table_t *find_table(table_id_t id)
   {
-    mysql_mutex_assert_owner(&mutex);
+    ut_ad(frozen());
     dict_table_t *table;
     ulint fold= ut_fold_ull(id);
     HASH_SEARCH(id_hash, &table_id_hash, fold, dict_table_t*, table,
@@ -1511,7 +1507,7 @@ public:
   {
     ut_ad(table);
     ut_ad(table->can_be_evicted == in_lru);
-    mysql_mutex_assert_owner(&mutex);
+    ut_ad(frozen());
     for (const dict_table_t* t= in_lru ? table_LRU.start : table_non_LRU.start;
          t; t = UT_LIST_GET_NEXT(table_LRU, t))
     {
@@ -1531,6 +1527,7 @@ public:
   @return whether the table was evictable */
   bool prevent_eviction(dict_table_t *table)
   {
+    ut_d(locked());
     ut_ad(find(table));
     if (!table->can_be_evicted)
       return false;
@@ -1542,56 +1539,78 @@ public:
   /** Move a table from the non-LRU list to the LRU list. */
   void allow_eviction(dict_table_t *table)
   {
+    ut_d(locked());
     ut_ad(find(table));
     ut_ad(!table->can_be_evicted);
     table->can_be_evicted= true;
     UT_LIST_REMOVE(table_non_LRU, table);
-    UT_LIST_ADD_LAST(table_LRU, table);
+    UT_LIST_ADD_FIRST(table_LRU, table);
   }
 
-  /** Acquire a reference to a cached table. */
-  inline void acquire(dict_table_t *table);
-
-  /** Assert that the mutex is locked */
-  void assert_locked() const { mysql_mutex_assert_owner(&mutex); }
-  /** Assert that the mutex is not locked */
-  void assert_not_locked() const { mysql_mutex_assert_not_owner(&mutex); }
-#ifdef SAFE_MUTEX
-  bool mutex_is_locked() const { return mysql_mutex_is_owner(&mutex); }
+#ifdef UNIV_DEBUG
+  /** @return whether any thread (not necessarily the current thread)
+  is holding the latch; that is, this check may return false
+  positives */
+  bool frozen() const { return latch_readers || locked(); }
+  /** @return whether any thread (not necessarily the current thread)
+  is holding the exclusive latch; that is, this check may return false
+  positives */
+  bool locked() const { return latch_ex; }
 #endif
 private:
-  /** Acquire the mutex */
-  ATTRIBUTE_NOINLINE void mutex_lock_wait();
+  /** Acquire the exclusive latch */
+  ATTRIBUTE_NOINLINE
+  void lock_wait(SRW_LOCK_ARGS(const char *file, unsigned line));
 public:
-  /** @return the my_hrtime_coarse().val of the oldest mutex_lock_wait() start,
+  /** @return the my_hrtime_coarse().val of the oldest lock_wait() start,
   assuming that requests are served on a FIFO basis */
   ulonglong oldest_wait() const
-  { return mutex_wait_start.load(std::memory_order_relaxed); }
+  { return latch_ex_wait_start.load(std::memory_order_relaxed); }
 
-#ifdef HAVE_PSI_MUTEX_INTERFACE
-  /** Acquire the mutex */
-  ATTRIBUTE_NOINLINE void mutex_lock();
-  /** Release the mutex */
-  ATTRIBUTE_NOINLINE void mutex_unlock();
-#else
-  /** Acquire the mutex */
-  void mutex_lock() { if (mysql_mutex_trylock(&mutex)) mutex_lock_wait(); }
-
-  /** Release the mutex */
-  void mutex_unlock() { mysql_mutex_unlock(&mutex); }
-#endif
-
-  /** Lock the data dictionary cache. */
-  void lock(SRW_LOCK_ARGS(const char *file, unsigned line));
+  /** Exclusively lock the dictionary cache. */
+  void lock(SRW_LOCK_ARGS(const char *file, unsigned line))
+  {
+    if (latch.wr_lock_try())
+    {
+      ut_ad(!latch_readers);
+      ut_ad(!latch_ex);
+      ut_d(latch_ex= true);
+    }
+    else
+      lock_wait(SRW_LOCK_ARGS(file, line));
+   }
 
+#ifdef UNIV_PFS_RWLOCK
+  /** Unlock the data dictionary cache. */
+  ATTRIBUTE_NOINLINE void unlock();
+  /** Acquire a shared lock on the dictionary cache. */
+  ATTRIBUTE_NOINLINE void freeze(const char *file, unsigned line);
+  /** Release a shared lock on the dictionary cache. */
+  ATTRIBUTE_NOINLINE void unfreeze();
+#else
   /** Unlock the data dictionary cache. */
   void unlock()
   {
     ut_ad(latch_ex);
+    ut_ad(!latch_readers);
     ut_d(latch_ex= false);
-    mutex_unlock();
     latch.wr_unlock();
   }
+  /** Acquire a shared lock on the dictionary cache. */
+  void freeze()
+  {
+    latch.rd_lock();
+    ut_ad(!latch_ex);
+    ut_d(latch_readers++);
+  }
+  /** Release a shared lock on the dictionary cache. */
+  void unfreeze()
+  {
+    ut_ad(!latch_ex);
+    ut_ad(latch_readers--);
+    latch.rd_unlock();
+  }
+#endif
 
   /** Estimate the used memory occupied by the data dictionary
   table and index objects.
@@ -1621,7 +1640,7 @@ public:
   @retval nullptr if not found */
   dict_table_t *find_table(const span<const char> &name) const
   {
-    assert_locked();
+    ut_ad(frozen());
     for (dict_table_t *table= static_cast<dict_table_t*>
          (HASH_GET_FIRST(&table_hash, table_hash.calc_hash
                          (my_crc32c(0, name.data(), name.size()))));
@@ -1650,9 +1669,6 @@ public:
 /** the data dictionary cache */
 extern dict_sys_t	dict_sys;
 
-#define dict_sys_lock() dict_sys.lock(SRW_LOCK_CALL)
-#define dict_sys_unlock() dict_sys.unlock()
-
 /*********************************************************************//**
 Converts a database and table name from filesystem encoding
 (e.g. d@i1b/a@q1b@1Kc, same format as used in dict_table_t::name) in two
diff --git a/storage/innobase/include/dict0dict.ic b/storage/innobase/include/dict0dict.ic
index a494eefdc32..a98ca38d9b3 100644
--- a/storage/innobase/include/dict0dict.ic
+++ b/storage/innobase/include/dict0dict.ic
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2020, MariaDB Corporation.
+Copyright (c) 2013, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -1141,7 +1141,7 @@ dict_table_is_file_per_table(
 /** Acquire the table handle. */
 inline void dict_table_t::acquire()
 {
-  dict_sys.assert_locked();
+  ut_ad(dict_sys.frozen());
   n_ref_count++;
 }
 
diff --git a/storage/innobase/include/dict0load.h b/storage/innobase/include/dict0load.h
index 591dd30f211..072773694a9 100644
--- a/storage/innobase/include/dict0load.h
+++ b/storage/innobase/include/dict0load.h
@@ -53,12 +53,9 @@ We also scan the biggest space id, and store it to fil_system. */
 void dict_check_tablespaces_and_store_max_id();
 
 /** Make sure the data_file_name is saved in dict_table_t if needed.
-@param[in]	table		Table object
-@param[in]	dict_mutex_own	true if dict_sys.mutex is owned already */
-void
-dict_get_and_save_data_dir_path(
-	dict_table_t*	table,
-	bool		dict_mutex_own);
+@param[in,out]	table		Table object
+@param[in]	dict_locked	dict_sys.frozen() */
+void dict_get_and_save_data_dir_path(dict_table_t* table, bool dict_locked);
 
 /***********************************************************************//**
 Loads a table object based on the table id.
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 82bb2b29b65..45747c25565 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -1026,8 +1026,7 @@ struct dict_index_t {
 				/*!< enum online_index_status.
 				Transitions from ONLINE_INDEX_COMPLETE (to
 				ONLINE_INDEX_CREATION) are protected
-				by dict_sys.latch and
-				dict_sys.mutex. Other changes are
+				by dict_sys.latch. Other changes are
 				protected by index->lock. */
 	unsigned	uncommitted:1;
 				/*!< a flag that is set for secondary indexes
@@ -1935,7 +1934,7 @@ struct dict_table_t {
 	inline size_t get_overflow_field_local_len() const;
 
 	/** Parse the table file name into table name and database name.
-	@tparam		dict_locked	whether dict_sys.mutex is being held
+	@tparam		dict_locked	whether dict_sys.lock() was called
 	@param[in,out]	db_name		database name buffer
 	@param[in,out]	tbl_name	table name buffer
 	@param[out]	db_name_len	database name length
@@ -2178,7 +2177,7 @@ public:
 	dict_foreign_set			referenced_set;
 
 	/** Statistics for query optimization. Mostly protected by
-	dict_sys.mutex. @{ */
+	dict_sys.latch and stats_mutex_lock(). @{ */
 
 	/** TRUE if statistics have been calculated the first time after
 	database startup or table creation. */
@@ -2260,7 +2259,7 @@ public:
 
 	/** The state of the background stats thread wrt this table.
 	See BG_STAT_NONE, BG_STAT_IN_PROGRESS and BG_STAT_SHOULD_QUIT.
-	Writes are covered by dict_sys.mutex. Dirty reads are possible. */
+	Writes are covered by dict_sys.latch and stats_mutex_lock(). */
 	byte					stats_bg_flag;
 
 	bool		stats_error_printed;
diff --git a/storage/innobase/include/dict0stats.ic b/storage/innobase/include/dict0stats.ic
index 68592404253..e49153eb099 100644
--- a/storage/innobase/include/dict0stats.ic
+++ b/storage/innobase/include/dict0stats.ic
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2012, 2015, Oracle and/or its affiliates. All rights reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -75,7 +75,7 @@ dict_stats_is_persistent_enabled(const dict_table_t* table)
 	+ dict_stats_update(DICT_STATS_RECALC_TRANSIENT) on a table that has
 	  just been PS-enabled.
 	This is acceptable. Avoiding this would mean that we would have to
-	protect the stat_persistent with dict_sys.mutex like the
+	hold dict_sys.latch or stats_mutex_lock() like for accessing the
 	other ::stat_ members which would be too big performance penalty,
 	especially when this function is called from
 	dict_stats_update_if_needed(). */
diff --git a/storage/innobase/include/dict0stats_bg.h b/storage/innobase/include/dict0stats_bg.h
index e9318a0116d..0d5982b231a 100644
--- a/storage/innobase/include/dict0stats_bg.h
+++ b/storage/innobase/include/dict0stats_bg.h
@@ -50,10 +50,10 @@ dict_stats_recalc_pool_del(
 /** Yield the data dictionary latch when waiting
 for the background thread to stop accessing a table.
 @param trx	transaction holding the data dictionary locks */
-#define DICT_BG_YIELD(trx)	do {	\
-	row_mysql_unlock_data_dictionary(trx);	\
+#define DICT_BG_YIELD	do {	\
+	dict_sys.unlock();	\
 	std::this_thread::sleep_for(std::chrono::milliseconds(250));	\
-	row_mysql_lock_data_dictionary(trx);	\
+	dict_sys.lock(SRW_LOCK_CALL); \
 } while (0)
 
 /*****************************************************************//**
@@ -67,31 +67,27 @@ dict_stats_stop_bg(
 	dict_table_t*	table)	/*!< in/out: table */
 {
 	ut_ad(!srv_read_only_mode);
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	if (!(table->stats_bg_flag & BG_STAT_IN_PROGRESS)) {
 		return(true);
 	}
 
+	/* In dict_stats_update_persistent() this flag is being read
+	while holding the mutex, not dict_sys.latch. */
+	table->stats_mutex_lock();
 	table->stats_bg_flag |= BG_STAT_SHOULD_QUIT;
+	table->stats_mutex_unlock();
 	return(false);
 }
 
 /*****************************************************************//**
 Wait until background stats thread has stopped using the specified table.
-The caller must have locked the data dictionary using
-row_mysql_lock_data_dictionary() and this function may unlock it temporarily
-and restore the lock before it exits.
 The background stats thread is guaranteed not to start using the specified
-table after this function returns and before the caller unlocks the data
-dictionary because it sets the BG_STAT_IN_PROGRESS bit in table->stats_bg_flag
-under dict_sys.mutex. */
-void
-dict_stats_wait_bg_to_stop_using_table(
-/*===================================*/
-	dict_table_t*	table,	/*!< in/out: table */
-	trx_t*		trx);	/*!< in/out: transaction to use for
-				unlocking/locking the data dict */
+table after this function returns and before the caller releases
+dict_sys.latch. */
+void dict_stats_wait_bg_to_stop_using_table(dict_table_t *table);
+
 /*****************************************************************//**
 Initialize global variables needed for the operation of dict_stats_thread().
 Must be called before dict_stats task is started. */
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 9fd0b076520..445a8085320 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -1660,7 +1660,7 @@ right in it. If does not succeed, prints an error message to the .err log. This
 function is used to open a tablespace when we start up mysqld, and also in
 IMPORT TABLESPACE.
 NOTE that we assume this operation is used either at the database startup
-or under the protection of the dictionary mutex, so that two users cannot
+or under the protection of dict_sys.latch, so that two users cannot
 race here. This operation does not leave the file associated with the
 tablespace open, but closes it after we have looked at the space id in it.
 
diff --git a/storage/innobase/include/fts0fts.h b/storage/innobase/include/fts0fts.h
index 95b5131a526..0fb01dd850e 100644
--- a/storage/innobase/include/fts0fts.h
+++ b/storage/innobase/include/fts0fts.h
@@ -318,7 +318,7 @@ public:
 
 	/** Whether the ADDED table record sync-ed after crash recovery */
 	unsigned	added_synced:1;
-	/** Whether the table holds dict_sys.mutex */
+	/** Whether the table holds dict_sys.latch */
 	unsigned	dict_locked:1;
 
 	/** Work queue for scheduling jobs for the FTS 'Add' thread, or NULL
@@ -741,7 +741,7 @@ FTS auxiliary INDEX table and clear the cache at the end.
 dberr_t fts_sync_table(dict_table_t* table, bool wait = true);
 
 /****************************************************************//**
-Free the query graph but check whether dict_sys.mutex is already
+Free the query graph but check whether dict_sys.latch is already
 held */
 void
 fts_que_graph_free_check_lock(
diff --git a/storage/innobase/include/fts0priv.h b/storage/innobase/include/fts0priv.h
index fbe2acdc712..a8647b4dd1d 100644
--- a/storage/innobase/include/fts0priv.h
+++ b/storage/innobase/include/fts0priv.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2011, 2018, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -135,7 +135,7 @@ fts_eval_sql(
 /** Construct the name of an internal FTS table for the given table.
 @param[in]	fts_table	metadata on fulltext-indexed table
 @param[out]	table_name	a name up to MAX_FULL_NAME_LEN
-@param[in]	dict_locked	whether dict_sys.mutex is being held */
+@param[in]	dict_locked	whether dict_sys.latch is being held */
 void fts_get_table_name(const fts_table_t* fts_table, char* table_name,
 			bool dict_locked = false)
 	MY_ATTRIBUTE((nonnull));
@@ -295,16 +295,6 @@ fts_trx_table_id_cmp(
 #define fts_sql_commit(trx) trx_commit_for_mysql(trx)
 #define fts_sql_rollback(trx) (trx)->rollback()
 /******************************************************************//**
-Parse an SQL string. %s is replaced with the table's id. Don't acquire
-the dict mutex
-@return query graph */
-que_t*
-fts_parse_sql_no_dict_lock(
-/*=======================*/
-	pars_info_t*	info,		/*!< in: parser info */
-	const char*	sql)		/*!< in: SQL string to evaluate */
-	MY_ATTRIBUTE((nonnull(2), malloc, warn_unused_result));
-/******************************************************************//**
 Get value from config table. The caller must ensure that enough
 space is allocated for value to hold the column contents
 @return DB_SUCCESS or error code */
@@ -469,12 +459,6 @@ fts_get_table_id(
 					FTS_AUX_MIN_TABLE_ID_LENGTH bytes
 					long */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
-/** Construct the name of an internal FTS table for the given table.
-@param[in]	fts_table	metadata on fulltext-indexed table
-@param[in]	dict_locked	whether dict_sys.mutex is being held
-@return	the prefix, must be freed with ut_free() */
-char* fts_get_table_name_prefix(const fts_table_t* fts_table)
-	MY_ATTRIBUTE((nonnull, malloc, warn_unused_result));
 /******************************************************************//**
 Add node positions. */
 void
diff --git a/storage/innobase/include/log0log.h b/storage/innobase/include/log0log.h
index ca792f9e439..a651a56e8a0 100644
--- a/storage/innobase/include/log0log.h
+++ b/storage/innobase/include/log0log.h
@@ -77,7 +77,7 @@ log_reserve_and_write_fast(
 Checks if there is need for a log buffer flush or a new checkpoint, and does
 this if yes. Any database operation should call this when it has modified
 more than about 4 pages. NOTE that this function may only be called when the
-OS thread owns no synchronization objects except the dictionary mutex. */
+OS thread owns no synchronization objects except dict_sys.latch. */
 UNIV_INLINE
 void
 log_free_check(void);
diff --git a/storage/innobase/include/log0log.ic b/storage/innobase/include/log0log.ic
index aebe511d659..c29c0bfa55f 100644
--- a/storage/innobase/include/log0log.ic
+++ b/storage/innobase/include/log0log.ic
@@ -294,7 +294,7 @@ log_reserve_and_write_fast(
 Checks if there is need for a log buffer flush or a new checkpoint, and does
 this if yes. Any database operation should call this when it has modified
 more than about 4 pages. NOTE that this function may only be called when the
-OS thread owns no synchronization objects except the dictionary mutex. */
+OS thread owns no synchronization objects except dict_sys.latch. */
 UNIV_INLINE
 void
 log_free_check(void)
diff --git a/storage/innobase/include/que0que.h b/storage/innobase/include/que0que.h
index 1a30ea45194..d81c77e3d36 100644
--- a/storage/innobase/include/que0que.h
+++ b/storage/innobase/include/que0que.h
@@ -207,9 +207,9 @@ que_eval_sql(
 /*=========*/
 	pars_info_t*	info,	/*!< in: info struct, or NULL */
 	const char*	sql,	/*!< in: SQL string */
-	bool		reserve_dict_mutex,
+	bool		lock_dict,
 				/*!< in: whether to acquire/release
-				dict_sys.mutex around call to pars_sql. */
+				dict_sys.latch around call to pars_sql(). */
 	trx_t*		trx);	/*!< in: trx */
 
 /**********************************************************************//**
diff --git a/storage/innobase/include/univ.i b/storage/innobase/include/univ.i
index eb32e91298d..88f0b2d185b 100644
--- a/storage/innobase/include/univ.i
+++ b/storage/innobase/include/univ.i
@@ -537,7 +537,6 @@ typedef unsigned int mysql_pfs_key_t;
 # ifdef UNIV_PFS_MUTEX
 extern mysql_pfs_key_t buf_pool_mutex_key;
 extern mysql_pfs_key_t dict_foreign_err_mutex_key;
-extern mysql_pfs_key_t dict_sys_mutex_key;
 extern mysql_pfs_key_t fil_system_mutex_key;
 extern mysql_pfs_key_t flush_list_mutex_key;
 extern mysql_pfs_key_t fts_cache_mutex_key;
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index dccad0f3442..f3f5ce42372 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -3846,7 +3846,7 @@ released:
 #if defined SAFE_MUTEX && defined UNIV_DEBUG
   if (to_evict.empty())
     return;
-  dict_sys.mutex_lock();
+  dict_sys.lock(SRW_LOCK_CALL);
   LockMutexGuard g{SRW_LOCK_CALL};
   for (const table_id_t id : to_evict)
   {
@@ -3854,7 +3854,7 @@ released:
       if (!table->get_ref_count() && !UT_LIST_GET_LEN(table->locks))
         dict_sys.remove(table, true);
   }
-  dict_sys.mutex_unlock();
+  dict_sys.unlock();
 #endif
 }
 
diff --git a/storage/innobase/pars/pars0pars.cc b/storage/innobase/pars/pars0pars.cc
index 74c33b0cd9c..d49e5668aa1 100644
--- a/storage/innobase/pars/pars0pars.cc
+++ b/storage/innobase/pars/pars0pars.cc
@@ -1973,7 +1973,7 @@ pars_sql(
 	heap = mem_heap_create(16000);
 
 	/* Currently, the parser is not reentrant: */
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	pars_sym_tab_global = sym_tab_create(heap);
 
diff --git a/storage/innobase/pars/pars0sym.cc b/storage/innobase/pars/pars0sym.cc
index b65021a375a..d68e352e795 100644
--- a/storage/innobase/pars/pars0sym.cc
+++ b/storage/innobase/pars/pars0sym.cc
@@ -67,7 +67,7 @@ sym_tab_free_private(
 	sym_node_t*	sym;
 	func_node_t*	func;
 
-	dict_sys.assert_locked();
+	ut_ad(dict_sys.locked());
 
 	for (sym = UT_LIST_GET_FIRST(sym_tab->sym_list);
 	     sym != NULL;
diff --git a/storage/innobase/que/que0que.cc b/storage/innobase/que/que0que.cc
index d97466a9219..074f8c54e24 100644
--- a/storage/innobase/que/que0que.cc
+++ b/storage/innobase/que/que0que.cc
@@ -748,9 +748,9 @@ que_eval_sql(
 /*=========*/
 	pars_info_t*	info,	/*!< in: info struct, or NULL */
 	const char*	sql,	/*!< in: SQL string */
-	bool		reserve_dict_mutex,
+	bool		lock_dict,
 				/*!< in: whether to acquire/release
-				dict_sys.mutex around call to pars_sql. */
+				dict_sys.latch around call to pars_sql(). */
 	trx_t*		trx)	/*!< in: trx */
 {
 	que_thr_t*	thr;
@@ -761,14 +761,14 @@ que_eval_sql(
 
 	ut_a(trx->error_state == DB_SUCCESS);
 
-	if (reserve_dict_mutex) {
-		dict_sys.mutex_lock();
+	if (lock_dict) {
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
 	graph = pars_sql(info, sql);
 
-	if (reserve_dict_mutex) {
-		dict_sys.mutex_unlock();
+	if (lock_dict) {
+		dict_sys.unlock();
 	}
 
 	graph->trx = trx;
@@ -778,14 +778,14 @@ que_eval_sql(
 
 	que_run_threads(thr);
 
-	if (reserve_dict_mutex) {
-		dict_sys.mutex_lock();
+	if (lock_dict) {
+		dict_sys.lock(SRW_LOCK_CALL);
 	}
 
 	que_graph_free(graph);
 
-	if (reserve_dict_mutex) {
-		dict_sys.mutex_unlock();
+	if (lock_dict) {
+		dict_sys.unlock();
 	}
 
 	DBUG_RETURN(trx->error_state);
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 710e1910d1b..0d380405e5d 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -1446,8 +1446,6 @@ row_import::set_root_by_heuristic() UNIV_NOTHROW
 			" the tablespace has " << m_n_indexes << " indexes";
 	}
 
-	dict_sys.mutex_lock();
-
 	ulint	i = 0;
 	dberr_t	err = DB_SUCCESS;
 
@@ -1487,8 +1485,6 @@ row_import::set_root_by_heuristic() UNIV_NOTHROW
 		}
 	}
 
-	dict_sys.mutex_unlock();
-
 	return(err);
 }
 
@@ -4211,8 +4207,6 @@ row_import_for_mysql(
 	}
 
 	/* Open the tablespace so that we can access via the buffer pool.
-	We set the 2nd param (fix_dict = true) here because we already
-	have locked dict_sys.latch and dict_sys.mutex.
 	The tablespace is initially opened as a temporary one, because
 	we will not be writing any redo log for it before we have invoked
 	fil_space_t::set_imported() to declare it a persistent tablespace. */
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 3d1f07093fd..eb5a3cc91b8 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -3040,9 +3040,7 @@ row_ins_sec_index_entry_low(
 			if (!index->is_committed()) {
 				ut_ad(!thr_get_trx(thr)
 				      ->dict_operation_lock_mode);
-				dict_sys.mutex_lock();
-				dict_set_corrupted_index_cache_only(index);
-				dict_sys.mutex_unlock();
+				index->type |= DICT_CORRUPT;
 				/* Do not return any error to the
 				caller. The duplicate will be reported
 				by ALTER TABLE or CREATE UNIQUE INDEX.
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index c6bd09e43fd..459f0db3997 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -3680,7 +3680,7 @@ row_merge_drop_index_dict(
 	ut_ad(!srv_read_only_mode);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 	ut_ad(trx->dict_operation);
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	info = pars_info_create();
 	pars_info_add_ull_literal(info, "indexid", index_id);
@@ -3742,7 +3742,7 @@ row_merge_drop_indexes_dict(
 	ut_ad(!srv_read_only_mode);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 	ut_ad(trx->dict_operation);
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	/* It is possible that table->n_ref_count > 1 when
 	locked=TRUE. In this case, all code that should have an open
@@ -3815,7 +3815,7 @@ row_merge_drop_indexes(
 	ut_ad(!srv_read_only_mode);
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 	ut_ad(trx->dict_operation);
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	index = dict_table_get_first_index(table);
 	ut_ad(dict_index_is_clust(index));
@@ -3900,7 +3900,7 @@ row_merge_drop_indexes(
 				index->lock.x_unlock();
 
 				DEBUG_SYNC_C("merge_drop_index_after_abort");
-				/* covered by dict_sys.mutex */
+				/* covered by dict_sys.latch */
 				MONITOR_INC(MONITOR_BACKGROUND_DROP_INDEX);
 				/* fall through */
 			case ONLINE_INDEX_ABORTED:
@@ -3967,7 +3967,7 @@ row_merge_drop_indexes(
 				break;
 			case ONLINE_INDEX_ABORTED:
 			case ONLINE_INDEX_ABORTED_DROPPED:
-				/* covered by dict_sys.mutex */
+				/* covered by dict_sys.latch */
 				MONITOR_DEC(MONITOR_BACKGROUND_DROP_INDEX);
 			}
 
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index a5c8f7047b6..b38ce54b6ca 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -1421,7 +1421,8 @@ error_exit:
 		srv_stats.n_rows_inserted.inc(size_t(trx->id));
 	}
 
-	/* Not protected by dict_sys.mutex for performance
+	/* Not protected by dict_sys.latch or table->stats_mutex_lock()
+	for performance
 	reasons, we would rather get garbage in stat_n_rows (which is
 	just an estimate anyway) than protecting the following code
 	with a latch. */
@@ -1758,7 +1759,8 @@ row_update_for_mysql(row_prebuilt_t* prebuilt)
 	ut_ad(is_delete == (node->is_delete == PLAIN_DELETE));
 
 	if (is_delete) {
-		/* Not protected by dict_sys.mutex for performance
+		/* Not protected by dict_sys.latch
+		or prebuilt->table->stats_mutex_lock() for performance
 		reasons, we would rather get garbage in stat_n_rows (which is
 		just an estimate anyway) than protecting the following code
 		with a latch. */
@@ -2105,7 +2107,8 @@ row_update_cascade_for_mysql(
 			bool stats;
 
 			if (node->is_delete == PLAIN_DELETE) {
-				/* Not protected by dict_sys.mutex for
+				/* Not protected by dict_sys.latch
+				or node->table->stats_mutex_lock() for
 				performance reasons, we would rather
 				get garbage in stat_n_rows (which is
 				just an estimate anyway) than
@@ -2181,7 +2184,7 @@ row_create_table_for_mysql(
 	que_thr_t*	thr;
 	dberr_t		err;
 
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 	ut_ad(trx->dict_operation_lock_mode == RW_X_LATCH);
 
 	DEBUG_SYNC_C("create_table");
@@ -2256,7 +2259,7 @@ row_create_index_for_mysql(
 	ulint		len;
 	dict_table_t*	table = index->table;
 
-	ut_d(dict_sys.assert_locked());
+	ut_ad(dict_sys.locked());
 
 	for (i = 0; i < index->n_def; i++) {
 		/* Check that prefix_len and actual length
@@ -2545,7 +2548,6 @@ dberr_t row_discard_tablespace_for_mysql(dict_table_t *table, trx_t *trx)
     if (err != DB_SUCCESS)
     {
 rollback:
-      table->stats_bg_flag= BG_STAT_NONE;
       if (fts_exist)
       {
         purge_sys.resume_FTS();
@@ -2557,17 +2559,18 @@ rollback:
   }
 
   row_mysql_lock_data_dictionary(trx);
-  dict_stats_wait_bg_to_stop_using_table(table, trx);
+  dict_stats_wait_bg_to_stop_using_table(table);
 
   trx->op_info = "discarding tablespace";
   trx->dict_operation= true;
 
-  /* Serialize data dictionary operations with dictionary mutex:
+  /* We serialize data dictionary operations with dict_sys.latch:
   this is to avoid deadlocks during data dictionary operations */
 
   err= row_discard_tablespace_foreign_key_checks(trx, table);
   if (err != DB_SUCCESS)
   {
+    table->stats_bg_flag= BG_STAT_NONE;
     row_mysql_unlock_data_dictionary(trx);
     goto rollback;
   }
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 082bf69476c..7e10dcdfe9c 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -114,11 +114,11 @@ row_purge_remove_clust_if_poss_low(
 	if (table_id) {
 retry:
 		purge_sys.check_stop_FTS();
-		dict_sys.mutex_lock();
+		dict_sys.lock(SRW_LOCK_CALL);
 		table = dict_table_open_on_id(
 			table_id, true, DICT_TABLE_OP_OPEN_ONLY_IF_CACHED);
 		if (!table) {
-			dict_sys.mutex_unlock();
+			dict_sys.unlock();
 		} else if (table->n_rec_locks) {
 			for (dict_index_t* ind = UT_LIST_GET_FIRST(
 				     table->indexes); ind;
@@ -142,7 +142,7 @@ removed:
 close_and_exit:
 		if (table) {
 			dict_table_close(table, true, false);
-			dict_sys.mutex_unlock();
+			dict_sys.unlock();
 		}
 		return success;
 	}
@@ -172,7 +172,7 @@ close_and_exit:
 					table->space = nullptr;
 					table->file_unreadable = true;
 				}
-				dict_sys.mutex_unlock();
+				dict_sys.unlock();
 				table = nullptr;
 			}
 			f = fil_delete_tablespace(space_id);
@@ -182,7 +182,7 @@ close_and_exit:
 
 		if (table) {
 			dict_table_close(table, true, false);
-			dict_sys.mutex_unlock();
+			dict_sys.unlock();
 			table = nullptr;
 		}
 
diff --git a/storage/innobase/row/row0quiesce.cc b/storage/innobase/row/row0quiesce.cc
index 74c7b9b35fc..04de3465dcf 100644
--- a/storage/innobase/row/row0quiesce.cc
+++ b/storage/innobase/row/row0quiesce.cc
@@ -486,8 +486,6 @@ row_quiesce_table_has_fts_index(
 {
 	bool			exists = false;
 
-	dict_sys.mutex_lock();
-
 	for (const dict_index_t* index = UT_LIST_GET_FIRST(table->indexes);
 	     index != 0;
 	     index = UT_LIST_GET_NEXT(indexes, index)) {
@@ -498,8 +496,6 @@ row_quiesce_table_has_fts_index(
 		}
 	}
 
-	dict_sys.mutex_unlock();
-
 	return(exists);
 }
 
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 704c067fc43..2c7408c950a 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -182,7 +182,7 @@ restart:
 					lock_release_on_rollback(node->trx,
 								 table);
 					if (!dict_locked) {
-						dict_sys.mutex_lock();
+						dict_sys.lock(SRW_LOCK_CALL);
 					}
 					if (table->release()) {
 						dict_sys.remove(table);
@@ -192,7 +192,7 @@ restart:
 						table->file_unreadable = true;
 					}
 					if (!dict_locked) {
-						dict_sys.mutex_unlock();
+						dict_sys.unlock();
 					}
 					table = nullptr;
 					if (!mdl_ticket);
@@ -432,9 +432,9 @@ static bool row_undo_ins_parse_undo_rec(undo_node_t* node, bool dict_locked)
 		node->table = dict_table_open_on_id(table_id, dict_locked,
 						    DICT_TABLE_OP_NORMAL);
 	} else if (!dict_locked) {
-		dict_sys.mutex_lock();
+		dict_sys.freeze(SRW_LOCK_CALL);
 		node->table = dict_sys.acquire_temporary_table(table_id);
-		dict_sys.mutex_unlock();
+		dict_sys.unfreeze();
 	} else {
 		node->table = dict_sys.acquire_temporary_table(table_id);
 	}
@@ -642,21 +642,19 @@ row_undo_ins(
 
 		log_free_check();
 
-		if (node->table->id == DICT_INDEXES_ID) {
-			ut_ad(!node->table->is_temporary());
-			if (!dict_locked) {
-				dict_sys.mutex_lock();
-			}
+		if (!dict_locked && node->table->id == DICT_INDEXES_ID) {
+			dict_sys.lock(SRW_LOCK_CALL);
 			err = row_undo_ins_remove_clust_rec(node);
-			if (!dict_locked) {
-				dict_sys.mutex_unlock();
-			}
+			dict_sys.unlock();
 		} else {
+			ut_ad(node->table->id != DICT_INDEXES_ID
+			      || !node->table->is_temporary());
 			err = row_undo_ins_remove_clust_rec(node);
 		}
 
 		if (err == DB_SUCCESS && node->table->stat_initialized) {
-			/* Not protected by dict_sys.mutex for
+			/* Not protected by dict_sys.latch
+			or table->stats_mutex_lock() for
 			performance reasons, we would rather get garbage
 			in stat_n_rows (which is just an estimate anyway)
 			than protecting the following code with a latch. */
@@ -665,7 +663,7 @@ row_undo_ins(
 			/* Do not attempt to update statistics when
 			executing ROLLBACK in the InnoDB SQL
 			interpreter, because in that case we would
-			already be holding dict_sys.mutex, which
+			already be holding dict_sys.latch, which
 			would be acquired when updating statistics. */
 			if (!dict_locked) {
 				dict_stats_update_if_needed(node->table,
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index ba675c927eb..5039df2e1a6 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -1215,9 +1215,9 @@ static bool row_undo_mod_parse_undo_rec(undo_node_t* node, bool dict_locked)
 		node->table = dict_table_open_on_id(table_id, dict_locked,
 						    DICT_TABLE_OP_NORMAL);
 	} else if (!dict_locked) {
-		dict_sys.mutex_lock();
+		dict_sys.freeze(SRW_LOCK_CALL);
 		node->table = dict_sys.acquire_temporary_table(table_id);
-		dict_sys.mutex_unlock();
+		dict_sys.unfreeze();
 	} else {
 		node->table = dict_sys.acquire_temporary_table(table_id);
 	}
@@ -1391,7 +1391,7 @@ rollback_clust:
 			/* Do not attempt to update statistics when
 			executing ROLLBACK in the InnoDB SQL
 			interpreter, because in that case we would
-			already be holding dict_sys.mutex, which
+			already be holding dict_sys.latch, which
 			would be acquired when updating statistics. */
 			if (update_statistics && !dict_locked) {
 				dict_stats_update_if_needed(node->table,
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index db666cc5b28..7261636a9ee 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -162,7 +162,7 @@ wsrep_row_upd_index_is_foreign(
   /* No MDL protects dereferencing the members of table->foreign_set. */
   const bool no_lock= !trx->dict_operation_lock_mode;
   if (no_lock)
-    dict_sys.mutex_lock();
+    dict_sys.freeze(SRW_LOCK_CALL);
 
   auto end= table->foreign_set.end();
   const bool is_referenced= end !=
@@ -170,7 +170,7 @@ wsrep_row_upd_index_is_foreign(
                  [index](const dict_foreign_t* f)
                  {return f->foreign_index == index;});
   if (no_lock)
-    dict_sys.mutex_unlock();
+    dict_sys.unfreeze();
 
   return is_referenced;
 }
diff --git a/storage/innobase/srv/srv0srv.cc b/storage/innobase/srv/srv0srv.cc
index 88cf5c0396a..80a412a85bf 100644
--- a/storage/innobase/srv/srv0srv.cc
+++ b/storage/innobase/srv/srv0srv.cc
@@ -1297,7 +1297,7 @@ void srv_monitor_task(void*)
 			    || waited == threshold / 2
 			    || waited == threshold / 4 * 3) {
 				ib::warn() << "Long wait (" << waited
-					   << " seconds) for dict_sys.mutex";
+					   << " seconds) for dict_sys.latch";
 			}
 		}
 	}
@@ -1371,8 +1371,6 @@ void purge_sys_t::stop_SYS()
 /** Stop purge during FLUSH TABLES FOR EXPORT */
 void purge_sys_t::stop()
 {
-  dict_sys.assert_not_locked();
-
   for (;;)
   {
     latch.wr_lock(SRW_LOCK_CALL);
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index 4a6c7ed42a0..4f01de11183 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -604,10 +604,10 @@ trx_resurrect_table_locks(
 		if (dict_table_t* table = dict_table_open_on_id(
 			    p.first, FALSE, DICT_TABLE_OP_LOAD_TABLESPACE)) {
 			if (!table->is_readable()) {
-				dict_sys.mutex_lock();
+				dict_sys.lock(SRW_LOCK_CALL);
 				dict_table_close(table, TRUE, FALSE);
 				dict_sys.remove(table);
-				dict_sys.mutex_unlock();
+				dict_sys.unlock();
 				continue;
 			}