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

merge from 0.7 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1052105 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-12-22 23:56:23 +00:00
parent 98d9eae14e b17aefbcf5
commit 44f0214b1d
26 changed files with 296 additions and 183 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.rat-excludes
View File

@ -13,7 +13,10 @@ build/**
lib/licenses/*.txt
.settings/**
contrib/pig/example-script.pig
contrib/redhat/cassandra
**/cassandra.yaml
**/*.db
redhat/apache-cassandra.spec
redhat/cassandra
redhat/cassandra.conf
redhat/cassandra.in.sh
redhat/default

7
CHANGES.txt
View File

@ -4,6 +4,13 @@
0.7-dev
* fix cli crash after backgrounding (CASSANDRA-1875)
* count timeouts in storageproxy latencies, and include latency
histograms in StorageProxyMBean (CASSANDRA-1893)
* check log4j configuration for changes every 10s (CASSANDRA-1525)
0.7.0-rc3
* expose getNaturalEndpoints in StorageServiceMBean taking byte[]
key; RMI cannot serialize ByteBuffer (CASSANDRA-1833)
* infer org.apache.cassandra.locator for replication strategy classes

2
bin/cassandra
View File

@ -108,7 +108,7 @@ launch_service()
foreground=$2
props=$3
class=$4
cassandra_parms="-Dlog4j.configuration=log4j-server.properties"
cassandra_parms="-Dlog4j.configuration=log4j-server.properties -Dlog4j.defaultInitOverride=true"
if [ "x$pidpath" != "x" ]; then
cassandra_parms="$cassandra_parms -Dcassandra-pidfile=$pidpath"

3
bin/cassandra.bat
View File

@ -37,7 +37,8 @@ set JAVA_OPTS=^
-Dcom.sun.management.jmxremote.port=8080^
-Dcom.sun.management.jmxremote.ssl=false^
-Dcom.sun.management.jmxremote.authenticate=false^
-Dlog4j.configuration=log4j-server.properties
-Dlog4j.configuration=log4j-server.properties^
-Dlog4j.defaultInitOverride=true
REM ***** CLASSPATH library setting *****

2
build.xml
View File

@ -47,7 +47,7 @@
<property name="test.unit.src" value="${test.dir}/unit"/>
<property name="test.long.src" value="${test.dir}/long"/>
<property name="dist.dir" value="${build.dir}/dist"/>
<property name="base.version" value="0.7.0-rc2"/>
<property name="base.version" value="0.7.0-rc3"/>
<condition property="version" value="${base.version}">
<isset property="release"/>
</condition>

7
conf/log4j-server.properties
View File

@ -35,6 +35,7 @@ log4j.appender.R.layout.ConversionPattern=%5p [%t] %d{ISO8601} %F (line %L) %m%n
log4j.appender.R.File=/var/log/cassandra/system.log
# Application logging options
#log4j.logger.com.facebook=DEBUG
#log4j.logger.com.facebook.infrastructure.gms=DEBUG
#log4j.logger.com.facebook.infrastructure.db=DEBUG
#log4j.logger.org.apache.cassandra=DEBUG
#log4j.logger.org.apache.cassandra.db=DEBUG
#log4j.logger.org.apache.cassandra.service.StorageProxy=DEBUG

6
debian/changelog vendored
View File

@ -1,3 +1,9 @@
cassandra (0.7.0~rc3) unstable; urgency=low
* Release candidate release.
-- Eric Evans <eevans@apache.org> Tue, 21 Dec 2010 17:36:31 -0600
cassandra (0.7.0~rc2) unstable; urgency=low
* Release candidate release.

20
redhat/apache-cassandra.spec
View File

@ -2,22 +2,27 @@
%global username cassandra
%define relname %{name}-%{version}-%{release}
Name: apache-cassandra
Version: 0.7.0
Release: 1%{?dist}
Release: rc3
Summary: Cassandra is a highly scalable, eventually consistent, distributed, structured key-value store.
Group: Development/Libraries
License: BSD
License: Apache Software License
URL: http://cassandra.apache.org/
Source0: http://www.ibiblio.org/pub/mirrors/apache/%{username}/%{version}/%{name}-%{version}-src.tar.gz
BuildRoot: %{_tmppath}/%{name}-%{version}-%{release}-root-%(%{__id_u} -n)
Source0: http://www.ibiblio.org/pub/mirrors/apache/%{username}/%{version}/%{relname}-src.tar.gz
BuildRoot: %{_tmppath}/%{relname}-root-%(%{__id_u} -n)
BuildRequires: java-devel
BuildRequires: jpackage-utils
BuildRequires: ant
BuildRequires: ant-nodeps
Conflicts: cassandra
Obsoletes: cassandra07
Requires: java >= 1.6.0
Requires: jna >= 3.2.7
Requires: jpackage-utils
@ -35,10 +40,10 @@ and the data model from Google's BigTable. Like Dynamo, Cassandra is
eventually consistent. Like BigTable, Cassandra provides a ColumnFamily-based
data model richer than typical key/value systems.
For more information see http://incubator.apache.org/cassandra
For more information see http://cassandra.apache.org/
%prep
%setup -q -n %{name}-%{version}-src
%setup -q -n %{relname}-src
%build
ant clean jar -Drelease=true
@ -66,7 +71,8 @@ rm bin/cassandra.in.sh
mv bin/cassandra %{buildroot}/usr/sbin
rm bin/*.bat
cp -p bin/* %{buildroot}/usr/bin
cp build/%{name}-%{version}.jar %{buildroot}/usr/share/%{username}/lib
# Handle the case of interim SNAPHOST builds
cp build/%{relname}*jar %{buildroot}/usr/share/%{username}/lib
mkdir -p %{buildroot}/var/lib/%{username}/commitlog
mkdir -p %{buildroot}/var/lib/%{username}/data
mkdir -p %{buildroot}/var/lib/%{username}/saved_caches

9
src/java/org/apache/cassandra/auth/SimpleAuthenticator.java
View File

@ -28,6 +28,7 @@ import java.util.Properties;
import java.util.Map;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.thrift.AuthenticationException;
import org.apache.cassandra.utils.FBUtilities;
@ -92,12 +93,12 @@ public class SimpleAuthenticator implements IAuthenticator
boolean authenticated = false;
InputStream in = null;
try
{
FileInputStream in = new FileInputStream(pfilename);
in = new BufferedInputStream(new FileInputStream(pfilename));
Properties props = new Properties();
props.load(in);
in.close();
// note we keep the message here and for the wrong password exactly the same to prevent attackers from guessing what users are valid
if (null == props.getProperty(username)) throw new AuthenticationException(authenticationErrorMessage(mode, username));
@ -125,6 +126,10 @@ public class SimpleAuthenticator implements IAuthenticator
{
throw new RuntimeException("Unexpected authentication problem", e);
}
finally
{
FileUtils.closeQuietly(in);
}
if (!authenticated) throw new AuthenticationException(authenticationErrorMessage(mode, username));

9
src/java/org/apache/cassandra/cli/CliMain.java
View File

@ -325,7 +325,14 @@ public class CliMain
{
prompt = (inCompoundStatement) ? "...\t" : getPrompt(cliClient);
line = reader.readLine(prompt);
try
{
line = reader.readLine(prompt);
}
catch (IOException e)
{
// retry on I/O Exception
}
if (line == null)
return;

8
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -267,12 +267,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Set<DecoratedKey> keys = new TreeSet<DecoratedKey>();
if (path.exists())
{
ObjectInputStream in = null;
try
{
long start = System.currentTimeMillis();
logger.info(String.format("reading saved cache %s", path));
ObjectInputStream in = new ObjectInputStream(new BufferedInputStream(new FileInputStream(path)));
in = new ObjectInputStream(new BufferedInputStream(new FileInputStream(path)));
while (in.available() > 0)
{
int size = in.readInt();
@ -280,7 +281,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
in.readFully(bytes);
keys.add(StorageService.getPartitioner().decorateKey(ByteBuffer.wrap(bytes)));
}
in.close();
if (logger.isDebugEnabled())
logger.debug(String.format("completed reading (%d ms; %d keys) saved cache %s",
System.currentTimeMillis() - start, keys.size(), path));
@ -289,6 +289,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
logger.warn(String.format("error reading saved cache %s", path.getAbsolutePath()), ioe);
}
finally
{
FileUtils.closeQuietly(in);
}
}
return keys;
}

44
src/java/org/apache/cassandra/db/ReadResponse.java
View File

@ -47,20 +47,21 @@ private static ICompactSerializer<ReadResponse> serializer_;
return serializer_;
}
private Row row_;
private ByteBuffer digest_ = FBUtilities.EMPTY_BYTE_BUFFER;
private boolean isDigestQuery_ = false;
private final Row row_;
private final ByteBuffer digest_;
public ReadResponse(ByteBuffer digest )
{
assert digest != null;
digest_= digest;
row_ = null;
}
public ReadResponse(Row row)
{
assert row != null;
row_ = row;
digest_ = null;
}
public Row row()
@ -75,12 +76,7 @@ private static ICompactSerializer<ReadResponse> serializer_;
public boolean isDigestQuery()
{
return isDigestQuery_;
}
public void setIsDigestQuery(boolean isDigestQuery)
{
isDigestQuery_ = isDigestQuery;
return digest_ != null;
}
}
@ -88,31 +84,35 @@ class ReadResponseSerializer implements ICompactSerializer<ReadResponse>
{
public void serialize(ReadResponse rm, DataOutputStream dos) throws IOException
{
dos.writeInt(rm.digest().remaining());
dos.write(rm.digest().array(), rm.digest().position() + rm.digest().arrayOffset(), rm.digest().remaining());
dos.writeInt(rm.isDigestQuery() ? rm.digest().remaining() : 0);
ByteBuffer buffer = rm.isDigestQuery() ? rm.digest() : FBUtilities.EMPTY_BYTE_BUFFER;
dos.write(buffer.array(), buffer.position() + buffer.arrayOffset(), buffer.remaining());
dos.writeBoolean(rm.isDigestQuery());
if( !rm.isDigestQuery() && rm.row() != null )
{
if (!rm.isDigestQuery())
{
Row.serializer().serialize(rm.row(), dos);
}
}
}
}
public ReadResponse deserialize(DataInputStream dis) throws IOException
{
byte[] digest = null;
int digestSize = dis.readInt();
byte[] digest = new byte[digestSize];
dis.read(digest, 0 , digestSize);
if (digestSize > 0)
{
digest = new byte[digestSize];
dis.readFully(digest, 0, digestSize);
}
boolean isDigest = dis.readBoolean();
assert isDigest == digestSize > 0;
Row row = null;
if (!isDigest)
{
row = Row.serializer().deserialize(dis);
}
ReadResponse rmsg = isDigest ? new ReadResponse(ByteBuffer.wrap(digest)) : new ReadResponse(row);
rmsg.setIsDigestQuery(isDigest);
return rmsg;
return isDigest ? new ReadResponse(ByteBuffer.wrap(digest)) : new ReadResponse(row);
}
}

1
src/java/org/apache/cassandra/db/ReadVerbHandler.java
View File

@ -82,7 +82,6 @@ public class ReadVerbHandler implements IVerbHandler
{
readResponse = new ReadResponse(row);
}
readResponse.setIsDigestQuery(command.isDigestQuery());
/* serialize the ReadResponseMessage. */
readCtx.bufOut_.reset();

