You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2010/01/15 20:55:00 UTC
svn commit: r899791 - in /incubator/cassandra/trunk:
src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/dht/
src/java/org/apache/cassandra/gms/ src/java/org/apache/cassandra/io/
src/java/org/apache/cassandra/locator/ src/java/org/apache/ca...
Author: jbellis
Date: Fri Jan 15 19:54:58 2010
New Revision: 899791
URL: http://svn.apache.org/viewvc?rev=899791&view=rev
Log:
replace instance() accessors with public final fields. patch by Jeff Hodges; reviewed by Gary Dusbabek and jbellis for CASSANDRA-700
Modified:
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/HintedHandOffManager.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ReadVerbHandler.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BootStrapper.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/ApplicationState.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/EndPointState.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/FailureDetector.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraDaemon.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraServer.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/LoadDisseminator.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadRepairManager.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageLoadBalancer.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/StreamingTest.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/MoveTest.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/StorageServiceTest.java
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java Fri Jan 15 19:54:58 2010
@@ -319,7 +319,7 @@
writer = new SSTableWriter(newFilename, expectedBloomFilterSize, StorageService.getPartitioner());
// validate the CF as we iterate over it
- AntiEntropyService.IValidator validator = AntiEntropyService.instance().getValidator(table.name, cfs.getColumnFamilyName(), null, major);
+ AntiEntropyService.IValidator validator = AntiEntropyService.instance.getValidator(table.name, cfs.getColumnFamilyName(), null, major);
validator.prepare();
while (nni.hasNext())
{
@@ -433,7 +433,7 @@
private void doCleanupCompaction(ColumnFamilyStore cfs) throws IOException
{
Collection<SSTableReader> originalSSTables = cfs.getSSTables();
- List<SSTableReader> sstables = doAntiCompaction(cfs, originalSSTables, StorageService.instance().getLocalRanges(), null);
+ List<SSTableReader> sstables = doAntiCompaction(cfs, originalSSTables, StorageService.instance.getLocalRanges(), null);
if (!sstables.isEmpty())
{
cfs.replaceCompactedSSTables(originalSSTables, sstables);
@@ -455,7 +455,7 @@
Iterator<CompactionIterator.CompactedRow> nni = new FilterIterator(ci, PredicateUtils.notNullPredicate());
// validate the CF as we iterate over it
- AntiEntropyService.IValidator validator = AntiEntropyService.instance().getValidator(cfs.getTable().name, cfs.getColumnFamilyName(), initiator, true);
+ AntiEntropyService.IValidator validator = AntiEntropyService.instance.getValidator(cfs.getTable().name, cfs.getColumnFamilyName(), initiator, true);
validator.prepare();
while (nni.hasNext())
{
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/HintedHandOffManager.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/HintedHandOffManager.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/HintedHandOffManager.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/HintedHandOffManager.java Fri Jan 15 19:54:58 2010
@@ -82,8 +82,8 @@
public class HintedHandOffManager
{
- private static volatile HintedHandOffManager instance_;
- private static final Lock lock_ = new ReentrantLock();
+ public static final HintedHandOffManager instance = new HintedHandOffManager();
+
private static final Logger logger_ = Logger.getLogger(HintedHandOffManager.class);
final static long INTERVAL_IN_MS = 3600 * 1000; // check for ability to deliver hints this often
public static final String HINTS_CF = "HintsColumnFamily";
@@ -91,26 +91,7 @@
private final ExecutorService executor_ = new JMXEnabledThreadPoolExecutor("HINTED-HANDOFF-POOL");
-
- public static HintedHandOffManager instance()
- {
- if (instance_ == null)
- {
- lock_.lock();
- try
- {
- if (instance_ == null)
- instance_ = new HintedHandOffManager();
- }
- finally
- {
- lock_.unlock();
- }
- }
- return instance_;
- }
-
- public HintedHandOffManager()
+ protected HintedHandOffManager()
{
new Thread(new WrappedRunnable()
{
@@ -127,12 +108,12 @@
private static boolean sendMessage(InetAddress endPoint, String tableName, String key) throws IOException
{
- if (!Gossiper.instance().isKnownEndpoint(endPoint))
+ if (!Gossiper.instance.isKnownEndpoint(endPoint))
{
logger_.warn("Hints found for endpoint " + endPoint + " which is not part of the gossip network. discarding.");
return true;
}
- if (!FailureDetector.instance().isAlive(endPoint))
+ if (!FailureDetector.instance.isAlive(endPoint))
{
return false;
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ReadVerbHandler.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ReadVerbHandler.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ReadVerbHandler.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ReadVerbHandler.java Fri Jan 15 19:54:58 2010
@@ -70,7 +70,7 @@
try
{
- if (StorageService.instance().isBootstrapMode())
+ if (StorageService.instance.isBootstrapMode())
{
/* Don't service reads! */
throw new RuntimeException("Cannot service reads while bootstrapping!");
@@ -106,11 +106,11 @@
/* Do read repair if header of the message says so */
if (message.getHeader(ReadCommand.DO_REPAIR) != null)
{
- List<InetAddress> endpoints = StorageService.instance().getLiveNaturalEndpoints(command.key);
+ List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(command.key);
/* Remove the local storage endpoint from the list. */
endpoints.remove(FBUtilities.getLocalAddress());
if (endpoints.size() > 0 && DatabaseDescriptor.getConsistencyCheck())
- StorageService.instance().doConsistencyCheck(row, endpoints, command);
+ StorageService.instance.doConsistencyCheck(row, endpoints, command);
}
}
catch (IOException ex)
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BootStrapper.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BootStrapper.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BootStrapper.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BootStrapper.java Fri Jan 15 19:54:58 2010
@@ -87,7 +87,7 @@
for (Map.Entry<InetAddress, Collection<Range>> entry : getWorkMap(rangesWithSourceTarget).asMap().entrySet())
{
InetAddress source = entry.getKey();
- StorageService.instance().addBootstrapSource(source);
+ StorageService.instance.addBootstrapSource(source);
if (logger.isDebugEnabled())
logger.debug("Requesting from " + source + " ranges " + StringUtils.join(entry.getValue(), ", "));
Streaming.requestRanges(source, entry.getValue());
@@ -187,7 +187,7 @@
static Multimap<InetAddress, Range> getWorkMap(Multimap<Range, InetAddress> rangesWithSourceTarget)
{
- return getWorkMap(rangesWithSourceTarget, FailureDetector.instance());
+ return getWorkMap(rangesWithSourceTarget, FailureDetector.instance);
}
static Multimap<InetAddress, Range> getWorkMap(Multimap<Range, InetAddress> rangesWithSourceTarget, IFailureDetector failureDetector)
@@ -217,7 +217,7 @@
{
public void doVerb(Message message)
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
List<String> tokens = ss.getSplits(2);
assert tokens.size() == 3 : tokens.size();
Message response;
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/ApplicationState.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/ApplicationState.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/ApplicationState.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/ApplicationState.java Fri Jan 15 19:54:58 2010
@@ -34,7 +34,7 @@
* e.g. if we want to disseminate load information for node A do the following:
*
* ApplicationState loadState = new ApplicationState(<string representation of load>);
- * Gossiper.instance().addApplicationState("LOAD STATE", loadState);
+ * Gossiper.instance.addApplicationState("LOAD STATE", loadState);
*/
public class ApplicationState implements Comparable<ApplicationState>
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/EndPointState.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/EndPointState.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/EndPointState.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/EndPointState.java Fri Jan 15 19:54:58 2010
@@ -84,7 +84,7 @@
void addApplicationState(String key, ApplicationState appState)
{
- assert !StorageService.instance().isClientMode();
+ assert !StorageService.instance.isClientMode();
applicationState_.put(key, appState);
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/FailureDetector.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/FailureDetector.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/FailureDetector.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/FailureDetector.java Fri Jan 15 19:54:58 2010
@@ -42,37 +42,15 @@
*/
public class FailureDetector implements IFailureDetector, FailureDetectorMBean
{
+ public static final IFailureDetector instance = new FailureDetector();
private static Logger logger_ = Logger.getLogger(FailureDetector.class);
private static final int sampleSize_ = 1000;
private static final int phiConvictThreshold_ = 8;
/* The Failure Detector has to have been up for at least 1 min. */
private static final long uptimeThreshold_ = 60000;
- private static volatile IFailureDetector failureDetector_;
- /* Used to lock the factory for creation of FailureDetector instance */
- private static Lock createLock_ = new ReentrantLock();
/* The time when the module was instantiated. */
private static long creationTime_;
- public static IFailureDetector instance()
- {
- if ( failureDetector_ == null )
- {
- FailureDetector.createLock_.lock();
- try
- {
- if ( failureDetector_ == null )
- {
- failureDetector_ = new FailureDetector();
- }
- }
- finally
- {
- createLock_.unlock();
- }
- }
- return failureDetector_;
- }
-
private Map<InetAddress, ArrivalWindow> arrivalSamples_ = new Hashtable<InetAddress, ArrivalWindow>();
private List<IFailureDetectionEventListener> fdEvntListeners_ = new ArrayList<IFailureDetectionEventListener>();
@@ -141,7 +119,7 @@
return true;
/* Incoming port is assumed to be the Storage port. We need to change it to the control port */
- EndPointState epState = Gossiper.instance().getEndPointStateForEndPoint(ep);
+ EndPointState epState = Gossiper.instance.getEndPointStateForEndPoint(ep);
return epState.isAlive();
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/gms/Gossiper.java Fri Jan 15 19:54:58 2010
@@ -49,12 +49,12 @@
{
try
{
- synchronized( Gossiper.instance() )
+ synchronized( Gossiper.instance )
{
/* Update the local heartbeat counter. */
endPointStateMap_.get(localEndPoint_).getHeartBeatState().updateHeartBeat();
List<GossipDigest> gDigests = new ArrayList<GossipDigest>();
- Gossiper.instance().makeRandomGossipDigest(gDigests);
+ Gossiper.instance.makeRandomGossipDigest(gDigests);
if ( gDigests.size() > 0 )
{
@@ -108,16 +108,7 @@
final static String GOSSIP_DIGEST_ACK2_VERB = "GA2V";
public final static int intervalInMillis_ = 1000;
private static Logger logger_ = Logger.getLogger(Gossiper.class);
- private static volatile Gossiper gossiper_;
-
- public synchronized static Gossiper instance()
- {
- if ( gossiper_ == null )
- {
- gossiper_ = new Gossiper();
- }
- return gossiper_;
- }
+ public static final Gossiper instance = new Gossiper();
private Timer gossipTimer_;
private InetAddress localEndPoint_;
@@ -144,7 +135,7 @@
gossipTimer_ = new Timer(false);
aVeryLongTime_ = 259200 * 1000;
/* register with the Failure Detector for receiving Failure detector events */
- FailureDetector.instance().registerFailureDetectionEventListener(this);
+ FailureDetector.instance.registerFailureDetectionEventListener(this);
/* register the verbs */
MessagingService.instance().registerVerbHandlers(JOIN_VERB_HANDLER, new JoinVerbHandler());
MessagingService.instance().registerVerbHandlers(GOSSIP_DIGEST_SYN_VERB, new GossipDigestSynVerbHandler());
@@ -382,7 +373,7 @@
if ( endpoint.equals(localEndPoint_) )
continue;
- FailureDetector.instance().interpret(endpoint);
+ FailureDetector.instance.interpret(endpoint);
EndPointState epState = endPointStateMap_.get(endpoint);
if ( epState != null )
{
@@ -460,7 +451,7 @@
void notifyFailureDetector(List<GossipDigest> gDigests)
{
- IFailureDetector fd = FailureDetector.instance();
+ IFailureDetector fd = FailureDetector.instance;
for ( GossipDigest gDigest : gDigests )
{
EndPointState localEndPointState = endPointStateMap_.get(gDigest.endPoint_);
@@ -494,7 +485,7 @@
void notifyFailureDetector(Map<InetAddress, EndPointState> remoteEpStateMap)
{
- IFailureDetector fd = FailureDetector.instance();
+ IFailureDetector fd = FailureDetector.instance;
Set<InetAddress> endpoints = remoteEpStateMap.keySet();
for ( InetAddress endpoint : endpoints )
{
@@ -857,7 +848,7 @@
}
if ( joinMessage.clusterId_.equals( DatabaseDescriptor.getClusterName() ) )
{
- Gossiper.instance().join(from);
+ Gossiper.instance.join(from);
}
}
}
@@ -884,16 +875,16 @@
List<GossipDigest> gDigestList = gDigestMessage.getGossipDigests();
/* Notify the Failure Detector */
- Gossiper.instance().notifyFailureDetector(gDigestList);
+ Gossiper.instance.notifyFailureDetector(gDigestList);
doSort(gDigestList);
List<GossipDigest> deltaGossipDigestList = new ArrayList<GossipDigest>();
Map<InetAddress, EndPointState> deltaEpStateMap = new HashMap<InetAddress, EndPointState>();
- Gossiper.instance().examineGossiper(gDigestList, deltaGossipDigestList, deltaEpStateMap);
+ Gossiper.instance.examineGossiper(gDigestList, deltaGossipDigestList, deltaEpStateMap);
GossipDigestAckMessage gDigestAck = new GossipDigestAckMessage(deltaGossipDigestList, deltaEpStateMap);
- Message gDigestAckMessage = Gossiper.instance().makeGossipDigestAckMessage(gDigestAck);
+ Message gDigestAckMessage = Gossiper.instance.makeGossipDigestAckMessage(gDigestAck);
if (logger_.isTraceEnabled())
logger_.trace("Sending a GossipDigestAckMessage to " + from);
MessagingService.instance().sendUdpOneWay(gDigestAckMessage, from);
@@ -928,8 +919,8 @@
for ( GossipDigest gDigest : gDigestList )
{
InetAddress ep = gDigest.getEndPoint();
- EndPointState epState = Gossiper.instance().getEndPointStateForEndPoint(ep);
- int version = (epState != null) ? Gossiper.instance().getMaxEndPointStateVersion( epState ) : 0;
+ EndPointState epState = Gossiper.instance.getEndPointStateForEndPoint(ep);
+ int version = (epState != null) ? Gossiper.instance.getMaxEndPointStateVersion( epState ) : 0;
int diffVersion = Math.abs(version - gDigest.getMaxVersion() );
diffDigests.add( new GossipDigest(ep, gDigest.getGeneration(), diffVersion) );
}
@@ -970,8 +961,8 @@
if ( epStateMap.size() > 0 )
{
/* Notify the Failure Detector */
- Gossiper.instance().notifyFailureDetector(epStateMap);
- Gossiper.instance().applyStateLocally(epStateMap);
+ Gossiper.instance.notifyFailureDetector(epStateMap);
+ Gossiper.instance.applyStateLocally(epStateMap);
}
/* Get the state required to send to this gossipee - construct GossipDigestAck2Message */
@@ -979,13 +970,13 @@
for( GossipDigest gDigest : gDigestList )
{
InetAddress addr = gDigest.getEndPoint();
- EndPointState localEpStatePtr = Gossiper.instance().getStateForVersionBiggerThan(addr, gDigest.getMaxVersion());
+ EndPointState localEpStatePtr = Gossiper.instance.getStateForVersionBiggerThan(addr, gDigest.getMaxVersion());
if ( localEpStatePtr != null )
deltaEpStateMap.put(addr, localEpStatePtr);
}
GossipDigestAck2Message gDigestAck2 = new GossipDigestAck2Message(deltaEpStateMap);
- Message gDigestAck2Message = Gossiper.instance().makeGossipDigestAck2Message(gDigestAck2);
+ Message gDigestAck2Message = Gossiper.instance.makeGossipDigestAck2Message(gDigestAck2);
if (logger_.isTraceEnabled())
logger_.trace("Sending a GossipDigestAck2Message to " + from);
MessagingService.instance().sendUdpOneWay(gDigestAck2Message, from);
@@ -1020,8 +1011,8 @@
}
Map<InetAddress, EndPointState> remoteEpStateMap = gDigestAck2Message.getEndPointStateMap();
/* Notify the Failure Detector */
- Gossiper.instance().notifyFailureDetector(remoteEpStateMap);
- Gossiper.instance().applyStateLocally(remoteEpStateMap);
+ Gossiper.instance.notifyFailureDetector(remoteEpStateMap);
+ Gossiper.instance.applyStateLocally(remoteEpStateMap);
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java Fri Jan 15 19:54:58 2010
@@ -159,7 +159,7 @@
*/
public static List<DecoratedKey> getIndexedDecoratedKeys()
{
- final Range range = StorageService.instance().getLocalPrimaryRange();
+ final Range range = StorageService.instance.getLocalPrimaryRange();
Predicate<SSTable> cfpred = Predicates.alwaysTrue();
return getIndexedDecoratedKeysFor(cfpred, new Predicate<DecoratedKey>()
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java Fri Jan 15 19:54:58 2010
@@ -169,11 +169,11 @@
StreamInitiateMessage biMsg = StreamInitiateMessage.serializer().deserialize(new DataInputStream(bufIn));
StreamContextManager.StreamContext[] streamContexts = biMsg.getStreamContext();
- if (streamContexts.length == 0 && StorageService.instance().isBootstrapMode())
+ if (streamContexts.length == 0 && StorageService.instance.isBootstrapMode())
{
if (logger.isDebugEnabled())
logger.debug("no data needed from " + message.getFrom());
- StorageService.instance().removeBootstrapSource(message.getFrom());
+ StorageService.instance.removeBootstrapSource(message.getFrom());
return;
}
@@ -313,9 +313,9 @@
MessagingService.instance().sendOneWay(message, host);
/* If we're done with everything for this host, remove from bootstrap sources */
- if (StreamContextManager.isDone(host) && StorageService.instance().isBootstrapMode())
+ if (StreamContextManager.isDone(host) && StorageService.instance.isBootstrapMode())
{
- StorageService.instance().removeBootstrapSource(host);
+ StorageService.instance.removeBootstrapSource(host);
}
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java Fri Jan 15 19:54:58 2010
@@ -112,7 +112,7 @@
for (InetAddress ep : targets)
{
- if (FailureDetector.instance().isAlive(ep))
+ if (FailureDetector.instance.isAlive(ep))
{
map.put(ep, ep);
usedEndpoints.add(ep);
@@ -135,7 +135,7 @@
for (int i = startIndex, count = 1; count < totalNodes; ++count, i = (i + 1) % totalNodes)
{
InetAddress tmpEndPoint = tokenMetadata_.getEndPoint((Token) tokens.get(i));
- if (FailureDetector.instance().isAlive(tmpEndPoint) && !targets.contains(tmpEndPoint) && !usedEndpoints.contains(tmpEndPoint))
+ if (FailureDetector.instance.isAlive(tmpEndPoint) && !targets.contains(tmpEndPoint) && !usedEndpoints.contains(tmpEndPoint))
{
hintLocation = tmpEndPoint;
break;
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java Fri Jan 15 19:54:58 2010
@@ -68,7 +68,7 @@
*/
private synchronized void loadEndPoints(TokenMetadata metadata) throws IOException
{
- endPointSnitch = (DatacenterEndPointSnitch) StorageService.instance().getEndPointSnitch();
+ endPointSnitch = (DatacenterEndPointSnitch) StorageService.instance.getEndPointSnitch();
this.tokens = new ArrayList<Token>(tokens);
String localDC = endPointSnitch.getLocation(InetAddress.getLocalHost());
dcMap = new HashMap<String, List<Token>>();
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java Fri Jan 15 19:54:58 2010
@@ -70,7 +70,7 @@
return endpoints;
}
startIndex = (index + 1)%totalNodes;
- EndPointSnitch endPointSnitch = (EndPointSnitch) StorageService.instance().getEndPointSnitch();
+ EndPointSnitch endPointSnitch = (EndPointSnitch) StorageService.instance.getEndPointSnitch();
for (int i = startIndex, count = 1; count < totalNodes && foundCount < replicas_; ++count, i = (i + 1) % totalNodes)
{
@@ -121,4 +121,4 @@
}
return endpoints;
}
-}
\ No newline at end of file
+}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java Fri Jan 15 19:54:58 2010
@@ -179,7 +179,7 @@
SelectionKey key = SelectorManager.getSelectorManager().register(serverChannel, handler, SelectionKey.OP_ACCEPT);
listenSockets_.put(localEp, key);
- FailureDetector.instance().registerFailureDetectionEventListener(this);
+ FailureDetector.instance.registerFailureDetectionEventListener(this);
}
/**
@@ -422,7 +422,7 @@
logger_.info("Shutting down ...");
synchronized (MessagingService.class)
{
- FailureDetector.instance().unregisterFailureDetectionEventListener(MessagingService.instance());
+ FailureDetector.instance.unregisterFailureDetectionEventListener(MessagingService.instance());
/* Stop listening on any TCP socket */
for (SelectionKey skey : listenSockets_.values())
{
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/AntiEntropyService.java Fri Jan 15 19:54:58 2010
@@ -98,7 +98,7 @@
public final static long NATURAL_REPAIR_FREQUENCY = 3600000;
// singleton enforcement
- private static volatile AntiEntropyService aeService;
+ public static final AntiEntropyService instance = new AntiEntropyService();
/**
* Map of CFPair to timestamp of the beginning of the last natural repair.
@@ -115,25 +115,10 @@
*/
private final Map<CFPair, Cachetable<InetAddress, TreePair>> trees;
- public static AntiEntropyService instance()
- {
- if (aeService == null)
- {
- synchronized (AntiEntropyService.class)
- {
- if (aeService == null)
- {
- aeService = new AntiEntropyService();
- }
- }
- }
- return aeService;
- }
-
/**
- * Private constructor. Use AntiEntropyService.instance()
+ * Protected constructor. Use AntiEntropyService.instance.
*/
- private AntiEntropyService()
+ protected AntiEntropyService()
{
MessagingService.instance().registerVerbHandlers(TREE_REQUEST_VERB, new TreeRequestVerbHandler());
MessagingService.instance().registerVerbHandlers(TREE_RESPONSE_VERB, new TreeResponseVerbHandler());
@@ -165,7 +150,7 @@
private static Collection<InetAddress> getNeighbors()
{
InetAddress local = FBUtilities.getLocalAddress();
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
return Collections2.filter(ss.getNaturalEndpoints(ss.getLocalToken()),
Predicates.not(Predicates.equalTo(local)));
}
@@ -311,7 +296,7 @@
{
if (!major || table.equals(Table.SYSTEM_TABLE))
return new NoopValidator();
- if (StorageService.instance().getTokenMetadata().sortedTokens().size() < 1)
+ if (StorageService.instance.getTokenMetadata().sortedTokens().size() < 1)
// gossiper isn't started
return new NoopValidator();
CFPair cfpair = new CFPair(table, cf);
@@ -495,7 +480,7 @@
*/
public Object call() throws Exception
{
- AntiEntropyService aes = AntiEntropyService.instance();
+ AntiEntropyService aes = AntiEntropyService.instance;
InetAddress local = FBUtilities.getLocalAddress();
Collection<InetAddress> neighbors = getNeighbors();
@@ -567,7 +552,7 @@
*/
public void run()
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
// restore partitioners (in case we were serialized)
if (ltree.partitioner() == null)
@@ -775,7 +760,7 @@
{
// deserialize the remote tree, and register it
Validator rvalidator = this.deserialize(new DataInputStream(buffer));
- AntiEntropyService.instance().rendezvous(rvalidator.cf, message.getFrom(), rvalidator.tree);
+ AntiEntropyService.instance.rendezvous(rvalidator.cf, message.getFrom(), rvalidator.tree);
}
catch (IOException e)
{
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraDaemon.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraDaemon.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraDaemon.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraDaemon.java Fri Jan 15 19:54:58 2010
@@ -95,7 +95,7 @@
CompactionManager.instance.checkAllColumnFamilies();
// start server internals
- StorageService.instance().initServer();
+ StorageService.instance.initServer();
// now we start listening for clients
CassandraServer cassandraServer = new CassandraServer();
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraServer.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraServer.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraServer.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/CassandraServer.java Fri Jan 15 19:54:58 2010
@@ -57,7 +57,7 @@
public CassandraServer()
{
- storageService = StorageService.instance();
+ storageService = StorageService.instance;
}
protected Map<String, ColumnFamily> readColumnFamily(List<ReadCommand> commands, ConsistencyLevel consistency_level)
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/LoadDisseminator.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/LoadDisseminator.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/LoadDisseminator.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/LoadDisseminator.java Fri Jan 15 19:54:58 2010
@@ -31,9 +31,9 @@
public void run()
{
- String diskUtilization = String.valueOf(StorageService.instance().getLoad());
+ String diskUtilization = String.valueOf(StorageService.instance.getLoad());
if (logger_.isDebugEnabled())
logger_.debug("Disseminating load info ...");
- Gossiper.instance().addApplicationState(LoadDisseminator.loadInfo_, new ApplicationState(diskUtilization));
+ Gossiper.instance.addApplicationState(LoadDisseminator.loadInfo_, new ApplicationState(diskUtilization));
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadRepairManager.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadRepairManager.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadRepairManager.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadRepairManager.java Fri Jan 15 19:54:58 2010
@@ -48,9 +48,8 @@
class ReadRepairManager
{
private static final Logger logger_ = Logger.getLogger(ReadRepairManager.class);
- private static final long expirationTimeInMillis = 2000;
- private static final Lock lock_ = new ReentrantLock();
- private static volatile ReadRepairManager self_ = null;
+ private static final long expirationTimeInMillis = 2000;
+ public static final ReadRepairManager instance = new ReadRepairManager();
/*
* This is the internal class which actually
@@ -87,24 +86,6 @@
}
- public static ReadRepairManager instance()
- {
- if (self_ == null)
- {
- lock_.lock();
- try
- {
- if (self_ == null)
- self_ = new ReadRepairManager();
- }
- finally
- {
- lock_.unlock();
- }
- }
- return self_;
- }
-
/*
* Schedules a read repair.
* @param target endpoint on which the read repair should happen
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java Fri Jan 15 19:54:58 2010
@@ -134,7 +134,7 @@
RowMutation rowMutation = new RowMutation(table, key);
rowMutation.add(diffCf);
RowMutationMessage rowMutationMessage = new RowMutationMessage(rowMutation);
- ReadRepairManager.instance().schedule(endPoints.get(i), rowMutationMessage);
+ ReadRepairManager.instance.schedule(endPoints.get(i), rowMutationMessage);
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageLoadBalancer.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageLoadBalancer.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageLoadBalancer.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageLoadBalancer.java Fri Jan 15 19:54:58 2010
@@ -69,14 +69,14 @@
/*
int threshold = (int)(StorageLoadBalancer.TOPHEAVY_RATIO * averageSystemLoad());
int myLoad = localLoad();
- InetAddress predecessor = StorageService.instance().getPredecessor(StorageService.getLocalStorageEndPoint());
+ InetAddress predecessor = StorageService.instance.getPredecessor(StorageService.getLocalStorageEndPoint());
if (logger_.isDebugEnabled())
logger_.debug("Trying to relocate the predecessor " + predecessor);
boolean value = tryThisNode(myLoad, threshold, predecessor);
if ( !value )
{
loadInfo2_.remove(predecessor);
- InetAddress successor = StorageService.instance().getSuccessor(StorageService.getLocalStorageEndPoint());
+ InetAddress successor = StorageService.instance.getSuccessor(StorageService.getLocalStorageEndPoint());
if (logger_.isDebugEnabled())
logger_.debug("Trying to relocate the successor " + successor);
value = tryThisNode(myLoad, threshold, successor);
@@ -166,13 +166,7 @@
private static final long BROADCAST_INTERVAL = 5 * 60 * 1000L;
- private static StorageLoadBalancer instance_;
-
- public static synchronized StorageLoadBalancer instance()
- {
- return instance_ == null ? (instance_ = new StorageLoadBalancer()) : instance_;
- }
-
+ public static final StorageLoadBalancer instance = new StorageLoadBalancer();
private static final Logger logger_ = Logger.getLogger(StorageLoadBalancer.class);
private static final String moveMessageVerbHandler_ = "MOVE-MESSAGE-VERB-HANDLER";
@@ -197,7 +191,7 @@
private StorageLoadBalancer()
{
MessagingService.instance().registerVerbHandlers(StorageLoadBalancer.moveMessageVerbHandler_, new MoveMessageVerbHandler());
- Gossiper.instance().register(this);
+ Gossiper.instance.register(this);
}
public void onChange(InetAddress endpoint, String stateName, ApplicationState state)
@@ -238,7 +232,7 @@
if ( !isMoveable_.get() )
return false;
int myload = localLoad();
- InetAddress successor = StorageService.instance().getSuccessor(StorageService.getLocalStorageEndPoint());
+ InetAddress successor = StorageService.instance.getSuccessor(StorageService.getLocalStorageEndPoint());
LoadInfo li = loadInfo2_.get(successor);
// "load" is NULL means that the successor node has not
// yet gossiped its load information. We should return
@@ -301,7 +295,7 @@
}
else
{
- InetAddress successor = StorageService.instance().getSuccessor(target);
+ InetAddress successor = StorageService.instance.getSuccessor(target);
double sLoad = loadInfo2_.get(successor);
double targetLoad = loadInfo2_.get(target);
return (sLoad + targetLoad) <= threshold;
@@ -310,11 +304,11 @@
private boolean isANeighbour(InetAddress neighbour)
{
- InetAddress predecessor = StorageService.instance().getPredecessor(FBUtilities.getLocalAddress());
+ InetAddress predecessor = StorageService.instance.getPredecessor(FBUtilities.getLocalAddress());
if ( predecessor.equals(neighbour) )
return true;
- InetAddress successor = StorageService.instance().getSuccessor(FBUtilities.getLocalAddress());
+ InetAddress successor = StorageService.instance.getSuccessor(FBUtilities.getLocalAddress());
if ( successor.equals(neighbour) )
return true;
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java Fri Jan 15 19:54:58 2010
@@ -103,9 +103,9 @@
for (final RowMutation rm: mutations)
{
try
- {
- List<InetAddress> naturalEndpoints = StorageService.instance().getNaturalEndpoints(rm.key());
- Map<InetAddress, InetAddress> endpointMap = StorageService.instance().getHintedEndpointMap(rm.key(), naturalEndpoints);
+ {
+ List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(rm.key());
+ Map<InetAddress, InetAddress> endpointMap = StorageService.instance.getHintedEndpointMap(rm.key(), naturalEndpoints);
Message unhintedMessage = null; // lazy initialize for non-local, unhinted writes
// 3 cases:
@@ -173,15 +173,15 @@
for (RowMutation rm: mutations)
{
mostRecentRowMutation = rm;
- List<InetAddress> naturalEndpoints = StorageService.instance().getNaturalEndpoints(rm.key());
- Map<InetAddress, InetAddress> endpointMap = StorageService.instance().getHintedEndpointMap(rm.key(), naturalEndpoints);
+ List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(rm.key());
+ Map<InetAddress, InetAddress> endpointMap = StorageService.instance.getHintedEndpointMap(rm.key(), naturalEndpoints);
int blockFor = determineBlockFor(naturalEndpoints.size(), endpointMap.size(), consistency_level);
// avoid starting a write we know can't achieve the required consistency
assureSufficientLiveNodes(endpointMap, blockFor);
// send out the writes, as in insert() above, but this time with a callback that tracks responses
- final WriteResponseHandler responseHandler = StorageService.instance().getWriteResponseHandler(blockFor, consistency_level);
+ final WriteResponseHandler responseHandler = StorageService.instance.getWriteResponseHandler(blockFor, consistency_level);
responseHandlers.add(responseHandler);
Message unhintedMessage = null;
for (Map.Entry<InetAddress, InetAddress> entry : endpointMap.entrySet())
@@ -320,7 +320,7 @@
for (ReadCommand command: commands)
{
- InetAddress endPoint = StorageService.instance().findSuitableEndPoint(command.key);
+ InetAddress endPoint = StorageService.instance.findSuitableEndPoint(command.key);
Message message = command.makeReadMessage();
if (logger.isDebugEnabled())
@@ -359,10 +359,10 @@
for (ReadCommand command: commands)
{
- List<InetAddress> endpoints = StorageService.instance().getNaturalEndpoints(command.key);
+ List<InetAddress> endpoints = StorageService.instance.getNaturalEndpoints(command.key);
boolean foundLocal = endpoints.contains(FBUtilities.getLocalAddress());
//TODO: Throw InvalidRequest if we're in bootstrap mode?
- if (foundLocal && !StorageService.instance().isBootstrapMode())
+ if (foundLocal && !StorageService.instance.isBootstrapMode())
{
localCommands.add(command);
}
@@ -417,8 +417,8 @@
Message message = command.makeReadMessage();
Message messageDigestOnly = readMessageDigestOnly.makeReadMessage();
- InetAddress dataPoint = StorageService.instance().findSuitableEndPoint(command.key);
- List<InetAddress> endpointList = StorageService.instance().getLiveNaturalEndpoints(command.key);
+ InetAddress dataPoint = StorageService.instance.findSuitableEndPoint(command.key);
+ List<InetAddress> endpointList = StorageService.instance.getLiveNaturalEndpoints(command.key);
if (endpointList.size() < responseCount)
throw new UnavailableException();
@@ -517,17 +517,17 @@
static List<Pair<String, ColumnFamily>> getRangeSlice(RangeSliceCommand command, ConsistencyLevel consistency_level) throws IOException, UnavailableException, TimeoutException
{
long startTime = System.currentTimeMillis();
- TokenMetadata tokenMetadata = StorageService.instance().getTokenMetadata();
+ TokenMetadata tokenMetadata = StorageService.instance.getTokenMetadata();
- InetAddress endPoint = StorageService.instance().getPrimary(command.startKey.token);
+ InetAddress endPoint = StorageService.instance.getPrimary(command.startKey.token);
InetAddress startEndpoint = endPoint;
int responseCount = determineBlockFor(DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getReplicationFactor(), consistency_level);
Map<String, ColumnFamily> rows = new HashMap<String, ColumnFamily>(command.max_keys);
do
{
- Range primaryRange = StorageService.instance().getPrimaryRangeForEndPoint(endPoint);
- List<InetAddress> endpoints = StorageService.instance().getLiveNaturalEndpoints(primaryRange.right());
+ Range primaryRange = StorageService.instance.getPrimaryRangeForEndPoint(endPoint);
+ List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(primaryRange.right());
if (endpoints.size() < responseCount)
throw new UnavailableException();
@@ -595,10 +595,10 @@
static List<String> getKeyRange(RangeCommand command) throws IOException, UnavailableException, TimeoutException
{
long startTime = System.currentTimeMillis();
- TokenMetadata tokenMetadata = StorageService.instance().getTokenMetadata();
+ TokenMetadata tokenMetadata = StorageService.instance.getTokenMetadata();
Set<String> uniqueKeys = new HashSet<String>(command.maxResults);
- InetAddress endPoint = StorageService.instance().findSuitableEndPoint(command.startWith);
+ InetAddress endPoint = StorageService.instance.findSuitableEndPoint(command.startWith);
InetAddress startEndpoint = endPoint;
do
@@ -628,7 +628,7 @@
do
{
endPoint = tokenMetadata.getSuccessor(endPoint);
- } while (!FailureDetector.instance().isAlive(endPoint));
+ } while (!FailureDetector.instance.isAlive(endPoint));
} while (!endPoint.equals(startEndpoint));
rangeStats.add(System.currentTimeMillis() - startTime);
@@ -680,7 +680,7 @@
public Object call() throws IOException
{
- List<InetAddress> endpoints = StorageService.instance().getLiveNaturalEndpoints(command.key);
+ List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(command.key);
/* Remove the local storage endpoint from the list. */
endpoints.remove(FBUtilities.getLocalAddress());
@@ -692,7 +692,7 @@
// Do the consistency checks in the background and return the non NULL row
if (endpoints.size() > 0 && DatabaseDescriptor.getConsistencyCheck())
- StorageService.instance().doConsistencyCheck(row, endpoints, command);
+ StorageService.instance.doConsistencyCheck(row, endpoints, command);
return row;
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java Fri Jan 15 19:54:58 2010
@@ -89,7 +89,7 @@
private static IPartitioner partitioner_ = DatabaseDescriptor.getPartitioner();
- private static volatile StorageService instance_;
+ public static final StorageService instance = new StorageService();
public static IPartitioner getPartitioner() {
return partitioner_;
@@ -106,25 +106,6 @@
}
/*
- * Factory method that gets an instance of the StorageService
- * class.
- */
- public static StorageService instance()
- {
- if (instance_ == null)
- {
- synchronized (StorageService.class)
- {
- if (instance_ == null)
- {
- instance_ = new StorageService();
- }
- }
- }
- return instance_;
- }
-
- /*
* This is the endpoint snitch which depends on the network architecture. We
* need to keep this information for each endpoint so that we make decisions
* while doing things like replication etc.
@@ -176,7 +157,7 @@
isBootstrapMode = false;
SystemTable.setBootstrapped(true);
setToken(getLocalToken());
- Gossiper.instance().addApplicationState(MOVE_STATE, new ApplicationState(STATE_NORMAL + Delimiter + partitioner_.getTokenFactory().toString(getLocalToken())));
+ Gossiper.instance.addApplicationState(MOVE_STATE, new ApplicationState(STATE_NORMAL + Delimiter + partitioner_.getTokenFactory().toString(getLocalToken())));
logger_.info("Bootstrap/move completed! Now serving reads.");
}
@@ -240,8 +221,8 @@
public void stopClient()
{
- Gossiper.instance().unregister(this);
- Gossiper.instance().stop();
+ Gossiper.instance.unregister(this);
+ Gossiper.instance.stop();
MessagingService.shutdown();
}
@@ -255,8 +236,8 @@
SelectorManager.getSelectorManager().start();
SelectorManager.getUdpSelectorManager().start();
- Gossiper.instance().register(this);
- Gossiper.instance().start(FBUtilities.getLocalAddress(), (int)(System.currentTimeMillis() / 1000)); // needed for node-ring gathering.
+ Gossiper.instance.register(this);
+ Gossiper.instance.start(FBUtilities.getLocalAddress(), (int)(System.currentTimeMillis() / 1000)); // needed for node-ring gathering.
}
public void initServer() throws IOException
@@ -274,27 +255,26 @@
SelectorManager.getSelectorManager().start();
SelectorManager.getUdpSelectorManager().start();
- AntiEntropyService.instance();
- StorageLoadBalancer.instance().startBroadcasting();
+ StorageLoadBalancer.instance.startBroadcasting();
// have to start the gossip service before we can see any info on other nodes. this is necessary
// for bootstrap to get the load info it needs.
// (we won't be part of the storage ring though until we add a nodeId to our state, below.)
- Gossiper.instance().register(this);
- Gossiper.instance().start(FBUtilities.getLocalAddress(), storageMetadata_.getGeneration()); // needed for node-ring gathering.
+ Gossiper.instance.register(this);
+ Gossiper.instance.start(FBUtilities.getLocalAddress(), storageMetadata_.getGeneration()); // needed for node-ring gathering.
if (DatabaseDescriptor.isAutoBootstrap()
&& !(DatabaseDescriptor.getSeeds().contains(FBUtilities.getLocalAddress()) || SystemTable.isBootstrapped()))
{
logger_.info("Starting in bootstrap mode (first, sleeping to get load information)");
- StorageLoadBalancer.instance().waitForLoadInfo();
+ StorageLoadBalancer.instance.waitForLoadInfo();
logger_.info("... got load info");
if (tokenMetadata_.isMember(FBUtilities.getLocalAddress()))
{
String s = "This node is already a member of the token ring; bootstrap aborted. (If replacing a dead node, remove the old one from the ring first.)";
throw new UnsupportedOperationException(s);
}
- Token token = BootStrapper.getBootstrapToken(tokenMetadata_, StorageLoadBalancer.instance().getLoadInfo());
+ Token token = BootStrapper.getBootstrapToken(tokenMetadata_, StorageLoadBalancer.instance.getLoadInfo());
startBootstrap(token);
// don't finish startup (enabling thrift) until after bootstrap is done
while (isBootstrapMode)
@@ -314,7 +294,7 @@
SystemTable.setBootstrapped(true);
Token token = storageMetadata_.getToken();
tokenMetadata_.updateNormalToken(token, FBUtilities.getLocalAddress());
- Gossiper.instance().addApplicationState(MOVE_STATE, new ApplicationState(STATE_NORMAL + Delimiter + partitioner_.getTokenFactory().toString(token)));
+ Gossiper.instance.addApplicationState(MOVE_STATE, new ApplicationState(STATE_NORMAL + Delimiter + partitioner_.getTokenFactory().toString(token)));
}
assert tokenMetadata_.sortedTokens().size() > 0;
@@ -324,7 +304,7 @@
{
isBootstrapMode = true;
SystemTable.updateToken(token); // DON'T use setToken, that makes us part of the ring locally which is incorrect until we are done bootstrapping
- Gossiper.instance().addApplicationState(MOVE_STATE, new ApplicationState(STATE_BOOTSTRAPPING + Delimiter + partitioner_.getTokenFactory().toString(token)));
+ Gossiper.instance.addApplicationState(MOVE_STATE, new ApplicationState(STATE_BOOTSTRAPPING + Delimiter + partitioner_.getTokenFactory().toString(token)));
logger_.info("bootstrap sleeping " + Streaming.RING_DELAY);
try
{
@@ -675,7 +655,7 @@
Multimap<Range, InetAddress> rangeAddresses = replicationStrategy_.getRangeAddresses(tokenMetadata_);
Multimap<InetAddress, Range> sourceRanges = HashMultimap.create();
- IFailureDetector failureDetector = FailureDetector.instance();
+ IFailureDetector failureDetector = FailureDetector.instance;
// find alive sources for our new ranges
for (Range myNewRange : myNewRanges)
@@ -799,7 +779,7 @@
public Map<String, String> getLoadMap()
{
Map<String, String> map = new HashMap<String, String>();
- for (Map.Entry<InetAddress,Double> entry : StorageLoadBalancer.instance().getLoadInfo().entrySet())
+ for (Map.Entry<InetAddress,Double> entry : StorageLoadBalancer.instance.getLoadInfo().entrySet())
{
map.put(entry.getKey().getHostAddress(), FileUtils.stringifyFileSize(entry.getValue()));
}
@@ -814,7 +794,7 @@
*/
public final void deliverHints(InetAddress endpoint)
{
- HintedHandOffManager.instance().deliverHints(endpoint);
+ HintedHandOffManager.instance.deliverHints(endpoint);
}
public Token getLocalToken()
@@ -831,12 +811,12 @@
public Set<String> getLiveNodes()
{
- return stringify(Gossiper.instance().getLiveMembers());
+ return stringify(Gossiper.instance.getLiveMembers());
}
public Set<String> getUnreachableNodes()
{
- return stringify(Gossiper.instance().getUnreachableMembers());
+ return stringify(Gossiper.instance.getUnreachableMembers());
}
private Set<String> stringify(Set<InetAddress> endPoints)
@@ -861,7 +841,7 @@
public int getCurrentGenerationNumber()
{
- return Gossiper.instance().getCurrentGenerationNumber(FBUtilities.getLocalAddress());
+ return Gossiper.instance.getCurrentGenerationNumber(FBUtilities.getLocalAddress());
}
public void forceTableCleanup() throws IOException
@@ -1150,7 +1130,7 @@
for (InetAddress endpoint : endpoints)
{
- if (FailureDetector.instance().isAlive(endpoint))
+ if (FailureDetector.instance.isAlive(endpoint))
liveEps.add(endpoint);
}
@@ -1178,7 +1158,7 @@
endPointSnitch_.sortByProximity(FBUtilities.getLocalAddress(), endpoints);
for (InetAddress endpoint : endpoints)
{
- if (FailureDetector.instance().isAlive(endpoint))
+ if (FailureDetector.instance.isAlive(endpoint))
return endpoint;
}
throw new UnavailableException(); // no nodes that could contain key are alive
@@ -1245,7 +1225,7 @@
*/
private void startLeaving()
{
- Gossiper.instance().addApplicationState(MOVE_STATE, new ApplicationState(STATE_LEAVING + Delimiter + getLocalToken().toString()));
+ Gossiper.instance.addApplicationState(MOVE_STATE, new ApplicationState(STATE_LEAVING + Delimiter + getLocalToken().toString()));
tokenMetadata_.addLeavingEndPoint(FBUtilities.getLocalAddress());
calculatePendingRanges();
}
@@ -1268,7 +1248,7 @@
{
public void run()
{
- Gossiper.instance().stop();
+ Gossiper.instance.stop();
MessagingService.shutdown();
logger_.info("DECOMMISSION FINISHED.");
// let op be responsible for killing the process
@@ -1285,7 +1265,7 @@
if (logger_.isDebugEnabled())
logger_.debug("");
- Gossiper.instance().addApplicationState(MOVE_STATE, new ApplicationState(STATE_LEFT + Delimiter + LEFT_NORMALLY + Delimiter + getLocalToken().toString()));
+ Gossiper.instance.addApplicationState(MOVE_STATE, new ApplicationState(STATE_LEFT + Delimiter + LEFT_NORMALLY + Delimiter + getLocalToken().toString()));
try
{
Thread.sleep(2 * Gossiper.intervalInMillis_);
@@ -1369,7 +1349,7 @@
{
Token bootstrapToken = token;
if (bootstrapToken == null)
- bootstrapToken = BootStrapper.getBalancedToken(tokenMetadata_, StorageLoadBalancer.instance().getLoadInfo());
+ bootstrapToken = BootStrapper.getBalancedToken(tokenMetadata_, StorageLoadBalancer.instance.getLoadInfo());
logger_.info("re-bootstrapping to new token " + bootstrapToken);
startBootstrap(bootstrapToken);
}
@@ -1390,7 +1370,7 @@
{
// Let's make sure however that we're not removing a live
// token (member)
- if (Gossiper.instance().getLiveMembers().contains(endPoint))
+ if (Gossiper.instance.getLiveMembers().contains(endPoint))
throw new UnsupportedOperationException("Node " + endPoint + " is alive and owns this token. Use decommission command to remove it from the ring");
restoreReplicaCount(endPoint);
@@ -1405,7 +1385,7 @@
// not good. REMOVE_TOKEN|LEFT_NORMALLY is used to distinguish
// between removetoken command and normal state left, so it is
// not so bad.
- Gossiper.instance().addApplicationState(MOVE_STATE, new ApplicationState(STATE_LEFT + Delimiter + REMOVE_TOKEN + Delimiter + token.toString()));
+ Gossiper.instance.addApplicationState(MOVE_STATE, new ApplicationState(STATE_LEFT + Delimiter + REMOVE_TOKEN + Delimiter + token.toString()));
}
public WriteResponseHandler getWriteResponseHandler(int blockFor, ConsistencyLevel consistency_level)
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/BootStrapperTest.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/BootStrapperTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/BootStrapperTest.java Fri Jan 15 19:54:58 2010
@@ -42,7 +42,7 @@
@Test
public void testGuessToken() throws IOException
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
generateFakeEndpoints(3);
@@ -79,7 +79,7 @@
private void testSourceTargetComputation(int numOldNodes) throws UnknownHostException
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
generateFakeEndpoints(numOldNodes);
Token myToken = StorageService.getPartitioner().getRandomToken();
@@ -119,7 +119,7 @@
private void generateFakeEndpoints(int numOldNodes) throws UnknownHostException
{
- TokenMetadata tmd = StorageService.instance().getTokenMetadata();
+ TokenMetadata tmd = StorageService.instance.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner<?> p = StorageService.getPartitioner();
@@ -129,4 +129,4 @@
tmd.updateNormalToken(p.getRandomToken(), InetAddress.getByName("127.0.0." + (i + 1)));
}
}
-}
\ No newline at end of file
+}
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/StreamingTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/StreamingTest.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/StreamingTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/StreamingTest.java Fri Jan 15 19:54:58 2010
@@ -43,7 +43,7 @@
@Test
public void testTransferTable() throws Exception
{
- StorageService.instance().initServer();
+ StorageService.instance.initServer();
// write a temporary SSTable, but don't register it
Set<String> content = new HashSet<String>();
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java Fri Jan 15 19:54:58 2010
@@ -63,9 +63,9 @@
// bump the replication factor so that local overlaps with REMOTE below
DatabaseDescriptorTest.setReplicationFactor(2);
- StorageService.instance().initServer();
+ StorageService.instance.initServer();
// generate a fake endpoint for which we can spoof receiving/sending trees
- TokenMetadata tmd = StorageService.instance().getTokenMetadata();
+ TokenMetadata tmd = StorageService.instance.getTokenMetadata();
IPartitioner part = StorageService.getPartitioner();
REMOTE = InetAddress.getByName("127.0.0.2");
tmd.updateNormalToken(part.getMinimumToken(), REMOTE);
@@ -75,14 +75,14 @@
cfname = Table.open(tablename).getColumnFamilies().iterator().next();
initialized = true;
}
- aes = AntiEntropyService.instance();
+ aes = AntiEntropyService.instance;
}
@Test
public void testInstance() throws Throwable
{
assert null != aes;
- assert aes == AntiEntropyService.instance();
+ assert aes == AntiEntropyService.instance;
}
@Test
@@ -218,7 +218,7 @@
MerkleTree rtree = validator.tree;
// change a range in one of the trees
- Token min = StorageService.instance().getPartitioner().getMinimumToken();
+ Token min = StorageService.instance.getPartitioner().getMinimumToken();
ltree.invalidate(min);
MerkleTree.TreeRange changed = ltree.invalids(new Range(min, min)).next();
changed.hash("non-empty hash!".getBytes());
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/MoveTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/MoveTest.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/MoveTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/MoveTest.java Fri Jan 15 19:54:58 2010
@@ -47,7 +47,7 @@
@Test
public void testWriteEndPointsDuringLeave() throws UnknownHostException
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
@@ -104,7 +104,7 @@
@Test
public void testSimultaneousMove() throws UnknownHostException
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
@@ -286,7 +286,7 @@
@Test
public void testStateJumpToBootstrap() throws UnknownHostException
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
@@ -355,7 +355,7 @@
@Test
public void testStateJumpToNormal() throws UnknownHostException
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
@@ -399,7 +399,7 @@
@Test
public void testStateJumpToLeaving() throws UnknownHostException
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
@@ -449,7 +449,7 @@
@Test
public void testStateJumpToLeft() throws UnknownHostException
{
- StorageService ss = StorageService.instance();
+ StorageService ss = StorageService.instance;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/StorageServiceTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/StorageServiceTest.java?rev=899791&r1=899790&r2=899791&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/StorageServiceTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/service/StorageServiceTest.java Fri Jan 15 19:54:58 2010
@@ -36,14 +36,14 @@
{
CleanupHelper.mkdirs();
CleanupHelper.cleanup();
- StorageService.instance().initClient();
+ StorageService.instance.initClient();
// verify that no storage directories were created.
for (String path : DatabaseDescriptor.getAllDataFileLocations())
{
assertFalse(new File(path).exists());
}
- StorageService.instance().stopClient();
+ StorageService.instance.stopClient();
NetPackageAccessor.resetSelectorManager();
}
@@ -52,7 +52,7 @@
{
CleanupHelper.mkdirs();
CleanupHelper.cleanup();
- StorageService.instance().initServer();
+ StorageService.instance.initServer();
for (String path : DatabaseDescriptor.getAllDataFileLocations())
{
// verify that storage directories are there.
@@ -61,8 +61,8 @@
// a proper test would be to call decommission here, but decommission() mixes both shutdown and datatransfer
// calls. This test is only interested in the shutdown-related items which a properly handled by just
// stopping the client.
- //StorageService.instance().decommission();
- StorageService.instance().stopClient();
+ //StorageService.instance.decommission();
+ StorageService.instance.stopClient();
NetPackageAccessor.resetSelectorManager();
}
}