Fix race in function mergeToViaEmplace (#2617)

pingcap · Aug 9, 2021 · 9776ea6 · 9776ea6
1 parent 1376786
commit 9776ea6
Showing 1 changed file with 5 additions and 5 deletions.
diff --git a/dbms/src/Interpreters/Aggregator.cpp b/dbms/src/Interpreters/Aggregator.cpp
@@ -1572,8 +1572,8 @@ class MergingAndConvertingBlockInputStream : public IProfilingBlockInputStream
     bool final;
     size_t threads;
 
-    Int32 current_bucket_num = -1;
-    Int32 max_scheduled_bucket_num = -1;
+    std::atomic<Int32> current_bucket_num = -1;
+    std::atomic<Int32> max_scheduled_bucket_num = -1;
     static constexpr Int32 NUM_BUCKETS = 256;
 
     struct ParallelMergeData
@@ -1591,12 +1591,12 @@ class MergingAndConvertingBlockInputStream : public IProfilingBlockInputStream
 
     void scheduleThreadForNextBucket()
     {
-        ++max_scheduled_bucket_num;
-        if (max_scheduled_bucket_num >= NUM_BUCKETS)
+        int num = max_scheduled_bucket_num.fetch_add(1) + 1;
+        if (num >= NUM_BUCKETS)
             return;
 
         parallel_merge_data->pool.schedule(
-            ThreadFactory(true, "MergingAggregtd").newJob([this]{ thread(max_scheduled_bucket_num); }));
+            ThreadFactory(true, "MergingAggregtd").newJob([this, num]{ thread(num); }));
     }
 
     void thread(Int32 bucket_num)