You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2015/07/02 11:37:21 UTC
[1/6] cassandra git commit: Ensure memtable book keeping is not
corrupted in the event we shrink usage
Repository: cassandra
Updated Branches:
refs/heads/cassandra-2.1 c5f03a988 -> b757db148
refs/heads/cassandra-2.2 92e2e4e46 -> 99f7ce9bf
refs/heads/trunk 03f72acd5 -> dea6ab1b7
Ensure memtable book keeping is not corrupted in the event we shrink usage
patch by benedict; reviewed by tjake for CASSANDRA-9681
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b757db14
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b757db14
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b757db14
Branch: refs/heads/cassandra-2.1
Commit: b757db1484473b264bf25ca5541f080d54a579a2
Parents: c5f03a9
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Jul 2 10:27:07 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Jul 2 10:27:07 2015 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/AtomicBTreeColumns.java | 2 +-
.../apache/cassandra/db/ColumnFamilyStore.java | 22 +++++-
src/java/org/apache/cassandra/db/Memtable.java | 15 ++--
.../org/apache/cassandra/utils/FBUtilities.java | 10 +++
.../apache/cassandra/utils/memory/HeapPool.java | 4 +-
.../utils/memory/MemtableAllocator.java | 39 +++++++----
.../cassandra/utils/memory/MemtablePool.java | 73 ++++++++++++--------
.../utils/memory/NativeAllocatorTest.java | 18 ++++-
9 files changed, 132 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 762b88b..25f7c1d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.8
+ * Ensure memtable book keeping is not corrupted in the event we shrink usage (CASSANDRA-9681)
* Update internal python driver for cqlsh (CASSANDRA-9064)
* Fix IndexOutOfBoundsException when inserting tuple with too many
elements using the string literal notation (CASSANDRA-9559)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
index 47f0b85..d9eb29c 100644
--- a/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
+++ b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
@@ -505,7 +505,7 @@ public class AtomicBTreeColumns extends ColumnFamily
protected void finish()
{
- allocator.onHeap().allocate(heapSize, writeOp);
+ allocator.onHeap().adjust(heapSize, writeOp);
reclaimer.commit();
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index fa527c7..8e67cdc 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -35,6 +35,7 @@ import com.google.common.collect.*;
import com.google.common.util.concurrent.*;
import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.utils.memory.MemtablePool;
import org.json.simple.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1157,6 +1158,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
float largestRatio = 0f;
Memtable largest = null;
+ float liveOnHeap = 0, liveOffHeap = 0;
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
// we take a reference to the current main memtable for the CF prior to snapping its ownership ratios
@@ -1181,19 +1183,37 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
float ratio = Math.max(onHeap, offHeap);
-
if (ratio > largestRatio)
{
largest = current;
largestRatio = ratio;
}
+
+ liveOnHeap += onHeap;
+ liveOffHeap += offHeap;
}
if (largest != null)
+ {
+ float usedOnHeap = Memtable.MEMORY_POOL.onHeap.usedRatio();
+ float usedOffHeap = Memtable.MEMORY_POOL.offHeap.usedRatio();
+ float flushingOnHeap = Memtable.MEMORY_POOL.onHeap.reclaimingRatio();
+ float flushingOffHeap = Memtable.MEMORY_POOL.offHeap.reclaimingRatio();
+ float thisOnHeap = largest.getAllocator().onHeap().ownershipRatio();
+ float thisOffHeap = largest.getAllocator().onHeap().ownershipRatio();
+ logger.info("Flushing largest {} to free up room. Used total: {}, live: {}, flushing: {}, this: {}",
+ largest.cfs, ratio(usedOnHeap, usedOffHeap), ratio(liveOnHeap, liveOffHeap),
+ ratio(flushingOnHeap, flushingOffHeap), ratio(thisOnHeap, thisOffHeap));
largest.cfs.switchMemtableIfCurrent(largest);
+ }
}
}
+ private static String ratio(float onHeap, float offHeap)
+ {
+ return String.format("%.0f/%.0f", onHeap, offHeap);
+ }
+
public void maybeUpdateRowCache(DecoratedKey key)
{
if (!isRowCacheEnabled())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/Memtable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java
index a50a614..9f6cf9b 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -239,7 +239,8 @@ public class Memtable
public String toString()
{
return String.format("Memtable-%s@%s(%s serialized bytes, %s ops, %.0f%%/%.0f%% of on/off-heap limit)",
- cfs.name, hashCode(), liveDataSize, currentOperations, 100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
+ cfs.name, hashCode(), FBUtilities.prettyPrintMemory(liveDataSize.get()), currentOperations,
+ 100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
}
/**
@@ -378,19 +379,21 @@ public class Memtable
if (writer.getFilePointer() > 0)
{
- writer.isolateReferences();
+ logger.info(String.format("Completed flushing %s (%s) for commitlog position %s",
+ writer.getFilename(),
+ FBUtilities.prettyPrintMemory(writer.getOnDiskFilePointer()),
+ context));
+ writer.isolateReferences();
// temp sstables should contain non-repaired data.
ssTable = writer.closeAndOpenReader();
- logger.info(String.format("Completed flushing %s (%d bytes) for commitlog position %s",
- ssTable.getFilename(), new File(ssTable.getFilename()).length(), context));
}
else
{
+ logger.info("Completed flushing %s; nothing needed to be retained. Commitlog position was {}",
+ writer.getFilename(), context);
writer.abort();
ssTable = null;
- logger.info("Completed flushing; nothing needed to be retained. Commitlog position was {}",
- context);
}
if (heavilyContendedRowCount > 0)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index 4c81b2a..68eb864 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -28,6 +28,7 @@ import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+import java.text.NumberFormat;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -570,6 +571,15 @@ public class FBUtilities
}
}
+ public static String prettyPrintMemory(long size)
+ {
+ if (size >= 1 << 30)
+ return String.format("%.3fGiB", size / (double) (1 << 30));
+ if (size >= 1 << 20)
+ return String.format("%.3fMiB", size / (double) (1 << 20));
+ return String.format("%.3fKiB", size / (double) (1 << 10));
+ }
+
/**
* Starts and waits for the given @param pb to finish.
* @throws java.io.IOException on non-zero exit code
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/HeapPool.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/HeapPool.java b/src/java/org/apache/cassandra/utils/memory/HeapPool.java
index a04947c..2a19d9c 100644
--- a/src/java/org/apache/cassandra/utils/memory/HeapPool.java
+++ b/src/java/org/apache/cassandra/utils/memory/HeapPool.java
@@ -75,13 +75,13 @@ public class HeapPool extends MemtablePool
public Reclaimer reclaimImmediately(Cell cell)
{
- onHeap().release(cell.name().dataSize() + cell.value().remaining());
+ onHeap().released(cell.name().dataSize() + cell.value().remaining());
return this;
}
public Reclaimer reclaimImmediately(DecoratedKey key)
{
- onHeap().release(key.getKey().remaining());
+ onHeap().released(key.getKey().remaining());
return this;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
index e814b4d..f5e743c 100644
--- a/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
+++ b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
@@ -160,13 +160,24 @@ public abstract class MemtableAllocator
// currently no corroboration/enforcement of this is performed.
void releaseAll()
{
- parent.adjustAcquired(-ownsUpdater.getAndSet(this, 0), false);
- parent.adjustReclaiming(-reclaimingUpdater.getAndSet(this, 0));
+ parent.released(ownsUpdater.getAndSet(this, 0));
+ parent.reclaimed(reclaimingUpdater.getAndSet(this, 0));
+ }
+
+ // like allocate, but permits allocations to be negative
+ public void adjust(long size, OpOrder.Group opGroup)
+ {
+ if (size <= 0)
+ released(-size);
+ else
+ allocate(size, opGroup);
}
// allocate memory in the tracker, and mark ourselves as owning it
public void allocate(long size, OpOrder.Group opGroup)
{
+ assert size >= 0;
+
while (true)
{
if (parent.tryAllocate(size))
@@ -190,23 +201,23 @@ public abstract class MemtableAllocator
}
}
- // retroactively mark an amount allocated amd acquired in the tracker, and owned by us
- void allocated(long size)
+ // retroactively mark an amount allocated and acquired in the tracker, and owned by us
+ private void allocated(long size)
{
- parent.adjustAcquired(size, true);
+ parent.allocated(size);
ownsUpdater.addAndGet(this, size);
}
// retroactively mark an amount acquired in the tracker, and owned by us
- void acquired(long size)
+ private void acquired(long size)
{
- parent.adjustAcquired(size, false);
+ parent.acquired(size);
ownsUpdater.addAndGet(this, size);
}
- void release(long size)
+ void released(long size)
{
- parent.adjustAcquired(-size, false);
+ parent.released(size);
ownsUpdater.addAndGet(this, -size);
}
@@ -217,11 +228,11 @@ public abstract class MemtableAllocator
{
long cur = owns;
long prev = reclaiming;
- if (reclaimingUpdater.compareAndSet(this, prev, cur))
- {
- parent.adjustReclaiming(cur - prev);
- return;
- }
+ if (!reclaimingUpdater.compareAndSet(this, prev, cur))
+ continue;
+
+ parent.reclaiming(cur - prev);
+ return;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
index 1d219bb..bb85884 100644
--- a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
+++ b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
@@ -130,10 +130,8 @@ public abstract class MemtablePool
* apply the size adjustment to allocated, bypassing any limits or constraints. If this reduces the
* allocated total, we will signal waiters
*/
- void adjustAllocated(long size)
+ private void adjustAllocated(long size)
{
- if (size == 0)
- return;
while (true)
{
long cur = allocated;
@@ -142,38 +140,43 @@ public abstract class MemtablePool
}
}
- // 'acquires' an amount of memory, and maybe also marks it allocated. This method is meant to be overridden
- // by implementations with a separate concept of acquired/allocated. As this method stands, an acquire
- // without an allocate is a no-op (acquisition is achieved through allocation), however a release (where size < 0)
- // is always processed and accounted for in allocated.
- void adjustAcquired(long size, boolean alsoAllocated)
+ void allocated(long size)
{
- if (size > 0 || alsoAllocated)
- {
- if (alsoAllocated)
- adjustAllocated(size);
- maybeClean();
- }
- else if (size < 0)
- {
- adjustAllocated(size);
- hasRoom.signalAll();
- }
+ assert size >= 0;
+ if (size == 0)
+ return;
+
+ adjustAllocated(size);
+ maybeClean();
+ }
+
+ void acquired(long size)
+ {
+ maybeClean();
+ }
+
+ void released(long size)
+ {
+ assert size >= 0;
+ adjustAllocated(-size);
+ hasRoom.signalAll();
}
- // space reclaimed should be released prior to calling this, to avoid triggering unnecessary cleans
- void adjustReclaiming(long reclaiming)
+ void reclaiming(long size)
{
- if (reclaiming == 0)
+ if (size == 0)
return;
- reclaimingUpdater.addAndGet(this, reclaiming);
- if (reclaiming < 0 && updateNextClean() && cleaner != null)
- cleaner.trigger();
+ reclaimingUpdater.addAndGet(this, size);
}
- public long allocated()
+ void reclaimed(long size)
{
- return allocated;
+ if (size == 0)
+ return;
+
+ reclaimingUpdater.addAndGet(this, -size);
+ if (updateNextClean() && cleaner != null)
+ cleaner.trigger();
}
public long used()
@@ -181,6 +184,22 @@ public abstract class MemtablePool
return allocated;
}
+ public float reclaimingRatio()
+ {
+ float r = reclaiming / (float) limit;
+ if (Float.isNaN(r))
+ return 0;
+ return r;
+ }
+
+ public float usedRatio()
+ {
+ float r = allocated / (float) limit;
+ if (Float.isNaN(r))
+ return 0;
+ return r;
+ }
+
public MemtableAllocator.SubAllocator newAllocator()
{
return new MemtableAllocator.SubAllocator(this);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java b/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
index 83d6c0c..7f87fcd 100644
--- a/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
+++ b/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
@@ -56,7 +56,7 @@ public class NativeAllocatorTest
}
if (isClean.getCount() > 0)
{
- allocatorRef.get().offHeap().release(80);
+ allocatorRef.get().offHeap().released(80);
isClean.countDown();
}
}
@@ -79,6 +79,22 @@ public class NativeAllocatorTest
// allocate normal, check accounted and not cleaned
allocator.allocate(10, group);
Assert.assertEquals(10, allocator.offHeap().owns());
+ // confirm adjustment works
+ allocator.offHeap().adjust(-10, group);
+ Assert.assertEquals(0, allocator.offHeap().owns());
+ allocator.offHeap().adjust(10, group);
+ Assert.assertEquals(10, allocator.offHeap().owns());
+ // confirm we cannot allocate negative
+ boolean success = false;
+ try
+ {
+ allocator.offHeap().allocate(-10, group);
+ }
+ catch (AssertionError e)
+ {
+ success = true;
+ }
+ Assert.assertTrue(success);
Uninterruptibles.sleepUninterruptibly(10L, TimeUnit.MILLISECONDS);
Assert.assertEquals(1, isClean.getCount());
[2/6] cassandra git commit: Ensure memtable book keeping is not
corrupted in the event we shrink usage
Posted by be...@apache.org.
Ensure memtable book keeping is not corrupted in the event we shrink usage
patch by benedict; reviewed by tjake for CASSANDRA-9681
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b757db14
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b757db14
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b757db14
Branch: refs/heads/cassandra-2.2
Commit: b757db1484473b264bf25ca5541f080d54a579a2
Parents: c5f03a9
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Jul 2 10:27:07 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Jul 2 10:27:07 2015 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/AtomicBTreeColumns.java | 2 +-
.../apache/cassandra/db/ColumnFamilyStore.java | 22 +++++-
src/java/org/apache/cassandra/db/Memtable.java | 15 ++--
.../org/apache/cassandra/utils/FBUtilities.java | 10 +++
.../apache/cassandra/utils/memory/HeapPool.java | 4 +-
.../utils/memory/MemtableAllocator.java | 39 +++++++----
.../cassandra/utils/memory/MemtablePool.java | 73 ++++++++++++--------
.../utils/memory/NativeAllocatorTest.java | 18 ++++-
9 files changed, 132 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 762b88b..25f7c1d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.8
+ * Ensure memtable book keeping is not corrupted in the event we shrink usage (CASSANDRA-9681)
* Update internal python driver for cqlsh (CASSANDRA-9064)
* Fix IndexOutOfBoundsException when inserting tuple with too many
elements using the string literal notation (CASSANDRA-9559)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
index 47f0b85..d9eb29c 100644
--- a/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
+++ b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
@@ -505,7 +505,7 @@ public class AtomicBTreeColumns extends ColumnFamily
protected void finish()
{
- allocator.onHeap().allocate(heapSize, writeOp);
+ allocator.onHeap().adjust(heapSize, writeOp);
reclaimer.commit();
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index fa527c7..8e67cdc 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -35,6 +35,7 @@ import com.google.common.collect.*;
import com.google.common.util.concurrent.*;
import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.utils.memory.MemtablePool;
import org.json.simple.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1157,6 +1158,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
float largestRatio = 0f;
Memtable largest = null;
+ float liveOnHeap = 0, liveOffHeap = 0;
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
// we take a reference to the current main memtable for the CF prior to snapping its ownership ratios
@@ -1181,19 +1183,37 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
float ratio = Math.max(onHeap, offHeap);
-
if (ratio > largestRatio)
{
largest = current;
largestRatio = ratio;
}
+
+ liveOnHeap += onHeap;
+ liveOffHeap += offHeap;
}
if (largest != null)
+ {
+ float usedOnHeap = Memtable.MEMORY_POOL.onHeap.usedRatio();
+ float usedOffHeap = Memtable.MEMORY_POOL.offHeap.usedRatio();
+ float flushingOnHeap = Memtable.MEMORY_POOL.onHeap.reclaimingRatio();
+ float flushingOffHeap = Memtable.MEMORY_POOL.offHeap.reclaimingRatio();
+ float thisOnHeap = largest.getAllocator().onHeap().ownershipRatio();
+ float thisOffHeap = largest.getAllocator().onHeap().ownershipRatio();
+ logger.info("Flushing largest {} to free up room. Used total: {}, live: {}, flushing: {}, this: {}",
+ largest.cfs, ratio(usedOnHeap, usedOffHeap), ratio(liveOnHeap, liveOffHeap),
+ ratio(flushingOnHeap, flushingOffHeap), ratio(thisOnHeap, thisOffHeap));
largest.cfs.switchMemtableIfCurrent(largest);
+ }
}
}
+ private static String ratio(float onHeap, float offHeap)
+ {
+ return String.format("%.0f/%.0f", onHeap, offHeap);
+ }
+
public void maybeUpdateRowCache(DecoratedKey key)
{
if (!isRowCacheEnabled())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/Memtable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java
index a50a614..9f6cf9b 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -239,7 +239,8 @@ public class Memtable
public String toString()
{
return String.format("Memtable-%s@%s(%s serialized bytes, %s ops, %.0f%%/%.0f%% of on/off-heap limit)",
- cfs.name, hashCode(), liveDataSize, currentOperations, 100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
+ cfs.name, hashCode(), FBUtilities.prettyPrintMemory(liveDataSize.get()), currentOperations,
+ 100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
}
/**
@@ -378,19 +379,21 @@ public class Memtable
if (writer.getFilePointer() > 0)
{
- writer.isolateReferences();
+ logger.info(String.format("Completed flushing %s (%s) for commitlog position %s",
+ writer.getFilename(),
+ FBUtilities.prettyPrintMemory(writer.getOnDiskFilePointer()),
+ context));
+ writer.isolateReferences();
// temp sstables should contain non-repaired data.
ssTable = writer.closeAndOpenReader();
- logger.info(String.format("Completed flushing %s (%d bytes) for commitlog position %s",
- ssTable.getFilename(), new File(ssTable.getFilename()).length(), context));
}
else
{
+ logger.info("Completed flushing %s; nothing needed to be retained. Commitlog position was {}",
+ writer.getFilename(), context);
writer.abort();
ssTable = null;
- logger.info("Completed flushing; nothing needed to be retained. Commitlog position was {}",
- context);
}
if (heavilyContendedRowCount > 0)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index 4c81b2a..68eb864 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -28,6 +28,7 @@ import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+import java.text.NumberFormat;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -570,6 +571,15 @@ public class FBUtilities
}
}
+ public static String prettyPrintMemory(long size)
+ {
+ if (size >= 1 << 30)
+ return String.format("%.3fGiB", size / (double) (1 << 30));
+ if (size >= 1 << 20)
+ return String.format("%.3fMiB", size / (double) (1 << 20));
+ return String.format("%.3fKiB", size / (double) (1 << 10));
+ }
+
/**
* Starts and waits for the given @param pb to finish.
* @throws java.io.IOException on non-zero exit code
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/HeapPool.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/HeapPool.java b/src/java/org/apache/cassandra/utils/memory/HeapPool.java
index a04947c..2a19d9c 100644
--- a/src/java/org/apache/cassandra/utils/memory/HeapPool.java
+++ b/src/java/org/apache/cassandra/utils/memory/HeapPool.java
@@ -75,13 +75,13 @@ public class HeapPool extends MemtablePool
public Reclaimer reclaimImmediately(Cell cell)
{
- onHeap().release(cell.name().dataSize() + cell.value().remaining());
+ onHeap().released(cell.name().dataSize() + cell.value().remaining());
return this;
}
public Reclaimer reclaimImmediately(DecoratedKey key)
{
- onHeap().release(key.getKey().remaining());
+ onHeap().released(key.getKey().remaining());
return this;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
index e814b4d..f5e743c 100644
--- a/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
+++ b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
@@ -160,13 +160,24 @@ public abstract class MemtableAllocator
// currently no corroboration/enforcement of this is performed.
void releaseAll()
{
- parent.adjustAcquired(-ownsUpdater.getAndSet(this, 0), false);
- parent.adjustReclaiming(-reclaimingUpdater.getAndSet(this, 0));
+ parent.released(ownsUpdater.getAndSet(this, 0));
+ parent.reclaimed(reclaimingUpdater.getAndSet(this, 0));
+ }
+
+ // like allocate, but permits allocations to be negative
+ public void adjust(long size, OpOrder.Group opGroup)
+ {
+ if (size <= 0)
+ released(-size);
+ else
+ allocate(size, opGroup);
}
// allocate memory in the tracker, and mark ourselves as owning it
public void allocate(long size, OpOrder.Group opGroup)
{
+ assert size >= 0;
+
while (true)
{
if (parent.tryAllocate(size))
@@ -190,23 +201,23 @@ public abstract class MemtableAllocator
}
}
- // retroactively mark an amount allocated amd acquired in the tracker, and owned by us
- void allocated(long size)
+ // retroactively mark an amount allocated and acquired in the tracker, and owned by us
+ private void allocated(long size)
{
- parent.adjustAcquired(size, true);
+ parent.allocated(size);
ownsUpdater.addAndGet(this, size);
}
// retroactively mark an amount acquired in the tracker, and owned by us
- void acquired(long size)
+ private void acquired(long size)
{
- parent.adjustAcquired(size, false);
+ parent.acquired(size);
ownsUpdater.addAndGet(this, size);
}
- void release(long size)
+ void released(long size)
{
- parent.adjustAcquired(-size, false);
+ parent.released(size);
ownsUpdater.addAndGet(this, -size);
}
@@ -217,11 +228,11 @@ public abstract class MemtableAllocator
{
long cur = owns;
long prev = reclaiming;
- if (reclaimingUpdater.compareAndSet(this, prev, cur))
- {
- parent.adjustReclaiming(cur - prev);
- return;
- }
+ if (!reclaimingUpdater.compareAndSet(this, prev, cur))
+ continue;
+
+ parent.reclaiming(cur - prev);
+ return;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
index 1d219bb..bb85884 100644
--- a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
+++ b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
@@ -130,10 +130,8 @@ public abstract class MemtablePool
* apply the size adjustment to allocated, bypassing any limits or constraints. If this reduces the
* allocated total, we will signal waiters
*/
- void adjustAllocated(long size)
+ private void adjustAllocated(long size)
{
- if (size == 0)
- return;
while (true)
{
long cur = allocated;
@@ -142,38 +140,43 @@ public abstract class MemtablePool
}
}
- // 'acquires' an amount of memory, and maybe also marks it allocated. This method is meant to be overridden
- // by implementations with a separate concept of acquired/allocated. As this method stands, an acquire
- // without an allocate is a no-op (acquisition is achieved through allocation), however a release (where size < 0)
- // is always processed and accounted for in allocated.
- void adjustAcquired(long size, boolean alsoAllocated)
+ void allocated(long size)
{
- if (size > 0 || alsoAllocated)
- {
- if (alsoAllocated)
- adjustAllocated(size);
- maybeClean();
- }
- else if (size < 0)
- {
- adjustAllocated(size);
- hasRoom.signalAll();
- }
+ assert size >= 0;
+ if (size == 0)
+ return;
+
+ adjustAllocated(size);
+ maybeClean();
+ }
+
+ void acquired(long size)
+ {
+ maybeClean();
+ }
+
+ void released(long size)
+ {
+ assert size >= 0;
+ adjustAllocated(-size);
+ hasRoom.signalAll();
}
- // space reclaimed should be released prior to calling this, to avoid triggering unnecessary cleans
- void adjustReclaiming(long reclaiming)
+ void reclaiming(long size)
{
- if (reclaiming == 0)
+ if (size == 0)
return;
- reclaimingUpdater.addAndGet(this, reclaiming);
- if (reclaiming < 0 && updateNextClean() && cleaner != null)
- cleaner.trigger();
+ reclaimingUpdater.addAndGet(this, size);
}
- public long allocated()
+ void reclaimed(long size)
{
- return allocated;
+ if (size == 0)
+ return;
+
+ reclaimingUpdater.addAndGet(this, -size);
+ if (updateNextClean() && cleaner != null)
+ cleaner.trigger();
}
public long used()
@@ -181,6 +184,22 @@ public abstract class MemtablePool
return allocated;
}
+ public float reclaimingRatio()
+ {
+ float r = reclaiming / (float) limit;
+ if (Float.isNaN(r))
+ return 0;
+ return r;
+ }
+
+ public float usedRatio()
+ {
+ float r = allocated / (float) limit;
+ if (Float.isNaN(r))
+ return 0;
+ return r;
+ }
+
public MemtableAllocator.SubAllocator newAllocator()
{
return new MemtableAllocator.SubAllocator(this);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java b/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
index 83d6c0c..7f87fcd 100644
--- a/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
+++ b/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
@@ -56,7 +56,7 @@ public class NativeAllocatorTest
}
if (isClean.getCount() > 0)
{
- allocatorRef.get().offHeap().release(80);
+ allocatorRef.get().offHeap().released(80);
isClean.countDown();
}
}
@@ -79,6 +79,22 @@ public class NativeAllocatorTest
// allocate normal, check accounted and not cleaned
allocator.allocate(10, group);
Assert.assertEquals(10, allocator.offHeap().owns());
+ // confirm adjustment works
+ allocator.offHeap().adjust(-10, group);
+ Assert.assertEquals(0, allocator.offHeap().owns());
+ allocator.offHeap().adjust(10, group);
+ Assert.assertEquals(10, allocator.offHeap().owns());
+ // confirm we cannot allocate negative
+ boolean success = false;
+ try
+ {
+ allocator.offHeap().allocate(-10, group);
+ }
+ catch (AssertionError e)
+ {
+ success = true;
+ }
+ Assert.assertTrue(success);
Uninterruptibles.sleepUninterruptibly(10L, TimeUnit.MILLISECONDS);
Assert.assertEquals(1, isClean.getCount());
[6/6] cassandra git commit: Merge branch 'cassandra-2.2' into trunk
Posted by be...@apache.org.
Merge branch 'cassandra-2.2' into trunk
Conflicts:
src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
src/java/org/apache/cassandra/utils/memory/HeapPool.java
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/dea6ab1b
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/dea6ab1b
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/dea6ab1b
Branch: refs/heads/trunk
Commit: dea6ab1b769943eedaeb590d545d7c476c4a2466
Parents: 03f72ac 99f7ce9
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Jul 2 10:34:46 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Jul 2 10:34:46 2015 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/ColumnFamilyStore.java | 22 +++++-
src/java/org/apache/cassandra/db/Memtable.java | 15 ++--
.../db/partitions/AtomicBTreePartition.java | 2 +-
.../org/apache/cassandra/utils/FBUtilities.java | 10 +++
.../utils/memory/MemtableAllocator.java | 39 +++++++----
.../cassandra/utils/memory/MemtablePool.java | 73 ++++++++++++--------
.../utils/memory/NativeAllocatorTest.java | 18 ++++-
8 files changed, 131 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dea6ab1b/CHANGES.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dea6ab1b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dea6ab1b/src/java/org/apache/cassandra/db/Memtable.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/Memtable.java
index e82c35e,6e4802f..b6341aa
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@@ -253,35 -256,23 +253,36 @@@ public class Memtable implements Compar
public String toString()
{
return String.format("Memtable-%s@%s(%s serialized bytes, %s ops, %.0f%%/%.0f%% of on/off-heap limit)",
- cfs.name, hashCode(), liveDataSize, currentOperations, 100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
+ cfs.name, hashCode(), FBUtilities.prettyPrintMemory(liveDataSize.get()), currentOperations,
+ 100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
}
- /**
- * @param startWith Include data in the result from and including this key and to the end of the memtable
- * @return An iterator of entries with the data from the start key
- */
- public Iterator<Map.Entry<DecoratedKey, ColumnFamily>> getEntryIterator(final RowPosition startWith, final RowPosition stopAt)
+ public UnfilteredPartitionIterator makePartitionIterator(final ColumnFilter columnFilter, final DataRange dataRange, final boolean isForThrift)
{
- return new Iterator<Map.Entry<DecoratedKey, ColumnFamily>>()
- {
- private Iterator<? extends Map.Entry<? extends RowPosition, AtomicBTreeColumns>> iter = stopAt.isMinimum()
- ? rows.tailMap(startWith).entrySet().iterator()
- : rows.subMap(startWith, true, stopAt, true).entrySet().iterator();
+ AbstractBounds<PartitionPosition> keyRange = dataRange.keyRange();
+
+ boolean startIsMin = keyRange.left.isMinimum();
+ boolean stopIsMin = keyRange.right.isMinimum();
+
+ boolean isBound = keyRange instanceof Bounds;
+ boolean includeStart = isBound || keyRange instanceof IncludingExcludingBounds;
+ boolean includeStop = isBound || keyRange instanceof Range;
+ Map<PartitionPosition, AtomicBTreePartition> subMap;
+ if (startIsMin)
+ subMap = stopIsMin ? partitions : partitions.headMap(keyRange.right, includeStop);
+ else
+ subMap = stopIsMin
+ ? partitions.tailMap(keyRange.left, includeStart)
+ : partitions.subMap(keyRange.left, includeStart, keyRange.right, includeStop);
+
+ final Iterator<Map.Entry<PartitionPosition, AtomicBTreePartition>> iter = subMap.entrySet().iterator();
- private Map.Entry<? extends RowPosition, ? extends ColumnFamily> currentEntry;
+ return new AbstractUnfilteredPartitionIterator()
+ {
+ public boolean isForThrift()
+ {
+ return isForThrift;
+ }
public boolean hasNext()
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dea6ab1b/src/java/org/apache/cassandra/db/partitions/AtomicBTreePartition.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/partitions/AtomicBTreePartition.java
index 4edd707,0000000..777e7b4
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/db/partitions/AtomicBTreePartition.java
+++ b/src/java/org/apache/cassandra/db/partitions/AtomicBTreePartition.java
@@@ -1,819 -1,0 +1,819 @@@
+/*
+ * 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.db.partitions;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.Iterators;
+
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.rows.*;
+import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.index.SecondaryIndexManager;
+import org.apache.cassandra.utils.btree.BTree;
+import org.apache.cassandra.utils.btree.BTreeSearchIterator;
+import org.apache.cassandra.utils.btree.UpdateFunction;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.SearchIterator;
+import org.apache.cassandra.utils.concurrent.OpOrder;
+import org.apache.cassandra.utils.concurrent.Locks;
+import org.apache.cassandra.utils.memory.MemtableAllocator;
+import org.apache.cassandra.utils.memory.HeapAllocator;
+import org.apache.cassandra.service.StorageService;
+import static org.apache.cassandra.db.index.SecondaryIndexManager.Updater;
+
+/**
+ * A thread-safe and atomic Partition implementation.
+ *
+ * Operations (in particular addAll) on this implementation are atomic and
+ * isolated (in the sense of ACID). Typically a addAll is guaranteed that no
+ * other thread can see the state where only parts but not all rows have
+ * been added.
+ */
+public class AtomicBTreePartition implements Partition
+{
+ private static final Logger logger = LoggerFactory.getLogger(AtomicBTreePartition.class);
+
+ public static final long EMPTY_SIZE = ObjectSizes.measure(new AtomicBTreePartition(CFMetaData.createFake("keyspace", "table"),
+ StorageService.getPartitioner().decorateKey(ByteBuffer.allocate(1)),
+ null));
+
+ // Reserved values for wasteTracker field. These values must not be consecutive (see avoidReservedValues)
+ private static final int TRACKER_NEVER_WASTED = 0;
+ private static final int TRACKER_PESSIMISTIC_LOCKING = Integer.MAX_VALUE;
+
+ // The granularity with which we track wasted allocation/work; we round up
+ private static final int ALLOCATION_GRANULARITY_BYTES = 1024;
+ // The number of bytes we have to waste in excess of our acceptable realtime rate of waste (defined below)
+ private static final long EXCESS_WASTE_BYTES = 10 * 1024 * 1024L;
+ private static final int EXCESS_WASTE_OFFSET = (int) (EXCESS_WASTE_BYTES / ALLOCATION_GRANULARITY_BYTES);
+ // Note this is a shift, because dividing a long time and then picking the low 32 bits doesn't give correct rollover behavior
+ private static final int CLOCK_SHIFT = 17;
+ // CLOCK_GRANULARITY = 1^9ns >> CLOCK_SHIFT == 132us == (1/7.63)ms
+
+ /**
+ * (clock + allocation) granularity are combined to give us an acceptable (waste) allocation rate that is defined by
+ * the passage of real time of ALLOCATION_GRANULARITY_BYTES/CLOCK_GRANULARITY, or in this case 7.63Kb/ms, or 7.45Mb/s
+ *
+ * in wasteTracker we maintain within EXCESS_WASTE_OFFSET before the current time; whenever we waste bytes
+ * we increment the current value if it is within this window, and set it to the min of the window plus our waste
+ * otherwise.
+ */
+ private volatile int wasteTracker = TRACKER_NEVER_WASTED;
+
+ private static final AtomicIntegerFieldUpdater<AtomicBTreePartition> wasteTrackerUpdater = AtomicIntegerFieldUpdater.newUpdater(AtomicBTreePartition.class, "wasteTracker");
+
+ private static final DeletionInfo LIVE = DeletionInfo.live();
+ // This is a small optimization: DeletionInfo is mutable, but we know that we will always copy it in that class,
+ // so we can safely alias one DeletionInfo.live() reference and avoid some allocations.
+ private static final Holder EMPTY = new Holder(BTree.empty(), LIVE, null, RowStats.NO_STATS);
+
+ private final CFMetaData metadata;
+ private final DecoratedKey partitionKey;
+ private final MemtableAllocator allocator;
+
+ private volatile Holder ref;
+
+ private static final AtomicReferenceFieldUpdater<AtomicBTreePartition, Holder> refUpdater = AtomicReferenceFieldUpdater.newUpdater(AtomicBTreePartition.class, Holder.class, "ref");
+
+ public AtomicBTreePartition(CFMetaData metadata, DecoratedKey partitionKey, MemtableAllocator allocator)
+ {
+ this.metadata = metadata;
+ this.partitionKey = partitionKey;
+ this.allocator = allocator;
+ this.ref = EMPTY;
+ }
+
+ public boolean isEmpty()
+ {
+ return ref.deletionInfo.isLive() && BTree.isEmpty(ref.tree) && ref.staticRow == null;
+ }
+
+ public CFMetaData metadata()
+ {
+ return metadata;
+ }
+
+ public DecoratedKey partitionKey()
+ {
+ return partitionKey;
+ }
+
+ public DeletionTime partitionLevelDeletion()
+ {
+ return ref.deletionInfo.getPartitionDeletion();
+ }
+
+ public PartitionColumns columns()
+ {
+ // We don't really know which columns will be part of the update, so assume it's all of them
+ return metadata.partitionColumns();
+ }
+
+ public boolean hasRows()
+ {
+ return !BTree.isEmpty(ref.tree);
+ }
+
+ public RowStats stats()
+ {
+ return ref.stats;
+ }
+
+ public Row getRow(Clustering clustering)
+ {
+ Row row = searchIterator(ColumnFilter.selection(columns()), false).next(clustering);
+ // Note that for statics, this will never return null, this will return an empty row. However,
+ // it's more consistent for this method to return null if we don't really have a static row.
+ return row == null || (clustering == Clustering.STATIC_CLUSTERING && row.isEmpty()) ? null : row;
+ }
+
+ public SearchIterator<Clustering, Row> searchIterator(final ColumnFilter columns, final boolean reversed)
+ {
+ // TODO: we could optimize comparison for "NativeRow" à la #6755
+ final Holder current = ref;
+ return new SearchIterator<Clustering, Row>()
+ {
+ private final SearchIterator<Clustering, MemtableRowData> rawIter = new BTreeSearchIterator<>(current.tree, metadata.comparator, !reversed);
+ private final MemtableRowData.ReusableRow row = allocator.newReusableRow();
+ private final ReusableFilteringRow filter = new ReusableFilteringRow(columns.fetchedColumns().regulars, columns);
+ private final long partitionDeletion = current.deletionInfo.getPartitionDeletion().markedForDeleteAt();
+
+ public boolean hasNext()
+ {
+ return rawIter.hasNext();
+ }
+
+ public Row next(Clustering key)
+ {
+ if (key == Clustering.STATIC_CLUSTERING)
+ return makeStatic(columns, current, allocator);
+
+ MemtableRowData data = rawIter.next(key);
+ // We also need to find if there is a range tombstone covering this key
+ RangeTombstone rt = current.deletionInfo.rangeCovering(key);
+
+ if (data == null)
+ {
+ // If we have a range tombstone but not data, "fake" the RT by return a row deletion
+ // corresponding to the tombstone.
+ if (rt != null && rt.deletionTime().markedForDeleteAt() > partitionDeletion)
+ return filter.setRowDeletion(rt.deletionTime()).setTo(emptyDeletedRow(key, rt.deletionTime()));
+ return null;
+ }
+
+ row.setTo(data);
+
+ filter.setRowDeletion(null);
+ if (rt == null || rt.deletionTime().markedForDeleteAt() < partitionDeletion)
+ {
+ filter.setDeletionTimestamp(partitionDeletion);
+ }
+ else
+ {
+ filter.setDeletionTimestamp(rt.deletionTime().markedForDeleteAt());
+ // If we have a range tombstone covering that row and it's bigger than the row deletion itself, then
+ // we replace the row deletion by the tombstone deletion as a way to return the tombstone.
+ if (rt.deletionTime().supersedes(row.deletion()))
+ filter.setRowDeletion(rt.deletionTime());
+ }
+
+ return filter.setTo(row);
+ }
+ };
+ }
+
+ private static Row emptyDeletedRow(Clustering clustering, DeletionTime deletion)
+ {
+ return new AbstractRow()
+ {
+ public Columns columns()
+ {
+ return Columns.NONE;
+ }
+
+ public LivenessInfo primaryKeyLivenessInfo()
+ {
+ return LivenessInfo.NONE;
+ }
+
+ public DeletionTime deletion()
+ {
+ return deletion;
+ }
+
+ public boolean isEmpty()
+ {
+ return true;
+ }
+
+ public boolean hasComplexDeletion()
+ {
+ return false;
+ }
+
+ public Clustering clustering()
+ {
+ return clustering;
+ }
+
+ public Cell getCell(ColumnDefinition c)
+ {
+ return null;
+ }
+
+ public Cell getCell(ColumnDefinition c, CellPath path)
+ {
+ return null;
+ }
+
+ public Iterator<Cell> getCells(ColumnDefinition c)
+ {
+ return null;
+ }
+
+ public DeletionTime getDeletion(ColumnDefinition c)
+ {
+ return DeletionTime.LIVE;
+ }
+
+ public Iterator<Cell> iterator()
+ {
+ return Iterators.<Cell>emptyIterator();
+ }
+
+ public SearchIterator<ColumnDefinition, ColumnData> searchIterator()
+ {
+ return new SearchIterator<ColumnDefinition, ColumnData>()
+ {
+ public boolean hasNext()
+ {
+ return false;
+ }
+
+ public ColumnData next(ColumnDefinition column)
+ {
+ return null;
+ }
+ };
+ }
+
+ public Row takeAlias()
+ {
+ return this;
+ }
+ };
+ }
+
+ public UnfilteredRowIterator unfilteredIterator()
+ {
+ return unfilteredIterator(ColumnFilter.selection(columns()), Slices.ALL, false);
+ }
+
+ public UnfilteredRowIterator unfilteredIterator(ColumnFilter selection, Slices slices, boolean reversed)
+ {
+ if (slices.size() == 0)
+ {
+ Holder current = ref;
+ DeletionTime partitionDeletion = current.deletionInfo.getPartitionDeletion();
+ if (selection.fetchedColumns().statics.isEmpty() && partitionDeletion.isLive())
+ return UnfilteredRowIterators.emptyIterator(metadata, partitionKey, reversed);
+
+ return new AbstractUnfilteredRowIterator(metadata,
+ partitionKey,
+ partitionDeletion,
+ selection.fetchedColumns(),
+ makeStatic(selection, current, allocator),
+ reversed,
+ current.stats)
+ {
+ protected Unfiltered computeNext()
+ {
+ return endOfData();
+ }
+ };
+ }
+
+ return slices.size() == 1
+ ? new SingleSliceIterator(metadata, partitionKey, ref, selection, slices.get(0), reversed, allocator)
+ : new SlicesIterator(metadata, partitionKey, ref, selection, slices, reversed, allocator);
+ }
+
+ private static Row makeStatic(ColumnFilter selection, Holder holder, MemtableAllocator allocator)
+ {
+ Columns statics = selection.fetchedColumns().statics;
+ if (statics.isEmpty() || holder.staticRow == null)
+ return Rows.EMPTY_STATIC_ROW;
+
+ return new ReusableFilteringRow(statics, selection)
+ .setDeletionTimestamp(holder.deletionInfo.getPartitionDeletion().markedForDeleteAt())
+ .setTo(allocator.newReusableRow().setTo(holder.staticRow));
+ }
+
+ private static class ReusableFilteringRow extends FilteringRow
+ {
+ private final Columns columns;
+ private final ColumnFilter selection;
+ private ColumnFilter.Tester tester;
+ private long deletionTimestamp;
+
+ // Used by searchIterator in case the row is covered by a tombstone.
+ private DeletionTime rowDeletion;
+
+ public ReusableFilteringRow(Columns columns, ColumnFilter selection)
+ {
+ this.columns = columns;
+ this.selection = selection;
+ }
+
+ public ReusableFilteringRow setDeletionTimestamp(long timestamp)
+ {
+ this.deletionTimestamp = timestamp;
+ return this;
+ }
+
+ public ReusableFilteringRow setRowDeletion(DeletionTime rowDeletion)
+ {
+ this.rowDeletion = rowDeletion;
+ return this;
+ }
+
+ @Override
+ public DeletionTime deletion()
+ {
+ return rowDeletion == null ? super.deletion() : rowDeletion;
+ }
+
+ @Override
+ protected boolean include(LivenessInfo info)
+ {
+ return info.timestamp() > deletionTimestamp;
+ }
+
+ @Override
+ protected boolean include(ColumnDefinition def)
+ {
+ return columns.contains(def);
+ }
+
+ @Override
+ protected boolean include(DeletionTime dt)
+ {
+ return dt.markedForDeleteAt() > deletionTimestamp;
+ }
+
+ @Override
+ protected boolean include(ColumnDefinition c, DeletionTime dt)
+ {
+ return dt.markedForDeleteAt() > deletionTimestamp;
+ }
+
+ @Override
+ protected boolean include(Cell cell)
+ {
+ return selection.includes(cell);
+ }
+ }
+
+ private static class SingleSliceIterator extends AbstractUnfilteredRowIterator
+ {
+ private final Iterator<Unfiltered> iterator;
+ private final ReusableFilteringRow row;
+
+ private SingleSliceIterator(CFMetaData metadata,
+ DecoratedKey key,
+ Holder holder,
+ ColumnFilter selection,
+ Slice slice,
+ boolean isReversed,
+ MemtableAllocator allocator)
+ {
+ super(metadata,
+ key,
+ holder.deletionInfo.getPartitionDeletion(),
+ selection.fetchedColumns(),
+ makeStatic(selection, holder, allocator),
+ isReversed,
+ holder.stats);
+
+ Iterator<Row> rowIter = rowIter(metadata,
+ holder,
+ slice,
+ !isReversed,
+ allocator);
+
+ this.iterator = new RowAndTombstoneMergeIterator(metadata.comparator, isReversed)
+ .setTo(rowIter, holder.deletionInfo.rangeIterator(slice, isReversed));
+
+ this.row = new ReusableFilteringRow(selection.fetchedColumns().regulars, selection)
+ .setDeletionTimestamp(partitionLevelDeletion.markedForDeleteAt());
+ }
+
+ private Iterator<Row> rowIter(CFMetaData metadata,
+ Holder holder,
+ Slice slice,
+ boolean forwards,
+ final MemtableAllocator allocator)
+ {
+ Slice.Bound start = slice.start() == Slice.Bound.BOTTOM ? null : slice.start();
+ Slice.Bound end = slice.end() == Slice.Bound.TOP ? null : slice.end();
+ final Iterator<MemtableRowData> dataIter = BTree.slice(holder.tree, metadata.comparator, start, true, end, true, forwards);
+ return new AbstractIterator<Row>()
+ {
+ private final MemtableRowData.ReusableRow row = allocator.newReusableRow();
+
+ protected Row computeNext()
+ {
+ return dataIter.hasNext() ? row.setTo(dataIter.next()) : endOfData();
+ }
+ };
+ }
+
+ protected Unfiltered computeNext()
+ {
+ while (iterator.hasNext())
+ {
+ Unfiltered next = iterator.next();
+ if (next.kind() == Unfiltered.Kind.ROW)
+ {
+ row.setTo((Row)next);
+ if (!row.isEmpty())
+ return row;
+ }
+ else
+ {
+ RangeTombstoneMarker marker = (RangeTombstoneMarker)next;
+
+ long deletion = partitionLevelDeletion().markedForDeleteAt();
+ if (marker.isOpen(isReverseOrder()))
+ deletion = Math.max(deletion, marker.openDeletionTime(isReverseOrder()).markedForDeleteAt());
+ row.setDeletionTimestamp(deletion);
+ return marker;
+ }
+ }
+ return endOfData();
+ }
+ }
+
+ public static class SlicesIterator extends AbstractUnfilteredRowIterator
+ {
+ private final Holder holder;
+ private final MemtableAllocator allocator;
+ private final ColumnFilter selection;
+ private final Slices slices;
+
+ private int idx;
+ private UnfilteredRowIterator currentSlice;
+
+ private SlicesIterator(CFMetaData metadata,
+ DecoratedKey key,
+ Holder holder,
+ ColumnFilter selection,
+ Slices slices,
+ boolean isReversed,
+ MemtableAllocator allocator)
+ {
+ super(metadata, key, holder.deletionInfo.getPartitionDeletion(), selection.fetchedColumns(), makeStatic(selection, holder, allocator), isReversed, holder.stats);
+ this.holder = holder;
+ this.selection = selection;
+ this.allocator = allocator;
+ this.slices = slices;
+ }
+
+ protected Unfiltered computeNext()
+ {
+ while (true)
+ {
+ if (currentSlice == null)
+ {
+ if (idx >= slices.size())
+ return endOfData();
+
+ int sliceIdx = isReverseOrder ? slices.size() - idx - 1 : idx;
+ currentSlice = new SingleSliceIterator(metadata,
+ partitionKey,
+ holder,
+ selection,
+ slices.get(sliceIdx),
+ isReverseOrder,
+ allocator);
+ idx++;
+ }
+
+ if (currentSlice.hasNext())
+ return currentSlice.next();
+
+ currentSlice = null;
+ }
+ }
+ }
+
+ /**
+ * Adds a given update to this in-memtable partition.
+ *
+ * @return an array containing first the difference in size seen after merging the updates, and second the minimum
+ * time detla between updates.
+ */
+ public long[] addAllWithSizeDelta(final PartitionUpdate update, OpOrder.Group writeOp, Updater indexer)
+ {
+ RowUpdater updater = new RowUpdater(this, allocator, writeOp, indexer);
+ DeletionInfo inputDeletionInfoCopy = null;
+
+ boolean monitorOwned = false;
+ try
+ {
+ if (usePessimisticLocking())
+ {
+ Locks.monitorEnterUnsafe(this);
+ monitorOwned = true;
+ }
+ while (true)
+ {
+ Holder current = ref;
+ updater.ref = current;
+ updater.reset();
+
+ DeletionInfo deletionInfo;
+ if (update.deletionInfo().mayModify(current.deletionInfo))
+ {
+ if (inputDeletionInfoCopy == null)
+ inputDeletionInfoCopy = update.deletionInfo().copy(HeapAllocator.instance);
+
+ deletionInfo = current.deletionInfo.copy().add(inputDeletionInfoCopy);
+ updater.allocated(deletionInfo.unsharedHeapSize() - current.deletionInfo.unsharedHeapSize());
+ }
+ else
+ {
+ deletionInfo = current.deletionInfo;
+ }
+
+ Row newStatic = update.staticRow();
+ MemtableRowData staticRow = newStatic == Rows.EMPTY_STATIC_ROW
+ ? current.staticRow
+ : (current.staticRow == null ? updater.apply(newStatic) : updater.apply(current.staticRow, newStatic));
+ Object[] tree = BTree.<Clusterable, Row, MemtableRowData>update(current.tree, update.metadata().comparator, update, update.rowCount(), updater);
+ RowStats newStats = current.stats.mergeWith(update.stats());
+
+ if (tree != null && refUpdater.compareAndSet(this, current, new Holder(tree, deletionInfo, staticRow, newStats)))
+ {
+ indexer.updateRowLevelIndexes();
+ updater.finish();
+ return new long[]{ updater.dataSize, updater.colUpdateTimeDelta };
+ }
+ else if (!monitorOwned)
+ {
+ boolean shouldLock = usePessimisticLocking();
+ if (!shouldLock)
+ {
+ shouldLock = updateWastedAllocationTracker(updater.heapSize);
+ }
+ if (shouldLock)
+ {
+ Locks.monitorEnterUnsafe(this);
+ monitorOwned = true;
+ }
+ }
+ }
+ }
+ finally
+ {
+ if (monitorOwned)
+ Locks.monitorExitUnsafe(this);
+ }
+
+ }
+
+ public boolean usePessimisticLocking()
+ {
+ return wasteTracker == TRACKER_PESSIMISTIC_LOCKING;
+ }
+
+ /**
+ * Update the wasted allocation tracker state based on newly wasted allocation information
+ *
+ * @param wastedBytes the number of bytes wasted by this thread
+ * @return true if the caller should now proceed with pessimistic locking because the waste limit has been reached
+ */
+ private boolean updateWastedAllocationTracker(long wastedBytes)
+ {
+ // Early check for huge allocation that exceeds the limit
+ if (wastedBytes < EXCESS_WASTE_BYTES)
+ {
+ // We round up to ensure work < granularity are still accounted for
+ int wastedAllocation = ((int) (wastedBytes + ALLOCATION_GRANULARITY_BYTES - 1)) / ALLOCATION_GRANULARITY_BYTES;
+
+ int oldTrackerValue;
+ while (TRACKER_PESSIMISTIC_LOCKING != (oldTrackerValue = wasteTracker))
+ {
+ // Note this time value has an arbitrary offset, but is a constant rate 32 bit counter (that may wrap)
+ int time = (int) (System.nanoTime() >>> CLOCK_SHIFT);
+ int delta = oldTrackerValue - time;
+ if (oldTrackerValue == TRACKER_NEVER_WASTED || delta >= 0 || delta < -EXCESS_WASTE_OFFSET)
+ delta = -EXCESS_WASTE_OFFSET;
+ delta += wastedAllocation;
+ if (delta >= 0)
+ break;
+ if (wasteTrackerUpdater.compareAndSet(this, oldTrackerValue, avoidReservedValues(time + delta)))
+ return false;
+ }
+ }
+ // We have definitely reached our waste limit so set the state if it isn't already
+ wasteTrackerUpdater.set(this, TRACKER_PESSIMISTIC_LOCKING);
+ // And tell the caller to proceed with pessimistic locking
+ return true;
+ }
+
+ private static int avoidReservedValues(int wasteTracker)
+ {
+ if (wasteTracker == TRACKER_NEVER_WASTED || wasteTracker == TRACKER_PESSIMISTIC_LOCKING)
+ return wasteTracker + 1;
+ return wasteTracker;
+ }
+
+ private static final class Holder
+ {
+ final DeletionInfo deletionInfo;
+ // the btree of rows
+ final Object[] tree;
+ final MemtableRowData staticRow;
+ final RowStats stats;
+
+ Holder(Object[] tree, DeletionInfo deletionInfo, MemtableRowData staticRow, RowStats stats)
+ {
+ this.tree = tree;
+ this.deletionInfo = deletionInfo;
+ this.staticRow = staticRow;
+ this.stats = stats;
+ }
+
+ Holder with(DeletionInfo info)
+ {
+ return new Holder(this.tree, info, this.staticRow, this.stats);
+ }
+ }
+
+ // the function we provide to the btree utilities to perform any column replacements
+ private static final class RowUpdater implements UpdateFunction<Row, MemtableRowData>
+ {
+ final AtomicBTreePartition updating;
+ final MemtableAllocator allocator;
+ final OpOrder.Group writeOp;
+ final Updater indexer;
+ final int nowInSec;
+ Holder ref;
+ long dataSize;
+ long heapSize;
+ long colUpdateTimeDelta = Long.MAX_VALUE;
+ final MemtableRowData.ReusableRow row;
+ final MemtableAllocator.DataReclaimer reclaimer;
+ final MemtableAllocator.RowAllocator rowAllocator;
+ List<MemtableRowData> inserted; // TODO: replace with walk of aborted BTree
+
+ private RowUpdater(AtomicBTreePartition updating, MemtableAllocator allocator, OpOrder.Group writeOp, Updater indexer)
+ {
+ this.updating = updating;
+ this.allocator = allocator;
+ this.writeOp = writeOp;
+ this.indexer = indexer;
+ this.nowInSec = FBUtilities.nowInSeconds();
+ this.row = allocator.newReusableRow();
+ this.reclaimer = allocator.reclaimer();
+ this.rowAllocator = allocator.newRowAllocator(updating.metadata(), writeOp);
+ }
+
+ public MemtableRowData apply(Row insert)
+ {
+ rowAllocator.allocateNewRow(insert.clustering().size(), insert.columns(), insert.isStatic());
+ insert.copyTo(rowAllocator);
+ MemtableRowData data = rowAllocator.allocatedRowData();
+
+ insertIntoIndexes(insert);
+
+ this.dataSize += data.dataSize();
+ this.heapSize += data.unsharedHeapSizeExcludingData();
+ if (inserted == null)
+ inserted = new ArrayList<>();
+ inserted.add(data);
+ return data;
+ }
+
+ public MemtableRowData apply(MemtableRowData existing, Row update)
+ {
+ Columns mergedColumns = existing.columns().mergeTo(update.columns());
+ rowAllocator.allocateNewRow(update.clustering().size(), mergedColumns, update.isStatic());
+
+ colUpdateTimeDelta = Math.min(colUpdateTimeDelta, Rows.merge(row.setTo(existing), update, mergedColumns, rowAllocator, nowInSec, indexer));
+
+ MemtableRowData reconciled = rowAllocator.allocatedRowData();
+
+ dataSize += reconciled.dataSize() - existing.dataSize();
+ heapSize += reconciled.unsharedHeapSizeExcludingData() - existing.unsharedHeapSizeExcludingData();
+ if (inserted == null)
+ inserted = new ArrayList<>();
+ inserted.add(reconciled);
+ discard(existing);
+
+ return reconciled;
+ }
+
+ private void insertIntoIndexes(Row toInsert)
+ {
+ if (indexer == SecondaryIndexManager.nullUpdater)
+ return;
+
+ maybeIndexPrimaryKeyColumns(toInsert);
+ Clustering clustering = toInsert.clustering();
+ for (Cell cell : toInsert)
+ indexer.insert(clustering, cell);
+ }
+
+ private void maybeIndexPrimaryKeyColumns(Row row)
+ {
+ // We want to update a primary key index with the most up to date info contains in that inserted row (if only for
+ // backward compatibility). Note that if there is an index but not a partition key or clustering column one, we've
+ // wasting this work. We might be able to avoid that if row indexing was pushed in the index updater.
+ long timestamp = row.primaryKeyLivenessInfo().timestamp();
+ int ttl = row.primaryKeyLivenessInfo().ttl();
+
+ for (Cell cell : row)
+ {
+ long cellTimestamp = cell.livenessInfo().timestamp();
+ if (cell.isLive(nowInSec))
+ {
+ if (cellTimestamp > timestamp)
+ {
+ timestamp = cellTimestamp;
+ ttl = cell.livenessInfo().ttl();
+ }
+ }
+ }
+
+ indexer.maybeIndex(row.clustering(), timestamp, ttl, row.deletion());
+ }
+
+ protected void reset()
+ {
+ this.dataSize = 0;
+ this.heapSize = 0;
+ if (inserted != null)
+ {
+ for (MemtableRowData row : inserted)
+ abort(row);
+ inserted.clear();
+ }
+ reclaimer.cancel();
+ }
+
+ protected void abort(MemtableRowData abort)
+ {
+ reclaimer.reclaimImmediately(abort);
+ }
+
+ protected void discard(MemtableRowData discard)
+ {
+ reclaimer.reclaim(discard);
+ }
+
+ public boolean abortEarly()
+ {
+ return updating.ref != ref;
+ }
+
+ public void allocated(long heapSize)
+ {
+ this.heapSize += heapSize;
+ }
+
+ protected void finish()
+ {
- allocator.onHeap().allocate(heapSize, writeOp);
++ allocator.onHeap().adjust(heapSize, writeOp);
+ reclaimer.commit();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dea6ab1b/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dea6ab1b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/dea6ab1b/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
----------------------------------------------------------------------
[3/6] cassandra git commit: Ensure memtable book keeping is not
corrupted in the event we shrink usage
Posted by be...@apache.org.
Ensure memtable book keeping is not corrupted in the event we shrink usage
patch by benedict; reviewed by tjake for CASSANDRA-9681
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b757db14
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b757db14
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b757db14
Branch: refs/heads/trunk
Commit: b757db1484473b264bf25ca5541f080d54a579a2
Parents: c5f03a9
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Jul 2 10:27:07 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Jul 2 10:27:07 2015 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/AtomicBTreeColumns.java | 2 +-
.../apache/cassandra/db/ColumnFamilyStore.java | 22 +++++-
src/java/org/apache/cassandra/db/Memtable.java | 15 ++--
.../org/apache/cassandra/utils/FBUtilities.java | 10 +++
.../apache/cassandra/utils/memory/HeapPool.java | 4 +-
.../utils/memory/MemtableAllocator.java | 39 +++++++----
.../cassandra/utils/memory/MemtablePool.java | 73 ++++++++++++--------
.../utils/memory/NativeAllocatorTest.java | 18 ++++-
9 files changed, 132 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 762b88b..25f7c1d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.8
+ * Ensure memtable book keeping is not corrupted in the event we shrink usage (CASSANDRA-9681)
* Update internal python driver for cqlsh (CASSANDRA-9064)
* Fix IndexOutOfBoundsException when inserting tuple with too many
elements using the string literal notation (CASSANDRA-9559)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
index 47f0b85..d9eb29c 100644
--- a/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
+++ b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
@@ -505,7 +505,7 @@ public class AtomicBTreeColumns extends ColumnFamily
protected void finish()
{
- allocator.onHeap().allocate(heapSize, writeOp);
+ allocator.onHeap().adjust(heapSize, writeOp);
reclaimer.commit();
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index fa527c7..8e67cdc 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -35,6 +35,7 @@ import com.google.common.collect.*;
import com.google.common.util.concurrent.*;
import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.utils.memory.MemtablePool;
import org.json.simple.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1157,6 +1158,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
float largestRatio = 0f;
Memtable largest = null;
+ float liveOnHeap = 0, liveOffHeap = 0;
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
// we take a reference to the current main memtable for the CF prior to snapping its ownership ratios
@@ -1181,19 +1183,37 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
float ratio = Math.max(onHeap, offHeap);
-
if (ratio > largestRatio)
{
largest = current;
largestRatio = ratio;
}
+
+ liveOnHeap += onHeap;
+ liveOffHeap += offHeap;
}
if (largest != null)
+ {
+ float usedOnHeap = Memtable.MEMORY_POOL.onHeap.usedRatio();
+ float usedOffHeap = Memtable.MEMORY_POOL.offHeap.usedRatio();
+ float flushingOnHeap = Memtable.MEMORY_POOL.onHeap.reclaimingRatio();
+ float flushingOffHeap = Memtable.MEMORY_POOL.offHeap.reclaimingRatio();
+ float thisOnHeap = largest.getAllocator().onHeap().ownershipRatio();
+ float thisOffHeap = largest.getAllocator().onHeap().ownershipRatio();
+ logger.info("Flushing largest {} to free up room. Used total: {}, live: {}, flushing: {}, this: {}",
+ largest.cfs, ratio(usedOnHeap, usedOffHeap), ratio(liveOnHeap, liveOffHeap),
+ ratio(flushingOnHeap, flushingOffHeap), ratio(thisOnHeap, thisOffHeap));
largest.cfs.switchMemtableIfCurrent(largest);
+ }
}
}
+ private static String ratio(float onHeap, float offHeap)
+ {
+ return String.format("%.0f/%.0f", onHeap, offHeap);
+ }
+
public void maybeUpdateRowCache(DecoratedKey key)
{
if (!isRowCacheEnabled())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/db/Memtable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java
index a50a614..9f6cf9b 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -239,7 +239,8 @@ public class Memtable
public String toString()
{
return String.format("Memtable-%s@%s(%s serialized bytes, %s ops, %.0f%%/%.0f%% of on/off-heap limit)",
- cfs.name, hashCode(), liveDataSize, currentOperations, 100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
+ cfs.name, hashCode(), FBUtilities.prettyPrintMemory(liveDataSize.get()), currentOperations,
+ 100 * allocator.onHeap().ownershipRatio(), 100 * allocator.offHeap().ownershipRatio());
}
/**
@@ -378,19 +379,21 @@ public class Memtable
if (writer.getFilePointer() > 0)
{
- writer.isolateReferences();
+ logger.info(String.format("Completed flushing %s (%s) for commitlog position %s",
+ writer.getFilename(),
+ FBUtilities.prettyPrintMemory(writer.getOnDiskFilePointer()),
+ context));
+ writer.isolateReferences();
// temp sstables should contain non-repaired data.
ssTable = writer.closeAndOpenReader();
- logger.info(String.format("Completed flushing %s (%d bytes) for commitlog position %s",
- ssTable.getFilename(), new File(ssTable.getFilename()).length(), context));
}
else
{
+ logger.info("Completed flushing %s; nothing needed to be retained. Commitlog position was {}",
+ writer.getFilename(), context);
writer.abort();
ssTable = null;
- logger.info("Completed flushing; nothing needed to be retained. Commitlog position was {}",
- context);
}
if (heavilyContendedRowCount > 0)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index 4c81b2a..68eb864 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -28,6 +28,7 @@ import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+import java.text.NumberFormat;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -570,6 +571,15 @@ public class FBUtilities
}
}
+ public static String prettyPrintMemory(long size)
+ {
+ if (size >= 1 << 30)
+ return String.format("%.3fGiB", size / (double) (1 << 30));
+ if (size >= 1 << 20)
+ return String.format("%.3fMiB", size / (double) (1 << 20));
+ return String.format("%.3fKiB", size / (double) (1 << 10));
+ }
+
/**
* Starts and waits for the given @param pb to finish.
* @throws java.io.IOException on non-zero exit code
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/HeapPool.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/HeapPool.java b/src/java/org/apache/cassandra/utils/memory/HeapPool.java
index a04947c..2a19d9c 100644
--- a/src/java/org/apache/cassandra/utils/memory/HeapPool.java
+++ b/src/java/org/apache/cassandra/utils/memory/HeapPool.java
@@ -75,13 +75,13 @@ public class HeapPool extends MemtablePool
public Reclaimer reclaimImmediately(Cell cell)
{
- onHeap().release(cell.name().dataSize() + cell.value().remaining());
+ onHeap().released(cell.name().dataSize() + cell.value().remaining());
return this;
}
public Reclaimer reclaimImmediately(DecoratedKey key)
{
- onHeap().release(key.getKey().remaining());
+ onHeap().released(key.getKey().remaining());
return this;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
index e814b4d..f5e743c 100644
--- a/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
+++ b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java
@@ -160,13 +160,24 @@ public abstract class MemtableAllocator
// currently no corroboration/enforcement of this is performed.
void releaseAll()
{
- parent.adjustAcquired(-ownsUpdater.getAndSet(this, 0), false);
- parent.adjustReclaiming(-reclaimingUpdater.getAndSet(this, 0));
+ parent.released(ownsUpdater.getAndSet(this, 0));
+ parent.reclaimed(reclaimingUpdater.getAndSet(this, 0));
+ }
+
+ // like allocate, but permits allocations to be negative
+ public void adjust(long size, OpOrder.Group opGroup)
+ {
+ if (size <= 0)
+ released(-size);
+ else
+ allocate(size, opGroup);
}
// allocate memory in the tracker, and mark ourselves as owning it
public void allocate(long size, OpOrder.Group opGroup)
{
+ assert size >= 0;
+
while (true)
{
if (parent.tryAllocate(size))
@@ -190,23 +201,23 @@ public abstract class MemtableAllocator
}
}
- // retroactively mark an amount allocated amd acquired in the tracker, and owned by us
- void allocated(long size)
+ // retroactively mark an amount allocated and acquired in the tracker, and owned by us
+ private void allocated(long size)
{
- parent.adjustAcquired(size, true);
+ parent.allocated(size);
ownsUpdater.addAndGet(this, size);
}
// retroactively mark an amount acquired in the tracker, and owned by us
- void acquired(long size)
+ private void acquired(long size)
{
- parent.adjustAcquired(size, false);
+ parent.acquired(size);
ownsUpdater.addAndGet(this, size);
}
- void release(long size)
+ void released(long size)
{
- parent.adjustAcquired(-size, false);
+ parent.released(size);
ownsUpdater.addAndGet(this, -size);
}
@@ -217,11 +228,11 @@ public abstract class MemtableAllocator
{
long cur = owns;
long prev = reclaiming;
- if (reclaimingUpdater.compareAndSet(this, prev, cur))
- {
- parent.adjustReclaiming(cur - prev);
- return;
- }
+ if (!reclaimingUpdater.compareAndSet(this, prev, cur))
+ continue;
+
+ parent.reclaiming(cur - prev);
+ return;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
index 1d219bb..bb85884 100644
--- a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
+++ b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java
@@ -130,10 +130,8 @@ public abstract class MemtablePool
* apply the size adjustment to allocated, bypassing any limits or constraints. If this reduces the
* allocated total, we will signal waiters
*/
- void adjustAllocated(long size)
+ private void adjustAllocated(long size)
{
- if (size == 0)
- return;
while (true)
{
long cur = allocated;
@@ -142,38 +140,43 @@ public abstract class MemtablePool
}
}
- // 'acquires' an amount of memory, and maybe also marks it allocated. This method is meant to be overridden
- // by implementations with a separate concept of acquired/allocated. As this method stands, an acquire
- // without an allocate is a no-op (acquisition is achieved through allocation), however a release (where size < 0)
- // is always processed and accounted for in allocated.
- void adjustAcquired(long size, boolean alsoAllocated)
+ void allocated(long size)
{
- if (size > 0 || alsoAllocated)
- {
- if (alsoAllocated)
- adjustAllocated(size);
- maybeClean();
- }
- else if (size < 0)
- {
- adjustAllocated(size);
- hasRoom.signalAll();
- }
+ assert size >= 0;
+ if (size == 0)
+ return;
+
+ adjustAllocated(size);
+ maybeClean();
+ }
+
+ void acquired(long size)
+ {
+ maybeClean();
+ }
+
+ void released(long size)
+ {
+ assert size >= 0;
+ adjustAllocated(-size);
+ hasRoom.signalAll();
}
- // space reclaimed should be released prior to calling this, to avoid triggering unnecessary cleans
- void adjustReclaiming(long reclaiming)
+ void reclaiming(long size)
{
- if (reclaiming == 0)
+ if (size == 0)
return;
- reclaimingUpdater.addAndGet(this, reclaiming);
- if (reclaiming < 0 && updateNextClean() && cleaner != null)
- cleaner.trigger();
+ reclaimingUpdater.addAndGet(this, size);
}
- public long allocated()
+ void reclaimed(long size)
{
- return allocated;
+ if (size == 0)
+ return;
+
+ reclaimingUpdater.addAndGet(this, -size);
+ if (updateNextClean() && cleaner != null)
+ cleaner.trigger();
}
public long used()
@@ -181,6 +184,22 @@ public abstract class MemtablePool
return allocated;
}
+ public float reclaimingRatio()
+ {
+ float r = reclaiming / (float) limit;
+ if (Float.isNaN(r))
+ return 0;
+ return r;
+ }
+
+ public float usedRatio()
+ {
+ float r = allocated / (float) limit;
+ if (Float.isNaN(r))
+ return 0;
+ return r;
+ }
+
public MemtableAllocator.SubAllocator newAllocator()
{
return new MemtableAllocator.SubAllocator(this);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b757db14/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java b/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
index 83d6c0c..7f87fcd 100644
--- a/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
+++ b/test/unit/org/apache/cassandra/utils/memory/NativeAllocatorTest.java
@@ -56,7 +56,7 @@ public class NativeAllocatorTest
}
if (isClean.getCount() > 0)
{
- allocatorRef.get().offHeap().release(80);
+ allocatorRef.get().offHeap().released(80);
isClean.countDown();
}
}
@@ -79,6 +79,22 @@ public class NativeAllocatorTest
// allocate normal, check accounted and not cleaned
allocator.allocate(10, group);
Assert.assertEquals(10, allocator.offHeap().owns());
+ // confirm adjustment works
+ allocator.offHeap().adjust(-10, group);
+ Assert.assertEquals(0, allocator.offHeap().owns());
+ allocator.offHeap().adjust(10, group);
+ Assert.assertEquals(10, allocator.offHeap().owns());
+ // confirm we cannot allocate negative
+ boolean success = false;
+ try
+ {
+ allocator.offHeap().allocate(-10, group);
+ }
+ catch (AssertionError e)
+ {
+ success = true;
+ }
+ Assert.assertTrue(success);
Uninterruptibles.sleepUninterruptibly(10L, TimeUnit.MILLISECONDS);
Assert.assertEquals(1, isClean.getCount());
[4/6] cassandra git commit: Merge branch 'cassandra-2.1' into
cassandra-2.2
Posted by be...@apache.org.
Merge branch 'cassandra-2.1' into cassandra-2.2
Conflicts:
CHANGES.txt
src/java/org/apache/cassandra/db/Memtable.java
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/99f7ce9b
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/99f7ce9b
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/99f7ce9b
Branch: refs/heads/trunk
Commit: 99f7ce9bfb03ad5eda21d3604b3844fc193d0f6f
Parents: 92e2e4e b757db1
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Jul 2 10:33:51 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Jul 2 10:33:51 2015 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/AtomicBTreeColumns.java | 2 +-
.../apache/cassandra/db/ColumnFamilyStore.java | 22 +++++-
src/java/org/apache/cassandra/db/Memtable.java | 15 ++--
.../org/apache/cassandra/utils/FBUtilities.java | 10 +++
.../apache/cassandra/utils/memory/HeapPool.java | 4 +-
.../utils/memory/MemtableAllocator.java | 39 +++++++----
.../cassandra/utils/memory/MemtablePool.java | 73 ++++++++++++--------
.../utils/memory/NativeAllocatorTest.java | 18 ++++-
9 files changed, 133 insertions(+), 51 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 720133a,25f7c1d..a282fd7
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,23 -1,5 +1,24 @@@
-2.1.8
+2.2.0-rc2
+ * (cqlsh) Allow setting the initial connection timeout (CASSANDRA-9601)
+ * BulkLoader has --transport-factory option but does not use it (CASSANDRA-9675)
+ * Allow JMX over SSL directly from nodetool (CASSANDRA-9090)
+ * Update cqlsh for UDFs (CASSANDRA-7556)
+ * Change Windows kernel default timer resolution (CASSANDRA-9634)
+ * Deprected sstable2json and json2sstable (CASSANDRA-9618)
+ * Allow native functions in user-defined aggregates (CASSANDRA-9542)
+ * Don't repair system_distributed by default (CASSANDRA-9621)
+ * Fix mixing min, max, and count aggregates for blob type (CASSANRA-9622)
+ * Rename class for DATE type in Java driver (CASSANDRA-9563)
+ * Duplicate compilation of UDFs on coordinator (CASSANDRA-9475)
+ * Fix connection leak in CqlRecordWriter (CASSANDRA-9576)
+ * Mlockall before opening system sstables & remove boot_without_jna option (CASSANDRA-9573)
+ * Add functions to convert timeuuid to date or time, deprecate dateOf and unixTimestampOf (CASSANDRA-9229)
+ * Make sure we cancel non-compacting sstables from LifecycleTransaction (CASSANDRA-9566)
+ * Fix deprecated repair JMX API (CASSANDRA-9570)
+ * Add logback metrics (CASSANDRA-9378)
+ * Update and refactor ant test/test-compression to run the tests in parallel (CASSANDRA-9583)
+Merged from 2.1:
+ * Ensure memtable book keeping is not corrupted in the event we shrink usage (CASSANDRA-9681)
* Update internal python driver for cqlsh (CASSANDRA-9064)
* Fix IndexOutOfBoundsException when inserting tuple with too many
elements using the string literal notation (CASSANDRA-9559)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 926cba2,8e67cdc..1374071
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@@ -34,11 -34,8 +34,12 @@@ import com.google.common.base.Throwable
import com.google.common.collect.*;
import com.google.common.util.concurrent.*;
+import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
+import org.apache.cassandra.db.lifecycle.View;
+import org.apache.cassandra.db.lifecycle.Tracker;
+import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.io.FSWriteError;
+ import org.apache.cassandra.utils.memory.MemtablePool;
import org.json.simple.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/db/Memtable.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/Memtable.java
index ccf92be,9f6cf9b..6e4802f
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@@ -393,10 -379,14 +394,13 @@@ public class Memtable implements Compar
if (writer.getFilePointer() > 0)
{
+ logger.info(String.format("Completed flushing %s (%s) for commitlog position %s",
+ writer.getFilename(),
+ FBUtilities.prettyPrintMemory(writer.getOnDiskFilePointer()),
+ context));
+
- writer.isolateReferences();
// temp sstables should contain non-repaired data.
- ssTable = writer.closeAndOpenReader();
+ ssTable = writer.finish(true);
- logger.info(String.format("Completed flushing %s (%d bytes) for commitlog position %s",
- ssTable.getFilename(), new File(ssTable.getFilename()).length(), context));
}
else
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/utils/FBUtilities.java
index 17edeb0,68eb864..df65572
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@@ -26,9 -28,12 +26,10 @@@ import java.net.*
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+ import java.text.NumberFormat;
import java.util.*;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
+import java.util.zip.Adler32;
import java.util.zip.Checksum;
import com.google.common.base.Joiner;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/utils/memory/HeapPool.java
----------------------------------------------------------------------
[5/6] cassandra git commit: Merge branch 'cassandra-2.1' into
cassandra-2.2
Posted by be...@apache.org.
Merge branch 'cassandra-2.1' into cassandra-2.2
Conflicts:
CHANGES.txt
src/java/org/apache/cassandra/db/Memtable.java
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/99f7ce9b
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/99f7ce9b
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/99f7ce9b
Branch: refs/heads/cassandra-2.2
Commit: 99f7ce9bfb03ad5eda21d3604b3844fc193d0f6f
Parents: 92e2e4e b757db1
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Jul 2 10:33:51 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Jul 2 10:33:51 2015 +0100
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/AtomicBTreeColumns.java | 2 +-
.../apache/cassandra/db/ColumnFamilyStore.java | 22 +++++-
src/java/org/apache/cassandra/db/Memtable.java | 15 ++--
.../org/apache/cassandra/utils/FBUtilities.java | 10 +++
.../apache/cassandra/utils/memory/HeapPool.java | 4 +-
.../utils/memory/MemtableAllocator.java | 39 +++++++----
.../cassandra/utils/memory/MemtablePool.java | 73 ++++++++++++--------
.../utils/memory/NativeAllocatorTest.java | 18 ++++-
9 files changed, 133 insertions(+), 51 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 720133a,25f7c1d..a282fd7
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,23 -1,5 +1,24 @@@
-2.1.8
+2.2.0-rc2
+ * (cqlsh) Allow setting the initial connection timeout (CASSANDRA-9601)
+ * BulkLoader has --transport-factory option but does not use it (CASSANDRA-9675)
+ * Allow JMX over SSL directly from nodetool (CASSANDRA-9090)
+ * Update cqlsh for UDFs (CASSANDRA-7556)
+ * Change Windows kernel default timer resolution (CASSANDRA-9634)
+ * Deprected sstable2json and json2sstable (CASSANDRA-9618)
+ * Allow native functions in user-defined aggregates (CASSANDRA-9542)
+ * Don't repair system_distributed by default (CASSANDRA-9621)
+ * Fix mixing min, max, and count aggregates for blob type (CASSANRA-9622)
+ * Rename class for DATE type in Java driver (CASSANDRA-9563)
+ * Duplicate compilation of UDFs on coordinator (CASSANDRA-9475)
+ * Fix connection leak in CqlRecordWriter (CASSANDRA-9576)
+ * Mlockall before opening system sstables & remove boot_without_jna option (CASSANDRA-9573)
+ * Add functions to convert timeuuid to date or time, deprecate dateOf and unixTimestampOf (CASSANDRA-9229)
+ * Make sure we cancel non-compacting sstables from LifecycleTransaction (CASSANDRA-9566)
+ * Fix deprecated repair JMX API (CASSANDRA-9570)
+ * Add logback metrics (CASSANDRA-9378)
+ * Update and refactor ant test/test-compression to run the tests in parallel (CASSANDRA-9583)
+Merged from 2.1:
+ * Ensure memtable book keeping is not corrupted in the event we shrink usage (CASSANDRA-9681)
* Update internal python driver for cqlsh (CASSANDRA-9064)
* Fix IndexOutOfBoundsException when inserting tuple with too many
elements using the string literal notation (CASSANDRA-9559)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 926cba2,8e67cdc..1374071
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@@ -34,11 -34,8 +34,12 @@@ import com.google.common.base.Throwable
import com.google.common.collect.*;
import com.google.common.util.concurrent.*;
+import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
+import org.apache.cassandra.db.lifecycle.View;
+import org.apache.cassandra.db.lifecycle.Tracker;
+import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.io.FSWriteError;
+ import org.apache.cassandra.utils.memory.MemtablePool;
import org.json.simple.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/db/Memtable.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/Memtable.java
index ccf92be,9f6cf9b..6e4802f
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@@ -393,10 -379,14 +394,13 @@@ public class Memtable implements Compar
if (writer.getFilePointer() > 0)
{
+ logger.info(String.format("Completed flushing %s (%s) for commitlog position %s",
+ writer.getFilename(),
+ FBUtilities.prettyPrintMemory(writer.getOnDiskFilePointer()),
+ context));
+
- writer.isolateReferences();
// temp sstables should contain non-repaired data.
- ssTable = writer.closeAndOpenReader();
+ ssTable = writer.finish(true);
- logger.info(String.format("Completed flushing %s (%d bytes) for commitlog position %s",
- ssTable.getFilename(), new File(ssTable.getFilename()).length(), context));
}
else
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/utils/FBUtilities.java
index 17edeb0,68eb864..df65572
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@@ -26,9 -28,12 +26,10 @@@ import java.net.*
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+ import java.text.NumberFormat;
import java.util.*;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
+import java.util.zip.Adler32;
import java.util.zip.Checksum;
import com.google.common.base.Joiner;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/99f7ce9b/src/java/org/apache/cassandra/utils/memory/HeapPool.java
----------------------------------------------------------------------