Eliminate array copy (#2886)

[Part C] Improve produce performance in M/S mode.
apache · Jul 6, 2021 · 3183122 · 3183122
1 parent d65778f
commit 3183122
Show file tree

Hide file tree

Showing 6 changed files with 14 additions and 14 deletions.
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/plugin/AbstractPluginMessageStore.java b/broker/src/main/java/org/apache/rocketmq/broker/plugin/AbstractPluginMessageStore.java
@@ -180,8 +180,8 @@ public SelectMappedBufferResult getCommitLogData(long offset) {
     }
 
     @Override
-    public boolean appendToCommitLog(long startOffset, byte[] data) {
-        return next.appendToCommitLog(startOffset, data);
+    public boolean appendToCommitLog(long startOffset, byte[] data, int dataStart, int dataLength) {
+        return next.appendToCommitLog(startOffset, data, dataStart, dataLength);
     }
 
     @Override

diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -1185,7 +1185,7 @@ public void destroy() {
         this.mappedFileQueue.destroy();
     }
 
-    public boolean appendData(long startOffset, byte[] data) {
+    public boolean appendData(long startOffset, byte[] data, int dataStart, int dataLength) {
         putMessageLock.lock();
         try {
             MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile(startOffset);
@@ -1194,7 +1194,7 @@ public boolean appendData(long startOffset, byte[] data) {
                 return false;
             }
 
-            return mappedFile.appendMessage(data);
+            return mappedFile.appendMessage(data, dataStart, dataLength);
         } finally {
             putMessageLock.unlock();
         }

diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -921,13 +921,13 @@ public SelectMappedBufferResult getCommitLogData(final long offset) {
     }
 
     @Override
-    public boolean appendToCommitLog(long startOffset, byte[] data) {
+    public boolean appendToCommitLog(long startOffset, byte[] data, int dataStart, int dataLength) {
         if (this.shutdown) {
             log.warn("message store has shutdown, so appendToPhyQueue is forbidden");
             return false;
         }
 
-        boolean result = this.commitLog.appendData(startOffset, data);
+        boolean result = this.commitLog.appendData(startOffset, data, dataStart, dataLength);
         if (result) {
             this.reputMessageService.wakeup();
         } else {

diff --git a/store/src/main/java/org/apache/rocketmq/store/MessageStore.java b/store/src/main/java/org/apache/rocketmq/store/MessageStore.java
@@ -245,9 +245,11 @@ GetMessageResult getMessage(final String group, final String topic, final int qu
      *
      * @param startOffset starting offset.
      * @param data data to append.
+     * @param dataStart the start index of data array
+     * @param dataLength the length of data array
      * @return true if success; false otherwise.
      */
-    boolean appendToCommitLog(final long startOffset, final byte[] data);
+    boolean appendToCommitLog(final long startOffset, final byte[] data, int dataStart, int dataLength);
 
     /**
      * Execute file deletion manually.

diff --git a/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java b/store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java
@@ -905,7 +905,7 @@ public void destroy() {
     }
 
     @Override
-    public boolean appendData(long startOffset, byte[] data) {
+    public boolean appendData(long startOffset, byte[] data, int dataStart, int dataLength) {
         //the old ha service will invoke method, here to prevent it
         return false;
     }

diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
@@ -440,7 +440,6 @@ private boolean processReadEvent() {
 
         private boolean dispatchReadRequest() {
             final int msgHeaderSize = 8 + 4; // phyoffset + size
-            int readSocketPos = this.byteBufferRead.position();
 
             while (true) {
                 int diff = this.byteBufferRead.position() - this.dispatchPosition;
@@ -459,13 +458,12 @@ private boolean dispatchReadRequest() {
                     }
 
                     if (diff >= (msgHeaderSize + bodySize)) {
-                        byte[] bodyData = new byte[bodySize];
-                        this.byteBufferRead.position(this.dispatchPosition + msgHeaderSize);
-                        this.byteBufferRead.get(bodyData);
+                        byte[] bodyData = byteBufferRead.array();
+                        int dataStart = this.dispatchPosition + msgHeaderSize;
 
-                        HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData);
+                        HAService.this.defaultMessageStore.appendToCommitLog(
+                                masterPhyOffset, bodyData, dataStart, bodySize);
 
-                        this.byteBufferRead.position(readSocketPos);
                         this.dispatchPosition += msgHeaderSize + bodySize;
 
                         if (!reportSlaveMaxOffsetPlus()) {