franz1981 commented on a change in pull request #2845: ARTEMIS-2336 Use zero
copy to replicate journal/page/large message file (AGAIN)
URL: https://github.com/apache/activemq-artemis/pull/2845#discussion_r328073825
##########
File path:
artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java
##########
@@ -560,49 +590,52 @@ private void
sendLargeFile(AbstractJournalStorageManager.JournalContent content,
if (!file.isOpen()) {
file.open();
}
- int size = 32 * 1024;
+ final int size = 1024 * 1024;
+ long fileSize = file.size();
int flowControlSize = 10;
int packetsSent = 0;
FlushAction action = new FlushAction();
+ long offset = 0;
+ RandomAccessFile raf = null;
+ FileChannel fileChannel = null;
try {
- try (FileInputStream fis = new FileInputStream(file.getJavaFile());
FileChannel channel = fis.getChannel()) {
-
- // We can afford having a single buffer here for this entire loop
- // because sendReplicatePacket will encode the packet as a
NettyBuffer
- // through ActiveMQBuffer class leaving this buffer free to be
reused on the next copy
- while (true) {
- final ByteBuf buffer =
PooledByteBufAllocator.DEFAULT.directBuffer(size, size);
- buffer.clear();
- ByteBuffer byteBuffer =
buffer.writerIndex(size).readerIndex(0).nioBuffer();
- final int bytesRead = channel.read(byteBuffer);
- int toSend = bytesRead;
- if (bytesRead > 0) {
- if (bytesRead >= maxBytesToSend) {
- toSend = (int) maxBytesToSend;
- maxBytesToSend = 0;
- } else {
- maxBytesToSend = maxBytesToSend - bytesRead;
- }
- }
- logger.debug("sending " + buffer.writerIndex() + " bytes on
file " + file.getFileName());
- // sending -1 or 0 bytes will close the file at the backup
- // We cannot simply send everything of a file through the
executor,
- // otherwise we would run out of memory.
- // so we don't use the executor here
- sendReplicatePacket(new ReplicationSyncFileMessage(content,
pageStore, id, toSend, buffer), true);
- packetsSent++;
-
- if (packetsSent % flowControlSize == 0) {
- flushReplicationStream(action);
+ raf = new RandomAccessFile(file.getJavaFile(), "r");
Review comment:
I see that :
```java
if (syncFileMessage.getFileId() != -1 &&
syncFileMessage.getDataSize() > 0) {
replicatingChannel.send(syncFileMessage,
syncFileMessage.getFileChannel(),
syncFileMessage.getOffset(),
syncFileMessage.getDataSize(),
lastChunk ? (Channel.Callback)
success -> syncFileMessage.release() : null);
```
So we never send syncFileMessage with file size == 0 and when we will do it
with
```java
replicatingChannel.send(syncFileMessage);
```
We don't release it. It seems a leak, but is not, because we will likely to
call it on the previous iteration and it should be already ready to be release
or released upon completion of the previous call.
I don't think we never send 0 bytes file sized on the wire AFAIK, but *if*
it would happen we're not releasing it correctly: we should add
``java
replicatingChannel.send(syncFileMessage, lastChunk ?
(Channel.Callback) success -> syncFileMessage.release() : null);
```
It would cause on the normal path (file::size > 0) to have
`syncFileMessage.release` called twice ATM (we mark lastChunk == true for the
last 2 sent packets), but `RandomAccessFile::close` should be idempotent
so...no harm :) wdyt?
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
