MDEV-36886 log_t::get_lsn_approx() isn't lower bound

If the execution of the two reads in log_t::get_lsn_approx() is
interleaved with concurrent writes of those fields in
log_t::write_buf() or log_t::persist(), the returned approximation
will be an upper bound. If log_t::append_prepare_wait() is pending,
the approximation could be a lower bound.

We must adjust each caller of log_t::get_lsn_approx() for the
possibility that the return value is larger than
MAX(oldest_modification) in buf_pool.flush_list.

af_needed_for_redo(): Add a comment that explains why the glitch
is not a problem.

page_cleaner_flush_pages_recommendation(): Revise the logic for
the unlikely case cur_lsn < oldest_lsn.  The original logic would have
invoked af_get_pct_for_lsn() with a very large age value, which
would likely cause an overflow of the local variable lsn_age_factor,
and make pct_for_lsn a "random number".  Based on that value,
total_ratio would be normalized to something between 0.0 and 1.0.
Nothing extremely bad should have happened in this case;
the innodb_io_capacity_max should not be exceeded.

Author: dr-m

storage/innobase/buf/buf0flu.cc

@@ -2294,6 +2294,12 @@ redo log capacity filled threshold.
 @return true if adaptive flushing is recommended. */
 static bool af_needed_for_redo(lsn_t oldest_lsn) noexcept
 {
+  /* We may have oldest_lsn > log_sys.get_lsn_approx() if
+  log_t::write_buf() or log_t::persist() are executing concurrently
+  with this. In that case, age > af_lwm should hold, and
+  buf_flush_page_cleaner() would execute one more timed wait. (Not a
+  big problem.) */
+
   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);
@@ -2355,8 +2361,10 @@ static ulint page_cleaner_flush_pages_recommendation(ulint last_pages_in,
 	ulint			n_pages = 0;
 
 	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);
+	/* We may have cur_lsn < oldest_lsn if
+	log_t::write_buf() or log_t::persist() were executing concurrently. */
+	ulint pct_for_lsn = cur_lsn < oldest_lsn
+                ? 0 : af_get_pct_for_lsn(cur_lsn - oldest_lsn);
 	time_t curr_time = time(nullptr);
 	const double max_pct = srv_max_buf_pool_modified_pct;
 

storage/innobase/include/log0log.h

@@ -406,9 +406,10 @@ struct log_t
   @param encrypted  whether the log is encrypted */
   static void header_write(byte *buf, lsn_t lsn, bool encrypted) noexcept;
 
-  /** @return a lower bound estimate of get_lsn(),
+  /** @return an estimate of get_lsn(),
   using acquire-release ordering with write_buf() or persist();
-  this is exact unless append_prepare_wait() is pending */
+  an upper bound if said functions have updated only one of the fields,
+  a lower bound if append_prepare_wait() is pending, otherwise exact */
   lsn_t get_lsn_approx() const noexcept
   {
     /* acquire-release ordering with write_buf() and persist() */