10
src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
View File

@ -39,6 +39,7 @@ import org.apache.cassandra.io.sstable.IndexHelper;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.FileMark;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
@ -78,14 +79,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement
}
finally
{
try
{
file.close();
}
catch (IOException ioe)
{
logger.warn("error closing " + file.getPath());
}
FileUtils.closeQuietly(file);
}
}

2
src/java/org/apache/cassandra/db/commitlog/CommitLog.java
View File

@ -339,7 +339,7 @@ public class CommitLog
}
finally
{
reader.close();
FileUtils.closeQuietly(reader);
logger.info("Finished reading " + file);
}
}

7
src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java
View File

@ -25,8 +25,8 @@ 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.ICompactSerializer2;
import org.apache.cassandra.io.util.FileUtils;
public class CommitLogHeader
{
@ -132,13 +132,12 @@ public class CommitLogHeader
DataInputStream reader = null;
try
{
reader = new DataInputStream(new FileInputStream(headerFile));
reader = new DataInputStream(new BufferedInputStream(new FileInputStream(headerFile)));
return serializer.deserialize(reader);
}
finally
{
if (reader != null)
reader.close();
FileUtils.closeQuietly(reader);
}
}

13
src/java/org/apache/cassandra/gms/FailureDetector.java
View File

@ -18,10 +18,12 @@
package org.apache.cassandra.gms;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.IOError;
import java.io.OutputStream;
import java.lang.management.ManagementFactory;
import java.util.*;
import javax.management.MBeanServer;
@ -31,6 +33,7 @@ import org.apache.commons.lang.StringUtils;
import java.net.InetAddress;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.BoundedStatsDeque;
import org.slf4j.Logger;
@ -83,17 +86,21 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean
*/
public void dumpInterArrivalTimes()
{
OutputStream os = null;
try
{
File file = File.createTempFile("failuredetector-", ".dat");
FileOutputStream fos = new FileOutputStream(file, true);
fos.write(toString().getBytes());
fos.close();
os = new BufferedOutputStream(new FileOutputStream(file, true));
os.write(toString().getBytes());
}
catch (IOException e)
{
throw new IOError(e);
}
finally
{
FileUtils.closeQuietly(os);
}
}
public void setPhiConvictThreshold(int phi)

