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

avoid allocation-per-row in log replay 

patch by jbellis; reviewed by mdenns for CASSANDRA-1219

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@957105 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-06-23 04:07:14 +00:00
parent 45254e10cc
commit 442c34b557
2 changed files with 12 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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

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

@ -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",