cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject [7/15] git commit: serializedSize implementations, part 1 (gossip and streaming packages) patch by jbellis; reviewed by yukim for CASSANDRA-3617
Date Tue, 08 May 2012 17:56:15 GMT
serializedSize implementations, part 1 (gossip and streaming packages)
patch by jbellis; reviewed by yukim for CASSANDRA-3617


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

Branch: refs/heads/trunk
Commit: 5b9fc26c51161837f01a9383aad8a2786445a4bd
Parents: 9471e8d
Author: Jonathan Ellis <jbellis@apache.org>
Authored: Mon Mar 26 17:53:59 2012 -0500
Committer: Jonathan Ellis <jbellis@apache.org>
Committed: Tue May 8 12:40:53 2012 -0500

----------------------------------------------------------------------
 .../org/apache/cassandra/dht/AbstractBounds.java   |   32 ++++++++---
 .../org/apache/cassandra/gms/EndpointState.java    |   13 ++++-
 .../org/apache/cassandra/gms/GossipDigest.java     |    9 ++-
 .../org/apache/cassandra/gms/GossipDigestAck.java  |   44 ++++++++++++---
 .../org/apache/cassandra/gms/GossipDigestAck2.java |   32 +++++++++--
 .../org/apache/cassandra/gms/GossipDigestSyn.java  |   38 +++---------
 .../cassandra/gms/GossipShutdownMessage.java       |    2 +-
 .../org/apache/cassandra/gms/HeartBeatState.java   |    5 +-
 .../org/apache/cassandra/gms/VersionedValue.java   |   21 ++++---
 .../apache/cassandra/service/MigrationManager.java |    4 +-
 .../apache/cassandra/streaming/PendingFile.java    |   18 +++++-
 .../apache/cassandra/streaming/StreamHeader.java   |   14 ++++-
 .../apache/cassandra/streaming/StreamReply.java    |    6 +-
 .../apache/cassandra/streaming/StreamRequest.java  |   28 +++++++--
 .../cassandra/streaming/StreamingRepairTask.java   |   11 +++-
 .../org/apache/cassandra/utils/FBUtilities.java    |    7 ++
 .../org/apache/cassandra/utils/MerkleTree.java     |   20 +++++--
 src/java/org/apache/cassandra/utils/UUIDGen.java   |    4 +-
 18 files changed, 218 insertions(+), 90 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/dht/AbstractBounds.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/AbstractBounds.java b/src/java/org/apache/cassandra/dht/AbstractBounds.java
index cdca2b2..44344cc 100644
--- a/src/java/org/apache/cassandra/dht/AbstractBounds.java
+++ b/src/java/org/apache/cassandra/dht/AbstractBounds.java
@@ -23,6 +23,7 @@ import java.io.IOException;
 import java.io.Serializable;
 import java.util.*;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.db.RowPosition;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.net.MessagingService;
@@ -118,12 +119,8 @@ public abstract class AbstractBounds<T extends RingPosition> implements
Serializ
              * The first int tells us if it's a range or bounds (depending on the value)
_and_ if it's tokens or keys (depending on the
              * sign). We use negative kind for keys so as to preserve the serialization of
token from older version.
              */
-            boolean isToken = range.left instanceof Token;
-            int kind = range instanceof Range ? Type.RANGE.ordinal() : Type.BOUNDS.ordinal();
-            if (!isToken)
-                kind = -(kind+1);
-            out.writeInt(kind);
-            if (isToken)
+            out.writeInt(kindInt(range));
+            if (range.left instanceof Token)
             {
                 Token.serializer().serialize((Token)range.left, out);
                 Token.serializer().serialize((Token)range.right, out);
@@ -135,6 +132,14 @@ public abstract class AbstractBounds<T extends RingPosition> implements
Serializ
             }
         }
 
+        private int kindInt(AbstractBounds<?> ab)
+        {
+            int kind = ab instanceof Range ? Type.RANGE.ordinal() : Type.BOUNDS.ordinal();
+            if (!(ab.left instanceof Token))
+                kind = -(kind + 1);
+            return kind;
+        }
+
         public AbstractBounds<?> deserialize(DataInput in, int version) throws IOException
         {
             int kind = in.readInt();
@@ -159,9 +164,20 @@ public abstract class AbstractBounds<T extends RingPosition> implements
Serializ
             return new Bounds(left, right);
         }
 
