[iotdb] branch rel/1.1 updated: [To rel/1.1]Remove unecessary synchronized on decrementCnt of SinkListener

[jackietien]

This is an automated email from the ASF dual-hosted git repository.

jackietien pushed a commit to branch rel/1.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/rel/1.1 by this push:
     new dedc9087eb [To rel/1.1]Remove unecessary synchronized on decrementCnt 
of SinkListener
dedc9087eb is described below

commit dedc9087eb8812cddb63d87530a49ede69181bc7
Author: Liao Lanyu <1435078...@qq.com>
AuthorDate: Tue Apr 25 11:22:39 2023 +0800

    [To rel/1.1]Remove unecessary synchronized on decrementCnt of SinkListener
---
 .../iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java   | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git 
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java
 
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java
index 67f8bda606..1e25adb71c 100644
--- 
a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java
+++ 
b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/MPPDataExchangeManager.java
@@ -63,6 +63,7 @@ import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -401,7 +402,7 @@ public class MPPDataExchangeManager implements 
IMPPDataExchangeManager {
 
     private final AtomicInteger cnt;
 
-    private volatile boolean hasDecremented = false;
+    private final AtomicBoolean hasDecremented = new AtomicBoolean(false);
 
     public ISinkChannelListenerImpl(
         TFragmentInstanceId localFragmentInstanceId,
@@ -441,9 +442,8 @@ public class MPPDataExchangeManager implements 
IMPPDataExchangeManager {
       }
     }
 
-    private synchronized void decrementCnt() {
-      if (!hasDecremented) {
-        hasDecremented = true;
+    private void decrementCnt() {
+      if (hasDecremented.compareAndSet(false, true)) {
         if (cnt.decrementAndGet() == 0) {
           closeShuffleSinkHandle();
         }