MDEV-21923 LSN allocation is a bottleneck

extra/mariabackup/xtrabackup.cc

@@ -2684,7 +2684,7 @@ static bool innodb_init()
   }
 
   recv_sys.lsn= log_sys.next_checkpoint_lsn=
-    log_sys.get_lsn() - SIZE_OF_FILE_CHECKPOINT;
+    log_get_lsn() - SIZE_OF_FILE_CHECKPOINT;
   log_sys.set_latest_format(false); // not encrypted
   log_hdr_init();
   byte *b= &log_hdr_buf[log_t::START_OFFSET];

mysql-test/suite/sys_vars/r/sysvars_innodb.result

@@ -1092,7 +1092,7 @@ SESSION_VALUE	NULL
 DEFAULT_VALUE	0
 VARIABLE_SCOPE	GLOBAL
 VARIABLE_TYPE	INT UNSIGNED
-VARIABLE_COMMENT	Delay between log buffer spin lock polls (0 to use a blocking latch)
+VARIABLE_COMMENT	Deprecated parameter with no effect
 NUMERIC_MIN_VALUE	0
 NUMERIC_MAX_VALUE	6000
 NUMERIC_BLOCK_SIZE	0

storage/innobase/buf/buf0buf.cc

@@ -558,16 +558,18 @@ buf_page_is_checksum_valid_crc32(
 }
 
 #ifndef UNIV_INNOCHECKSUM
-/** Checks whether the lsn present in the page is lesser than the
-peek current lsn.
-@param check_lsn    lsn to check
+/** Check whether a page is newer than the durable LSN.
+@param check_lsn    whether to check the LSN
 @param read_buf     page frame
-@return whether the FIL_PAGE_LSN is invalid */
-static bool buf_page_check_lsn(bool check_lsn, const byte *read_buf)
+@return whether the FIL_PAGE_LSN is invalid (ahead of the durable LSN) */
+static bool buf_page_check_lsn(bool check_lsn, const byte *read_buf) noexcept
 {
   if (!check_lsn)
     return false;
-  lsn_t current_lsn= log_sys.get_lsn();
+  /* A page may not be read before it is written, and it may not be
+  written before the corresponding log has been durably written.
+  Hence, we refer to the current durable LSN here */
+  lsn_t current_lsn= log_sys.get_flushed_lsn(std::memory_order_relaxed);
   if (UNIV_UNLIKELY(current_lsn == log_sys.FIRST_LSN) &&
       srv_force_recovery == SRV_FORCE_NO_LOG_REDO)
     return false;

storage/innobase/buf/buf0dblwr.cc

@@ -365,7 +365,7 @@ void buf_dblwr_t::recover() noexcept
   ut_ad(log_sys.last_checkpoint_lsn);
   if (!is_created())
     return;
-  const lsn_t max_lsn{log_sys.get_lsn()};
+  const lsn_t max_lsn{log_sys.get_flushed_lsn(std::memory_order_relaxed)};
   ut_ad(recv_sys.scanned_lsn == max_lsn);
   ut_ad(recv_sys.scanned_lsn >= recv_sys.lsn);
 
@@ -780,6 +780,9 @@ void buf_dblwr_t::flush_buffered_writes_completed(const IORequest &request)
     ut_ad(lsn);
     ut_ad(lsn >= bpage->oldest_modification());
     log_write_up_to(lsn, true);
+    ut_ad(!e.request.node->space->full_crc32() ||
+          !buf_page_is_corrupted(true, static_cast<const byte*>(frame),
+                                 e.request.node->space->flags));
     e.request.node->space->io(e.request, bpage->physical_offset(), e_size,
                               frame, bpage);
   }

storage/innobase/buf/buf0flu.cc

@@ -694,7 +694,6 @@ static byte *buf_page_encrypt(fil_space_t *space, buf_page_t *bpage, byte *s,
     {
       static_assert(FIL_PAGE_FCRC32_CHECKSUM == 4, "alignment");
       mach_write_to_4(tmp + len - 4, my_crc32c(0, tmp, len - 4));
-      ut_ad(!buf_page_is_corrupted(true, tmp, space->flags));
     }
 
     d= tmp;
@@ -854,6 +853,8 @@ bool buf_page_t::flush(fil_space_t *space) noexcept
     if (!space->is_temporary() && !space->is_being_imported() &&
         lsn > log_sys.get_flushed_lsn())
       log_write_up_to(lsn, true);
+    ut_ad(space->is_temporary() || !space->full_crc32() ||
+          !buf_page_is_corrupted(true, write_frame, space->flags));
     space->io(IORequest{type, this, slot}, physical_offset(), size,
               write_frame, this);
   }