-        public long serializedSize(AbstractBounds<?> abstractBounds, int version)
+        public long serializedSize(AbstractBounds<?> ab, int version)
         {
-            throw new UnsupportedOperationException();
+            int size = DBTypeSizes.NATIVE.sizeof(kindInt(ab));
+            if (ab.left instanceof Token)
+            {
+                size += Token.serializer().serializedSize((Token) ab.left, DBTypeSizes.NATIVE);
+                size += Token.serializer().serializedSize((Token) ab.right, DBTypeSizes.NATIVE);
+            }
+            else
+            {
+                size += RowPosition.serializer().serializedSize((RowPosition) ab.left, DBTypeSizes.NATIVE);
+                size += RowPosition.serializer().serializedSize((RowPosition) ab.right, DBTypeSizes.NATIVE);
+            }
+            return size;
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/gms/EndpointState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/EndpointState.java b/src/java/org/apache/cassandra/gms/EndpointState.java
index 05757ee..31eaafd 100644
--- a/src/java/org/apache/cassandra/gms/EndpointState.java
+++ b/src/java/org/apache/cassandra/gms/EndpointState.java
@@ -23,6 +23,7 @@ import java.util.Map;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.cliffc.high_scale_lib.NonBlockingHashMap;
 
@@ -150,8 +151,16 @@ class EndpointStateSerializer implements IVersionedSerializer<EndpointState>
         return epState;
     }
 
