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

Assert the local node is never hinted and make PAXOS commit not hint 

patch by aweisberg; reviewed by slebresne for CASSANDRA-10477
This commit is contained in:
Ariel Weisberg 2015-11-17 17:34:43 -05:00 committed by Sylvain Lebresne
parent 10bbcfcd9c
commit df342b952c
4 changed files with 74 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.1.13
* Avoid AssertionError while submitting hint with LWT (CASSANDRA-10477)
* If CompactionMetadata is not in stats file, use index summary instead (CASSANDRA-10676)
* Retry sending gossip syn multiple times during shadow round (CASSANDRA-8072)
* Fix pending range calculation during moves (CASSANDRA-10887)

3
src/java/org/apache/cassandra/net/WriteCallbackInfo.java
View File

@ -25,6 +25,7 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.utils.FBUtilities;
public class WriteCallbackInfo extends CallbackInfo
{
@ -44,6 +45,8 @@ public class WriteCallbackInfo extends CallbackInfo
this.sentMessage = message;
this.consistencyLevel = consistencyLevel;
this.allowHints = allowHints;
//Local writes shouldn't go through messaging service (https://issues.apache.org/jira/browse/CASSANDRA-10477)
assert (!target.equals(FBUtilities.getBroadcastAddress()));
}
Mutation mutation()

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

@ -76,7 +76,7 @@ public abstract class AbstractReadExecutor
private static boolean isLocalRequest(InetAddress replica)
{
return replica.equals(FBUtilities.getBroadcastAddress()) && StorageProxy.OPTIMIZE_LOCAL_REQUESTS;
return replica.equals(FBUtilities.getBroadcastAddress());
}
protected void makeDataRequests(Iterable<InetAddress> endpoints)