@@ -1773,8 +1774,9 @@ inline void log_t::write_checkpoint(lsn_t end_lsn) noexcept
 {
   ut_ad(!srv_read_only_mode);
   ut_ad(end_lsn >= next_checkpoint_lsn);
-  ut_ad(end_lsn <= get_lsn());
-  ut_ad(end_lsn + SIZE_OF_FILE_CHECKPOINT <= get_lsn() ||
+  ut_d(const lsn_t current_lsn{get_lsn()});
+  ut_ad(end_lsn <= current_lsn);
+  ut_ad(end_lsn + SIZE_OF_FILE_CHECKPOINT <= current_lsn ||
         srv_shutdown_state > SRV_SHUTDOWN_INITIATED);
 
   DBUG_PRINT("ib_log",
@@ -1903,7 +1905,8 @@ inline void log_t::write_checkpoint(lsn_t end_lsn) noexcept
         ut_ad(!is_opened());
         my_munmap(buf, file_size);
         buf= resize_buf;
-        set_buf_free(START_OFFSET + (get_lsn() - resizing));
+        buf_size= unsigned(std::min<uint64_t>(resize_target - START_OFFSET,
+                                              buf_size_max));
       }
       else
 #endif
@@ -2038,18 +2041,16 @@ static bool log_checkpoint() noexcept
 }
 
 /** Make a checkpoint. */
-ATTRIBUTE_COLD void log_make_checkpoint()
+ATTRIBUTE_COLD void log_make_checkpoint() noexcept
 {
-  buf_flush_wait_flushed(log_sys.get_lsn(std::memory_order_acquire));
+  buf_flush_wait_flushed(log_get_lsn());
   while (!log_checkpoint());
 }
 
 /** Wait for all dirty pages up to an LSN to be written out.
 NOTE: The calling thread is not allowed to hold any buffer page latches! */
 static void buf_flush_wait(lsn_t lsn) noexcept
 {
-  ut_ad(lsn <= log_sys.get_lsn());
-
   lsn_t oldest_lsn;
 
   while ((oldest_lsn= buf_pool.get_oldest_modification(lsn)) < lsn)
@@ -2258,13 +2259,13 @@ static void buf_flush_sync_for_checkpoint(lsn_t lsn) noexcept
   mysql_mutex_unlock(&buf_pool.flush_list_mutex);
 }
 
-/** Check if the adpative flushing threshold is recommended based on
+/** Check if the adaptive flushing threshold is recommended based on
 redo log capacity filled threshold.
 @param oldest_lsn     buf_pool.get_oldest_modification()
 @return true if adaptive flushing is recommended. */
 static bool af_needed_for_redo(lsn_t oldest_lsn) noexcept
 {
-  lsn_t age= (log_sys.get_lsn() - oldest_lsn);
+  lsn_t age= log_sys.get_lsn_approx() - oldest_lsn;
   lsn_t af_lwm= static_cast<lsn_t>(srv_adaptive_flushing_lwm *
     static_cast<double>(log_sys.log_capacity) / 100);
 
@@ -2324,7 +2325,7 @@ static ulint page_cleaner_flush_pages_recommendation(ulint last_pages_in,
 	lsn_t			lsn_rate;
 	ulint			n_pages = 0;
 
-	const lsn_t cur_lsn = log_sys.get_lsn();
+	const lsn_t cur_lsn = log_sys.get_lsn_approx();
 	ut_ad(oldest_lsn <= cur_lsn);
 	ulint pct_for_lsn = af_get_pct_for_lsn(cur_lsn - oldest_lsn);
 	time_t curr_time = time(nullptr);
@@ -2796,7 +2797,7 @@ ATTRIBUTE_COLD void buf_flush_buffer_pool() noexcept
 NOTE: The calling thread is not allowed to hold any buffer page latches! */
 void buf_flush_sync_batch(lsn_t lsn) noexcept
 {
-  lsn= std::max(lsn, log_sys.get_lsn());
+  lsn= std::max(lsn, log_get_lsn());
   mysql_mutex_lock(&buf_pool.flush_list_mutex);
   buf_flush_wait(lsn);
   mysql_mutex_unlock(&buf_pool.flush_list_mutex);
@@ -2815,20 +2816,26 @@ void buf_flush_sync() noexcept
 
   thd_wait_begin(nullptr, THD_WAIT_DISKIO);
   tpool::tpool_wait_begin();
-  mysql_mutex_lock(&buf_pool.flush_list_mutex);
-  for (;;)
+  log_sys.latch.wr_lock(SRW_LOCK_CALL);
+
+  for (lsn_t lsn= log_sys.get_lsn();;)
   {
-    const lsn_t lsn= log_sys.get_lsn();
+    log_sys.latch.wr_unlock();
+    mysql_mutex_lock(&buf_pool.flush_list_mutex);
     buf_flush_wait(lsn);
     /* Wait for the page cleaner to be idle (for log resizing at startup) */
     while (buf_flush_sync_lsn)
       my_cond_wait(&buf_pool.done_flush_list,
                    &buf_pool.flush_list_mutex.m_mutex);
-    if (lsn == log_sys.get_lsn())
+    mysql_mutex_unlock(&buf_pool.flush_list_mutex);
+    log_sys.latch.wr_lock(SRW_LOCK_CALL);
+    lsn_t new_lsn= log_sys.get_lsn();
+    if (lsn == new_lsn)
       break;
+    lsn= new_lsn;
   }
 
-  mysql_mutex_unlock(&buf_pool.flush_list_mutex);
+  log_sys.latch.wr_unlock();
   tpool::tpool_wait_end();
   thd_wait_end(nullptr);
 }
@@ -2848,7 +2855,7 @@ ATTRIBUTE_COLD void buf_pool_t::print_flush_info() const noexcept
     "-------------------",
     lru_size, free_size, dirty_size, dirty_pct);
 
-  lsn_t lsn= log_sys.get_lsn();
+  lsn_t lsn= log_get_lsn();
   lsn_t clsn= log_sys.last_checkpoint_lsn;
   sql_print_information("InnoDB: LSN flush parameters\n"
     "-------------------\n"

storage/innobase/handler/ha_innodb.cc

@@ -243,11 +243,11 @@ static void innodb_max_purge_lag_wait_update(THD *thd, st_mysql_sys_var *,
     if (thd_kill_level(thd))
       break;
     /* Adjust for purge_coordinator_state::refresh() */
-    log_sys.latch.rd_lock(SRW_LOCK_CALL);
+    log_sys.latch.wr_lock(SRW_LOCK_CALL);
     const lsn_t last= log_sys.last_checkpoint_lsn,
       max_age= log_sys.max_checkpoint_age;
-    log_sys.latch.rd_unlock();
     const lsn_t lsn= log_sys.get_lsn();
+    log_sys.latch.wr_unlock();
     if ((lsn - last) / 4 >= max_age / 5)
       buf_flush_ahead(last + max_age / 5, false);
     purge_sys.wake_if_not_active();
@@ -1155,7 +1155,7 @@ innobase_rollback_to_savepoint_can_release_mdl(
 					be rolled back to savepoint */
 
 /** Request notification of log writes */
-static void innodb_log_flush_request(void *cookie);
+static void innodb_log_flush_request(void *cookie) noexcept;
 
 /** Requests for log flushes */
 struct log_flush_request
@@ -4749,11 +4749,13 @@ void log_flush_notify(lsn_t flush_lsn)
 We put the request in a queue, so that we can notify upper layer about
 checkpoint complete when we have flushed the redo log.
 If we have already flushed all relevant redo log, we notify immediately.*/
-static void innodb_log_flush_request(void *cookie)
+static void innodb_log_flush_request(void *cookie) noexcept
 {
+  log_sys.latch.wr_lock(SRW_LOCK_CALL);
   lsn_t flush_lsn= log_sys.get_flushed_lsn();
   /* Load lsn relaxed after flush_lsn was loaded from the same cache line */
   const lsn_t lsn= log_sys.get_lsn();
+  log_sys.latch.wr_unlock();
 
   if (flush_lsn >= lsn)
     /* All log is already persistent. */;
@@ -18440,11 +18442,16 @@ checkpoint_now_set(THD* thd, st_mysql_sys_var*, void*, const void *save)
   const auto size= log_sys.is_encrypted()
     ? SIZE_OF_FILE_CHECKPOINT + 8 : SIZE_OF_FILE_CHECKPOINT;
   mysql_mutex_unlock(&LOCK_global_system_variables);
-  lsn_t lsn;
-  while (!thd_kill_level(thd) &&
-         log_sys.last_checkpoint_lsn.load(std::memory_order_acquire) + size <
-         (lsn= log_sys.get_lsn(std::memory_order_acquire)))
+  while (!thd_kill_level(thd))
+  {
+    log_sys.latch.wr_lock(SRW_LOCK_CALL);
+    lsn_t cp= log_sys.last_checkpoint_lsn.load(std::memory_order_relaxed),
+      lsn= log_sys.get_lsn();
+    log_sys.latch.wr_unlock();
+    if (cp + size >= lsn)
+      break;
     log_make_checkpoint();
+  }
 
   mysql_mutex_lock(&LOCK_global_system_variables);
 }
@@ -18664,35 +18671,23 @@ static void innodb_log_file_size_update(THD *thd, st_mysql_sys_var*,
         mysql_mutex_unlock(&buf_pool.flush_list_mutex);
         if (!resizing || !log_sys.resize_running(thd))
           break;
-        if (resizing > log_sys.get_lsn())
+        log_sys.latch.wr_lock(SRW_LOCK_CALL);
+        while (resizing > log_sys.get_lsn())
         {
           ut_ad(!log_sys.is_mmap());
           /* The server is almost idle. Write dummy FILE_CHECKPOINT records
           to ensure that the log resizing will complete. */
-          log_sys.latch.wr_lock(SRW_LOCK_CALL);
-          while (resizing > log_sys.get_lsn())
-          {
-            mtr_t mtr;
-            mtr.start();
-            mtr.commit_files(log_sys.last_checkpoint_lsn);
-          }
-          log_sys.latch.wr_unlock();
+          mtr_t mtr;
+          mtr.start();
+          mtr.commit_files(log_sys.last_checkpoint_lsn);
         }
+        log_sys.latch.wr_unlock();
       }
     }
   }
   mysql_mutex_lock(&LOCK_global_system_variables);
 }
 
-static void innodb_log_spin_wait_delay_update(THD *, st_mysql_sys_var*,
-                                              void *, const void *save)
-{
-  log_sys.latch.wr_lock(SRW_LOCK_CALL);
-  mtr_t::spin_wait_delay= *static_cast<const unsigned*>(save);
-  mtr_t::finisher_update();
-  log_sys.latch.wr_unlock();
-}
-
 /** Update innodb_status_output or innodb_status_output_locks,
 which control InnoDB "status monitor" output to the error log.
 @param[out]	var	current value
@@ -19551,11 +19546,12 @@ static MYSQL_SYSVAR_ULONGLONG(log_file_size, srv_log_file_size,
   nullptr, innodb_log_file_size_update,
   96 << 20, 4 << 20, std::numeric_limits<ulonglong>::max(), 4096);
 
-static MYSQL_SYSVAR_UINT(log_spin_wait_delay, mtr_t::spin_wait_delay,
-  PLUGIN_VAR_OPCMDARG,
-  "Delay between log buffer spin lock polls (0 to use a blocking latch)",
-  nullptr, innodb_log_spin_wait_delay_update,
-  0, 0, 6000, 0);
+static uint innodb_log_spin_wait_delay;
+
+static MYSQL_SYSVAR_UINT(log_spin_wait_delay, innodb_log_spin_wait_delay,
+  PLUGIN_VAR_OPCMDARG | PLUGIN_VAR_DEPRECATED,
+  "Deprecated parameter with no effect",
+  nullptr, nullptr, 0, 0, 6000, 0);
 
 static MYSQL_SYSVAR_UINT(old_blocks_pct, innobase_old_blocks_pct,
   PLUGIN_VAR_RQCMDARG,

storage/innobase/include/fil0fil.h

@@ -351,7 +351,7 @@ struct fil_space_t final
   /** fil_system.spaces chain node */
   fil_space_t *hash= nullptr;
   /** log_sys.get_lsn() of the most recent fil_names_write_if_was_clean().
-  Reset to 0 by fil_names_clear(). Protected by log_sys.mutex.
+  Reset to 0 by fil_names_clear(). Protected by log_sys.latch_have_wr().
   If and only if this is nonzero, the tablespace will be in named_spaces. */
   lsn_t max_lsn= 0;
   /** base node for the chain of data files; multiple entries are