http://git-wip-us.apache.org/repos/asf/cassandra/blob/17dd4ccc/src/java/org/apache/cassandra/utils/memory/BufferPool.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java new file mode 100644 index 0000000..0731301 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java @@ -0,0 +1,858 @@ +/* + * 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.lang.ref.PhantomReference; +import java.lang.ref.ReferenceQueue; +import java.nio.ByteBuffer; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; + +import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.io.compress.BufferType; +import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.utils.NoSpamLogger; + +import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.metrics.BufferPoolMetrics; +import org.apache.cassandra.utils.concurrent.Ref; + +/** + * A pool of ByteBuffers that can be recycled. + */ +public class BufferPool +{ + /** The size of a page aligned buffer, 64kbit */ + static final int CHUNK_SIZE = 64 << 10; + + @VisibleForTesting + public static long MEMORY_USAGE_THRESHOLD = DatabaseDescriptor.getFileCacheSizeInMB() * 1024L * 1024L; + + @VisibleForTesting + public static boolean ALLOCATE_ON_HEAP_WHEN_EXAHUSTED = DatabaseDescriptor.getBufferPoolUseHeapIfExhausted(); + + @VisibleForTesting + public static boolean DISABLED = Boolean.parseBoolean(System.getProperty("cassandra.test.disable_buffer_pool", "false")); + + @VisibleForTesting + public static boolean DEBUG = false; + + private static final Logger logger = LoggerFactory.getLogger(BufferPool.class); + private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 15L, TimeUnit.MINUTES); + private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocateDirect(0); + + /** A global pool of chunks (page aligned buffers) */ + private static final GlobalPool globalPool = new GlobalPool(); + + /** A thread local pool of chunks, where chunks come from the global pool */ + private static final ThreadLocal<LocalPool> localPool = new ThreadLocal<LocalPool>() { + @Override + protected LocalPool initialValue() + { + return new LocalPool(); + } + }; + + public static ByteBuffer get(int size) + { + if (DISABLED) + return allocate(size, ALLOCATE_ON_HEAP_WHEN_EXAHUSTED); + else + return takeFromPool(size, ALLOCATE_ON_HEAP_WHEN_EXAHUSTED); + } + + public static ByteBuffer get(int size, BufferType bufferType) + { + boolean direct = bufferType == BufferType.OFF_HEAP; + if (DISABLED | !direct) + return allocate(size, !direct); + else + return takeFromPool(size, !direct); + } + + /** Unlike the get methods, this will return null if the pool is exhausted */ + public static ByteBuffer tryGet(int size) + { + if (DISABLED) + return allocate(size, ALLOCATE_ON_HEAP_WHEN_EXAHUSTED); + else + return maybeTakeFromPool(size, ALLOCATE_ON_HEAP_WHEN_EXAHUSTED); + } + + private static ByteBuffer allocate(int size, boolean onHeap) + { + return onHeap + ? ByteBuffer.allocate(size) + : ByteBuffer.allocateDirect(size); + } + + private static ByteBuffer takeFromPool(int size, boolean allocateOnHeapWhenExhausted) + { + ByteBuffer ret = maybeTakeFromPool(size, allocateOnHeapWhenExhausted); + if (ret != null) + return ret; + + if (logger.isTraceEnabled()) + logger.trace("Requested buffer size {} has been allocated directly due to lack of capacity", size); + + return localPool.get().allocate(size, allocateOnHeapWhenExhausted); + } + + private static ByteBuffer maybeTakeFromPool(int size, boolean allocateOnHeapWhenExhausted) + { + if (size < 0) + throw new IllegalArgumentException("Size must be positive (" + size + ")"); + + if (size == 0) + return EMPTY_BUFFER; + + if (size > CHUNK_SIZE) + { + if (logger.isTraceEnabled()) + logger.trace("Requested buffer size {} is bigger than {}, allocating directly", size, CHUNK_SIZE); + + return localPool.get().allocate(size, allocateOnHeapWhenExhausted); + } + + return localPool.get().get(size); + } + + public static void put(ByteBuffer buffer) + { + if (!(DISABLED | buffer.hasArray())) + localPool.get().put(buffer); + } + + /** This is not thread safe and should only be used for unit testing. */ + @VisibleForTesting + static void reset() + { + localPool.get().reset(); + globalPool.reset(); + } + + @VisibleForTesting + static Chunk currentChunk() + { + return localPool.get().chunks[0]; + } + + @VisibleForTesting + static int numChunks() + { + int ret = 0; + for (Chunk chunk : localPool.get().chunks) + { + if (chunk != null) + ret++; + } + return ret; + } + + @VisibleForTesting + static void assertAllRecycled() + { + globalPool.debug.check(); + } + + public static long sizeInBytes() + { + return globalPool.sizeInBytes(); + } + + static final class Debug + { + long recycleRound = 1; + final Queue<Chunk> allChunks = new ConcurrentLinkedQueue<>(); + void register(Chunk chunk) + { + allChunks.add(chunk); + } + void recycle(Chunk chunk) + { + chunk.lastRecycled = recycleRound; + } + void check() + { + for (Chunk chunk : allChunks) + assert chunk.lastRecycled == recycleRound; + recycleRound++; + } + } + + /** + * A queue of page aligned buffers, the chunks, which have been sliced from bigger chunks, + * the macro-chunks, also page aligned. Macro-chunks are allocated as long as we have not exceeded the + * memory maximum threshold, MEMORY_USAGE_THRESHOLD and are never released. + * + * This class is shared by multiple thread local pools and must be thread-safe. + */ + static final class GlobalPool + { + /** The size of a bigger chunk, 1-mbit, must be a multiple of CHUNK_SIZE */ + static final int MACRO_CHUNK_SIZE = 1 << 20; + + static + { + assert Integer.bitCount(CHUNK_SIZE) == 1; // must be a power of 2 + assert Integer.bitCount(MACRO_CHUNK_SIZE) == 1; // must be a power of 2 + assert MACRO_CHUNK_SIZE % CHUNK_SIZE == 0; // must be a multiple + + if (DISABLED) + logger.info("Global buffer pool is disabled, allocating {}", ALLOCATE_ON_HEAP_WHEN_EXAHUSTED ? "on heap" : "off heap"); + else + logger.info("Global buffer pool is enabled, when pool is exahusted (max is {} mb) it will allocate {}", + MEMORY_USAGE_THRESHOLD / (1024L * 1024L), + ALLOCATE_ON_HEAP_WHEN_EXAHUSTED ? "on heap" : "off heap"); + } + + private final Debug debug = new Debug(); + private final Queue<Chunk> macroChunks = new ConcurrentLinkedQueue<>(); + // TODO (future): it would be preferable to use a CLStack to improve cache occupancy; it would also be preferable to use "CoreLocal" storage + private final Queue<Chunk> chunks = new ConcurrentLinkedQueue<>(); + private final AtomicLong memoryUsage = new AtomicLong(); + + /** Return a chunk, the caller will take owership of the parent chunk. */ + public Chunk get() + { + while (true) + { + Chunk chunk = chunks.poll(); + if (chunk != null) + return chunk; + + if (!allocateMoreChunks()) + // give it one last attempt, in case someone else allocated before us + return chunks.poll(); + } + } + + /** + * This method might be called by multiple threads and that's fine if we add more + * than one chunk at the same time as long as we don't exceed the MEMORY_USAGE_THRESHOLD. + */ + private boolean allocateMoreChunks() + { + while (true) + { + long cur = memoryUsage.get(); + if (cur + MACRO_CHUNK_SIZE > MEMORY_USAGE_THRESHOLD) + { + noSpamLogger.info("Maximum memory usage reached ({} bytes), cannot allocate chunk of {} bytes", + MEMORY_USAGE_THRESHOLD, MACRO_CHUNK_SIZE); + return false; + } + if (memoryUsage.compareAndSet(cur, cur + MACRO_CHUNK_SIZE)) + break; + } + + // allocate a large chunk + Chunk chunk = new Chunk(allocateDirectAligned(MACRO_CHUNK_SIZE)); + chunk.acquire(null); + macroChunks.add(chunk); + for (int i = 0 ; i < MACRO_CHUNK_SIZE ; i += CHUNK_SIZE) + { + Chunk add = new Chunk(chunk.get(CHUNK_SIZE)); + chunks.add(add); + if (DEBUG) + debug.register(add); + } + + return true; + } + + public void recycle(Chunk chunk) + { + chunks.add(chunk); + } + + public long sizeInBytes() + { + return memoryUsage.get(); + } + + /** This is not thread safe and should only be used for unit testing. */ + @VisibleForTesting + void reset() + { + while (!chunks.isEmpty()) + chunks.poll().reset(); + + while (!macroChunks.isEmpty()) + macroChunks.poll().reset(); + + memoryUsage.set(0); + } + } + + /** + * A thread local class that grabs chunks from the global pool for this thread allocations. + * Only one thread can do the allocations but multiple threads can release the allocations. + */ + static final class LocalPool + { + private final static BufferPoolMetrics metrics = new BufferPoolMetrics(); + // a microqueue of Chunks: + // * if any are null, they are at the end; + // * new Chunks are added to the last null index + // * if no null indexes available, the smallest is swapped with the last index, and this replaced + // * this results in a queue that will typically be visited in ascending order of available space, so that + // small allocations preferentially slice from the Chunks with the smallest space available to furnish them + // WARNING: if we ever change the size of this, we must update removeFromLocalQueue, and addChunk + private final Chunk[] chunks = new Chunk[3]; + private byte chunkCount = 0; + + public LocalPool() + { + localPoolReferences.add(new LocalPoolRef(this, localPoolRefQueue)); + } + + private Chunk addChunkFromGlobalPool() + { + Chunk chunk = globalPool.get(); + if (chunk == null) + return null; + + addChunk(chunk); + return chunk; + } + + private void addChunk(Chunk chunk) + { + chunk.acquire(this); + + if (chunkCount < 3) + { + chunks[chunkCount++] = chunk; + return; + } + + int smallestChunkIdx = 0; + if (chunks[1].free() < chunks[0].free()) + smallestChunkIdx = 1; + if (chunks[2].free() < chunks[smallestChunkIdx].free()) + smallestChunkIdx = 2; + + chunks[smallestChunkIdx].release(); + if (smallestChunkIdx != 2) + chunks[smallestChunkIdx] = chunks[2]; + chunks[2] = chunk; + } + + public ByteBuffer get(int size) + { + for (Chunk chunk : chunks) + { // first see if our own chunks can serve this buffer + if (chunk == null) + break; + + ByteBuffer buffer = chunk.get(size); + if (buffer != null) + return buffer; + } + + // else ask the global pool + Chunk chunk = addChunkFromGlobalPool(); + if (chunk != null) + return chunk.get(size); + + return null; + } + + private ByteBuffer allocate(int size, boolean onHeap) + { + metrics.misses.mark(); + return BufferPool.allocate(size, onHeap); + } + + public void put(ByteBuffer buffer) + { + Chunk chunk = Chunk.getParentChunk(buffer); + if (chunk == null) + { + FileUtils.clean(buffer); + return; + } + + LocalPool owner = chunk.owner; + // ask the free method to take exclusive ownership of the act of recycling + // if we are either: already not owned by anyone, or owned by ourselves + long free = chunk.free(buffer, owner == null | owner == this); + if (free == 0L) + { + // 0L => we own recycling responsibility, so must recycle; + chunk.recycle(); + // if we are also the owner, we must remove the Chunk from our local queue + if (owner == this) + removeFromLocalQueue(chunk); + } + else if (((free == -1L) & owner != this) && chunk.owner == null) + { + // although we try to take recycle ownership cheaply, it is not always possible to do so if the owner is racing to unset. + // we must also check after completely freeing if the owner has since been unset, and try to recycle + chunk.tryRecycle(); + } + } + + private void removeFromLocalQueue(Chunk chunk) + { + // since we only have three elements in the queue, it is clearer, easier and faster to just hard code the options + if (chunks[0] == chunk) + { // remove first by shifting back second two + chunks[0] = chunks[1]; + chunks[1] = chunks[2]; + } + else if (chunks[1] == chunk) + { // remove second by shifting back last + chunks[1] = chunks[2]; + } + else assert chunks[2] == chunk; + // whatever we do, the last element myst be null + chunks[2] = null; + chunkCount--; + } + + @VisibleForTesting + void reset() + { + chunkCount = 0; + for (int i = 0; i < chunks.length; i++) + { + if (chunks[i] != null) + { + chunks[i].owner = null; + chunks[i].freeSlots = 0L; + chunks[i].recycle(); + chunks[i] = null; + } + } + } + } + + private static final class LocalPoolRef extends PhantomReference<LocalPool> + { + private final Chunk[] chunks; + public LocalPoolRef(LocalPool localPool, ReferenceQueue<? super LocalPool> q) + { + super(localPool, q); + chunks = localPool.chunks; + } + + public void release() + { + for (int i = 0 ; i < chunks.length ; i++) + { + if (chunks[i] != null) + { + chunks[i].release(); + chunks[i] = null; + } + } + } + } + + private static final ConcurrentLinkedQueue<LocalPoolRef> localPoolReferences = new ConcurrentLinkedQueue<>(); + + private static final ReferenceQueue<Object> localPoolRefQueue = new ReferenceQueue<>(); + private static final ExecutorService EXEC = Executors.newFixedThreadPool(1, new NamedThreadFactory("LocalPool-Cleaner")); + static + { + EXEC.execute(new Runnable() + { + public void run() + { + try + { + while (true) + { + Object obj = localPoolRefQueue.remove(); + if (obj instanceof LocalPoolRef) + { + ((LocalPoolRef) obj).release(); + localPoolReferences.remove(obj); + } + } + } + catch (InterruptedException e) + { + } + finally + { + EXEC.execute(this); + } + } + }); + } + + private static ByteBuffer allocateDirectAligned(int capacity) + { + int align = MemoryUtil.pageSize(); + if (Integer.bitCount(align) != 1) + throw new IllegalArgumentException("Alignment must be a power of 2"); + + ByteBuffer buffer = ByteBuffer.allocateDirect(capacity + align); + long address = MemoryUtil.getAddress(buffer); + long offset = address & (align -1); // (address % align) + + if (offset == 0) + { // already aligned + buffer.limit(capacity); + } + else + { // shift by offset + int pos = (int)(align - offset); + buffer.position(pos); + buffer.limit(pos + capacity); + } + + return buffer.slice(); + } + + /** + * A memory chunk: it takes a buffer (the slab) and slices it + * into smaller buffers when requested. + * + * It divides the slab into 64 units and keeps a long mask, freeSlots, + * indicating if a unit is in use or not. Each bit in freeSlots corresponds + * to a unit, if the bit is set then the unit is free (available for allocation) + * whilst if it is not set then the unit is in use. + * + * When we receive a request of a given size we round up the size to the nearest + * multiple of allocation units required. Then we search for n consecutive free units, + * where n is the number of units required. We also align to page boundaries. + * + * When we reiceve a release request we work out the position by comparing the buffer + * address to our base address and we simply release the units. + */ + final static class Chunk + { + private final ByteBuffer slab; + private final long baseAddress; + private final int shift; + + private volatile long freeSlots; + private static final AtomicLongFieldUpdater<Chunk> freeSlotsUpdater = AtomicLongFieldUpdater.newUpdater(Chunk.class, "freeSlots"); + + // the pool that is _currently allocating_ from this Chunk + // if this is set, it means the chunk may not be recycled because we may still allocate from it; + // if it has been unset the local pool has finished with it, and it may be recycled + private volatile LocalPool owner; + private long lastRecycled; + private final Chunk original; + + Chunk(Chunk recycle) + { + assert recycle.freeSlots == 0L; + this.slab = recycle.slab; + this.baseAddress = recycle.baseAddress; + this.shift = recycle.shift; + this.freeSlots = -1L; + this.original = recycle.original; + if (DEBUG) + globalPool.debug.recycle(original); + } + + Chunk(ByteBuffer slab) + { + assert !slab.hasArray(); + this.slab = slab; + this.baseAddress = MemoryUtil.getAddress(slab); + + // The number of bits by which we need to shift to obtain a unit + // "31 &" is because numberOfTrailingZeros returns 32 when the capacity is zero + this.shift = 31 & (Integer.numberOfTrailingZeros(slab.capacity() / 64)); + // -1 means all free whilst 0 means all in use + this.freeSlots = slab.capacity() == 0 ? 0L : -1L; + this.original = DEBUG ? this : null; + } + + /** + * Acquire the chunk for future allocations: set the owner and prep + * the free slots mask. + */ + void acquire(LocalPool owner) + { + assert this.owner == null; + this.owner = owner; + } + + /** + * Set the owner to null and return the chunk to the global pool if the chunk is fully free. + * This method must be called by the LocalPool when it is certain that + * the local pool shall never try to allocate any more buffers from this chunk. + */ + void release() + { + this.owner = null; + tryRecycle(); + } + + void tryRecycle() + { + assert owner == null; + if (isFree() && freeSlotsUpdater.compareAndSet(this, -1L, 0L)) + recycle(); + } + + void recycle() + { + assert freeSlots == 0L; + globalPool.recycle(new Chunk(this)); + } + + /** + * We stash the chunk in the attachment of a buffer + * that was returned by get(), this method simply + * retrives the chunk that sliced a buffer, if any. + */ + static Chunk getParentChunk(ByteBuffer buffer) + { + Object attachment = MemoryUtil.getAttachment(buffer); + + if (attachment instanceof Chunk) + return (Chunk) attachment; + + if (attachment instanceof Ref) + return ((Ref<Chunk>) attachment).get(); + + return null; + } + + ByteBuffer setAttachment(ByteBuffer buffer) + { + if (Ref.DEBUG_ENABLED) + MemoryUtil.setAttachment(buffer, new Ref<>(this, null)); + else + MemoryUtil.setAttachment(buffer, this); + + return buffer; + } + + boolean releaseAttachment(ByteBuffer buffer) + { + Object attachment = MemoryUtil.getAttachment(buffer); + if (attachment == null) + return false; + + if (attachment instanceof Ref) + ((Ref<Chunk>) attachment).release(); + + return true; + } + + @VisibleForTesting + void reset() + { + Chunk parent = getParentChunk(slab); + if (parent != null) + parent.free(slab, false); + else + FileUtils.clean(slab); + } + + @VisibleForTesting + long setFreeSlots(long val) + { + long ret = freeSlots; + freeSlots = val; + return ret; + } + + int capacity() + { + return 64 << shift; + } + + final int unit() + { + return 1 << shift; + } + + final boolean isFree() + { + return freeSlots == -1L; + } + + /** The total free size */ + int free() + { + return Long.bitCount(freeSlots) * unit(); + } + + /** + * Return the next available slice of this size. If + * we have exceeded the capacity we return null. + */ + ByteBuffer get(int size) + { + // how many multiples of our units is the size? + // we add (unit - 1), so that when we divide by unit (>>> shift), we effectively round up + int slotCount = (size - 1 + unit()) >>> shift; + + // if we require more than 64 slots, we cannot possibly accommodate the allocation + if (slotCount > 64) + return null; + + // convert the slotCount into the bits needed in the bitmap, but at the bottom of the register + long slotBits = -1L >>> (64 - slotCount); + + // in order that we always allocate page aligned results, we require that any allocation is "somewhat" aligned + // i.e. any single unit allocation can go anywhere; any 2 unit allocation must begin in one of the first 3 slots + // of a page; a 3 unit must go in the first two slots; and any four unit allocation must be fully page-aligned + + // to achieve this, we construct a searchMask that constrains the bits we find to those we permit starting + // a match from. as we find bits, we remove them from the mask to continue our search. + // this has an odd property when it comes to concurrent alloc/free, as we can safely skip backwards if + // a new slot is freed up, but we always make forward progress (i.e. never check the same bits twice), + // so running time is bounded + long searchMask = 0x1111111111111111L; + searchMask *= 15L >>> ((slotCount - 1) & 3); + // i.e. switch (slotCount & 3) + // case 1: searchMask = 0xFFFFFFFFFFFFFFFFL + // case 2: searchMask = 0x7777777777777777L + // case 3: searchMask = 0x3333333333333333L + // case 0: searchMask = 0x1111111111111111L + + // truncate the mask, removing bits that have too few slots proceeding them + searchMask &= -1L >>> (slotCount - 1); + + // this loop is very unroll friendly, and would achieve high ILP, but not clear if the compiler will exploit this. + // right now, not worth manually exploiting, but worth noting for future + while (true) + { + long cur = freeSlots; + // find the index of the lowest set bit that also occurs in our mask (i.e. is permitted alignment, and not yet searched) + // we take the index, rather than finding the lowest bit, since we must obtain it anyway, and shifting is more efficient + // than multiplication + int index = Long.numberOfTrailingZeros(cur & searchMask); + + // if no bit was actually found, we cannot serve this request, so return null. + // due to truncating the searchMask this immediately terminates any search when we run out of indexes + // that could accommodate the allocation, i.e. is equivalent to checking (64 - index) < slotCount + if (index == 64) + return null; + + // remove this bit from our searchMask, so we don't return here next round + searchMask ^= 1L << index; + // if our bits occur starting at the index, remove ourselves from the bitmask and return + long candidate = slotBits << index; + if ((candidate & cur) == candidate) + { + // here we are sure we will manage to CAS successfully without changing candidate because + // there is only one thread allocating at the moment, the concurrency is with the release + // operations only + while (true) + { + // clear the candidate bits (freeSlots &= ~candidate) + if (freeSlotsUpdater.compareAndSet(this, cur, cur & ~candidate)) + break; + + cur = freeSlots; + // make sure no other thread has cleared the candidate bits + assert ((candidate & cur) == candidate); + } + return get(index << shift, size); + } + } + } + + private ByteBuffer get(int offset, int size) + { + slab.limit(offset + size); + slab.position(offset); + + return setAttachment(slab.slice()); + } + + /** + * Round the size to the next unit multiple. + */ + int roundUp(int v) + { + return BufferPool.roundUp(v, unit()); + } + + /** + * Release a buffer. Return: + * 0L if the buffer must be recycled after the call; + * -1L if it is free (and so we should tryRecycle if owner is now null) + * some other value otherwise + **/ + long free(ByteBuffer buffer, boolean tryRelease) + { + if (!releaseAttachment(buffer)) + return 1L; + + long address = MemoryUtil.getAddress(buffer); + assert (address >= baseAddress) & (address <= baseAddress + capacity()); + + int position = (int)(address - baseAddress); + int size = roundUp(buffer.capacity()); + + position >>= shift; + int slotCount = size >> shift; + + long slotBits = (1L << slotCount) - 1; + long shiftedSlotBits = (slotBits << position); + + if (slotCount == 64) + { + assert size == capacity(); + assert position == 0; + shiftedSlotBits = -1L; + } + + long next; + while (true) + { + long cur = freeSlots; + next = cur | shiftedSlotBits; + assert next == (cur ^ shiftedSlotBits); // ensure no double free + if (tryRelease & (next == -1L)) + next = 0L; + if (freeSlotsUpdater.compareAndSet(this, cur, next)) + return next; + } + } + + @Override + public String toString() + { + return String.format("[slab %s, slots bitmap %s, capacity %d, free %d]", slab, Long.toBinaryString(freeSlots), capacity(), free()); + } + } + + @VisibleForTesting + public static int roundUpNormal(int size) + { + return roundUp(size, CHUNK_SIZE / 64); + } + + private static int roundUp(int size, int unit) + { + int mask = unit - 1; + return (size + mask) & ~mask; + } +}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/17dd4ccc/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java b/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java index d2b2879..5671fa9 100644 --- a/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java +++ b/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java @@ -36,6 +36,7 @@ public abstract class MemoryUtil private static final long DIRECT_BYTE_BUFFER_CAPACITY_OFFSET; private static final long DIRECT_BYTE_BUFFER_LIMIT_OFFSET; private static final long DIRECT_BYTE_BUFFER_POSITION_OFFSET; + private static final long DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET; private static final Class<?> BYTE_BUFFER_CLASS; private static final long BYTE_BUFFER_OFFSET_OFFSET; private static final long BYTE_BUFFER_HB_OFFSET; @@ -62,6 +63,7 @@ public abstract class MemoryUtil DIRECT_BYTE_BUFFER_CAPACITY_OFFSET = unsafe.objectFieldOffset(Buffer.class.getDeclaredField("capacity")); DIRECT_BYTE_BUFFER_LIMIT_OFFSET = unsafe.objectFieldOffset(Buffer.class.getDeclaredField("limit")); DIRECT_BYTE_BUFFER_POSITION_OFFSET = unsafe.objectFieldOffset(Buffer.class.getDeclaredField("position")); + DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET = unsafe.objectFieldOffset(clazz.getDeclaredField("att")); DIRECT_BYTE_BUFFER_CLASS = clazz; clazz = ByteBuffer.allocate(0).getClass(); @@ -77,6 +79,17 @@ public abstract class MemoryUtil } } + public static int pageSize() + { + return unsafe.pageSize(); + } + + public static long getAddress(ByteBuffer buffer) + { + assert buffer.getClass() == DIRECT_BYTE_BUFFER_CLASS; + return unsafe.getLong(buffer, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET); + } + public static long allocate(long size) { return Native.malloc(size); @@ -177,24 +190,25 @@ public abstract class MemoryUtil unsafe.putInt(instance, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, length); } - public static ByteBuffer duplicateDirectByteBuffer(ByteBuffer source, ByteBuffer hollowBuffer) + public static Object getAttachment(ByteBuffer instance) { - assert(source.isDirect()); - unsafe.putLong(hollowBuffer, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, unsafe.getLong(source, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET)); - unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_POSITION_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_POSITION_OFFSET)); - unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_LIMIT_OFFSET)); - unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET)); - return hollowBuffer; + assert instance.getClass() == DIRECT_BYTE_BUFFER_CLASS; + return unsafe.getObject(instance, DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET); + } + + public static void setAttachment(ByteBuffer instance, Object next) + { + assert instance.getClass() == DIRECT_BYTE_BUFFER_CLASS; + unsafe.putObject(instance, DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET, next); } - public static ByteBuffer duplicateByteBuffer(ByteBuffer source, ByteBuffer hollowBuffer) + public static ByteBuffer duplicateDirectByteBuffer(ByteBuffer source, ByteBuffer hollowBuffer) { - assert(!source.isDirect()); + assert source.getClass() == DIRECT_BYTE_BUFFER_CLASS; + unsafe.putLong(hollowBuffer, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, unsafe.getLong(source, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET)); unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_POSITION_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_POSITION_OFFSET)); unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_LIMIT_OFFSET)); unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET)); - unsafe.putInt(hollowBuffer, BYTE_BUFFER_OFFSET_OFFSET, unsafe.getInt(source, BYTE_BUFFER_OFFSET_OFFSET)); - unsafe.putObject(hollowBuffer, BYTE_BUFFER_HB_OFFSET, unsafe.getObject(source, BYTE_BUFFER_HB_OFFSET)); return hollowBuffer; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/17dd4ccc/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java ---------------------------------------------------------------------- diff --git a/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java new file mode 100644 index 0000000..17ac569 --- /dev/null +++ b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java @@ -0,0 +1,454 @@ +/* + * 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.text.DateFormat; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; + +import com.google.common.util.concurrent.Uninterruptibles; +import org.junit.Test; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.utils.DynamicList; + +import static org.junit.Assert.*; + +public class LongBufferPoolTest +{ + private static final Logger logger = LoggerFactory.getLogger(LongBufferPoolTest.class); + + @Test + public void testAllocate() throws InterruptedException, ExecutionException + { + testAllocate(Runtime.getRuntime().availableProcessors() * 2, TimeUnit.MINUTES.toNanos(2L), 16 << 20); + } + + private static final class BufferCheck + { + final ByteBuffer buffer; + final long val; + DynamicList.Node<BufferCheck> listnode; + + private BufferCheck(ByteBuffer buffer, long val) + { + this.buffer = buffer; + this.val = val; + } + + void validate() + { + ByteBuffer read = buffer.duplicate(); + while (read.remaining() > 8) + assert read.getLong() == val; + } + + void init() + { + ByteBuffer write = buffer.duplicate(); + while (write.remaining() > 8) + write.putLong(val); + } + } + + public void testAllocate(int threadCount, long duration, int poolSize) throws InterruptedException, ExecutionException + { + final int avgBufferSize = 16 << 10; + final int stdevBufferSize = 10 << 10; // picked to ensure exceeding buffer size is rare, but occurs + final DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss"); + + System.out.println(String.format("%s - testing %d threads for %dm", + dateFormat.format(new Date()), + threadCount, + TimeUnit.NANOSECONDS.toMinutes(duration))); + + final long until = System.nanoTime() + duration; + final CountDownLatch latch = new CountDownLatch(threadCount); + final SPSCQueue<BufferCheck>[] sharedRecycle = new SPSCQueue[threadCount]; + final AtomicBoolean[] makingProgress = new AtomicBoolean[threadCount]; + for (int i = 0 ; i < sharedRecycle.length ; i++) + { + sharedRecycle[i] = new SPSCQueue<>(); + makingProgress[i] = new AtomicBoolean(true); + } + + ExecutorService executorService = Executors.newFixedThreadPool(threadCount + 2); + List<Future<Boolean>> ret = new ArrayList<>(threadCount); + long prevPoolSize = BufferPool.MEMORY_USAGE_THRESHOLD; + BufferPool.MEMORY_USAGE_THRESHOLD = poolSize; + BufferPool.DEBUG = true; + // sum(1..n) = n/2 * (n + 1); we set zero to CHUNK_SIZE, so have n=threadCount-1 + int targetSizeQuanta = ((threadCount) * (threadCount - 1)) / 2; + // fix targetSizeQuanta at 1/64th our poolSize, so that we only consciously exceed our pool size limit + targetSizeQuanta = (targetSizeQuanta * poolSize) / 64; + + { + // setup some high churn allocate/deallocate, without any checking + final SPSCQueue<ByteBuffer> burn = new SPSCQueue<>(); + final CountDownLatch doneAdd = new CountDownLatch(1); + executorService.submit(new TestUntil(until) + { + int count = 0; + void testOne() throws Exception + { + if (count * BufferPool.CHUNK_SIZE >= poolSize / 10) + { + if (burn.exhausted) + count = 0; + else + Thread.yield(); + return; + } + + ByteBuffer buffer = BufferPool.tryGet(BufferPool.CHUNK_SIZE); + if (buffer == null) + { + Thread.yield(); + return; + } + + BufferPool.put(buffer); + burn.add(buffer); + count++; + } + void cleanup() + { + doneAdd.countDown(); + } + }); + executorService.submit(new TestUntil(until) + { + void testOne() throws Exception + { + ByteBuffer buffer = burn.poll(); + if (buffer == null) + { + Thread.yield(); + return; + } + BufferPool.put(buffer); + } + void cleanup() + { + Uninterruptibles.awaitUninterruptibly(doneAdd); + } + }); + } + + for (int t = 0; t < threadCount; t++) + { + final int threadIdx = t; + final int targetSize = t == 0 ? BufferPool.CHUNK_SIZE : targetSizeQuanta * t; + + ret.add(executorService.submit(new TestUntil(until) + { + final SPSCQueue<BufferCheck> shareFrom = sharedRecycle[threadIdx]; + final DynamicList<BufferCheck> checks = new DynamicList<>((int) Math.max(1, targetSize / (1 << 10))); + final SPSCQueue<BufferCheck> shareTo = sharedRecycle[(threadIdx + 1) % threadCount]; + final ThreadLocalRandom rand = ThreadLocalRandom.current(); + int totalSize = 0; + int freeingSize = 0; + int size = 0; + + void checkpoint() + { + if (!makingProgress[threadIdx].get()) + makingProgress[threadIdx].set(true); + } + + void testOne() throws Exception + { + + long currentTargetSize = rand.nextInt(poolSize / 1024) == 0 ? 0 : targetSize; + int spinCount = 0; + while (totalSize > currentTargetSize - freeingSize) + { + // free buffers until we're below our target size + if (checks.size() == 0) + { + // if we're out of buffers to free, we're waiting on our neighbour to free them; + // first check if the consuming neighbour has caught up, and if so mark that free + if (shareTo.exhausted) + { + totalSize -= freeingSize; + freeingSize = 0; + } + else if (!recycleFromNeighbour()) + { + if (++spinCount > 1000 && System.nanoTime() > until) + return; + // otherwise, free one of our other neighbour's buffers if can; and otherwise yield + Thread.yield(); + } + continue; + } + + // pick a random buffer, with preference going to earlier ones + BufferCheck check = sample(); + checks.remove(check.listnode); + check.validate(); + + size = BufferPool.roundUpNormal(check.buffer.capacity()); + if (size > BufferPool.CHUNK_SIZE) + size = 0; + + // either share to free, or free immediately + if (rand.nextBoolean()) + { + shareTo.add(check); + freeingSize += size; + // interleave this with potentially messing with the other neighbour's stuff + recycleFromNeighbour(); + } + else + { + check.validate(); + BufferPool.put(check.buffer); + totalSize -= size; + } + } + + // allocate a new buffer + size = (int) Math.max(1, avgBufferSize + (stdevBufferSize * rand.nextGaussian())); + if (size <= BufferPool.CHUNK_SIZE) + { + totalSize += BufferPool.roundUpNormal(size); + allocate(size); + } + else if (rand.nextBoolean()) + { + allocate(size); + } + else + { + // perform a burst allocation to exhaust all available memory + while (totalSize < poolSize) + { + size = (int) Math.max(1, avgBufferSize + (stdevBufferSize * rand.nextGaussian())); + if (size <= BufferPool.CHUNK_SIZE) + { + allocate(size); + totalSize += BufferPool.roundUpNormal(size); + } + } + } + + // validate a random buffer we have stashed + checks.get(rand.nextInt(checks.size())).validate(); + + // free all of our neighbour's remaining shared buffers + while (recycleFromNeighbour()); + } + + void cleanup() + { + while (checks.size() > 0) + { + BufferCheck check = checks.get(0); + BufferPool.put(check.buffer); + checks.remove(check.listnode); + } + latch.countDown(); + } + + boolean recycleFromNeighbour() + { + BufferCheck check = shareFrom.poll(); + if (check == null) + return false; + check.validate(); + BufferPool.put(check.buffer); + return true; + } + + BufferCheck allocate(int size) + { + ByteBuffer buffer = BufferPool.get(size); + assertNotNull(buffer); + BufferCheck check = new BufferCheck(buffer, rand.nextLong()); + assertEquals(size, buffer.capacity()); + assertEquals(0, buffer.position()); + check.init(); + check.listnode = checks.append(check); + return check; + } + + BufferCheck sample() + { + // sample with preference to first elements: + // element at index n will be selected with likelihood (size - n) / sum1ToN(size) + int size = checks.size(); + + // pick a random number between 1 and sum1toN(size) + int sampleRange = sum1toN(size); + int sampleIndex = rand.nextInt(sampleRange); + + // then binary search for the N, such that [sum1ToN(N), sum1ToN(N+1)) contains this random number + int moveBy = Math.max(size / 4, 1); + int index = size / 2; + while (true) + { + int baseSampleIndex = sum1toN(index); + int endOfSampleIndex = sum1toN(index + 1); + if (sampleIndex >= baseSampleIndex) + { + if (sampleIndex < endOfSampleIndex) + break; + index += moveBy; + } + else index -= moveBy; + moveBy = Math.max(moveBy / 2, 1); + } + + // this gives us the inverse of our desired value, so just subtract it from the last index + index = size - (index + 1); + + return checks.get(index); + } + + private int sum1toN(int n) + { + return (n * (n + 1)) / 2; + } + })); + } + + boolean first = true; + while (!latch.await(10L, TimeUnit.SECONDS)) + { + if (!first) + BufferPool.assertAllRecycled(); + first = false; + for (AtomicBoolean progress : makingProgress) + { + assert progress.get(); + progress.set(false); + } + } + + for (SPSCQueue<BufferCheck> queue : sharedRecycle) + { + BufferCheck check; + while ( null != (check = queue.poll()) ) + { + check.validate(); + BufferPool.put(check.buffer); + } + } + + assertEquals(0, executorService.shutdownNow().size()); + + BufferPool.MEMORY_USAGE_THRESHOLD = prevPoolSize; + for (Future<Boolean> r : ret) + assertTrue(r.get()); + + System.out.println(String.format("%s - finished.", + dateFormat.format(new Date()))); + } + + static abstract class TestUntil implements Callable<Boolean> + { + final long until; + protected TestUntil(long until) + { + this.until = until; + } + + abstract void testOne() throws Exception; + void checkpoint() {} + void cleanup() {} + + public Boolean call() throws Exception + { + try + { + while (System.nanoTime() < until) + { + checkpoint(); + for (int i = 0 ; i < 100 ; i++) + testOne(); + } + } + catch (Exception ex) + { + logger.error("Got exception {}, current chunk {}", + ex.getMessage(), + BufferPool.currentChunk()); + ex.printStackTrace(); + return false; + } + finally + { + cleanup(); + } + return true; + } + } + + public static void main(String[] args) throws InterruptedException, ExecutionException + { + new LongBufferPoolTest().testAllocate(Runtime.getRuntime().availableProcessors(), TimeUnit.HOURS.toNanos(2L), 16 << 20); + } + + /** + * A single producer, single consumer queue. + */ + private static final class SPSCQueue<V> + { + static final class Node<V> + { + volatile Node<V> next; + final V value; + Node(V value) + { + this.value = value; + } + } + + private volatile boolean exhausted = true; + Node<V> head = new Node<>(null); + Node<V> tail = head; + + void add(V value) + { + exhausted = false; + tail = tail.next = new Node<>(value); + } + + V poll() + { + Node<V> next = head.next; + if (next == null) + { + // this is racey, but good enough for our purposes + exhausted = true; + return null; + } + head = next; + return next.value; + } + } + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/cassandra/blob/17dd4ccc/test/unit/org/apache/cassandra/io/compress/CompressorTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/io/compress/CompressorTest.java b/test/unit/org/apache/cassandra/io/compress/CompressorTest.java index 6018cc7..8ff0074 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressorTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressorTest.java @@ -33,6 +33,7 @@ import org.junit.Test; import org.apache.cassandra.io.util.RandomAccessReader; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.memory.BufferPool; public class CompressorTest { http://git-wip-us.apache.org/repos/asf/cassandra/blob/17dd4ccc/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java b/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java index 0c1583d..3c9aa0e 100644 --- a/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java +++ b/test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java @@ -18,8 +18,6 @@ * */ package org.apache.cassandra.io.util; - -import org.apache.cassandra.service.FileCacheService; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.SyncUtil; @@ -30,10 +28,6 @@ import java.nio.ByteBuffer; import java.nio.channels.ClosedChannelException; import java.util.Arrays; import java.util.concurrent.Callable; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.cassandra.Util.expectEOF; import static org.apache.cassandra.Util.expectException; @@ -48,6 +42,7 @@ public class BufferedRandomAccessFileTest public void testReadAndWrite() throws Exception { SequentialWriter w = createTempFile("braf"); + ChannelProxy channel = new ChannelProxy(w.getPath()); // writting string of data to the file byte[] data = "Hello".getBytes(); @@ -58,7 +53,7 @@ public class BufferedRandomAccessFileTest w.sync(); // reading small amount of data from file, this is handled by initial buffer - RandomAccessReader r = RandomAccessReader.open(w); + RandomAccessReader r = RandomAccessReader.open(channel); byte[] buffer = new byte[data.length]; assertEquals(data.length, r.read(buffer)); @@ -81,7 +76,7 @@ public class BufferedRandomAccessFileTest w.sync(); - r = RandomAccessReader.open(w); // re-opening file in read-only mode + r = RandomAccessReader.open(channel); // re-opening file in read-only mode // reading written buffer r.seek(initialPosition); // back to initial (before write) position @@ -130,6 +125,7 @@ public class BufferedRandomAccessFileTest w.finish(); r.close(); + channel.close(); } @Test @@ -142,7 +138,8 @@ public class BufferedRandomAccessFileTest byte[] in = generateByteArray(RandomAccessReader.DEFAULT_BUFFER_SIZE); w.write(in); - RandomAccessReader r = RandomAccessReader.open(w); + ChannelProxy channel = new ChannelProxy(w.getPath()); + RandomAccessReader r = RandomAccessReader.open(channel); // Read it into a same size array. byte[] out = new byte[RandomAccessReader.DEFAULT_BUFFER_SIZE]; @@ -154,6 +151,7 @@ public class BufferedRandomAccessFileTest r.close(); w.finish(); + channel.close(); } @Test @@ -181,9 +179,11 @@ public class BufferedRandomAccessFileTest w.finish(); // will use cachedlength - RandomAccessReader r = RandomAccessReader.open(tmpFile); - assertEquals(lessThenBuffer.length + biggerThenBuffer.length, r.length()); - r.close(); + try (ChannelProxy channel = new ChannelProxy(tmpFile); + RandomAccessReader r = RandomAccessReader.open(channel)) + { + assertEquals(lessThenBuffer.length + biggerThenBuffer.length, r.length()); + } } @Test @@ -201,7 +201,8 @@ public class BufferedRandomAccessFileTest w.write(data); w.sync(); - final RandomAccessReader r = RandomAccessReader.open(w); + final ChannelProxy channel = new ChannelProxy(w.getPath()); + final RandomAccessReader r = RandomAccessReader.open(channel); ByteBuffer content = r.readBytes((int) r.length()); @@ -225,6 +226,7 @@ public class BufferedRandomAccessFileTest w.finish(); r.close(); + channel.close(); } @Test @@ -235,7 +237,8 @@ public class BufferedRandomAccessFileTest w.write(data); w.finish(); - final RandomAccessReader file = RandomAccessReader.open(w); + final ChannelProxy channel = new ChannelProxy(w.getPath()); + final RandomAccessReader file = RandomAccessReader.open(channel); file.seek(0); assertEquals(file.getFilePointer(), 0); @@ -265,6 +268,7 @@ public class BufferedRandomAccessFileTest }, IllegalArgumentException.class); // throws IllegalArgumentException file.close(); + channel.close(); } @Test @@ -274,7 +278,8 @@ public class BufferedRandomAccessFileTest w.write(generateByteArray(RandomAccessReader.DEFAULT_BUFFER_SIZE * 2)); w.finish(); - RandomAccessReader file = RandomAccessReader.open(w); + ChannelProxy channel = new ChannelProxy(w.getPath()); + RandomAccessReader file = RandomAccessReader.open(channel); file.seek(0); // back to the beginning of the file assertEquals(file.skipBytes(10), 10); @@ -294,6 +299,7 @@ public class BufferedRandomAccessFileTest assertEquals(file.bytesRemaining(), file.length()); file.close(); + channel.close(); } @Test @@ -308,7 +314,8 @@ public class BufferedRandomAccessFileTest w.sync(); - RandomAccessReader r = RandomAccessReader.open(w); + ChannelProxy channel = new ChannelProxy(w.getPath()); + RandomAccessReader r = RandomAccessReader.open(channel); // position should change after skip bytes r.seek(0); @@ -322,6 +329,7 @@ public class BufferedRandomAccessFileTest w.finish(); r.close(); + channel.close(); } @Test @@ -344,7 +352,7 @@ public class BufferedRandomAccessFileTest { File file1 = writeTemporaryFile(new byte[16]); try (final ChannelProxy channel = new ChannelProxy(file1); - final RandomAccessReader file = RandomAccessReader.open(channel, bufferSize, null)) + final RandomAccessReader file = RandomAccessReader.open(channel, bufferSize, -1L)) { expectEOF(new Callable<Object>() { @@ -362,7 +370,7 @@ public class BufferedRandomAccessFileTest { File file1 = writeTemporaryFile(new byte[16]); try (final ChannelProxy channel = new ChannelProxy(file1); - final RandomAccessReader file = RandomAccessReader.open(channel, bufferSize, null)) + final RandomAccessReader file = RandomAccessReader.open(channel, bufferSize, -1L)) { expectEOF(new Callable<Object>() { @@ -397,7 +405,8 @@ public class BufferedRandomAccessFileTest w.sync(); - RandomAccessReader r = RandomAccessReader.open(w); + ChannelProxy channel = new ChannelProxy(w.getPath()); + RandomAccessReader r = RandomAccessReader.open(channel); assertEquals(r.bytesRemaining(), toWrite); @@ -413,6 +422,7 @@ public class BufferedRandomAccessFileTest w.finish(); r.close(); + channel.close(); } @Test @@ -483,7 +493,8 @@ public class BufferedRandomAccessFileTest w.finish(); - RandomAccessReader file = RandomAccessReader.open(w); + ChannelProxy channel = new ChannelProxy(w.getPath()); + RandomAccessReader file = RandomAccessReader.open(channel); file.seek(10); FileMark mark = file.mark(); @@ -508,6 +519,7 @@ public class BufferedRandomAccessFileTest assertEquals(file.bytesPastMark(), 0); file.close(); + channel.close(); } @Test (expected = AssertionError.class) @@ -518,7 +530,8 @@ public class BufferedRandomAccessFileTest w.write(new byte[30]); w.flush(); - try (RandomAccessReader r = RandomAccessReader.open(w)) + try (ChannelProxy channel = new ChannelProxy(w.getPath()); + RandomAccessReader r = RandomAccessReader.open(channel)) { r.seek(10); r.mark(); @@ -530,71 +543,6 @@ public class BufferedRandomAccessFileTest } @Test - public void testFileCacheService() throws IOException, InterruptedException - { - //see https://issues.apache.org/jira/browse/CASSANDRA-7756 - - final FileCacheService.CacheKey cacheKey = new FileCacheService.CacheKey(); - final int THREAD_COUNT = 40; - ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT); - - SequentialWriter w1 = createTempFile("fscache1"); - SequentialWriter w2 = createTempFile("fscache2"); - - w1.write(new byte[30]); - w1.finish(); - - w2.write(new byte[30]); - w2.finish(); - - for (int i = 0; i < 20; i++) - { - - - RandomAccessReader r1 = RandomAccessReader.open(w1); - RandomAccessReader r2 = RandomAccessReader.open(w2); - - - FileCacheService.instance.put(cacheKey, r1); - FileCacheService.instance.put(cacheKey, r2); - - final CountDownLatch finished = new CountDownLatch(THREAD_COUNT); - final AtomicBoolean hadError = new AtomicBoolean(false); - - for (int k = 0; k < THREAD_COUNT; k++) - { - executorService.execute( new Runnable() - { - @Override - public void run() - { - try - { - long size = FileCacheService.instance.sizeInBytes(); - - while (size > 0) - size = FileCacheService.instance.sizeInBytes(); - } - catch (Throwable t) - { - t.printStackTrace(); - hadError.set(true); - } - finally - { - finished.countDown(); - } - } - }); - - } - - finished.await(); - assert !hadError.get(); - } - } - - @Test public void testReadOnly() throws IOException { SequentialWriter file = createTempFile("brafReadOnlyTest");
