cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r939053 - in /cassandra/trunk: src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java test/unit/org/apache/cassandra/db/commitlog/CommitLogHeaderTest.java
Date Wed, 28 Apr 2010 19:05:10 GMT
Author: jbellis
Date: Wed Apr 28 19:05:10 2010
New Revision: 939053

URL: http://svn.apache.org/viewvc?rev=939053&view=rev
Log:
redo "add crc to commitlogheader," with fix for header serialization.
patch by jbellis; reviewed by gdusbabek for CASSANDRA-999

Modified:
    cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java
    cassandra/trunk/test/unit/org/apache/cassandra/db/commitlog/CommitLogHeaderTest.java

Modified: cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java?rev=939053&r1=939052&r2=939053&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java Wed Apr
28 19:05:10 2010
@@ -19,10 +19,13 @@
 package org.apache.cassandra.db.commitlog;
 
 import java.io.*;
+import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.zip.CRC32;
+import java.util.zip.Checksum;
 
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.io.ICompactSerializer;
@@ -31,7 +34,7 @@ import org.apache.cassandra.utils.Pair;
 
 class CommitLogHeader
 {    
-    private static CommitLogHeaderSerializer serializer = new CommitLogHeaderSerializer();
+    static CommitLogHeaderSerializer serializer = new CommitLogHeaderSerializer();
 
     static int getLowestPosition(CommitLogHeader clheader)
     {
@@ -52,7 +55,6 @@ class CommitLogHeader
     private final byte[] serializedCfMap; // serialized. only needed during commit log recovery.
     private final int cfCount; // we keep this in case cfcount changes in the interim (size
of lastFlushedAt is not a good indication).
     
-    private transient final int maxSerializedSize;
     private transient Map<Pair<String, String>, Integer> cfIdMap; // only needed
during recovery. created from this.serializedCfMap.
     
     CommitLogHeader()
@@ -71,8 +73,6 @@ class CommitLogHeader
         this.lastFlushedAt = lastFlushedAt;
         this.serializedCfMap = serializedCfMap;
         assert lastFlushedAt.size() <= cfCount;
-        // (size of lastFlushedAt) + (size of map buf) + (size of cfCount int)
-        maxSerializedSize = (8 * cfCount + 4) + (serializedCfMap.length + 4) + (4);
     }
         
     boolean isDirty(int cfId)
@@ -150,14 +150,11 @@ class CommitLogHeader
 
     byte[] toByteArray() throws IOException
     {
-        ByteArrayOutputStream bos = new ByteArrayOutputStream(maxSerializedSize);
-        DataOutputStream dos = new DataOutputStream(bos);        
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(bos);
         serializer.serialize(this, dos);
-        byte[] src = bos.toByteArray();
-        assert src.length <= maxSerializedSize;
-        byte[] dst = new byte[maxSerializedSize];
-        System.arraycopy(src, 0, dst, 0, src.length);
-        return dst;
+        dos.flush();
+        return bos.toByteArray();
     }
     
     // we use cf ids. getting the cf names would be pretty pretty expensive.
@@ -195,27 +192,70 @@ class CommitLogHeader
         public void serialize(CommitLogHeader clHeader, DataOutputStream dos) throws IOException
         {
             assert clHeader.lastFlushedAt.size() <= clHeader.cfCount;
+            Checksum checksum = new CRC32();
+
+            // write the first checksum after the fixed-size part, so we won't OOM allocating
a bogus cfmap buffer
             dos.writeInt(clHeader.cfCount); // 4
             dos.writeInt(clHeader.serializedCfMap.length); // 4
-            dos.write(clHeader.serializedCfMap); // colMap.length
             dos.writeInt(clHeader.lastFlushedAt.size()); // 4
+            checksum.update(clHeader.cfCount);
+            checksum.update(clHeader.serializedCfMap.length);
+            checksum.update(clHeader.lastFlushedAt.size());
+            dos.writeLong(checksum.getValue());
+
+            // write the 2nd checksum after the cfmap and lastflushedat map
+            dos.write(clHeader.serializedCfMap); // colMap.length
+            checksum.update(clHeader.serializedCfMap, 0, clHeader.serializedCfMap.length);
             for (Map.Entry<Integer, Integer> entry : clHeader.lastFlushedAt.entrySet())
             {
                 dos.writeInt(entry.getKey()); // 4
+                checksum.update(entry.getKey());
                 dos.writeInt(entry.getValue()); // 4
+                checksum.update(entry.getValue());
+            }
+            dos.writeLong(checksum.getValue());
+
+            // keep the size constant by padding for missing flushed-at entries.  these do
not affect checksum.
+            for (int i = clHeader.lastFlushedAt.entrySet().size(); i < clHeader.cfCount;
i++)
+            {
+                dos.writeInt(0);
+                dos.writeInt(0);
             }
         }
 
         public CommitLogHeader deserialize(DataInputStream dis) throws IOException
         {
-            int colCount = dis.readInt();
-            byte[] map = new byte[dis.readInt()];
-            dis.readFully(map);
-            int size = dis.readInt();
+            Checksum checksum = new CRC32();
+
+            int cfCount = dis.readInt();
+            checksum.update(cfCount);
+            int cfMapLength = dis.readInt();
+            checksum.update(cfMapLength);
+            int lastFlushedAtSize = dis.readInt();
+            checksum.update(lastFlushedAtSize);
+            if (checksum.getValue() != dis.readLong())
+            {
+                throw new IOException("Invalid or corrupt commitlog header");
+            }
+
+            byte[] cfMap = new byte[cfMapLength];
+            dis.readFully(cfMap);
+            checksum.update(cfMap, 0, cfMap.length);
             Map<Integer, Integer> lastFlushedAt = new HashMap<Integer, Integer>();
-            for (int i = 0; i < size; i++)
-                lastFlushedAt.put(dis.readInt(), dis.readInt());
-            return new CommitLogHeader(lastFlushedAt, map, colCount);
+            for (int i = 0; i < lastFlushedAtSize; i++)
+            {
+                int key = dis.readInt();
+                checksum.update(key);
+                int value = dis.readInt();
+                checksum.update(value);
+                lastFlushedAt.put(key, value);
+            }
+            if (checksum.getValue() != dis.readLong())
+            {
+                throw new IOException("Invalid or corrupt commitlog header");
+            }
+
+            return new CommitLogHeader(lastFlushedAt, cfMap, cfCount);
         }
     }
 }

Modified: cassandra/trunk/test/unit/org/apache/cassandra/db/commitlog/CommitLogHeaderTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/db/commitlog/CommitLogHeaderTest.java?rev=939053&r1=939052&r2=939053&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/db/commitlog/CommitLogHeaderTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/db/commitlog/CommitLogHeaderTest.java Wed
Apr 28 19:05:10 2010
@@ -80,6 +80,16 @@ public class CommitLogHeaderTest extends
         
         assert one.length == two.length;
     }
+
+    @Test
+    public void byteArray() throws IOException
+    {
+        SchemaLoader.loadSchemaFromYaml();
+        CommitLogHeader clh = new CommitLogHeader();
+        assert clh.getCfIdMap().size() > 0;
+        CommitLogHeader clh2 = CommitLogHeader.serializer.deserialize(new DataInputStream(new
ByteArrayInputStream(clh.toByteArray())));
+        assert clh.getCfIdMap().equals(clh2.getCfIdMap());
+    }
     
     @Test
     public void cfMapSerialization() throws IOException



Mime
View raw message