35
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -19,6 +19,7 @@
package org.apache.cassandra.io.sstable;
import java.io.BufferedInputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.File;
@ -53,6 +54,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.io.ICompactSerializer2;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.SegmentedFile;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.*;
@ -174,11 +176,18 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
File statsFile = new File(descriptor.filenameFor(SSTable.COMPONENT_STATS));
if (statsFile.exists())
{
logger.debug("Load statistics for {}", descriptor);
DataInputStream dis = new DataInputStream(new FileInputStream(statsFile));
rowSizes = EstimatedHistogram.serializer.deserialize(dis);
columnCounts = EstimatedHistogram.serializer.deserialize(dis);
dis.close();
DataInputStream dis = null;
try
{
logger.debug("Load statistics for {}", descriptor);
dis = new DataInputStream(new BufferedInputStream(new FileInputStream(statsFile)));
rowSizes = EstimatedHistogram.serializer.deserialize(dis);
columnCounts = EstimatedHistogram.serializer.deserialize(dis);
}
finally
{
FileUtils.closeQuietly(dis);
}
}
else
{
@ -253,14 +262,15 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
void loadBloomFilter() throws IOException
{
DataInputStream stream = new DataInputStream(new FileInputStream(descriptor.filenameFor(Component.FILTER)));
DataInputStream stream = null;
try
{
stream = new DataInputStream(new BufferedInputStream(new FileInputStream(descriptor.filenameFor(Component.FILTER))));
bf = BloomFilter.serializer().deserialize(stream);
}
finally
{
stream.close();
FileUtils.closeQuietly(stream);
}
}
@ -316,7 +326,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
}
finally
{
input.close();
FileUtils.closeQuietly(input);
}
// finalize the state of the reader
@ -497,14 +507,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
}
finally
{
try
{
input.close();
}
catch (IOException e)
{
logger.error("error closing file", e);
}
FileUtils.closeQuietly(input);
}
}

