totalCommitLogSize;
+ /** Time spent waiting for a CLS to be allocated - under normal conditions this should be zero */
+ public final Timer waitingOnSegmentAllocation;
+ /** The time spent waiting on CL sync; for Periodic this is only occurs when the sync is lagging its sync interval */
+ public final Timer waitingOnCommit;
public CommitLogMetrics(final AbstractCommitLogService service, final CommitLogSegmentManager allocator)
{
@@ -60,5 +67,7 @@ public class CommitLogMetrics
return allocator.bytesUsed();
}
});
+ waitingOnSegmentAllocation = Metrics.newTimer(factory.createMetricName("WaitingOnSegmentAllocation"), TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
+ waitingOnCommit = Metrics.newTimer(factory.createMetricName("WaitingOnCommit"), TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
}
}
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 424dbfa58e..2903fc4090 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -45,7 +45,6 @@ import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.db.MeteredFlusher;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.compaction.CompactionManager;
@@ -258,10 +257,6 @@ public class CassandraDaemon
logger.warn("Unable to start GCInspector (currently only supported on the Sun JVM)");
}
- // MeteredFlusher can block if flush queue fills up, so don't put on scheduledTasks
- // Start it before commit log, so memtables can flush during commit log replay
- StorageService.optionalTasks.scheduleWithFixedDelay(new MeteredFlusher(), 1000, 1000, TimeUnit.MILLISECONDS);
-
// replay the log if necessary
try
{
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index 182ed21df5..0a94cc0736 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -245,13 +245,6 @@ public class FBUtilities
return out;
}
- public static BigInteger hashToBigInteger(ByteBuffer data)
- {
- byte[] result = hash(data);
- BigInteger hash = new BigInteger(result);
- return hash.abs();
- }
-
public static byte[] hash(ByteBuffer... data)
{
MessageDigest messageDigest = localMD5Digest.get();
@@ -266,6 +259,11 @@ public class FBUtilities
return messageDigest.digest();
}
+ public static BigInteger hashToBigInteger(ByteBuffer data)
+ {
+ return new BigInteger(hash(data)).abs();
+ }
+
@Deprecated
public static void serialize(TSerializer serializer, TBase struct, DataOutput out)
throws IOException
diff --git a/src/java/org/apache/cassandra/utils/ObjectSizes.java b/src/java/org/apache/cassandra/utils/ObjectSizes.java
index 28ddffda6b..3720385dbc 100644
--- a/src/java/org/apache/cassandra/utils/ObjectSizes.java
+++ b/src/java/org/apache/cassandra/utils/ObjectSizes.java
@@ -21,242 +21,143 @@ package org.apache.cassandra.utils;
*/
-import java.lang.management.ManagementFactory;
-import java.lang.management.MemoryPoolMXBean;
import java.nio.ByteBuffer;
+import org.github.jamm.MemoryLayoutSpecification;
import org.github.jamm.MemoryMeter;
/**
- * Modified version of the code from.
- * https://github.com/twitter/commons/blob/master
- * /src/java/com/twitter/common/objectsize/ObjectSizeCalculator.java
- *
- * Difference is that we don't use reflection.
+ * A convenience class for wrapping access to MemoryMeter
*/
public class ObjectSizes
{
- public static final MemoryLayoutSpecification SPEC = getEffectiveMemoryLayoutSpecification();
- private static final MemoryMeter meter = new MemoryMeter().omitSharedBufferOverhead();
+ private static final MemoryMeter meter = new MemoryMeter()
+ .omitSharedBufferOverhead()
+ .withGuessing(MemoryMeter.Guess.FALLBACK_UNSAFE);
- /**
- * Describes constant memory overheads for various constructs in a JVM
- * implementation.
- */
- public interface MemoryLayoutSpecification
- {
- int getArrayHeaderSize();
-
- int getObjectHeaderSize();
-
- int getObjectPadding();
-
- int getReferenceSize();
-
- int getSuperclassFieldPadding();
- }
-
- /**
- * Memory a class consumes, including the object header and the size of the fields.
- * @param fieldsSize Total size of the primitive fields of a class
- * @return Total in-memory size of the class
- */
- public static long getFieldSize(long fieldsSize)
- {
- return roundTo(SPEC.getObjectHeaderSize() + fieldsSize, SPEC.getObjectPadding());
- }
-
- /**
- * Memory a super class consumes, given the primitive field sizes
- * @param fieldsSize Total size of the primitive fields of the super class
- * @return Total additional in-memory that the super class takes up
- */
- public static long getSuperClassFieldSize(long fieldsSize)
- {
- return roundTo(fieldsSize, SPEC.getSuperclassFieldPadding());
- }
-
- /**
- * Memory an array will consume
- * @param length Number of elements in the array
- * @param elementSize In-memory size of each element's primitive stored
- * @return In-memory size of the array
- */
- public static long getArraySize(int length, long elementSize)
- {
- return roundTo(SPEC.getArrayHeaderSize() + length * elementSize, SPEC.getObjectPadding());
- }
+ private static final long BUFFER_EMPTY_SIZE = measure(ByteBufferUtil.EMPTY_BYTE_BUFFER);
+ private static final long STRING_EMPTY_SIZE = measure("");
/**
* Memory a byte array consumes
* @param bytes byte array to get memory size
- * @return In-memory size of the array
+ * @return heap-size of the array
*/
- public static long getArraySize(byte[] bytes)
+ public static long sizeOfArray(byte[] bytes)
{
- return getArraySize(bytes.length, 1);
+ return sizeOfArray(bytes.length, 1);
+ }
+
+ /**
+ * Memory a long array consumes
+ * @param longs byte array to get memory size
+ * @return heap-size of the array
+ */
+ public static long sizeOfArray(long[] longs)
+ {
+ return sizeOfArray(longs.length, 8);
+ }
+
+ /**
+ * Memory an int array consumes
+ * @param ints byte array to get memory size
+ * @return heap-size of the array
+ */
+ public static long sizeOfArray(int[] ints)
+ {
+ return sizeOfArray(ints.length, 4);
+ }
+
+ /**
+ * Memory a reference array consumes
+ * @param length the length of the reference array
+ * @return heap-size of the array
+ */
+ public static long sizeOfReferenceArray(int length)
+ {
+ return sizeOfArray(length, MemoryLayoutSpecification.SPEC.getReferenceSize());
+ }
+
+ /**
+ * Memory a reference array consumes itself only
+ * @param objects the array to size
+ * @return heap-size of the array (excluding memory retained by referenced objects)
+ */
+ public static long sizeOfArray(Object[] objects)
+ {
+ return sizeOfReferenceArray(objects.length);
+ }
+
+ private static long sizeOfArray(int length, long elementSize)
+ {
+ return MemoryLayoutSpecification.sizeOfArray(length, elementSize);
}
/**
* Memory a ByteBuffer array consumes.
*/
- public static long getArraySize(ByteBuffer[] array)
+ public static long sizeOnHeapOf(ByteBuffer[] array)
{
long allElementsSize = 0;
for (int i = 0; i < array.length; i++)
if (array[i] != null)
- allElementsSize += getSize(array[i]);
+ allElementsSize += sizeOnHeapOf(array[i]);
- return allElementsSize + getArraySize(array.length, getReferenceSize());
+ return allElementsSize + sizeOfArray(array);
}
+ public static long sizeOnHeapExcludingData(ByteBuffer[] array)
+ {
+ return BUFFER_EMPTY_SIZE * array.length + sizeOfArray(array);
+ }
/**
* Memory a byte buffer consumes
* @param buffer ByteBuffer to calculate in memory size
* @return Total in-memory size of the byte buffer
*/
- public static long getSize(ByteBuffer buffer)
+ public static long sizeOnHeapOf(ByteBuffer buffer)
{
- long size = 0;
- /* BB Class */
- // final byte[] hb;
- // final int offset;
- // boolean isReadOnly;
- size += ObjectSizes.getFieldSize(1L + 4 + ObjectSizes.getReferenceSize() + ObjectSizes.getArraySize(buffer.capacity(), 1));
- /* Super Class */
- // private int mark;
- // private int position;
- // private int limit;
- // private int capacity;
- size += ObjectSizes.getSuperClassFieldSize(4L + 4 + 4 + 4 + 8);
- return size;
+ if (buffer.isDirect())
+ return BUFFER_EMPTY_SIZE;
+ // if we're only referencing a sub-portion of the ByteBuffer, don't count the array overhead (assume it's slab
+ // allocated, so amortized over all the allocations the overhead is negligible and better to undercount than over)
+ if (buffer.capacity() > buffer.remaining())
+ return buffer.remaining();
+ return BUFFER_EMPTY_SIZE + sizeOfArray(buffer.capacity(), 1);
}
- public static long roundTo(long x, int multiple)
+ public static long sizeOnHeapExcludingData(ByteBuffer buffer)
{
- return ((x + multiple - 1) / multiple) * multiple;
+ return BUFFER_EMPTY_SIZE;
}
/**
- * @return Memory a reference consumes on the current architecture.
+ * Memory a String consumes
+ * @param str String to calculate memory size of
+ * @return Total in-memory size of the String
*/
- public static int getReferenceSize()
+ public static long sizeOf(String str)
{
- return SPEC.getReferenceSize();
- }
-
- private static MemoryLayoutSpecification getEffectiveMemoryLayoutSpecification()
- {
- final String dataModel = System.getProperty("sun.arch.data.model");
- if ("32".equals(dataModel))
- {
- // Running with 32-bit data model
- return new MemoryLayoutSpecification()
- {
- public int getArrayHeaderSize()
- {
- return 12;
- }
-
- public int getObjectHeaderSize()
- {
- return 8;
- }
-
- public int getObjectPadding()
- {
- return 8;
- }
-
- public int getReferenceSize()
- {
- return 4;
- }
-
- public int getSuperclassFieldPadding()
- {
- return 4;
- }
- };
- }
-
- final String strVmVersion = System.getProperty("java.vm.version");
- final int vmVersion = Integer.parseInt(strVmVersion.substring(0, strVmVersion.indexOf('.')));
- if (vmVersion >= 17)
- {
- long maxMemory = 0;
- for (MemoryPoolMXBean mp : ManagementFactory.getMemoryPoolMXBeans())
- {
- maxMemory += mp.getUsage().getMax();
- }
- if (maxMemory < 30L * 1024 * 1024 * 1024)
- {
- // HotSpot 17.0 and above use compressed OOPs below 30GB of RAM
- // total for all memory pools (yes, including code cache).
- return new MemoryLayoutSpecification()
- {
- public int getArrayHeaderSize()
- {
- return 16;
- }
-
- public int getObjectHeaderSize()
- {
- return 12;
- }
-
- public int getObjectPadding()
- {
- return 8;
- }
-
- public int getReferenceSize()
- {
- return 4;
- }
-
- public int getSuperclassFieldPadding()
- {
- return 4;
- }
- };
- }
- }
-
- /* Worst case we over count. */
-
- // In other cases, it's a 64-bit uncompressed OOPs object model
- return new MemoryLayoutSpecification()
- {
- public int getArrayHeaderSize()
- {
- return 24;
- }
-
- public int getObjectHeaderSize()
- {
- return 16;
- }
-
- public int getObjectPadding()
- {
- return 8;
- }
-
- public int getReferenceSize()
- {
- return 8;
- }
-
- public int getSuperclassFieldPadding()
- {
- return 8;
- }
- };
+ return STRING_EMPTY_SIZE + sizeOfArray(str.length(), 2);
}
+ /**
+ * @param pojo the object to measure
+ * @return the size on the heap of the instance and all retained heap referenced by it, excluding portions of
+ * ByteBuffer that are not directly referenced by it but including any other referenced that may also be retained
+ * by other objects.
+ */
public static long measureDeep(Object pojo)
{
return meter.measureDeep(pojo);
}
+
+ /**
+ * @param pojo the object to measure
+ * @return the size on the heap of the instance only, excluding any referenced objects
+ */
+ public static long measure(Object pojo)
+ {
+ return meter.measure(pojo);
+ }
}
diff --git a/src/java/org/apache/cassandra/utils/WaitQueue.java b/src/java/org/apache/cassandra/utils/WaitQueue.java
deleted file mode 100644
index 01b2559930..0000000000
--- a/src/java/org/apache/cassandra/utils/WaitQueue.java
+++ /dev/null
@@ -1,264 +0,0 @@
-package org.apache.cassandra.utils;
-
-import java.util.Iterator;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.locks.AbstractQueuedSynchronizer;
-import java.util.concurrent.locks.LockSupport;
-
-/**
- * A relatively easy to use utility for general purpose thread signalling.
- * Usage on a thread awaiting a state change using a WaitQueue q is:
- *
- * {@code
- * while (!conditionMet())
- * WaitSignal s = q.register();
- * if (!conditionMet()) // or, perhaps more correctly, !conditionChanged()
- * s.await();
- * else
- * s.cancel();
- * }
- *
- * A signalling thread, AFTER changing the state, then calls q.signal() to wake up one, or q.signalAll()
- * to wake up all, waiting threads.
- *
- * A few notes on utilisation:
- * 1. A thread will only exit await() when it has been signalled, but this does
- * not guarantee the condition has not been altered since it was signalled,
- * and depending on your design it is likely the outer condition will need to be
- * checked in a loop, though this is not always the case.
- * 2. Each signal is single use, so must be re-registered after each await(). This is true even if it times out.
- * 3. If you choose not to wait on the signal (because the condition has been met before you waited on it)
- * you must cancel() the signal if the signalling thread uses signal() to awake waiters; otherwise signals will be
- * lost
- * 4. Care must be taken when selecting conditionMet() to ensure we are waiting on the condition that actually
- * indicates progress is possible. In some complex cases it may be tempting to wait on a condition that is only indicative
- * of local progress, not progress on the task we are aiming to complete, and a race may leave us waiting for a condition
- * to be met that we no longer need.
- *
5. This scheme is not fair
- * 6. Only the thread that calls register() may call await()
- * To understand intuitively how this class works, the idea is simply that a thread, once it considers itself
- * incapable of making progress, registers itself to be awoken once that condition changes. However, that condition
- * could have changed between checking and registering (in which case a thread updating the state would have been unable to signal it),
- * so before going to sleep on the signal, it checks the condition again, sleeping only if it hasn't changed.
- */
-// TODO : switch to a Lock Free queue
-public final class WaitQueue
-{
- public final class Signal
- {
- private final Thread thread = Thread.currentThread();
- volatile int signalled;
-
- private boolean isSignalled()
- {
- return signalled == 1;
- }
-
- public boolean isCancelled()
- {
- return signalled == -1;
- }
-
- private boolean signal()
- {
- if (signalledUpdater.compareAndSet(this, 0, 1))
- {
- LockSupport.unpark(thread);
- return true;
- }
- return false;
- }
-
- public void awaitUninterruptibly()
- {
- assert !isCancelled();
- if (thread != Thread.currentThread())
- throw new IllegalStateException();
- boolean interrupted = false;
- while (!isSignalled())
- {
- if (Thread.interrupted())
- interrupted = true;
- LockSupport.park();
- }
- if (interrupted)
- thread.interrupt();
- }
-
- public void await() throws InterruptedException
- {
- assert !isCancelled();
- while (!isSignalled())
- {
- if (Thread.interrupted())
- {
- checkAndClear();
- throw new InterruptedException();
- }
- if (thread != Thread.currentThread())
- throw new IllegalStateException();
- LockSupport.park();
- }
- }
-
- public long awaitNanos(long nanosTimeout) throws InterruptedException
- {
- assert signalled != -1;
- long start = System.nanoTime();
- while (!isSignalled())
- {
- if (Thread.interrupted())
- {
- checkAndClear();
- throw new InterruptedException();
- }
- LockSupport.parkNanos(nanosTimeout);
- }
- return nanosTimeout - (System.nanoTime() - start);
- }
-
- public boolean await(long time, TimeUnit unit) throws InterruptedException
- {
- // ignores nanos atm
- long until = System.currentTimeMillis() + unit.toMillis(time);
- if (until < 0)
- until = Long.MAX_VALUE;
- return awaitUntil(until);
- }
-
- public boolean awaitUntil(long until) throws InterruptedException
- {
- assert !isCancelled();
- while (until < System.currentTimeMillis() && !isSignalled())
- {
- if (Thread.interrupted())
- {
- checkAndClear();
- throw new InterruptedException();
- }
- LockSupport.parkUntil(until);
- }
- return checkAndClear();
- }
-
- private boolean checkAndClear()
- {
- if (isSignalled())
- {
- signalled = -1;
- return true;
- }
- else if (signalledUpdater.compareAndSet(this, 0, -1))
- {
- cleanUpCancelled();
- return false;
- }
- else
- {
- // must now be signalled, as checkAndClear() can only be called by
- // owning thread if used correctly
- signalled = -1;
- return true;
- }
- }
-
- public void cancel()
- {
- if (signalled < 0)
- return;
- if (!signalledUpdater.compareAndSet(this, 0, -1))
- {
- signalled = -1;
- signal();
- cleanUpCancelled();
- }
- }
-
- }
-
- private static final AtomicIntegerFieldUpdater signalledUpdater = AtomicIntegerFieldUpdater.newUpdater(Signal.class, "signalled");
-
- // the waiting signals
- private final ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>();
-
- /**
- * The calling thread MUST be the thread that uses the signal (for now)
- * @return
- */
- public Signal register()
- {
- Signal signal = new Signal();
- queue.add(signal);
- return signal;
- }
-
- /**
- * Signal one waiting thread
- */
- public void signal()
- {
- if (queue.isEmpty())
- return;
- Iterator iter = queue.iterator();
- while (iter.hasNext())
- {
- Signal next = iter.next();
- if (next.signal())
- {
- iter.remove();
- return;
- }
- }
- }
-
- /**
- * Signal all waiting threads
- */
- public void signalAll()
- {
- if (queue.isEmpty())
- return;
- Iterator iter = queue.iterator();
- while (iter.hasNext())
- {
- Signal next = iter.next();
- if (next.signal())
- iter.remove();
- }
- }
-
- private void cleanUpCancelled()
- {
- Iterator iter = queue.iterator();
- while (iter.hasNext())
- {
- Signal next = iter.next();
- if (next.isCancelled())
- iter.remove();
- }
- }
-
- /**
- * Return how many threads are waiting
- * @return
- */
- public int getWaiting()
- {
- if (queue.isEmpty())
- return 0;
- Iterator iter = queue.iterator();
- int count = 0;
- while (iter.hasNext())
- {
- Signal next = iter.next();
- if (next.isCancelled())
- iter.remove();
- else
- count++;
- }
- return count;
- }
-
-}
diff --git a/src/java/org/apache/cassandra/utils/btree/BTree.java b/src/java/org/apache/cassandra/utils/btree/BTree.java
index 1721fb0e9b..eb7e5588fc 100644
--- a/src/java/org/apache/cassandra/utils/btree/BTree.java
+++ b/src/java/org/apache/cassandra/utils/btree/BTree.java
@@ -5,6 +5,7 @@ import java.util.Collection;
import java.util.Comparator;
import com.google.common.base.Function;
+import org.apache.cassandra.utils.ObjectSizes;
import com.google.common.collect.Collections2;
public class BTree
@@ -73,7 +74,7 @@ public class BTree
* @param
* @return
*/
- public static Object[] build(Collection source, Comparator comparator, boolean sorted)
+ public static Object[] build(Collection source, Comparator comparator, boolean sorted, UpdateFunction updateF)
{
int size = source.size();
@@ -84,6 +85,12 @@ public class BTree
// inline sorting since we're already calling toArray
if (!sorted)
Arrays.sort(values, 0, size, comparator);
+ if (updateF != null)
+ {
+ for (int i = 0 ; i < size ; i++)
+ values[i] = updateF.apply(values[i]);
+ updateF.allocated(ObjectSizes.sizeOfArray(values));
+ }
return values;
}
@@ -105,7 +112,7 @@ public class BTree
*/
public static Object[] update(Object[] btree, Comparator comparator, Collection updateWith, boolean updateWithIsSorted)
{
- return update(btree, comparator, updateWith, updateWithIsSorted, null, null);
+ return update(btree, comparator, updateWith, updateWithIsSorted, null);
}
/**
@@ -115,9 +122,7 @@ public class BTree
* @param comparator the comparator that defines the ordering over the items in the tree
* @param updateWith the items to either insert / update
* @param updateWithIsSorted if false, updateWith will be copied and sorted to facilitate construction
- * @param replaceF a function to apply to a pair we are swapping
- * @param terminateEarly a function that returns Boolean.TRUE if we should terminate before finishing our work.
- * the argument to terminateEarly is ignored.
+ * @param updateF the update function to apply to any pairs we are swapping, and maybe abort early
* @param
* @return
*/
@@ -125,15 +130,10 @@ public class BTree
Comparator comparator,
Collection updateWith,
boolean updateWithIsSorted,
- ReplaceFunction replaceF,
- Function terminateEarly)
+ UpdateFunction updateF)
{
if (btree.length == 0)
- {
- if (replaceF != null)
- updateWith = Collections2.transform(updateWith, replaceF);
- return build(updateWith, comparator, updateWithIsSorted);
- }
+ return build(updateWith, comparator, updateWithIsSorted, updateF);
if (!updateWithIsSorted)
updateWith = sorted(updateWith, comparator, updateWith.size());
@@ -167,13 +167,13 @@ public class BTree
{
// apply replaceF if it matches an existing element
btreeOffset++;
- if (replaceF != null)
- v = replaceF.apply((V) btree[i], v);
+ if (updateF != null)
+ v = updateF.apply((V) btree[i], v);
}
- else if (replaceF != null)
+ else if (updateF != null)
{
// new element but still need to apply replaceF to handle indexing and size-tracking
- v = replaceF.apply(v);
+ v = updateF.apply(v);
}
merged[mergedCount++] = v;
@@ -187,19 +187,15 @@ public class BTree
mergedCount += count;
}
- if (mergedCount > FAN_FACTOR)
- {
- // TODO this code will never execute since QUICK_MERGE_LIMIT == FAN_FACTOR
- int mid = (mergedCount >> 1) & ~1; // divide by two, rounding down to an even number
- return new Object[] { merged[mid],
- Arrays.copyOfRange(merged, 0, mid),
- Arrays.copyOfRange(merged, 1 + mid, mergedCount + ((mergedCount + 1) & 1)), };
- }
+ assert mergedCount <= FAN_FACTOR;
- return Arrays.copyOfRange(merged, 0, mergedCount + (mergedCount & 1));
+ Object[] r = Arrays.copyOfRange(merged, 0, mergedCount + (mergedCount & 1));
+ if (updateF != null)
+ updateF.allocated(ObjectSizes.sizeOfArray(r) - (btree.length == 0 ? 0 : ObjectSizes.sizeOfArray(btree)));
+ return r;
}
- return modifier.get().update(btree, comparator, updateWith, replaceF, terminateEarly);
+ return modifier.get().update(btree, comparator, updateWith, updateF);
}
/**
diff --git a/src/java/org/apache/cassandra/utils/btree/BTreeSet.java b/src/java/org/apache/cassandra/utils/btree/BTreeSet.java
index 1730cfc8a4..4452663b0e 100644
--- a/src/java/org/apache/cassandra/utils/btree/BTreeSet.java
+++ b/src/java/org/apache/cassandra/utils/btree/BTreeSet.java
@@ -20,7 +20,7 @@ public class BTreeSet implements NavigableSet
public BTreeSet update(Collection updateWith, boolean isSorted)
{
- return new BTreeSet<>(BTree.update(tree, comparator, updateWith, isSorted, null, null), comparator);
+ return new BTreeSet<>(BTree.update(tree, comparator, updateWith, isSorted, null), comparator);
}
@Override
diff --git a/src/java/org/apache/cassandra/utils/btree/Builder.java b/src/java/org/apache/cassandra/utils/btree/Builder.java
index 0865146473..5ba923bc18 100644
--- a/src/java/org/apache/cassandra/utils/btree/Builder.java
+++ b/src/java/org/apache/cassandra/utils/btree/Builder.java
@@ -3,9 +3,6 @@ package org.apache.cassandra.utils.btree;
import java.util.Collection;
import java.util.Comparator;
-import com.google.common.base.Function;
-
-import static org.apache.cassandra.utils.btree.BTree.EMPTY_BRANCH;
import static org.apache.cassandra.utils.btree.BTree.EMPTY_LEAF;
import static org.apache.cassandra.utils.btree.BTree.FAN_SHIFT;
import static org.apache.cassandra.utils.btree.BTree.POSITIVE_INFINITY;
@@ -40,21 +37,21 @@ final class Builder
* we assume @param source has been sorted, e.g. by BTree.update, so the update of each key resumes where
* the previous left off.
*/
- public Object[] update(Object[] btree, Comparator comparator, Collection source, ReplaceFunction replaceF, Function terminateEarly)
+ public Object[] update(Object[] btree, Comparator comparator, Collection source, UpdateFunction updateF)
{
NodeBuilder current = rootBuilder;
- current.reset(btree, POSITIVE_INFINITY);
+ current.reset(btree, POSITIVE_INFINITY, updateF, comparator);
for (V key : source)
{
while (true)
{
- if (terminateEarly != null && terminateEarly.apply(null) == Boolean.TRUE)
+ if (updateF != null && updateF.abortEarly())
{
rootBuilder.clear();
return null;
}
- NodeBuilder next = current.update(key, comparator, replaceF);
+ NodeBuilder next = current.update(key);
if (next == null)
break;
// we were in a subtree from a previous key that didn't contain this new key;
@@ -66,7 +63,7 @@ final class Builder
// finish copying any remaining keys from the original btree
while (true)
{
- NodeBuilder next = current.update(POSITIVE_INFINITY, comparator, replaceF);
+ NodeBuilder next = current.update(POSITIVE_INFINITY);
if (next == null)
break;
current = next;
@@ -88,9 +85,9 @@ final class Builder
while ((size >>= FAN_SHIFT) > 0)
current = current.ensureChild();
- current.reset(EMPTY_LEAF, POSITIVE_INFINITY);
+ current.reset(EMPTY_LEAF, POSITIVE_INFINITY, null, null);
for (V key : source)
- current.addNewKey(key, null);
+ current.addNewKey(key);
current = current.ascendToRoot();
diff --git a/src/java/org/apache/cassandra/utils/btree/NodeBuilder.java b/src/java/org/apache/cassandra/utils/btree/NodeBuilder.java
index e526394933..5a47f35006 100644
--- a/src/java/org/apache/cassandra/utils/btree/NodeBuilder.java
+++ b/src/java/org/apache/cassandra/utils/btree/NodeBuilder.java
@@ -1,5 +1,7 @@
package org.apache.cassandra.utils.btree;
+import org.apache.cassandra.utils.ObjectSizes;
+
import java.util.Arrays;
import java.util.Comparator;
@@ -38,6 +40,9 @@ final class NodeBuilder
// the index of the first child node in copyFrom that has not yet been copied into the build arrays
private int copyFromChildPosition;
+ private UpdateFunction updateFunction;
+ private Comparator comparator;
+
// upper bound of range owned by this level; lets us know if we need to ascend back up the tree
// for the next key we update when bsearch gives an insertion point past the end of the values
// in the current node
@@ -51,7 +56,7 @@ final class NodeBuilder
{
if (current.upperBound != null)
{
- current.reset(null, null);
+ current.reset(null, null, null, null);
Arrays.fill(current.buildKeys, 0, current.maxBuildKeyPosition, null);
Arrays.fill(current.buildChildren, 0, current.maxBuildChildPosition, null);
current.maxBuildChildPosition = current.maxBuildKeyPosition = 0;
@@ -61,10 +66,12 @@ final class NodeBuilder
}
// reset counters/setup to copy from provided node
- void reset(Object[] copyFrom, Object upperBound)
+ void reset(Object[] copyFrom, Object upperBound, UpdateFunction updateFunction, Comparator comparator)
{
this.copyFrom = copyFrom;
this.upperBound = upperBound;
+ this.updateFunction = updateFunction;
+ this.comparator = comparator;
maxBuildKeyPosition = Math.max(maxBuildKeyPosition, buildKeyPosition);
maxBuildChildPosition = Math.max(maxBuildChildPosition, buildChildPosition);
buildKeyPosition = 0;
@@ -81,7 +88,7 @@ final class NodeBuilder
* a parent if we do not -- we got here from an earlier key -- and we need to ascend back up),
* or null if we finished the update in this node.
*/
- NodeBuilder update(Object key, Comparator comparator, ReplaceFunction replaceF)
+ NodeBuilder update(Object key)
{
assert copyFrom != null;
int copyFromKeyEnd = getKeyEnd(copyFrom);
@@ -104,9 +111,9 @@ final class NodeBuilder
if (owns)
{
if (found)
- replaceNextKey(key, replaceF);
+ replaceNextKey(key);
else
- addNewKey(key, replaceF); // handles splitting parent if necessary via ensureRoom
+ addNewKey(key); // handles splitting parent if necessary via ensureRoom
// done, so return null
return null;
@@ -122,7 +129,7 @@ final class NodeBuilder
if (found)
{
copyKeys(i);
- replaceNextKey(key, replaceF);
+ replaceNextKey(key);
copyChildren(i + 1);
return null;
}
@@ -135,7 +142,7 @@ final class NodeBuilder
// so descend into the owning child
Object newUpperBound = i < copyFromKeyEnd ? copyFrom[i] : upperBound;
Object[] descendInto = (Object[]) copyFrom[copyFromKeyEnd + i];
- ensureChild().reset(descendInto, newUpperBound);
+ ensureChild().reset(descendInto, newUpperBound, updateFunction, comparator);
return child;
}
else
@@ -177,7 +184,7 @@ final class NodeBuilder
Object[] toNode()
{
assert buildKeyPosition <= FAN_FACTOR && buildKeyPosition > 0 : buildKeyPosition;
- return buildFromRange(0, buildKeyPosition, isLeaf(copyFrom));
+ return buildFromRange(0, buildKeyPosition, isLeaf(copyFrom), false);
}
// finish up this level and pass any constructed children up to our parent, ensuring a parent exists
@@ -189,12 +196,12 @@ final class NodeBuilder
{
// split current node and move the midpoint into parent, with the two halves as children
int mid = buildKeyPosition / 2;
- parent.addExtraChild(buildFromRange(0, mid, isLeaf), buildKeys[mid]);
- parent.finishChild(buildFromRange(mid + 1, buildKeyPosition - (mid + 1), isLeaf));
+ parent.addExtraChild(buildFromRange(0, mid, isLeaf, true), buildKeys[mid]);
+ parent.finishChild(buildFromRange(mid + 1, buildKeyPosition - (mid + 1), isLeaf, false));
}
else
{
- parent.finishChild(buildFromRange(0, buildKeyPosition, isLeaf));
+ parent.finishChild(buildFromRange(0, buildKeyPosition, isLeaf, false));
}
return parent;
}
@@ -215,23 +222,23 @@ final class NodeBuilder
}
// skips the next key in copyf, and puts the provided key in the builder instead
- private void replaceNextKey(Object with, ReplaceFunction replaceF)
+ private void replaceNextKey(Object with)
{
// (this first part differs from addNewKey in that we pass the replaced object to replaceF as well)
ensureRoom(buildKeyPosition + 1);
- if (replaceF != null)
- with = replaceF.apply((V) copyFrom[copyFromKeyPosition], (V) with);
+ if (updateFunction != null)
+ with = updateFunction.apply((V) copyFrom[copyFromKeyPosition], (V) with);
buildKeys[buildKeyPosition++] = with;
copyFromKeyPosition++;
}
// puts the provided key in the builder, with no impact on treatment of data from copyf
- void addNewKey(Object key, ReplaceFunction replaceF)
+ void addNewKey(Object key)
{
ensureRoom(buildKeyPosition + 1);
- if (replaceF != null)
- key = replaceF.apply((V) key);
+ if (updateFunction != null)
+ key = updateFunction.apply((V) key);
buildKeys[buildKeyPosition++] = key;
}
@@ -269,7 +276,7 @@ final class NodeBuilder
return;
// flush even number of items so we don't waste leaf space repeatedly
- Object[] flushUp = buildFromRange(0, FAN_FACTOR, isLeaf(copyFrom));
+ Object[] flushUp = buildFromRange(0, FAN_FACTOR, isLeaf(copyFrom), true);
ensureParent().addExtraChild(flushUp, buildKeys[FAN_FACTOR]);
int size = FAN_FACTOR + 1;
assert size <= buildKeyPosition : buildKeyPosition + "," + nextBuildKeyPosition;
@@ -285,7 +292,7 @@ final class NodeBuilder
}
// builds and returns a node from the buffered objects in the given range
- private Object[] buildFromRange(int offset, int keyLength, boolean isLeaf)
+ private Object[] buildFromRange(int offset, int keyLength, boolean isLeaf, boolean isExtra)
{
Object[] a;
if (isLeaf)
@@ -299,6 +306,14 @@ final class NodeBuilder
System.arraycopy(buildKeys, offset, a, 0, keyLength);
System.arraycopy(buildChildren, offset, a, keyLength, keyLength + 1);
}
+ if (updateFunction != null)
+ {
+ if (isExtra)
+ updateFunction.allocated(ObjectSizes.sizeOfArray(a));
+ else if (a.length != copyFrom.length)
+ updateFunction.allocated(ObjectSizes.sizeOfArray(a) -
+ (copyFrom.length == 0 ? 0 : ObjectSizes.sizeOfArray(copyFrom)));
+ }
return a;
}
@@ -313,7 +328,7 @@ final class NodeBuilder
parent.child = this;
}
if (parent.upperBound == null)
- parent.reset(EMPTY_BRANCH, upperBound);
+ parent.reset(EMPTY_BRANCH, upperBound, updateFunction, comparator);
return parent;
}
diff --git a/src/java/org/apache/cassandra/utils/btree/ReplaceFunction.java b/src/java/org/apache/cassandra/utils/btree/ReplaceFunction.java
deleted file mode 100644
index a193c31890..0000000000
--- a/src/java/org/apache/cassandra/utils/btree/ReplaceFunction.java
+++ /dev/null
@@ -1,17 +0,0 @@
-package org.apache.cassandra.utils.btree;
-
-import com.google.common.base.Function;
-
-/**
- * An interface defining a function to be applied to both the object we are replacing in a BTree and
- * the object that is intended to replace it, returning the object to actually replace it.
- *
- * If this is a new insertion, that is there is no object to replace, the one argument variant of
- * the function will be called.
- *
- * @param
- */
-public interface ReplaceFunction