franz1981 commented on a change in pull request #2633: ARTEMIS-2317 Avoid long
TTSP caused by Page::read using mmap read
URL: https://github.com/apache/activemq-artemis/pull/2633#discussion_r278136368
##########
File path:
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java
##########
@@ -120,105 +110,121 @@ public void setLiveCache(LivePageCache pageCache) {
throw ActiveMQMessageBundle.BUNDLE.invalidPageIO();
}
- final List<PagedMessage> messages = new ArrayList<>();
-
size.lazySet((int) file.size());
- if (this.canBeMapped) {
- readFromMapped(storage, messages);
- // if the file is open to be written
- // it needs to updated the position
- file.position(file.size());
- } else {
- readFromSequentialFile(storage, messages);
- }
+ final List<PagedMessage> messages = readFromSequentialFile(storage);
numberOfMessages.lazySet(messages.size());
return messages;
}
- private void readFromSequentialFile(StorageManager storage,
List<PagedMessage> messages) throws Exception {
- final int fileSize = (int) file.size();
- //doesn't need to be a direct buffer: that case is covered using the
MMAP read
- final ByteBuffer buffer = this.fileFactory.newBuffer(fileSize);
- try {
- file.position(0);
- file.read(buffer);
- buffer.rewind();
- assert (buffer.limit() == fileSize) : "buffer doesn't contains the
whole file";
- ChannelBufferWrapper activeMQBuffer = wrapBuffer(fileSize, buffer);
- read(storage, activeMQBuffer, messages);
- } finally {
- this.fileFactory.releaseBuffer(buffer);
- }
+ private static void decodeInto(ByteBuffer fileBuffer, int encodedSize,
PagedMessageImpl msg) {
+ final ActiveMQBuffer wrappedBuffer =
ActiveMQBuffers.wrappedBuffer(fileBuffer);
+ wrappedBuffer.writerIndex(encodedSize);
+ msg.decode(wrappedBuffer);
}
- private ChannelBufferWrapper wrapBuffer(int fileSize, ByteBuffer buffer) {
- ChannelBufferWrapper activeMQBuffer = new
ChannelBufferWrapper(Unpooled.wrappedBuffer(buffer));
- return activeMQBuffer;
+ private ByteBuffer allocateAndReadIntoFileBuffer(SequentialFile file,
+ ByteBuffer fileBuffer,
+ int requiredBytes) throws
Exception {
+ final ByteBuffer newFileBuffer =
fileFactory.newBuffer(Math.max(requiredBytes, Env.osPageSize()));
+ newFileBuffer.put(fileBuffer);
+ fileFactory.releaseBuffer(fileBuffer);
+ fileBuffer = newFileBuffer;
+ //move the limit to allow reading as much as possible from the file
+ fileBuffer.limit(fileBuffer.capacity());
+ file.read(fileBuffer);
+ fileBuffer.position(0);
+ return fileBuffer;
}
- private static MappedByteBuffer mapFileForRead(File file, int fileSize) {
- try (FileChannel channel = FileChannel.open(file.toPath(),
StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE)) {
- return channel.map(FileChannel.MapMode.READ_ONLY, 0, fileSize);
- } catch (Exception e) {
- throw new IllegalStateException(e);
+ private static void compactAndReadIntoFileBuffer(SequentialFile file,
ByteBuffer fileBuffer) throws Exception {
+ if (fileBuffer.remaining() > 0) {
+ fileBuffer.compact();
}
+ fileBuffer.limit(fileBuffer.capacity());
+ file.read(fileBuffer);
+ fileBuffer.position(0);
}
- private int readFromMapped(StorageManager storage, List<PagedMessage>
messages) throws IOException {
- file.position(0);
- //use a readonly mapped view of the file
- final int mappedSize = size.get();
- final MappedByteBuffer mappedByteBuffer =
mapFileForRead(this.file.getJavaFile(), mappedSize);
- ChannelBufferWrapper activeMQBuffer = wrapBuffer(mappedSize,
mappedByteBuffer);
- try {
- return read(storage, activeMQBuffer, messages);
- } finally {
- //unmap the file after read it to avoid GC to take care of it
- PlatformDependent.freeDirectBuffer(mappedByteBuffer);
+ private ByteBuffer readIntoFileBufferIfNecessary(SequentialFile file,
+ ByteBuffer fileBuffer,
+ int requiredBytes) throws
Exception {
+ final int remaining = fileBuffer.remaining();
+ if (remaining < requiredBytes) {
+ final int capacity = fileBuffer.capacity();
+ if (capacity >= requiredBytes) {
+ final int bytesToBeRead = requiredBytes - remaining;
+ final int availableCapacity = capacity - fileBuffer.limit();
+ if (availableCapacity >= bytesToBeRead) {
+ final int originalPosition = fileBuffer.position();
+ fileBuffer.limit(capacity);
+ file.read(fileBuffer);
+ fileBuffer.position(originalPosition);
+ } else {
+ compactAndReadIntoFileBuffer(file, fileBuffer);
+ }
+ } else {
+ fileBuffer = allocateAndReadIntoFileBuffer(file, fileBuffer,
requiredBytes);
+ }
}
+ return fileBuffer;
}
- private int read(StorageManager storage, ActiveMQBuffer fileBuffer,
List<PagedMessage> messages) {
- int readMessages = 0;
- while (fileBuffer.readable()) {
- final int position = fileBuffer.readerIndex();
-
- byte byteRead = fileBuffer.readByte();
-
- if (byteRead == Page.START_BYTE) {
- if (fileBuffer.readerIndex() + DataConstants.SIZE_INT <
fileBuffer.capacity()) {
- int messageSize = fileBuffer.readInt();
- int oldPos = fileBuffer.readerIndex();
- if (fileBuffer.readerIndex() + messageSize <
fileBuffer.capacity() && fileBuffer.getByte(oldPos + messageSize) ==
Page.END_BYTE) {
- PagedMessage msg = new PagedMessageImpl(storageManager);
- msg.decode(fileBuffer);
- byte b = fileBuffer.readByte();
- if (b != Page.END_BYTE) {
- // Sanity Check: This would only happen if there is a bug
on decode or any internal code, as
- // this
- // constraint was already checked
- throw new IllegalStateException("Internal error, it
wasn't possible to locate END_BYTE " + b);
- }
+ private List<PagedMessage> readFromSequentialFile(StorageManager storage)
throws Exception {
+ final List<PagedMessage> messages = new ArrayList<>();
+ final int fileSize = (int) file.size();
+ file.position(0);
+ final int minimumMessagePersistentSize = DataConstants.SIZE_INT + 2;
+ int position = 0;
+ ByteBuffer fileBuffer = null;
+ try {
+ while ((fileSize - position) >= minimumMessagePersistentSize) {
+ if (fileBuffer == null) {
+ fileBuffer = fileFactory.newBuffer(Math.min((fileSize -
position), Env.osPageSize()));
+ fileBuffer.limit(0);
+ }
+ fileBuffer = readIntoFileBufferIfNecessary(file, fileBuffer,
minimumMessagePersistentSize);
+ final byte startByte = fileBuffer.get();
+ if (startByte == Page.START_BYTE) {
+ final int encodedSize = fileBuffer.getInt();
+ fileBuffer = readIntoFileBufferIfNecessary(file, fileBuffer,
encodedSize + 1);
+ final int endPosition = fileBuffer.position() + encodedSize;
+ //this check must be performed upfront decoding
+ if (fileBuffer.get(endPosition) == Page.END_BYTE) {
+ final PagedMessageImpl msg = new
PagedMessageImpl(storageManager);
+ decodeInto(fileBuffer, encodedSize, msg);
+ fileBuffer.position(endPosition + 1);
+ assert fileBuffer.get(endPosition) == Page.END_BYTE :
"decoding cannot change end byte";
msg.initMessage(storage);
if (logger.isTraceEnabled()) {
- logger.trace("Reading message " + msg + " on pageId=" +
this.pageId + " for address=" + storeName);
+ logger.tracef("Reading message %s on pageId=%d for
address=%s", msg, pageId, storeName);
}
- readMessages++;
messages.add(msg);
+ position += minimumMessagePersistentSize + encodedSize;
Review comment:
I understand but the point of this PR is to maintain a similar speed of mmap
without incurring into the time-to-safepoint issue; that's why the logic has
changed.
The original implementation (pre-mmap) was instantiating for each file a
giant direct `ByteBuffer`, impacting on the overall native memory footprint and
causing a long context switch while making the kernel fill that buffer from its
OS (file) page cache.
It has been solved using mmap, to avoid allocations and copies, but
`MappedByteBuffer::get(byte)` could pause the JVM for long time: that's why I'm
addressing both the issues at once: memory footprint and long time to
safepoints (to save memory).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
