diff --git a/build.xml b/build.xml index 2965a4b263..8df98df1ab 100644 --- a/build.xml +++ b/build.xml @@ -361,7 +361,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} - + @@ -948,7 +948,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} - + diff --git a/conf/cassandra-env.sh b/conf/cassandra-env.sh index 8fd1fa9a62..803f521f28 100644 --- a/conf/cassandra-env.sh +++ b/conf/cassandra-env.sh @@ -100,7 +100,7 @@ JVM_OPTS="$JVM_OPTS -ea" check_openjdk=`"${JAVA:-java}" -version 2>&1 | awk '{if (NR == 2) {print $1}}'` if [ "$check_openjdk" != "OpenJDK" ] then - JVM_OPTS="$JVM_OPTS -javaagent:$CASSANDRA_HOME/lib/jamm-0.2.2.jar" + JVM_OPTS="$JVM_OPTS -javaagent:$CASSANDRA_HOME/lib/jamm-0.2.4.jar" fi # enable thread priorities, primarily so we can give periodic tasks diff --git a/lib/jamm-0.2.2.jar b/lib/jamm-0.2.2.jar deleted file mode 100644 index ce4227a950..0000000000 Binary files a/lib/jamm-0.2.2.jar and /dev/null differ diff --git a/lib/jamm-0.2.4.jar b/lib/jamm-0.2.4.jar new file mode 100644 index 0000000000..e69a774254 Binary files /dev/null and b/lib/jamm-0.2.4.jar differ diff --git a/src/java/org/apache/cassandra/db/AbstractColumnContainer.java b/src/java/org/apache/cassandra/db/AbstractColumnContainer.java index 6f77d32996..c10bf0c964 100644 --- a/src/java/org/apache/cassandra/db/AbstractColumnContainer.java +++ b/src/java/org/apache/cassandra/db/AbstractColumnContainer.java @@ -35,7 +35,9 @@ import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.io.ICompactSerializer2; import org.apache.cassandra.io.util.IIterableColumns; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.HeapAllocator; public abstract class AbstractColumnContainer implements IColumnContainer, IIterableColumns { @@ -105,15 +107,20 @@ public abstract class AbstractColumnContainer implements IColumnContainer, IIter /** * We need to go through each column in the column container and resolve it before adding */ - public void addAll(AbstractColumnContainer cc) + public void addAll(AbstractColumnContainer cc, Allocator allocator) { - columns.addAll(cc.columns); + columns.addAll(cc.columns, allocator); delete(cc); } public void addColumn(IColumn column) { - columns.addColumn(column); + addColumn(column, HeapAllocator.instance); + } + + public void addColumn(IColumn column, Allocator allocator) + { + columns.addColumn(column, allocator); } public IColumn getColumn(ByteBuffer name) diff --git a/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java b/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java index 1416e4639f..cc07f83ca5 100644 --- a/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java +++ b/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java @@ -21,6 +21,7 @@ import java.nio.ByteBuffer; import java.util.*; import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.utils.Allocator; /** * A ISortedColumns backed by an ArrayList. @@ -100,7 +101,7 @@ public class ArrayBackedSortedColumns extends ArrayList implements ISor * without knowing about (we can revisit that decision later if we have * use cases where most insert are in sorted order but a few are not). */ - public void addColumn(IColumn column) + public void addColumn(IColumn column, Allocator allocator) { if (isEmpty()) { @@ -122,13 +123,13 @@ public class ArrayBackedSortedColumns extends ArrayList implements ISor else if (c == 0) { // Resolve against last - resolveAgainst(size() - 1, column); + resolveAgainst(size() - 1, column, allocator); } else { int pos = binarySearch(column.name()); if (pos >= 0) - resolveAgainst(pos, column); + resolveAgainst(pos, column, allocator); else add(-pos-1, column); } @@ -138,19 +139,19 @@ public class ArrayBackedSortedColumns extends ArrayList implements ISor * Resolve against element at position i. * Assume that i is a valid position. */ - private void resolveAgainst(int i, IColumn column) + private void resolveAgainst(int i, IColumn column, Allocator allocator) { IColumn oldColumn = get(i); if (oldColumn instanceof SuperColumn) { // Delegated to SuperColumn assert column instanceof SuperColumn; - ((SuperColumn) oldColumn).putColumn((SuperColumn)column); + ((SuperColumn) oldColumn).putColumn((SuperColumn)column, allocator); } else { // calculate reconciled col from old (existing) col and new col - IColumn reconciledColumn = column.reconcile(oldColumn); + IColumn reconciledColumn = column.reconcile(oldColumn, allocator); set(i, reconciledColumn); } } @@ -186,7 +187,7 @@ public class ArrayBackedSortedColumns extends ArrayList implements ISor return -mid - (result < 0 ? 1 : 2); } - public void addAll(ISortedColumns cm) + public void addAll(ISortedColumns cm, Allocator allocator) { if (cm.isEmpty()) return; @@ -214,7 +215,7 @@ public class ArrayBackedSortedColumns extends ArrayList implements ISor else // c == 0 { add(copy[idx]); - resolveAgainst(size() - 1, otherColumn); + resolveAgainst(size() - 1, otherColumn, allocator); idx++; otherColumn = other.hasNext() ? other.next() : null; } diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java index f978c1130a..376cb0d276 100644 --- a/src/java/org/apache/cassandra/db/Column.java +++ b/src/java/org/apache/cassandra/db/Column.java @@ -30,8 +30,9 @@ import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.ByteBufferUtil; - +import org.apache.cassandra.utils.HeapAllocator; /** * Column is immutable, which prevents all kinds of confusion in a multithreaded environment. @@ -146,6 +147,11 @@ public class Column implements IColumn } public void addColumn(IColumn column) + { + addColumn(null, null); + } + + public void addColumn(IColumn column, Allocator allocator) { throw new UnsupportedOperationException("This operation is not supported for simple columns."); } @@ -183,6 +189,11 @@ public class Column implements IColumn } public IColumn reconcile(IColumn column) + { + return reconcile(column, HeapAllocator.instance); + } + + public IColumn reconcile(IColumn column, Allocator allocator) { // tombstones take precedence. (if both are tombstones, then it doesn't matter which one we use.) if (isMarkedForDelete()) @@ -225,9 +236,14 @@ public class Column implements IColumn public IColumn localCopy(ColumnFamilyStore cfs) { - return new Column(cfs.internOrCopy(name), ByteBufferUtil.clone(value), timestamp); + return localCopy(cfs, HeapAllocator.instance); } + public IColumn localCopy(ColumnFamilyStore cfs, Allocator allocator) + { + return new Column(cfs.internOrCopy(name, allocator), allocator.clone(value), timestamp); + } + public String getString(AbstractType comparator) { StringBuilder sb = new StringBuilder(); diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index ee02864039..09b0ca5614 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -31,7 +31,9 @@ import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.io.IColumnSerializer; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.HeapAllocator; public class ColumnFamily extends AbstractColumnContainer { @@ -121,6 +123,10 @@ public class ColumnFamily extends AbstractColumnContainer return cfm; } + /** + * FIXME: shouldn't need to hold a reference to a serializer; worse, for super cfs, + * it will be a _unique_ serializer object per row + */ public IColumnSerializer getColumnSerializer() { return cfm.getColumnSerializer(); @@ -301,11 +307,16 @@ public class ColumnFamily extends AbstractColumnContainer } public void resolve(ColumnFamily cf) + { + resolve(cf, HeapAllocator.instance); + } + + public void resolve(ColumnFamily cf, Allocator allocator) { // Row _does_ allow null CF objects :( seems a necessary evil for efficiency if (cf == null) return; - addAll(cf); + addAll(cf, allocator); } public long serializedSize() diff --git a/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java b/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java index 4c07819064..5c02336956 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java @@ -110,10 +110,10 @@ public class ColumnFamilySerializer implements ICompactSerializer3 public ColumnFamily deserialize(DataInput dis) throws IOException { - return deserialize(dis, false, false, ThreadSafeSortedColumns.FACTORY); + return deserialize(dis, false, ThreadSafeSortedColumns.FACTORY); } - public ColumnFamily deserialize(DataInput dis, boolean intern, boolean fromRemote, ISortedColumns.Factory factory) throws IOException + public ColumnFamily deserialize(DataInput dis, boolean fromRemote, ISortedColumns.Factory factory) throws IOException { if (!dis.readBoolean()) return null; @@ -124,23 +124,22 @@ public class ColumnFamilySerializer implements ICompactSerializer3 throw new UnserializableColumnFamilyException("Couldn't find cfId=" + cfId, cfId); ColumnFamily cf = ColumnFamily.create(cfId, factory); deserializeFromSSTableNoColumns(cf, dis); - deserializeColumns(dis, cf, intern, fromRemote); + deserializeColumns(dis, cf, fromRemote); return cf; } - public void deserializeColumns(DataInput dis, ColumnFamily cf, boolean intern, boolean fromRemote) throws IOException + public void deserializeColumns(DataInput dis, ColumnFamily cf, boolean fromRemote) throws IOException { - int count = dis.readInt(); - deserializeColumns(dis, cf, count, intern, fromRemote); + int size = dis.readInt(); + deserializeColumns(dis, cf, size, fromRemote); } /* column count is already read from DataInput */ - public void deserializeColumns(DataInput dis, ColumnFamily cf, int count, boolean intern, boolean fromRemote) throws IOException + public void deserializeColumns(DataInput dis, ColumnFamily cf, int size, boolean fromRemote) throws IOException { - ColumnFamilyStore interner = intern ? Table.open(CFMetaData.getCF(cf.id()).left).getColumnFamilyStore(cf.id()) : null; - for (int i = 0; i < count; ++i) + for (int i = 0; i < size; ++i) { - IColumn column = cf.getColumnSerializer().deserialize(dis, interner, fromRemote, (int) (System.currentTimeMillis() / 1000)); + IColumn column = cf.getColumnSerializer().deserialize(dis, fromRemote, (int) (System.currentTimeMillis() / 1000)); cf.addColumn(column); } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 12e9056198..e9404e0fca 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -749,7 +749,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { ColumnFamily cachedRow = getRawCachedRow(key); if (cachedRow != null) - cachedRow.addAll(columnFamily); + cachedRow.addAll(columnFamily, HeapAllocator.instance); } } @@ -1211,7 +1211,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { ColumnFamily cf = cached.cloneMeShallow(); if (sc != null) - cf.addColumn(sc); + cf.addColumn(sc, HeapAllocator.instance); return removeDeleted(cf, gcBefore); } } @@ -1919,10 +1919,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean return internedName; } - public ByteBuffer internOrCopy(ByteBuffer name) + public ByteBuffer internOrCopy(ByteBuffer name, Allocator allocator) { if (internedNames.size() >= INTERN_CUTOFF) - return ByteBufferUtil.clone(name); + return allocator.clone(name); return intern(name); } @@ -1930,7 +1930,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public ByteBuffer maybeIntern(ByteBuffer name) { if (internedNames.size() >= INTERN_CUTOFF) - return name; + return null; return intern(name); } diff --git a/src/java/org/apache/cassandra/db/ColumnSerializer.java b/src/java/org/apache/cassandra/db/ColumnSerializer.java index bcbccb7671..d65cc5fbe3 100644 --- a/src/java/org/apache/cassandra/db/ColumnSerializer.java +++ b/src/java/org/apache/cassandra/db/ColumnSerializer.java @@ -69,7 +69,7 @@ public class ColumnSerializer implements IColumnSerializer public Column deserialize(DataInput dis) throws IOException { - return deserialize(dis, null, false); + return deserialize(dis, false); } /* @@ -77,18 +77,16 @@ public class ColumnSerializer implements IColumnSerializer * deserialize comes from a remote host. If it does, then we must clear * the delta. */ - public Column deserialize(DataInput dis, ColumnFamilyStore interner, boolean fromRemote) throws IOException + public Column deserialize(DataInput dis, boolean fromRemote) throws IOException { - return deserialize(dis, interner, fromRemote, (int) (System.currentTimeMillis() / 1000)); + return deserialize(dis, fromRemote, (int) (System.currentTimeMillis() / 1000)); } - public Column deserialize(DataInput dis, ColumnFamilyStore interner, boolean fromRemote, int expireBefore) throws IOException + public Column deserialize(DataInput dis, boolean fromRemote, int expireBefore) throws IOException { ByteBuffer name = ByteBufferUtil.readWithShortLength(dis); if (name.remaining() <= 0) throw new CorruptColumnException("invalid column name length " + name.remaining()); - if (interner != null) - name = interner.maybeIntern(name); int b = dis.readUnsignedByte(); if ((b & COUNTER_MASK) != 0) diff --git a/src/java/org/apache/cassandra/db/CounterColumn.java b/src/java/org/apache/cassandra/db/CounterColumn.java index 749ef2d9c5..fb796343d7 100644 --- a/src/java/org/apache/cassandra/db/CounterColumn.java +++ b/src/java/org/apache/cassandra/db/CounterColumn.java @@ -31,7 +31,9 @@ import org.apache.cassandra.db.context.IContext.ContextRelationship; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.HeapAllocator; import org.apache.cassandra.utils.NodeId; /** @@ -47,12 +49,12 @@ public class CounterColumn extends Column public CounterColumn(ByteBuffer name, long value, long timestamp) { - this(name, contextManager.create(value), timestamp); + this(name, contextManager.create(value, HeapAllocator.instance), timestamp); } public CounterColumn(ByteBuffer name, long value, long timestamp, long timestampOfLastDelete) { - this(name, contextManager.create(value), timestamp, timestampOfLastDelete); + this(name, contextManager.create(value, HeapAllocator.instance), timestamp, timestampOfLastDelete); } public CounterColumn(ByteBuffer name, ByteBuffer value, long timestamp) @@ -135,7 +137,7 @@ public class CounterColumn extends Column } @Override - public IColumn reconcile(IColumn column) + public IColumn reconcile(IColumn column, Allocator allocator) { assert (column instanceof CounterColumn) || (column instanceof DeletedColumn) : "Wrong class type."; @@ -162,7 +164,7 @@ public class CounterColumn extends Column // live + live: merge clocks; update value return new CounterColumn( name(), - contextManager.merge(value(), column.value()), + contextManager.merge(value(), column.value(), allocator), Math.max(timestamp(), column.timestamp()), Math.max(timestampOfLastDelete(), ((CounterColumn)column).timestampOfLastDelete())); } @@ -185,7 +187,13 @@ public class CounterColumn extends Column @Override public IColumn localCopy(ColumnFamilyStore cfs) { - return new CounterColumn(cfs.internOrCopy(name), ByteBufferUtil.clone(value), timestamp, timestampOfLastDelete); + return new CounterColumn(cfs.internOrCopy(name, HeapAllocator.instance), ByteBufferUtil.clone(value), timestamp, timestampOfLastDelete); + } + + @Override + public IColumn localCopy(ColumnFamilyStore cfs, Allocator allocator) + { + return new CounterColumn(cfs.internOrCopy(name, allocator), allocator.clone(value), timestamp, timestampOfLastDelete); } @Override diff --git a/src/java/org/apache/cassandra/db/CounterMutation.java b/src/java/org/apache/cassandra/db/CounterMutation.java index 1f0a8da510..b3ba646e85 100644 --- a/src/java/org/apache/cassandra/db/CounterMutation.java +++ b/src/java/org/apache/cassandra/db/CounterMutation.java @@ -40,7 +40,9 @@ import org.apache.cassandra.io.ICompactSerializer; import org.apache.cassandra.io.util.FastByteArrayOutputStream; import org.apache.cassandra.net.Message; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.HeapAllocator; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.thrift.ConsistencyLevel; @@ -144,7 +146,7 @@ public class CounterMutation implements IMutation localMutation.add(merger); localMutation.apply(); - cf.addAll(merger); + cf.addAll(merger, HeapAllocator.instance); } } return row; @@ -216,8 +218,7 @@ public class CounterMutation implements IMutation public void apply() throws IOException { - // We need to transform all CounterUpdateColumn to CounterColumn and we need to deepCopy. Both are done - // below since CUC.asCounterColumn() does a deep copy. + // transform all CounterUpdateColumn to CounterColumn: accomplished by localCopy RowMutation rm = new RowMutation(rowMutation.getTable(), ByteBufferUtil.clone(rowMutation.key())); Table table = Table.open(rm.getTable()); @@ -227,7 +228,7 @@ public class CounterMutation implements IMutation ColumnFamilyStore cfs = table.getColumnFamilyStore(cf.id()); for (IColumn column : cf_) { - cf.addColumn(column.localCopy(cfs)); + cf.addColumn(column.localCopy(cfs), HeapAllocator.instance); } rm.add(cf); } diff --git a/src/java/org/apache/cassandra/db/CounterUpdateColumn.java b/src/java/org/apache/cassandra/db/CounterUpdateColumn.java index d6d3d9a32f..94b8ed84de 100644 --- a/src/java/org/apache/cassandra/db/CounterUpdateColumn.java +++ b/src/java/org/apache/cassandra/db/CounterUpdateColumn.java @@ -21,7 +21,9 @@ package org.apache.cassandra.db; import java.nio.ByteBuffer; import org.apache.cassandra.db.context.CounterContext; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.HeapAllocator; /** * A counter update while it hasn't been applied yet by the leader replica. @@ -55,7 +57,7 @@ public class CounterUpdateColumn extends Column } @Override - public IColumn reconcile(IColumn column) + public IColumn reconcile(IColumn column, Allocator allocator) { // The only time this could happen is if a batchAdd ships two // increment for the same column. Hence we simply sums the delta. @@ -80,8 +82,17 @@ public class CounterUpdateColumn extends Column @Override public CounterColumn localCopy(ColumnFamilyStore cfs) { - return new CounterColumn(cfs.internOrCopy(name), - CounterContext.instance().create(delta()), + return new CounterColumn(cfs.internOrCopy(name, HeapAllocator.instance), + CounterContext.instance().create(delta(), HeapAllocator.instance), + timestamp(), + Long.MIN_VALUE); + } + + @Override + public IColumn localCopy(ColumnFamilyStore cfs, Allocator allocator) + { + return new CounterColumn(cfs.internOrCopy(name, allocator), + CounterContext.instance().create(delta(), allocator), timestamp(), Long.MIN_VALUE); } diff --git a/src/java/org/apache/cassandra/db/DeletedColumn.java b/src/java/org/apache/cassandra/db/DeletedColumn.java index 8f2de11bf6..eff8c5c712 100644 --- a/src/java/org/apache/cassandra/db/DeletedColumn.java +++ b/src/java/org/apache/cassandra/db/DeletedColumn.java @@ -22,7 +22,9 @@ import java.nio.ByteBuffer; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.db.marshal.MarshalException; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.HeapAllocator; public class DeletedColumn extends Column { @@ -55,17 +57,23 @@ public class DeletedColumn extends Column } @Override - public IColumn reconcile(IColumn column) + public IColumn reconcile(IColumn column, Allocator allocator) { if (column instanceof DeletedColumn) - return super.reconcile(column); - return column.reconcile(this); + return super.reconcile(column, allocator); + return column.reconcile(this, allocator); } @Override public IColumn localCopy(ColumnFamilyStore cfs) { - return new DeletedColumn(cfs.internOrCopy(name), ByteBufferUtil.clone(value), timestamp); + return new DeletedColumn(cfs.internOrCopy(name, HeapAllocator.instance), ByteBufferUtil.clone(value), timestamp); + } + + @Override + public IColumn localCopy(ColumnFamilyStore cfs, Allocator allocator) + { + return new DeletedColumn(cfs.internOrCopy(name, allocator), allocator.clone(value), timestamp); } @Override diff --git a/src/java/org/apache/cassandra/db/ExpiringColumn.java b/src/java/org/apache/cassandra/db/ExpiringColumn.java index 39a3cd579d..68ffd31033 100644 --- a/src/java/org/apache/cassandra/db/ExpiringColumn.java +++ b/src/java/org/apache/cassandra/db/ExpiringColumn.java @@ -26,7 +26,9 @@ import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.HeapAllocator; /** * Alternative to Column that have an expiring time. @@ -117,7 +119,16 @@ public class ExpiringColumn extends Column @Override public IColumn localCopy(ColumnFamilyStore cfs) { - return new ExpiringColumn(cfs.internOrCopy(name), ByteBufferUtil.clone(value), timestamp, timeToLive, localExpirationTime); + return new ExpiringColumn(cfs.internOrCopy(name, HeapAllocator.instance), ByteBufferUtil.clone(value), timestamp, timeToLive, localExpirationTime); + } + + @Override + public IColumn localCopy(ColumnFamilyStore cfs, Allocator allocator) + { + ByteBuffer clonedName = cfs.maybeIntern(name); + if (clonedName == null) + clonedName = allocator.clone(name); + return new ExpiringColumn(clonedName, allocator.clone(value), timestamp, timeToLive, localExpirationTime); } @Override diff --git a/src/java/org/apache/cassandra/db/IColumn.java b/src/java/org/apache/cassandra/db/IColumn.java index 6bf932a14c..d6fb656e86 100644 --- a/src/java/org/apache/cassandra/db/IColumn.java +++ b/src/java/org/apache/cassandra/db/IColumn.java @@ -25,8 +25,10 @@ import java.util.Collection; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.MarshalException; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.FBUtilities; +/** TODO: rename */ public interface IColumn { public static final int MAX_NAME_LENGTH = FBUtilities.MAX_UNSIGNED_SHORT; @@ -43,17 +45,25 @@ public interface IColumn public Collection getSubColumns(); public IColumn getSubColumn(ByteBuffer columnName); public void addColumn(IColumn column); + public void addColumn(IColumn column, Allocator allocator); public IColumn diff(IColumn column); public IColumn reconcile(IColumn column); + public IColumn reconcile(IColumn column, Allocator allocator); public void updateDigest(MessageDigest digest); public int getLocalDeletionTime(); // for tombstone GC, so int is sufficient granularity public String getString(AbstractType comparator); public void validateFields(CFMetaData metadata) throws MarshalException; - /** clones the column, interning column names and making copies of other underlying byte buffers - * @param cfs*/ + /** clones the column for the row cache, interning column names and making copies of other underlying byte buffers */ IColumn localCopy(ColumnFamilyStore cfs); + /** + * clones the column for the memtable, interning column names and making copies of other underlying byte buffers. + * Unlike the other localCopy, this uses Allocator to allocate values in contiguous memory regions, + * which helps avoid heap fragmentation. + */ + IColumn localCopy(ColumnFamilyStore cfs, Allocator allocator); + /** * For a simple column, live == !isMarkedForDelete. * For a supercolumn, live means it has at least one subcolumn whose timestamp is greater than the diff --git a/src/java/org/apache/cassandra/db/IColumnContainer.java b/src/java/org/apache/cassandra/db/IColumnContainer.java index 6eb6afd775..1e6b42d9fb 100644 --- a/src/java/org/apache/cassandra/db/IColumnContainer.java +++ b/src/java/org/apache/cassandra/db/IColumnContainer.java @@ -24,10 +24,12 @@ import java.nio.ByteBuffer; import java.util.Collection; import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.utils.Allocator; public interface IColumnContainer { public void addColumn(IColumn column); + public void addColumn(IColumn column, Allocator allocator); public void remove(ByteBuffer columnName); /** diff --git a/src/java/org/apache/cassandra/db/ISortedColumns.java b/src/java/org/apache/cassandra/db/ISortedColumns.java index 10300bf69a..37f5a605ba 100644 --- a/src/java/org/apache/cassandra/db/ISortedColumns.java +++ b/src/java/org/apache/cassandra/db/ISortedColumns.java @@ -25,6 +25,7 @@ import java.util.SortedSet; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.io.util.IIterableColumns; +import org.apache.cassandra.utils.Allocator; /** * A sorted map of columns. @@ -45,7 +46,7 @@ public interface ISortedColumns extends IIterableColumns * If a column with the same name is already present in the map, it will * be replaced by the newly added column. */ - public void addColumn(IColumn column); + public void addColumn(IColumn column, Allocator allocator); /** * Adds all the columns of a given column map to this column map. @@ -56,7 +57,7 @@ public interface ISortedColumns extends IIterableColumns * * but is potentially faster. */ - public void addAll(ISortedColumns cm); + public void addAll(ISortedColumns cm, Allocator allocator); /** * Replace oldColumn if present by newColumn. diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index 906abe5f96..6b00c84199 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -43,6 +43,7 @@ import org.apache.cassandra.db.filter.SliceQueryFilter; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.io.sstable.SSTableReader; import org.apache.cassandra.io.sstable.SSTableWriter; +import org.apache.cassandra.utils.SlabAllocator; import org.apache.cassandra.utils.WrappedRunnable; import org.github.jamm.MemoryMeter; @@ -54,7 +55,7 @@ public class Memtable private static final double MIN_SANE_LIVE_RATIO = 1.0; // max liveratio seen w/ 1-byte columns on a 64-bit jvm was 19. If it gets higher than 64 something is probably broken. private static final double MAX_SANE_LIVE_RATIO = 64.0; - private static final MemoryMeter meter = new MemoryMeter(); + private static final MemoryMeter meter = new MemoryMeter().omitSharedBufferOverhead(); // we're careful to only allow one count to run at a time because counting is slow // (can be minutes, for a large memtable and a busy server), so we could keep memtables @@ -69,6 +70,8 @@ public class Memtable } }; + volatile static Memtable activelyMeasuring; + private volatile boolean isFrozen; private final AtomicLong currentThroughput = new AtomicLong(0); private final AtomicLong currentOperations = new AtomicLong(0); @@ -78,7 +81,7 @@ public class Memtable private final long THRESHOLD; private final long THRESHOLD_COUNT; - volatile static Memtable activelyMeasuring; + private SlabAllocator allocator = new SlabAllocator(); public Memtable(ColumnFamilyStore cfs) { @@ -89,8 +92,9 @@ public class Memtable public long getLiveSize() { - // 25% fudge factor - return (long) (currentThroughput.get() * cfs.liveRatio * 1.25); + // 25% fudge factor on the base throughput * liveRatio calculation. (Based on observed + // pre-slabbing behavior -- not sure what accounts for this. May have changed with introduction of slabbing.) + return (long) (currentThroughput.get() * cfs.liveRatio * 1.25) + allocator.size(); } public long getSerializedSize() @@ -192,11 +196,25 @@ public class Memtable ? cf.isMarkedForDelete() ? 1 : 0 : cf.getColumnCount()); - ColumnFamily oldCf = columnFamilies.putIfAbsent(key, cf); - if (oldCf == null) - return; + ColumnFamily clonedCf = columnFamilies.get(key); + // if the row doesn't exist yet in the memtable, clone cf to our allocator. + if (clonedCf == null) + { + clonedCf = cf.cloneMeShallow(); + for (IColumn column : cf.getSortedColumns()) + clonedCf.addColumn(column.localCopy(cfs, allocator)); + clonedCf = columnFamilies.putIfAbsent(new DecoratedKey(key.token, allocator.clone(key.key)), clonedCf); + if (clonedCf == null) + return; + // else there was a race and the other thread won. fall through to updating his CF object + } - oldCf.resolve(cf); + // we duplicate the funcationality of CF.resolve here to avoid having to either pass the Memtable in for + // the cloning operation, or cloning the CF container as well as the Columns. fortunately, resolve + // is really quite simple: + clonedCf.delete(cf); + for (IColumn column : cf.getSortedColumns()) + clonedCf.addColumn(column.localCopy(cfs, allocator), allocator); } // for debugging diff --git a/src/java/org/apache/cassandra/db/Row.java b/src/java/org/apache/cassandra/db/Row.java index b1d62a2ed0..25a9f2fb85 100644 --- a/src/java/org/apache/cassandra/db/Row.java +++ b/src/java/org/apache/cassandra/db/Row.java @@ -66,7 +66,7 @@ public class Row public Row deserialize(DataInputStream dis, int version, boolean fromRemote, ISortedColumns.Factory factory) throws IOException { return new Row(StorageService.getPartitioner().decorateKey(ByteBufferUtil.readWithShortLength(dis)), - ColumnFamily.serializer().deserialize(dis, false, fromRemote, factory)); + ColumnFamily.serializer().deserialize(dis, fromRemote, factory)); } public Row deserialize(DataInputStream dis, int version) throws IOException diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index 7eede71865..2635186e84 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -373,23 +373,6 @@ public class RowMutation implements IMutation, MessageProducer return rm; } - public RowMutation localCopy() - { - RowMutation rm = new RowMutation(table_, ByteBufferUtil.clone(key_)); - - Table table = Table.open(table_); - for (Map.Entry entry : modifications_.entrySet()) - { - ColumnFamily cf = entry.getValue().cloneMeShallow(); - ColumnFamilyStore cfs = table.getColumnFamilyStore(cf.id()); - for (IColumn col : entry.getValue()) - cf.addColumn(col.localCopy(cfs)); - rm.modifications_.put(entry.getKey(), cf); - } - - return rm; - } - public static class RowMutationSerializer implements ICompactSerializer { public void serialize(RowMutation rm, DataOutputStream dos, int version) throws IOException @@ -419,7 +402,7 @@ public class RowMutation implements IMutation, MessageProducer for (int i = 0; i < size; ++i) { Integer cfid = Integer.valueOf(dis.readInt()); - ColumnFamily cf = ColumnFamily.serializer().deserialize(dis, true, fromRemote, ThreadSafeSortedColumns.FACTORY); + ColumnFamily cf = ColumnFamily.serializer().deserialize(dis, fromRemote, ThreadSafeSortedColumns.FACTORY); modifications.put(cfid, cf); } return new RowMutation(table, key, modifications); diff --git a/src/java/org/apache/cassandra/db/SuperColumn.java b/src/java/org/apache/cassandra/db/SuperColumn.java index ad28f024dd..d877464b32 100644 --- a/src/java/org/apache/cassandra/db/SuperColumn.java +++ b/src/java/org/apache/cassandra/db/SuperColumn.java @@ -25,9 +25,6 @@ import java.nio.ByteBuffer; import java.security.MessageDigest; import java.util.Collection; import java.util.Comparator; -import java.util.Iterator; -import java.util.Map; -import java.util.concurrent.ConcurrentSkipListMap; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.db.marshal.AbstractType; @@ -35,10 +32,11 @@ import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.io.IColumnSerializer; import org.apache.cassandra.io.util.ColumnSortedMap; import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.HeapAllocator; import org.cliffc.high_scale_lib.NonBlockingHashMap; - public class SuperColumn extends AbstractColumnContainer implements IColumn { private static NonBlockingHashMap serializers = new NonBlockingHashMap(); @@ -160,21 +158,21 @@ public class SuperColumn extends AbstractColumnContainer implements IColumn } @Override - public void addColumn(IColumn column) + public void addColumn(IColumn column, Allocator allocator) { assert column instanceof Column : "A super column can only contain simple columns"; - super.addColumn((Column)column); + super.addColumn(column, allocator); } /* * Go through each sub column if it exists then as it to resolve itself * if the column does not exist then create it. */ - void putColumn(SuperColumn column) + void putColumn(SuperColumn column, Allocator allocator) { for (IColumn subColumn : column.getSubColumns()) { - addColumn(subColumn); + addColumn(subColumn, allocator); } delete(column); } @@ -255,29 +253,33 @@ public class SuperColumn extends AbstractColumnContainer implements IColumn return mostRecentLiveChangeAt() > getMarkedForDeleteAt(); } - public IColumn shallowCopy() - { - SuperColumn sc = new SuperColumn(ByteBufferUtil.clone(name()), this.getComparator()); - // since deletion info is immutable, aliasing it is fine - sc.deletionInfo.set(deletionInfo.get()); - return sc; - } - public IColumn localCopy(ColumnFamilyStore cfs) + { + return localCopy(cfs, HeapAllocator.instance); + } + + public IColumn localCopy(ColumnFamilyStore cfs, Allocator allocator) { // we don't try to intern supercolumn names, because if we're using Cassandra correctly it's almost // certainly just going to pollute our interning map with unique, dynamic values - SuperColumn sc = (SuperColumn)shallowCopy(); + SuperColumn sc = new SuperColumn(allocator.clone(name), this.getComparator()); + // since deletion info is immutable, aliasing it is fine + sc.deletionInfo.set(deletionInfo.get()); for(IColumn c : columns) { - sc.addColumn(c.localCopy(cfs)); + sc.addColumn(c.localCopy(cfs, allocator)); } return sc; } public IColumn reconcile(IColumn c) + { + return reconcile(null, null); + } + + public IColumn reconcile(IColumn c, Allocator allocator) { throw new UnsupportedOperationException("This operation is unsupported on super columns."); } @@ -335,20 +337,15 @@ class SuperColumnSerializer implements IColumnSerializer public IColumn deserialize(DataInput dis) throws IOException { - return deserialize(dis, null, false); + return deserialize(dis, false); } - public IColumn deserialize(DataInput dis, ColumnFamilyStore interner) throws IOException + public IColumn deserialize(DataInput dis, boolean fromRemote) throws IOException { - return deserialize(dis, interner, false); + return deserialize(dis, fromRemote, (int)(System.currentTimeMillis() / 1000)); } - public IColumn deserialize(DataInput dis, ColumnFamilyStore interner, boolean fromRemote) throws IOException - { - return deserialize(dis, interner, fromRemote, (int)(System.currentTimeMillis() / 1000)); - } - - public IColumn deserialize(DataInput dis, ColumnFamilyStore interner, boolean fromRemote, int expireBefore) throws IOException + public IColumn deserialize(DataInput dis, boolean fromRemote, int expireBefore) throws IOException { ByteBuffer name = ByteBufferUtil.readWithShortLength(dis); int localDeleteTime = dis.readInt(); @@ -361,7 +358,7 @@ class SuperColumnSerializer implements IColumnSerializer /* read the number of columns */ int size = dis.readInt(); ColumnSerializer serializer = Column.serializer(); - ColumnSortedMap preSortedMap = new ColumnSortedMap(comparator, serializer, dis, interner, size, fromRemote, expireBefore); + ColumnSortedMap preSortedMap = new ColumnSortedMap(comparator, serializer, dis, size, fromRemote, expireBefore); SuperColumn superColumn = new SuperColumn(name, ThreadSafeSortedColumns.factory().fromSorted(preSortedMap, false)); if (localDeleteTime != Integer.MIN_VALUE && localDeleteTime <= 0) { diff --git a/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java b/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java index cbfaf77e7c..941aa7b6ae 100644 --- a/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java +++ b/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java @@ -18,15 +18,14 @@ package org.apache.cassandra.db; import java.nio.ByteBuffer; -import java.util.Comparator; import java.util.Collection; import java.util.Iterator; -import java.util.Map; import java.util.SortedMap; import java.util.SortedSet; import java.util.concurrent.ConcurrentSkipListMap; import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.utils.Allocator; public class ThreadSafeSortedColumns extends ConcurrentSkipListMap implements ISortedColumns { @@ -72,7 +71,7 @@ public class ThreadSafeSortedColumns extends ConcurrentSkipListMap { - public IColumn deserialize(DataInput in, ColumnFamilyStore interner, boolean fromRemote, int expireBefore) throws IOException; + public IColumn deserialize(DataInput in, boolean fromRemote, int expireBefore) throws IOException; } diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java index bf32995429..63f765477b 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java @@ -175,7 +175,7 @@ public class SSTableIdentityIterator implements Comparable bufferSize) sync(); diff --git a/src/java/org/apache/cassandra/io/util/ColumnSortedMap.java b/src/java/org/apache/cassandra/io/util/ColumnSortedMap.java index 4115ac6ae7..f83d5b86a8 100644 --- a/src/java/org/apache/cassandra/io/util/ColumnSortedMap.java +++ b/src/java/org/apache/cassandra/io/util/ColumnSortedMap.java @@ -43,15 +43,13 @@ public class ColumnSortedMap implements SortedMap private final DataInput dis; private final Comparator comparator; private final int length; - private final ColumnFamilyStore interner; private final boolean fromRemote; private final int expireBefore; - public ColumnSortedMap(Comparator comparator, ColumnSerializer serializer, DataInput dis, ColumnFamilyStore interner, int length, boolean fromRemote, int expireBefore) + public ColumnSortedMap(Comparator comparator, ColumnSerializer serializer, DataInput dis, int length, boolean fromRemote, int expireBefore) { this.comparator = comparator; this.serializer = serializer; - this.interner = interner; this.dis = dis; this.length = length; this.fromRemote = fromRemote; @@ -145,7 +143,7 @@ public class ColumnSortedMap implements SortedMap public Set> entrySet() { - return new ColumnSet(serializer, dis, interner, length, fromRemote, expireBefore); + return new ColumnSet(serializer, dis, length, fromRemote, expireBefore); } } @@ -154,15 +152,13 @@ class ColumnSet implements Set> private final ColumnSerializer serializer; private final DataInput dis; private final int length; - private final ColumnFamilyStore interner; private boolean fromRemote; private final int expireBefore; - public ColumnSet(ColumnSerializer serializer, DataInput dis, ColumnFamilyStore interner, int length, boolean fromRemote, int expireBefore) + public ColumnSet(ColumnSerializer serializer, DataInput dis, int length, boolean fromRemote, int expireBefore) { this.serializer = serializer; this.dis = dis; - this.interner = interner; this.length = length; this.fromRemote = fromRemote; this.expireBefore = expireBefore; @@ -185,7 +181,7 @@ class ColumnSet implements Set> public Iterator> iterator() { - return new ColumnIterator(serializer, dis, interner, length, fromRemote, expireBefore); + return new ColumnIterator(serializer, dis, length, fromRemote, expireBefore); } public Object[] toArray() @@ -240,14 +236,12 @@ class ColumnIterator implements Iterator> private final int length; private final boolean fromRemote; private int count = 0; - private ColumnFamilyStore interner; private final int expireBefore; - public ColumnIterator(ColumnSerializer serializer, DataInput dis, ColumnFamilyStore interner, int length, boolean fromRemote, int expireBefore) + public ColumnIterator(ColumnSerializer serializer, DataInput dis, int length, boolean fromRemote, int expireBefore) { this.dis = dis; this.serializer = serializer; - this.interner = interner; this.length = length; this.fromRemote = fromRemote; this.expireBefore = expireBefore; @@ -258,7 +252,7 @@ class ColumnIterator implements Iterator> try { count++; - return serializer.deserialize(dis, interner, fromRemote, expireBefore); + return serializer.deserialize(dis, fromRemote, expireBefore); } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index eb448fc25c..7eff1f3a18 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -359,7 +359,7 @@ public class StorageProxy implements StorageProxyMBean { public void runMayThrow() throws IOException { - rm.localCopy().apply(); + rm.apply(); responseHandler.response(null); } }; diff --git a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java index 22883cb406..2688792b4a 100644 --- a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java +++ b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java @@ -148,7 +148,7 @@ public class IncomingStreamReader // restore ColumnFamily cf = ColumnFamily.create(cfs.metadata); ColumnFamily.serializer().deserializeFromSSTableNoColumns(cf, in); - ColumnFamily.serializer().deserializeColumns(in, cf, true, true); + ColumnFamily.serializer().deserializeColumns(in, cf, true); // write key and cf writer.append(key, cf); diff --git a/src/java/org/apache/cassandra/utils/Allocator.java b/src/java/org/apache/cassandra/utils/Allocator.java new file mode 100644 index 0000000000..671111c23d --- /dev/null +++ b/src/java/org/apache/cassandra/utils/Allocator.java @@ -0,0 +1,41 @@ +/** + * Copyright 2011 The Apache Software Foundation + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.utils; + +import java.nio.ByteBuffer; + +public abstract class Allocator +{ + /** + * Allocate a slice of the given length. + */ + public ByteBuffer clone(ByteBuffer buffer) + { + assert buffer != null; + ByteBuffer cloned = allocate(buffer.remaining()); + + cloned.mark(); + cloned.put(buffer.duplicate()); + cloned.reset(); + return cloned; + } + + public abstract ByteBuffer allocate(int size); +} diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index 860efa62dc..220fea0fc4 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java @@ -169,11 +169,11 @@ public class ByteBufferUtil if (buffer.hasArray()) { - int start = buffer.position(); - if (buffer.arrayOffset() == 0 && start == 0 && length == buffer.array().length) + int boff = buffer.arrayOffset() + buffer.position(); + if (boff == 0 && length == buffer.array().length) return buffer.array(); else - return Arrays.copyOfRange(buffer.array(), start + buffer.arrayOffset(), start + length + buffer.arrayOffset()); + return Arrays.copyOfRange(buffer.array(), boff, boff + length); } // else, DirectByteBuffer.get() is the fastest route byte[] bytes = new byte[length]; @@ -297,9 +297,8 @@ public class ByteBufferUtil throw new IndexOutOfBoundsException(); for (int i = 0; i < length; i++) - { + // TODO: ByteBuffer.put is polymorphic, and might be slow here dst.put(dstPos++, src.get(srcPos++)); - } } } diff --git a/src/java/org/apache/cassandra/utils/HeapAllocator.java b/src/java/org/apache/cassandra/utils/HeapAllocator.java new file mode 100644 index 0000000000..2353829cc7 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/HeapAllocator.java @@ -0,0 +1,34 @@ +/** + * Copyright 2011 The Apache Software Foundation + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.utils; + +import java.nio.ByteBuffer; + +public final class HeapAllocator extends Allocator +{ + public static final HeapAllocator instance = new HeapAllocator(); + + private HeapAllocator() {} + + public ByteBuffer allocate(int size) + { + return ByteBuffer.allocate(size); + } +} diff --git a/src/java/org/apache/cassandra/utils/SlabAllocator.java b/src/java/org/apache/cassandra/utils/SlabAllocator.java new file mode 100644 index 0000000000..c463b90507 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/SlabAllocator.java @@ -0,0 +1,232 @@ +/** + * Copyright 2011 The Apache Software Foundation + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.utils; + +import java.nio.ByteBuffer; +import java.util.Collection; +import java.util.Collections; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Iterables; + +/** + * The SlabAllocator is a bump-the-pointer allocator that allocates + * large (2MB by default) regions and then doles them out to threads that request + * slices into the array. + *

