natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve write path performance 

Patch by tjake; reviewed by Stefania Alborghetti for CASSANDRA-12269
This commit is contained in:
T Jake Luciani 2016-06-21 21:53:43 -04:00
parent 2c0edce094
commit dc9ed46341
35 changed files with 826 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.10
* Faster write path (CASSANDRA-12269)
* Option to leave omitted columns in INSERT JSON unset (CASSANDRA-11424)
* Support json/yaml output in nodetool tpstats (CASSANDRA-12035)
* Expose metrics for successful/failed authentication attempts (CASSANDRA-10635)

1
conf/jvm.options
View File

@ -118,6 +118,7 @@
# resize them at runtime.
-XX:+UseTLAB
-XX:+ResizeTLAB
-XX:+UseNUMA
# http://www.evanjones.ca/jvm-mmap-pause.html
-XX:+PerfDisableSharedMem

18
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -47,6 +47,7 @@ import org.apache.cassandra.cql3.statements.CFStatement;
import org.apache.cassandra.cql3.statements.CreateTableStatement;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.exceptions.ConfigurationException;
@ -122,6 +123,9 @@ public final class CFMetaData
public final DataResource resource;
//For hot path serialization it's often easier to store this info here
private volatile ColumnFilter allColumnFilter;
/*
* All of these methods will go away once CFMetaData becomes completely immutable.
*/
@ -294,9 +298,12 @@ public final class CFMetaData
this.clusteringColumns = clusteringColumns;
this.partitionColumns = partitionColumns;
this.serializers = new Serializers(this);
this.resource = DataResource.table(ksName, cfName);
//This needs to happen before serializers are set
//because they use comparator.subtypes()
rebuild();
this.resource = DataResource.table(ksName, cfName);
this.serializers = new Serializers(this);
}
// This rebuild informations that are intrinsically duplicate of the table definition but
@ -323,6 +330,8 @@ public final class CFMetaData
if (isCompactTable())
this.compactValueColumn = CompactTables.getCompactValueColumn(partitionColumns, isSuper());
this.allColumnFilter = ColumnFilter.all(this);
}
public Indexes getIndexes()
@ -330,6 +339,11 @@ public final class CFMetaData
return indexes;
}
public ColumnFilter getAllColumnFilter()
{
return allColumnFilter;
}
public static CFMetaData create(String ksName,
String name,
UUID cfId,

11
src/java/org/apache/cassandra/db/Columns.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.db;
import java.io.IOException;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
@ -366,6 +367,16 @@ public class Columns extends AbstractCollection<ColumnDefinition> implements Col
digest.update(c.name.bytes.duplicate());
}
/**
* Apply a function to each column definition in forwards or reversed order.
* @param function
* @param reversed
*/
public void apply(Consumer<ColumnDefinition> function, boolean reversed)
{
BTree.apply(columns, function, reversed);
}
@Override
public boolean equals(Object other)
{

12
src/java/org/apache/cassandra/db/SerializationHeader.java
View File

@ -21,9 +21,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.db.filter.ColumnFilter;
@ -108,17 +105,12 @@ public class SerializationHeader
{
this(isForSSTable,
metadata.getKeyValidator(),
typesOf(metadata.clusteringColumns()),
metadata.comparator.subtypes(),
columns,
stats,
null);
}
private static List<AbstractType<?>> typesOf(List<ColumnDefinition> columns)
{
return ImmutableList.copyOf(Lists.transform(columns, column -> column.type));
}
public PartitionColumns columns()
{
return columns;
@ -398,7 +390,7 @@ public class SerializationHeader
EncodingStats stats = EncodingStats.serializer.deserialize(in);
AbstractType<?> keyType = metadata.getKeyValidator();
List<AbstractType<?>> clusteringTypes = typesOf(metadata.clusteringColumns());
List<AbstractType<?>> clusteringTypes = metadata.comparator.subtypes();
Columns statics, regulars;
if (selection == null)

2
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -652,7 +652,7 @@ public final class SystemKeyspace
private static Map<UUID, ByteBuffer> truncationAsMapEntry(ColumnFamilyStore cfs, long truncatedAt, CommitLogPosition position)
{
DataOutputBuffer out = null;
try (DataOutputBuffer ignored = out = DataOutputBuffer.RECYCLER.get())
try (DataOutputBuffer ignored = out = DataOutputBuffer.scratchBuffer.get())
{
CommitLogPosition.serializer.serialize(position, out);
out.writeLong(truncatedAt);

66
src/java/org/apache/cassandra/db/commitlog/CommitLog.java
View File

@ -38,6 +38,7 @@ import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.compress.ICompressor;
import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputBufferFixed;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.metrics.CommitLogMetrics;
@ -243,43 +244,56 @@ public class CommitLog implements CommitLogMBean
{
assert mutation != null;
int size = (int) Mutation.serializer.serializedSize(mutation, MessagingService.current_version);
int totalSize = size + ENTRY_OVERHEAD_SIZE;
if (totalSize > MAX_MUTATION_SIZE)
DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get();
try
{
throw new IllegalArgumentException(String.format("Mutation of %s is too large for the maximum size of %s",
FBUtilities.prettyPrintMemory(totalSize),
FBUtilities.prettyPrintMemory(MAX_MUTATION_SIZE)));
}
Mutation.serializer.serialize(mutation, dob, MessagingService.current_version);
int size = dob.getLength();
Allocation alloc = segmentManager.allocate(mutation, totalSize);
int totalSize = size + ENTRY_OVERHEAD_SIZE;
if (totalSize > MAX_MUTATION_SIZE)
{
throw new IllegalArgumentException(String.format("Mutation of %s is too large for the maximum size of %s",
FBUtilities.prettyPrintMemory(totalSize),
FBUtilities.prettyPrintMemory(MAX_MUTATION_SIZE)));
}
CRC32 checksum = new CRC32();
final ByteBuffer buffer = alloc.getBuffer();
try (BufferedDataOutputStreamPlus dos = new DataOutputBufferFixed(buffer))
{
// checksummed length
dos.writeInt(size);
updateChecksumInt(checksum, size);
buffer.putInt((int) checksum.getValue());
Allocation alloc = segmentManager.allocate(mutation, totalSize);
// checksummed mutation
Mutation.serializer.serialize(mutation, dos, MessagingService.current_version);
updateChecksum(checksum, buffer, buffer.position() - size, size);
buffer.putInt((int) checksum.getValue());
CRC32 checksum = new CRC32();
final ByteBuffer buffer = alloc.getBuffer();
try (BufferedDataOutputStreamPlus dos = new DataOutputBufferFixed(buffer))
{
// checksummed length
dos.writeInt(size);
updateChecksumInt(checksum, size);
buffer.putInt((int) checksum.getValue());
// checksummed mutation
dos.write(dob.getData(), 0, size);
updateChecksum(checksum, buffer, buffer.position() - size, size);
buffer.putInt((int) checksum.getValue());
}
catch (IOException e)
{
throw new FSWriteError(e, alloc.getSegment().getPath());
}
finally
{
alloc.markWritten();
}
executor.finishWriteFor(alloc);
return alloc.getCommitLogPosition();
}
catch (IOException e)
{
throw new FSWriteError(e, alloc.getSegment().getPath());
throw new FSWriteError(e, segmentManager.allocatingFrom().getPath());
}
finally
{
alloc.markWritten();
dob.recycle();
}
executor.finishWriteFor(alloc);
return alloc.getCommitLogPosition();
}
/**

3
src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java
View File

@ -169,7 +169,7 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row>
public UnfilteredRowIterator unfilteredIterator()
{
return unfilteredIterator(ColumnFilter.all(metadata()), Slices.ALL, false);
return unfilteredIterator(metadata().getAllColumnFilter(), Slices.ALL, false);
}
public UnfilteredRowIterator unfilteredIterator(ColumnFilter selection, Slices slices, boolean reversed)
@ -197,7 +197,6 @@ public abstract class AbstractBTreePartition implements Partition, Iterable<Row>
ClusteringBound end = slice.end() == ClusteringBound.TOP ? null : slice.end();
Iterator<Row> rowIter = BTree.slice(current.tree, metadata.comparator, start, true, end, true, desc(reversed));
Iterator<RangeTombstone> deleteIter = current.deletionInfo.rangeIterator(slice, reversed);
return merge(rowIter, deleteIter, selection, reversed, current, staticRow);
}

49
src/java/org/apache/cassandra/db/rows/BTreeRow.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.db.rows;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Predicate;
import com.google.common.base.Function;
@ -32,9 +33,7 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.BTreeSearchIterator;
import org.apache.cassandra.utils.btree.UpdateFunction;
@ -168,16 +167,23 @@ public class BTreeRow extends AbstractRow
return cd.column().isSimple() ? minDeletionTime((Cell) cd) : minDeletionTime((ComplexColumnData)cd);
}
public void apply(Consumer<ColumnData> function, boolean reversed)
{
BTree.apply(btree, function, reversed);
}
public void apply(Consumer<ColumnData> funtion, com.google.common.base.Predicate<ColumnData> stopCondition, boolean reversed)
{
BTree.apply(btree, funtion, stopCondition, reversed);
}
private static int minDeletionTime(Object[] btree, LivenessInfo info, DeletionTime rowDeletion)
{
int min = Math.min(minDeletionTime(info), minDeletionTime(rowDeletion));
for (ColumnData cd : BTree.<ColumnData>iterable(btree))
{
min = Math.min(min, minDeletionTime(cd));
if (min == Integer.MIN_VALUE)
break;
}
return min;
//we have to wrap this for the lambda
final WrappedInt min = new WrappedInt(Math.min(minDeletionTime(info), minDeletionTime(rowDeletion)));
BTree.<ColumnData>apply(btree, cd -> min.set( Math.min(min.get(), minDeletionTime(cd)) ), cd -> min.get() == Integer.MIN_VALUE, false);
return min.get();
}
public Clustering clustering()
@ -324,17 +330,26 @@ public class BTreeRow extends AbstractRow
public boolean hasComplexDeletion()
{
final WrappedBoolean result = new WrappedBoolean(false);
// We start by the end cause we know complex columns sort before simple ones
for (ColumnData cd : BTree.<ColumnData>iterable(btree, BTree.Dir.DESC))
{
if (cd.column().isSimple())
return false;
apply(c -> {}, cd -> {
if (cd.column.isSimple())
{
result.set(false);
return true;
}
if (!((ComplexColumnData) cd).complexDeletion().isLive())
{
result.set(true);
return true;
}
}
return false;
return false;
}, true);
return result.get();
}
public Row markCounterLocalToBeCleared()

13
src/java/org/apache/cassandra/db/rows/Row.java
View File

@ -19,6 +19,9 @@ package org.apache.cassandra.db.rows;
import java.util.*;
import java.security.MessageDigest;
import java.util.function.Consumer;
import com.google.common.base.Predicate;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
@ -250,6 +253,16 @@ public interface Row extends Unfiltered, Collection<ColumnData>
public String toString(CFMetaData metadata, boolean fullDetails);
/**
* Apply a function to every column in a row
*/
public void apply(Consumer<ColumnData> function, boolean reverse);
/**
* Apply a funtion to every column in a row until a stop condition is reached
*/
public void apply(Consumer<ColumnData> function, Predicate<ColumnData> stopCondition, boolean reverse);
/**
* A row deletion/tombstone.
* <p>

24
src/java/org/apache/cassandra/db/rows/Rows.java
View File

@ -26,6 +26,7 @@ import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.PartitionStatisticsCollector;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.WrappedInt;
/**
* Static utilities to work on Row objects.
@ -72,14 +73,15 @@ public abstract class Rows
collector.update(row.primaryKeyLivenessInfo());
collector.update(row.deletion().time());
int columnCount = 0;
int cellCount = 0;
for (ColumnData cd : row)
{
//we have to wrap these for the lambda
final WrappedInt columnCount = new WrappedInt(0);
final WrappedInt cellCount = new WrappedInt(0);
row.apply(cd -> {
if (cd.column().isSimple())
{
++columnCount;
++cellCount;
columnCount.increment();
cellCount.increment();
Cells.collectStats((Cell) cd, collector);
}
else
@ -88,18 +90,18 @@ public abstract class Rows
collector.update(complexData.complexDeletion());
if (complexData.hasCells())
{
++columnCount;
columnCount.increment();
for (Cell cell : complexData)
{
++cellCount;
cellCount.increment();
Cells.collectStats(cell, collector);
}
}
}
}, false);
}
collector.updateColumnSetPerRow(columnCount);
return cellCount;
collector.updateColumnSetPerRow(columnCount.get());
return cellCount.get();
}
/**

3
src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java
View File

@ -80,12 +80,15 @@ public class UnfilteredRowIteratorSerializer
}
// Should only be used for the on-wire format.
public void serialize(UnfilteredRowIterator iterator, ColumnFilter selection, DataOutputPlus out, int version, int rowEstimate) throws IOException
{
SerializationHeader header = new SerializationHeader(false,
iterator.metadata(),
iterator.columns(),
iterator.stats());
serialize(iterator, header, selection, out, version, rowEstimate);
}

77
src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java
View File

@ -28,6 +28,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.WrappedException;
/**
* Serialize/deserialize a single Unfiltered (both on-wire and on-disk).
@ -181,7 +182,7 @@ public class UnfilteredSerializer
if (header.isForSSTable())
{
DataOutputBuffer dob = DataOutputBuffer.RECYCLER.get();
DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get();
try
{
serializeRowBody(row, flags, header, dob);
@ -190,7 +191,7 @@ public class UnfilteredSerializer
// We write the size of the previous unfiltered to make reverse queries more efficient (and simpler).
// This is currently not used however and using it is tbd.
out.writeUnsignedVInt(previousUnfilteredSize);
out.write(dob.buffer());
out.write(dob.getData(), 0, dob.getLength());
}
finally
{
@ -227,20 +228,37 @@ public class UnfilteredSerializer
Columns.serializer.serializeSubset(Collections2.transform(row, ColumnData::column), headerColumns, out);
SearchIterator<ColumnDefinition, ColumnDefinition> si = headerColumns.iterator();
for (ColumnData data : row)
{
// We can obtain the column for data directly from data.column(). However, if the cell/complex data
// originates from a sstable, the column we'll get will have the type used when the sstable was serialized,
// and if that type have been recently altered, that may not be the type we want to serialize the column
// with. So we use the ColumnDefinition from the "header" which is "current". Also see #11810 for what
// happens if we don't do that.
ColumnDefinition column = si.next(data.column());
assert column != null;
if (data.column.isSimple())
Cell.serializer.serialize((Cell) data, column, out, pkLiveness, header);
else
writeComplexColumn((ComplexColumnData) data, column, (flags & HAS_COMPLEX_DELETION) != 0, pkLiveness, header, out);
try
{
row.apply(cd -> {
// We can obtain the column for data directly from data.column(). However, if the cell/complex data
// originates from a sstable, the column we'll get will have the type used when the sstable was serialized,
// and if that type have been recently altered, that may not be the type we want to serialize the column
// with. So we use the ColumnDefinition from the "header" which is "current". Also see #11810 for what
// happens if we don't do that.
ColumnDefinition column = si.next(cd.column());
assert column != null : cd.column.toString();
try
{
if (cd.column.isSimple())
Cell.serializer.serialize((Cell) cd, column, out, pkLiveness, header);
else
writeComplexColumn((ComplexColumnData) cd, column, (flags & HAS_COMPLEX_DELETION) != 0, pkLiveness, header, out);
}
catch (IOException e)
{
throw new WrappedException(e);
}
}, false);
}
catch (WrappedException e)
{
if (e.getCause() instanceof IOException)
throw (IOException) e.getCause();
throw e;
}
}
@ -496,12 +514,31 @@ public class UnfilteredSerializer
builder.addRowDeletion(hasDeletion ? new Row.Deletion(header.readDeletionTime(in), deletionIsShadowable) : Row.Deletion.LIVE);
Columns columns = hasAllColumns ? headerColumns : Columns.serializer.deserializeSubset(headerColumns, in);
for (ColumnDefinition column : columns)
final LivenessInfo livenessInfo = rowLiveness;
try
{
if (column.isSimple())
readSimpleColumn(column, in, header, helper, builder, rowLiveness);
else
readComplexColumn(column, in, header, helper, hasComplexDeletion, builder, rowLiveness);
columns.apply(column -> {
try
{
if (column.isSimple())
readSimpleColumn(column, in, header, helper, builder, livenessInfo);
else
readComplexColumn(column, in, header, helper, hasComplexDeletion, builder, livenessInfo);
}
catch (IOException e)
{
throw new WrappedException(e);
}
}, false);
}
catch (WrappedException e)
{
if (e.getCause() instanceof IOException)
throw (IOException) e.getCause();
throw e;
}
return builder.build();

2
src/java/org/apache/cassandra/hints/HintsWriter.java
View File

@ -75,7 +75,7 @@ class HintsWriter implements AutoCloseable
CRC32 crc = new CRC32();
DataOutputBuffer dob = null;
try (DataOutputBuffer ignored = dob = DataOutputBuffer.RECYCLER.get())
try (DataOutputBuffer ignored = dob = DataOutputBuffer.scratchBuffer.get())
{
// write the descriptor
descriptor.serialize(dob);

3
src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
View File

@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit;
import com.google.common.base.Throwables;
import com.datastax.shaded.netty.util.concurrent.FastThreadLocalThread;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.Row;
@ -189,7 +190,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
//// typedef
static class Buffer extends TreeMap<DecoratedKey, PartitionUpdate> {}
private class DiskWriter extends Thread
private class DiskWriter extends FastThreadLocalThread
{
volatile Throwable exception = null;

42
src/java/org/apache/cassandra/io/util/DataOutputBuffer.java
View File

@ -24,7 +24,7 @@ import java.nio.channels.WritableByteChannel;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import io.netty.util.Recycler;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.config.Config;
/**
@ -43,25 +43,22 @@ public class DataOutputBuffer extends BufferedDataOutputStreamPlus
/*
* Only recycle OutputBuffers up to 1Mb. Larger buffers will be trimmed back to this size.
*/
private static final int MAX_RECYCLE_BUFFER_SIZE = 1024 * 1024;
private static final int MAX_RECYCLE_BUFFER_SIZE = Integer.getInteger(Config.PROPERTY_PREFIX + "dob_max_recycle_bytes", 1024 * 1024);
private static final int DEFAULT_INITIAL_BUFFER_SIZE = 128;
public static final Recycler<DataOutputBuffer> RECYCLER = new Recycler<DataOutputBuffer>()
/**
* Scratch buffers used mostly for serializing in memory. It's important to call #recycle() when finished
* to keep the memory overhead from being too large in the system.
*/
public static final FastThreadLocal<DataOutputBuffer> scratchBuffer = new FastThreadLocal<DataOutputBuffer>()
{
protected DataOutputBuffer newObject(Handle handle)
protected DataOutputBuffer initialValue() throws Exception
{
return new DataOutputBuffer(handle);
return new DataOutputBuffer();
}
};
private final Recycler.Handle handle;
private DataOutputBuffer(Recycler.Handle handle)
{
this(DEFAULT_INITIAL_BUFFER_SIZE, handle);
}
public DataOutputBuffer()
{
this(DEFAULT_INITIAL_BUFFER_SIZE);
@ -69,28 +66,23 @@ public class DataOutputBuffer extends BufferedDataOutputStreamPlus
public DataOutputBuffer(int size)
{
this(size, null);
super(ByteBuffer.allocate(size));
}
protected DataOutputBuffer(int size, Recycler.Handle handle)
{
this(ByteBuffer.allocate(size), handle);
}
protected DataOutputBuffer(ByteBuffer buffer, Recycler.Handle handle)
public DataOutputBuffer(ByteBuffer buffer)
{
super(buffer);
this.handle = handle;
}
public void recycle()
{
assert handle != null;
if (buffer().capacity() <= MAX_RECYCLE_BUFFER_SIZE)
if (buffer.capacity() <= MAX_RECYCLE_BUFFER_SIZE)
{
buffer.rewind();
RECYCLER.recycle(this, handle);
buffer.clear();
}
else
{
buffer = ByteBuffer.allocate(DEFAULT_INITIAL_BUFFER_SIZE);
}
}

4
src/java/org/apache/cassandra/io/util/DataOutputBufferFixed.java
View File

@ -38,12 +38,12 @@ public class DataOutputBufferFixed extends DataOutputBuffer
public DataOutputBufferFixed(int size)
{
super(size, null);
super(size);
}
public DataOutputBufferFixed(ByteBuffer buffer)
{
super(buffer, null);
super(buffer);
}
@Override

2
src/java/org/apache/cassandra/io/util/SafeMemoryWriter.java
View File

@ -33,7 +33,7 @@ public class SafeMemoryWriter extends DataOutputBuffer
private SafeMemoryWriter(SafeMemory memory)
{
super(tailBuffer(memory).order(ByteOrder.BIG_ENDIAN), null);
super(tailBuffer(memory).order(ByteOrder.BIG_ENDIAN));
this.memory = memory;
}

3
src/java/org/apache/cassandra/net/IncomingTcpConnection.java
View File

@ -29,6 +29,7 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.util.concurrent.FastThreadLocalThread;
import net.jpountz.lz4.LZ4BlockInputStream;
import net.jpountz.lz4.LZ4FastDecompressor;
import net.jpountz.lz4.LZ4Factory;
@ -42,7 +43,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
import org.apache.cassandra.io.util.NIODataInputStream;
public class IncomingTcpConnection extends Thread implements Closeable
public class IncomingTcpConnection extends FastThreadLocalThread implements Closeable
{
private static final Logger logger = LoggerFactory.getLogger(IncomingTcpConnection.class);

2
src/java/org/apache/cassandra/net/MessageIn.java
View File

@ -73,7 +73,7 @@ public class MessageIn<T>
{
InetAddress from = CompactEndpointSerializationHelper.deserialize(in);
MessagingService.Verb verb = MessagingService.Verb.values()[in.readInt()];
MessagingService.Verb verb = MessagingService.verbValues[in.readInt()];
int parameterCount = in.readInt();
Map<String, byte[]> parameters;
if (parameterCount == 0)

24
src/java/org/apache/cassandra/net/MessageOut.java
View File

@ -30,6 +30,7 @@ import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
@ -113,11 +114,26 @@ public class MessageOut<T>
out.write(entry.getValue());
}
long longSize = payloadSize(version);
assert longSize <= Integer.MAX_VALUE; // larger values are supported in sstables but not messages
out.writeInt((int) longSize);
if (payload != null)
serializer.serialize(payload, out, version);
{
DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get();
try
{
serializer.serialize(payload, dob, version);
int size = dob.getLength();
out.writeInt(size);
out.write(dob.getData(), 0, size);
}
finally
{
dob.recycle();
}
}
else
{
out.writeInt(0);
}
}
public int serializedSize(int version)

2
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -153,6 +153,8 @@ public final class MessagingService implements MessagingServiceMBean
;
}
public static final Verb[] verbValues = Verb.values();
public static final EnumMap<MessagingService.Verb, Stage> verbStages = new EnumMap<MessagingService.Verb, Stage>(MessagingService.Verb.class)
{{
put(Verb.MUTATION, Stage.MUTATION);

3
src/java/org/apache/cassandra/net/OutboundTcpConnection.java
View File

@ -40,6 +40,7 @@ import javax.net.ssl.SSLHandshakeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.util.concurrent.FastThreadLocalThread;
import net.jpountz.lz4.LZ4BlockOutputStream;
import net.jpountz.lz4.LZ4Compressor;
import net.jpountz.lz4.LZ4Factory;
@ -63,7 +64,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import com.google.common.util.concurrent.Uninterruptibles;
public class OutboundTcpConnection extends Thread
public class OutboundTcpConnection extends FastThreadLocalThread
{
private static final Logger logger = LoggerFactory.getLogger(OutboundTcpConnection.class);

3
src/java/org/apache/cassandra/service/ClientWarn.java
View File

@ -20,13 +20,14 @@ package org.apache.cassandra.service;
import java.util.ArrayList;
import java.util.List;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.concurrent.ExecutorLocal;
import org.apache.cassandra.utils.FBUtilities;
public class ClientWarn implements ExecutorLocal<ClientWarn.State>
{
private static final String TRUNCATED = " [truncated]";
private static final ThreadLocal<ClientWarn.State> warnLocal = new ThreadLocal<>();
private static final FastThreadLocal<State> warnLocal = new FastThreadLocal<>();
public static ClientWarn instance = new ClientWarn();
private ClientWarn()

3
src/java/org/apache/cassandra/tracing/Tracing.java
View File

@ -31,6 +31,7 @@ import com.google.common.collect.ImmutableMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.concurrent.ExecutorLocal;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.marshal.TimeUUIDType;
@ -84,7 +85,7 @@ public abstract class Tracing implements ExecutorLocal<TraceState>
private final InetAddress localAddress = FBUtilities.getLocalAddress();
private final ThreadLocal<TraceState> state = new ThreadLocal<>();
private final FastThreadLocal<TraceState> state = new FastThreadLocal<>();
protected final ConcurrentMap<UUID, TraceState> sessions = new ConcurrentHashMap<>();

13
src/java/org/apache/cassandra/utils/ChecksumType.java
View File

@ -22,6 +22,8 @@ import java.util.zip.Checksum;
import java.util.zip.CRC32;
import java.util.zip.Adler32;
import io.netty.util.concurrent.FastThreadLocal;
public enum ChecksumType
{
Adler32
@ -60,12 +62,13 @@ public enum ChecksumType
public abstract Checksum newInstance();
public abstract void update(Checksum checksum, ByteBuffer buf);
private ThreadLocal<Checksum> instances = ThreadLocal.withInitial(this::newInstance);
public Checksum threadLocalInstance()
private FastThreadLocal<Checksum> instances = new FastThreadLocal<Checksum>()
{
return instances.get();
}
protected Checksum initialValue() throws Exception
{
return newInstance();
}
};
public long of(ByteBuffer buf)
{

48
src/java/org/apache/cassandra/utils/Wrapped.java Normal file
View File

@ -0,0 +1,48 @@
/*
* 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;
/**
* Simple wrapper class to be used when a lambda function
* needs to modify a variable outside it's scope.
*/
public class Wrapped<T>
{
private T value;
public static <V> Wrapped<V> create(V initial)
{
return new Wrapped<>(initial);
}
private Wrapped(T initial)
{
this.value = initial;
}
public T get()
{
return value;
}
public void set(T value)
{
this.value = value;
}
}

42
src/java/org/apache/cassandra/utils/WrappedBoolean.java Normal file
View File

@ -0,0 +1,42 @@
/*
* 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;
/**
* Simple wrapper for native boolean type
*/
public class WrappedBoolean
{
private boolean value;
public WrappedBoolean(boolean initial)
{
this.value = initial;
}
public boolean get()
{
return value;
}
public void set(boolean value)
{
this.value = value;
}
}

30
src/java/org/apache/cassandra/utils/WrappedException.java Normal file
View File

@ -0,0 +1,30 @@
/*
* 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;
/**
* Wrapped runtime exception for lambda functions
*/
public class WrappedException extends RuntimeException
{
public WrappedException(Exception cause)
{
super(cause);
}
}

52
src/java/org/apache/cassandra/utils/WrappedInt.java Normal file
View File

@ -0,0 +1,52 @@
/*
* 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;
/**
* Simple wrapper for native int type
*/
public class WrappedInt
{
private int value;
public WrappedInt(int initial)
{
this.value = initial;
}
public int get()
{
return value;
}
public void set(int value)
{
this.value = value;
}
public void increment()
{
++value;
}
public void decrement()
{
--value;
}
}

131
src/java/org/apache/cassandra/utils/btree/BTree.java
View File

@ -19,8 +19,10 @@
package org.apache.cassandra.utils.btree;
import java.util.*;
import java.util.function.Consumer;
import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterators;
import com.google.common.collect.Ordering;
@ -136,7 +138,8 @@ public class BTree
for (K k : source)
values[i++] = updateF.apply(k);
}
updateF.allocated(ObjectSizes.sizeOfArray(values));
if (updateF != UpdateFunction.noOp())
updateF.allocated(ObjectSizes.sizeOfArray(values));
return values;
}
@ -188,7 +191,7 @@ public class BTree
tree1 = tree2;
tree2 = tmp;
}
return update(tree1, comparator, new BTreeSet<K>(tree2, comparator), updateF);
return update(tree1, comparator, new BTreeSet<>(tree2, comparator), updateF);
}
public static <V> Iterator<V> iterator(Object[] btree)
@ -1159,4 +1162,128 @@ public class BTree
}
return compare(cmp, previous, max) < 0;
}
/**
* Simple method to walk the btree forwards or reversed and apply a function to each element
*
* Public method
*
*/
public static <V> void apply(Object[] btree, Consumer<V> function, boolean reversed)
{
if (reversed)
applyReverse(btree, function, null);
else
applyForwards(btree, function, null);
}
/**
* Simple method to walk the btree forwards or reversed and apply a function till a stop condition is reached
*
* Public method
*
*/
public static <V> void apply(Object[] btree, Consumer<V> function, Predicate<V> stopCondition, boolean reversed)
{
if (reversed)
applyReverse(btree, function, stopCondition);
else
applyForwards(btree, function, stopCondition);
}
/**
* Simple method to walk the btree forwards and apply a function till a stop condition is reached
*
* Private method
*
* @param btree
* @param function
* @param stopCondition
*/
private static <V> boolean applyForwards(Object[] btree, Consumer<V> function, Predicate<V> stopCondition)
{
boolean isLeaf = isLeaf(btree);
int childOffset = isLeaf ? Integer.MAX_VALUE : getChildStart(btree);
int limit = isLeaf ? getLeafKeyEnd(btree) : btree.length - 1;
for (int i = 0 ; i < limit ; i++)
{
// we want to visit in iteration order, so we visit our key nodes inbetween our children
int idx = isLeaf ? i : (i / 2) + (i % 2 == 0 ? childOffset : 0);
Object current = btree[idx];
if (idx < childOffset)
{
V castedCurrent = (V) current;
if (stopCondition != null && stopCondition.apply(castedCurrent))
return true;
function.accept(castedCurrent);
}
else
{
if (applyForwards((Object[]) current, function, stopCondition))
return true;
}
}
return false;
}
/**
* Simple method to walk the btree in reverse and apply a function till a stop condition is reached
*
* Private method
*
* @param btree
* @param function
* @param stopCondition
*/
private static <V> boolean applyReverse(Object[] btree, Consumer<V> function, Predicate<V> stopCondition)
{
boolean isLeaf = isLeaf(btree);
int childOffset = isLeaf ? 0 : getChildStart(btree);
int limit = isLeaf ? getLeafKeyEnd(btree) : btree.length - 1;
for (int i = limit - 1, visited = 0; i >= 0 ; i--, visited++)
{
int idx = i;
// we want to visit in reverse iteration order, so we visit our children nodes inbetween our keys
if (!isLeaf)
{
int typeOffset = visited / 2;
if (i % 2 == 0)
{
// This is a child branch. Since children are in the second half of the array, we must
// adjust for the key's we've visited along the way
idx += typeOffset;
}
else
{
// This is a key. Since the keys are in the first half of the array and we are iterating
// in reverse we subtract the childOffset and adjust for children we've walked so far
idx = i - childOffset + typeOffset;
}
}
Object current = btree[idx];
if (isLeaf || idx < childOffset)
{
V castedCurrent = (V) current;
if (stopCondition != null && stopCondition.apply(castedCurrent))
return true;
function.accept(castedCurrent);
}
else
{
if (applyReverse((Object[]) current, function, stopCondition))
return true;
}
}
return false;
}
}

4
test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java
View File

@ -35,6 +35,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import io.netty.util.concurrent.FastThreadLocalThread;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.UpdateBuilder;
@ -408,7 +409,8 @@ public class CommitLogStressTest
return slice;
}
public class CommitlogThread extends Thread {
public class CommitlogThread extends FastThreadLocalThread
{
final AtomicLong counter = new AtomicLong();
int hash = 0;
int cells = 0;

96
test/microbench/org/apache/cassandra/test/microbench/FastThreadExecutor.java Normal file
View File

@ -0,0 +1,96 @@
/*
* 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.test.microbench;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import io.netty.util.concurrent.FastThreadLocalThread;
/**
* Created to test perf of FastThreadLocal
*
* Used in MutationBench via:
* jvmArgsAppend = {"-Djmh.executor=CUSTOM", "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor"}
*/
public class FastThreadExecutor extends AbstractExecutorService
{
final FastThreadLocalThread thread;
final LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
final CountDownLatch shutdown = new CountDownLatch(1);
public FastThreadExecutor(int size, String name)
{
assert size == 1;
thread = new FastThreadLocalThread(() -> {
Runnable work = null;
try
{
while ((work = queue.take()) != null)
work.run();
}
catch (InterruptedException e)
{
shutdown.countDown();
}
});
thread.setName(name + "-1");
thread.setDaemon(true);
thread.start();
}
public void shutdown()
{
thread.interrupt();
}
public List<Runnable> shutdownNow()
{
thread.interrupt();
return Collections.emptyList();
}
public boolean isShutdown()
{
return shutdown.getCount() == 0;
}
public boolean isTerminated()
{
return shutdown.getCount() == 0;
}
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException
{
return shutdown.await(timeout, unit);
}
public void execute(Runnable command)
{
while(!queue.add(command));
}
}

148
test/microbench/org/apache/cassandra/test/microbench/MutationBench.java Normal file
View File

@ -0,0 +1,148 @@
/*
* 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.test.microbench;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.concurrent.*;
import org.apache.cassandra.UpdateBuilder;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputBufferFixed;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.openjdk.jmh.annotations.*;
import org.openjdk.jmh.profile.StackProfiler;
import org.openjdk.jmh.results.Result;
import org.openjdk.jmh.results.RunResult;
import org.openjdk.jmh.runner.Runner;
import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 5, time = 2, timeUnit = TimeUnit.SECONDS)
@Fork(value = 1
, jvmArgsAppend = {"-Djmh.executor=CUSTOM", "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor"
//,"-XX:+UnlockCommercialFeatures", "-XX:+FlightRecorder","-XX:+UnlockDiagnosticVMOptions", "-XX:+DebugNonSafepoints",
// "-XX:StartFlightRecording=duration=60s,filename=./profiling-data.jfr,name=profile,settings=profile",
// "-XX:FlightRecorderOptions=settings=/home/jake/workspace/cassandra/profiling-advanced.jfc,samplethreads=true"
}
)
@Threads(1)
@State(Scope.Benchmark)
public class MutationBench
{
static
{
Config.setClientMode(true);
// Partitioner is not set in client mode.
if (DatabaseDescriptor.getPartitioner() == null)
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
}
static String keyspace = "keyspace1";
private Mutation mutation;
private MessageOut<Mutation> messageOut;
private ByteBuffer buffer;
private DataOutputBuffer outputBuffer;
private DataInputBuffer inputBuffer;
@State(Scope.Thread)
public static class ThreadState
{
MessageIn<Mutation> in;
int counter = 0;
}
@Setup
public void setup() throws IOException
{
Schema.instance.load(KeyspaceMetadata.create(keyspace, KeyspaceParams.simple(1)));
KeyspaceMetadata ksm = Schema.instance.getKSMetaData(keyspace);
CFMetaData metadata = CFMetaData.compile("CREATE TABLE userpics " +
"( userid bigint," +
"picid bigint," +
"commentid bigint, " +
"PRIMARY KEY(userid, picid))", keyspace);
Schema.instance.load(metadata);
Schema.instance.setKeyspaceMetadata(ksm.withSwapped(ksm.tables.with(metadata)));
mutation = (Mutation)UpdateBuilder.create(metadata, 1L).newRow(1L).add("commentid", 32L).makeMutation();
messageOut = mutation.createMessage();
buffer = ByteBuffer.allocate(messageOut.serializedSize(MessagingService.current_version));
outputBuffer = new DataOutputBufferFixed(buffer);
inputBuffer = new DataInputBuffer(buffer, false);
messageOut.serialize(outputBuffer, MessagingService.current_version);
}
@Benchmark
public void serialize(ThreadState state) throws IOException
{
buffer.rewind();
messageOut.serialize(outputBuffer, MessagingService.current_version);
state.counter++;
}
@Benchmark
public void deserialize(ThreadState state) throws IOException
{
buffer.rewind();
state.in = MessageIn.read(inputBuffer, MessagingService.current_version, 0);
state.counter++;
}
public static void main(String... args) throws Exception {
Options opts = new OptionsBuilder()
.include(".*"+MutationBench.class.getSimpleName()+".*")
.jvmArgs("-server")
.forks(1)
.mode(Mode.Throughput)
.addProfiler(StackProfiler.class)
.build();
Collection<RunResult> records = new Runner(opts).run();
for ( RunResult result : records) {
Result r = result.getPrimaryResult();
System.out.println("API replied benchmark score: "
+ r.getScore() + " "
+ r.getScoreUnit() + " over "
+ r.getStatistics().getN() + " iterations");
}
}
}

25
test/unit/org/apache/cassandra/utils/BTreeTest.java
View File

@ -21,6 +21,7 @@ import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import org.junit.Test;
import junit.framework.Assert;
@ -128,6 +129,30 @@ public class BTreeTest
checkResult(i, BTree.update(BTree.build(seq(i), noOp), CMP, seq(i), updateF));
}
@Test
public void testApplyForwards()
{
List<Integer> input = seq(71);
Object[] btree = BTree.build(input, noOp);
final List<Integer> result = new ArrayList<>();
BTree.<Integer>apply(btree, i -> result.add(i), false);
org.junit.Assert.assertArrayEquals(input.toArray(),result.toArray());
}
@Test
public void testApplyReverse()
{
List<Integer> input = seq(71);
Object[] btree = BTree.build(input, noOp);
final List<Integer> result = new ArrayList<>();
BTree.<Integer>apply(btree, i -> result.add(i), true);
org.junit.Assert.assertArrayEquals(Lists.reverse(input).toArray(),result.toArray());
}
/**
* Tests that the apply method of the <code>UpdateFunction</code> is only called once with each key update.
* (see CASSANDRA-8018).