cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject [6/7] git commit: avoid serializing to byte[] on commitlog append patch by jbellis; reviewed by yukim for CASSANDRA-5199
Date Tue, 05 Feb 2013 14:46:05 GMT
avoid serializing to byte[] on commitlog append
patch by jbellis; reviewed by yukim for CASSANDRA-5199


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/baca0bc7
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/baca0bc7
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/baca0bc7

Branch: refs/heads/trunk
Commit: baca0bc7fe4b13bf8970f974c15d83701e02c048
Parents: 1dcf18c
Author: Jonathan Ellis <jbellis@apache.org>
Authored: Tue Feb 5 15:44:32 2013 +0100
Committer: Jonathan Ellis <jbellis@apache.org>
Committed: Tue Feb 5 15:45:34 2013 +0100

----------------------------------------------------------------------
 CHANGES.txt                                        |    1 +
 .../cassandra/db/commitlog/CommitLogReplayer.java  |    7 +++-
 .../cassandra/db/commitlog/CommitLogSegment.java   |   25 +++++++----
 .../cassandra/io/util/ByteBufferOutputStream.java  |   25 ++++++++++
 .../cassandra/io/util/ChecksummedOutputStream.java |   35 +++++++++++++++
 .../org/apache/cassandra/utils/FBUtilities.java    |    9 ++++
 6 files changed, 92 insertions(+), 10 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/baca0bc7/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index bf95ae1..b1c066c 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 1.3
+ * avoid serializing to byte[] on commitlog append (CASSANDRA-5199)
  * make index_interval configurable per columnfamily (CASSANDRA-3961)
  * add default_tim_to_live (CASSANDRA-3974)
  * add memtable_flush_period_in_ms (CASSANDRA-4237)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/baca0bc7/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
index 705053b..107e9b7 100644
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
@@ -167,9 +167,14 @@ public class CommitLogReplayer
                     // This prevents CRC by being fooled by special-case garbage in the file;
see CASSANDRA-2128
                     if (serializedSize < 10)
                         break;
+
                     long claimedSizeChecksum = reader.readLong();
                     checksum.reset();
-                    checksum.update(serializedSize);
+                    if (CommitLogDescriptor.current_version < CommitLogDescriptor.VERSION_20)
+                        checksum.update(serializedSize);
+                    else
+                        FBUtilities.updateChecksumInt(checksum, serializedSize);
+
                     if (checksum.getValue() != claimedSizeChecksum)
                         break; // entry wasn't synced correctly/fully. that's
                                // ok.

http://git-wip-us.apache.org/repos/asf/cassandra/blob/baca0bc7/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
index 469ab99..9256214 100644
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.db.commitlog;
 
+import java.io.DataOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
@@ -38,6 +39,8 @@ import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.db.ColumnFamily;
 import org.apache.cassandra.db.RowMutation;
 import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.io.util.ByteBufferOutputStream;
+import org.apache.cassandra.io.util.ChecksummedOutputStream;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.utils.FBUtilities;
@@ -69,6 +72,8 @@ public class CommitLogSegment
     private boolean needsSync = false;
 
     private final MappedByteBuffer buffer;
+    private final Checksum checksum;
+    private final DataOutputStream bufferStream;
     private boolean closed;
 
     public final CommitLogDescriptor descriptor;
@@ -122,6 +127,8 @@ public class CommitLogSegment
             logFileAccessor.setLength(DatabaseDescriptor.getCommitLogSegmentSize());
 
             buffer = logFileAccessor.getChannel().map(FileChannel.MapMode.READ_WRITE, 0,
DatabaseDescriptor.getCommitLogSegmentSize());
+            checksum = new PureJavaCrc32();
+            bufferStream = new DataOutputStream(new ChecksummedOutputStream(new ByteBufferOutputStream(buffer),
checksum));
             buffer.putInt(CommitLog.END_OF_SEGMENT_MARKER);
             buffer.position(0);
 
