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

merge from 1.0 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1220672 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Pavel Yaskevich 2011-12-19 09:33:17 +00:00
parent f9170645b1 c6e1e5029b
commit 9b7421ad05
8 changed files with 183 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
build.xml
View File

@ -376,7 +376,9 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<dependency groupId="com.thoughtworks.paranamer" artifactId="paranamer-ant" version="2.1"/>
<dependency groupId="junit" artifactId="junit" version="4.6" />
<dependency groupId="commons-logging" artifactId="commons-logging" version="1.1.1"/>
<dependency groupId="org.apache.rat" artifactId="apache-rat" version="0.6" />
<dependency groupId="org.apache.rat" artifactId="apache-rat" version="0.6">
<exclusion groupId="commons-lang" artifactId="commons-lang"/>
</dependency>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core" version="0.20.203.0"/>
<dependency groupId="net.sf.jopt-simple" artifactId="jopt-simple" version="3.2"/>
<dependency groupId="net.java.dev.jna" artifactId="jna" version="3.2.7"/>

13
src/java/org/apache/cassandra/db/context/CounterContext.java
View File

@ -19,15 +19,18 @@ package org.apache.cassandra.db.context;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.apache.log4j.Logger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.marshal.MarshalException;
import org.apache.cassandra.db.DBConstants;
import org.apache.cassandra.db.marshal.MarshalException;
import org.apache.cassandra.utils.Allocator;
import org.apache.cassandra.utils.HeapAllocator;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.HeapAllocator;
import org.apache.cassandra.utils.NodeId;
/**
@ -73,7 +76,7 @@ public class CounterContext implements IContext
private static final int COUNT_LENGTH = DBConstants.longSize;
private static final int STEP_LENGTH = NodeId.LENGTH + CLOCK_LENGTH + COUNT_LENGTH;
private static final Logger logger = Logger.getLogger(CounterContext.class);
private static final Logger logger = LoggerFactory.getLogger(CounterContext.class);
// lazy-load singleton
private static class LazyHolder

3
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -930,7 +930,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
}
else
{
// this is a new node
// this is a new node, report it to the FD in case it is the first time we are seeing it AND it's not alive
FailureDetector.instance.report(ep);
handleMajorStateChange(ep, remoteState);
}
}

11
src/java/org/apache/cassandra/io/util/SequentialWriter.java
View File

@ -39,6 +39,9 @@ public class SequentialWriter extends OutputStream
protected byte[] buffer;
private final boolean skipIOCache;
private final int fd;
private final int directoryFD;
// directory should be synced only after first file sync, in other words, only once per file
private boolean directorySynced = false;
protected long current = 0, bufferOffset;
protected int validBufferBytes;
@ -60,6 +63,7 @@ public class SequentialWriter extends OutputStream
buffer = new byte[bufferSize];
this.skipIOCache = skipIOCache;
fd = CLibrary.getfd(out.getFD());
directoryFD = CLibrary.tryOpenDirectory(file.getParent());
stream = new DataOutputStream(this);
}
@ -148,6 +152,12 @@ public class SequentialWriter extends OutputStream
flushInternal();
out.getFD().sync();
if (!directorySynced)
{
CLibrary.trySync(directoryFD);
directorySynced = true;
}
syncNeeded = false;
}
}
@ -288,6 +298,7 @@ public class SequentialWriter extends OutputStream
CLibrary.trySkipCache(fd, 0, 0);
out.close();
CLibrary.tryCloseFD(directoryFD);
}
/**

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

@ -23,6 +23,8 @@ import java.io.IOException;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
@ -46,6 +48,9 @@ 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.Mx4jTool;
import org.apache.commons.lang.ArrayUtils;
import com.google.common.collect.Iterables;
/**
* The <code>CassandraDaemon</code> is an abstraction for a Cassandra daemon
@ -117,10 +122,10 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
*/
protected void setup() throws IOException
{
logger.info("JVM vendor/version: {}/{}", System.getProperty("java.vm.name"), System.getProperty("java.version") );
logger.info("JVM vendor/version: {}/{}", System.getProperty("java.vm.name"), System.getProperty("java.version") );
logger.info("Heap size: {}/{}", Runtime.getRuntime().totalMemory(), Runtime.getRuntime().maxMemory());
logger.info("Classpath: {}", System.getProperty("java.class.path"));
CLibrary.tryMlockall();
logger.info("Classpath: {}", System.getProperty("java.class.path"));
CLibrary.tryMlockall();
listenPort = DatabaseDescriptor.getRpcPort();
listenAddr = DatabaseDescriptor.getRpcAddress();
@ -139,7 +144,20 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
}
}
});
// check all directories(data, commitlog, saved cache) for existence and permission
Iterable<String> dirs = Iterables.concat(Arrays.asList(DatabaseDescriptor.getAllDataFileLocations()),
Arrays.asList(new String[] {DatabaseDescriptor.getCommitLogLocation(),
DatabaseDescriptor.getSavedCachesLocation()}));
for (String dataDir : dirs)
{
logger.debug("Checking directory {}", dataDir);
File dir = new File(dataDir);
if (dir.exists())
assert dir.isDirectory() && dir.canRead() && dir.canWrite() && dir.canExecute()
: String.format("Directory %s is not accessible.", dataDir);
}
// check the system table to keep user from shooting self in foot by changing partitioner, cluster name, etc.
// we do a one-off scrub of the system table first; we can't load the list of the rest of the tables,
// until system table is opened.

14
src/java/org/apache/cassandra/service/MigrationManager.java
View File

