bytesPool;
+ private final ByteBufferPool bytesPool;
public NIOSequentialFileFactory(final File journalDir, final int maxIO) {
this(journalDir, null, maxIO);
@@ -76,7 +77,7 @@ public class NIOSequentialFileFactory extends AbstractSequentialFileFactory {
final CriticalAnalyzer analyzer) {
super(journalDir, buffered, bufferSize, bufferTimeout, maxIO, logRates, listener, analyzer);
this.bufferPooling = true;
- this.bytesPool = new ThreadLocal<>();
+ this.bytesPool = ByteBufferPool.threadLocal(true);
}
public static ByteBuffer allocateDirectByteBuffer(final int size) {
@@ -123,13 +124,9 @@ public class NIOSequentialFileFactory extends AbstractSequentialFileFactory {
return timedBuffer != null;
}
- private static int align(final int value, final int pow2alignment) {
- return (value + (pow2alignment - 1)) & ~(pow2alignment - 1);
- }
-
@Override
public ByteBuffer allocateDirectBuffer(final int size) {
- final int requiredCapacity = align(size, DEFAULT_CAPACITY_ALIGNMENT);
+ final int requiredCapacity = PowerOf2Util.align(size, DEFAULT_CAPACITY_ALIGNMENT);
final ByteBuffer byteBuffer = ByteBuffer.allocateDirect(requiredCapacity);
byteBuffer.limit(size);
return byteBuffer;
@@ -141,43 +138,18 @@ public class NIOSequentialFileFactory extends AbstractSequentialFileFactory {
}
@Override
- public ByteBuffer newBuffer(final int size) {
+ public ByteBuffer newBuffer(int size) {
if (!this.bufferPooling) {
return allocateDirectBuffer(size);
} else {
- final int requiredCapacity = align(size, DEFAULT_CAPACITY_ALIGNMENT);
- ByteBuffer byteBuffer = bytesPool.get();
- if (byteBuffer == null || requiredCapacity > byteBuffer.capacity()) {
- //do not free the old one (if any) until the new one will be released into the pool!
- byteBuffer = ByteBuffer.allocateDirect(requiredCapacity);
- } else {
- bytesPool.set(null);
- PlatformDependent.setMemory(PlatformDependent.directBufferAddress(byteBuffer), size, (byte) 0);
- byteBuffer.clear();
- }
- byteBuffer.limit(size);
- return byteBuffer;
+ return bytesPool.borrow(size, true);
}
}
@Override
public void releaseBuffer(ByteBuffer buffer) {
if (this.bufferPooling) {
- if (buffer.isDirect()) {
- final ByteBuffer byteBuffer = bytesPool.get();
- if (byteBuffer != buffer) {
- //replace with the current pooled only if greater or null
- if (byteBuffer == null || buffer.capacity() > byteBuffer.capacity()) {
- if (byteBuffer != null) {
- //free the smaller one
- PlatformDependent.freeDirectBuffer(byteBuffer);
- }
- bytesPool.set(buffer);
- } else {
- PlatformDependent.freeDirectBuffer(buffer);
- }
- }
- }
+ bytesPool.release(buffer);
}
}
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ByteBufferPool.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ByteBufferPool.java
new file mode 100644
index 0000000000..bebb514186
--- /dev/null
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ByteBufferPool.java
@@ -0,0 +1,51 @@
+/*
+ * 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.activemq.artemis.core.io.util;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Object Pool that allows to borrow and release {@link ByteBuffer}s according to a specific type (direct/heap).
+ * The suggested usage pattern is:
+ * {@code
+ * ByteBuffer buffer = pool.borrow(size);
+ * //...using buffer...
+ * pool.release(buffer);
+ * }
+ */
+public interface ByteBufferPool {
+
+ /**
+ * It returns a {@link ByteBuffer} with {@link ByteBuffer#capacity()}>={@code size}.
+ * The {@code buffer} is zeroed until {@code size} if {@code zeroed=true}, with {@link ByteBuffer#position()}=0 and {@link ByteBuffer#limit()}={@code size}.
+ */
+ ByteBuffer borrow(int size, boolean zeroed);
+
+ /**
+ * It pools or free {@code buffer} that cannot be used anymore.
+ * If {@code buffer} is of a type different from the one that the pool can borrow, it will ignore it.
+ */
+ void release(ByteBuffer buffer);
+
+ /**
+ * Factory method that creates a thread-local pool of capacity 1 of {@link ByteBuffer}s of the specified type (direct/heap).
+ */
+ static ByteBufferPool threadLocal(boolean direct) {
+ return new ThreadLocalByteBufferPool(direct);
+ }
+
+}
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPool.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPool.java
new file mode 100644
index 0000000000..eee04668bd
--- /dev/null
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPool.java
@@ -0,0 +1,80 @@
+/*
+ * 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.activemq.artemis.core.io.util;
+
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+import io.netty.util.internal.PlatformDependent;
+import org.apache.activemq.artemis.utils.PowerOf2Util;
+import org.apache.activemq.artemis.utils.ByteUtil;
+import org.apache.activemq.artemis.utils.Env;
+
+final class ThreadLocalByteBufferPool implements ByteBufferPool {
+
+ private final ThreadLocal bytesPool;
+ private final boolean direct;
+
+ ThreadLocalByteBufferPool(boolean direct) {
+ this.bytesPool = new ThreadLocal<>();
+ this.direct = direct;
+ }
+
+ @Override
+ public ByteBuffer borrow(final int size, boolean zeroed) {
+ final int requiredCapacity = PowerOf2Util.align(size, Env.osPageSize());
+ ByteBuffer byteBuffer = bytesPool.get();
+ if (byteBuffer == null || requiredCapacity > byteBuffer.capacity()) {
+ //do not free the old one (if any) until the new one will be released into the pool!
+ byteBuffer = direct ? ByteBuffer.allocateDirect(requiredCapacity) : ByteBuffer.allocate(requiredCapacity);
+ } else {
+ bytesPool.set(null);
+ if (zeroed) {
+ ByteUtil.zeros(byteBuffer, 0, size);
+ }
+ byteBuffer.clear();
+ }
+ byteBuffer.limit(size);
+ return byteBuffer;
+ }
+
+ @Override
+ public void release(ByteBuffer buffer) {
+ Objects.requireNonNull(buffer);
+ boolean directBuffer = buffer.isDirect();
+ if (directBuffer == direct && !buffer.isReadOnly()) {
+ final ByteBuffer byteBuffer = bytesPool.get();
+ if (byteBuffer != buffer) {
+ //replace with the current pooled only if greater or null
+ if (byteBuffer == null || buffer.capacity() > byteBuffer.capacity()) {
+ if (byteBuffer != null) {
+ //free the smaller one
+ if (directBuffer) {
+ PlatformDependent.freeDirectBuffer(byteBuffer);
+ }
+ }
+ bytesPool.set(buffer);
+ } else {
+ if (directBuffer) {
+ PlatformDependent.freeDirectBuffer(buffer);
+ }
+ }
+ }
+ }
+ }
+
+}
diff --git a/artemis-journal/src/test/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPoolTest.java b/artemis-journal/src/test/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPoolTest.java
new file mode 100644
index 0000000000..190d7de401
--- /dev/null
+++ b/artemis-journal/src/test/java/org/apache/activemq/artemis/core/io/util/ThreadLocalByteBufferPoolTest.java
@@ -0,0 +1,174 @@
+/*
+ * 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.activemq.artemis.core.io.util;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import io.netty.util.internal.PlatformDependent;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class ThreadLocalByteBufferPoolTest {
+
+ //testing using heap buffers to avoid killing the test suite
+ private static final boolean isDirect = false;
+ private final ByteBufferPool pool = ByteBufferPool.threadLocal(isDirect);
+ private final boolean zeroed;
+
+ public ThreadLocalByteBufferPoolTest(boolean zeroed) {
+ this.zeroed = zeroed;
+ }
+
+ @Parameterized.Parameters(name = "zeroed={0}")
+ public static Collection getParams() {
+ return Arrays.asList(new Object[][]{{Boolean.TRUE}, {Boolean.FALSE}});
+ }
+
+ private static void assertZeroed(ByteBuffer buffer) {
+ ByteBuffer bb = buffer.slice();
+ final byte[] content = new byte[bb.remaining()];
+ bb.get(content);
+ final byte[] zeroed = new byte[content.length];
+ Arrays.fill(zeroed, (byte) 0);
+ Assert.assertArrayEquals(zeroed, content);
+ }
+
+ @Test
+ public void shouldBorrowOnlyBuffersOfTheCorrectType() {
+ Assert.assertEquals(isDirect, pool.borrow(0, zeroed).isDirect());
+ }
+
+ @Test
+ public void shouldBorrowZeroedBuffer() {
+ final int size = 32;
+ final ByteBuffer buffer = pool.borrow(size, zeroed);
+ Assert.assertEquals(0, buffer.position());
+ Assert.assertEquals(size, buffer.limit());
+ if (zeroed) {
+ assertZeroed(buffer);
+ }
+ }
+
+ @Test
+ public void shouldBorrowTheSameBuffer() {
+ final int size = 32;
+ final ByteBuffer buffer = pool.borrow(size, zeroed);
+ buffer.put(0, (byte) 1);
+ buffer.position(1);
+ buffer.limit(2);
+ pool.release(buffer);
+ final int newSize = size - 1;
+ final ByteBuffer sameBuffer = pool.borrow(newSize, zeroed);
+ Assert.assertSame(buffer, sameBuffer);
+ Assert.assertEquals(0, sameBuffer.position());
+ Assert.assertEquals(newSize, sameBuffer.limit());
+ if (zeroed) {
+ assertZeroed(sameBuffer);
+ }
+ }
+
+ @Test
+ public void shouldBorrowNewBufferIfExceedPooledCapacity() {
+ final int size = 32;
+ final ByteBuffer buffer = pool.borrow(size, zeroed);
+ pool.release(buffer);
+ final int newSize = buffer.capacity() + 1;
+ final ByteBuffer differentBuffer = pool.borrow(newSize, zeroed);
+ Assert.assertNotSame(buffer, differentBuffer);
+ }
+
+ @Test
+ public void shouldPoolTheBiggestBuffer() {
+ final int size = 32;
+ final ByteBuffer small = pool.borrow(size, zeroed);
+ final ByteBuffer big = pool.borrow(small.capacity() + 1, zeroed);
+ pool.release(small);
+ big.limit(0);
+ pool.release(big);
+ Assert.assertSame(big, pool.borrow(big.capacity(), zeroed));
+ }
+
+ @Test
+ public void shouldNotPoolTheSmallestBuffer() {
+ final int size = 32;
+ final ByteBuffer small = pool.borrow(size, zeroed);
+ final ByteBuffer big = pool.borrow(small.capacity() + 1, zeroed);
+ big.limit(0);
+ pool.release(big);
+ pool.release(small);
+ Assert.assertSame(big, pool.borrow(big.capacity(), zeroed));
+ }
+
+ @Test
+ public void shouldNotPoolBufferOfDifferentType() {
+ final int size = 32;
+ final ByteBuffer buffer = isDirect ? ByteBuffer.allocate(size) : ByteBuffer.allocateDirect(size);
+ try {
+ pool.release(buffer);
+ Assert.assertNotSame(buffer, pool.borrow(size, zeroed));
+ } catch (Throwable t) {
+ if (PlatformDependent.hasUnsafe()) {
+ if (buffer.isDirect()) {
+ PlatformDependent.freeDirectBuffer(buffer);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void shouldNotPoolReadOnlyBuffer() {
+ final int size = 32;
+ final ByteBuffer borrow = pool.borrow(size, zeroed);
+ final ByteBuffer readOnlyBuffer = borrow.asReadOnlyBuffer();
+ pool.release(readOnlyBuffer);
+ Assert.assertNotSame(readOnlyBuffer, pool.borrow(size, zeroed));
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void shouldFailPoolingNullBuffer() {
+ pool.release(null);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void shouldFailPoolingNullBufferIfNotEmpty() {
+ final int size = 32;
+ pool.release(pool.borrow(size, zeroed));
+ pool.release(null);
+ }
+
+ @Test
+ public void shouldBorrowOnlyThreadLocalBuffers() throws ExecutionException, InterruptedException {
+ final int size = 32;
+ final ByteBuffer buffer = pool.borrow(size, zeroed);
+ pool.release(buffer);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Assert.assertNotSame(buffer, executor.submit(() -> pool.borrow(size, zeroed)).get());
+ } finally {
+ executor.shutdown();
+ }
+ }
+
+}