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@1030290 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-11-02 23:40:55 +00:00
parent 2e27927111 85d6ce67f3
commit 00a0df163f
12 changed files with 161 additions and 221 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGES.txt
View File

@ -10,6 +10,11 @@ dev
to client requests (CASSANDRA-1685)
* log tpstats when dropping messages (CASSANDRA-1660)
* include unreachable nodes in describeSchemaVersions (CASSANDRA-1678)
* Avoid dropping messages off the client request path (CASSANDRA-1676)
* fix jna errno reporting (CASSANDRA-1694)
* add friendlier error for UnknownHostException on startup (CASSANDRA-1697)
* include jna dependency in RPM package (CASSANDRA-1690)
* add --skip-keys option to stress.py (CASSANDRA-1696)
0.7.0-beta3

4
build.xml
View File

@ -251,7 +251,7 @@
-->
<target name="gen-thrift-java" depends="rat-init" description="Generate Thrift Java artifacts">
<echo>Generating Thrift Java code from ${basedir}/interface/cassandra.thrift ....</echo>
<exec executable="thrift" dir="${basedir}/interface">
<exec executable="thrift" dir="${basedir}/interface" failonerror="true">
<arg line="--gen java:hashcode" />
<arg line="-o ${interface.thrift.dir}" />
<arg line="cassandra.thrift" />
@ -266,7 +266,7 @@
</target>
<target name="gen-thrift-py" description="Generate Thrift Python artifacts">
<echo>Generating Thrift Python code from ${basedir}/interface/cassandra.thrift ....</echo>
<exec executable="thrift" dir="${basedir}/interface">
<exec executable="thrift" dir="${basedir}/interface" failonerror="true">
<arg line="--gen py" />
<arg line="-o ${interface.thrift.dir}" />
<arg line="cassandra.thrift" />

5
contrib/py_stress/stress.py
View File

@ -61,6 +61,8 @@ except ImportError:
parser = OptionParser()
parser.add_option('-n', '--num-keys', type="int", dest="numkeys",
help="Number of keys", default=1000**2)
parser.add_option('-N', '--skip-keys', type="float", dest="skipkeys",
help="Fraction of keys to skip initially", default=0)
parser.add_option('-t', '--threads', type="int", dest="threads",
help="Number of threads/procs to use", default=50)
parser.add_option('-c', '--columns', type="int", dest="columns",
@ -189,7 +191,8 @@ class Operation(Thread):
def __init__(self, i, opcounts, keycounts, latencies):
Thread.__init__(self)
# generator of the keys to be used
self.range = xrange(keys_per_thread * i, keys_per_thread * (i + 1))
self.range = xrange(int(keys_per_thread * (i + options.skipkeys)),
keys_per_thread * (i + 1))
# we can't use a local counter, since that won't be visible to the parent
# under multiprocessing. instead, the parent passes a "opcounts" array
# and an index that is our assigned counter.

1
redhat/apache-cassandra.spec
View File

@ -19,6 +19,7 @@ BuildRequires: ant
BuildRequires: ant-nodeps
Requires: java >= 1.6.0
Requires: jna >= 3.2.7
Requires: jpackage-utils
Requires(pre): user(cassandra)
Requires(pre): group(cassandra)

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

@ -367,11 +367,17 @@ public class DatabaseDescriptor
{
throw new ConfigurationException("seeds missing; a minimum of one seed is required.");
}
for( int i = 0; i < conf.seeds.length; ++i )
for (String seedString : conf.seeds)
{
seeds.add(InetAddress.getByName(conf.seeds[i]));
seeds.add(InetAddress.getByName(seedString));
}
}
catch (UnknownHostException e)
{
logger.error("Fatal error: " + e.getMessage());
System.err.println("Unable to start with unknown hosts configured. Use IP addresses instead of hostnames.");
System.exit(2);
}
catch (ConfigurationException e)
{
logger.error("Fatal error: " + e.getMessage());

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

@ -18,46 +18,22 @@
package org.apache.cassandra.db;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FilenameFilter;
import java.io.IOError;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.*;
import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.collect.Iterables;
import org.apache.commons.collections.IteratorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.concurrent.RetryingScheduledThreadPoolExecutor;
@ -66,47 +42,20 @@ import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.columniterator.IColumnIterator;
import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogSegment;
import org.apache.cassandra.db.filter.IFilter;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.filter.SliceQueryFilter;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.LocalByPartionerType;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.LocalToken;
import org.apache.cassandra.dht.OrderPreservingPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.ReducingKeyIterator;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableTracker;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.IndexClause;
import org.apache.cassandra.thrift.IndexExpression;
import org.apache.cassandra.thrift.IndexOperator;
import org.apache.cassandra.utils.EstimatedHistogram;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.LatencyTracker;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.commons.collections.IteratorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Iterables;
import org.apache.cassandra.utils.*;
public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
@ -682,11 +631,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public void forceFlushIfExpired()
{
if (memtable.isExpired())
{
logger.info("Memtable for {} has reached memtable_flush_after_mins {}, enqueueing flush",
memtable.cfs.getColumnFamilyName(), memtable.cfs.getMemtableFlushAfterMins());
forceFlush();
}
}
public Future<?> forceFlush()
@ -1540,7 +1485,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
File sourceFile = new File(ssTable.descriptor.filenameFor(component));
File targetLink = new File(snapshotDirectoryPath, sourceFile.getName());
FileUtils.createHardLink(sourceFile, targetLink);
CLibrary.createHardLink(sourceFile, targetLink);
}
if (logger.isDebugEnabled())
logger.debug("Snapshot for " + table + " keyspace data file " + ssTable.getFilename() +
@ -1550,7 +1495,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
throw new IOError(e);
}
}
}

