Use txn but not WAL entry in WAL flush thread (#2469)

### What problem does this PR solve?

Use txn but not WAL entry in WAL flush thread

### Type of change

- [x] Refactoring

Signed-off-by: Jin Hai <[email protected]>

Author: JinHai-CN

src/storage/txn/txn_manager.cpp

@@ -194,27 +194,27 @@ void TxnManager::SendToWAL(Txn *txn) {
 
     std::lock_guard guard(locker_);
     if (wait_conflict_ck_.empty()) {
-        String error_message = fmt::format("WalManager::PutEntry wait_conflict_ck_ is empty, txn->CommitTS() {}", txn->CommitTS());
+        String error_message = fmt::format("TxnManager::SendToWAL wait_conflict_ck_ is empty, txn->CommitTS() {}", txn->CommitTS());
         UnrecoverableError(error_message);
     }
     if (wait_conflict_ck_.begin()->first > commit_ts) {
-        String error_message = fmt::format("WalManager::PutEntry wait_conflict_ck_.begin()->first {} > txn->CommitTS() {}",
+        String error_message = fmt::format("TxnManager::SendToWAL wait_conflict_ck_.begin()->first {} > txn->CommitTS() {}",
                                            wait_conflict_ck_.begin()->first,
                                            txn->CommitTS());
         UnrecoverableError(error_message);
     }
     if (wal_entry) {
-        wait_conflict_ck_.at(commit_ts) = wal_entry;
+        wait_conflict_ck_.at(commit_ts) = txn;
     } else {
         wait_conflict_ck_.erase(commit_ts); // rollback
     }
     if (!wait_conflict_ck_.empty() && wait_conflict_ck_.begin()->second != nullptr) {
-        Vector<WalEntry *> wal_entries;
+        Vector<Txn *> txn_array;
         do {
-            wal_entries.push_back(wait_conflict_ck_.begin()->second);
+            txn_array.push_back(wait_conflict_ck_.begin()->second);
             wait_conflict_ck_.erase(wait_conflict_ck_.begin());
         } while (!wait_conflict_ck_.empty() && wait_conflict_ck_.begin()->second != nullptr);
-        wal_mgr_->PutEntries(wal_entries);
+        wal_mgr_->SubmitTxn(txn_array);
     }
 }
 

src/storage/txn/txn_manager.cppm

@@ -123,7 +123,7 @@ private:
     Map<TxnTimeStamp, Txn *> committing_txns_; // the txns in committing stage
     Set<TxnTimeStamp> checking_ts_{};          // the begin ts of txn that is used to check conflict
 
-    Map<TxnTimeStamp, WalEntry *> wait_conflict_ck_{}; // sorted by commit ts
+    Map<TxnTimeStamp, Txn *> wait_conflict_ck_{}; // sorted by commit ts
 
     Atomic<TxnTimeStamp> current_ts_{}; // The next txn ts
     Atomic<TxnTimeStamp> max_committed_ts_{};

src/storage/wal/wal_manager.cpp

@@ -150,13 +150,11 @@ void WalManager::Stop() {
     LOG_INFO("WAL manager is stopped.");
 }
 
-// Session request to persist an entry. Assuming txn_id of the entry has
-// been initialized.
-void WalManager::PutEntries(Vector<WalEntry *> wal_entries) {
+void WalManager::SubmitTxn(Vector<Txn *> &txn_array) {
     if (!running_.load()) {
         return;
     }
-    wait_flush_.EnqueueBulk(wal_entries);
+    wait_flush_.EnqueueBulk(txn_array);
 }
 
 TxnTimeStamp WalManager::LastCheckpointTS() const {
@@ -292,24 +290,27 @@ Vector<SharedPtr<String>> WalManager::GetDiffWalEntryString(TxnTimeStamp start_t
 void WalManager::Flush() {
     LOG_TRACE("WalManager::Flush log mainloop begin");
 
-    Deque<WalEntry *> log_batch{};
+    Deque<Txn *> txn_batch{};
     TxnManager *txn_mgr = storage_->txn_manager();
     ClusterManager *cluster_manager = nullptr;
     while (running_.load()) {
-        wait_flush_.DequeueBulk(log_batch);
-        if (log_batch.empty()) {
+        wait_flush_.DequeueBulk(txn_batch);
+        if (txn_batch.empty()) {
             LOG_WARN("WalManager::Dequeue empty batch logs");
             continue;
         }
 
-        for (const auto &entry : log_batch) {
-            // Empty WalEntry (read-only transactions) shouldn't go into WalManager.
-            if (entry == nullptr) {
+        for (const auto &txn : txn_batch) {
+            if (txn == nullptr) {
                 // terminate entry
+                LOG_INFO("Terminate WAL Manager flush thread");
                 running_ = false;
                 break;
             }
 
+            WalEntry *entry = txn->GetWALEntry();
+            // Empty WalEntry (read-only transactions) shouldn't go into WalManager.
+
             if (entry->cmds_.empty()) {
                 continue;
                 // UnrecoverableError(fmt::format("WalEntry of txn_id {} commands is empty", entry->txn_id_));
@@ -371,13 +372,12 @@ void WalManager::Flush() {
             cluster_manager->SyncLogs();
         }
 
-        for (const auto &entry : log_batch) {
-            Txn *txn = txn_mgr->GetTxn(entry->txn_id_);
-            if (txn != nullptr) {
-                txn->CommitBottom();
-            }
+        // Commit bottom
+        for (const auto &txn : txn_batch) {
+            txn->CommitBottom();
+            // TODO: if txn is checkpoint, swap WAL file
         }
-        log_batch.clear();
+        txn_batch.clear();
 
         // Check if the wal file is too large, swap to a new one.
         try {

src/storage/wal/wal_manager.cppm

@@ -79,9 +79,7 @@ public:
 
     void Stop();
 
-    // Session request to persist an entry. Assuming txn_id of the entry has
-    // been initialized.
-    void PutEntries(Vector<WalEntry *> wal_entries);
+    void SubmitTxn(Vector<Txn *> &txn_batch);
 
     // Flush is scheduled regularly. It collects a batch of transactions, sync
     // wal and do parallel committing. Each sync cost ~1s. Each checkpoint cost
@@ -173,7 +171,7 @@ private:
     Thread flush_thread_{};
 
     // TxnManager and Flush thread access following members
-    BlockingQueue<WalEntry *> wait_flush_{"WalManager"};
+    BlockingQueue<Txn *> wait_flush_{"WalManager"};
 
     // Only Flush thread access following members
     std::ofstream ofs_{};