-    public long serializedSize(EndpointState endpointState, int version)
+    public long serializedSize(EndpointState epState, int version)
     {
-        throw new UnsupportedOperationException();
+        long size = HeartBeatState.serializer().serializedSize(epState.getHeartBeatState(),
version);
+        size += DBTypeSizes.NATIVE.sizeof(epState.applicationState.size());
+        for (Map.Entry<ApplicationState, VersionedValue> entry : epState.applicationState.entrySet())
+        {
+            VersionedValue value = entry.getValue();
+            size += DBTypeSizes.NATIVE.sizeof(entry.getKey().ordinal());
+            size += VersionedValue.serializer.serializedSize(value, version);
+        }
+        return size;
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/gms/GossipDigest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/GossipDigest.java b/src/java/org/apache/cassandra/gms/GossipDigest.java
index 3dc8294..fcc598e 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigest.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigest.java
@@ -20,8 +20,10 @@ package org.apache.cassandra.gms;
 import java.io.*;
 import java.net.InetAddress;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.net.CompactEndpointSerializationHelper;
+import org.apache.cassandra.utils.FBUtilities;
 
 /**
  * Contains information about a specified list of Endpoints and the largest version
@@ -103,8 +105,11 @@ class GossipDigestSerializer implements IVersionedSerializer<GossipDigest>
         return new GossipDigest(endpoint, generation, maxVersion);
     }
 
-    public long serializedSize(GossipDigest gossipDigest, int version)
+    public long serializedSize(GossipDigest gDigest, int version)
     {
-        throw new UnsupportedOperationException();
+        long size = CompactEndpointSerializationHelper.serializedSize(gDigest.endpoint);
+        size += DBTypeSizes.NATIVE.sizeof(gDigest.generation);
+        size += DBTypeSizes.NATIVE.sizeof(gDigest.maxVersion);
+        return size;
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/gms/GossipDigestAck.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAck.java b/src/java/org/apache/cassandra/gms/GossipDigestAck.java
index 64a6e3c..0e01d60 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestAck.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestAck.java
@@ -17,13 +17,18 @@
  */
 package org.apache.cassandra.gms;
 
-import java.io.*;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
 import java.net.InetAddress;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
-
+import org.apache.cassandra.net.CompactEndpointSerializationHelper;
+import org.apache.cassandra.net.MessagingService;
 
 
 /**
@@ -69,20 +74,43 @@ class GossipDigestAckSerializer implements IVersionedSerializer<GossipDigestAck>
     public void serialize(GossipDigestAck gDigestAckMessage, DataOutput dos, int version)
throws IOException
     {
         GossipDigestSerializationHelper.serialize(gDigestAckMessage.gDigestList, dos, version);
-        dos.writeBoolean(true); // 0.6 compatibility
-        EndpointStatesSerializationHelper.serialize(gDigestAckMessage.epStateMap, dos, version);
+        if (version <= MessagingService.VERSION_10)
+            dos.writeBoolean(true); // 0.6 compatibility
+        dos.writeInt(gDigestAckMessage.epStateMap.size());
+        for (Map.Entry<InetAddress, EndpointState> entry : gDigestAckMessage.epStateMap.entrySet())
+        {
+            InetAddress ep = entry.getKey();
+            CompactEndpointSerializationHelper.serialize(ep, dos);
+            EndpointState.serializer().serialize(entry.getValue(), dos, version);
+        }
     }
 
     public GossipDigestAck deserialize(DataInput dis, int version) throws IOException
     {
         List<GossipDigest> gDigestList = GossipDigestSerializationHelper.deserialize(dis,
version);
-        dis.readBoolean(); // 0.6 compatibility
-        Map<InetAddress, EndpointState> epStateMap = EndpointStatesSerializationHelper.deserialize(dis,
version);
+        if (version <= MessagingService.VERSION_10)
+            dis.readBoolean(); // 0.6 compatibility
+        int size = dis.readInt();
+        Map<InetAddress, EndpointState> epStateMap = new HashMap<InetAddress, EndpointState>(size);
+
+        for (int i = 0; i < size; ++i)
+        {
+            InetAddress ep = CompactEndpointSerializationHelper.deserialize(dis);
+            EndpointState epState = EndpointState.serializer().deserialize(dis, version);
+            epStateMap.put(ep, epState);
+        }
         return new GossipDigestAck(gDigestList, epStateMap);
     }
 
-    public long serializedSize(GossipDigestAck gossipDigestAckMessage, int version)
+    public long serializedSize(GossipDigestAck ack, int version)
     {
-        throw new UnsupportedOperationException();
+        int size = GossipDigestSerializationHelper.serializedSize(ack.gDigestList, version);
+        if (version <= MessagingService.VERSION_11)
+            size += DBTypeSizes.NATIVE.sizeof(true);
+        size += DBTypeSizes.NATIVE.sizeof(ack.epStateMap.size());
+        for (Map.Entry<InetAddress, EndpointState> entry : ack.epStateMap.entrySet())
+            size += CompactEndpointSerializationHelper.serializedSize(entry.getKey())
+                  + EndpointState.serializer().serializedSize(entry.getValue(), version);
+        return size;
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/gms/GossipDigestAck2.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAck2.java b/src/java/org/apache/cassandra/gms/GossipDigestAck2.java
index a95f43a..c77c223 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestAck2.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestAck2.java
@@ -19,9 +19,12 @@ package org.apache.cassandra.gms;
 
 import java.io.*;
 import java.net.InetAddress;
+import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.net.CompactEndpointSerializationHelper;
 
 
 /**
@@ -57,21 +60,38 @@ public class GossipDigestAck2
 
 class GossipDigestAck2Serializer implements IVersionedSerializer<GossipDigestAck2>
 {
-    public void serialize(GossipDigestAck2 gDigestAck2Message, DataOutput dos, int version)
throws IOException
+    public void serialize(GossipDigestAck2 ack2, DataOutput dos, int version) throws IOException
     {
-        /* Use the EndpointState */
-        EndpointStatesSerializationHelper.serialize(gDigestAck2Message.epStateMap, dos, version);
+        dos.writeInt(ack2.epStateMap.size());
+        for (Map.Entry<InetAddress, EndpointState> entry : ack2.epStateMap.entrySet())
+        {
+            InetAddress ep = entry.getKey();
+            CompactEndpointSerializationHelper.serialize(ep, dos);
+            EndpointState.serializer().serialize(entry.getValue(), dos, version);
+        }
     }
 
     public GossipDigestAck2 deserialize(DataInput dis, int version) throws IOException
     {
-        Map<InetAddress, EndpointState> epStateMap = EndpointStatesSerializationHelper.deserialize(dis,
version);
+        int size = dis.readInt();
+        Map<InetAddress, EndpointState> epStateMap = new HashMap<InetAddress, EndpointState>(size);
+
+        for (int i = 0; i < size; ++i)
+        {
+            InetAddress ep = CompactEndpointSerializationHelper.deserialize(dis);
+            EndpointState epState = EndpointState.serializer().deserialize(dis, version);
+            epStateMap.put(ep, epState);
+        }
         return new GossipDigestAck2(epStateMap);
     }
 
