From 5e6539addbca00d82484a60d0ca574f0223bfe05 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 28 Apr 2011 13:24:22 +0000 Subject: [PATCH 1/2] fix incorrect use ofNBHM.size in ReadCallback patch by jbellis; reviewed by stuhood for CASSANDRA-2552 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1097448 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 2 ++ .../cassandra/service/AbstractRowResolver.java | 5 ----- .../cassandra/service/AsyncRepairCallback.java | 10 ++++++---- .../service/DatacenterReadCallback.java | 18 +++++++----------- .../cassandra/service/IResponseResolver.java | 3 --- .../service/RangeSliceResponseResolver.java | 5 ----- .../apache/cassandra/service/ReadCallback.java | 16 ++++++---------- .../cassandra/service/RepairCallback.java | 6 ++++-- 8 files changed, 25 insertions(+), 40 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index c1046b123d..8e79dcd720 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,6 +1,8 @@ 0.7.6 * force GC to reclaim disk space on flush, if necessary (CASSANDRA-2404) * move gossip heartbeat back to its own thread (CASSANDRA-2554) + * fix incorrect use of NBHM.size in ReadCallback that could cause + reads to time out even when responses were received (CASSAMDRA-2552) 0.7.5 diff --git a/src/java/org/apache/cassandra/service/AbstractRowResolver.java b/src/java/org/apache/cassandra/service/AbstractRowResolver.java index 23dfe23682..b13151c5a5 100644 --- a/src/java/org/apache/cassandra/service/AbstractRowResolver.java +++ b/src/java/org/apache/cassandra/service/AbstractRowResolver.java @@ -83,9 +83,4 @@ public abstract class AbstractRowResolver implements IResponseResolver { return replies.keySet(); } - - public int getMessageCount() - { - return replies.size(); - } } diff --git a/src/java/org/apache/cassandra/service/AsyncRepairCallback.java b/src/java/org/apache/cassandra/service/AsyncRepairCallback.java index 6c925877e4..72a4b0929b 100644 --- a/src/java/org/apache/cassandra/service/AsyncRepairCallback.java +++ b/src/java/org/apache/cassandra/service/AsyncRepairCallback.java @@ -22,6 +22,7 @@ package org.apache.cassandra.service; import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; @@ -32,18 +33,19 @@ import org.apache.cassandra.utils.WrappedRunnable; public class AsyncRepairCallback implements IAsyncCallback { private final RowRepairResolver repairResolver; - private final int count; + private final int blockfor; + protected final AtomicInteger received = new AtomicInteger(0); - public AsyncRepairCallback(RowRepairResolver repairResolver, int count) + public AsyncRepairCallback(RowRepairResolver repairResolver, int blockfor) { this.repairResolver = repairResolver; - this.count = count; + this.blockfor = blockfor; } public void response(Message message) { repairResolver.preprocess(message); - if (repairResolver.getMessageCount() == count) + if (received.incrementAndGet() == blockfor) { StageManager.getStage(Stage.READ_REPAIR).execute(new WrappedRunnable() { diff --git a/src/java/org/apache/cassandra/service/DatacenterReadCallback.java b/src/java/org/apache/cassandra/service/DatacenterReadCallback.java index 9cdfa19c64..53323d3692 100644 --- a/src/java/org/apache/cassandra/service/DatacenterReadCallback.java +++ b/src/java/org/apache/cassandra/service/DatacenterReadCallback.java @@ -23,7 +23,6 @@ package org.apache.cassandra.service; import java.net.InetAddress; import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ReadResponse; @@ -42,12 +41,10 @@ public class DatacenterReadCallback extends ReadCallback { private static final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch(); private static final String localdc = snitch.getDatacenter(FBUtilities.getLocalAddress()); - private AtomicInteger localResponses; - + public DatacenterReadCallback(IResponseResolver resolver, ConsistencyLevel consistencyLevel, IReadCommand command, List endpoints) { super(resolver, consistencyLevel, command, endpoints); - localResponses = new AtomicInteger(blockfor); } @Override @@ -56,12 +53,13 @@ public class DatacenterReadCallback extends ReadCallback resolver.preprocess(message); int n = localdc.equals(snitch.getDatacenter(message.getFrom())) - ? localResponses.decrementAndGet() - : localResponses.get(); + ? received.incrementAndGet() + : received.get(); - if (n == 0 && resolver.isDataPresent()) + if (n == blockfor && resolver.isDataPresent()) { condition.signal(); + maybeResolveForRepair(); } } @@ -70,13 +68,11 @@ public class DatacenterReadCallback extends ReadCallback { ((RowDigestResolver) resolver).injectPreProcessed(result); - int n = localResponses.decrementAndGet(); - if (n == 0 && resolver.isDataPresent()) + if (received.incrementAndGet() == blockfor && resolver.isDataPresent()) { condition.signal(); + maybeResolveForRepair(); } - - maybeResolveForRepair(); } @Override diff --git a/src/java/org/apache/cassandra/service/IResponseResolver.java b/src/java/org/apache/cassandra/service/IResponseResolver.java index 870ee16ab1..f4f972a304 100644 --- a/src/java/org/apache/cassandra/service/IResponseResolver.java +++ b/src/java/org/apache/cassandra/service/IResponseResolver.java @@ -43,7 +43,4 @@ public interface IResponseResolver { public void preprocess(Message message); public Iterable getMessages(); - - /** Potentially called by multiple response threads, so must be threadsafe. */ - public int getMessageCount(); } diff --git a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java index 28923895d6..09a4d17982 100644 --- a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java +++ b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java @@ -146,9 +146,4 @@ public class RangeSliceResponseResolver implements IResponseResolver implements IAsyncCallback private final long startTime; protected final int blockfor; private final IReadCommand command; + protected final AtomicInteger received = new AtomicInteger(0); /** the list of endpoints that StorageProxy should send requests to */ final List endpoints; @@ -117,7 +119,7 @@ public class ReadCallback implements IAsyncCallback StringBuilder sb = new StringBuilder(""); for (Message message : resolver.getMessages()) sb.append(message.getFrom()).append(", "); - throw new TimeoutException("Operation timed out - received only " + resolver.getMessageCount() + " responses from " + sb.toString() + " ."); + throw new TimeoutException("Operation timed out - received only " + received.get() + " responses from " + sb.toString() + " ."); } return blockfor == 1 ? resolver.getData() : resolver.resolve(); @@ -126,10 +128,7 @@ public class ReadCallback implements IAsyncCallback public void response(Message message) { resolver.preprocess(message); - assert resolver.getMessageCount() <= endpoints.size() : "Got " + resolver.getMessageCount() + " replies but requests were only sent to " + endpoints.size() + " endpoints"; - if (resolver.getMessageCount() < blockfor) - return; - if (resolver.isDataPresent()) + if (received.incrementAndGet() >= blockfor && resolver.isDataPresent()) { condition.signal(); maybeResolveForRepair(); @@ -139,10 +138,7 @@ public class ReadCallback implements IAsyncCallback public void response(ReadResponse result) { ((RowDigestResolver) resolver).injectPreProcessed(result); - assert resolver.getMessageCount() <= endpoints.size(); - if (resolver.getMessageCount() < blockfor) - return; - if (resolver.isDataPresent()) + if (received.incrementAndGet() >= blockfor && resolver.isDataPresent()) { condition.signal(); maybeResolveForRepair(); @@ -155,7 +151,7 @@ public class ReadCallback implements IAsyncCallback */ protected void maybeResolveForRepair() { - if (blockfor < endpoints.size() && resolver.getMessageCount() == endpoints.size()) + if (blockfor < endpoints.size() && received.get() == endpoints.size()) { assert resolver.isDataPresent(); StageManager.getStage(Stage.READ_REPAIR).execute(new AsyncRepairRunner()); diff --git a/src/java/org/apache/cassandra/service/RepairCallback.java b/src/java/org/apache/cassandra/service/RepairCallback.java index 2b946223a9..d79ea1dcd6 100644 --- a/src/java/org/apache/cassandra/service/RepairCallback.java +++ b/src/java/org/apache/cassandra/service/RepairCallback.java @@ -26,6 +26,7 @@ import java.net.InetAddress; import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.net.IAsyncCallback; @@ -38,6 +39,7 @@ public class RepairCallback implements IAsyncCallback private final List endpoints; private final SimpleCondition condition = new SimpleCondition(); private final long startTime; + protected final AtomicInteger received = new AtomicInteger(0); /** * The main difference between this and ReadCallback is, ReadCallback has a ConsistencyLevel @@ -66,13 +68,13 @@ public class RepairCallback implements IAsyncCallback throw new AssertionError(ex); } - return resolver.getMessageCount() > 1 ? resolver.resolve() : null; + return received.get() > 1 ? resolver.resolve() : null; } public void response(Message message) { resolver.preprocess(message); - if (resolver.getMessageCount() == endpoints.size()) + if (received.incrementAndGet() == endpoints.size()) condition.signal(); } From 0251a8f0f66055960c2dc5502b6496313bf074f3 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Thu, 28 Apr 2011 13:36:07 +0000 Subject: [PATCH 2/2] trigger read repair correctly forLOCAL_QUORUM reads patch by jbellis; reviewed by slebresne for CASSANDRA-2556 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1097455 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + .../service/DatacenterReadCallback.java | 30 +++++-------------- .../cassandra/service/ReadCallback.java | 27 +++++++++++++++-- 3 files changed, 34 insertions(+), 24 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 8e79dcd720..ac99d927f0 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -3,6 +3,7 @@ * move gossip heartbeat back to its own thread (CASSANDRA-2554) * fix incorrect use of NBHM.size in ReadCallback that could cause reads to time out even when responses were received (CASSAMDRA-2552) + * trigger read repair correctly for LOCAL_QUORUM reads (CASSANDRA-2556) 0.7.5 diff --git a/src/java/org/apache/cassandra/service/DatacenterReadCallback.java b/src/java/org/apache/cassandra/service/DatacenterReadCallback.java index 53323d3692..3e6a37ba99 100644 --- a/src/java/org/apache/cassandra/service/DatacenterReadCallback.java +++ b/src/java/org/apache/cassandra/service/DatacenterReadCallback.java @@ -48,31 +48,17 @@ public class DatacenterReadCallback extends ReadCallback } @Override - public void response(Message message) + protected boolean waitingFor(Message message) { - resolver.preprocess(message); - - int n = localdc.equals(snitch.getDatacenter(message.getFrom())) - ? received.incrementAndGet() - : received.get(); - - if (n == blockfor && resolver.isDataPresent()) - { - condition.signal(); - maybeResolveForRepair(); - } + return localdc.equals(snitch.getDatacenter(message.getFrom())); } - - @Override - public void response(ReadResponse result) - { - ((RowDigestResolver) resolver).injectPreProcessed(result); - if (received.incrementAndGet() == blockfor && resolver.isDataPresent()) - { - condition.signal(); - maybeResolveForRepair(); - } + @Override + protected boolean waitingFor(ReadResponse response) + { + // cheat and leverage our knowledge that a local read is the only way the ReadResponse + // version of this method gets called + return true; } @Override diff --git a/src/java/org/apache/cassandra/service/ReadCallback.java b/src/java/org/apache/cassandra/service/ReadCallback.java index c482084ab0..a3e929c01b 100644 --- a/src/java/org/apache/cassandra/service/ReadCallback.java +++ b/src/java/org/apache/cassandra/service/ReadCallback.java @@ -128,17 +128,40 @@ public class ReadCallback implements IAsyncCallback public void response(Message message) { resolver.preprocess(message); - if (received.incrementAndGet() >= blockfor && resolver.isDataPresent()) + int n = waitingFor(message) + ? received.incrementAndGet() + : received.get(); + if (n >= blockfor && resolver.isDataPresent()) { condition.signal(); maybeResolveForRepair(); } } + /** + * @return true if the message counts towards the blockfor threshold + * TODO turn the Message into a response so we don't need two versions of this method + */ + protected boolean waitingFor(Message message) + { + return true; + } + + /** + * @return true if the response counts towards the blockfor threshold + */ + protected boolean waitingFor(ReadResponse response) + { + return true; + } + public void response(ReadResponse result) { ((RowDigestResolver) resolver).injectPreProcessed(result); - if (received.incrementAndGet() >= blockfor && resolver.isDataPresent()) + int n = waitingFor(result) + ? received.incrementAndGet() + : received.get(); + if (n >= blockfor && resolver.isDataPresent()) { condition.signal(); maybeResolveForRepair();