Author: jbellis
Date: Wed Jun 23 04:07:14 2010
New Revision: 957105
URL: http://svn.apache.org/viewvc?rev=957105&view=rev
Log:
avoid allocation-per-row in log replay
patch by jbellis; reviewed by mdenns for CASSANDRA-1219
Modified:
cassandra/trunk/CHANGES.txt
cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
Modified: cassandra/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/cassandra/trunk/CHANGES.txt?rev=957105&r1=957104&r2=957105&view=diff
==============================================================================
--- cassandra/trunk/CHANGES.txt (original)
+++ cassandra/trunk/CHANGES.txt Wed Jun 23 04:07:14 2010
@@ -32,6 +32,7 @@ dev
* efficient Streaming (no more anticompaction) (CASSANDRA-579)
* split commitlog header into separate file and add size checksum to
mutations (CASSANDRA-1179)
+ * avoid allocating a new byte[] for each mutation on replay (CASSANDRA-1219)
0.6.3
Modified:
cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
URL:
http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLog.java?rev=957105&r1=957104&r2=957105&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
(original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
Wed Jun 23 04:07:14 2010
@@ -181,6 +181,8 @@ public class CommitLog
{
Set<Table> tablesRecovered = new HashSet<Table>();
final AtomicInteger counter = new AtomicInteger(0);
+ byte[] bytes = new byte[4096];
+
for (File file : clogs)
{
CommitLogHeader clHeader = null;
@@ -211,20 +213,21 @@ public class CommitLog
logger.debug("Reading mutation at " +
reader.getFilePointer());
long claimedCRC32;
- byte[] bytes;
Checksum checksum = new CRC32();
+ int serializedSize;
try
{
// any of the reads may hit EOF
- int size = reader.readInt();
+ serializedSize = reader.readInt();
long claimedSizeChecksum = reader.readLong();
- checksum.update(size);
- if (checksum.getValue() != claimedSizeChecksum || size <=
0)
+ checksum.update(serializedSize);
+ if (checksum.getValue() != claimedSizeChecksum ||
serializedSize <= 0)
break; // entry wasn't synced correctly/fully. that's
ok.
- bytes = new byte[size];
- reader.readFully(bytes);
+ if (serializedSize > bytes.length)
+ bytes = new byte[(int) (1.2 * serializedSize)];
+ reader.readFully(bytes, 0, serializedSize);
claimedCRC32 = reader.readLong();
}
catch(EOFException eof)
@@ -232,7 +235,7 @@ public class CommitLog
break; // last CL entry didn't get completely written.
that's ok.
}
- checksum.update(bytes, 0, bytes.length);
+ checksum.update(bytes, 0, serializedSize);
if (claimedCRC32 != checksum.getValue())
{
// this entry must not have been fsynced. probably the
rest is bad too,
@@ -241,7 +244,7 @@ public class CommitLog
}
/* deserialize the commit log entry */
- ByteArrayInputStream bufIn = new ByteArrayInputStream(bytes);
+ ByteArrayInputStream bufIn = new ByteArrayInputStream(bytes,
0, serializedSize);
final RowMutation rm =
RowMutation.serializer().deserialize(new DataInputStream(bufIn));
if (logger.isDebugEnabled())
logger.debug(String.format("replaying mutation for %s.%s:
%s",