+ * The purpose of this class is to combat heap fragmentation in long lived + * objects: by ensuring that all allocations with similar lifetimes + * only to large regions of contiguous memory, we ensure that large blocks + * get freed up at the same time. + *

+ * Otherwise, variable length byte arrays allocated end up + * interleaved throughout the heap, and the old generation gets progressively + * more fragmented until a stop-the-world compacting collection occurs. + */ +public class SlabAllocator extends Allocator +{ + private final static int REGION_SIZE = 2 * 1024 * 1024; + private final static int MAX_CLONED_SIZE = 256 * 1024; // bigger than this don't go in the region + + private final AtomicReference currentRegion = new AtomicReference(); + private final Collection filledRegions = new LinkedBlockingQueue(); + + /** @return Total number of bytes allocated by this allocator. */ + public long size() + { + Iterable regions = filledRegions; + if (currentRegion.get() != null) + regions = Iterables.concat(regions, Collections.singleton(currentRegion.get())); + + long total = 0; + for (Region region : regions) + total += region.size; + return total; + } + + public ByteBuffer allocate(int size) + { + assert size >= 0; + if (size == 0) + return ByteBufferUtil.EMPTY_BYTE_BUFFER; + + // satisfy large allocations directly from JVM since they don't cause fragmentation + // as badly, and fill up our regions quickly + if (size > MAX_CLONED_SIZE) + return ByteBuffer.allocate(size); + + while (true) + { + Region region = getRegion(); + + // Try to allocate from this region + ByteBuffer cloned = region.allocate(size); + if (cloned != null) + return cloned; + + // not enough space! + tryRetireRegion(region); + } + } + + /** + * Try to retire the current region if it is still region. + * Postcondition is that curRegion.get() != region + */ + private void tryRetireRegion(Region region) + { + if (currentRegion.compareAndSet(region, null)) + { + filledRegions.add(region); + } + } + + /** + * Get the current region, or, if there is no current region, allocate a new one + */ + private Region getRegion() + { + while (true) + { + // Try to get the region + Region region = currentRegion.get(); + if (region != null) + return region; + + // No current region, so we want to allocate one. We race + // against other allocators to CAS in an uninitialized region + // (which is cheap to allocate) + region = new Region(REGION_SIZE); + if (currentRegion.compareAndSet(null, region)) + { + // we won race - now we need to actually do the expensive allocation step + region.init(); + return region; + } + // someone else won race - that's fine, we'll try to grab theirs + // in the next iteration of the loop. + } + } + + /** + * A region of memory out of which allocations are sliced. + * + * This serves two purposes: + * - to provide a step between initialization and allocation, so that racing to CAS a + * new region in is harmless + * - encapsulates the allocation offset + */ + private static class Region + { + /** + * Actual underlying data + */ + private ByteBuffer data; + + private static final int UNINITIALIZED = -1; + /** + * Offset for the next allocation, or the sentinel value -1 + * which implies that the region is still uninitialized. + */ + private AtomicInteger nextFreeOffset = new AtomicInteger(UNINITIALIZED); + + /** + * Total number of allocations satisfied from this buffer + */ + private AtomicInteger allocCount = new AtomicInteger(); + + /** + * Size of region in bytes + */ + private final int size; + + /** + * Create an uninitialized region. Note that memory is not allocated yet, so + * this is cheap. + * + * @param size in bytes + */ + private Region(int size) + { + this.size = size; + } + + /** + * Actually claim the memory for this region. This should only be called from + * the thread that constructed the region. It is thread-safe against other + * threads calling alloc(), who will block until the allocation is complete. + */ + public void init() + { + assert nextFreeOffset.get() == UNINITIALIZED; + data = ByteBuffer.allocate(size); + assert data.remaining() == data.capacity(); + // Mark that it's ready for use + boolean initted = nextFreeOffset.compareAndSet(UNINITIALIZED, 0); + // We should always succeed the above CAS since only one thread calls init()! + Preconditions.checkState(initted, "Multiple threads tried to init same region"); + } + + /** + * Try to allocate size bytes from the region. + * + * @return the successful allocation, or null to indicate not-enough-space + */ + public ByteBuffer allocate(int size) + { + while (true) + { + int oldOffset = nextFreeOffset.get(); + if (oldOffset == UNINITIALIZED) + { + // The region doesn't have its data allocated yet. + // Since we found this in currentRegion, we know that whoever + // CAS-ed it there is allocating it right now. So spin-loop + // shouldn't spin long! + Thread.yield(); + continue; + } + + if (oldOffset + size > data.capacity()) // capacity == remaining + return null; + + // Try to atomically claim this region + if (nextFreeOffset.compareAndSet(oldOffset, oldOffset + size)) + { + // we got the alloc + allocCount.incrementAndGet(); + return (ByteBuffer) data.duplicate().position(oldOffset).limit(oldOffset + size); + } + // we raced and lost alloc, try again + } + } + + @Override + public String toString() + { + return "Region@" + System.identityHashCode(this) + + " allocs=" + allocCount.get() + "waste=" + + (data.capacity() - nextFreeOffset.get()); + } + } +} diff --git a/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java b/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java index 1121b91774..0a546d201c 100644 --- a/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java +++ b/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java @@ -1,21 +1,15 @@ package org.apache.cassandra.db; -import java.io.ByteArrayInputStream; -import java.io.DataInputStream; -import java.io.IOException; import java.nio.ByteBuffer; import java.util.*; -import org.apache.cassandra.SchemaLoader; import org.junit.Test; -import org.apache.cassandra.io.util.DataOutputBuffer; -import org.apache.cassandra.db.filter.QueryPath; -import static org.apache.cassandra.Util.column; import static org.junit.Assert.*; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.utils.HeapAllocator; public class ArrayBackedSortedColumnsTest { @@ -32,7 +26,7 @@ public class ArrayBackedSortedColumnsTest int[] values = new int[]{ 1, 2, 2, 3 }; for (int i = 0; i < values.length; ++i) - map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i]))); + map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i])), HeapAllocator.instance); Iterator iter = map.iterator(); assertEquals("1st column", 1, iter.next().name().getInt(0)); @@ -56,12 +50,12 @@ public class ArrayBackedSortedColumnsTest int[] values2 = new int[]{ 2, 4, 5, 6 }; for (int i = 0; i < values1.length; ++i) - map.addColumn(new Column(ByteBufferUtil.bytes(values1[reversed ? values1.length - 1 - i : i]))); + map.addColumn(new Column(ByteBufferUtil.bytes(values1[reversed ? values1.length - 1 - i : i])), HeapAllocator.instance); for (int i = 0; i < values2.length; ++i) - map2.addColumn(new Column(ByteBufferUtil.bytes(values2[reversed ? values2.length - 1 - i : i]))); + map2.addColumn(new Column(ByteBufferUtil.bytes(values2[reversed ? values2.length - 1 - i : i])), HeapAllocator.instance); - map2.addAll(map); + map2.addAll(map, HeapAllocator.instance); Iterator iter = map2.iterator(); assertEquals("1st column", 1, iter.next().name().getInt(0)); @@ -91,7 +85,7 @@ public class ArrayBackedSortedColumnsTest Collections.reverse(reverseSorted); for (int i = 0; i < values.length; ++i) - map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i]))); + map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i])), HeapAllocator.instance); assertSame(sorted, map.getSortedColumns()); assertSame(reverseSorted, map.getReverseSortedColumns()); @@ -113,7 +107,7 @@ public class ArrayBackedSortedColumnsTest names.add(ByteBufferUtil.bytes(v)); for (int i = 0; i < values.length; ++i) - map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i]))); + map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i])), HeapAllocator.instance); assertSame(names, map.getColumnNames()); } diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyTest.java index 95d6b09ba4..08149c7e96 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyTest.java @@ -31,6 +31,7 @@ import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.db.filter.QueryPath; import static org.apache.cassandra.Util.column; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.HeapAllocator; public class ColumnFamilyTest extends SchemaLoader @@ -126,8 +127,8 @@ public class ColumnFamilyTest extends SchemaLoader cf_old.addColumn(QueryPath.column(ByteBufferUtil.bytes("col2")), val2, 1); cf_old.addColumn(QueryPath.column(ByteBufferUtil.bytes("col3")), val2, 2); - cf_result.addAll(cf_new); - cf_result.addAll(cf_old); + cf_result.addAll(cf_new, HeapAllocator.instance); + cf_result.addAll(cf_old, HeapAllocator.instance); assert 3 == cf_result.getColumnCount() : "Count is " + cf_new.getColumnCount(); //addcolumns will only add if timestamp >= old timestamp diff --git a/test/unit/org/apache/cassandra/db/CounterColumnTest.java b/test/unit/org/apache/cassandra/db/CounterColumnTest.java index 1b37675f5d..355482ab0d 100644 --- a/test/unit/org/apache/cassandra/db/CounterColumnTest.java +++ b/test/unit/org/apache/cassandra/db/CounterColumnTest.java @@ -40,6 +40,8 @@ import org.apache.cassandra.db.context.CounterContext; import static org.apache.cassandra.db.context.CounterContext.ContextState; import org.apache.cassandra.db.marshal.*; import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.utils.Allocator; +import org.apache.cassandra.utils.HeapAllocator; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.NodeId; @@ -205,6 +207,7 @@ public class CounterColumnTest extends SchemaLoader @Test public void testDiff() throws UnknownHostException { + Allocator allocator = HeapAllocator.instance; ContextState left; ContextState right; @@ -226,7 +229,7 @@ public class CounterColumnTest extends SchemaLoader assert null == rightCol.diff(leftCol); // equality: equal nodes, all counts same - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 3L, 0L); left.writeElement(NodeId.fromInt(6), 2L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); @@ -237,13 +240,13 @@ public class CounterColumnTest extends SchemaLoader assert null == leftCol.diff(rightCol); // greater than: left has superset of nodes (counts equal) - left = ContextState.allocate(4, 0); + left = ContextState.allocate(4, 0, allocator); left.writeElement(NodeId.fromInt(3), 3L, 0L); left.writeElement(NodeId.fromInt(6), 2L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); left.writeElement(NodeId.fromInt(12), 0L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 3L, 0L); right.writeElement(NodeId.fromInt(6), 2L, 0L); right.writeElement(NodeId.fromInt(9), 1L, 0L); @@ -256,12 +259,12 @@ public class CounterColumnTest extends SchemaLoader assert leftCol == rightCol.diff(leftCol); // disjoint: right and left have disjoint node sets - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 1L, 0L); left.writeElement(NodeId.fromInt(4), 1L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 1L, 0L); right.writeElement(NodeId.fromInt(6), 1L, 0L); right.writeElement(NodeId.fromInt(9), 1L, 0L); @@ -275,7 +278,8 @@ public class CounterColumnTest extends SchemaLoader @Test public void testSerializeDeserialize() throws IOException { - CounterContext.ContextState state = CounterContext.ContextState.allocate(4, 2); + Allocator allocator = HeapAllocator.instance; + CounterContext.ContextState state = CounterContext.ContextState.allocate(4, 2, allocator); state.writeElement(NodeId.fromInt(1), 4L, 4L); state.writeElement(NodeId.fromInt(2), 4L, 4L, true); state.writeElement(NodeId.fromInt(3), 4L, 4L); @@ -291,7 +295,7 @@ public class CounterColumnTest extends SchemaLoader assert original.equals(deserialized); bufIn = new ByteArrayInputStream(serialized, 0, serialized.length); - CounterColumn deserializedOnRemote = (CounterColumn)Column.serializer().deserialize(new DataInputStream(bufIn), null, true); + CounterColumn deserializedOnRemote = (CounterColumn)Column.serializer().deserialize(new DataInputStream(bufIn), true); assert deserializedOnRemote.name().equals(original.name()); assert deserializedOnRemote.total() == original.total(); assert deserializedOnRemote.value().equals(cc.clearAllDelta(original.value())); @@ -302,10 +306,11 @@ public class CounterColumnTest extends SchemaLoader @Test public void testUpdateDigest() throws Exception { + Allocator allocator = HeapAllocator.instance; MessageDigest digest1 = MessageDigest.getInstance("md5"); MessageDigest digest2 = MessageDigest.getInstance("md5"); - CounterContext.ContextState state = CounterContext.ContextState.allocate(4, 2); + CounterContext.ContextState state = CounterContext.ContextState.allocate(4, 2, allocator); state.writeElement(NodeId.fromInt(1), 4L, 4L); state.writeElement(NodeId.fromInt(2), 4L, 4L, true); state.writeElement(NodeId.fromInt(3), 4L, 4L); diff --git a/test/unit/org/apache/cassandra/db/CounterMutationTest.java b/test/unit/org/apache/cassandra/db/CounterMutationTest.java index 70e12dfae9..15b638d608 100644 --- a/test/unit/org/apache/cassandra/db/CounterMutationTest.java +++ b/test/unit/org/apache/cassandra/db/CounterMutationTest.java @@ -164,7 +164,7 @@ public class CounterMutationTest extends CleanupHelper } // Check that if we merge old and clean on another node, we keep the right count - ByteBuffer onRemote = ctx.merge(ctx.clearAllDelta(state.context), ctx.clearAllDelta(cleaned)); + ByteBuffer onRemote = ctx.merge(ctx.clearAllDelta(state.context), ctx.clearAllDelta(cleaned), HeapAllocator.instance); assert ctx.total(onRemote) == 11; } } diff --git a/test/unit/org/apache/cassandra/db/ReadMessageTest.java b/test/unit/org/apache/cassandra/db/ReadMessageTest.java index a1edc86149..d5b1066477 100644 --- a/test/unit/org/apache/cassandra/db/ReadMessageTest.java +++ b/test/unit/org/apache/cassandra/db/ReadMessageTest.java @@ -95,7 +95,7 @@ public class ReadMessageTest extends SchemaLoader ReadCommand command = new SliceByNamesReadCommand("Keyspace1", dk.key, new QueryPath("Standard1"), Arrays.asList(ByteBufferUtil.bytes("Column1"))); Row row = command.getRow(table); IColumn col = row.cf.getColumn(ByteBufferUtil.bytes("Column1")); - assert Arrays.equals(col.value().array(), "abcd".getBytes()); + assertEquals(col.value(), ByteBuffer.wrap("abcd".getBytes())); } @Test diff --git a/test/unit/org/apache/cassandra/db/TimeSortTest.java b/test/unit/org/apache/cassandra/db/TimeSortTest.java index ab06755173..59bdabf3e7 100644 --- a/test/unit/org/apache/cassandra/db/TimeSortTest.java +++ b/test/unit/org/apache/cassandra/db/TimeSortTest.java @@ -110,8 +110,8 @@ public class TimeSortTest extends CleanupHelper columnNames.add(getBytes(10)); columnNames.add(getBytes(0)); cf = cfStore.getColumnFamily(QueryFilter.getNamesFilter(Util.dk("900"), new QueryPath("StandardLong1"), columnNames)); - assert "c".equals(new String(cf.getColumn(getBytes(0)).value().array())); - assert "c".equals(new String(cf.getColumn(getBytes(10)).value().array())); + assert "c".equals(ByteBufferUtil.string(cf.getColumn(getBytes(0)).value())); + assert "c".equals(ByteBufferUtil.string(cf.getColumn(getBytes(10)).value())); } private void validateTimeSort(Table table) throws IOException diff --git a/test/unit/org/apache/cassandra/db/context/CounterContextTest.java b/test/unit/org/apache/cassandra/db/context/CounterContextTest.java index dc48398ad7..9ebc3677b9 100644 --- a/test/unit/org/apache/cassandra/db/context/CounterContextTest.java +++ b/test/unit/org/apache/cassandra/db/context/CounterContextTest.java @@ -25,17 +25,15 @@ import static org.junit.Assert.*; import java.net.InetAddress; import java.net.UnknownHostException; import java.nio.ByteBuffer; -import java.util.*; - -import org.apache.commons.lang.ArrayUtils; +import java.util.ArrayList; +import java.util.List; import org.junit.Test; - import org.apache.cassandra.Util; + import org.apache.cassandra.db.context.IContext.ContextRelationship; import static org.apache.cassandra.db.context.CounterContext.ContextState; -import org.apache.cassandra.utils.ByteBufferUtil; -import org.apache.cassandra.utils.NodeId; +import org.apache.cassandra.utils.*; public class CounterContextTest { @@ -55,17 +53,37 @@ public class CounterContextTest stepLength = idLength + clockLength + countLength; } + /** Allocates 1 byte from a new SlabAllocator and returns it. */ + private Allocator bumpedSlab() + { + SlabAllocator allocator = new SlabAllocator(); + allocator.allocate(1); + return allocator; + } + @Test public void testCreate() { - ByteBuffer context = cc.create(4); - assert context.remaining() == stepLength + 4; + runCreate(HeapAllocator.instance); + runCreate(bumpedSlab()); + } + + private void runCreate(Allocator allocator) + { + ByteBuffer bytes = cc.create(4, allocator); + assertEquals(stepLength + 4, bytes.remaining()); } @Test public void testDiff() { - ContextState left = ContextState.allocate(3, 0); + runDiff(HeapAllocator.instance); + runDiff(bumpedSlab()); + } + + private void runDiff(Allocator allocator) + { + ContextState left = ContextState.allocate(3, 0, allocator); ContextState right; // equality: equal nodes, all counts same @@ -78,13 +96,13 @@ public class CounterContextTest cc.diff(left.context, right.context); // greater than: left has superset of nodes (counts equal) - left = ContextState.allocate(4, 0); + left = ContextState.allocate(4, 0, allocator); left.writeElement(NodeId.fromInt(3), 3L, 0L); left.writeElement(NodeId.fromInt(6), 2L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); left.writeElement(NodeId.fromInt(12), 0L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 3L, 0L); right.writeElement(NodeId.fromInt(6), 2L, 0L); right.writeElement(NodeId.fromInt(9), 1L, 0L); @@ -93,12 +111,12 @@ public class CounterContextTest cc.diff(left.context, right.context); // less than: left has subset of nodes (counts equal) - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 3L, 0L); left.writeElement(NodeId.fromInt(6), 2L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); - right = ContextState.allocate(4, 0); + right = ContextState.allocate(4, 0, allocator); right.writeElement(NodeId.fromInt(3), 3L, 0L); right.writeElement(NodeId.fromInt(6), 2L, 0L); right.writeElement(NodeId.fromInt(9), 1L, 0L); @@ -108,12 +126,12 @@ public class CounterContextTest cc.diff(left.context, right.context); // greater than: equal nodes, but left has higher counts - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 3L, 0L); left.writeElement(NodeId.fromInt(6), 2L, 0L); left.writeElement(NodeId.fromInt(9), 3L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 3L, 0L); right.writeElement(NodeId.fromInt(6), 2L, 0L); right.writeElement(NodeId.fromInt(9), 1L, 0L); @@ -122,12 +140,12 @@ public class CounterContextTest cc.diff(left.context, right.context); // less than: equal nodes, but right has higher counts - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 3L, 0L); left.writeElement(NodeId.fromInt(6), 2L, 0L); left.writeElement(NodeId.fromInt(9), 3L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 3L, 0L); right.writeElement(NodeId.fromInt(6), 9L, 0L); right.writeElement(NodeId.fromInt(9), 3L, 0L); @@ -136,12 +154,12 @@ public class CounterContextTest cc.diff(left.context, right.context); // disjoint: right and left have disjoint node sets - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 1L, 0L); left.writeElement(NodeId.fromInt(4), 1L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 1L, 0L); right.writeElement(NodeId.fromInt(6), 1L, 0L); right.writeElement(NodeId.fromInt(9), 1L, 0L); @@ -149,12 +167,12 @@ public class CounterContextTest assert ContextRelationship.DISJOINT == cc.diff(left.context, right.context); - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 1L, 0L); left.writeElement(NodeId.fromInt(4), 1L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(2), 1L, 0L); right.writeElement(NodeId.fromInt(6), 1L, 0L); right.writeElement(NodeId.fromInt(12), 1L, 0L); @@ -163,12 +181,12 @@ public class CounterContextTest cc.diff(left.context, right.context); // disjoint: equal nodes, but right and left have higher counts in differing nodes - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 1L, 0L); left.writeElement(NodeId.fromInt(6), 3L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 1L, 0L); right.writeElement(NodeId.fromInt(6), 1L, 0L); right.writeElement(NodeId.fromInt(9), 5L, 0L); @@ -176,12 +194,12 @@ public class CounterContextTest assert ContextRelationship.DISJOINT == cc.diff(left.context, right.context); - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 2L, 0L); left.writeElement(NodeId.fromInt(6), 3L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 1L, 0L); right.writeElement(NodeId.fromInt(6), 9L, 0L); right.writeElement(NodeId.fromInt(9), 5L, 0L); @@ -190,13 +208,13 @@ public class CounterContextTest cc.diff(left.context, right.context); // disjoint: left has more nodes, but lower counts - left = ContextState.allocate(4, 0); + left = ContextState.allocate(4, 0, allocator); left.writeElement(NodeId.fromInt(3), 2L, 0L); left.writeElement(NodeId.fromInt(6), 3L, 0L); left.writeElement(NodeId.fromInt(9), 1L, 0L); left.writeElement(NodeId.fromInt(12), 1L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 4L, 0L); right.writeElement(NodeId.fromInt(6), 9L, 0L); right.writeElement(NodeId.fromInt(9), 5L, 0L); @@ -205,12 +223,12 @@ public class CounterContextTest cc.diff(left.context, right.context); // disjoint: left has less nodes, but higher counts - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 5L, 0L); left.writeElement(NodeId.fromInt(6), 3L, 0L); left.writeElement(NodeId.fromInt(9), 2L, 0L); - right = ContextState.allocate(4, 0); + right = ContextState.allocate(4, 0, allocator); right.writeElement(NodeId.fromInt(3), 4L, 0L); right.writeElement(NodeId.fromInt(6), 3L, 0L); right.writeElement(NodeId.fromInt(9), 2L, 0L); @@ -220,12 +238,12 @@ public class CounterContextTest cc.diff(left.context, right.context); // disjoint: mixed nodes and counts - left = ContextState.allocate(3, 0); + left = ContextState.allocate(3, 0, allocator); left.writeElement(NodeId.fromInt(3), 5L, 0L); left.writeElement(NodeId.fromInt(6), 2L, 0L); left.writeElement(NodeId.fromInt(9), 2L, 0L); - right = ContextState.allocate(4, 0); + right = ContextState.allocate(4, 0, allocator); right.writeElement(NodeId.fromInt(3), 4L, 0L); right.writeElement(NodeId.fromInt(6), 3L, 0L); right.writeElement(NodeId.fromInt(9), 2L, 0L); @@ -234,13 +252,13 @@ public class CounterContextTest assert ContextRelationship.DISJOINT == cc.diff(left.context, right.context); - left = ContextState.allocate(4, 0); + left = ContextState.allocate(4, 0, allocator); left.writeElement(NodeId.fromInt(3), 5L, 0L); left.writeElement(NodeId.fromInt(6), 2L, 0L); left.writeElement(NodeId.fromInt(7), 2L, 0L); left.writeElement(NodeId.fromInt(9), 2L, 0L); - right = ContextState.allocate(3, 0); + right = ContextState.allocate(3, 0, allocator); right.writeElement(NodeId.fromInt(3), 4L, 0L); right.writeElement(NodeId.fromInt(6), 3L, 0L); right.writeElement(NodeId.fromInt(9), 2L, 0L); @@ -251,61 +269,73 @@ public class CounterContextTest @Test public void testMerge() + { + runMerge(HeapAllocator.instance); + runMerge(bumpedSlab()); + } + + private void runMerge(Allocator allocator) { // note: local counts aggregated; remote counts are reconciled (i.e. take max) - ContextState left = ContextState.allocate(4, 1); + ContextState left = ContextState.allocate(4, 1, allocator); left.writeElement(NodeId.fromInt(1), 1L, 1L); left.writeElement(NodeId.fromInt(2), 2L, 2L); left.writeElement(NodeId.fromInt(4), 6L, 3L); left.writeElement(NodeId.getLocalId(), 7L, 3L, true); - ContextState right = ContextState.allocate(3, 1); + ContextState right = ContextState.allocate(3, 1, allocator); right.writeElement(NodeId.fromInt(4), 4L, 4L); right.writeElement(NodeId.fromInt(5), 5L, 5L); right.writeElement(NodeId.getLocalId(), 2L, 9L, true); - ByteBuffer merged = cc.merge(left.context, right.context); + ByteBuffer merged = cc.merge(left.context, right.context, allocator); int hd = 4; assertEquals(hd + 5 * stepLength, merged.remaining()); // local node id's counts are aggregated assert Util.equalsNodeId(NodeId.getLocalId(), merged, hd + 4*stepLength); - assertEquals( 9L, merged.getLong(hd + 4*stepLength + idLength)); - assertEquals(12L, merged.getLong(hd + 4*stepLength + idLength + clockLength)); + assertEquals( 9L, merged.getLong(merged.position() + hd + 4*stepLength + idLength)); + assertEquals(12L, merged.getLong(merged.position() + hd + 4*stepLength + idLength + clockLength)); // remote node id counts are reconciled (i.e. take max) assert Util.equalsNodeId(NodeId.fromInt(4), merged, hd + 2*stepLength); - assertEquals( 6L, merged.getLong(hd + 2*stepLength + idLength)); - assertEquals( 3L, merged.getLong(hd + 2*stepLength + idLength + clockLength)); + assertEquals( 6L, merged.getLong(merged.position() + hd + 2*stepLength + idLength)); + assertEquals( 3L, merged.getLong(merged.position() + hd + 2*stepLength + idLength + clockLength)); assert Util.equalsNodeId(NodeId.fromInt(5), merged, hd + 3*stepLength); - assertEquals( 5L, merged.getLong(hd + 3*stepLength + idLength)); - assertEquals( 5L, merged.getLong(hd + 3*stepLength + idLength + clockLength)); + assertEquals( 5L, merged.getLong(merged.position() + hd + 3*stepLength + idLength)); + assertEquals( 5L, merged.getLong(merged.position() + hd + 3*stepLength + idLength + clockLength)); assert Util.equalsNodeId(NodeId.fromInt(2), merged, hd + 1*stepLength); - assertEquals( 2L, merged.getLong(hd + 1*stepLength + idLength)); - assertEquals( 2L, merged.getLong(hd + 1*stepLength + idLength + clockLength)); + assertEquals( 2L, merged.getLong(merged.position() + hd + 1*stepLength + idLength)); + assertEquals( 2L, merged.getLong(merged.position() + hd + 1*stepLength + idLength + clockLength)); assert Util.equalsNodeId(NodeId.fromInt(1), merged, hd + 0*stepLength); - assertEquals( 1L, merged.getLong(hd + 0*stepLength + idLength)); - assertEquals( 1L, merged.getLong(hd + 0*stepLength + idLength + clockLength)); + assertEquals( 1L, merged.getLong(merged.position() + hd + 0*stepLength + idLength)); + assertEquals( 1L, merged.getLong(merged.position() + hd + 0*stepLength + idLength + clockLength)); } @Test public void testTotal() { - ContextState left = ContextState.allocate(4, 1); + runTotal(HeapAllocator.instance); + runTotal(bumpedSlab()); + } + + private void runTotal(Allocator allocator) + { + ContextState left = ContextState.allocate(4, 1, allocator); left.writeElement(NodeId.fromInt(1), 1L, 1L); left.writeElement(NodeId.fromInt(2), 2L, 2L); left.writeElement(NodeId.fromInt(4), 3L, 3L); left.writeElement(NodeId.getLocalId(), 3L, 3L, true); - ContextState right = ContextState.allocate(3, 1); + ContextState right = ContextState.allocate(3, 1, allocator); right.writeElement(NodeId.fromInt(4), 4L, 4L); right.writeElement(NodeId.fromInt(5), 5L, 5L); right.writeElement(NodeId.getLocalId(), 9L, 9L, true); - ByteBuffer merged = cc.merge(left.context, right.context); + ByteBuffer merged = cc.merge(left.context, right.context, allocator); // 127.0.0.1: 12 (3+9) // 0.0.0.1: 1 @@ -318,6 +348,12 @@ public class CounterContextTest @Test public void testMergeOldShards() + { + runMergeOldShards(HeapAllocator.instance); + runMergeOldShards(bumpedSlab()); + } + + private void runMergeOldShards(Allocator allocator) { long now = System.currentTimeMillis(); NodeId id1 = NodeId.fromInt(1); @@ -327,7 +363,7 @@ public class CounterContextTest records.add(new NodeId.NodeIdRecord(id3, 4L)); // Destination of merge is a delta - ContextState ctx = ContextState.allocate(5, 2); + ContextState ctx = ContextState.allocate(5, 2, allocator); ctx.writeElement(id1, 1L, 1L); ctx.writeElement(NodeId.fromInt(2), 2L, 2L); ctx.writeElement(id3, 3L, 3L, true); @@ -344,41 +380,47 @@ public class CounterContextTest assert m.getNodeId().equals(id3); assert m.getClock() == 4L; assert m.getCount() == 1L; - assert cc.total(ctx.context) == cc.total(cc.merge(ctx.context, merger)); + assert cc.total(ctx.context) == cc.total(cc.merge(ctx.context, merger, allocator)); // Source of merge is a delta - ctx = ContextState.allocate(4, 1); + ctx = ContextState.allocate(4, 1, allocator); ctx.writeElement(id1, 1L, 1L, true); ctx.writeElement(NodeId.fromInt(2), 2L, 2L); ctx.writeElement(id3, 3L, 3L); ctx.writeElement(NodeId.fromInt(4), 6L, 3L); merger = cc.computeOldShardMerger(ctx.context, records); - assert cc.total(ctx.context) == cc.total(cc.merge(ctx.context, merger)); + assert cc.total(ctx.context) == cc.total(cc.merge(ctx.context, merger, allocator)); // source and destination of merge are deltas - ctx = ContextState.allocate(4, 2); + ctx = ContextState.allocate(4, 2, allocator); ctx.writeElement(id1, 1L, 1L, true); ctx.writeElement(NodeId.fromInt(2), 2L, 2L); ctx.writeElement(id3, 3L, 3L, true); ctx.writeElement(NodeId.fromInt(4), 6L, 3L); merger = cc.computeOldShardMerger(ctx.context, records); - assert cc.total(ctx.context) == cc.total(cc.merge(ctx.context, merger)); + assert cc.total(ctx.context) == cc.total(cc.merge(ctx.context, merger, allocator)); // none of source and destination of merge are deltas - ctx = ContextState.allocate(4, 0); + ctx = ContextState.allocate(4, 0, allocator); ctx.writeElement(id1, 1L, 1L); ctx.writeElement(NodeId.fromInt(2), 2L, 2L); ctx.writeElement(id3, 3L, 3L); ctx.writeElement(NodeId.fromInt(4), 6L, 3L); merger = cc.computeOldShardMerger(ctx.context, records); - assert cc.total(ctx.context) == cc.total(cc.merge(ctx.context, merger)); + assert cc.total(ctx.context) == cc.total(cc.merge(ctx.context, merger, allocator)); } @Test public void testRemoveOldShards() + { + runRemoveOldShards(HeapAllocator.instance); + runRemoveOldShards(bumpedSlab()); + } + + private void runRemoveOldShards(Allocator allocator) { NodeId id1 = NodeId.fromInt(1); NodeId id3 = NodeId.fromInt(3); @@ -388,7 +430,7 @@ public class CounterContextTest records.add(new NodeId.NodeIdRecord(id3, 4L)); records.add(new NodeId.NodeIdRecord(id6, 10L)); - ContextState ctx = ContextState.allocate(6, 2); + ContextState ctx = ContextState.allocate(6, 2, allocator); ctx.writeElement(id1, 1L, 1L); ctx.writeElement(NodeId.fromInt(2), 2L, 2L); ctx.writeElement(id3, 3L, 3L, true); @@ -397,7 +439,7 @@ public class CounterContextTest ctx.writeElement(id6, 5L, 6L); ByteBuffer merger = cc.computeOldShardMerger(ctx.context, records); - ByteBuffer merged = cc.merge(ctx.context, merger); + ByteBuffer merged = cc.merge(ctx.context, merger, allocator); assert cc.total(ctx.context) == cc.total(merged); ByteBuffer cleaned = cc.removeOldShards(merged, (int)(System.currentTimeMillis() / 1000) + 1); @@ -405,12 +447,11 @@ public class CounterContextTest assert cleaned.remaining() == ctx.context.remaining() - stepLength; merger = cc.computeOldShardMerger(cleaned, records); - merged = cc.merge(cleaned, merger); + merged = cc.merge(cleaned, merger, allocator); assert cc.total(ctx.context) == cc.total(merged); cleaned = cc.removeOldShards(merged, (int)(System.currentTimeMillis() / 1000) + 1); assert cc.total(ctx.context) == cc.total(cleaned); assert cleaned.remaining() == ctx.context.remaining() - 2 * stepLength - 2; - } }