squash! 9967583

mtr_t::finish_writer(): Specialize for innodb_log_archive=ON.

TODO: Specialize log_t::resizing_and_latch.

Author: dr-m

storage/innobase/handler/ha_innodb.cc

@@ -19435,14 +19435,19 @@ static void innodb_log_archive_update(THD *, st_mysql_sys_var*,
                                       void *, const void *save) noexcept
 {
   const my_bool archive= *static_cast<const my_bool*>(save);
-  log_sys.latch.rd_lock(SRW_LOCK_CALL);
-  if (archive && UNIV_UNLIKELY(log_sys.resize_in_progress() != 0))
+  log_sys.latch.wr_lock(SRW_LOCK_CALL);
+  const lsn_t resizing{log_sys.resize_in_progress()};
+  if (archive && UNIV_UNLIKELY(resizing != 0))
     my_printf_error(ER_WRONG_USAGE,
                     "SET GLOBAL innodb_log_file_size is in progress", MYF(0));
   else
+  {
     log_sys.archive= archive;
-  log_sys.archived_lsn= 0; // FIXME: remove this
-  log_sys.latch.rd_unlock();
+    if (!resizing)
+      mtr_t::finisher_update();
+  }
+  log_sys.archived_lsn= 0; // FIXME: move this to log_t::write_checkpoint()
+  log_sys.latch.wr_unlock();
 }
 
 static MYSQL_SYSVAR_BOOL(log_archive, log_sys.archive,

storage/innobase/include/log0log.h

@@ -273,8 +273,7 @@ struct log_t
   uint write_size;
   /** format of the redo log: e.g., FORMAT_10_8 */
   uint32_t format;
-  /** the current value of innodb_log_archive;
-  protected by LOCK_global_system_variable and shared latch */
+  /** the current value of innodb_log_archive; protected by latch.wr_lock() */
   my_bool archive;
   /** whether the memory-mapped interface is enabled for the log */
   my_bool log_mmap;

storage/innobase/include/mtr0mtr.h

@@ -701,12 +701,22 @@ struct mtr_t {
   @return {start_lsn,flush_ahead_lsn} */
   std::pair<lsn_t,lsn_t> do_write() noexcept;
 
+  /** How to write log */
+  enum finish_writing {
+    /** circular memory-mapped writing when log_sys.is_mmap() */
+    CIRCULAR_MMAP,
+    /** memory-mapped log for log_sys.archive */
+    ARCHIVED_MMAP,
+    /** normal writing !log_sys.is_mmap() */
+    WRITE_NORMAL
+  };
+
   /** Append the redo log records to the redo log buffer.
-  @tparam mmap log_sys.is_mmap()
+  @tparam how  how to write
   @param mtr   mini-transaction
   @param len   number of bytes to write
   @return {start_lsn,flush_ahead_lsn} */
-  template<bool mmap> static
+  template<finish_writing how> static
   std::pair<lsn_t,lsn_t> finish_writer(mtr_t *mtr, size_t len);
 
   /** The applicable variant of commit_log() */

storage/innobase/mtr/mtr0mtr.cc

@@ -52,12 +52,14 @@ void mtr_t::finisher_update()
   if (log_sys.is_mmap())
   {
     commit_logger= mtr_t::commit_log<true>;
-    finisher= mtr_t::finish_writer<true>;
+    finisher= log_sys.archive
+      ? mtr_t::finish_writer<ARCHIVED_MMAP>
+      : mtr_t::finish_writer<CIRCULAR_MMAP>;
     return;
   }
   commit_logger= mtr_t::commit_log<false>;
 #endif
-  finisher= mtr_t::finish_writer<false>;
+  finisher= mtr_t::finish_writer<WRITE_NORMAL>;
 }
 
 void mtr_memo_slot_t::release() const
@@ -920,6 +922,7 @@ ATTRIBUTE_COLD void log_t::append_prepare_wait(bool late, bool ex) noexcept
     {
       ut_ad(lsn - get_flushed_lsn(std::memory_order_relaxed) < capacity() ||
             overwrite_warned);
+      ut_a(!archive); // FIXME: create, allocate and attach a new file
       persist(lsn);
     }
 #endif
@@ -1214,25 +1217,30 @@ inline void log_t::append(byte *&d, const void *s, size_t size) noexcept
   d+= size;
 }
 
-template<bool mmap>
-std::pair<lsn_t,lsn_t> mtr_t::finish_writer(mtr_t *mtr, size_t len)
+template<mtr_t::finish_writing how>
+std::pair<lsn_t,lsn_t>
+mtr_t::finish_writer(mtr_t *mtr, size_t len)
 {
   ut_ad(log_sys.is_latest());
   ut_ad(!recv_no_log_write);
   ut_ad(mtr->is_logged());
   ut_ad(mtr->m_latch_ex ? log_sys.latch_have_wr() : log_sys.latch_have_rd());
   ut_ad(len < recv_sys.MTR_SIZE_MAX);
+  ut_ad(how == WRITE_NORMAL || log_sys.archive == (how == ARCHIVED_MMAP));
 
   const size_t size{mtr->m_commit_lsn ? 5U + 8U : 5U};
   std::pair<lsn_t, byte*> start=
-    log_sys.append_prepare<mmap>(len, mtr->m_latch_ex);
+    log_sys.append_prepare<how != WRITE_NORMAL>(len, mtr->m_latch_ex);
 
-  if (!mmap)
+  if (how == WRITE_NORMAL ||
+      UNIV_LIKELY(start.second + len <= &log_sys.buf[log_sys.file_size]))
   {
     for (const mtr_buf_t::block_t &b : mtr->m_log)
       log_sys.append(start.second, b.begin(), b.used());
 
+#ifdef HAVE_PMEM
   write_trailer:
+#endif
     *start.second++= log_sys.get_sequence_bit(start.first + len - size);
     if (mtr->m_commit_lsn)
     {
@@ -1245,12 +1253,10 @@ std::pair<lsn_t,lsn_t> mtr_t::finish_writer(mtr_t *mtr, size_t len)
   }
   else
   {
-    if (UNIV_LIKELY(start.second + len <= &log_sys.buf[log_sys.file_size]))
-    {
-      for (const mtr_buf_t::block_t &b : mtr->m_log)
-        log_sys.append(start.second, b.begin(), b.used());
-      goto write_trailer;
-    }
+#ifndef HAVE_PMEM
+    static_assert(how == WRITE_NORMAL, "");
+#else
+    ut_a(how == CIRCULAR_MMAP); // FIXME: implement ARCHIVED_MMAP
     for (const mtr_buf_t::block_t &b : mtr->m_log)
     {
       size_t size{b.used()};
@@ -1288,9 +1294,13 @@ std::pair<lsn_t,lsn_t> mtr_t::finish_writer(mtr_t *mtr, size_t len)
     start.second= log_sys.buf +
       ((size >= size_left) ? log_sys.START_OFFSET : log_sys.file_size) +
       (size - size_left);
+#endif
   }
 
-  log_sys.resize_write(start.first, start.second, len, size);
+  if (how == ARCHIVED_MMAP)
+    ut_ad(!log_sys.resize_in_progress());
+  else
+    log_sys.resize_write(start.first, start.second, len, size);
 
   mtr->m_commit_lsn= start.first + len;
   return {start.first, log_close(mtr->m_commit_lsn)};