From 4a4b8bebf76e7b4bc494a020435277f47b0bcfaf Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Fri, 9 Oct 2009 17:07:24 +0000 Subject: [PATCH] fix regressions patch by jbellis reviewed by Eric Evans for CASSANDRA-477 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@823617 13f79535-47bb-0310-9956-ffa450edef68 --- .../apache/cassandra/dht/BootStrapper.java | 2 ++ .../org/apache/cassandra/gms/Gossiper.java | 4 ++- .../locator/AbstractReplicationStrategy.java | 2 +- .../service/StorageLoadBalancer.java | 5 +-- .../cassandra/service/StorageService.java | 32 +++++++++++++------ 5 files changed, 32 insertions(+), 13 deletions(-) diff --git a/src/java/org/apache/cassandra/dht/BootStrapper.java b/src/java/org/apache/cassandra/dht/BootStrapper.java index 3592c411cd..7fd3e52c52 100644 --- a/src/java/org/apache/cassandra/dht/BootStrapper.java +++ b/src/java/org/apache/cassandra/dht/BootStrapper.java @@ -37,6 +37,7 @@ package org.apache.cassandra.dht; import org.apache.commons.lang.ArrayUtils; import org.apache.cassandra.locator.TokenMetadata; + import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.net.*; import org.apache.cassandra.net.io.StreamContextManager; import org.apache.cassandra.net.io.IStreamComplete; @@ -208,6 +209,7 @@ public class BootStrapper implements Runnable if (!maxEndpoint.equals(StorageService.getLocalStorageEndPoint())) { + StorageService.instance().retrofitPorts(Arrays.asList(maxEndpoint)); Token t = getBootstrapTokenFrom(maxEndpoint); logger_.info("Setting token to " + t + " to assume load from " + maxEndpoint.getHost()); ss.updateToken(t); diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 4b4e2ae4b4..7d9b2c671a 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -95,7 +95,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC final static String GOSSIP_DIGEST_ACK_VERB = "GAV"; /* GA2V - abbreviation for GOSSIP-DIGEST-ACK2-VERB */ final static String GOSSIP_DIGEST_ACK2_VERB = "GA2V"; - final static int intervalInMillis_ = 1000; + public final static int intervalInMillis_ = 1000; private static Logger logger_ = Logger.getLogger(Gossiper.class); static Gossiper gossiper_; @@ -522,6 +522,8 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC { reqdEndPointState = new EndPointState(epState.getHeartBeatState()); } + if (logger_.isTraceEnabled()) + logger_.trace("Adding state " + key + ": " + appState.getState()); reqdEndPointState.addApplicationState(key, appState); } } diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index fe85bbcea4..cb6b103629 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -73,7 +73,7 @@ public abstract class AbstractReplicationStrategy * This method changes the ports of the endpoints from * the control port to the storage ports. */ - protected void retrofitPorts(List eps) + public void retrofitPorts(List eps) { for ( EndPoint ep : eps ) { diff --git a/src/java/org/apache/cassandra/service/StorageLoadBalancer.java b/src/java/org/apache/cassandra/service/StorageLoadBalancer.java index cc34577ead..82ccac5f78 100644 --- a/src/java/org/apache/cassandra/service/StorageLoadBalancer.java +++ b/src/java/org/apache/cassandra/service/StorageLoadBalancer.java @@ -351,8 +351,9 @@ public final class StorageLoadBalancer implements IEndPointStateChangeSubscriber public void startBroadcasting() { - /* starts a load timer thread */ - loadTimer_.schedule(new LoadDisseminator(), BROADCAST_INTERVAL, BROADCAST_INTERVAL); + // send the first broadcast "right away" (i.e., in 2 gossip heartbeats, when we should have someone to talk to); + // after that send every BROADCAST_INTERVAL. + loadTimer_.schedule(new LoadDisseminator(), 2 * Gossiper.intervalInMillis_, BROADCAST_INTERVAL); } /** wait for node information to be available. if the rest of the cluster just came up, diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 9c50710630..3ee24d7a18 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -226,12 +226,12 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto MessagingService.instance().registerVerbHandlers(mbrshipCleanerVerbHandler_, new MembershipCleanerVerbHandler() ); MessagingService.instance().registerVerbHandlers(rangeVerbHandler_, new RangeVerbHandler()); // see BootStrapper for a summary of how the bootstrap verbs interact + MessagingService.instance().registerVerbHandlers(bootstrapTokenVerbHandler_, new BootStrapper.BootstrapTokenVerbHandler()); MessagingService.instance().registerVerbHandlers(bootstrapMetadataVerbHandler_, new BootstrapMetadataVerbHandler() ); MessagingService.instance().registerVerbHandlers(bootStrapInitiateVerbHandler_, new BootStrapper.BootStrapInitiateVerbHandler()); MessagingService.instance().registerVerbHandlers(bootStrapInitiateDoneVerbHandler_, new BootStrapper.BootstrapInitiateDoneVerbHandler()); MessagingService.instance().registerVerbHandlers(bootStrapTerminateVerbHandler_, new BootStrapper.BootstrapTerminateVerbHandler()); - MessagingService.instance().registerVerbHandlers(bootstrapTokenVerbHandler_, new BootStrapper.BootstrapTokenVerbHandler()); - + StageManager.registerStage(StorageService.mutationStage_, new MultiThreadedStage(StorageService.mutationStage_, DatabaseDescriptor.getConcurrentWriters())); StageManager.registerStage(StorageService.readStage_, @@ -266,15 +266,24 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto StorageLoadBalancer.instance().startBroadcasting(); - if (isBootstrapMode) - { - BootStrapper.startBootstrap(); - } - + // 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(udpAddr_, storageMetadata_.getGeneration()); - /* Make sure this token gets gossiped around. */ - tokenMetadata_.update(storageMetadata_.getToken(), StorageService.tcpAddr_, isBootstrapMode); + + if (isBootstrapMode) + { + BootStrapper.startBootstrap(); // handles token update + } + else + { + tokenMetadata_.update(storageMetadata_.getToken(), StorageService.tcpAddr_, isBootstrapMode); + } + + // Gossip my token. + // note that before we do this we've (a) finalized what the token is actually going to be, and + // (b) added a bootstrap state (done by startBootstrap) ApplicationState state = new ApplicationState(StorageService.getPartitioner().getTokenFactory().toString(storageMetadata_.getToken())); Gossiper.instance().addApplicationState(StorageService.nodeId_, state); } @@ -935,6 +944,11 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto return nodePicker_.getHintedStorageEndPoints(partitioner_.getToken(key)); } + public void retrofitPorts(List eps) + { + nodePicker_.retrofitPorts(eps); + } + /** * This function finds the most suitable endpoint given a key. * It checks for locality and alive test.