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

add inter-node compression 

patch by Marcus Eriksson; reviewed by jbellis for CASSANDRA-3127
This commit is contained in:
Jonathan Ellis 2012-06-05 21:27:44 -05:00
parent 90ce7ef210
commit 625e61396b
7 changed files with 72 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
1.2-dev
* add inter-node message compression (CASSANDRA-3127)
* enforce 1m min keycache for auto (CASSANDRA-4306)
* remove COPP (CASSANDRA-2479)
* Track tombstone expiration and compact when tombstone content is

7
conf/cassandra.yaml
View File

@ -553,3 +553,10 @@ encryption_options:
# algorithm: SunX509
# store_type: JKS
# cipher_suites: [TLS_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA]
# internode_compression controls whether traffic between nodes is
# compressed.
# can be: all - all traffic is compressed
# dc - traffic between different datacenters is compressed
# none - nothing is compressed.
internode_compression: all

13
src/java/org/apache/cassandra/config/Config.java
View File

@ -112,6 +112,8 @@ public class Config
public EncryptionOptions encryption_options = new EncryptionOptions();
public InternodeCompression internode_compression = InternodeCompression.none;
public Integer index_interval = 128;
public Double flush_largest_memtables_at = 1.0;
@ -158,12 +160,19 @@ public class Config
loadYaml = value;
}
public static enum CommitLogSync {
public static enum CommitLogSync
{
periodic,
batch
}
public static enum DiskAccessMode {
public static enum InternodeCompression
{
all, none, dc
}
public static enum DiskAccessMode
{
auto,
mmap,
mmap_index_only,

5
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -1059,4 +1059,9 @@ public class DatabaseDescriptor
{
return conf.populate_io_cache_on_flush;
}
public static Config.InternodeCompression internodeCompression()
{
return conf.internode_compression;
}
}

17
src/java/org/apache/cassandra/net/IncomingTcpConnection.java
View File

@ -26,6 +26,7 @@ import java.net.Socket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xerial.snappy.SnappyInputStream;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.util.FastByteArrayInputStream;
@ -53,17 +54,15 @@ public class IncomingTcpConnection extends Thread
@Override
public void run()
{
DataInputStream input;
boolean isStream;
int version;
try
{
// determine the connection type to decide whether to buffer
input = new DataInputStream(socket.getInputStream());
DataInputStream input = new DataInputStream(socket.getInputStream());
MessagingService.validateMagic(input.readInt());
int header = input.readInt();
isStream = MessagingService.getBits(header, 3, 1) == 1;
version = MessagingService.getBits(header, 15, 8);
boolean compressed = MessagingService.getBits(header, 2, 1) == 1;
boolean isStream = MessagingService.getBits(header, 3, 1) == 1;
int version = MessagingService.getBits(header, 15, 8);
if (isStream)
{
if (version == MessagingService.current_version)
@ -97,7 +96,11 @@ public class IncomingTcpConnection extends Thread
}
Gossiper.instance.setVersion(from, version);
logger.debug("set version for {} to {}", from, version);
if (compressed)
{
logger.debug("Upgrading incoming connection to be compressed");
input = new DataInputStream(new SnappyInputStream(input));
}
// loop to get the next message.
while (true)
{

40
src/java/org/apache/cassandra/net/OutboundTcpConnection.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.net;
import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
@ -27,10 +28,11 @@ import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xerial.snappy.SnappyOutputStream;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.utils.FBUtilities;
public class OutboundTcpConnection extends Thread
{
@ -51,13 +53,23 @@ public class OutboundTcpConnection extends Thread
private Socket socket;
private volatile long completed;
private final AtomicLong dropped = new AtomicLong();
private boolean isUpgraded = false;
private boolean writeToLocalDC;
public OutboundTcpConnection(OutboundTcpConnectionPool pool)
{
super("WRITE-" + pool.endPoint());
this.writeToLocalDC = isSameDC(pool.endPoint());
this.poolReference = pool;
}
private static boolean isSameDC(InetAddress targetHost)
{
String remoteDC = DatabaseDescriptor.getEndpointSnitch().getDatacenter(targetHost);
String thisDC = DatabaseDescriptor.getEndpointSnitch().getDatacenter(DatabaseDescriptor.getRpcAddress());
return remoteDC.equals(thisDC);
}
public void enqueue(MessageOut<?> message, String id)
{
expireMessages();
@ -137,6 +149,15 @@ public class OutboundTcpConnection extends Thread
return dropped.get();
}
private boolean shouldUpgradeConnection()
{
if(Gossiper.instance.getVersion(poolReference.endPoint()) >= MessagingService.current_version &&
(DatabaseDescriptor.internodeCompression() == Config.InternodeCompression.all ||
(DatabaseDescriptor.internodeCompression() == Config.InternodeCompression.dc && !writeToLocalDC)))
return true;
return false;
}
private void writeConnected(MessageOut<?> message, String id)
{
try
@ -147,6 +168,13 @@ public class OutboundTcpConnection extends Thread
{
out.flush();
}
if(!isUpgraded && shouldUpgradeConnection())
{
out.flush();
logger.debug("Upgrading OutputStream to be compressed");
out = new DataOutputStream(new SnappyOutputStream(new BufferedOutputStream(socket.getOutputStream())));
isUpgraded = true;
}
}
catch (Exception e)
{
@ -161,10 +189,10 @@ public class OutboundTcpConnection extends Thread
public void write(MessageOut<?> message, String id, DataOutputStream out) throws IOException
{
write(message, id, out, Gossiper.instance.getVersion(poolReference.endPoint()));
write(message, id, out, Gossiper.instance.getVersion(poolReference.endPoint()), shouldUpgradeConnection());
}
public static void write(MessageOut message, String id, DataOutputStream out, int version) throws IOException
public static void write(MessageOut message, String id, DataOutputStream out, int version, boolean compressionEnabled) throws IOException
{
/*
Setting up the protocol header. This is 4 bytes long
@ -180,7 +208,7 @@ public class OutboundTcpConnection extends Thread
// Setting up the serializer bit
header |= MessagingService.serializerType.ordinal();
// set compression bit.
if (false)
if (compressionEnabled)
header |= 4;
// Setting up the version bit
header |= (version << 8);
@ -216,6 +244,7 @@ public class OutboundTcpConnection extends Thread
private boolean connect()
{
isUpgraded = false;
if (logger.isDebugEnabled())
logger.debug("attempting to connect to " + poolReference.endPoint());
long start = System.currentTimeMillis();
@ -244,6 +273,9 @@ public class OutboundTcpConnection extends Thread
}
}
}
// nodes can change dc during the lifetime of this OutboundTcpConnection, but we cannot upgrade a
// connection mid-flight, so it is safe to set here and dont change until the next connect()
this.writeToLocalDC = isSameDC(poolReference.endPoint());
return false;
}

4
src/java/org/apache/cassandra/streaming/StreamInSession.java
View File

@ -171,7 +171,7 @@ public class StreamInSession extends AbstractStreamSession
OutboundTcpConnection.write(message,
String.valueOf(getSessionId()),
out,
Gossiper.instance.getVersion(getHost()));
Gossiper.instance.getVersion(getHost()), false);
out.flush();
}
@ -222,7 +222,7 @@ public class StreamInSession extends AbstractStreamSession
OutboundTcpConnection.write(reply.createMessage(),
context.right.toString(),
new DataOutputStream(socket.getOutputStream()),
Gossiper.instance.getVersion(getHost()));
Gossiper.instance.getVersion(getHost()), false);
else
logger.debug("No socket to reply to {} with!", getHost());
}