@ -53,6 +53,8 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
// avoids re-pushing migrations that we're waiting on target to apply already
private static Map<InetAddress,UUID> lastPushed = new MapMaker().expiration(1, TimeUnit.MINUTES).makeMap();
private static UUID highestKnown;
public void onJoin(InetAddress endpoint, EndpointState epState) {
VersionedValue value = epState.getApplicationState(ApplicationState.SCHEMA);
@ -94,6 +96,7 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
*/
public static void rectify(UUID theirVersion, InetAddress endpoint)
{
updateHighestKnown(theirVersion);
UUID myVersion = Schema.instance.getVersion();
if (theirVersion.timestamp() < myVersion.timestamp()
&& !StorageService.instance.isClientMode())
@ -112,6 +115,17 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
}
}
}
private static void updateHighestKnown(UUID theirversion)
{
if (highestKnown == null || theirversion.timestamp() > highestKnown.timestamp())
highestKnown = theirversion;
}
public static boolean isReadyForBootstrap()
{
return highestKnown.compareTo(Schema.instance.getVersion()) == 0;
}
private static void pushMigrations(InetAddress endpoint, Collection<IColumn> migrations)
{

17
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -524,6 +524,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
|| !Schema.instance.getNonSystemTables().isEmpty()))
{
setMode(Mode.JOINING, "waiting for ring and schema information", true);
// first sleep the delay to make sure we see the schema
try
{
Thread.sleep(delay);
@ -532,6 +533,22 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
{
throw new AssertionError(e);
}
// now if our schema hasn't matched, keep sleeping until it does
while (!MigrationManager.isReadyForBootstrap())
{
setMode(Mode.JOINING, "waiting for schema information to complete", true);
try
{
Thread.sleep(delay);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
}
setMode(Mode.JOINING, "schema complete, ready to bootstrap", true);
if (logger_.isDebugEnabled())
logger_.debug("... got ring + schema info");

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

@ -18,14 +18,18 @@
*/
package org.apache.cassandra.utils;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileDescriptor;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.reflect.Field;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.commons.lang.StringUtils;
import com.sun.jna.LastErrorException;
import com.sun.jna.Native;
@ -42,6 +46,7 @@ public final class CLibrary
private static final int F_SETFL = 4; /* set file status flags */
private static final int F_NOCACHE = 48; /* Mac OS X specific flag, turns cache on/off */
private static final int O_DIRECT = 040000; /* fcntl.h */
private static final int O_RDONLY = 00000000; /* fcntl.h */
private static final int POSIX_FADV_NORMAL = 0; /* fadvise.h */
private static final int POSIX_FADV_RANDOM = 1; /* fadvise.h */
@ -80,7 +85,11 @@ public final class CLibrary
// fadvice
public static native int posix_fadvise(int fd, long offset, int len, int flag) throws LastErrorException;
public static native int open(String path, int flags) throws LastErrorException;
public static native int fsync(int fd) throws LastErrorException;
public static native int close(int fd) throws LastErrorException;
private static int errno(RuntimeException e)
{
assert e instanceof LastErrorException;
@ -148,6 +157,7 @@ public final class CLibrary
}
catch (RuntimeException e)
{
logger.error("Unable to create hard link", e);
if (!(e instanceof LastErrorException))
throw e;
// there are 17 different error codes listed on the man page. punt until/unless we find which
@ -178,10 +188,37 @@ public final class CLibrary
pb = new ProcessBuilder("ln", sourceFile.getAbsolutePath(), destinationFile.getAbsolutePath());
pb.redirectErrorStream(true);
}
try
{
exec(pb);
}
catch (IOException ex)
{
logger.error("Unable to create hard link", ex);
throw ex;
}
}
private static void exec(ProcessBuilder pb) throws IOException
{
Process p = pb.start();
try
{
p.waitFor();
int errCode = p.waitFor();
if (errCode != 0)
{
BufferedReader in = new BufferedReader(new InputStreamReader(p.getInputStream()));
BufferedReader err = new BufferedReader(new InputStreamReader(p.getErrorStream()));
StringBuffer buff = new StringBuffer();
String str;
while ((str = in.readLine()) != null)
buff.append(str).append(System.getProperty("line.separator"));
while ((str = err.readLine()) != null)
buff.append(str).append(System.getProperty("line.separator"));
throw new IOException("Exception while executing the command: "+ StringUtils.join(pb.command(), " ") +
", command error Code: " + errCode +
", command output: "+ buff.toString());
}
}
catch (InterruptedException e)
{
@ -229,6 +266,73 @@ public final class CLibrary
return result;
}
public static int tryOpenDirectory(String path)
{
int fd = -1;
try
{
return open(path, O_RDONLY);
}
catch (UnsatisfiedLinkError e)
{
// JNA is unavailable just skipping Direct I/O
}
catch (RuntimeException e)
{
if (!(e instanceof LastErrorException))
throw e;
logger.warn(String.format("open(%s, O_RDONLY) failed, errno (%d).", path, CLibrary.errno(e)));
}
return fd;
}
public static void trySync(int fd)
{
if (fd == -1)
return;
try
{
fsync(fd);
}
catch (UnsatisfiedLinkError e)
{
// JNA is unavailable just skipping Direct I/O
}
catch (RuntimeException e)
{
if (!(e instanceof LastErrorException))
throw e;
logger.warn(String.format("fsync(%d) failed, errno (%d).", fd, CLibrary.errno(e)));
}
}
public static void tryCloseFD(int fd)
{
if (fd == -1)
return;
try
{
close(fd);
}
catch (UnsatisfiedLinkError e)
{
// JNA is unavailable just skipping Direct I/O
}
catch (RuntimeException e)
{
if (!(e instanceof LastErrorException))
throw e;
logger.warn(String.format("close(%d) failed, errno (%d).", fd, CLibrary.errno(e)));
}
}
/**
* Get system file descriptor from FileDescriptor object.
* @param descriptor - FileDescriptor objec to get fd from