@@ -204,24 +211,24 @@ public class CommitLogSegment
    /**
      * Appends a row mutation onto the commit log.  Requres that hasCapacityFor has already
been checked.
      *
-     * @param   rowMutation   the mutation to append to the commit log.
+     * @param   mutation   the mutation to append to the commit log.
      * @return  the position of the appended mutation
      */
-    public ReplayPosition write(RowMutation rowMutation) throws IOException
+    public ReplayPosition write(RowMutation mutation) throws IOException
     {
         assert !closed;
         ReplayPosition repPos = getContext();
-        markDirty(rowMutation, repPos);
+        markDirty(mutation, repPos);
 
-        Checksum checksum = new PureJavaCrc32();
-        byte[] serializedRow = FBUtilities.serialize(rowMutation, RowMutation.serializer,
MessagingService.current_version);
+        checksum.reset();
 
-        checksum.update(serializedRow.length);
-        buffer.putInt(serializedRow.length);
+        // checksummed length
+        int length = (int) RowMutation.serializer.serializedSize(mutation, MessagingService.current_version);
+        bufferStream.writeInt(length);
         buffer.putLong(checksum.getValue());
 
-        buffer.put(serializedRow);
-        checksum.update(serializedRow, 0, serializedRow.length);
+        // checksummed mutation
+        RowMutation.serializer.serialize(mutation, bufferStream, MessagingService.current_version);
         buffer.putLong(checksum.getValue());
 
         if (buffer.remaining() >= 4)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/baca0bc7/src/java/org/apache/cassandra/io/util/ByteBufferOutputStream.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/util/ByteBufferOutputStream.java b/src/java/org/apache/cassandra/io/util/ByteBufferOutputStream.java
new file mode 100644
index 0000000..9a15a1c
--- /dev/null
+++ b/src/java/org/apache/cassandra/io/util/ByteBufferOutputStream.java
@@ -0,0 +1,25 @@
+package org.apache.cassandra.io.util;
+
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+public class ByteBufferOutputStream extends OutputStream
+{
+    private final ByteBuffer buffer;
+
+    public ByteBufferOutputStream(ByteBuffer buffer)
+    {
+        this.buffer = buffer;
+    }
+
+    public void write(int b)
+    {
+        buffer.put((byte) b);
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len)
+    {
+        buffer.put(b, off, len);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/baca0bc7/src/java/org/apache/cassandra/io/util/ChecksummedOutputStream.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/util/ChecksummedOutputStream.java b/src/java/org/apache/cassandra/io/util/ChecksummedOutputStream.java
new file mode 100644
index 0000000..c7f2df3
--- /dev/null
+++ b/src/java/org/apache/cassandra/io/util/ChecksummedOutputStream.java
@@ -0,0 +1,35 @@
+package org.apache.cassandra.io.util;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.zip.Checksum;
+
+public class ChecksummedOutputStream extends OutputStream
+{
+    private final OutputStream out;
+    private final Checksum checksum;
+
+    public ChecksummedOutputStream(OutputStream out, Checksum checksum)
+    {
+        this.out = out;
+        this.checksum = checksum;
+    }
+
+    public void resetChecksum()
+    {
+        checksum.reset();
+    }
+
+    public void write(int b) throws IOException
+    {
+        out.write(b);
+        checksum.update(b);
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) throws IOException
+    {
+        out.write(b, off, len);
+        checksum.update(b, off, len);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/baca0bc7/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index cafd856..15f7fb0 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -34,6 +34,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.zip.Checksum;
 
 import com.google.common.base.Joiner;
 import com.google.common.collect.AbstractIterator;
@@ -609,6 +610,14 @@ public class FBUtilities
         }
     }
 
+    public static void updateChecksumInt(Checksum checksum, int v)
+    {
+        checksum.update((v >>> 24) & 0xFF);
+        checksum.update((v >>> 16) & 0xFF);
+        checksum.update((v >>> 8) & 0xFF);
+        checksum.update((v >>> 0) & 0xFF);
+    }
+
     private static final class WrappedCloseableIterator<T>
         extends AbstractIterator<T> implements CloseableIterator<T>
     {


Mime
View raw message