fix random timeout issue for mpp execution (#1307) (#1318)

pingcap · Dec 29, 2020 · f4700d1 · f4700d1
1 parent f2761fe
commit f4700d1
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 3 deletions.
diff --git a/dbms/src/DataStreams/CreatingSetsBlockInputStream.cpp b/dbms/src/DataStreams/CreatingSetsBlockInputStream.cpp
@@ -189,13 +189,14 @@ void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
         }
     }
 
+    if (join_build_thread_pool != nullptr)
+        join_build_thread_pool->wait();
+
     if (subquery.join)
         subquery.join->setFinishBuildTable(true);
 
     if (table_out)
         table_out->writeSuffix();
-    if (join_build_thread_pool != nullptr)
-        join_build_thread_pool->wait();
 
     watch.stop();
 

diff --git a/dbms/src/Flash/Mpp/ExchangeReceiver.cpp b/dbms/src/Flash/Mpp/ExchangeReceiver.cpp
@@ -89,6 +89,7 @@ void ExchangeReceiver::ReadLoop(const String & meta_raw)
         meet_error = true;
         err = Exception("fatal error");
     }
+    std::lock_guard<std::mutex> lock(mu);
     live_connections--;
     cv.notify_all();
     LOG_DEBUG(log, "read thread end!!! live connections: " << std::to_string(live_connections));

diff --git a/dbms/src/Flash/Mpp/ExchangeReceiver.h b/dbms/src/Flash/Mpp/ExchangeReceiver.h
@@ -40,7 +40,7 @@ class ExchangeReceiver
     std::mutex mu;
     std::condition_variable cv;
     std::queue<Block> block_buffer;
-    std::atomic_int live_connections;
+    Int32 live_connections;
     bool inited;
     bool meet_error;
     Exception err;