1 files changed, 77 insertions, 54 deletions

diff --git a/storage/innobase/mtr/mtr0mtr.cc b/storage/innobase/mtr/mtr0mtr.cc
index 1834a164..01641f74 100644
--- a/storage/innobase/mtr/mtr0mtr.cc
+++ b/storage/innobase/mtr/mtr0mtr.cc
@@ -308,6 +308,22 @@ void mtr_t::release()
   m_memo.clear();
 }
 
+inline lsn_t log_t::get_write_target() const
+{
+#ifndef SUX_LOCK_GENERIC
+  ut_ad(latch.is_locked());
+#endif
+  if (UNIV_LIKELY(buf_free < max_buf_free))
+    return 0;
+  ut_ad(!is_pmem());
+  /* The LSN corresponding to the end of buf is
+  write_lsn - (first_lsn & 4095) + buf_free,
+  but we use simpler arithmetics to return a smaller write target in
+  order to minimize waiting in log_write_up_to(). */
+  ut_ad(max_buf_free >= 4096 * 4);
+  return write_lsn + max_buf_free / 2;
+}
+
 /** Commit a mini-transaction. */
 void mtr_t::commit()
 {
@@ -331,6 +347,7 @@ void mtr_t::commit()
     std::pair<lsn_t,page_flush_ahead> lsns{do_write()};
     process_freed_pages();
     size_t modified= 0;
+    const lsn_t write_lsn= log_sys.get_write_target();
 
     if (m_made_dirty)
     {
@@ -408,7 +425,8 @@ void mtr_t::commit()
           break;
         default:
           buf_page_t *bpage= static_cast<buf_page_t*>(slot.object);
-          const auto s= bpage->unfix();
+          ut_d(const auto s=)
+            bpage->unfix();
           if (slot.type & MTR_MEMO_MODIFY)
           {
             ut_ad(slot.type == MTR_MEMO_PAGE_X_MODIFY ||
@@ -420,13 +438,10 @@ void mtr_t::commit()
             ut_ad(s < buf_page_t::READ_FIX);
             ut_ad(mach_read_from_8(bpage->frame + FIL_PAGE_LSN) <=
                   m_commit_lsn);
-            if (s >= buf_page_t::UNFIXED)
-            {
-              mach_write_to_8(bpage->frame + FIL_PAGE_LSN, m_commit_lsn);
-              if (UNIV_LIKELY_NULL(bpage->zip.data))
-                memcpy_aligned<8>(FIL_PAGE_LSN + bpage->zip.data,
-                                  FIL_PAGE_LSN + bpage->frame, 8);
-            }
+            mach_write_to_8(bpage->frame + FIL_PAGE_LSN, m_commit_lsn);
+            if (UNIV_LIKELY_NULL(bpage->zip.data))
+              memcpy_aligned<8>(FIL_PAGE_LSN + bpage->zip.data,
+                                FIL_PAGE_LSN + bpage->frame, 8);
             modified++;
           }
           switch (auto latch= slot.type & ~MTR_MEMO_MODIFY) {
@@ -451,6 +466,9 @@ void mtr_t::commit()
 
     if (UNIV_UNLIKELY(lsns.second != PAGE_FLUSH_NO))
       buf_flush_ahead(m_commit_lsn, lsns.second == PAGE_FLUSH_SYNC);
+
+    if (UNIV_UNLIKELY(write_lsn != 0))
+      log_write_up_to(write_lsn, false);
   }
   else
   {
@@ -492,9 +510,20 @@ void mtr_t::rollback_to_savepoint(ulint begin, ulint end)
   m_memo.erase(m_memo.begin() + begin, m_memo.begin() + end);
 }
 
+/** Set create_lsn. */
+inline void fil_space_t::set_create_lsn(lsn_t lsn)
+{
+#ifndef SUX_LOCK_GENERIC
+  /* Concurrent log_checkpoint_low() must be impossible. */
+  ut_ad(latch.is_write_locked());
+#endif
+  create_lsn= lsn;
+}
+
 /** Commit a mini-transaction that is shrinking a tablespace.
-@param space   tablespace that is being shrunk */
-void mtr_t::commit_shrink(fil_space_t &space)
+@param space   tablespace that is being shrunk
+@param size    new size in pages */
+void mtr_t::commit_shrink(fil_space_t &space, uint32_t size)
 {
   ut_ad(is_active());
   ut_ad(!is_inside_ibuf());
@@ -514,6 +543,15 @@ void mtr_t::commit_shrink(fil_space_t &space)
   const lsn_t start_lsn= do_write().first;
   ut_d(m_log.erase());
 
+  fil_node_t *file= UT_LIST_GET_LAST(space.chain);
+  mysql_mutex_lock(&fil_system.mutex);
+  ut_ad(file->is_open());
+  space.size= file->size= size;
+  space.set_create_lsn(m_commit_lsn);
+  mysql_mutex_unlock(&fil_system.mutex);
+
+  space.clear_freed_ranges();
+
   /* Durably write the reduced FSP_SIZE before truncating the data file. */
   log_write_and_flush();
 #ifndef SUX_LOCK_GENERIC
@@ -521,11 +559,11 @@ void mtr_t::commit_shrink(fil_space_t &space)
 #endif
 
   os_file_truncate(space.chain.start->name, space.chain.start->handle,
-                   os_offset_t{space.size} << srv_page_size_shift, true);
+                   os_offset_t{size} << srv_page_size_shift, true);
 
   space.clear_freed_ranges();
 
-  const page_id_t high{space.id, space.size};
+  const page_id_t high{space.id, size};
   size_t modified= 0;
   auto it= m_memo.rbegin();
   mysql_mutex_lock(&buf_pool.flush_list_mutex);
@@ -586,13 +624,6 @@ void mtr_t::commit_shrink(fil_space_t &space)
   log_sys.latch.wr_unlock();
   m_latch_ex= false;
 
-  mysql_mutex_lock(&fil_system.mutex);
-  ut_ad(space.is_being_truncated);
-  ut_ad(space.is_stopping_writes());
-  space.clear_stopping();
-  space.is_being_truncated= false;
-  mysql_mutex_unlock(&fil_system.mutex);
-
   release();
   release_resources();
 }
@@ -680,7 +711,7 @@ The caller must hold exclusive log_sys.latch.
 This is to be used at log_checkpoint().
 @param checkpoint_lsn   the log sequence number of a checkpoint, or 0
 @return current LSN */
-lsn_t mtr_t::commit_files(lsn_t checkpoint_lsn)
+ATTRIBUTE_COLD lsn_t mtr_t::commit_files(lsn_t checkpoint_lsn)
 {
 #ifndef SUX_LOCK_GENERIC
   ut_ad(log_sys.latch.is_write_locked());
@@ -840,26 +871,26 @@ ATTRIBUTE_COLD static void log_overwrite_warning(lsn_t lsn)
 }
 
 /** Wait in append_prepare() for buffer to become available
+@param lsn  log sequence number to write up to
 @param ex   whether log_sys.latch is exclusively locked */
-ATTRIBUTE_COLD void log_t::append_prepare_wait(bool ex) noexcept
+ATTRIBUTE_COLD void log_t::append_prepare_wait(lsn_t lsn, bool ex) noexcept
 {
-  log_sys.waits++;
-  log_sys.unlock_lsn();
+  waits++;
+  unlock_lsn();
 
   if (ex)
-    log_sys.latch.wr_unlock();
+    latch.wr_unlock();
   else
-    log_sys.latch.rd_unlock();
+    latch.rd_unlock();
 
-  DEBUG_SYNC_C("log_buf_size_exceeded");
-  log_buffer_flush_to_disk(log_sys.is_pmem());
+  log_write_up_to(lsn, is_pmem());
 
   if (ex)
-    log_sys.latch.wr_lock(SRW_LOCK_CALL);
+    latch.wr_lock(SRW_LOCK_CALL);
   else
-    log_sys.latch.rd_lock(SRW_LOCK_CALL);
+    latch.rd_lock(SRW_LOCK_CALL);
 
-  log_sys.lock_lsn();
+  lock_lsn();
 }
 
 /** Reserve space in the log buffer for appending data.
@@ -878,34 +909,30 @@ std::pair<lsn_t,byte*> log_t::append_prepare(size_t size, bool ex) noexcept
 # endif
 #endif
   ut_ad(pmem == is_pmem());
-  const lsn_t checkpoint_margin{last_checkpoint_lsn + log_capacity - size};
-  const size_t avail{(pmem ? size_t(capacity()) : buf_size) - size};
   lock_lsn();
   write_to_buf++;
 
-  for (ut_d(int count= 50);
-       UNIV_UNLIKELY((pmem
-                      ? size_t(get_lsn() -
-                               get_flushed_lsn(std::memory_order_relaxed))
-                      : size_t{buf_free}) > avail); )
+  const lsn_t l{lsn.load(std::memory_order_relaxed)}, end_lsn{l + size};
+  size_t b{buf_free};
+
+  if (UNIV_UNLIKELY(pmem
+                    ? (end_lsn -
+                       get_flushed_lsn(std::memory_order_relaxed)) > capacity()
+                    : b + size >= buf_size))
   {
-    append_prepare_wait(ex);
-    ut_ad(count--);
+    append_prepare_wait(l, ex);
+    b= buf_free;
   }
 
-  const lsn_t l{lsn.load(std::memory_order_relaxed)};
-  lsn.store(l + size, std::memory_order_relaxed);
-  const size_t b{buf_free};
-  size_t new_buf_free{b};
-  new_buf_free+= size;
+  lsn.store(end_lsn, std::memory_order_relaxed);
+  size_t new_buf_free= b + size;
   if (pmem && new_buf_free >= file_size)
     new_buf_free-= size_t(capacity());
   buf_free= new_buf_free;
   unlock_lsn();
 
-  if (UNIV_UNLIKELY(l > checkpoint_margin) ||
-      (!pmem && b >= max_buf_free))
-    set_check_flush_or_checkpoint();
+  if (UNIV_UNLIKELY(end_lsn >= last_checkpoint_lsn + log_capacity))
+    set_check_for_checkpoint();
 
   return {l, &buf[b]};
 }
@@ -930,7 +957,7 @@ static mtr_t::page_flush_ahead log_close(lsn_t lsn) noexcept
   else if (UNIV_LIKELY(checkpoint_age <= log_sys.max_checkpoint_age))
     return mtr_t::PAGE_FLUSH_ASYNC;
 
-  log_sys.set_check_flush_or_checkpoint();
+  log_sys.set_check_for_checkpoint();
   return mtr_t::PAGE_FLUSH_SYNC;
 }
 
@@ -989,10 +1016,9 @@ std::pair<lsn_t,mtr_t::page_flush_ahead> mtr_t::do_write()
 #ifndef DBUG_OFF
   do
   {
-    if (m_log_mode != MTR_LOG_ALL)
+    if (m_log_mode != MTR_LOG_ALL ||
+        _db_keyword_(nullptr, "skip_page_checksum", 1))
       continue;
-    DBUG_EXECUTE_IF("skip_page_checksum", continue;);
-
     for (const mtr_memo_slot_t& slot : m_memo)
       if (slot.type & MTR_MEMO_MODIFY)
       {
@@ -1150,9 +1176,6 @@ inline void log_t::resize_write(lsn_t lsn, const byte *end, size_t len,
   }
 }
 
-/** Write the mini-transaction log to the redo log buffer.
-@param len   number of bytes to write
-@return {start_lsn,flush_ahead} */
 std::pair<lsn_t,mtr_t::page_flush_ahead>
 mtr_t::finish_write(size_t len)
 {