19
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.db;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
@ -99,20 +100,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable
boolean isThresholdViolated()
{
if (currentThroughput.get() >= THRESHOLD)
{
logger.info("Memtable for CF {} has reached memtable_throughput_in_mb {}, enqueueing flush",
cfs.getColumnFamilyName(), THRESHOLD);
return true;
}
if (currentOperations.get() >= THRESHOLD_COUNT)
{
logger.info("Memtable for CF {} has reached memtable_operations_in_millions {}, enqueueing flush",
cfs.getColumnFamilyName(), THRESHOLD_COUNT);
return true;
}
// default case, threshold is not violated.
return false;
return currentThroughput.get() >= this.THRESHOLD || currentOperations.get() >= this.THRESHOLD_COUNT;
}
boolean isFrozen()
@ -171,7 +159,8 @@ public class Memtable implements Comparable<Memtable>, IFlushable
writer.append(entry.getKey(), entry.getValue());
SSTableReader ssTable = writer.closeAndOpenReader();
logger.info("Completed flushing " + ssTable.getFilename());
logger.info(String.format("Completed flushing %s (%d bytes)",
ssTable.getFilename(), new File(ssTable.getFilename()).length()));
return ssTable;
}

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

@ -18,17 +18,16 @@
package org.apache.cassandra.io.util;
import java.io.*;
import java.io.File;
import java.io.IOException;
import java.text.DecimalFormat;
import java.util.*;
import java.util.Comparator;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.sun.jna.Native;
import org.apache.cassandra.utils.CLibrary;
import com.sun.jna.Native;
import com.sun.jna.LastErrorException;
import org.apache.cassandra.utils.CLibrary;
@ -164,29 +163,6 @@ public class FileUtils
}
}
/**
* calculate the total space used by a file or directory
*
* @param path the path
* @return total space used.
*/
public static long getUsedDiskSpaceForPath(String path)
{
File file = new File(path);
if (file.isFile())
{
return file.length();
}
long diskSpace = 0;
for (File childFile: file.listFiles())
{
diskSpace += getUsedDiskSpaceForPath(childFile.getPath());
}
return diskSpace;
}
/**
* Deletes all files and subdirectories under "dir".
* @param dir Directory to be deleted
@ -206,68 +182,4 @@ public class FileUtils
// The directory is now empty so now it can be smoked
deleteWithConfirm(dir);
}
/**
* Create a hard link for a given file.
*
* @param sourceFile The name of the source file.
* @param destinationFile The name of the destination file.
*
* @throws IOException if an error has occurred while creating the link.
*/
public static void createHardLink(File sourceFile, File destinationFile) throws IOException
{
int errno = Integer.MIN_VALUE;
try
{
int result = CLibrary.link(sourceFile.getAbsolutePath(), destinationFile.getAbsolutePath());
if (result != 0)
errno = Native.getLastError();
}
catch (UnsatisfiedLinkError e)
{
createHardLinkWithExec(sourceFile, destinationFile);
return;
}
if (errno != Integer.MIN_VALUE)
{
// there are 17 different error codes listed on the man page. punt until/unless we find which
// ones actually turn up in practice.
throw new IOException(String.format("Unable to create hard link from %s to %s (errno %d)",
sourceFile, destinationFile, errno));
}
}
private static void createHardLinkWithExec(File sourceFile, File destinationFile) throws IOException
{
String osname = System.getProperty("os.name");
ProcessBuilder pb;
if (osname.startsWith("Windows"))
{
float osversion = Float.parseFloat(System.getProperty("os.version"));
if (osversion >= 6.0f)
{
pb = new ProcessBuilder("cmd", "/c", "mklink", "/H", destinationFile.getAbsolutePath(), sourceFile.getAbsolutePath());
}
else
{
pb = new ProcessBuilder("fsutil", "hardlink", "create", destinationFile.getAbsolutePath(), sourceFile.getAbsolutePath());
}
}
else
{
pb = new ProcessBuilder("ln", sourceFile.getAbsolutePath(), destinationFile.getAbsolutePath());
pb.redirectErrorStream(true);
}
Process p = pb.start();
try
{
p.waitFor();
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
}
}

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