91
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -59,6 +59,7 @@ import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.*;
import org.apache.cassandra.net.MessagingService.Verb;
import org.apache.cassandra.service.paxos.*;
import org.apache.cassandra.sink.SinkManager;
import org.apache.cassandra.tracing.Tracing;
@ -69,7 +70,6 @@ public class StorageProxy implements StorageProxyMBean
{
public static final String MBEAN_NAME = "org.apache.cassandra.db:type=StorageProxy";
private static final Logger logger = LoggerFactory.getLogger(StorageProxy.class);
static final boolean OPTIMIZE_LOCAL_REQUESTS = true; // set to false to test messagingservice path on single node
public static final String UNREACHABLE = "UNREACHABLE";
@ -493,12 +493,20 @@ public class StorageProxy implements StorageProxyMBean
MessageOut<Commit> message = new MessageOut<Commit>(MessagingService.Verb.PAXOS_COMMIT, proposal, Commit.serializer);
for (InetAddress destination : Iterables.concat(naturalEndpoints, pendingEndpoints))
{
if (FailureDetector.instance.isAlive(destination))
{
if (shouldBlock)
MessagingService.instance().sendRR(message, destination, responseHandler, shouldHint);
{
if (destination.equals(FBUtilities.getBroadcastAddress()))
commitPaxosLocal(message, responseHandler);
else
MessagingService.instance().sendRR(message, destination, responseHandler, shouldHint);
}
else
{
MessagingService.instance().sendOneWay(message, destination);
}
}
else if (shouldHint)
{
@ -510,6 +518,30 @@ public class StorageProxy implements StorageProxyMBean
responseHandler.get();
}
/**
* Commit a PAXOS task locally, and if the task times out rather then submitting a real hint
* submit a fake one that executes immediately on the mutation stage, but generates the necessary backpressure
* signal for hints
*/
private static void commitPaxosLocal(final MessageOut<Commit> message, final AbstractWriteResponseHandler responseHandler)
{
StageManager.getStage(MessagingService.verbStages.get(MessagingService.Verb.PAXOS_COMMIT)).maybeExecuteImmediately(new LocalMutationRunnable()
{
public void runMayThrow()
{
PaxosState.commit(message.payload);
if (responseHandler != null)
responseHandler.response(null);
}
@Override
protected Verb verb()
{
return MessagingService.Verb.PAXOS_COMMIT;
}
});
}
/**
* Use this method to have these Mutations applied
* across all replicas. This method will take care
@ -692,7 +724,7 @@ public class StorageProxy implements StorageProxyMBean
for (InetAddress target : endpoints)
{
int targetVersion = MessagingService.instance().getVersion(target);
if (target.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS)
if (target.equals(FBUtilities.getBroadcastAddress()))
{
insertLocal(message.payload, handler);
}
@ -726,7 +758,7 @@ public class StorageProxy implements StorageProxyMBean
MessageOut<Mutation> message = mutation.createMessage();
for (InetAddress target : endpoints)
{
if (target.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS)
if (target.equals(FBUtilities.getBroadcastAddress()))
insertLocal(message.payload, handler);
else
MessagingService.instance().sendRR(message, target, handler, false);
@ -865,20 +897,11 @@ public class StorageProxy implements StorageProxyMBean
for (InetAddress destination : targets)
{
// avoid OOMing due to excess hints. we need to do this check even for "live" nodes, since we can
// still generate hints for those if it's overloaded or simply dead but not yet known-to-be-dead.
// The idea is that if we have over maxHintsInProgress hints in flight, this is probably due to
// a small number of nodes causing problems, so we should avoid shutting down writes completely to
// healthy nodes. Any node with no hintsInProgress is considered healthy.
if (StorageMetrics.totalHintsInProgress.count() > maxHintsInProgress
&& (getHintsInProgressFor(destination).get() > 0 && shouldHint(destination)))
{
throw new OverloadedException("Too many in flight hints: " + StorageMetrics.totalHintsInProgress.count());
}
checkHintOverload(destination);
if (FailureDetector.instance.isAlive(destination))
{
if (destination.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS)
if (destination.equals(FBUtilities.getBroadcastAddress()))
{
insertLocal = true;
} else
@ -929,6 +952,22 @@ public class StorageProxy implements StorageProxyMBean
}
}
private static void checkHintOverload(InetAddress destination) throws OverloadedException
{
// avoid OOMing due to excess hints. we need to do this check even for "live" nodes, since we can
// still generate hints for those if it's overloaded or simply dead but not yet known-to-be-dead.
// The idea is that if we have over maxHintsInProgress hints in flight, this is probably due to
// a small number of nodes causing problems, so we should avoid shutting down writes completely to
// healthy nodes. Any node with no hintsInProgress is considered healthy.
if (StorageMetrics.totalHintsInProgress.count() > maxHintsInProgress
&& (getHintsInProgressFor(destination).get() > 0 && shouldHint(destination)))
{
throw new OverloadedException("Too many in flight hints: " + StorageMetrics.totalHintsInProgress.count() +
" destination: " + destination +
" destination hints: " + getHintsInProgressFor(destination).get());
}
}
private static AtomicInteger getHintsInProgressFor(InetAddress destination)
{
try
@ -1043,6 +1082,12 @@ public class StorageProxy implements StorageProxyMBean
responseHandler.response(null);
}
}
@Override
protected Verb verb()
{
return MessagingService.Verb.MUTATION;
}
});
}
@ -1730,8 +1775,7 @@ public class StorageProxy implements StorageProxyMBean
handler.assureSufficientLiveNodes();
resolver.setSources(filteredEndpoints);
if (filteredEndpoints.size() == 1
&& filteredEndpoints.get(0).equals(FBUtilities.getBroadcastAddress())
&& OPTIMIZE_LOCAL_REQUESTS)
&& filteredEndpoints.get(0).equals(FBUtilities.getBroadcastAddress()))
{
StageManager.getStage(Stage.READ).execute(new LocalRangeSliceRunnable(nodeCmd, handler), Tracing.instance.get());
}
@ -2161,7 +2205,7 @@ public class StorageProxy implements StorageProxyMBean
{
return !Gossiper.instance.getUnreachableTokenOwners().isEmpty();
}
public interface WritePerformer
{
public void apply(IMutation mutation,
@ -2214,9 +2258,11 @@ public class StorageProxy implements StorageProxyMBean
public final void run()
{
if (System.currentTimeMillis() > constructionTime + DatabaseDescriptor.getTimeout(MessagingService.Verb.MUTATION))
final MessagingService.Verb verb = verb();
if (System.currentTimeMillis() > constructionTime + DatabaseDescriptor.getTimeout(verb))
{
MessagingService.instance().incrementDroppedMessages(MessagingService.Verb.MUTATION);
if (MessagingService.DROPPABLE_VERBS.contains(verb()))
MessagingService.instance().incrementDroppedMessages(verb);
HintRunnable runnable = new HintRunnable(FBUtilities.getBroadcastAddress())
{
protected void runMayThrow() throws Exception
@ -2238,6 +2284,7 @@ public class StorageProxy implements StorageProxyMBean
}
}
abstract protected MessagingService.Verb verb();
abstract protected void runMayThrow() throws Exception;
}
@ -2332,11 +2379,11 @@ public class StorageProxy implements StorageProxyMBean
public long getReadRepairAttempted() {
return ReadRepairMetrics.attempted.count();
}
public long getReadRepairRepairedBlocking() {
return ReadRepairMetrics.repairedBlocking.count();
}
public long getReadRepairRepairedBackground() {
return ReadRepairMetrics.repairedBackground.count();
}