content: /*
   * 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.cassandra.utils.memory;
  
  import java.nio.ByteBuffer;
  import java.util.*;
  import java.util.concurrent.*;
  
  import com.google.common.collect.Iterables;
  import org.junit.Before;
  import org.junit.Ignore;
  import org.junit.Test;
  
  import org.apache.cassandra.io.compress.BufferType;
  import org.apache.cassandra.io.util.RandomAccessReader;
  
  import static org.junit.Assert.*;
  
  public class BufferPoolTest
  {
      private BufferPool bufferPool;
  
      @Before
      public void setUp()
      {
          bufferPool = new BufferPool("test_pool", 8 * 1024 * 1024, true);
      }
  
      @Test
      public void testGetPut()
      {
          final int size = RandomAccessReader.DEFAULT_BUFFER_SIZE;
  
          ByteBuffer buffer = bufferPool.get(size, BufferType.OFF_HEAP);
          assertNotNull(buffer);
          assertEquals(size, buffer.capacity());
          assertEquals(true, buffer.isDirect());
          assertEquals(size, bufferPool.usedSizeInBytes());
  
          BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
          assertNotNull(chunk);
          assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
  
          bufferPool.put(buffer);
          assertEquals(null, bufferPool.unsafeCurrentChunk());
          assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
          assertEquals(0, bufferPool.usedSizeInBytes());
      }
  
      @Test
      public void testTryGet()
      {
          final int size = RandomAccessReader.DEFAULT_BUFFER_SIZE;
  
          ByteBuffer buffer = bufferPool.tryGet(size);
          assertNotNull(buffer);
          assertEquals(size, buffer.capacity());
          assertEquals(true, buffer.isDirect());
          assertEquals(size, bufferPool.usedSizeInBytes());
  
          BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
          assertNotNull(chunk);
          assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
  
          bufferPool.put(buffer);
          assertEquals(null, bufferPool.unsafeCurrentChunk());
          assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
          assertEquals(0, bufferPool.usedSizeInBytes());
      }
  
      @Test
      public void testPageAligned()
      {
          final int size = 1024;
          for (int i = size;
                   i <= BufferPool.NORMAL_CHUNK_SIZE;
                   i += size)
          {
              checkPageAligned(i);
          }
      }
  
      private void checkPageAligned(int size)
      {
          ByteBuffer buffer = bufferPool.get(size, BufferType.OFF_HEAP);
          assertNotNull(buffer);
          assertEquals(size, buffer.capacity());
          assertTrue(buffer.isDirect());
  
          long address = MemoryUtil.getAddress(buffer);
          assertTrue((address % MemoryUtil.pageSize()) == 0);
  
          bufferPool.put(buffer);
      }
  
      @Test
      public void testDifferentSizes() throws InterruptedException
      {
          final int size1 = 1024;
          final int size2 = 2048;
  
          ByteBuffer buffer1 = bufferPool.get(size1, BufferType.OFF_HEAP);
          assertNotNull(buffer1);
          assertEquals(size1, buffer1.capacity());
          assertEquals(size1, bufferPool.usedSizeInBytes());
  
          ByteBuffer buffer2 = bufferPool.get(size2, BufferType.OFF_HEAP);
          assertNotNull(buffer2);
          assertEquals(size2, buffer2.capacity());
          assertEquals(size1 + size2, bufferPool.usedSizeInBytes());
  
          BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
          assertNotNull(chunk);
          assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
  
          bufferPool.put(buffer1);
          bufferPool.put(buffer2);
  
          assertEquals(null, bufferPool.unsafeCurrentChunk());
          assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
          assertEquals(0, bufferPool.usedSizeInBytes());
      }
  
      @Test
      public void testMaxMemoryExceededDirect()
      {
          requestDoubleMaxMemory();
      }
  
      @Test
      public void testMaxMemoryExceededHeap()
      {
          requestDoubleMaxMemory();
      }
  
      @Test
      public void testMaxMemoryExceeded_SameAsChunkSize()
      {
          requestDoubleMaxMemory();
      }
  
      @Test
      public void testMaxMemoryExceeded_SmallerThanChunkSize()
      {
          bufferPool = new BufferPool("test_pool", BufferPool.GlobalPool.MACRO_CHUNK_SIZE / 2, false);
          requestDoubleMaxMemory();
      }
  
      @Test
      public void testRecycle()
      {
          requestUpToSize(RandomAccessReader.DEFAULT_BUFFER_SIZE, 3 * BufferPool.NORMAL_CHUNK_SIZE);
      }
  
      private void requestDoubleMaxMemory()
      {
          requestUpToSize(RandomAccessReader.DEFAULT_BUFFER_SIZE, (int)(2 * bufferPool.memoryUsageThreshold()));
      }
  
      private void requestUpToSize(int bufferSize, int totalSize)
      {
          final int numBuffers = totalSize / bufferSize;
  
          List<ByteBuffer> buffers = new ArrayList<>(numBuffers);
          for (int i = 0; i < numBuffers; i++)
          {
              ByteBuffer buffer = bufferPool.get(bufferSize, BufferType.OFF_HEAP);
              assertNotNull(buffer);
              assertEquals(bufferSize, buffer.capacity());
              assertTrue(buffer.isDirect());
              buffers.add(buffer);
          }
  
          for (ByteBuffer buffer : buffers)
              bufferPool.put(buffer);
      }
  
      @Test
      public void testBigRequest()
      {
          final int size = BufferPool.NORMAL_CHUNK_SIZE + 1;
  
          ByteBuffer buffer = bufferPool.get(size, BufferType.OFF_HEAP);
          assertNotNull(buffer);
          assertEquals(size, buffer.capacity());
          bufferPool.put(buffer);
      }
  
      @Test
      public void testFillUpChunks()
      {
          final int size = RandomAccessReader.DEFAULT_BUFFER_SIZE;
          final int numBuffers = BufferPool.NORMAL_CHUNK_SIZE / size;
  
          List<ByteBuffer> buffers1 = new ArrayList<>(numBuffers);
          List<ByteBuffer> buffers2 = new ArrayList<>(numBuffers);
          for (int i = 0; i < numBuffers; i++)
              buffers1.add(bufferPool.get(size, BufferType.OFF_HEAP));
  
          BufferPool.Chunk chunk1 = bufferPool.unsafeCurrentChunk();
          assertNotNull(chunk1);
  
          for (int i = 0; i < numBuffers; i++)
              buffers2.add(bufferPool.get(size, BufferType.OFF_HEAP));
  
          assertEquals(2, bufferPool.unsafeNumChunks());
  
          for (ByteBuffer buffer : buffers1)
              bufferPool.put(buffer);
  
          assertEquals(1, bufferPool.unsafeNumChunks());
  
          for (ByteBuffer buffer : buffers2)
              bufferPool.put(buffer);
  
          assertEquals(0, bufferPool.unsafeNumChunks());
  
          buffers2.clear();
      }
  
      @Test
      public void testOutOfOrderFrees()
      {
          final int size = 4096;
          final int maxFreeSlots = BufferPool.NORMAL_CHUNK_SIZE / size;
  
          final int[] idxs = new int[maxFreeSlots];
          for (int i = 0; i < maxFreeSlots; i++)
              idxs[i] = i;
  
          doTestFrees(size, maxFreeSlots, idxs);
      }
  
      @Test
      public void testInOrderFrees()
      {
          final int size = 4096;
          final int maxFreeSlots = BufferPool.NORMAL_CHUNK_SIZE / size;
  
          final int[] idxs = new int[maxFreeSlots];
          for (int i = 0; i < maxFreeSlots; i++)
              idxs[i] = maxFreeSlots - 1 - i;
  
          doTestFrees(size, maxFreeSlots, idxs);
      }
  
      @Test
      public void testRandomFrees()
      {
          doTestRandomFrees(12345567878L);
  
          bufferPool.unsafeReset();
          doTestRandomFrees(20452249587L);
  
          bufferPool.unsafeReset();
          doTestRandomFrees(82457252948L);
  
          bufferPool.unsafeReset();
          doTestRandomFrees(98759284579L);
  
          bufferPool.unsafeReset();
          doTestRandomFrees(19475257244L);
      }
  
      private void doTestRandomFrees(long seed)
      {
          final int size = 4096;
          final int maxFreeSlots = BufferPool.NORMAL_CHUNK_SIZE / size;
  
          final int[] idxs = new int[maxFreeSlots];
          for (int i = 0; i < maxFreeSlots; i++)
              idxs[i] = maxFreeSlots - 1 - i;
  
          Random rnd = new Random();
          rnd.setSeed(seed);
          for (int i = idxs.length - 1; i > 0; i--)
          {
              int idx = rnd.nextInt(i+1);
              int v = idxs[idx];
              idxs[idx] = idxs[i];
              idxs[i] = v;
          }
  
          doTestFrees(size, maxFreeSlots, idxs);
      }
  
      private void doTestFrees(final int size, final int maxFreeSlots, final int[] toReleaseIdxs)
      {
          List<ByteBuffer> buffers = new ArrayList<>(maxFreeSlots);
          for (int i = 0; i < maxFreeSlots; i++)
          {
              buffers.add(bufferPool.get(size, BufferType.OFF_HEAP));
          }
  
          BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
          assertFalse(chunk.isFree());
  
          int freeSize = BufferPool.NORMAL_CHUNK_SIZE - maxFreeSlots * size;
          assertEquals(freeSize, chunk.free());
  
          for (int i : toReleaseIdxs)
          {
              ByteBuffer buffer = buffers.get(i);
              assertNotNull(buffer);
              assertEquals(size, buffer.capacity());
  
              bufferPool.put(buffer);
  
              freeSize += size;
              if (freeSize == chunk.capacity())
                  assertEquals(0, chunk.free());
              else
                  assertEquals(freeSize, chunk.free());
          }
  
          assertFalse(chunk.isFree());
      }
  
      @Test
      public void testDifferentSizeBuffersOnOneChunk()
      {
          int[] sizes = new int[] {
              5, 1024, 4096, 8, 16000, 78, 512, 256, 63, 55, 89, 90, 255, 32, 2048, 128
          };
  
          int sum = 0;
          List<ByteBuffer> buffers = new ArrayList<>(sizes.length);
          for (int i = 0; i < sizes.length; i++)
          {
              ByteBuffer buffer = bufferPool.get(sizes[i], BufferType.OFF_HEAP);
              assertNotNull(buffer);
              assertTrue(buffer.capacity() >= sizes[i]);
              buffers.add(buffer);
  
              sum += bufferPool.unsafeCurrentChunk().roundUp(buffer.capacity());
          }
  
          // else the test will fail, adjust sizes as required
          assertTrue(sum <= BufferPool.GlobalPool.MACRO_CHUNK_SIZE);
  
          BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
          assertNotNull(chunk);
  
          Random rnd = new Random();
          rnd.setSeed(298347529L);
          while (buffers.size() > 1)
          {
              int index = rnd.nextInt(buffers.size());
              ByteBuffer buffer = buffers.remove(index);
  
              bufferPool.put(buffer);
          }
          bufferPool.put(buffers.remove(0));
  
          assertEquals(null, bufferPool.unsafeCurrentChunk());
          assertEquals(0, chunk.free());
      }
  
      @Test
      public void testChunkExhausted()
      {
          final int size = BufferPool.NORMAL_CHUNK_SIZE / 64; // 1kibibit
          int[] sizes = new int[128];
          Arrays.fill(sizes, size);
  
          int sum = 0;
          List<ByteBuffer> buffers = new ArrayList<>(sizes.length);
          for (int i = 0; i < sizes.length; i++)
          {
              ByteBuffer buffer = bufferPool.get(sizes[i], BufferType.OFF_HEAP);
              assertNotNull(buffer);
              assertTrue(buffer.capacity() >= sizes[i]);
              buffers.add(buffer);
  
              sum += buffer.capacity();
          }
  
          // else the test will fail, adjust sizes as required
          assertTrue(sum <= BufferPool.GlobalPool.MACRO_CHUNK_SIZE);
  
          BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
          assertNotNull(chunk);
  
          for (int i = 0; i < sizes.length; i++)
          {
              bufferPool.put(buffers.get(i));
          }
  
          assertEquals(null, bufferPool.unsafeCurrentChunk());
          assertEquals(0, chunk.free());
      }
  
      @Test
      public void testCompactIfOutOfCapacity()
      {
          final int size = 4096;
          final int numBuffersInChunk = BufferPool.GlobalPool.MACRO_CHUNK_SIZE / size;
  
          List<ByteBuffer> buffers = new ArrayList<>(numBuffersInChunk);
          Set<Long> addresses = new HashSet<>(numBuffersInChunk);
  
          for (int i = 0; i < numBuffersInChunk; i++)
          {
              ByteBuffer buffer = bufferPool.get(size, BufferType.OFF_HEAP);
              buffers.add(buffer);
              addresses.add(MemoryUtil.getAddress(buffer));
          }
  
          for (int i = numBuffersInChunk - 1; i >= 0; i--)
              bufferPool.put(buffers.get(i));
  
          buffers.clear();
  
          for (int i = 0; i < numBuffersInChunk; i++)
          {
              ByteBuffer buffer = bufferPool.get(size, BufferType.OFF_HEAP);
              assertNotNull(buffer);
              assertEquals(size, buffer.capacity());
              assert addresses.remove(MemoryUtil.getAddress(buffer));
  
              buffers.add(buffer);
          }
  
          assertTrue(addresses.isEmpty()); // all 5 released buffers were used
  
          for (ByteBuffer buffer : buffers)
              bufferPool.put(buffer);
      }
  
      @Test
      public void testHeapBuffer()
      {
          ByteBuffer buffer = bufferPool.get(1024, BufferType.ON_HEAP);
          assertNotNull(buffer);
          assertEquals(1024, buffer.capacity());
          assertFalse(buffer.isDirect());
          assertNotNull(buffer.array());
          bufferPool.put(buffer);
      }
  
      @Test
      public void testSingleBufferOneChunk()
      {
          checkBuffer(0);
  
          checkBuffer(1);
          checkBuffer(2);
          checkBuffer(4);
          checkBuffer(5);
          checkBuffer(8);
          checkBuffer(16);
          checkBuffer(32);
          checkBuffer(64);
  
          checkBuffer(65);
          checkBuffer(127);
          checkBuffer(128);
  
          checkBuffer(129);
          checkBuffer(255);
          checkBuffer(256);
  
          checkBuffer(512);
          checkBuffer(1024);
          checkBuffer(2048);
          checkBuffer(4096);
          checkBuffer(8192);
          checkBuffer(16384);
  
          checkBuffer(16385);
          checkBuffer(32767);
          checkBuffer(32768);
  
          checkBuffer(32769);
          checkBuffer(33172);
          checkBuffer(33553);
          checkBuffer(36000);
          checkBuffer(65535);
          checkBuffer(65536);
  
          checkBuffer(65537);
      }
  
      private void checkBuffer(int size)
      {
          ByteBuffer buffer = bufferPool.get(size, BufferType.OFF_HEAP);
          assertEquals(size, buffer.capacity());
  
          if (size > 0 && size < BufferPool.NORMAL_CHUNK_SIZE)
          {
              BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
              assertNotNull(chunk);
              assertEquals(chunk.capacity(), chunk.free() + chunk.roundUp(size));
          }
  
          bufferPool.put(buffer);
      }
  
      @Test
      public void testMultipleBuffersOneChunk()
      {
          checkBuffers(32768, 33553);
          checkBuffers(32768, 32768);
          checkBuffers(48450, 33172);
          checkBuffers(32768, 15682, 33172);
      }
  
      private void checkBuffers(int ... sizes)
      {
          List<ByteBuffer> buffers = new ArrayList<>(sizes.length);
  
          for (int size : sizes)
          {
              ByteBuffer buffer = bufferPool.get(size, BufferType.OFF_HEAP);
              assertEquals(size, buffer.capacity());
  
              buffers.add(buffer);
          }
  
          for (ByteBuffer buffer : buffers)
              bufferPool.put(buffer);
      }
  
      @Test
      public void testBuffersWithGivenSlots()
      {
          checkBufferWithGivenSlots(21241, (-1L << 27) ^ (1L << 40));
      }
  
      private void checkBufferWithGivenSlots(int size, long freeSlots)
      {
          //first allocate to make sure there is a chunk
          ByteBuffer buffer = bufferPool.get(size, BufferType.OFF_HEAP);
  
          // now get the current chunk and override the free slots mask
          BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
          assertNotNull(chunk);
          long oldFreeSlots = chunk.setFreeSlots(freeSlots);
  
          // now check we can still get the buffer with the free slots mask changed
          ByteBuffer buffer2 = bufferPool.get(size, BufferType.OFF_HEAP);
          assertEquals(size, buffer.capacity());
          bufferPool.put(buffer2);
  
          // unsafeReset the free slots
          chunk.setFreeSlots(oldFreeSlots);
          bufferPool.put(buffer);
      }
  
      @Test
      public void testZeroSizeRequest()
      {
          ByteBuffer buffer = bufferPool.get(0, BufferType.OFF_HEAP);
          assertNotNull(buffer);
          assertEquals(0, buffer.capacity());
          bufferPool.put(buffer);
      }
  
      @Test(expected = IllegalArgumentException.class)
      public void testNegativeSizeRequest()
      {
          bufferPool.get(-1, BufferType.OFF_HEAP);
      }
  
      @Test
      public void testMT_SameSizeImmediateReturn() throws InterruptedException
      {
          checkMultipleThreads(40, 1, true, RandomAccessReader.DEFAULT_BUFFER_SIZE);
      }
  
      @Test
      public void testMT_SameSizePostponedReturn() throws InterruptedException
      {
          checkMultipleThreads(40, 1, false, RandomAccessReader.DEFAULT_BUFFER_SIZE);
      }
  
      @Test
      public void testMT_TwoSizesOneBufferImmediateReturn() throws InterruptedException
      {
          checkMultipleThreads(40, 1, true, 1024, 2048);
      }
  
      @Test
      public void testMT_TwoSizesOneBufferPostponedReturn() throws InterruptedException
      {
          checkMultipleThreads(40, 1, false, 1024, 2048);
      }
  
      @Test
      public void testMT_TwoSizesTwoBuffersImmediateReturn() throws InterruptedException
      {
          checkMultipleThreads(40, 2, true, 1024, 2048);
      }
  
      @Test
      public void testMT_TwoSizesTwoBuffersPostponedReturn() throws InterruptedException
      {
          checkMultipleThreads(40, 2, false, 1024, 2048);
      }
  
      @Test
      public void testMT_MultipleSizesOneBufferImmediateReturn() throws InterruptedException
      {
          checkMultipleThreads(40,
                               1,
                               true,
                               1024,
                               2048,
                               3072,
                               4096,
                               5120);
      }
  
      @Test
      public void testMT_MultipleSizesOneBufferPostponedReturn() throws InterruptedException
      {
          checkMultipleThreads(40,
                               1,
                               false,
                               1024,
                               2048,
                               3072,
                               4096,
                               5120);
      }
  
      @Test
      public void testMT_MultipleSizesMultipleBuffersImmediateReturn() throws InterruptedException
      {
          checkMultipleThreads(40,
                               4,
                               true,
                               1024,
                               2048,
                               3072,
                               4096,
                               5120);
      }
  
      @Test
      public void testMT_MultipleSizesMultipleBuffersPostponedReturn() throws InterruptedException
      {
          checkMultipleThreads(40,
                               3,
                               false,
                               1024,
                               2048,
                               3072,
                               4096,
                               5120);
      }
  
      private void checkMultipleThreads(int threadCount, int numBuffersPerThread, final boolean returnImmediately, final int ... sizes) throws InterruptedException
      {
          ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
          final CountDownLatch finished = new CountDownLatch(threadCount);
  
          for (int i = 0; i < threadCount; i++)
          {
              final int[] threadSizes = new int[numBuffersPerThread];
              for (int j = 0; j < threadSizes.length; j++)
                  threadSizes[j] = sizes[(i * numBuffersPerThread + j) % sizes.length];
  
              final Random rand = new Random();
              executorService.submit(new Runnable()
              {
                  @Override
                  public void run()
                  {
                      try
                      {
                          Thread.sleep(rand.nextInt(3));
  
                          List<ByteBuffer> toBeReturned = new ArrayList<ByteBuffer>(threadSizes.length);
  
                          for (int j = 0; j < threadSizes.length; j++)
                          {
                              ByteBuffer buffer = bufferPool.get(threadSizes[j], BufferType.OFF_HEAP);
                              assertNotNull(buffer);
                              assertEquals(threadSizes[j], buffer.capacity());
  
                              for (int i = 0; i < 10; i++)
                                  buffer.putInt(i);
  
                              buffer.rewind();
  
                              Thread.sleep(rand.nextInt(3));
  
                              for (int i = 0; i < 10; i++)
                                  assertEquals(i, buffer.getInt());
  
                              if (returnImmediately)
                                  bufferPool.put(buffer);
                              else
                                  toBeReturned.add(buffer);
  
                              assertTrue(bufferPool.sizeInBytes() > 0);
                          }
  
                          Thread.sleep(rand.nextInt(3));
  
                          for (ByteBuffer buffer : toBeReturned)
                              bufferPool.put(buffer);
                      }
                      catch (Exception ex)
                      {
                          ex.printStackTrace();
                          fail(ex.getMessage());
                      }
                      finally
                      {
                          finished.countDown();
                      }
                  }
              });
          }
  
          finished.await();
          assertEquals(0, executorService.shutdownNow().size());
  
          // Make sure thread local storage gets GC-ed
          for (int i = 0; i < 5; i++)
          {
              System.gc();
              Thread.sleep(100);
          }
      }
  
      @Ignore
      public void testMultipleThreadsReleaseSameBuffer() throws InterruptedException
      {
          doMultipleThreadsReleaseBuffers(45, 4096);
      }
  
      @Ignore
      public void testMultipleThreadsReleaseDifferentBuffer() throws InterruptedException
      {
          doMultipleThreadsReleaseBuffers(45, 4096, 8192);
      }
  
      private void doMultipleThreadsReleaseBuffers(final int threadCount, final int ... sizes) throws InterruptedException
      {
          final ByteBuffer[] buffers = new ByteBuffer[sizes.length];
          int sum = 0;
          for (int i = 0; i < sizes.length; i++)
          {
              buffers[i] = bufferPool.get(sizes[i], BufferType.OFF_HEAP);
              assertNotNull(buffers[i]);
              assertEquals(sizes[i], buffers[i].capacity());
              sum += bufferPool.unsafeCurrentChunk().roundUp(buffers[i].capacity());
          }
  
          final BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
          assertNotNull(chunk);
          assertFalse(chunk.isFree());
  
          // if we use multiple chunks the test will fail, adjust sizes accordingly
          assertTrue(sum < BufferPool.GlobalPool.MACRO_CHUNK_SIZE);
  
          ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
          final CountDownLatch finished = new CountDownLatch(threadCount);
  
          for (int i = 0; i < threadCount; i++)
          {
              final int idx = i % sizes.length;
              final ByteBuffer buffer = buffers[idx];
  
              executorService.submit(new Runnable()
              {
                  @Override
                  public void run()
                  {
                      try
                      {
                          assertNotSame(chunk, bufferPool.unsafeCurrentChunk());
                          bufferPool.put(buffer);
                      }
                      catch (AssertionError ex)
                      { //this is expected if we release a buffer more than once
                          ex.printStackTrace();
                      }
                      catch (Throwable t)
                      {
                          t.printStackTrace();
                          fail(t.getMessage());
                      }
                      finally
                      {
                          finished.countDown();
                      }
                  }
              });
          }
  
          finished.await();
          assertEquals(0, executorService.shutdownNow().size());
  
          executorService = null;
  
          // Make sure thread local storage gets GC-ed
          System.gc();
          System.gc();
          System.gc();
  
          assertTrue(bufferPool.unsafeCurrentChunk().isFree());
  
          //make sure the main thread can still allocate buffers
          ByteBuffer buffer = bufferPool.get(sizes[0], BufferType.OFF_HEAP);
          assertNotNull(buffer);
          assertEquals(sizes[0], buffer.capacity());
          bufferPool.put(buffer);
      }
  
      @Test
      public void testOverflowAllocation()
      {
          int macroChunkSize = BufferPool.GlobalPool.MACRO_CHUNK_SIZE;
          int allocationSize = BufferPool.NORMAL_CHUNK_SIZE;
          int allocations = BufferPool.GlobalPool.MACRO_CHUNK_SIZE / allocationSize;
  
          // occupy entire buffer pool
          List<ByteBuffer> buffers = new ArrayList<>();
          allocate(allocations, allocationSize, buffers);
  
          assertEquals(macroChunkSize, bufferPool.sizeInBytes());
          assertEquals(macroChunkSize, bufferPool.usedSizeInBytes());
          assertEquals(0, bufferPool.overflowMemoryInBytes());
  
          // allocate overflow due to pool exhaust
          ByteBuffer overflowBuffer = bufferPool.get(BufferPool.NORMAL_ALLOCATION_UNIT, BufferType.OFF_HEAP);
  
          assertEquals(macroChunkSize + overflowBuffer.capacity(), bufferPool.sizeInBytes());
          assertEquals(macroChunkSize + overflowBuffer.capacity(), bufferPool.usedSizeInBytes());
          assertEquals(overflowBuffer.capacity(), bufferPool.overflowMemoryInBytes());
  
          // free all buffer
          bufferPool.put(overflowBuffer);
          release(buffers);
  
          assertEquals(macroChunkSize, bufferPool.sizeInBytes());
          assertEquals(0, bufferPool.usedSizeInBytes());
          assertEquals(0, bufferPool.overflowMemoryInBytes());
  
          // allocate overflow due to on-heap
          overflowBuffer = bufferPool.get(BufferPool.NORMAL_ALLOCATION_UNIT, BufferType.ON_HEAP);
          assertEquals(macroChunkSize + overflowBuffer.capacity(), bufferPool.sizeInBytes());
          assertEquals(overflowBuffer.capacity(), bufferPool.usedSizeInBytes());
          assertEquals(overflowBuffer.capacity(), bufferPool.overflowMemoryInBytes());
          bufferPool.put(overflowBuffer);
  
          // allocate overflow due to over allocation size
          overflowBuffer = bufferPool.get(2 * BufferPool.NORMAL_CHUNK_SIZE, BufferType.ON_HEAP);
          assertEquals(macroChunkSize + overflowBuffer.capacity(), bufferPool.sizeInBytes());
          assertEquals(overflowBuffer.capacity(), bufferPool.usedSizeInBytes());
          assertEquals(overflowBuffer.capacity(), bufferPool.overflowMemoryInBytes());
      }
  
      @Test
      public void testRecyclePartialFreeChunk()
      {
          // normal chunk size is 128KiB
          int halfNormalChunk = BufferPool.NORMAL_CHUNK_SIZE / 2; // 64KiB, half of normal chunk
          List<ByteBuffer> toRelease = new ArrayList<>();
  
          // allocate three buffers on different chunks
          ByteBuffer buffer0 = bufferPool.get(halfNormalChunk, BufferType.OFF_HEAP);
          BufferPool.Chunk chunk0 = BufferPool.Chunk.getParentChunk(buffer0);
          assertFalse(chunk0.isFree());
          allocate(1, halfNormalChunk, toRelease); // allocate remaining buffers in the chunk
  
          ByteBuffer buffer1 = bufferPool.get(halfNormalChunk, BufferType.OFF_HEAP);
          BufferPool.Chunk chunk1 = BufferPool.Chunk.getParentChunk(buffer1);
          assertFalse(chunk1.isFree());
          assertNotEquals(chunk0, chunk1);
          allocate(1, halfNormalChunk, toRelease); // allocate remaining buffers in the chunk
  
          ByteBuffer buffer2 = bufferPool.get(halfNormalChunk, BufferType.OFF_HEAP);
          BufferPool.Chunk chunk2 = BufferPool.Chunk.getParentChunk(buffer2);
          assertFalse(chunk2.isFree());
          assertNotEquals(chunk0, chunk2);
          assertNotEquals(chunk1, chunk2);
          allocate(1, halfNormalChunk, toRelease); // allocate remaining buffers in the chunk
  
          // now all 3 chunks in local pool is full, allocate one more buffer to evict chunk2
          ByteBuffer buffer3 = bufferPool.get(halfNormalChunk, BufferType.OFF_HEAP);
          BufferPool.Chunk chunk3 = BufferPool.Chunk.getParentChunk(buffer3);
          assertNotEquals(chunk0, chunk3);
          assertNotEquals(chunk1, chunk3);
          assertNotEquals(chunk2, chunk3);
  
          // verify chunk2 got evicted, it doesn't have a owner
          assertNotNull(chunk0.owner());
          assertEquals(BufferPool.Chunk.Status.IN_USE, chunk0.status());
          assertNotNull(chunk1.owner());
          assertEquals(BufferPool.Chunk.Status.IN_USE, chunk1.status());
          assertNull(chunk2.owner());
          assertEquals(BufferPool.Chunk.Status.EVICTED, chunk2.status());
  
          // release half buffers for chunk0/1/2
          release(toRelease);
          BufferPool.Chunk partiallyFreed = chunk2;
  
          // try to recirculate chunk2 and verify freed space
          assertFalse(bufferPool.globalPool().isFullyFreed(partiallyFreed));
          assertTrue(bufferPool.globalPool().isPartiallyFreed(partiallyFreed));
          assertEquals(BufferPool.Chunk.Status.IN_USE, partiallyFreed.status());
          assertEquals(halfNormalChunk, partiallyFreed.free());
          ByteBuffer buffer = partiallyFreed.get(halfNormalChunk, false, null);
          assertEquals(halfNormalChunk, buffer.capacity());
  
          // cleanup allocated buffers
          for (ByteBuffer buf : Arrays.asList(buffer0, buffer1, buffer2, buffer3, buffer))
              bufferPool.put(buf);
  
          // verify that fully freed chunk are prioritized over partially freed chunks
          List<BufferPool.Chunk> remainingChunks = new ArrayList<>();
          BufferPool.Chunk chunkForAllocation;
          while ((chunkForAllocation = bufferPool.globalPool().get()) != null)
              remainingChunks.add(chunkForAllocation);
  
          int totalNormalChunks = BufferPool.GlobalPool.MACRO_CHUNK_SIZE / BufferPool.NORMAL_CHUNK_SIZE; // 64;
          assertEquals(totalNormalChunks, remainingChunks.size());
          assertSame(partiallyFreed, remainingChunks.get(remainingChunks.size() - 1)); // last one is partially freed
  
          // cleanup polled chunks
          remainingChunks.forEach(BufferPool.Chunk::release);
      }
  
      @Test
      public void testTinyPool()
      {
          int total = 0;
          final int size = BufferPool.TINY_ALLOCATION_UNIT;
          final int allocationPerChunk = 64;
  
          // occupy 3 tiny chunks
          List<ByteBuffer> buffers0 = new ArrayList<>();
          BufferPool.Chunk chunk0 = allocate(allocationPerChunk, size, buffers0);
          assertTrue(chunk0.owner().isTinyPool());
          List<ByteBuffer> buffers1 = new ArrayList<>();
          BufferPool.Chunk chunk1 = allocate(allocationPerChunk, size, buffers1);
          assertTrue(chunk1.owner().isTinyPool());
          List<ByteBuffer> buffers2 = new ArrayList<>();
          BufferPool.Chunk chunk2 = allocate(allocationPerChunk, size, buffers2);
          assertTrue(chunk2.owner().isTinyPool());
          total += 3 * BufferPool.TINY_CHUNK_SIZE;
          assertEquals(total, bufferPool.usedSizeInBytes());
  
          // allocate another tiny chunk.. chunk2 should be evicted
          List<ByteBuffer> buffers3 = new ArrayList<>();
          BufferPool.Chunk chunk3 = allocate(allocationPerChunk, size, buffers3);
          assertTrue(chunk3.owner().isTinyPool());
          total += BufferPool.TINY_CHUNK_SIZE;
          assertEquals(total, bufferPool.usedSizeInBytes());
  
          // verify chunk2 is full and evicted
          assertEquals(0, chunk2.free());
          assertNull(chunk2.owner());
  
          // release chunk2's buffer
          for (int i = 0; i < buffers2.size(); i++)
          {
              bufferPool.put(buffers2.get(i));
              total -= buffers2.get(i).capacity();
              assertEquals(total, bufferPool.usedSizeInBytes());
          }
  
          // cleanup allocated buffers
          for (ByteBuffer buffer : Iterables.concat(buffers0, buffers1, buffers3))
              bufferPool.put(buffer);
      }
  
      @Test
      public void testReleaseLocal()
      {
          final int size = BufferPool.TINY_ALLOCATION_UNIT;
          final int allocationPerChunk = 64;
  
          // occupy 3 tiny chunks
          List<ByteBuffer> buffers0 = new ArrayList<>();
          BufferPool.Chunk chunk0 = allocate(allocationPerChunk, size, buffers0);
          List<ByteBuffer> buffers1 = new ArrayList<>();
          BufferPool.Chunk chunk1 = allocate(allocationPerChunk, size, buffers1);
          List<ByteBuffer> buffers2 = new ArrayList<>();
          BufferPool.Chunk chunk2 = allocate(allocationPerChunk, size, buffers2);
  
          // release them from the pool
          bufferPool.releaseLocal();
  
          assertNull(chunk0.owner());
          assertNull(chunk1.owner());
          assertNull(chunk2.owner());
          assertEquals(BufferPool.Chunk.Status.EVICTED, chunk0.status());
          assertEquals(BufferPool.Chunk.Status.EVICTED, chunk1.status());
          assertEquals(BufferPool.Chunk.Status.EVICTED, chunk2.status());
  
          // cleanup allocated buffers, that should still work fine even though we released them from the localPool
          for (ByteBuffer buffer : Iterables.concat(buffers0, buffers1, buffers2))
              bufferPool.put(buffer);
  
          assertEquals(0, bufferPool.usedSizeInBytes());
      }
  
      @Test
      public void testPuttingUnusedPortion()
      {
          final int expectedCapacity = BufferPool.TINY_ALLOCATION_UNIT * 4;
          final int quarterUnit = BufferPool.TINY_ALLOCATION_UNIT / 4;
          final int requestedCapacity = expectedCapacity - 3 * quarterUnit;
  
          ByteBuffer buffer = bufferPool.getAtLeast(requestedCapacity, BufferType.OFF_HEAP);
          assertNotNull(buffer);
          assertEquals(expectedCapacity, buffer.capacity());
          assertEquals(expectedCapacity, bufferPool.usedSizeInBytes());
  
          buffer.limit(requestedCapacity); // 3.25 x unit
          bufferPool.putUnusedPortion(buffer);
  
          // the unused portion was too small to be returned, the buffer remains unchanged
          assertEquals(expectedCapacity, buffer.capacity());
          // used size is didn't change either
          assertEquals(expectedCapacity, bufferPool.usedSizeInBytes());
  
          buffer.limit(expectedCapacity - BufferPool.TINY_ALLOCATION_UNIT); // 3.0 x unit
          bufferPool.putUnusedPortion(buffer);
  
          // now we should notice a change
          assertEquals(BufferPool.TINY_ALLOCATION_UNIT * 3, buffer.capacity());
          assertEquals(BufferPool.TINY_ALLOCATION_UNIT * 3, bufferPool.usedSizeInBytes());
  
          bufferPool.put(buffer);
  
          assertEquals(0, bufferPool.usedSizeInBytes());
      }
  
      private BufferPool.Chunk allocate(int num, int bufferSize, List<ByteBuffer> buffers)
      {
          for (int i = 0; i < num; i++)
              buffers.add(bufferPool.get(bufferSize, BufferType.OFF_HEAP));
  
          return BufferPool.Chunk.getParentChunk(buffers.get(buffers.size() - 1));
      }
  
      private void release(List<ByteBuffer> toRelease)
      {
          for (ByteBuffer buffer : toRelease)
              bufferPool.put(buffer);
      }
  }
  