34
src/java/org/apache/cassandra/io/sstable/SSTableTracker.java
View File

@ -37,7 +37,9 @@ import org.apache.cassandra.cache.JMXInstrumentedCache;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -75,19 +77,29 @@ public class SSTableTracker implements Iterable<SSTableReader>
logger.info("saving " + msgSuffix);
int count = 0;
File tmpFile = File.createTempFile(savedCachePath.getName(), null, savedCachePath.getParentFile());
FileOutputStream fout = new FileOutputStream(tmpFile);
ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(fout));
FileDescriptor fd = fout.getFD();
for (K key : cache.getKeySet())
FileOutputStream fout = null;
ObjectOutputStream out = null;
try
{
ByteBuffer bytes = converter.apply(key);
out.writeInt(bytes.remaining());
out.write(bytes.array(),bytes.position()+bytes.arrayOffset(),bytes.remaining());
++count;
fout = new FileOutputStream(tmpFile);
out = new ObjectOutputStream(new BufferedOutputStream(fout));
FileDescriptor fd = fout.getFD();
for (K key : cache.getKeySet())
{
ByteBuffer bytes = converter.apply(key);
out.writeInt(bytes.remaining());
out.write(bytes.array(),bytes.position()+bytes.arrayOffset(),bytes.remaining());
++count;
}
out.flush();
fd.sync();
}
finally
{
FileUtils.closeQuietly(out);
FileUtils.closeQuietly(fout);
}
out.flush();
fd.sync();
out.close();
if (!tmpFile.renameTo(savedCachePath))
throw new IOException("Unable to rename cache to " + savedCachePath);
if (logger.isDebugEnabled())

