You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2018/07/17 08:08:01 UTC
[GitHub] lizhanhui closed pull request #195: [ROCKETMQ-317] Take advantage
of Linux native libaio
lizhanhui closed pull request #195: [ROCKETMQ-317] Take advantage of Linux native libaio
URL: https://github.com/apache/rocketmq/pull/195
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java
index a46cbff2e..49f054624 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java
@@ -21,8 +21,12 @@
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.FileRegion;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.activemq.artemis.jlibaio.SubmitInfo;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.filter.ConsumerFilterData;
@@ -57,10 +61,13 @@
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.netty.RequestTask;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.store.AccessDiskStrategy;
import org.apache.rocketmq.store.GetMessageResult;
+import org.apache.rocketmq.store.MappedFile;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.MessageFilter;
import org.apache.rocketmq.store.PutMessageResult;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.slf4j.Logger;
@@ -375,14 +382,15 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
getMessageResult.getBufferTotalSize());
this.brokerController.getBrokerStatsManager().incBrokerGetNums(getMessageResult.getMessageCount());
- if (this.brokerController.getBrokerConfig().isTransferMsgByHeap()) {
+ if (this.brokerController.getBrokerConfig().isTransferMsgByHeap()
+ || AccessDiskStrategy.HEAP == getMessageResult.getAccessDiskStrategy()) {
final long beginTimeMills = this.brokerController.getMessageStore().now();
final byte[] r = this.readGetMessageResult(getMessageResult, requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId());
this.brokerController.getBrokerStatsManager().incGroupGetLatency(requestHeader.getConsumerGroup(),
requestHeader.getTopic(), requestHeader.getQueueId(),
(int) (this.brokerController.getMessageStore().now() - beginTimeMills));
response.setBody(r);
- } else {
+ } else if (getMessageResult.getAccessDiskStrategy() == AccessDiskStrategy.MMAP) {
try {
FileRegion fileRegion =
new ManyMessageTransfer(response.encodeHeader(getMessageResult.getBufferTotalSize()), getMessageResult);
@@ -400,6 +408,67 @@ public void operationComplete(ChannelFuture future) throws Exception {
getMessageResult.release();
}
+ response = null;
+ } else {
+ final List<SelectMappedBufferResult> selectMappedBufferResults = getMessageResult.getMessageMappedList();
+ final FileRegion fileRegion = new ManyMessageTransfer(response.encodeHeader(getMessageResult.getBufferTotalSize()), getMessageResult);
+ final int opaque = request.getOpaque();
+
+ // IO completion callback.
+ SubmitInfo submitInfo = new SubmitInfo() {
+ private AtomicInteger count = new AtomicInteger(selectMappedBufferResults.size());
+
+ private AtomicBoolean failed = new AtomicBoolean(false);
+
+ @Override
+ public void onError(int errno, String message) {
+ log.error("AIO error. Error Number: {}, Msg: {}", errno, message);
+ failed.set(true);
+ count.decrementAndGet();
+ }
+
+ @Override
+ public void done() {
+ count.decrementAndGet();
+
+ if (count.get() == 0) {
+ if (!failed.get()) {
+ try {
+ channel.writeAndFlush(fileRegion).addListener(new ChannelFutureListener() {
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ getMessageResult.release();
+ if (!future.isSuccess()) {
+ log.error("transfer many message by pagecache failed, {}", channel.remoteAddress(), future.cause());
+ }
+ }
+ });
+ } catch (Throwable e) {
+ log.error("transfer many message by pagecache exception", e);
+ getMessageResult.release();
+ }
+ } else {
+ // Error handling.
+ RemotingCommand response = RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setOpaque(opaque);
+ channel.writeAndFlush(response);
+ }
+ }
+ }
+ };
+
+ // Submit asynchronous IO request through Linux AIO system call interface.
+ for (SelectMappedBufferResult selectMappedBufferResult : selectMappedBufferResults) {
+ MappedFile mappedFile = selectMappedBufferResult.getMappedFile();
+ try {
+ mappedFile.getLibaioFile().read(selectMappedBufferResult.getStartOffset(),
+ selectMappedBufferResult.getSize(), selectMappedBufferResult.getByteBuffer(),
+ submitInfo);
+ } catch (IOException e) {
+
+ }
+ }
response = null;
}
break;
diff --git a/store/pom.xml b/store/pom.xml
index dd035ea57..704f68ec8 100644
--- a/store/pom.xml
+++ b/store/pom.xml
@@ -46,5 +46,10 @@
<artifactId>logback-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>artemis-native</artifactId>
+ <version>2.4.0</version>
+ </dependency>
</dependencies>
</project>
diff --git a/store/src/main/java/org/apache/rocketmq/store/AccessDiskStrategy.java b/store/src/main/java/org/apache/rocketmq/store/AccessDiskStrategy.java
new file mode 100644
index 000000000..a85024f10
--- /dev/null
+++ b/store/src/main/java/org/apache/rocketmq/store/AccessDiskStrategy.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.store;
+
+public enum AccessDiskStrategy {
+ MMAP,
+ AIO,
+ HEAP
+}
diff --git a/store/src/main/java/org/apache/rocketmq/store/AllocateMappedFileService.java b/store/src/main/java/org/apache/rocketmq/store/AllocateMappedFileService.java
index ad8e65dca..90d9c1cb2 100644
--- a/store/src/main/java/org/apache/rocketmq/store/AllocateMappedFileService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/AllocateMappedFileService.java
@@ -174,10 +174,10 @@ private boolean mmapOperation() {
if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
try {
mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
- mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
+ mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool(), messageStore.getLibaioContext());
} catch (RuntimeException e) {
log.warn("Use default implementation.");
- mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
+ mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool(), messageStore.getLibaioContext());
}
} else {
mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
index 03d98d319..d4d26c392 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -65,7 +65,8 @@
public CommitLog(final DefaultMessageStore defaultMessageStore) {
this.mappedFileQueue = new MappedFileQueue(defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog(),
- defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog(), defaultMessageStore.getAllocateMappedFileService());
+ defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog(),
+ defaultMessageStore.getAllocateMappedFileService());
this.defaultMessageStore = defaultMessageStore;
if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
index 192d0972e..c018dbfb4 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -35,6 +35,7 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.activemq.artemis.jlibaio.LibaioContext;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ServiceThread;
@@ -110,6 +111,8 @@
private FileLock lock;
+ private LibaioContext libaioContext;
+
public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {
this.messageArrivingListener = messageArrivingListener;
@@ -148,6 +151,10 @@ public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final Br
File file = new File(StorePathConfigHelper.getLockFile(messageStoreConfig.getStorePathRootDir()));
MappedFile.ensureDirOK(file.getParent());
lockFile = new RandomAccessFile(file, "rw");
+
+ if (System.getProperty("os.name").equalsIgnoreCase("linux") && messageStoreConfig.isEnableAIO()) {
+ libaioContext = new LibaioContext(1048576, true, true);
+ }
}
public void truncateDirtyLogicFiles(long phyOffset) {
@@ -505,6 +512,10 @@ public GetMessageResult getMessage(final String group, final String topic, final
boolean isInDisk = checkInDiskByCommitOffset(offsetPy, maxOffsetPy);
+ if (isInDisk) {
+ getResult.setAccessDiskStrategy(null == libaioContext ? AccessDiskStrategy.HEAP : AccessDiskStrategy.AIO);
+ }
+
if (this.isTheBatchFull(sizePy, maxMsgNums, getResult.getBufferTotalSize(), getResult.getMessageCount(),
isInDisk)) {
break;
@@ -1404,6 +1415,10 @@ public void run() {
}, 6, TimeUnit.SECONDS);
}
+ public LibaioContext getLibaioContext() {
+ return libaioContext;
+ }
+
class CommitLogDispatcherBuildConsumeQueue implements CommitLogDispatcher {
@Override
diff --git a/store/src/main/java/org/apache/rocketmq/store/GetMessageResult.java b/store/src/main/java/org/apache/rocketmq/store/GetMessageResult.java
index 996e24d80..e1f986a16 100644
--- a/store/src/main/java/org/apache/rocketmq/store/GetMessageResult.java
+++ b/store/src/main/java/org/apache/rocketmq/store/GetMessageResult.java
@@ -23,8 +23,7 @@
public class GetMessageResult {
- private final List<SelectMappedBufferResult> messageMapedList =
- new ArrayList<SelectMappedBufferResult>(100);
+ private final List<SelectMappedBufferResult> messageMappedList = new ArrayList<SelectMappedBufferResult>(100);
private final List<ByteBuffer> messageBufferList = new ArrayList<ByteBuffer>(100);
@@ -39,7 +38,10 @@
private int msgCount4Commercial = 0;
+ private AccessDiskStrategy accessDiskStrategy;
+
public GetMessageResult() {
+ accessDiskStrategy = AccessDiskStrategy.MMAP;
}
public GetMessageStatus getStatus() {
@@ -74,8 +76,8 @@ public void setMaxOffset(long maxOffset) {
this.maxOffset = maxOffset;
}
- public List<SelectMappedBufferResult> getMessageMapedList() {
- return messageMapedList;
+ public List<SelectMappedBufferResult> getMessageMappedList() {
+ return messageMappedList;
}
public List<ByteBuffer> getMessageBufferList() {
@@ -83,15 +85,14 @@ public void setMaxOffset(long maxOffset) {
}
public void addMessage(final SelectMappedBufferResult mapedBuffer) {
- this.messageMapedList.add(mapedBuffer);
+ this.messageMappedList.add(mapedBuffer);
this.messageBufferList.add(mapedBuffer.getByteBuffer());
this.bufferTotalSize += mapedBuffer.getSize();
- this.msgCount4Commercial += (int) Math.ceil(
- mapedBuffer.getSize() / BrokerStatsManager.SIZE_PER_COUNT);
+ this.msgCount4Commercial += (int) Math.ceil(mapedBuffer.getSize() / BrokerStatsManager.SIZE_PER_COUNT);
}
public void release() {
- for (SelectMappedBufferResult select : this.messageMapedList) {
+ for (SelectMappedBufferResult select : this.messageMappedList) {
select.release();
}
}
@@ -105,7 +106,7 @@ public void setBufferTotalSize(int bufferTotalSize) {
}
public int getMessageCount() {
- return this.messageMapedList.size();
+ return this.messageMappedList.size();
}
public boolean isSuggestPullingFromSlave() {
@@ -124,6 +125,14 @@ public void setMsgCount4Commercial(int msgCount4Commercial) {
this.msgCount4Commercial = msgCount4Commercial;
}
+ public AccessDiskStrategy getAccessDiskStrategy() {
+ return accessDiskStrategy;
+ }
+
+ public void setAccessDiskStrategy(AccessDiskStrategy accessDiskStrategy) {
+ this.accessDiskStrategy = accessDiskStrategy;
+ }
+
@Override
public String toString() {
return "GetMessageResult [status=" + status + ", nextBeginOffset=" + nextBeginOffset + ", minOffset="
diff --git a/store/src/main/java/org/apache/rocketmq/store/MappedFile.java b/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
index 492ac5f26..30dcdc7da 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
@@ -31,6 +31,8 @@
import java.security.PrivilegedAction;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.activemq.artemis.jlibaio.LibaioContext;
+import org.apache.activemq.artemis.jlibaio.LibaioFile;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageExt;
@@ -66,6 +68,8 @@
private volatile long storeTimestamp = 0;
private boolean firstCreateInQueue = false;
+ private LibaioFile libaioFile;
+
public MappedFile() {
}
@@ -74,8 +78,8 @@ public MappedFile(final String fileName, final int fileSize) throws IOException
}
public MappedFile(final String fileName, final int fileSize,
- final TransientStorePool transientStorePool) throws IOException {
- init(fileName, fileSize, transientStorePool);
+ final TransientStorePool transientStorePool, final LibaioContext libaioContext) throws IOException {
+ init(fileName, fileSize, transientStorePool, libaioContext);
}
public static void ensureDirOK(final String dirName) {
@@ -144,10 +148,13 @@ public static long getTotalMappedVirtualMemory() {
}
public void init(final String fileName, final int fileSize,
- final TransientStorePool transientStorePool) throws IOException {
+ final TransientStorePool transientStorePool, final LibaioContext libaioContext) throws IOException {
init(fileName, fileSize);
this.writeBuffer = transientStorePool.borrowBuffer();
this.transientStorePool = transientStorePool;
+ if (null != libaioContext) {
+ libaioFile = libaioContext.openFile(new File(fileName), true);
+ }
}
private void init(final String fileName, final int fileSize) throws IOException {
@@ -456,6 +463,14 @@ public boolean destroy(final long intervalForcibly) {
log.warn("close file channel " + this.fileName + " Failed. ", e);
}
+ if (null != libaioFile) {
+ try {
+ libaioFile.close();
+ } catch (IOException e) {
+ log.error("Error while close LibAIO file", e);
+ }
+ }
+
return true;
} else {
log.warn("destroy mapped file[REF:" + this.getRefCount() + "] " + this.fileName
@@ -570,6 +585,14 @@ public void munlock() {
log.info("munlock {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret, System.currentTimeMillis() - beginTime);
}
+ public LibaioFile getLibaioFile() {
+ return libaioFile;
+ }
+
+ public void setLibaioFile(LibaioFile libaioFile) {
+ this.libaioFile = libaioFile;
+ }
+
@Override
public String toString() {
return this.fileName;
diff --git a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
index 02aa84a3e..98c781c5a 100644
--- a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+++ b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
@@ -143,6 +143,8 @@
private int transientStorePoolSize = 5;
private boolean fastFailIfNoBufferInStorePool = false;
+ private boolean enableAIO = true;
+
public boolean isDebugLockEnable() {
return debugLockEnable;
}
@@ -666,4 +668,11 @@ public void setCommitCommitLogThoroughInterval(final int commitCommitLogThorough
this.commitCommitLogThoroughInterval = commitCommitLogThoroughInterval;
}
+ public boolean isEnableAIO() {
+ return enableAIO;
+ }
+
+ public void setEnableAIO(boolean enableAIO) {
+ this.enableAIO = enableAIO;
+ }
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services