From 3949f6938e136f35e5e929588ef465ab74e82008 Mon Sep 17 00:00:00 2001 From: SSpirits Date: Mon, 9 Jan 2023 16:47:42 +0800 Subject: [PATCH] [ISSUE #5837] Deprecate MessageStore#checkInDiskByConsumeOffset (#5840) * deprecate MessageStore#checkInDiskByConsumeOffset and add checkInMemByConsumeOffset and checkInStoreByConsumeOffset in MessageStore * skip initializing isLoaded0 method in windows --- .../broker/offset/BroadcastOffsetManager.java | 2 +- .../processor/ConsumerManageProcessor.java | 4 +- .../rocketmq/store/DefaultMessageStore.java | 79 ++++++++++++++----- .../apache/rocketmq/store/MessageStore.java | 32 ++++++-- .../store/SelectMappedBufferResult.java | 11 ++- .../store/logfile/DefaultMappedFile.java | 55 +++++++++++++ .../rocketmq/store/logfile/MappedFile.java | 8 ++ .../plugin/AbstractPluginMessageStore.java | 11 +++ .../store/queue/BatchConsumeMessageTest.java | 4 +- 9 files changed, 176 insertions(+), 30 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/offset/BroadcastOffsetManager.java b/broker/src/main/java/org/apache/rocketmq/broker/offset/BroadcastOffsetManager.java index 16e70eed25..9896735dd1 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/offset/BroadcastOffsetManager.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/offset/BroadcastOffsetManager.java @@ -116,7 +116,7 @@ public class BroadcastOffsetManager extends ServiceThread { brokerController.getConsumerOffsetManager().queryOffset(broadcastGroupId(groupId), topic, queueId); } if (storeOffset < 0) { - if (!this.brokerController.getMessageStore().checkInDiskByConsumeOffset(topic, queueId, 0)) { + if (this.brokerController.getMessageStore().checkInMemByConsumeOffset(topic, queueId, 0, 1)) { storeOffset = 0; } else { storeOffset = brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId, true); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java index 395102c7ed..9c6d28d3dc 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java @@ -328,8 +328,8 @@ public class ConsumerManageProcessor implements NettyRequestProcessor { response.setCode(ResponseCode.QUERY_NOT_FOUND); response.setRemark("Not found, do not set to zero, maybe this group boot first"); } else if (minOffset <= 0 - && !this.brokerController.getMessageStore().checkInDiskByConsumeOffset( - requestHeader.getTopic(), requestHeader.getQueueId(), 0)) { + && this.brokerController.getMessageStore().checkInMemByConsumeOffset( + requestHeader.getTopic(), requestHeader.getQueueId(), 0, 1)) { responseHeader.setOffset(0L); response.setCode(ResponseCode.SUCCESS); response.setRemark(null); 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 93d0062458..b52982dc48 100644 --- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java @@ -795,13 +795,13 @@ public class DefaultMessageStore implements MessageStore { long offsetPy = cqUnit.getPos(); int sizePy = cqUnit.getSize(); - boolean isInDisk = checkInDiskByCommitOffset(offsetPy, maxOffsetPy); + boolean isInMem = estimateInMemByCommitOffset(offsetPy, maxOffsetPy); if ((cqUnit.getQueueOffset() - offset) * consumeQueue.getUnitSize() > maxFilterMessageSize) { break; } - if (this.isTheBatchFull(sizePy, cqUnit.getBatchNum(), maxMsgNums, maxPullSize, getResult.getBufferTotalSize(), getResult.getMessageCount(), isInDisk)) { + if (this.isTheBatchFull(sizePy, cqUnit.getBatchNum(), maxMsgNums, maxPullSize, getResult.getBufferTotalSize(), getResult.getMessageCount(), isInMem)) { break; } @@ -1418,6 +1418,7 @@ public class DefaultMessageStore implements MessageStore { } @Override + @Deprecated public boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset) { final long maxOffsetPy = this.commitLog.getMaxOffset(); @@ -1428,7 +1429,7 @@ public class DefaultMessageStore implements MessageStore { if (cqUnit != null) { long offsetPy = cqUnit.getPos(); - return checkInDiskByCommitOffset(offsetPy, maxOffsetPy); + return !estimateInMemByCommitOffset(offsetPy, maxOffsetPy); } else { return false; } @@ -1436,6 +1437,38 @@ public class DefaultMessageStore implements MessageStore { return false; } + @Override + public boolean checkInMemByConsumeOffset(final String topic, final int queueId, long consumeOffset, int batchSize) { + ConsumeQueueInterface consumeQueue = findConsumeQueue(topic, queueId); + if (consumeQueue != null) { + CqUnit firstCQItem = consumeQueue.get(consumeOffset); + if (firstCQItem == null) { + return false; + } + long startOffsetPy = firstCQItem.getPos(); + if (batchSize <= 1) { + int size = firstCQItem.getSize(); + return checkInMemByCommitOffset(startOffsetPy, size); + } + + CqUnit lastCQItem = consumeQueue.get(consumeOffset + batchSize); + if (lastCQItem == null) { + int size = firstCQItem.getSize(); + return checkInMemByCommitOffset(startOffsetPy, size); + } + long endOffsetPy = lastCQItem.getPos(); + int size = (int) (endOffsetPy - startOffsetPy) + lastCQItem.getSize(); + return checkInMemByCommitOffset(startOffsetPy, size); + } + return false; + } + + @Override + public boolean checkInStoreByConsumeOffset(String topic, int queueId, long consumeOffset) { + long commitLogOffset = getCommitLogOffsetInQueue(topic, queueId, consumeOffset); + return checkInDiskByCommitOffset(commitLogOffset); + } + @Override public long dispatchBehindBytes() { return this.reputMessageService.behind(); @@ -1600,13 +1633,29 @@ public class DefaultMessageStore implements MessageStore { return nextOffset; } - private boolean checkInDiskByCommitOffset(long offsetPy, long maxOffsetPy) { + private boolean estimateInMemByCommitOffset(long offsetPy, long maxOffsetPy) { long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0)); - return (maxOffsetPy - offsetPy) > memory; + return (maxOffsetPy - offsetPy) <= memory; + } + + private boolean checkInMemByCommitOffset(long offsetPy, int size) { + SelectMappedBufferResult message = this.commitLog.getMessage(offsetPy, size); + if (message != null) { + try { + return message.isInMem(); + } finally { + message.release(); + } + } + return false; + } + + public boolean checkInDiskByCommitOffset(long offsetPy) { + return offsetPy >= commitLog.getMinOffset(); } private boolean isTheBatchFull(int sizePy, int unitBatchNum, int maxMsgNums, long maxMsgSize, int bufferTotal, - int messageTotal, boolean isInDisk) { + int messageTotal, boolean isInMem) { if (0 == bufferTotal || 0 == messageTotal) { return false; @@ -1620,25 +1669,19 @@ public class DefaultMessageStore implements MessageStore { return true; } - if (isInDisk) { - if ((bufferTotal + sizePy) > this.messageStoreConfig.getMaxTransferBytesOnMessageInDisk()) { - return true; - } - - if (messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInDisk() - 1) { - return true; - } - } else { + if (isInMem) { if ((bufferTotal + sizePy) > this.messageStoreConfig.getMaxTransferBytesOnMessageInMemory()) { return true; } - if (messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInMemory() - 1) { + return messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInMemory() - 1; + } else { + if ((bufferTotal + sizePy) > this.messageStoreConfig.getMaxTransferBytesOnMessageInDisk()) { return true; } - } - return false; + return messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInDisk() - 1; + } } private void deleteFile(final String fileName) { diff --git a/store/src/main/java/org/apache/rocketmq/store/MessageStore.java b/store/src/main/java/org/apache/rocketmq/store/MessageStore.java index bbf2056ccd..8e86f8abe7 100644 --- a/store/src/main/java/org/apache/rocketmq/store/MessageStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/MessageStore.java @@ -129,7 +129,7 @@ public interface MessageStore { /** * Asynchronous get message - * @see org.apache.rocketmq.store.MessageStore#getMessage(String, String, int, long, int, MessageFilter) getMessage + * @see #getMessage(String, String, int, long, int, MessageFilter) getMessage * * @param group Consumer group that launches this query. * @param topic Topic to query. @@ -160,7 +160,7 @@ public interface MessageStore { /** * Asynchronous get message - * @see org.apache.rocketmq.store.MessageStore#getMessage(String, String, int, long, int, int, MessageFilter) getMessage + * @see #getMessage(String, String, int, long, int, int, MessageFilter) getMessage * * @param group Consumer group that launches this query. * @param topic Topic to query. @@ -312,7 +312,7 @@ public interface MessageStore { /** * Asynchronous get the store time of the earliest message in this store. - * @see org.apache.rocketmq.store.MessageStore#getEarliestMessageTime() getEarliestMessageTime + * @see #getEarliestMessageTime() getEarliestMessageTime * * @return timestamp of the earliest message in this store. */ @@ -330,7 +330,7 @@ public interface MessageStore { /** * Asynchronous get the store time of the message specified. - * @see org.apache.rocketmq.store.MessageStore#getMessageStoreTimeStamp(String, int, long) getMessageStoreTimeStamp + * @see #getMessageStoreTimeStamp(String, int, long) getMessageStoreTimeStamp * * @param topic message topic. * @param queueId queue ID. @@ -396,7 +396,7 @@ public interface MessageStore { /** * Asynchronous query messages by given key. - * @see org.apache.rocketmq.store.MessageStore#queryMessage(String, String, int, long, long) queryMessage + * @see #queryMessage(String, String, int, long, long) queryMessage * * @param topic topic of the message. * @param key message key. @@ -464,9 +464,31 @@ public interface MessageStore { * @param queueId queue ID. * @param consumeOffset consume queue offset. * @return true if the message is no longer in memory; false otherwise. + * @deprecated As of RIP-57, replaced by {@link #checkInMemByConsumeOffset(String, int, long, int)}, see this issue for more details */ + @Deprecated boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset); + /** + * Check if the given message is in the page cache. + * + * @param topic topic. + * @param queueId queue ID. + * @param consumeOffset consume queue offset. + * @return true if the message is in page cache; false otherwise. + */ + boolean checkInMemByConsumeOffset(final String topic, final int queueId, long consumeOffset, int batchSize); + + /** + * Check if the given message is in store. + * + * @param topic topic. + * @param queueId queue ID. + * @param consumeOffset consume queue offset. + * @return true if the message is in store; false otherwise. + */ + boolean checkInStoreByConsumeOffset(final String topic, final int queueId, long consumeOffset); + /** * Get number of the bytes that have been stored in commit log and not yet dispatched to consume queue. * diff --git a/store/src/main/java/org/apache/rocketmq/store/SelectMappedBufferResult.java b/store/src/main/java/org/apache/rocketmq/store/SelectMappedBufferResult.java index 76919f3d15..317f536058 100644 --- a/store/src/main/java/org/apache/rocketmq/store/SelectMappedBufferResult.java +++ b/store/src/main/java/org/apache/rocketmq/store/SelectMappedBufferResult.java @@ -16,9 +16,8 @@ */ package org.apache.rocketmq.store; -import org.apache.rocketmq.store.logfile.MappedFile; - import java.nio.ByteBuffer; +import org.apache.rocketmq.store.logfile.MappedFile; public class SelectMappedBufferResult { @@ -67,4 +66,12 @@ public class SelectMappedBufferResult { public long getStartOffset() { return startOffset; } + + public boolean isInMem() { + if (mappedFile == null) { + return true; + } + long pos = startOffset - mappedFile.getFileFromOffset(); + return mappedFile.isLoaded(pos, size); + } } diff --git a/store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java b/store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java index 7b56150f64..401c64539e 100644 --- a/store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java +++ b/store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java @@ -22,6 +22,8 @@ import java.io.File; import java.io.FileNotFoundException; import java.io.IOException; import java.io.RandomAccessFile; +import java.lang.reflect.Field; +import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.nio.MappedByteBuffer; import java.nio.channels.FileChannel; @@ -35,6 +37,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; +import org.apache.commons.lang3.SystemUtils; import org.apache.rocketmq.common.UtilAll; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.message.MessageExt; @@ -52,10 +55,15 @@ import org.apache.rocketmq.store.SelectMappedBufferResult; import org.apache.rocketmq.store.TransientStorePool; import org.apache.rocketmq.store.config.FlushDiskType; import org.apache.rocketmq.store.util.LibC; +import sun.misc.Unsafe; import sun.nio.ch.DirectBuffer; public class DefaultMappedFile extends AbstractMappedFile { public static final int OS_PAGE_SIZE = 1024 * 4; + public static final Unsafe UNSAFE = getUnsafe(); + private static final Method IS_LOADED_METHOD; + public static final int UNSAFE_PAGE_SIZE = UNSAFE == null ? OS_PAGE_SIZE : UNSAFE.pageSize(); + protected static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); protected static final AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0); @@ -92,6 +100,18 @@ public class DefaultMappedFile extends AbstractMappedFile { WROTE_POSITION_UPDATER = AtomicIntegerFieldUpdater.newUpdater(DefaultMappedFile.class, "wrotePosition"); COMMITTED_POSITION_UPDATER = AtomicIntegerFieldUpdater.newUpdater(DefaultMappedFile.class, "committedPosition"); FLUSHED_POSITION_UPDATER = AtomicIntegerFieldUpdater.newUpdater(DefaultMappedFile.class, "flushedPosition"); + + Method isLoaded0method = null; + // On the windows platform and openjdk 11 method isLoaded0 always returns false. + // see https://github.com/AdoptOpenJDK/openjdk-jdk11/blob/19fb8f93c59dfd791f62d41f332db9e306bc1422/src/java.base/windows/native/libnio/MappedByteBuffer.c#L34 + if (!SystemUtils.IS_OS_WINDOWS) { + try { + isLoaded0method = MappedByteBuffer.class.getDeclaredMethod("isLoaded0", long.class, long.class, int.class); + isLoaded0method.setAccessible(true); + } catch (NoSuchMethodException ignore) { + } + } + IS_LOADED_METHOD = isLoaded0method; } public DefaultMappedFile() { @@ -796,6 +816,41 @@ public class DefaultMappedFile extends AbstractMappedFile { return new Itr(startPos); } + public static Unsafe getUnsafe() { + try { + Field f = Unsafe.class.getDeclaredField("theUnsafe"); + f.setAccessible(true); + return (Unsafe) f.get(null); + } catch (Exception ignore) { + + } + return null; + } + + public static long mappingAddr(long addr) { + long offset = addr % UNSAFE_PAGE_SIZE; + offset = (offset >= 0) ? offset : (UNSAFE_PAGE_SIZE + offset); + return addr - offset; + } + + public static int pageCount(long size) { + return (int) (size + (long) UNSAFE_PAGE_SIZE - 1L) / UNSAFE_PAGE_SIZE; + } + + @Override + public boolean isLoaded(long position, int size) { + if (IS_LOADED_METHOD == null) { + return true; + } + try { + long addr = ((DirectBuffer) mappedByteBuffer).address() + position; + return (boolean) IS_LOADED_METHOD.invoke(mappedByteBuffer, mappingAddr(addr), size, pageCount(size)); + } catch (Exception e) { + log.info("invoke isLoaded0 of file {} error:", file.getAbsolutePath(), e); + } + return true; + } + private class Itr implements Iterator { private int start; private int current; diff --git a/store/src/main/java/org/apache/rocketmq/store/logfile/MappedFile.java b/store/src/main/java/org/apache/rocketmq/store/logfile/MappedFile.java index 64e1336e8c..dfcf66f088 100644 --- a/store/src/main/java/org/apache/rocketmq/store/logfile/MappedFile.java +++ b/store/src/main/java/org/apache/rocketmq/store/logfile/MappedFile.java @@ -360,4 +360,12 @@ public interface MappedFile { void init(String fileName, int fileSize, TransientStorePool transientStorePool) throws IOException; Iterator iterator(int pos); + + /** + * Check mapped file is loaded to memory with given position and size + * @param position start offset of data + * @param size data size + * @return data is resided in memory or not + */ + boolean isLoaded(long position, int size); } diff --git a/store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java index 47416a8736..77908c5fa6 100644 --- a/store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java @@ -272,10 +272,21 @@ public abstract class AbstractPluginMessageStore implements MessageStore { } @Override + @Deprecated public boolean checkInDiskByConsumeOffset(String topic, int queueId, long consumeOffset) { return next.checkInDiskByConsumeOffset(topic, queueId, consumeOffset); } + @Override + public boolean checkInMemByConsumeOffset(String topic, int queueId, long consumeOffset, int batchSize) { + return next.checkInMemByConsumeOffset(topic, queueId, consumeOffset, batchSize); + } + + @Override + public boolean checkInStoreByConsumeOffset(String topic, int queueId, long consumeOffset) { + return next.checkInStoreByConsumeOffset(topic, queueId, consumeOffset); + } + @Override public long dispatchBehindBytes() { return next.dispatchBehindBytes(); diff --git a/store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest.java b/store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest.java index 8e8fee278e..1a18f1ba1a 100644 --- a/store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/queue/BatchConsumeMessageTest.java @@ -281,7 +281,7 @@ public class BatchConsumeMessageTest extends QueueTestBase { Assert.assertTrue(commitLogOffset <= messageStore.getMaxPhyOffset()); Assert.assertEquals(commitLogMid, commitLogOffset); - Assert.assertFalse(messageStore.checkInDiskByConsumeOffset(topic, 0, 50)); + Assert.assertTrue(messageStore.checkInMemByConsumeOffset(topic, 0, 50, 1)); } @Test @@ -331,7 +331,7 @@ public class BatchConsumeMessageTest extends QueueTestBase { Assert.assertTrue(commitLogOffset >= messageStore.getMinPhyOffset()); Assert.assertTrue(commitLogOffset <= messageStore.getMaxPhyOffset()); - Assert.assertFalse(messageStore.checkInDiskByConsumeOffset(topic, 0, 300)); + Assert.assertTrue(messageStore.checkInMemByConsumeOffset(topic, 0, 300, 1)); //get the message Normally GetMessageResult getMessageResult = messageStore.getMessage("group", topic, 0, 0, 10 * batchNum, null);