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

merge from 0.5 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@908319 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-02-10 02:14:47 +00:00
parent dfdb54acce
commit 17bf5d8ef0
5 changed files with 24 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -28,6 +28,7 @@ dev
0.5.1
* ensure all files for an sstable are streamed to the same directory.
(CASSANDRA-716)
* more accurate load estimate for bootstrapping (CASSANDRA-762)
0.5.0 final

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

@ -163,7 +163,7 @@ public class StorageLoadBalancer implements IEndPointStateChangeSubscriber
}
}
private static final long BROADCAST_INTERVAL = 5 * 60 * 1000L;
private static final int BROADCAST_INTERVAL = 60 * 1000;
public static final StorageLoadBalancer instance = new StorageLoadBalancer();
@ -353,18 +353,17 @@ public class StorageLoadBalancer implements IEndPointStateChangeSubscriber
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,
this could be up to threshold_ ms (currently 5 minutes). */
/**
* Wait for at least BROADCAST_INTERVAL ms, to give all nodes enough time to
* report in.
*/
public void waitForLoadInfo()
{
int duration = BROADCAST_INTERVAL + StorageService.RING_DELAY;
try
{
while (loadInfo_.isEmpty())
{
Thread.sleep(100);
}
// one more sleep in case there are some stragglers
Thread.sleep(StorageService.RING_DELAY);
logger_.info("Sleeping " + duration + " ms to wait for load information...");
Thread.sleep(duration);
}
catch (InterruptedException e)
{

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

@ -67,7 +67,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
{
private static Logger logger_ = Logger.getLogger(StorageService.class);
public static final long RING_DELAY = 30 * 1000; // delay after which we assume ring has stablized
public static final int RING_DELAY = 30 * 1000; // delay after which we assume ring has stablized
public final static String MOVE_STATE = "MOVE";
@ -313,7 +313,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
if (DatabaseDescriptor.isAutoBootstrap()
&& !(DatabaseDescriptor.getSeeds().contains(FBUtilities.getLocalAddress()) || SystemTable.isBootstrapped()))
{
logger_.info("Starting in bootstrap mode (first, sleeping to get load information)");
logger_.info("Starting in bootstrap mode");
StorageLoadBalancer.instance.waitForLoadInfo();
logger_.info("... got load info");
if (tokenMetadata_.isMember(FBUtilities.getLocalAddress()))
@ -1271,6 +1271,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
throw new UnsupportedOperationException("data is currently moving to this node; unable to leave the ring");
}
// leave the ring
logger_.info("DECOMMISSIONING");
startLeaving();
logger_.info("decommission sleeping " + RING_DELAY);
@ -1362,7 +1363,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
onFinish.run();
}
public void move(String newToken) throws InterruptedException
public void move(String newToken) throws IOException, InterruptedException
{
move(partitioner_.getTokenFactory().fromString(newToken));
}
@ -1377,14 +1378,17 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
*
* @param token new token to boot to, or if null, find balanced token to boot to
*/
private void move(final Token token) throws InterruptedException
private void move(final Token token) throws IOException, InterruptedException
{
for (String table : DatabaseDescriptor.getTables())
{
if (tokenMetadata_.getPendingRanges(table, FBUtilities.getLocalAddress()).size() > 0)
throw new UnsupportedOperationException("data is currently moving to this node; unable to leave the ring");
}
if (token != null && tokenMetadata_.sortedTokens().contains(token))
throw new IOException("target token " + token + " is already owned by another node");
// leave the ring
logger_.info("starting move. leaving token " + getLocalToken());
startLeaving();
logger_.info("move sleeping " + RING_DELAY);
@ -1395,8 +1399,11 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
public void runMayThrow() throws IOException
{
Token bootstrapToken = token;
if (bootstrapToken == null)
bootstrapToken = BootStrapper.getBalancedToken(tokenMetadata_, StorageLoadBalancer.instance.getLoadInfo());
if (bootstrapToken == null)
{
StorageLoadBalancer.instance.waitForLoadInfo();
bootstrapToken = BootStrapper.getBalancedToken(tokenMetadata_, StorageLoadBalancer.instance.getLoadInfo());
}
logger_.info("re-bootstrapping to new token " + bootstrapToken);
startBootstrap(bootstrapToken);
}

2
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -141,7 +141,7 @@ public interface StorageServiceMBean
* @param newToken token to move this node to.
* This node will unload its data onto its neighbors, and bootstrap to the new token.
*/
public void move(String newToken) throws InterruptedException;
public void move(String newToken) throws IOException, InterruptedException;
/**
* This node will unload its data onto its neighbors, and bootstrap to share the range

2
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -325,7 +325,7 @@ public class NodeProbe
ssProxy.loadBalance();
}
public void move(String newToken) throws InterruptedException
public void move(String newToken) throws IOException, InterruptedException
{
ssProxy.move(newToken);
}