cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From xe...@apache.org
Subject svn commit: r1164689 - in /cassandra/trunk: ./ lib/ lib/licenses/ src/java/org/apache/cassandra/net/ src/java/org/apache/cassandra/streaming/
Date Fri, 02 Sep 2011 20:23:42 GMT
Author: xedin
Date: Fri Sep  2 20:23:42 2011
New Revision: 1164689

URL: http://svn.apache.org/viewvc?rev=1164689&view=rev
Log:
Streams Compression
patch by Pavel Yaskevich; reviewed by Jonathan Ellis for CASSANDRA-3015

Added:
    cassandra/trunk/lib/compress-lzf-0.8.4.jar
    cassandra/trunk/lib/licenses/compress-lzf-0.8.4.txt   (with props)
Modified:
    cassandra/trunk/CHANGES.txt
    cassandra/trunk/src/java/org/apache/cassandra/net/IncomingTcpConnection.java
    cassandra/trunk/src/java/org/apache/cassandra/streaming/FileStreamTask.java
    cassandra/trunk/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java

Modified: cassandra/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/CHANGES.txt?rev=1164689&r1=1164688&r2=1164689&view=diff
==============================================================================
--- cassandra/trunk/CHANGES.txt (original)
+++ cassandra/trunk/CHANGES.txt Fri Sep  2 20:23:42 2011
@@ -58,6 +58,7 @@
  * make the repair of a range repair all replica (CASSANDRA-2610)
  * expose the ability to repair the first range (as returned by the
    partitioner) of a node (CASSANDRA-2606)
+ * Streams Compression (CASSANDRA-3015)
 
 0.8.5
  * fix NPE when encryption_options is unspecified (CASSANDRA-3007)

Added: cassandra/trunk/lib/compress-lzf-0.8.4.jar
URL: http://svn.apache.org/viewvc/cassandra/trunk/lib/compress-lzf-0.8.4.jar?rev=1164689&view=auto
==============================================================================
Files cassandra/trunk/lib/compress-lzf-0.8.4.jar (added) and cassandra/trunk/lib/compress-lzf-0.8.4.jar
Fri Sep  2 20:23:42 2011 differ

Added: cassandra/trunk/lib/licenses/compress-lzf-0.8.4.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/lib/licenses/compress-lzf-0.8.4.txt?rev=1164689&view=auto
==============================================================================
--- cassandra/trunk/lib/licenses/compress-lzf-0.8.4.txt (added)
+++ cassandra/trunk/lib/licenses/compress-lzf-0.8.4.txt Fri Sep  2 20:23:42 2011
@@ -0,0 +1,11 @@
+Copyright 2009-2010 Ning, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License"); you may not
+use this file except in compliance with the License. You may obtain a copy of
+the License at http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,WITHOUT
+WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+License for the specific language governing permissions and limitations under
+the License.
\ No newline at end of file

Propchange: cassandra/trunk/lib/licenses/compress-lzf-0.8.4.txt
------------------------------------------------------------------------------
    svn:executable = *

Modified: cassandra/trunk/src/java/org/apache/cassandra/net/IncomingTcpConnection.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/net/IncomingTcpConnection.java?rev=1164689&r1=1164688&r2=1164689&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/net/IncomingTcpConnection.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/net/IncomingTcpConnection.java Fri Sep 
2 20:23:42 2011
@@ -22,6 +22,7 @@ package org.apache.cassandra.net;
 
 
 import java.io.*;
+import java.net.InetSocketAddress;
 import java.net.Socket;
 
 import org.apache.cassandra.gms.Gossiper;
@@ -78,7 +79,7 @@ public class IncomingTcpConnection exten
                 else
                 {
                     // streaming connections are per-session and have a fixed version.  we
can't do anything with a new-version stream connection, so drop it.
-                    logger.error("Received untranslated stream from newer protcol version.
Terminating connection!");
+                    logger.error("Received untranslated stream from newer protocol version.
Terminating connection!");
                 }
                 // We are done with this connection....
                 return;

Modified: cassandra/trunk/src/java/org/apache/cassandra/streaming/FileStreamTask.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/streaming/FileStreamTask.java?rev=1164689&r1=1164688&r2=1164689&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/streaming/FileStreamTask.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/streaming/FileStreamTask.java Fri Sep  2
20:23:42 2011
@@ -18,9 +18,9 @@
 
 package org.apache.cassandra.streaming;
 
-import java.io.DataOutputStream;
 import java.io.File;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
@@ -40,6 +40,8 @@ import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.Throttle;
 import org.apache.cassandra.utils.WrappedRunnable;
 
+import com.ning.compress.lzf.LZFOutputStream;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,7 +60,7 @@ public class FileStreamTask extends Wrap
     // communication socket
     private Socket socket;
     // socket's output stream
-    private DataOutputStream output;
+    private OutputStream output;
     // system encryption options if any
     private final EncryptionOptions encryptionOptions;
     // allocate buffer to use for transfers only once
@@ -119,7 +121,7 @@ public class FileStreamTask extends Wrap
     private void stream() throws IOException
     {
         ByteBuffer HeaderBuffer = MessagingService.instance().constructStreamHeader(header,
false, Gossiper.instance.getVersion(to));
-        // write header
+        // write header (this should not be compressed for compatibility with other messages)
         output.write(ByteBufferUtil.getArray(HeaderBuffer));
 
         if (header.file == null)
@@ -129,6 +131,9 @@ public class FileStreamTask extends Wrap
                                     ? CompressedRandomAccessReader.open(header.file.getFilename(),
true)
                                     : RandomAccessReader.open(new File(header.file.getFilename()),
CHUNK_SIZE, true);
 
+        // setting up data compression stream
+        output = new LZFOutputStream(output);
+
         try
         {
             // stream each of the required sections of the file
@@ -234,12 +239,12 @@ public class FileStreamTask extends Wrap
     protected void connect() throws IOException
     {
         socket.connect(new InetSocketAddress(to, DatabaseDescriptor.getStoragePort()));
-        output = new DataOutputStream(socket.getOutputStream());
+        output = socket.getOutputStream();
     }
 
     protected void close() throws IOException
     {
-        socket.close();
+        output.close();
     }
 
     public String toString()

Modified: cassandra/trunk/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java?rev=1164689&r1=1164688&r2=1164689&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java Fri
Sep  2 20:23:42 2011
@@ -23,9 +23,6 @@ import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.util.Collections;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ColumnFamily;
 import org.apache.cassandra.db.ColumnFamilyStore;
@@ -41,6 +38,11 @@ import org.apache.cassandra.utils.ByteBu
 import org.apache.cassandra.utils.BytesReadTracker;
 import org.apache.cassandra.utils.Pair;
 
+import com.ning.compress.lzf.LZFInputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 public class IncomingStreamReader
 {
     private static final Logger logger = LoggerFactory.getLogger(IncomingStreamReader.class);
@@ -79,7 +81,7 @@ public class IncomingStreamReader
             assert remoteFile.estimatedKeys > 0;
             SSTableReader reader = null;
             logger.debug("Estimated keys {}", remoteFile.estimatedKeys);
-            DataInputStream dis = new DataInputStream(socket.getInputStream());
+            DataInputStream dis = new DataInputStream(new LZFInputStream(socket.getInputStream()));
             try
             {
                 reader = streamIn(dis, localFile, remoteFile);



Mime
View raw message