-    public long serializedSize(GossipDigestAck2 gossipDigestAck2Message, int version)
+    public long serializedSize(GossipDigestAck2 ack2, int version)
     {
-        throw new UnsupportedOperationException();
+        long size = DBTypeSizes.NATIVE.sizeof(ack2.epStateMap.size());
+        for (Map.Entry<InetAddress, EndpointState> entry : ack2.epStateMap.entrySet())
+            size += CompactEndpointSerializationHelper.serializedSize(entry.getKey())
+                  + EndpointState.serializer().serializedSize(entry.getValue(), version);
+        return size;
     }
 }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/gms/GossipDigestSyn.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/GossipDigestSyn.java b/src/java/org/apache/cassandra/gms/GossipDigestSyn.java
index e1dd59d..aaec57a 100644
--- a/src/java/org/apache/cassandra/gms/GossipDigestSyn.java
+++ b/src/java/org/apache/cassandra/gms/GossipDigestSyn.java
@@ -25,8 +25,10 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.net.CompactEndpointSerializationHelper;
+import org.apache.cassandra.utils.FBUtilities;
 
 
 /**
@@ -84,33 +86,13 @@ class GossipDigestSerializationHelper
         }
         return gDigests;
     }
-}
-
-class EndpointStatesSerializationHelper
-{
-    static void serialize(Map<InetAddress, EndpointState> epStateMap, DataOutput dos,
int version) throws IOException
+    
+    static int serializedSize(List<GossipDigest> digests, int version)
     {
-        dos.writeInt(epStateMap.size());
-        for (Entry<InetAddress, EndpointState> entry : epStateMap.entrySet())
-        {
-            InetAddress ep = entry.getKey();
-            CompactEndpointSerializationHelper.serialize(ep, dos);
-            EndpointState.serializer().serialize(entry.getValue(), dos, version);
-        }
-    }
-
-    static Map<InetAddress, EndpointState> deserialize(DataInput dis, int version)
throws IOException
-    {
-        int size = dis.readInt();
-        Map<InetAddress, EndpointState> epStateMap = new HashMap<InetAddress, EndpointState>(size);
-
-        for ( int i = 0; i < size; ++i )
-        {
-            InetAddress ep = CompactEndpointSerializationHelper.deserialize(dis);
-            EndpointState epState = EndpointState.serializer().deserialize(dis, version);
-            epStateMap.put(ep, epState);
-        }
-        return epStateMap;
+        int size = DBTypeSizes.NATIVE.sizeof(digests.size());
+        for (GossipDigest digest : digests)
+            size += GossipDigest.serializer().serializedSize(digest, version);
+        return size;
     }
 }
 
@@ -129,9 +111,9 @@ class GossipDigestSynSerializer implements IVersionedSerializer<GossipDigestSyn>
         return new GossipDigestSyn(clusterId, gDigests);
     }
 
-    public long serializedSize(GossipDigestSyn gossipDigestSynMessage, int version)
+    public long serializedSize(GossipDigestSyn syn, int version)
     {
-        throw new UnsupportedOperationException();
+        return FBUtilities.serializedUTF8Size(syn.clusterId) + GossipDigestSerializationHelper.serializedSize(syn.gDigests,
version);
     }
 }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/gms/GossipShutdownMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/GossipShutdownMessage.java b/src/java/org/apache/cassandra/gms/GossipShutdownMessage.java
index 3122986..d3cdaf1 100644
--- a/src/java/org/apache/cassandra/gms/GossipShutdownMessage.java
+++ b/src/java/org/apache/cassandra/gms/GossipShutdownMessage.java
@@ -58,6 +58,6 @@ class GossipShutdownMessageSerializer implements IVersionedSerializer<GossipShut
 
     public long serializedSize(GossipShutdownMessage gossipShutdownMessage, int version)
     {
-        throw new UnsupportedOperationException();
+        return 0;
     }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/gms/HeartBeatState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/HeartBeatState.java b/src/java/org/apache/cassandra/gms/HeartBeatState.java
index 294d460..bdbb6a3 100644
--- a/src/java/org/apache/cassandra/gms/HeartBeatState.java
+++ b/src/java/org/apache/cassandra/gms/HeartBeatState.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.gms;
 
 import java.io.*;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 
 
@@ -88,8 +89,8 @@ class HeartBeatStateSerializer implements IVersionedSerializer<HeartBeatState>
         return new HeartBeatState(dis.readInt(), dis.readInt());
     }
 
-    public long serializedSize(HeartBeatState heartBeatState, int version)
+    public long serializedSize(HeartBeatState state, int version)
     {
-        throw new UnsupportedOperationException();
+        return DBTypeSizes.NATIVE.sizeof(state.getGeneration()) + DBTypeSizes.NATIVE.sizeof(state.getHeartBeatVersion());
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/gms/VersionedValue.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/VersionedValue.java b/src/java/org/apache/cassandra/gms/VersionedValue.java
index 10dbb6a..807eb88 100644
--- a/src/java/org/apache/cassandra/gms/VersionedValue.java
+++ b/src/java/org/apache/cassandra/gms/VersionedValue.java
@@ -21,6 +21,7 @@ import java.io.*;
 import java.net.InetAddress;
 import java.util.UUID;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.io.IVersionedSerializer;
@@ -199,6 +200,12 @@ public class VersionedValue implements Comparable<VersionedValue>
     {
         public void serialize(VersionedValue value, DataOutput dos, int version) throws IOException
         {
+            dos.writeUTF(outValue(value, version));
+            dos.writeInt(value.version);
+        }
+
+        private String outValue(VersionedValue value, int version)
+        {
             String outValue = value.value;
 
             if (version < MessagingService.VERSION_12)
@@ -206,20 +213,17 @@ public class VersionedValue implements Comparable<VersionedValue>
                 String[] pieces = value.value.split(DELIMITER_STR, -1);
                 String type = pieces[0];
 
-                if ((type == STATUS_NORMAL) || type == STATUS_BOOTSTRAPPING)
+                if ((type.equals(STATUS_NORMAL)) || type.equals(STATUS_BOOTSTRAPPING))
                 {
                     assert pieces.length >= 3;
                     outValue = versionString(pieces[0], pieces[2]);
                 }
 
-                if ((type == REMOVAL_COORDINATOR) || (type == REMOVING_TOKEN) || (type ==
REMOVED_TOKEN))
+                if ((type.equals(REMOVAL_COORDINATOR)) || (type.equals(REMOVING_TOKEN)) ||
(type.equals(REMOVED_TOKEN)))
                     throw new RuntimeException(String.format("Unable to serialize %s(%s...)
for nodes older than 1.2",
-                                                             VersionedValue.class.getName(),
-                                                             type));
+                                                             VersionedValue.class.getName(),
type));
             }
-
-            dos.writeUTF(outValue);
-            dos.writeInt(value.version);
+            return outValue;
         }
 
         public VersionedValue deserialize(DataInput dis, int version) throws IOException
@@ -231,7 +235,8 @@ public class VersionedValue implements Comparable<VersionedValue>
 
         public long serializedSize(VersionedValue value, int version)
         {
-            throw new UnsupportedOperationException();
+            int outLength = FBUtilities.encodedUTF8Length(outValue(value, version));
+            return DBTypeSizes.NATIVE.sizeof(outLength) + outLength + DBTypeSizes.NATIVE.sizeof(value.version);
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/service/MigrationManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/MigrationManager.java b/src/java/org/apache/cassandra/service/MigrationManager.java
index 3e88984..7a9b0d7 100644
--- a/src/java/org/apache/cassandra/service/MigrationManager.java
+++ b/src/java/org/apache/cassandra/service/MigrationManager.java
@@ -40,7 +40,6 @@ import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.filter.QueryFilter;
 import org.apache.cassandra.db.filter.QueryPath;
-import org.apache.cassandra.db.DBConstants;
 import org.apache.cassandra.gms.*;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.net.MessageOut;
@@ -328,11 +327,10 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
 
         public long serializedSize(Collection<RowMutation> schema, int version)
         {
-            int size = DBConstants.INT_SIZE;
+            int size = DBTypeSizes.NATIVE.sizeof(schema.size());
             for (RowMutation rm : schema)
                 size += RowMutation.serializer().serializedSize(rm, version);
             return size;
         }
     }
-
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/streaming/PendingFile.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/PendingFile.java b/src/java/org/apache/cassandra/streaming/PendingFile.java
index 1554f85..41ccf46 100644
--- a/src/java/org/apache/cassandra/streaming/PendingFile.java
+++ b/src/java/org/apache/cassandra/streaming/PendingFile.java
@@ -21,10 +21,12 @@ import java.io.*;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.SSTableReader;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Pair;
 
 /**
@@ -147,9 +149,21 @@ public class PendingFile
             return new PendingFile(null, desc, component, sections, type, estimatedKeys);
         }
 
-        public long serializedSize(PendingFile pendingFile, int version)
+        public long serializedSize(PendingFile pf, int version)
         {
-            throw new UnsupportedOperationException();
+            if (pf == null)
+                return DBTypeSizes.NATIVE.sizeof(0);
+
+            long size = FBUtilities.serializedUTF8Size(pf.desc.filenameFor(pf.component));
+            size += FBUtilities.serializedUTF8Size(pf.component);
+            size += DBTypeSizes.NATIVE.sizeof(pf.sections.size());
+            for (Pair<Long,Long> section : pf.sections)
+                size += DBTypeSizes.NATIVE.sizeof(section.left + DBTypeSizes.NATIVE.sizeof(section.right));
+            if (version > MessagingService.VERSION_07)
+                size += FBUtilities.serializedUTF8Size(pf.type.name());
+            if (version > MessagingService.VERSION_080)
+                size += DBTypeSizes.NATIVE.sizeof(pf.estimatedKeys);
+            return size;
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/streaming/StreamHeader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamHeader.java b/src/java/org/apache/cassandra/streaming/StreamHeader.java
index 00d0049..b98aab5 100644
--- a/src/java/org/apache/cassandra/streaming/StreamHeader.java
+++ b/src/java/org/apache/cassandra/streaming/StreamHeader.java
@@ -25,6 +25,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.net.CompactEndpointSerializationHelper;
 import org.apache.cassandra.net.MessagingService;
@@ -110,9 +111,16 @@ public class StreamHeader
             return new StreamHeader(table, sessionId, file, pendingFiles, bca);
         }
 
-        public long serializedSize(StreamHeader streamHeader, int version)
+        public long serializedSize(StreamHeader sh, int version)
         {
-            throw new UnsupportedOperationException();
-        }
+            long size = FBUtilities.serializedUTF8Size(sh.table);
+            size += DBTypeSizes.NATIVE.sizeof(sh.sessionId);
+            size += PendingFile.serializer().serializedSize(sh.file, version);
+            size += DBTypeSizes.NATIVE.sizeof(sh.pendingFiles.size());
+            for(PendingFile file : sh.pendingFiles)
+                size += PendingFile.serializer().serializedSize(file, version);
+            size += CompactEndpointSerializationHelper.serializedSize(sh.broadcastAddress);
+            return size;
+       }
     }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/streaming/StreamReply.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamReply.java b/src/java/org/apache/cassandra/streaming/StreamReply.java
index d207d6e..55ded17 100644
--- a/src/java/org/apache/cassandra/streaming/StreamReply.java
+++ b/src/java/org/apache/cassandra/streaming/StreamReply.java
@@ -21,9 +21,11 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.net.MessageOut;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.FBUtilities;
 
 public class StreamReply
 {
@@ -80,9 +82,9 @@ public class StreamReply
             return new StreamReply(targetFile, sessionId, action);
         }
 
-        public long serializedSize(StreamReply streamReply, int version)
+        public long serializedSize(StreamReply reply, int version)
         {
-            throw new UnsupportedOperationException();
+            return DBTypeSizes.NATIVE.sizeof(reply.sessionId) + FBUtilities.serializedUTF8Size(reply.file)
+ DBTypeSizes.NATIVE.sizeof(reply.action.ordinal());
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/streaming/StreamRequest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamRequest.java b/src/java/org/apache/cassandra/streaming/StreamRequest.java
index 3814049..fa1d864 100644
--- a/src/java/org/apache/cassandra/streaming/StreamRequest.java
+++ b/src/java/org/apache/cassandra/streaming/StreamRequest.java
@@ -28,6 +28,7 @@ import java.util.List;
 import com.google.common.collect.Iterables;
 
 import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.db.Table;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.Range;
@@ -36,6 +37,7 @@ import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.net.CompactEndpointSerializationHelper;
 import org.apache.cassandra.net.MessageOut;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.FBUtilities;
 
 /**
 * This class encapsulates the message that needs to be sent to nodes
@@ -138,9 +140,7 @@ public class StreamRequest
                 dos.writeUTF(srm.table);
                 dos.writeInt(srm.ranges.size());
                 for (Range<Token> range : srm.ranges)
-                {
                     AbstractBounds.serializer().serialize(range, dos, version);
-                }
 
                 if (version > MessagingService.VERSION_07)
                     dos.writeUTF(srm.type.name());
@@ -170,9 +170,7 @@ public class StreamRequest
                 int size = dis.readInt();
                 List<Range<Token>> ranges = (size == 0) ? null : new ArrayList<Range<Token>>(size);
                 for( int i = 0; i < size; ++i )
-                {
                     ranges.add((Range<Token>) AbstractBounds.serializer().deserialize(dis,
version).toTokenBounds());
-                }
                 OperationType type = OperationType.RESTORE_REPLICA_COUNT;
                 if (version > MessagingService.VERSION_07)
                     type = OperationType.valueOf(dis.readUTF());
@@ -189,9 +187,27 @@ public class StreamRequest
             }
         }
 
-        public long serializedSize(StreamRequest streamRequestMessage, int version)
+        public long serializedSize(StreamRequest sr, int version)
         {
-            throw new UnsupportedOperationException();
+            long size = DBTypeSizes.NATIVE.sizeof(sr.sessionId);
+            size += CompactEndpointSerializationHelper.serializedSize(sr.target);
+            size += DBTypeSizes.NATIVE.sizeof(true);
+            if (sr.file != null)
+                return size + PendingFile.serializer().serializedSize(sr.file, version);
+
+            size += FBUtilities.serializedUTF8Size(sr.table);
+            size += DBTypeSizes.NATIVE.sizeof(sr.ranges.size());
+            for (Range<Token> range : sr.ranges)
+                size += AbstractBounds.serializer().serializedSize(range, version);
+            if (version > MessagingService.VERSION_07)
+                size += FBUtilities.serializedUTF8Size(sr.type.name());
+            if (version > MessagingService.VERSION_080)
+            {
+                size += DBTypeSizes.NATIVE.sizeof(Iterables.size(sr.columnFamilies));
+                for (ColumnFamilyStore cfs : sr.columnFamilies)
+                    size += DBTypeSizes.NATIVE.sizeof(cfs.metadata.cfId);
+            }
+            return size;
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java b/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java
index 3074bc0..947de9b 100644
--- a/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java
+++ b/src/java/org/apache/cassandra/streaming/StreamingRepairTask.java
@@ -28,6 +28,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.db.Table;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.Range;
@@ -38,6 +39,7 @@ import org.apache.cassandra.net.*;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.UUIDGen;
 
+
 /**
  * Task that make two nodes exchange (stream) some ranges (for a given table/cf).
  * This handle the case where the local node is neither of the two nodes that
@@ -265,7 +267,14 @@ public class StreamingRepairTask implements Runnable
 
         public long serializedSize(StreamingRepairTask task, int version)
         {
-            throw new UnsupportedOperationException();
+            long size = UUIDGen.serializer.serializedSize(task.id, version);
+            size += 3 * CompactEndpointSerializationHelper.serializedSize(task.owner);
+            size += FBUtilities.serializedUTF8Size(task.tableName);
+            size += FBUtilities.serializedUTF8Size(task.cfName);
+            size += DBTypeSizes.NATIVE.sizeof(task.ranges.size());
+            for (Range<Token> range : task.ranges)
+                size += AbstractBounds.serializer().serializedSize(range, version);
+            return size;
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/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 ac55d08..8b5e2f3 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -42,6 +42,7 @@ import org.apache.cassandra.cache.IRowCacheProvider;
 import org.apache.cassandra.concurrent.CreationTimeAwareFuture;
 import org.apache.cassandra.config.ConfigurationException;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Range;
@@ -585,6 +586,12 @@ public class FBUtilities
         }
     }
 
+    public static int serializedUTF8Size(String st)
+    {
+        int length = encodedUTF8Length(st);
+        return DBTypeSizes.NATIVE.sizeof(length) + length;
+    }
+
     private static final class WrappedCloseableIterator<T>
         extends AbstractIterator<T> implements CloseableIterator<T>
     {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/utils/MerkleTree.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/MerkleTree.java b/src/java/org/apache/cassandra/utils/MerkleTree.java
index 9b56d76..2cbecb8 100644
--- a/src/java/org/apache/cassandra/utils/MerkleTree.java
+++ b/src/java/org/apache/cassandra/utils/MerkleTree.java
@@ -26,7 +26,7 @@ import java.util.*;
 import com.google.common.collect.AbstractIterator;
 import com.google.common.collect.PeekingIterator;
 
-import org.apache.cassandra.db.DBConstants;
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
@@ -103,9 +103,12 @@ public class MerkleTree implements Serializable
             return mt;
         }
 
-        public long serializedSize(MerkleTree merkleTree, int version)
+        public long serializedSize(MerkleTree mt, int version)
         {
-            return 1 + DBConstants.LONG_SIZE + DBConstants.LONG_SIZE + Hashable.serializer.serializedSize(merkleTree.root,
version);
+            return DBTypeSizes.NATIVE.sizeof(mt.hashdepth)
+                 + DBTypeSizes.NATIVE.sizeof(mt.maxsize)
+                 + DBTypeSizes.NATIVE.sizeof(mt.size)
+                 + Hashable.serializer.serializedSize(mt.root, version);
         }
     }
 
@@ -711,8 +714,11 @@ public class MerkleTree implements Serializable
 
             public long serializedSize(Inner inner, int version)
             {
-                int size = inner.hash == null ? DBConstants.INT_SIZE : DBConstants.INT_SIZE
+ inner.hash.length;
-                size += Token.serializer().serializedSize(inner.token)
+                int size = inner.hash == null
+                         ? DBTypeSizes.NATIVE.sizeof(-1)
+                         : DBTypeSizes.NATIVE.sizeof(inner.hash().length) + inner.hash().length;
+
+                size += Token.serializer().serializedSize(inner.token, DBTypeSizes.NATIVE)
                         + Hashable.serializer.serializedSize(inner.lchild, version)
                         + Hashable.serializer.serializedSize(inner.rchild, version);
                 return size;
@@ -790,7 +796,9 @@ public class MerkleTree implements Serializable
 
             public long serializedSize(Leaf leaf, int version)
             {
-                return leaf.hash == null ? DBConstants.INT_SIZE : DBConstants.INT_SIZE +
leaf.hash.length;
+                return leaf.hash == null
+                     ? DBTypeSizes.NATIVE.sizeof(-1)
+                     : DBTypeSizes.NATIVE.sizeof(leaf.hash().length) + leaf.hash().length;
             }
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b9fc26c/src/java/org/apache/cassandra/utils/UUIDGen.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/UUIDGen.java b/src/java/org/apache/cassandra/utils/UUIDGen.java
index 19f2e9b..b83d16d 100644
--- a/src/java/org/apache/cassandra/utils/UUIDGen.java
+++ b/src/java/org/apache/cassandra/utils/UUIDGen.java
@@ -27,7 +27,7 @@ import java.util.Map;
 import java.util.Random;
 import java.util.UUID;
 
-import org.apache.cassandra.db.DBConstants;
+import org.apache.cassandra.db.DBTypeSizes;
 import org.apache.cassandra.io.IVersionedSerializer;
 
 /**
@@ -110,7 +110,7 @@ public class UUIDGen
 
         public long serializedSize(UUID uuid, int version)
         {
-            return DBConstants.LONG_SIZE + DBConstants.LONG_SIZE;
+            return DBTypeSizes.NATIVE.sizeof(uuid.getMostSignificantBits()) + DBTypeSizes.NATIVE.sizeof(uuid.getLeastSignificantBits());
         }
     }
 


Mime
View raw message