19
src/java/org/apache/cassandra/io/util/FileUtils.java
View File

@ -18,14 +18,12 @@
package org.apache.cassandra.io.util;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.*;
import java.text.DecimalFormat;
import java.util.Comparator;
import java.util.List;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -85,6 +83,19 @@ public class FileUtils
}
}
public static void closeQuietly(Closeable c)
{
try
{
if (c != null)
c.close();
}
catch (Exception e)
{
logger_.warn("Failed closing stream", e);
}
}
public static class FileComparator implements Comparator<File>
{
public int compare(File f, File f2)

9
src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java
View File

@ -173,14 +173,7 @@ public class MmappedSegmentedFile extends SegmentedFile
}
finally
{
try
{
if (raf != null) raf.close();
}
catch (IOException e)
{
throw new IOError(e);
}
FileUtils.closeQuietly(raf);
}
return segments;
}

16
src/java/org/apache/cassandra/locator/PropertyFileSnitch.java
View File

@ -18,22 +18,24 @@
package org.apache.cassandra.locator;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.Reader;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.ResourceWatcher;
import org.apache.cassandra.utils.WrappedRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Used to determine if two IP's are in the same datacenter or on the same rack.
@ -112,14 +114,20 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
String rackPropertyFilename = FBUtilities.resourceToFile(RACK_PROPERTY_FILENAME);
Properties properties = new Properties();
Reader reader = null;
try
{
properties.load(new FileReader(rackPropertyFilename));
reader = new BufferedReader(new FileReader(rackPropertyFilename));
properties.load(reader);
}
catch (IOException e)
{
throw new ConfigurationException("Unable to read " + RACK_PROPERTY_FILENAME, e);
}
finally
{
FileUtils.closeQuietly(reader);
}
for (Map.Entry<Object, Object> entry : properties.entrySet())
{

13
src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java
View File

@ -42,6 +42,7 @@ import org.apache.cassandra.db.migration.Migration;
import org.apache.cassandra.utils.CLibrary;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Mx4jTool;
import org.apache.log4j.PropertyConfigurator;
import org.mortbay.thread.ThreadPool;
/**
@ -53,8 +54,16 @@ import org.mortbay.thread.ThreadPool;
*/
public abstract class AbstractCassandraDaemon implements CassandraDaemon
{
private static Logger logger = LoggerFactory
.getLogger(AbstractCassandraDaemon.class);
//Initialize logging in such a way that it checks for config changes every 10 seconds.
static
{
String config = System.getProperty("log4j.configuration", "log4j-server.properties");
PropertyConfigurator.configureAndWatch(ClassLoader.getSystemResource(config).getFile(), 10000);
org.apache.log4j.Logger.getLogger(AbstractCassandraDaemon.class).info("Logging initialized");
}
private static Logger logger = LoggerFactory.getLogger(AbstractCassandraDaemon.class);
protected InetAddress listenAddr;
protected int listenPort;

180
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -186,7 +186,6 @@ public class StorageProxy implements StorageProxyMBean
{
writeStats.addNano(System.nanoTime() - startTime);
}
}
/**
@ -265,19 +264,23 @@ public class StorageProxy implements StorageProxyMBean
if (StorageService.instance.isBootstrapMode())
throw new UnavailableException();
long startTime = System.nanoTime();
List<Row> rows;
if (consistency_level == ConsistencyLevel.ONE)
try
{
rows = weakRead(commands);
if (consistency_level == ConsistencyLevel.ONE)
{
rows = weakRead(commands);
}
else
{
assert consistency_level.getValue() >= ConsistencyLevel.QUORUM.getValue();
rows = strongRead(commands, consistency_level);
}
}
else
finally
{
assert consistency_level.getValue() >= ConsistencyLevel.QUORUM.getValue();
rows = strongRead(commands, consistency_level);
readStats.addNano(System.nanoTime() - startTime);
}
readStats.addNano(System.nanoTime() - startTime);
return rows;
}
@ -463,77 +466,82 @@ public class StorageProxy implements StorageProxyMBean
if (logger.isDebugEnabled())
logger.debug(command.toString());
long startTime = System.nanoTime();
List<AbstractBounds> ranges = getRestrictedRanges(command.range);
List<Row> rows;
// now scan until we have enough results
List<Row> rows = new ArrayList<Row>(command.max_keys);
for (AbstractBounds range : ranges)
try
{
List<InetAddress> liveEndpoints = StorageService.instance.getLiveNaturalEndpoints(command.keyspace, range.right);
if (consistency_level == ConsistencyLevel.ONE && liveEndpoints.contains(FBUtilities.getLocalAddress()))
rows = new ArrayList<Row>(command.max_keys);
List<AbstractBounds> ranges = getRestrictedRanges(command.range);
for (AbstractBounds range : ranges)
{
if (logger.isDebugEnabled())
logger.debug("local range slice");
ColumnFamilyStore cfs = Table.open(command.keyspace).getColumnFamilyStore(command.column_family);
try
{
rows.addAll(cfs.getRangeSlice(command.super_column,
range,
command.max_keys,
QueryFilter.getFilter(command.predicate, cfs.getComparator())));
}
catch (ExecutionException e)
{
throw new RuntimeException(e.getCause());
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
}
else
{
DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), liveEndpoints);
RangeSliceCommand c2 = new RangeSliceCommand(command.keyspace, command.column_family, command.super_column, command.predicate, range, command.max_keys);
Message message = c2.getMessage();
List<InetAddress> liveEndpoints = StorageService.instance.getLiveNaturalEndpoints(command.keyspace, range.right);
// collect replies and resolve according to consistency level
RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints);
AbstractReplicationStrategy rs = Table.open(command.keyspace).getReplicationStrategy();
QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver, consistency_level);
// TODO bail early if live endpoints can't satisfy requested consistency level
for (InetAddress endpoint : liveEndpoints)
if (consistency_level == ConsistencyLevel.ONE && liveEndpoints.contains(FBUtilities.getLocalAddress()))
{
MessagingService.instance.sendRR(message, endpoint, handler);
if (logger.isDebugEnabled())
logger.debug("reading " + c2 + " from " + message.getMessageId() + "@" + endpoint);
}
// TODO read repair on remaining replicas?
// if we're done, great, otherwise, move to the next range
try
{
if (logger.isDebugEnabled())
logger.debug("local range slice");
ColumnFamilyStore cfs = Table.open(command.keyspace).getColumnFamilyStore(command.column_family);
try
{
for (Row row : handler.get())
{
logger.debug("range slices read " + row.key);
}
}
rows.addAll(handler.get());
}
catch (DigestMismatchException e)
{
throw new AssertionError(e); // no digests in range slices yet
rows.addAll(cfs.getRangeSlice(command.super_column,
range,
command.max_keys,
QueryFilter.getFilter(command.predicate, cfs.getComparator())));
}
catch (ExecutionException e)
{
throw new RuntimeException(e.getCause());
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
}
}
if (rows.size() >= command.max_keys)
break;
}
else
{
DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), liveEndpoints);
RangeSliceCommand c2 = new RangeSliceCommand(command.keyspace, command.column_family, command.super_column, command.predicate, range, command.max_keys);
Message message = c2.getMessage();
rangeStats.addNano(System.nanoTime() - startTime);
// collect replies and resolve according to consistency level
RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints);
AbstractReplicationStrategy rs = Table.open(command.keyspace).getReplicationStrategy();
QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver, consistency_level);
// TODO bail early if live endpoints can't satisfy requested consistency level
for (InetAddress endpoint : liveEndpoints)
{
MessagingService.instance.sendRR(message, endpoint, handler);
if (logger.isDebugEnabled())
logger.debug("reading " + c2 + " from " + message.getMessageId() + "@" + endpoint);
}
// TODO read repair on remaining replicas?
// if we're done, great, otherwise, move to the next range
try
{
if (logger.isDebugEnabled())
{
for (Row row : handler.get())
{
logger.debug("range slices read " + row.key);
}
}
rows.addAll(handler.get());
}
catch (DigestMismatchException e)
{
throw new AssertionError(e); // no digests in range slices yet
}
}
if (rows.size() >= command.max_keys)
break;
}
}
finally
{
rangeStats.addNano(System.nanoTime() - startTime);
}
return rows.size() > command.max_keys ? rows.subList(0, command.max_keys) : rows;
}
@ -666,6 +674,16 @@ public class StorageProxy implements StorageProxyMBean
return readStats.getRecentLatencyMicros();
}
public long[] getTotalReadLatencyHistogramMicros()
{
return readStats.getTotalLatencyHistogramMicros();
}
public long[] getRecentReadLatencyHistogramMicros()
{
return readStats.getRecentLatencyHistogramMicros();
}
public long getRangeOperations()
{
return rangeStats.getOpCount();
@ -681,6 +699,16 @@ public class StorageProxy implements StorageProxyMBean
return rangeStats.getRecentLatencyMicros();
}
public long[] getTotalRangeLatencyHistogramMicros()
{
return rangeStats.getTotalLatencyHistogramMicros();
}
public long[] getRecentRangeLatencyHistogramMicros()
{
return rangeStats.getRecentLatencyHistogramMicros();
}
public long getWriteOperations()
{
return writeStats.getOpCount();
@ -696,6 +724,16 @@ public class StorageProxy implements StorageProxyMBean
return writeStats.getRecentLatencyMicros();
}
public long[] getTotalWriteLatencyHistogramMicros()
{
return writeStats.getTotalLatencyHistogramMicros();
}
public long[] getRecentWriteLatencyHistogramMicros()
{
return writeStats.getRecentLatencyHistogramMicros();
}
public static List<Row> scan(String keyspace, String column_family, IndexClause index_clause, SlicePredicate column_predicate, ConsistencyLevel consistency_level)
throws IOException, TimeoutException, UnavailableException
{

6
src/java/org/apache/cassandra/service/StorageProxyMBean.java
View File

@ -23,14 +23,20 @@ public interface StorageProxyMBean
public long getReadOperations();
public long getTotalReadLatencyMicros();
public double getRecentReadLatencyMicros();
public long[] getTotalReadLatencyHistogramMicros();
public long[] getRecentReadLatencyHistogramMicros();
public long getRangeOperations();
public long getTotalRangeLatencyMicros();
public double getRecentRangeLatencyMicros();
public long[] getTotalRangeLatencyHistogramMicros();
public long[] getRecentRangeLatencyHistogramMicros();
public long getWriteOperations();
public long getTotalWriteLatencyMicros();
public double getRecentWriteLatencyMicros();
public long[] getTotalWriteLatencyHistogramMicros();
public long[] getRecentWriteLatencyHistogramMicros();
public boolean getHintedHandoffEnabled();
public void setHintedHandoffEnabled(boolean b);

12
src/java/org/apache/cassandra/streaming/FileStreamTask.java
View File

@ -25,7 +25,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SocketChannel;
import org.apache.cassandra.streaming.StreamHeader;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -35,6 +35,7 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.WrappedRunnable;
public class FileStreamTask extends WrappedRunnable
{
private static Logger logger = LoggerFactory.getLogger( FileStreamTask.class );
@ -109,14 +110,7 @@ public class FileStreamTask extends WrappedRunnable
}
finally
{
try
{
raf.close();
}
catch (IOException e)
{
throw new AssertionError(e);
}
FileUtils.closeQuietly(raf);
}
}