[seekdb][PALF] Remove LogGroupBuffer copy from transaction log flush

src/logservice/CMakeLists.txt

@@ -52,6 +52,7 @@ ob_set_subtarget(ob_logservice palf
   palf/log_storage.cpp
   palf/log_task.cpp
   palf/log_writer_utils.cpp
+  palf/palf_log_buffer.cpp
   palf/lsn.cpp
   palf/lsn_allocator.cpp
   palf/palf_base_info.cpp

src/logservice/ob_log_handler.cpp

@@ -184,6 +184,25 @@ int ObLogHandler::append_big_log(const void *buffer,
   return ret;
 }
 
+int ObLogHandler::append_owned(palf::PalfLogBuffer &buffer,
+                               const SCN &ref_scn,
+                               const bool need_nonblock,
+                               AppendCb *cb,
+                               LSN &lsn,
+                               SCN &scn)
+{
+  int ret = OB_SUCCESS;
+  if (!buffer.is_valid() || !buffer.is_sealed() || buffer.get_size() <= 0
+      || buffer.get_size() > MAX_LOG_BODY_SIZE || NULL == cb || !ref_scn.is_valid()) {
+    ret = OB_INVALID_ARGUMENT;
+  } else if (OB_FAIL(append_owned_(buffer, ref_scn, need_nonblock, cb, lsn, scn))) {
+    if (OB_EAGAIN != ret) {
+      CLOG_LOG(WARN, "appending owned log fails", K(ret), K(buffer), K(ref_scn));
+    }
+  }
+  return ret;
+}
+
 int ObLogHandler::get_append_mode_initial_scn(share::SCN &ref_scn) const
 {
   int ret = OB_SUCCESS;
@@ -332,6 +351,27 @@ int ObLogHandler::append_(const void *buffer,
                           AppendCb *cb,
                           palf::LSN &lsn,
                           share::SCN &scn)
+{
+  int ret = OB_SUCCESS;
+  palf::PalfLogBuffer owned_buffer;
+  if (NULL == buffer || nbytes <= 0 || nbytes > MAX_LOG_BODY_SIZE
+      || NULL == cb || !ref_scn.is_valid()) {
+    ret = OB_INVALID_ARGUMENT;
+  } else if (OB_FAIL(owned_buffer.copy_from(static_cast<const char *>(buffer), nbytes,
+                                            palf::LogEntryHeader::HEADER_SER_SIZE))) {
+    CLOG_LOG(WARN, "copy legacy log into owned buffer failed", K(ret), K(nbytes));
+  } else {
+    ret = append_owned_(owned_buffer, ref_scn, need_nonblock, cb, lsn, scn);
+  }
+  return ret;
+}
+
+int ObLogHandler::append_owned_(palf::PalfLogBuffer &buffer,
+                                const share::SCN &ref_scn,
+                                const bool need_nonblock,
+                                AppendCb *cb,
+                                palf::LSN &lsn,
+                                share::SCN &scn)
 {
   int ret = OB_SUCCESS;
   int64_t wait_times = 0;
@@ -347,7 +387,7 @@ int ObLogHandler::append_(const void *buffer,
         ret = OB_NOT_INIT;
       } else if (is_in_stop_state_ || is_offline_) {
         ret = OB_NOT_RUNNING;
-      } else if (OB_FAIL(palf_handle_.append(opts, buffer, nbytes, ref_scn, lsn, scn))) {
+      } else if (OB_FAIL(palf_handle_.append(opts, buffer, ref_scn, lsn, scn))) {
         if (REACH_TIME_INTERVAL(1*1000*1000)) {
           CLOG_LOG(WARN, "palf_handle_ append failed", K(ret), KPC(this));
         }
@@ -356,7 +396,7 @@ int ObLogHandler::append_(const void *buffer,
         cb->__set_lsn(lsn);
         cb->__set_scn(scn);
         ret = apply_status_->push_append_cb(cb);
-        CLOG_LOG(TRACE, "palf_handle_ push_append_cb success", K(lsn), K(scn), K(nbytes));
+        CLOG_LOG(TRACE, "palf_handle_ push_append_cb success", K(lsn), K(scn));
       }
     } while (0);
     // check if need wait and retry append

src/logservice/ob_log_handler.h

@@ -25,6 +25,7 @@
 #include "palf/palf_handle.h"
 #include "palf/palf_base_info.h"
 #include "palf/palf_iterator.h"
+#include "palf/palf_log_buffer.h"
 
 namespace oceanbase
 {
@@ -158,6 +159,13 @@ class ObLogHandler : public ObILogHandler
                      palf::LSN &lsn,
                      share::SCN &scn) override final;
 
+  int append_owned(palf::PalfLogBuffer &buffer,
+                   const share::SCN &ref_scn,
+                   const bool need_nonblock,
+                   AppendCb *cb,
+                   palf::LSN &lsn,
+                   share::SCN &scn);
+
   // @description: get ref_scn of APPEND mode
   // @return
   // - OB_SUCCESS
@@ -337,6 +345,12 @@ class ObLogHandler : public ObILogHandler
               AppendCb *cb,
               palf::LSN &lsn,
               share::SCN &scn);
+  int append_owned_(palf::PalfLogBuffer &buffer,
+                    const share::SCN &ref_scn,
+                    const bool need_nonblock,
+                    AppendCb *cb,
+                    palf::LSN &lsn,
+                    share::SCN &scn);
 
   template<typename StartPoint, typename IteratorType>
   int seek_log_iterator_dispatch_(const StartPoint &start_point,

src/logservice/palf/log_block_handler.cpp

@@ -89,6 +89,9 @@ int LogDIOAlignedBuf::align_buf(const char *input,
         K(input_len), K(offset));
   } else if (false == need_align_()) {
     PALF_LOG(TRACE, "no need align", K(ret));
+  } else if (buf_write_offset_ + input_len + align_size_ - 1 > aligned_buf_size_) {
+    ret = OB_BUF_NOT_ENOUGH;
+    PALF_LOG(ERROR, "aligned buffer is not enough", K(ret), K(input_len), KPC(this));
   } else {
     int64_t start_ts = ObTimeUtility::fast_current_time();
     memcpy(static_cast<char*>(aligned_data_buf_) + buf_write_offset_, input, input_len);
@@ -103,6 +106,50 @@ int LogDIOAlignedBuf::align_buf(const char *input,
   return ret;
 }
 
+int LogDIOAlignedBuf::align_buf(const LogWriteBuf &write_buf,
+                                char *&output,
+                                int64_t &output_len,
+                                offset_t &offset)
+{
+  int ret = OB_SUCCESS;
+  const int64_t input_len = write_buf.get_total_size();
+  if (!write_buf.is_valid() || input_len <= 0) {
+    ret = OB_INVALID_ARGUMENT;
+  } else if (false == need_align_()) {
+    ret = OB_NOT_SUPPORTED;
+  } else if (buf_write_offset_ + input_len + align_size_ - 1 > aligned_buf_size_) {
+    ret = OB_BUF_NOT_ENOUGH;
+    PALF_LOG(ERROR, "aligned buffer is not enough for fragmented write", K(ret),
+        K(input_len), KPC(this), K(write_buf));
+  } else {
+    const int64_t start_ts = ObTimeUtility::fast_current_time();
+    const int64_t buf_cnt = write_buf.get_buf_count();
+    for (int64_t i = 0; OB_SUCC(ret) && i < buf_cnt; ++i) {
+      const char *buf = NULL;
+      int64_t buf_len = 0;
+      bool is_fill = false;
+      char fill_char = 0;
+      if (OB_FAIL(write_buf.get_write_buf(i, buf, buf_len, is_fill, fill_char))) {
+        PALF_LOG(ERROR, "get fragmented write buffer failed", K(ret), K(i), K(write_buf));
+      } else if (is_fill) {
+        MEMSET(aligned_data_buf_ + buf_write_offset_, fill_char, buf_len);
+        buf_write_offset_ += buf_len;
+      } else {
+        MEMCPY(aligned_data_buf_ + buf_write_offset_, buf, buf_len);
+        buf_write_offset_ += buf_len;
+      }
+    }
+    if (OB_SUCC(ret)) {
+      align_buf_();
+      output = aligned_data_buf_;
+      output_len = buf_write_offset_;
+      offset = lower_align(offset, align_size_);
+      aligned_used_ts_ += ObTimeUtility::fast_current_time() - start_ts;
+    }
+  }
+  return ret;
+}
+
 void LogDIOAlignedBuf::truncate_buf()
 {
   if (false == need_align_()) {
@@ -415,28 +462,94 @@ int LogBlockHandler::inner_write_once_(const offset_t offset,
   return ret;
 }
 
-int LogBlockHandler::inner_writev_once_(const offset_t offset,
-    const LogWriteBuf &write_buf)
+int LogBlockHandler::inner_write_fragments_(const offset_t offset,
+                                            const LogWriteBuf &write_buf)
 {
   int ret = OB_SUCCESS;
-  int64_t write_size = 0;
   const int64_t write_buf_cnt = write_buf.get_buf_count();
   offset_t curr_write_offset = offset;
   for (int64_t i = 0; OB_SUCC(ret) && i < write_buf_cnt; i++) {
     const char *buf = NULL;
     int64_t buf_len = 0;
-    if (OB_FAIL(write_buf.get_write_buf(i, buf, buf_len))) {
+    bool is_fill = false;
+    char fill_char = 0;
+    if (OB_FAIL(write_buf.get_write_buf(i, buf, buf_len, is_fill, fill_char))) {
       PALF_LOG(ERROR, "LogWriteBuf get_write_buf failed", K(ret), K(i));
-    } else if (OB_FAIL(inner_write_once_(curr_write_offset, buf, buf_len))) {
+    } else if (!is_fill && OB_FAIL(inner_write_once_(curr_write_offset, buf, buf_len))) {
       PALF_LOG(ERROR, "inner_write_once_ failed", K(ret), K(offset));
-    } else {
-      // NB: Advance write offset
+    } else if (is_fill) {
+      char fill_buf[LOG_DIO_ALIGN_SIZE];
+      MEMSET(fill_buf, fill_char, sizeof(fill_buf));
+      int64_t written_size = 0;
+      while (OB_SUCC(ret) && written_size < buf_len) {
+        const int64_t write_size = MIN(buf_len - written_size,
+                                       static_cast<int64_t>(sizeof(fill_buf)));
+        if (OB_FAIL(inner_write_once_(curr_write_offset + written_size,
+                                      fill_buf,
+                                      write_size))) {
+          PALF_LOG(ERROR, "write virtual fill fragment failed", K(ret), K(i),
+              K(curr_write_offset), K(written_size), K(write_size));
+        } else {
+          written_size += write_size;
+        }
+      }
+    }
+    if (OB_SUCC(ret)) {
       curr_write_offset += buf_len;
     }
   }
   return ret;
 }
 
+int LogBlockHandler::inner_writev_once_(const offset_t offset,
+    const LogWriteBuf &write_buf)
+{
+  int ret = OB_SUCCESS;
+  const int64_t write_buf_cnt = write_buf.get_buf_count();
+  if (write_buf_cnt <= 0) {
+    ret = OB_INVALID_ARGUMENT;
+  } else if (!dio_aligned_buf_.need_align()) {
+    ret = inner_write_fragments_(offset, write_buf);
+  } else {
+    char *aligned_buf = NULL;
+    int64_t aligned_buf_len = 0;
+    offset_t aligned_block_offset = offset;
+    const int64_t input_len = write_buf.get_total_size();
+    ret = dio_aligned_buf_.align_buf(write_buf, aligned_buf,
+                                     aligned_buf_len, aligned_block_offset);
+    if (OB_BUF_NOT_ENOUGH == ret) {
+      PALF_LOG(WARN, "DIO gather buffer is full, fall back to fragmented writes",
+          K(ret), K(offset), K(write_buf));
+      ret = inner_write_fragments_(offset, write_buf);
+    } else if (OB_FAIL(ret)) {
+      PALF_LOG(ERROR, "gather fragmented write buffer failed", K(ret), K(offset), K(write_buf));
+    } else if (OB_FAIL(inner_write_impl_(io_fd_, aligned_buf, aligned_buf_len,
+                                         aligned_block_offset))) {
+      PALF_LOG(ERROR, "pwrite gathered buffer failed", K(ret), K(offset),
+          K(aligned_block_offset), K(aligned_buf_len));
+    } else {
+      dio_aligned_buf_.truncate_buf();
+      const int64_t total_write_size = ATOMIC_AAF(&total_write_size_, input_len);
+      const int64_t total_write_size_after_dio =
+          ATOMIC_AAF(&total_write_size_after_dio_, aligned_buf_len);
+      const int64_t count = ATOMIC_AAF(&count_, 1);
+      const int64_t ob_pwrite_used_ts = ATOMIC_LOAD(&ob_pwrite_used_ts_);
+      if (palf_reach_time_interval(PALF_IO_STAT_PRINT_INTERVAL_US, trace_time_)) {
+        const int64_t each_pwrite_cost = ob_pwrite_used_ts / count;
+        PALF_LOG(INFO, "[PALF STAT WRITE LOG INFO TO DISK]", K(ret), K(offset), KPC(this),
+            K(aligned_buf_len), K(aligned_block_offset), "buf_len", input_len,
+            K(total_write_size), K(total_write_size_after_dio), K(ob_pwrite_used_ts),
+            K(count), K(each_pwrite_cost));
+        ATOMIC_STORE(&total_write_size_, 0);
+        ATOMIC_STORE(&total_write_size_after_dio_, 0);
+        ATOMIC_STORE(&count_, 0);
+        ATOMIC_STORE(&ob_pwrite_used_ts_, 0);
+      }
+    }
+  }
+  return ret;
+}
+
 int LogBlockHandler::inner_write_impl_(const ObIOFd &io_fd, const char *buf, const int64_t count, const int64_t offset)
 {
   int ret = OB_SUCCESS;

src/logservice/palf/log_block_handler.h

@@ -57,10 +57,18 @@ class LogDIOAlignedBuf {
                 int64_t &output_len,
                 offset_t &offset);
 
+  // Gather every fragment into the same aligned buffer and finish alignment
+  // once, so one logical batch is persisted by one pwrite.
+  int align_buf(const LogWriteBuf &write_buf,
+                char *&output,
+                int64_t &output_len,
+                offset_t &offset);
+
   // @brief this function used to truncate 'aligned_data_buf_', move
   // the tail unaligned part to head
   void truncate_buf();
   void reset_buf();
+  bool need_align() const { return need_align_(); }
 
   TO_STRING_KV(K_(buf_write_offset), K_(buf_padding_size), K_(align_size), K_(aligned_buf_size),
       K_(aligned_used_ts), K_(truncate_used_ts));
@@ -160,6 +168,8 @@ class LogBlockHandler {
   int inner_write_once_(const offset_t offset,
       const char *buf,
       const int64_t buf_len);
+  int inner_write_fragments_(const offset_t offset,
+      const LogWriteBuf &write_buf);
   int inner_writev_once_(const offset_t offset,
       const LogWriteBuf &write_buf);
   int inner_write_impl_(const ObIOFd &io_fd, const char *buf, const int64_t count, const int64_t offset);

src/logservice/palf/log_engine.cpp

@@ -327,6 +327,32 @@ int LogEngine::submit_flush_log_task(const FlushLogCbCtx &flush_log_cb_ctx,
   return ret;
 }
 
+int LogEngine::submit_flush_log_task(const FlushLogCbCtx &flush_log_cb_ctx,
+                                     LogGroupWriteBuf &group_write_buf)
+{
+  int ret = OB_SUCCESS;
+  LogIOFlushLogTask *flush_log_task = NULL;
+  if (IS_NOT_INIT) {
+    ret = OB_NOT_INIT;
+  } else if (false == flush_log_cb_ctx.is_valid() || !group_write_buf.is_valid()) {
+    ret = OB_INVALID_ARGUMENT;
+  } else if (OB_FAIL(generate_flush_log_task_(flush_log_cb_ctx, group_write_buf,
+                                              flush_log_task))) {
+    PALF_LOG(ERROR, "generate owned flush log task failed", K(ret), K(flush_log_cb_ctx));
+  } else if (OB_FAIL(log_io_worker_->submit_io_task(flush_log_task))) {
+    PALF_LOG(WARN, "submit owned io task failed", K(ret));
+    int tmp_ret = flush_log_task->move_group_write_buf_to(group_write_buf);
+    if (OB_SUCCESS != tmp_ret) {
+      PALF_LOG(ERROR, "restore owned group after enqueue failure failed", K(tmp_ret));
+    }
+  }
+  if (OB_FAIL(ret) && OB_NOT_NULL(flush_log_task)) {
+    alloc_mgr_->free_log_io_flush_log_task(flush_log_task);
+    flush_log_task = NULL;
+  }
+  return ret;
+}
+
 int LogEngine::submit_handle_submit_task()
 {
   int ret = OB_SUCCESS;
@@ -942,6 +968,32 @@ int LogEngine::generate_flush_log_task_(const FlushLogCbCtx &flush_log_cb_ctx,
   return ret;
 }
 
+int LogEngine::generate_flush_log_task_(const FlushLogCbCtx &flush_log_cb_ctx,
+                                        LogGroupWriteBuf &group_write_buf,
+                                        LogIOFlushLogTask *&flush_log_task)
+{
+  int ret = OB_SUCCESS;
+  flush_log_task = NULL;
+  if (false == flush_log_cb_ctx.is_valid() || !group_write_buf.is_valid()) {
+    ret = OB_INVALID_ARGUMENT;
+  } else if (NULL == (flush_log_task = alloc_mgr_->alloc_log_io_flush_log_task(palf_epoch_))) {
+    ret = OB_ALLOCATE_MEMORY_FAILED;
+  } else if (OB_FAIL(flush_log_task->init(flush_log_cb_ctx, group_write_buf))) {
+    PALF_LOG(ERROR, "init owned LogIOFlushLogTask failed", K(ret));
+    if (!group_write_buf.is_valid()) {
+      int tmp_ret = flush_log_task->move_group_write_buf_to(group_write_buf);
+      if (OB_SUCCESS != tmp_ret) {
+        PALF_LOG(ERROR, "restore group after task init failure failed", K(tmp_ret));
+      }
+    }
+  }
+  if (OB_FAIL(ret) && NULL != flush_log_task) {
+    alloc_mgr_->free_log_io_flush_log_task(flush_log_task);
+    flush_log_task = NULL;
+  }
+  return ret;
+}
+
 int LogEngine::generate_handle_submit_task_(LogHandleSubmitTask *&handle_submit_task)
 {
   int ret = OB_SUCCESS;

src/logservice/palf/log_engine.h

@@ -47,6 +47,7 @@ class FlushLogCbCtx;
 class FlushMetaCbCtx;
 class TruncatePrefixBlocksCbCtx;
 class LogWriteBuf;
+class LogGroupWriteBuf;
 class LogGroupEntryHeader;
 class TruncatePrefixBlocksCbCtx;
 class LogIOTruncatePrefixBlocksTask;
@@ -98,6 +99,8 @@ class LogEngine
                             const int64_t buf_len);
 
   virtual int submit_flush_log_task(const FlushLogCbCtx &flush_log_cb_ctx, const LogWriteBuf &write_buf);
+  virtual int submit_flush_log_task(const FlushLogCbCtx &flush_log_cb_ctx,
+                                    LogGroupWriteBuf &group_write_buf);
   virtual int submit_handle_submit_task();
 
   int submit_flush_snapshot_meta_task(const FlushMetaCbCtx &flush_meta_cb_ctx,
@@ -161,6 +164,9 @@ class LogEngine
   int generate_flush_log_task_(const FlushLogCbCtx &flush_log_cb_ctx,
                                const LogWriteBuf &write_buf,
                                LogIOFlushLogTask *&flush_log_task);
+  int generate_flush_log_task_(const FlushLogCbCtx &flush_log_cb_ctx,
+                               LogGroupWriteBuf &group_write_buf,
+                               LogIOFlushLogTask *&flush_log_task);
   int generate_handle_submit_task_(LogHandleSubmitTask *&handle_submit_task);
   int generate_truncate_prefix_blocks_task_(
       const TruncatePrefixBlocksCbCtx &truncate_prefix_blocks_ctx,