From a5bd08f421c647c3a472d231b24ae0d0a7101b9e Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Tue, 17 Nov 2009 13:34:06 +0000 Subject: [PATCH] add WriteResponseHandler combining the important parts of QuorumResponseHandler and WriteResponseResolver. In particular, not thate we (correctly) never send a write response of false, letting the timeout take care of that should-never-happen case. optimize local writes in insertBlocking, and fix HH. patch by jbellis; reviewed by Jaakko Laine for CASSANDRA-558 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@881281 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 3 +- .../cassandra/db/HintedHandOffManager.java | 18 ++- .../locator/AbstractReplicationStrategy.java | 5 +- .../locator/DatacenterShardStategy.java | 9 +- .../cassandra/net/MessagingService.java | 15 ++- .../DatacenterQuorumResponseHandler.java | 6 +- .../DatacenterQuorumSyncResponseHandler.java | 6 +- .../service/ReadResponseResolver.java | 2 +- .../cassandra/service/StorageProxy.java | 106 ++++++++++-------- .../cassandra/service/StorageService.java | 4 +- .../service/WriteResponseHandler.java | 106 ++++++++++++++++++ .../service/WriteResponseResolver.java | 73 ------------ 12 files changed, 209 insertions(+), 144 deletions(-) create mode 100644 src/java/org/apache/cassandra/service/WriteResponseHandler.java delete mode 100644 src/java/org/apache/cassandra/service/WriteResponseResolver.java diff --git a/CHANGES.txt b/CHANGES.txt index 201863a8b0..916063030e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -41,7 +41,8 @@ interfaces (CASSANDRA-546) * stress.py benchmarking tool improvements (several tickets) * optimized replica placement code (CASSANDRA-525) - * faster log replay on restart (CASSANDRA-539, -540) + * faster log replay on restart (CASSANDRA-539, CASSANDRA-540) + * optimized local-node writes (CASSANDRA-558) 0.4.2 diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index 7cd46621b5..f3a6c665ba 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -101,7 +101,7 @@ public class HintedHandOffManager return instance_; } - private static boolean sendMessage(InetAddress endPoint, String tableName, String key) throws DigestMismatchException, TimeoutException, IOException, InvalidRequestException + private static boolean sendMessage(InetAddress endPoint, String tableName, String key) throws IOException { if (!FailureDetector.instance().isAlive(endPoint)) { @@ -112,10 +112,18 @@ public class HintedHandOffManager Row row = table.get(key); RowMutation rm = new RowMutation(tableName, row); Message message = rm.makeRowMutationMessage(); - QuorumResponseHandler quorumResponseHandler = new QuorumResponseHandler(1, new WriteResponseResolver()); - MessagingService.instance().sendRR(message, new InetAddress[] { endPoint }, quorumResponseHandler); + WriteResponseHandler responseHandler = new WriteResponseHandler(1); + MessagingService.instance().sendRR(message, new InetAddress[] { endPoint }, responseHandler); - return quorumResponseHandler.get(); + try + { + responseHandler.get(); + } + catch (TimeoutException e) + { + return false; + } + return true; } private static void deleteEndPoint(byte[] endpointAddress, String tableName, byte[] key, long timestamp) throws IOException @@ -205,7 +213,7 @@ public class HintedHandOffManager Collection endpoints = keyColumn.getSubColumns(); for (IColumn hintEndPoint : endpoints) { - if (Arrays.equals(hintEndPoint.name(), targetEPBytes) && sendMessage(endPoint, null, keyStr)) + if (Arrays.equals(hintEndPoint.name(), targetEPBytes) && sendMessage(endPoint, tableName, keyStr)) { if (endpoints.size() == 1) { diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 4cd6479a62..aa5f8fda4e 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -31,6 +31,7 @@ import org.apache.cassandra.dht.Token; import org.apache.cassandra.gms.FailureDetector; import org.apache.cassandra.service.IResponseResolver; import org.apache.cassandra.service.QuorumResponseHandler; +import org.apache.cassandra.service.WriteResponseHandler; import org.apache.cassandra.utils.FBUtilities; /** @@ -55,9 +56,9 @@ public abstract class AbstractReplicationStrategy public abstract ArrayList getNaturalEndpoints(Token token, TokenMetadata metadata); - public QuorumResponseHandler getResponseHandler(IResponseResolver responseResolver, int blockFor, int consistency_level) + public WriteResponseHandler getWriteResponseHandler(int blockFor, int consistency_level) { - return new QuorumResponseHandler(blockFor, responseResolver); + return new WriteResponseHandler(blockFor); } public ArrayList getNaturalEndpoints(Token token) diff --git a/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java b/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java index f07123b476..01aae88c60 100644 --- a/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java +++ b/src/java/org/apache/cassandra/locator/DatacenterShardStategy.java @@ -202,17 +202,16 @@ public class DatacenterShardStategy extends AbstractReplicationStrategy * return a DCQRH with a map of all the DC rep facor. */ @Override - public QuorumResponseHandler getResponseHandler(IResponseResolver responseResolver, int blockFor, int consistency_level) + public WriteResponseHandler getWriteResponseHandler(int blockFor, int consistency_level) { if (consistency_level == ConsistencyLevel.DCQUORUM) { - List endpoints = getLocalEndPoints(); - return new DatacenterQuorumResponseHandler(locQFactor, responseResolver); + return new DatacenterQuorumResponseHandler(locQFactor); } else if (consistency_level == ConsistencyLevel.DCQUORUMSYNC) { - return new DatacenterQuorumSyncResponseHandler(getQuorumRepFactor(), responseResolver); + return new DatacenterQuorumSyncResponseHandler(getQuorumRepFactor()); } - return super.getResponseHandler(responseResolver, blockFor, consistency_level); + return super.getWriteResponseHandler(blockFor, consistency_level); } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index b975979e75..cabb375511 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -322,15 +322,20 @@ public class MessagingService */ public String sendRR(Message message, InetAddress[] to, IAsyncCallback cb) { - String messageId = message.getMessageId(); - callbackMap_.put(messageId, cb); + String messageId = message.getMessageId(); + addCallback(cb, messageId); for ( int i = 0; i < to.length; ++i ) { sendOneWay(message, to[i]); } return messageId; } - + + public void addCallback(IAsyncCallback cb, String messageId) + { + callbackMap_.put(messageId, cb); + } + /** * Send a message to a given endpoint. This method specifies a callback * which is invoked with the actual response. @@ -344,7 +349,7 @@ public class MessagingService public String sendRR(Message message, InetAddress to, IAsyncCallback cb) { String messageId = message.getMessageId(); - callbackMap_.put(messageId, cb); + addCallback(cb, messageId); sendOneWay(message, to); return messageId; } @@ -369,7 +374,7 @@ public class MessagingService throw new IllegalArgumentException("Number of messages and the number of endpoints need to be same."); } String groupId = GuidGenerator.guid(); - callbackMap_.put(groupId, cb); + addCallback(cb, groupId); for ( int i = 0; i < messages.length; ++i ) { messages[i].setMessageId(groupId); diff --git a/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java index 7ab4e62643..7ba15415ad 100644 --- a/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java @@ -16,16 +16,16 @@ import org.apache.cassandra.utils.FBUtilities; * provided in the input map. it will block till we recive response from (DC, n) * nodes. */ -public class DatacenterQuorumResponseHandler extends QuorumResponseHandler +public class DatacenterQuorumResponseHandler extends WriteResponseHandler { private int blockFor; private IEndPointSnitch endpointsnitch; private InetAddress localEndpoint; - public DatacenterQuorumResponseHandler(int blockFor, IResponseResolver responseResolver) + public DatacenterQuorumResponseHandler(int blockFor) { // Response is been managed by the map so the waitlist size really doesnt matter. - super(blockFor, responseResolver); + super(blockFor); this.blockFor = blockFor; endpointsnitch = DatabaseDescriptor.getEndPointSnitch(); localEndpoint = FBUtilities.getLocalAddress(); diff --git a/src/java/org/apache/cassandra/service/DatacenterQuorumSyncResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterQuorumSyncResponseHandler.java index ac9c61cffe..20c07de068 100644 --- a/src/java/org/apache/cassandra/service/DatacenterQuorumSyncResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterQuorumSyncResponseHandler.java @@ -15,15 +15,15 @@ import org.apache.cassandra.net.Message; * provided in the input map. it will block till we recive response from * n nodes in each of our data centers. */ -public class DatacenterQuorumSyncResponseHandler extends QuorumResponseHandler +public class DatacenterQuorumSyncResponseHandler extends WriteResponseHandler { private final Map dcResponses = new HashMap(); private final Map responseCounts; - public DatacenterQuorumSyncResponseHandler(Map responseCounts, IResponseResolver responseResolver) + public DatacenterQuorumSyncResponseHandler(Map responseCounts) { // Response is been managed by the map so make it 1 for the superclass. - super(1, responseResolver); + super(1); this.responseCounts = responseCounts; } diff --git a/src/java/org/apache/cassandra/service/ReadResponseResolver.java b/src/java/org/apache/cassandra/service/ReadResponseResolver.java index 246475173f..48df44051c 100644 --- a/src/java/org/apache/cassandra/service/ReadResponseResolver.java +++ b/src/java/org/apache/cassandra/service/ReadResponseResolver.java @@ -47,7 +47,7 @@ import org.apache.log4j.Logger; */ public class ReadResponseResolver implements IResponseResolver { - private static Logger logger_ = Logger.getLogger(WriteResponseResolver.class); + private static Logger logger_ = Logger.getLogger(ReadResponseResolver.class); /* * This method for resolving read data should look at the timestamps of each diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index bd2be67105..9e28465325 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -19,7 +19,6 @@ package org.apache.cassandra.service; import java.io.IOError; import java.io.IOException; -import java.io.IOError; import java.util.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -149,53 +148,85 @@ public class StorageProxy implements StorageProxyMBean } } - public static void insertBlocking(RowMutation rm, int consistency_level) throws UnavailableException + public static void insertBlocking(final RowMutation rm, int consistency_level) throws UnavailableException { long startTime = System.currentTimeMillis(); - Message message; - try - { - message = rm.makeRowMutationMessage(); - } - catch (IOException e) - { - throw new RuntimeException(e); - } try { List naturalEndpoints = StorageService.instance().getNaturalEndpoints(rm.key()); Map endpointMap = StorageService.instance().getHintedEndpointMap(rm.key(), naturalEndpoints); int blockFor = determineBlockFor(naturalEndpoints.size(), endpointMap.size(), consistency_level); - List primaryNodes = getUnhintedNodes(endpointMap); - if (primaryNodes.size() < blockFor) // guarantee blockFor = W live nodes. + + // avoid starting a write we know can't achieve the required consistency + int liveNodes = 0; + for (Map.Entry entry : endpointMap.entrySet()) + { + if (entry.getKey().equals(entry.getValue())) + { + liveNodes++; + } + } + if (liveNodes < blockFor) { throw new UnavailableException(); } - QuorumResponseHandler quorumResponseHandler = StorageService.instance().getResponseHandler(new WriteResponseResolver(), blockFor, consistency_level); - if (logger.isDebugEnabled()) - logger.debug("insertBlocking writing key " + rm.key() + " to " + message.getMessageId() + "@[" + StringUtils.join(endpointMap.values(), ", ") + "]"); - // Get all the targets and stick them in an array - MessagingService.instance().sendRR(message, primaryNodes.toArray(new InetAddress[primaryNodes.size()]), quorumResponseHandler); - try + // 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); + Message unhintedMessage = null; + for (Map.Entry entry : endpointMap.entrySet()) { - if (!quorumResponseHandler.get()) - throw new UnavailableException(); - } - catch (DigestMismatchException e) - { - throw new AssertionError(e); - } - if (primaryNodes.size() < endpointMap.size()) // Do we need to bother with Hinted Handoff? - { - for (Map.Entry e : endpointMap.entrySet()) + InetAddress target = entry.getKey(); + InetAddress hintedTarget = entry.getValue(); + + if (target.equals(hintedTarget)) { - if (!e.getKey().equals(e.getValue())) // Hinted Handoff to target + if (target.equals(FBUtilities.getLocalAddress())) { - MessagingService.instance().sendOneWay(message, e.getValue()); + if (logger.isDebugEnabled()) + logger.debug("insert writing local key " + rm.key()); + Runnable runnable = new Runnable() + { + public void run() + { + try + { + rm.apply(); + responseHandler.localResponse(); + } + catch (IOException e) + { + throw new IOError(e); + } + } + }; + StageManager.getStage(StageManager.mutationStage_).execute(runnable); + } + else + { + if (unhintedMessage == null) + { + unhintedMessage = rm.makeRowMutationMessage(); + MessagingService.instance().addCallback(responseHandler, unhintedMessage.getMessageId()); + } + if (logger.isDebugEnabled()) + logger.debug("insert writing key " + rm.key() + " to " + unhintedMessage.getMessageId() + "@" + target); + MessagingService.instance().sendOneWay(unhintedMessage, target); } } + else + { + // (hints aren't part of the callback since they don't count towards consistency until they are on the final destination node) + Message hintedMessage = rm.makeRowMutationMessage(); + hintedMessage.addHeader(RowMutation.HINT, target.getAddress()); + if (logger.isDebugEnabled()) + logger.debug("insert writing key " + rm.key() + " to " + hintedMessage.getMessageId() + "@" + hintedTarget + " for " + target); + MessagingService.instance().sendOneWay(hintedMessage, hintedTarget); + } } + + // wait for writes. throws timeoutexception if necessary + responseHandler.get(); } catch (TimeoutException e) { @@ -211,19 +242,6 @@ public class StorageProxy implements StorageProxyMBean } } - private static List getUnhintedNodes(Map endpointMap) - { - List liveEndPoints = new ArrayList(endpointMap.size()); - for (Map.Entry e : endpointMap.entrySet()) - { - if (e.getKey().equals(e.getValue())) - { - liveEndPoints.add(e.getKey()); - } - } - return liveEndPoints; - } - private static int determineBlockFor(int naturalTargets, int hintedTargets, int consistency_level) { assert naturalTargets >= 1; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index d4beeb8c9a..e6b7eed0c6 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -1037,9 +1037,9 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto unbootstrap(finishMoving); } - public QuorumResponseHandler getResponseHandler(IResponseResolver responseResolver, int blockFor, int consistency_level) + public WriteResponseHandler getWriteResponseHandler(int blockFor, int consistency_level) { - return replicationStrategy_.getResponseHandler(responseResolver, blockFor, consistency_level); + return replicationStrategy_.getWriteResponseHandler(blockFor, consistency_level); } public AbstractReplicationStrategy getReplicationStrategy() diff --git a/src/java/org/apache/cassandra/service/WriteResponseHandler.java b/src/java/org/apache/cassandra/service/WriteResponseHandler.java new file mode 100644 index 0000000000..76c543c359 --- /dev/null +++ b/src/java/org/apache/cassandra/service/WriteResponseHandler.java @@ -0,0 +1,106 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.service; + +import java.util.List; +import java.util.ArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.io.IOException; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.net.IAsyncCallback; +import org.apache.cassandra.net.Message; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.utils.SimpleCondition; + +import org.apache.log4j.Logger; + +public class WriteResponseHandler implements IAsyncCallback +{ + protected static final Logger logger = Logger.getLogger( WriteResponseHandler.class ); + protected final SimpleCondition condition = new SimpleCondition(); + private final int responseCount; + protected final List responses; + protected int localResponses; + private final long startTime; + + public WriteResponseHandler(int responseCount) + { + assert 1 <= responseCount && responseCount <= DatabaseDescriptor.getReplicationFactor() + : "invalid response count " + responseCount; + + this.responseCount = responseCount; + responses = new ArrayList(responseCount); + startTime = System.currentTimeMillis(); + } + + public void get() throws TimeoutException + { + try + { + long timeout = System.currentTimeMillis() - startTime + DatabaseDescriptor.getRpcTimeout(); + boolean success; + try + { + success = condition.await(timeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException ex) + { + throw new AssertionError(ex); + } + + if (!success) + { + throw new TimeoutException("Operation timed out - received only " + responses.size() + localResponses + " responses"); + } + } + finally + { + for (Message response : responses) + { + MessagingService.removeRegisteredCallback(response.getMessageId()); + } + } + } + + public synchronized void response(Message message) + { + if (condition.isSignaled()) + return; + responses.add(message); + maybeSignal(); + } + + public synchronized void localResponse() + { + if (condition.isSignaled()) + return; + localResponses++; + maybeSignal(); + } + + private void maybeSignal() + { + if (responses.size() + localResponses >= responseCount) + { + condition.signal(); + } + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/WriteResponseResolver.java b/src/java/org/apache/cassandra/service/WriteResponseResolver.java deleted file mode 100644 index cd96644afe..0000000000 --- a/src/java/org/apache/cassandra/service/WriteResponseResolver.java +++ /dev/null @@ -1,73 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.cassandra.service; - -import java.util.List; -import java.io.DataInputStream; -import java.io.ByteArrayInputStream; -import java.io.IOException; - -import org.apache.cassandra.db.WriteResponse; -import org.apache.cassandra.net.Message; -import org.apache.log4j.Logger; - -public class WriteResponseResolver implements IResponseResolver { - - private static Logger logger_ = Logger.getLogger(WriteResponseResolver.class); - - /* - * The resolve function for the Write looks at all the responses if all the - * responses returned are false then we have a problem since that means the - * key was not written to any of the servers we want to notify the client of - * this so in that case we should return a false saying that the write - * failed. - * - */ - public Boolean resolve(List responses) throws DigestMismatchException - { - // TODO: We need to log error responses here for example - // if a write fails for a key log that the key could not be replicated - boolean returnValue = false; - for (Message response : responses) { - WriteResponse writeResponseMessage = null; - try - { - writeResponseMessage = WriteResponse.serializer().deserialize(new DataInputStream(new ByteArrayInputStream(response.getMessageBody()))); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - boolean result = writeResponseMessage.isSuccess(); - if (!result) { - if (logger_.isDebugEnabled()) - logger_.debug("Write at " + response.getFrom() - + " may have failed for the key " + writeResponseMessage.key()); - } - returnValue |= result; - } - return returnValue; - } - - public boolean isDataPresent(List responses) - { - return true; - } - -}