[ISSUE #2883] [Part B] Improve produce performance in M/S mode. (#2885)

* Optimise lock in WaitNotifyObject

* Remove lock in HAService

* Remove lock in GroupCommitService

store/src/main/java/org/apache/rocketmq/store/CommitLog.java

@@ -19,8 +19,8 @@ package org.apache.rocketmq.store;
 import java.net.Inet6Address;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -1403,48 +1403,55 @@ public class CommitLog {
      * GroupCommit Service
      */
     class GroupCommitService extends FlushCommitLogService {
-        private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>();
-        private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>();
+        private volatile LinkedList<GroupCommitRequest> requestsWrite = new LinkedList<GroupCommitRequest>();
+        private volatile LinkedList<GroupCommitRequest> requestsRead = new LinkedList<GroupCommitRequest>();
+        private final PutMessageSpinLock lock = new PutMessageSpinLock();
 
         public synchronized void putRequest(final GroupCommitRequest request) {
-            synchronized (this.requestsWrite) {
+            lock.lock();
+            try {
                 this.requestsWrite.add(request);
+            } finally {
+                lock.unlock();
             }
             this.wakeup();
         }
 
         private void swapRequests() {
-            List<GroupCommitRequest> tmp = this.requestsWrite;
-            this.requestsWrite = this.requestsRead;
-            this.requestsRead = tmp;
+            lock.lock();
+            try {
+                LinkedList<GroupCommitRequest> tmp = this.requestsWrite;
+                this.requestsWrite = this.requestsRead;
+                this.requestsRead = tmp;
+            } finally {
+                lock.unlock();
+            }
         }
 
         private void doCommit() {
-            synchronized (this.requestsRead) {
-                if (!this.requestsRead.isEmpty()) {
-                    for (GroupCommitRequest req : this.requestsRead) {
-                        // There may be a message in the next file, so a maximum of
-                        // two times the flush
-                        boolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
-                        for (int i = 0; i < 2 && !flushOK; i++) {
-                            CommitLog.this.mappedFileQueue.flush(0);
-                            flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
-                        }
-
-                        req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);
+            if (!this.requestsRead.isEmpty()) {
+                for (GroupCommitRequest req : this.requestsRead) {
+                    // There may be a message in the next file, so a maximum of
+                    // two times the flush
+                    boolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
+                    for (int i = 0; i < 2 && !flushOK; i++) {
+                        CommitLog.this.mappedFileQueue.flush(0);
+                        flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
                     }
 
-                    long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
-                    if (storeTimestamp > 0) {
-                        CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
-                    }
+                    req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);
+                }
 
-                    this.requestsRead.clear();
-                } else {
-                    // Because of individual messages is set to not sync flush, it
-                    // will come to this process
-                    CommitLog.this.mappedFileQueue.flush(0);
+                long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
+                if (storeTimestamp > 0) {
+                    CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
                 }
+
+                this.requestsRead = new LinkedList<>();
+            } else {
+                // Because of individual messages is set to not sync flush, it
+                // will come to this process
+                CommitLog.this.mappedFileQueue.flush(0);
             }
         }
 

store/src/main/java/org/apache/rocketmq/store/ha/HAService.java

@@ -25,7 +25,6 @@ import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
-import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;
@@ -39,6 +38,7 @@ import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.remoting.common.RemotingUtil;
 import org.apache.rocketmq.store.CommitLog;
 import org.apache.rocketmq.store.DefaultMessageStore;
+import org.apache.rocketmq.store.PutMessageSpinLock;
 import org.apache.rocketmq.store.PutMessageStatus;
 
 public class HAService {
@@ -254,12 +254,16 @@ public class HAService {
     class GroupTransferService extends ServiceThread {
 
         private final WaitNotifyObject notifyTransferObject = new WaitNotifyObject();
-        private volatile List<CommitLog.GroupCommitRequest> requestsWrite = new ArrayList<>();
-        private volatile List<CommitLog.GroupCommitRequest> requestsRead = new ArrayList<>();
+        private final PutMessageSpinLock lock = new PutMessageSpinLock();
+        private volatile LinkedList<CommitLog.GroupCommitRequest> requestsWrite = new LinkedList<>();
+        private volatile LinkedList<CommitLog.GroupCommitRequest> requestsRead = new LinkedList<>();
 
-        public synchronized void putRequest(final CommitLog.GroupCommitRequest request) {
-            synchronized (this.requestsWrite) {
+        public void putRequest(final CommitLog.GroupCommitRequest request) {
+            lock.lock();
+            try {
                 this.requestsWrite.add(request);
+            } finally {
+                lock.unlock();
             }
             this.wakeup();
         }
@@ -269,32 +273,35 @@ public class HAService {
         }
 
         private void swapRequests() {
-            List<CommitLog.GroupCommitRequest> tmp = this.requestsWrite;
-            this.requestsWrite = this.requestsRead;
-            this.requestsRead = tmp;
+            lock.lock();
+            try {
+                LinkedList<CommitLog.GroupCommitRequest> tmp = this.requestsWrite;
+                this.requestsWrite = this.requestsRead;
+                this.requestsRead = tmp;
+            } finally {
+                lock.unlock();
+            }
         }
 
         private void doWaitTransfer() {
-            synchronized (this.requestsRead) {
-                if (!this.requestsRead.isEmpty()) {
-                    for (CommitLog.GroupCommitRequest req : this.requestsRead) {
-                        boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
-                        long waitUntilWhen = HAService.this.defaultMessageStore.getSystemClock().now()
+            if (!this.requestsRead.isEmpty()) {
+                for (CommitLog.GroupCommitRequest req : this.requestsRead) {
+                    boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
+                    long waitUntilWhen = HAService.this.defaultMessageStore.getSystemClock().now()
                             + HAService.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout();
-                        while (!transferOK && HAService.this.defaultMessageStore.getSystemClock().now() < waitUntilWhen) {
-                            this.notifyTransferObject.waitForRunning(1000);
-                            transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
-                        }
-
-                        if (!transferOK) {
-                            log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
-                        }
+                    while (!transferOK && HAService.this.defaultMessageStore.getSystemClock().now() < waitUntilWhen) {
+                        this.notifyTransferObject.waitForRunning(1000);
+                        transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
+                    }
 
-                        req.wakeupCustomer(transferOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
+                    if (!transferOK) {
+                        log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
                     }
 
-                    this.requestsRead.clear();
+                    req.wakeupCustomer(transferOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
                 }
+
+                this.requestsRead = new LinkedList<>();
             }
         }
 

store/src/main/java/org/apache/rocketmq/store/ha/WaitNotifyObject.java

@@ -20,40 +20,43 @@ import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class WaitNotifyObject {
     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
 
-    protected final HashMap<Long/* thread id */, Boolean/* notified */> waitingThreadTable =
-        new HashMap<Long, Boolean>(16);
+    protected final ConcurrentHashMap<Long/* thread id */, AtomicBoolean/* notified */> waitingThreadTable =
+        new ConcurrentHashMap<Long, AtomicBoolean>(16);
 
-    protected volatile boolean hasNotified = false;
+    protected AtomicBoolean hasNotified = new AtomicBoolean(false);
 
     public void wakeup() {
-        synchronized (this) {
-            if (!this.hasNotified) {
-                this.hasNotified = true;
+        boolean needNotify = hasNotified.compareAndSet(false, true);
+        if (needNotify) {
+            synchronized (this) {
                 this.notify();
             }
         }
     }
 
     protected void waitForRunning(long interval) {
+        if (this.hasNotified.compareAndSet(true, false)) {
+            this.onWaitEnd();
+            return;
+        }
         synchronized (this) {
-            if (this.hasNotified) {
-                this.hasNotified = false;
-                this.onWaitEnd();
-                return;
-            }
-
             try {
+                if (this.hasNotified.compareAndSet(true, false)) {
+                    this.onWaitEnd();
+                    return;
+                }
                 this.wait(interval);
             } catch (InterruptedException e) {
                 log.error("Interrupted", e);
             } finally {
-                this.hasNotified = false;
+                this.hasNotified.set(false);
                 this.onWaitEnd();
             }
         }
@@ -63,15 +66,14 @@ public class WaitNotifyObject {
     }
 
     public void wakeupAll() {
-        synchronized (this) {
-            boolean needNotify = false;
-
-            for (Map.Entry<Long,Boolean> entry : this.waitingThreadTable.entrySet()) {
-                needNotify = needNotify || !entry.getValue();
-                entry.setValue(true);
+        boolean needNotify = false;
+        for (Map.Entry<Long,AtomicBoolean> entry : this.waitingThreadTable.entrySet()) {
+            if (entry.getValue().compareAndSet(false, true)) {
+                needNotify = true;
             }
-
-            if (needNotify) {
+        }
+        if (needNotify) {
+            synchronized (this) {
                 this.notifyAll();
             }
         }
@@ -79,20 +81,22 @@ public class WaitNotifyObject {
 
     public void allWaitForRunning(long interval) {
         long currentThreadId = Thread.currentThread().getId();
+        AtomicBoolean notified = this.waitingThreadTable.computeIfAbsent(currentThreadId, k -> new AtomicBoolean(false));
+        if (notified.compareAndSet(true, false)) {
+            this.onWaitEnd();
+            return;
+        }
         synchronized (this) {
-            Boolean notified = this.waitingThreadTable.get(currentThreadId);
-            if (notified != null && notified) {
-                this.waitingThreadTable.put(currentThreadId, false);
-                this.onWaitEnd();
-                return;
-            }
-
             try {
+                if (notified.compareAndSet(true, false)) {
+                    this.onWaitEnd();
+                    return;
+                }
                 this.wait(interval);
             } catch (InterruptedException e) {
                 log.error("Interrupted", e);
             } finally {
-                this.waitingThreadTable.put(currentThreadId, false);
+                notified.set(false);
                 this.onWaitEnd();
             }
         }