Backport #47953 to 23.1: Fix tsan error lock-order-inversion

src/Interpreters/ProcessList.cpp

@@ -368,7 +368,7 @@ QueryStatus::~QueryStatus()
 {
 #if !defined(NDEBUG)
     /// Check that all executors were invalidated.
-    for (const auto & e : executors)
+    for (const auto & [_, e] : executors)
         assert(!e->executor);
 #endif
 
@@ -406,7 +406,9 @@ CancellationCode QueryStatus::cancelQuery(bool)
     {
         /// Create a snapshot of executors under a mutex.
         std::lock_guard lock(executors_mutex);
-        executors_snapshot = executors;
+        executors_snapshot.reserve(executors.size());
+        for (const auto & [_, e] : executors)
+            executors_snapshot.push_back(e);
     }
 
     /// We should call cancel() for each executor with unlocked executors_mutex, because
@@ -434,17 +436,24 @@ void QueryStatus::addPipelineExecutor(PipelineExecutor * e)
         throw Exception(ErrorCodes::QUERY_WAS_CANCELLED, "Query was cancelled");
 
     std::lock_guard lock(executors_mutex);
-    assert(std::find_if(executors.begin(), executors.end(), [e](const ExecutorHolderPtr & x){ return x->executor == e; }) == executors.end());
-    executors.push_back(std::make_shared<ExecutorHolder>(e));
+    assert(!executors.contains(e));
+    executors[e] = std::make_shared<ExecutorHolder>(e);
 }
 
 void QueryStatus::removePipelineExecutor(PipelineExecutor * e)
 {
-    std::lock_guard lock(executors_mutex);
-    auto it = std::find_if(executors.begin(), executors.end(), [e](const ExecutorHolderPtr & x){ return x->executor == e; });
-    assert(it != executors.end());
-    /// Invalidate executor pointer inside holder, but don't remove holder from the executors (to avoid race with cancelQuery)
-    (*it)->remove();
+    ExecutorHolderPtr executor_holder;
+
+    {
+        std::lock_guard lock(executors_mutex);
+        assert(executors.contains(e));
+        executor_holder = executors[e];
+        executors.erase(e);
+    }
+
+    /// Invalidate executor pointer inside holder.
+    /// We should do it with released executors_mutex to avoid possible lock order inversion.
+    executor_holder->remove();
 }
 
 bool QueryStatus::checkTimeLimit()

src/Interpreters/ProcessList.h

@@ -132,8 +132,8 @@ class QueryStatus : public WithContext
 
     using ExecutorHolderPtr = std::shared_ptr<ExecutorHolder>;
 
-    /// Array of PipelineExecutors to be cancelled when a cancelQuery is received
-    std::vector<ExecutorHolderPtr> executors;
+    /// Container of PipelineExecutors to be cancelled when a cancelQuery is received
+    std::unordered_map<PipelineExecutor *, ExecutorHolderPtr> executors;
 
     enum QueryStreamsStatus
     {