@ -39,6 +39,7 @@ import org.apache.cassandra.db.SystemTable;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.commitlog.CommitLog;
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.mortbay.thread.ThreadPool;
@ -70,7 +71,7 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
protected void setup() throws IOException
{
logger.info("Heap size: {}/{}", Runtime.getRuntime().totalMemory(), Runtime.getRuntime().maxMemory());
FBUtilities.tryMlockall();
CLibrary.tryMlockall();
listenPort = DatabaseDescriptor.getRpcPort();
listenAddr = DatabaseDescriptor.getRpcAddress();

2
src/java/org/apache/cassandra/thrift/CassandraDaemon.java
View File

@ -52,7 +52,7 @@ public class CassandraDaemon extends org.apache.cassandra.service.AbstractCassan
protected void setup() throws IOException
{
super.setup();
super.setup();
// now we start listening for clients
final CassandraServer cassandraServer = new CassandraServer();

121
src/java/org/apache/cassandra/utils/CLibrary.java
View File

@ -18,19 +18,23 @@
*/
package org.apache.cassandra.utils;
import java.io.File;
import java.io.IOException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.sun.jna.LastErrorException;
import com.sun.jna.Native;
public final class CLibrary
{
private static Logger logger = LoggerFactory.getLogger(CLibrary.class);
public static final int MCL_CURRENT = 1;
public static final int MCL_FUTURE = 2;
private static final int MCL_CURRENT = 1;
private static final int MCL_FUTURE = 2;
public static final int ENOMEM = 12;
private static final int ENOMEM = 12;
static
{
@ -48,10 +52,115 @@ public final class CLibrary
}
}
public static native int mlockall(int flags);
public static native int munlockall();
private static native int mlockall(int flags) throws LastErrorException;
private static native int munlockall() throws LastErrorException;
public static native int link(String from, String to);
private static native int link(String from, String to) throws LastErrorException;
private static int errno(RuntimeException e)
{
assert e instanceof LastErrorException;
try
{
return ((LastErrorException) e).getErrorCode();
}
catch (NoSuchMethodError x)
{
logger.warn("Obsolete version of JNA present; unable to read errno. Upgrade to JNA 3.2.7 or later");
return 0;
}
}
private CLibrary() {}
public static void tryMlockall()
{
try
{
int result = mlockall(MCL_CURRENT);
assert result == 0; // mlockall should always be zero on success
}
catch (UnsatisfiedLinkError e)
{
// this will have already been logged by CLibrary, no need to repeat it
}
catch (RuntimeException e)
{
if (!(e instanceof LastErrorException))
throw e;
if (errno(e) == ENOMEM && System.getProperty("os.name").toLowerCase().contains("linux"))
{
logger.warn("Unable to lock JVM memory (ENOMEM)."
+ " This can result in part of the JVM being swapped out, especially with mmapped I/O enabled."
+ " Increase RLIMIT_MEMLOCK or run Cassandra as root.");
}
else if (!System.getProperty("os.name").toLowerCase().contains("mac"))
{
// OS X allows mlockall to be called, but always returns an error
logger.warn("Unknown mlockall error " + errno(e));
}
}
}
/**
* Create a hard link for a given file.
*
* @param sourceFile The name of the source file.
* @param destinationFile The name of the destination file.
*
* @throws java.io.IOException if an error has occurred while creating the link.
*/
public static void createHardLink(File sourceFile, File destinationFile) throws IOException
{
try
{
int result = link(sourceFile.getAbsolutePath(), destinationFile.getAbsolutePath());
assert result == 0; // success is always zero
}
catch (UnsatisfiedLinkError e)
{
createHardLinkWithExec(sourceFile, destinationFile);
}
catch (RuntimeException e)
{
if (!(e instanceof LastErrorException))
throw e;
// there are 17 different error codes listed on the man page. punt until/unless we find which
// ones actually turn up in practice.
throw new IOException(String.format("Unable to create hard link from %s to %s (errno %d)",
sourceFile, destinationFile, errno(e)));
}
}
private static void createHardLinkWithExec(File sourceFile, File destinationFile) throws IOException
{
String osname = System.getProperty("os.name");
ProcessBuilder pb;
if (osname.startsWith("Windows"))
{
float osversion = Float.parseFloat(System.getProperty("os.version"));
if (osversion >= 6.0f)
{
pb = new ProcessBuilder("cmd", "/c", "mklink", "/H", destinationFile.getAbsolutePath(), sourceFile.getAbsolutePath());
}
else
{
pb = new ProcessBuilder("fsutil", "hardlink", "create", destinationFile.getAbsolutePath(), sourceFile.getAbsolutePath());
}
}
else
{
pb = new ProcessBuilder("ln", sourceFile.getAbsolutePath(), destinationFile.getAbsolutePath());
pb.redirectErrorStream(true);
}
Process p = pb.start();
try
{
p.waitFor();
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
}
}

32
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -41,6 +41,7 @@ import org.apache.commons.lang.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.sun.jna.LastErrorException;
import com.sun.jna.Native;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -637,37 +638,6 @@ public class FBUtilities
}
}
public static void tryMlockall()
{
int errno = Integer.MIN_VALUE;
try
{
int result = CLibrary.mlockall(CLibrary.MCL_CURRENT);
if (result != 0)
errno = Native.getLastError();
}
catch (UnsatisfiedLinkError e)
{
// this will have already been logged by CLibrary, no need to repeat it
return;
}
if (errno != Integer.MIN_VALUE)
{
if (errno == CLibrary.ENOMEM && System.getProperty("os.name").toLowerCase().contains("linux"))
{
logger_.warn("Unable to lock JVM memory (ENOMEM)."
+ " This can result in part of the JVM being swapped out, especially with mmapped I/O enabled."
+ " Increase RLIMIT_MEMLOCK or run Cassandra as root.");
}
else if (!System.getProperty("os.name").toLowerCase().contains("mac"))
{
// OS X allows mlockall to be called, but always returns an error
logger_.warn("Unknown mlockall error " + errno);
}
}
}
public static <T extends Comparable> SortedSet<T> singleton(T column)
{
return new TreeSet<T>(Arrays.asList(column));