squash! 6458cc2

Fix the ARCHIVED_MMAP write path, except for renaming files at
log checkpoint, and implementing crash recovery.

log_t::archived_mmap_switch_complete(): Attempt to complete
log_t::archived_mmap_switch_prepare().

Author: dr-m

storage/innobase/buf/buf0flu.cc

@@ -2172,7 +2172,8 @@ ATTRIBUTE_COLD void buf_flush_ahead(lsn_t lsn, bool furious) noexcept
   if (recv_recovery_is_on())
     recv_sys.apply(true);
 
-  DBUG_EXECUTE_IF("ib_log_checkpoint_avoid_hard", return;);
+  DBUG_EXECUTE_IF("ib_log_checkpoint_avoid_hard",
+                  if (!log_sys.archive) return;);
 
   Atomic_relaxed<lsn_t> &limit= furious
     ? buf_flush_sync_lsn : buf_flush_async_lsn;

storage/innobase/include/log0log.h

@@ -259,8 +259,8 @@ struct log_t
   of the next file created if archive==TRUE; protected by latch */
   lsn_t resize_target;
   /** Buffer for writing to resize_log; @see buf
-  Also a spare buffer between append_prepare_archived_mmap() and
-  archive_new_mmap() */
+  Also a spare buffer between archived_mmap_switch_prepare()
+  and archived_mmap_switch_complete() */
   byte *resize_buf;
   /** Buffer for writing to resize_log; @see flush_buf */
   byte *resize_flush_buf;
@@ -402,13 +402,6 @@ struct log_t
   @return whether an error occurred */
   static bool resize_rename() noexcept;
 
-  /** @return pointer for writing to resize_buf
-  @retval nullptr if no is_mmap() based resizing is active */
-  inline byte *resize_buf_begin(lsn_t lsn) const noexcept;
-  /** @return end of resize_buf */
-  inline const byte *resize_buf_end() const noexcept
-  { return resize_buf + resize_target; }
-
   /** Initialise the redo log subsystem. */
   void create() noexcept;
 
@@ -484,9 +477,23 @@ struct log_t
   /** Persist the log.
   @param lsn            desired new value of flushed_to_disk_lsn */
   void persist(lsn_t lsn) noexcept;
-  /** Switch the log buffers. */
-  inline void archive_new_mmap() noexcept;
+  /** @return the overflow buffer when ARCHIVED_MMAP is wrapping around */
+  byte *get_archived_mmap_switch() const noexcept
+  {
+    ut_ad(archived_mmap_switch());
+    return resize_buf + START_OFFSET;
+  }
 #endif
+  /** @return whether archived_mmap_switch_complete() needs to be called */
+  bool archived_mmap_switch() const noexcept
+  {
+    ut_ad(latch_have_any());
+    return UNIV_UNLIKELY(archive && resize_buf);
+  }
+  /** Attempt to finish archived_mmap_switch_prepare().
+  @return the current LSN in the new file
+  @retval 0 if no switch took place */
+  ATTRIBUTE_COLD lsn_t archived_mmap_switch_complete() noexcept;
   /** Create a new log file when the current one will fill up.
   @param buf     log records to append
   @param length  size of the log records, in bytes
@@ -539,7 +546,7 @@ struct log_t
   /** Wait in append_prepare<ARCHIVED_MMAP>() for buffer to become available
   @param late   whether the WRITE_BACKOFF flag had already been set
   @param ex     whether log_sys.latch is exclusively locked */
-  ATTRIBUTE_COLD void append_prepare_archived_mmap(bool late, bool ex)
+  ATTRIBUTE_COLD void archived_mmap_switch_prepare(bool late, bool ex)
     noexcept;
 #endif
 public:

storage/innobase/include/mtr0mtr.h

@@ -688,13 +688,15 @@ struct mtr_t {
   ATTRIBUTE_NOINLINE size_t crc32c() noexcept;
 
   /** Commit the mini-transaction log.
-  @tparam pmem log_sys.is_mmap()
+  @tparam mmap log_sys.is_mmap()
   @param mtr   mini-transaction
   @param lsns  {start_lsn,flush_ahead_lsn} */
-  template<bool pmem>
+  template<bool mmap>
   static void commit_log(mtr_t *mtr, std::pair<lsn_t,lsn_t> lsns) noexcept;
 
-  /** Release log_sys.latch. */
+  /** Release log_sys.latch.
+  @tparam mmap log_sys.is_mmap() */
+  template<bool mmap>
   void commit_log_release() noexcept;
 
   /** Append the redo log records to the redo log buffer.

storage/innobase/log/log0log.cc

@@ -967,8 +967,8 @@ static size_t log_pad(lsn_t lsn, size_t pad, byte *begin, byte *extra)
 #endif
 
 #ifdef HAVE_PMEM
-ATTRIBUTE_COLD void log_t::append_prepare_archived_mmap(bool late, bool ex)
-  noexcept
+ATTRIBUTE_COLD
+void log_t::archived_mmap_switch_prepare(bool late, bool ex) noexcept
 {
   ut_ad(archive);
   ut_ad(is_mmap());
@@ -1056,16 +1056,13 @@ ATTRIBUTE_COLD void log_t::append_prepare_archived_mmap(bool late, bool ex)
     }
     while (false);
 
-    // TODO: adjust this, and clear the WRITE_BACKOFF flag
-    ut_ad(lsn - get_flushed_lsn(std::memory_order_relaxed) < capacity() ||
-        overwrite_warned);
-    persist(lsn); // TODO: pmem_persist()
+    ut_ad(lsn - get_flushed_lsn(std::memory_order_relaxed) < capacity());
+    persist(lsn);
     /* Above we cleared the WRITE_BACKOFF flag,
     which our caller will recheck. */
     if (ex)
       return;
     latch.wr_unlock();
-    buf_flush_ahead(lsn, false);
   }
 
 done:

