Merge pull request #47245 from ClickHouse/fix_deadlock_with_transactions

Release shared ptrs after finishing a transaction

src/Interpreters/MergeTreeTransaction.cpp

@@ -315,6 +315,15 @@ bool MergeTreeTransaction::rollback() noexcept
     return true;
 }
 
+void MergeTreeTransaction::afterFinalize()
+{
+    std::lock_guard lock{mutex};
+    /// Release shared pointers just in case
+    storages.clear();
+    mutations.clear();
+    finalized = true;
+}
+
 void MergeTreeTransaction::onException()
 {
     TransactionLog::instance().rollbackTransaction(shared_from_this());
@@ -331,6 +340,11 @@ String MergeTreeTransaction::dumpDescription() const
     }
 
     std::lock_guard lock{mutex};
+    if (finalized)
+    {
+        res += ", cannot dump detailed description, transaction is finalized";
+        return res;
+    }
 
     res += fmt::format(", affects {} tables:", storages.size());
 

src/Interpreters/MergeTreeTransaction.h

@@ -65,6 +65,8 @@ private:
     scope_guard beforeCommit();
     void afterCommit(CSN assigned_csn) noexcept;
     bool rollback() noexcept;
+    void afterFinalize();
+
     void checkIsNotCancelled() const;
 
     mutable std::mutex mutex;
@@ -74,6 +76,8 @@ private:
     std::atomic<CSN> snapshot;
     const std::list<CSN>::iterator snapshot_in_use_it;
 
+    bool finalized TSA_GUARDED_BY(mutex) = false;
+
     /// Lists of changes made by transaction
     std::unordered_set<StoragePtr> storages TSA_GUARDED_BY(mutex);
     DataPartsVector creating_parts TSA_GUARDED_BY(mutex);

src/Interpreters/TransactionLog.cpp

@@ -350,7 +350,7 @@ void TransactionLog::tryFinalizeUnknownStateTransactions()
         /// CSNs must be already loaded, only need to check if the corresponding mapping exists.
         if (auto csn = getCSN(txn->tid))
         {
-            finalizeCommittedTransaction(txn, csn, state_guard);
+            finalizeCommittedTransaction(txn.get(), csn, state_guard);
         }
         else
         {
@@ -431,7 +431,7 @@ CSN TransactionLog::commitTransaction(const MergeTreeTransactionPtr & txn, bool
             /// The only thing we can do is to postpone its finalization.
             {
                 std::lock_guard lock{running_list_mutex};
-                unknown_state_list.emplace_back(txn.get(), std::move(state_guard));
+                unknown_state_list.emplace_back(txn, std::move(state_guard));
             }
             log_updated_event->set();
             if (throw_on_unknown_status)
@@ -487,6 +487,7 @@ CSN TransactionLog::finalizeCommittedTransaction(MergeTreeTransaction * txn, CSN
         }
     }
 
+    txn->afterFinalize();
     return allocated_csn;
 }
 
@@ -523,6 +524,7 @@ void TransactionLog::rollbackTransaction(const MergeTreeTransactionPtr & txn) no
     }
 
     tryWriteEventToSystemLog(log, global_context, TransactionsInfoLogElement::ROLLBACK, txn->tid);
+    txn->afterFinalize();
 }
 
 MergeTreeTransactionPtr TransactionLog::tryGetRunningTransaction(const TIDHash & tid)

src/Interpreters/TransactionLog.h

@@ -177,7 +177,7 @@ private:
     /// Transactions that are currently processed
     TransactionsList running_list TSA_GUARDED_BY(running_list_mutex);
     /// If we lost connection on attempt to create csn- node then we don't know transaction's state.
-    using UnknownStateList = std::vector<std::pair<MergeTreeTransaction *, scope_guard>>;
+    using UnknownStateList = std::vector<std::pair<MergeTreeTransactionPtr, scope_guard>>;
     UnknownStateList unknown_state_list TSA_GUARDED_BY(running_list_mutex);
     UnknownStateList unknown_state_list_loaded TSA_GUARDED_BY(running_list_mutex);
     /// Ordered list of snapshots that are currently used by some transactions. Needed for background cleanup.