[BUG] add callback logging blocked bug detect

mittest/simple_server/test_ob_minor_freeze.cpp

@@ -359,8 +359,15 @@ void ObMinorFreezeTest::logstream_freeze()
   const int64_t start = ObTimeUtility::current_time();
   while (ObTimeUtility::current_time() - start <= freeze_duration_) {
     for (int j = 0; j < OB_DEFAULT_TABLE_COUNT; ++j) {
+      int ret = OB_EAGAIN;
+      while (OB_EAGAIN == ret) {
+        ret = ls_handles_.at(j).get_ls()->logstream_freeze((i % 2 == 0) ? true : false);
 
-      ASSERT_EQ(OB_SUCCESS, ls_handles_.at(j).get_ls()->logstream_freeze((i % 2 == 0) ? true : false));
+        if (OB_EAGAIN == ret) {
+          ob_usleep(rand() % SLEEP_TIME);
+        }
+      }
+      ASSERT_EQ(OB_SUCCESS, ret);
     }
     i = i + 1;
   }
@@ -377,7 +384,14 @@ void ObMinorFreezeTest::tablet_freeze()
   const int64_t start = ObTimeUtility::current_time();
   while (ObTimeUtility::current_time() - start <= freeze_duration_) {
     for (int j = 0; j < OB_DEFAULT_TABLE_COUNT; ++j) {
-      ASSERT_EQ(OB_SUCCESS, ls_handles_.at(j).get_ls()->tablet_freeze(tablet_ids_.at(j), (i % 2 == 0) ? true : false));
+      int ret = OB_EAGAIN;
+      while (OB_EAGAIN == ret) {
+        ret = ls_handles_.at(j).get_ls()->tablet_freeze(tablet_ids_.at(j), (i % 2 == 0) ? true : false);
+        if (OB_EAGAIN == ret) {
+          ob_usleep(rand() % SLEEP_TIME);
+        }
+      }
+      ASSERT_EQ(OB_SUCCESS, ret);
     }
     i = i + 1;
   }
@@ -424,6 +438,7 @@ void ObMinorFreezeTest::insert_and_freeze()
 {
   std::thread tenant_freeze_thread([this]() { tenant_freeze(); });
   std::thread tablet_freeze_thread([this]() { tablet_freeze(); });
+  std::thread logstream_freeze_thread([this]() { logstream_freeze(); });
   std::thread tablet_freeze_for_replace_tablet_meta_thread([this]() { tablet_freeze_for_replace_tablet_meta(); });
   std::thread check_frozen_memtable_thread([this]() { check_frozen_memtable(); });
   std::thread batch_tablet_freeze_thread([this]() { batch_tablet_freeze(); });
@@ -435,6 +450,7 @@ void ObMinorFreezeTest::insert_and_freeze()
 
   tenant_freeze_thread.join();
   tablet_freeze_thread.join();
+  logstream_freeze_thread.join();
   tablet_freeze_for_replace_tablet_meta_thread.join();
   check_frozen_memtable_thread.join();
   batch_tablet_freeze_thread.join();

src/storage/memtable/ob_redo_log_generator.cpp

@@ -35,6 +35,7 @@ void ObRedoLogGenerator::reset()
   generate_cursor_.reset();
   callback_mgr_ = nullptr;
   mem_ctx_ = NULL;
+  last_logging_blocked_time_ = 0;
   if (clog_encrypt_meta_ != NULL) {
     op_free(clog_encrypt_meta_);
     clog_encrypt_meta_ = NULL;
@@ -58,6 +59,7 @@ int ObRedoLogGenerator::set(ObTransCallbackMgr *mgr, ObIMemtableCtx *mem_ctx)
   generate_cursor_ = mgr->begin();
   callback_mgr_ = mgr;
   mem_ctx_ = mem_ctx;
+  last_logging_blocked_time_ = 0;
   is_inited_ = true;
 
   return ret;
@@ -105,8 +107,20 @@ int ObRedoLogGenerator::fill_redo_log(char *buf,
           // Becasue the first callback is linked to a logging_blocked memtable
           transaction::ObPartTransCtx *part_ctx = static_cast<transaction::ObPartTransCtx *>(mem_ctx_->get_trans_ctx());
           part_ctx->set_block_frozen_memtable(static_cast<memtable::ObMemtable *>(iter->get_memtable()));
+
+          int64_t current_time = ObTimeUtility::current_time();
+          if (last_logging_blocked_time_ == 0) {
+            last_logging_blocked_time_ = current_time;
+          } else if (current_time - last_logging_blocked_time_ > 5 * 1_min) {
+            TRANS_LOG(WARN, "logging block cost too much time", KPC(part_ctx), KPC(iter));
+            if (REACH_TENANT_TIME_INTERVAL(1_min)) {
+              bug_detect_for_logging_blocked_();
+            }
+          }
         }
       } else {
+        last_logging_blocked_time_ = 0;
+
         bool fake_fill = false;
         if (MutatorType::MUTATOR_ROW == iter->get_mutator_type()) {
           ret = fill_row_redo(cursor, mmw, redo, log_for_lock_node, fake_fill, encrypt_info);
@@ -397,6 +411,22 @@ bool ObRedoLogGenerator::check_dup_tablet_(const ObITransCallback *callback_ptr)
   return is_dup_tablet;
 }
 
+void ObRedoLogGenerator::bug_detect_for_logging_blocked_()
+{
+  int ret = OB_SUCCESS;
+
+  ObITransCallbackIterator bug_detect_cursor;
+  int64_t count = 0;
+  for (bug_detect_cursor = generate_cursor_ + 1;
+       callback_mgr_->end() !=bug_detect_cursor
+         && count <= 5;
+       ++bug_detect_cursor) {
+    ObITransCallback *bug_detect_iter = (ObITransCallback *)*bug_detect_cursor;
+    count++;
+    TRANS_LOG(WARN, "logging block print callback", KPC(bug_detect_iter), K(count));
+  }
+}
+
 }; // end namespace memtable
 }; // end namespace oceanbase
 

src/storage/memtable/ob_redo_log_generator.h

@@ -101,6 +101,7 @@ private:
                            const bool log_for_lock_node,
                            bool &fake_fill);
   bool check_dup_tablet_(const ObITransCallback * callback_ptr) const;
+  void bug_detect_for_logging_blocked_();
 private:
   DISALLOW_COPY_AND_ASSIGN(ObRedoLogGenerator);
   bool is_inited_;
@@ -111,6 +112,9 @@ private:
   ObTransCallbackMgr *callback_mgr_;
   ObIMemtableCtx *mem_ctx_;
   transaction::ObTxEncryptMeta *clog_encrypt_meta_;
+
+  // logging block bug detector
+  int64_t last_logging_blocked_time_;
 };
 }; // end namespace memtable
 }; // end namespace oceanbase