storage/innobase/mtr/mtr0mtr.cc

@@ -338,15 +338,48 @@ void mtr_t::release()
   m_memo.clear();
 }
 
+#ifdef HAVE_PMEM
+ATTRIBUTE_COLD lsn_t log_t::archived_mmap_switch_complete() noexcept
+{
+  ut_ad(latch_have_wr());
+  if (!archive || !resize_buf)
+    return 0;
+  const lsn_t lsn{get_lsn()}, end_lsn{first_lsn + capacity()};
+  if (lsn < end_lsn)
+    return 0;
+  persist(lsn);
+  my_munmap(buf, file_size);
+  /* TODO: make the file read-only */
+  buf= resize_buf;
+  resize_buf= nullptr;
+  first_lsn= end_lsn;
+  file_size= resize_target;
+  return lsn;
+}
+#endif
+
+template<bool mmap>
 ATTRIBUTE_NOINLINE void mtr_t::commit_log_release() noexcept
 {
   if (m_latch_ex)
   {
+  completed:
+    const lsn_t lsn{mmap ? log_sys.archived_mmap_switch_complete() : 0};
     log_sys.latch.wr_unlock();
     m_latch_ex= false;
+    if (mmap && lsn)
+      buf_flush_ahead(lsn, true);
   }
   else
+  {
+    const bool retry{mmap && log_sys.archived_mmap_switch()};
     log_sys.latch.rd_unlock();
+    if (retry)
+    {
+      log_sys.latch.wr_lock(SRW_LOCK_CALL);
+      goto completed;
+    }
+  }
 }
 
 static ATTRIBUTE_NOINLINE ATTRIBUTE_COLD
@@ -397,12 +430,12 @@ void mtr_t::commit_log(mtr_t *mtr, std::pair<lsn_t,lsn_t> lsns) noexcept
     buf_pool.page_cleaner_wakeup();
     mysql_mutex_unlock(&buf_pool.flush_list_mutex);
 
-    mtr->commit_log_release();
+    mtr->commit_log_release<mmap>();
     mtr->release();
   }
   else
   {
-    mtr->commit_log_release();
+    mtr->commit_log_release<mmap>();
 
     for (auto it= mtr->m_memo.rbegin(); it != mtr->m_memo.rend(); )
     {
@@ -893,39 +926,20 @@ log_t::append_prepare<log_t::ARCHIVED_MMAP>(size_t size, bool ex) noexcept
                         (WRITE_TO_BUF - 1)) >=
                        capacity() -
                        (lsn= base_lsn.load(std::memory_order_relaxed)) -
-                       first_lsn - size))
+                       first_lsn - size) && !resize_buf)
   {
     /* The following is inlined here instead of being part of
     append_prepare_wait(), in order to increase the locality of reference
     and to set the WRITE_BACKOFF flag as soon as possible. */
     bool late(write_lsn_offset.fetch_or(WRITE_BACKOFF) & WRITE_BACKOFF);
     /* Subtract our LSN overshoot. */
     write_lsn_offset.fetch_sub(size);
-    append_prepare_archived_mmap(late, ex);
+    archived_mmap_switch_prepare(late, ex);
   }
 
   lsn+= l;
   return {lsn, buf + FIRST_LSN + (lsn - first_lsn)};
 }
-
-inline void log_t::archive_new_mmap() noexcept
-{
-  ut_ad(latch_have_any());
-  ut_ad(!resize_log.is_opened());
-  ut_ad(!resize_in_progress());
-  ut_ad(resize_target >= 4U << 20);
-  ut_ad(is_latest());
-
-  resize_wrap_mutex.wr_lock();
-  if (resize_buf)
-  {
-    my_munmap(buf, size_t(file_size));
-    buf= resize_buf;
-    resize_buf= nullptr;
-    file_size= resize_target;
-  }
-  resize_wrap_mutex.wr_unlock();
-}
 #endif
 
 ATTRIBUTE_COLD void log_t::append_prepare_wait(bool late, bool ex) noexcept
@@ -1299,14 +1313,14 @@ mtr_t::finish_writer(mtr_t *mtr, size_t len)
 #ifdef HAVE_PMEM
   else
   {
-    byte *const end= &log_sys.buf[log_sys.file_size];
+    const size_t file_size= log_sys.file_size;
+    byte *const buf{log_sys.buf};
+    byte *const end= &buf[file_size];
     if (UNIV_LIKELY(start.second + len <= end))
       goto write_normal;
-    if (mode == log_t::CIRCULAR_MMAP)
-      log_sys.archived_lsn= 0;
-    else
-      log_sys.archive_new_mmap();
-    byte *const begin= &log_sys.buf[log_sys.START_OFFSET];
+    byte *const begin= mode == log_t::ARCHIVED_MMAP
+      ? log_sys.get_archived_mmap_switch()
+      : buf + log_sys.START_OFFSET;
     for (const mtr_buf_t::block_t &b : mtr->m_log)
     {
       size_t size{b.used()};