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

add PBSPredictor consistency modeler 

patch by Peter Bailis and Shivaram Venkataraman; reviewed by jbellis for CASSANDRA-4261
This commit is contained in:
Jonathan Ellis 2012-09-27 15:28:25 -05:00
parent 6ca75ef9ba
commit 0b94b191d8
11 changed files with 1023 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
1.2-beta2
* add PBSPredictor consistency modeler (CASSANDRA-4261)
* remove vestiges of Thrift unframed mode (CASSANDRA-4729)
* optimize single-row PK lookups (CASSANDRA-4710)
* adjust blockFor calculation to account for pending ranges due to node

5
build.xml
View File

@ -1134,6 +1134,11 @@
</testmacro>
</target>
<target name="pbs-test" depends="build-test" description="Tests PBS predictor">
<testmacro suitename="unit" inputdir="${test.unit.src}"
timeout="15000" filter="**/PBSPredictorTest.java"/>
</target>
<target name="long-test" depends="build-test" description="Execute functional tests">
<testmacro suitename="long" inputdir="${test.long.src}"
timeout="${test.long.timeout}">

34
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -36,18 +36,17 @@ import javax.management.ObjectName;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.db.*;
import org.apache.cassandra.dht.BootStrapper;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.gms.GossipDigestAck;
import org.apache.cassandra.gms.GossipDigestAck2;
import org.apache.cassandra.gms.GossipDigestSyn;
@ -58,14 +57,12 @@ import org.apache.cassandra.metrics.ConnectionMetrics;
import org.apache.cassandra.metrics.DroppedMessageMetrics;
import org.apache.cassandra.net.sink.SinkManager;
import org.apache.cassandra.security.SSLFactory;
import org.apache.cassandra.service.AntiEntropyService;
import org.apache.cassandra.service.MigrationManager;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.*;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.streaming.compress.CompressedFileStreamTask;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.*;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
public final class MessagingService implements MessagingServiceMBean
{
@ -555,6 +552,16 @@ public final class MessagingService implements MessagingServiceMBean
public String sendRR(MessageOut message, InetAddress to, IMessageCallback cb, long timeout)
{
String id = addCallback(cb, message, to, timeout);
if (cb instanceof AbstractWriteResponseHandler)
{
PBSPredictor.instance().startWriteOperation(id);
}
else if (cb instanceof ReadCallback)
{
PBSPredictor.instance().startReadOperation(id);
}
sendOneWay(message, id, to);
return id;
}
@ -695,6 +702,21 @@ public final class MessagingService implements MessagingServiceMBean
Runnable runnable = new MessageDeliveryTask(message, id, timestamp);
ExecutorService stage = StageManager.getStage(message.getMessageType());
assert stage != null : "No stage for message type " + message.verb;
if (message.verb == Verb.REQUEST_RESPONSE && PBSPredictor.instance().isLoggingEnabled())
{
IMessageCallback cb = MessagingService.instance().getRegisteredCallback(id).callback;
if (cb instanceof AbstractWriteResponseHandler)
{
PBSPredictor.instance().logWriteResponse(id, timestamp);
}
else if (cb instanceof ReadCallback)
{
PBSPredictor.instance().logReadResponse(id, timestamp);
}
}
stage.execute(runnable);
}

127
src/java/org/apache/cassandra/service/PBSPredictionResult.java Normal file
View File

@ -0,0 +1,127 @@
/**
* 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.io.Serializable;
public class PBSPredictionResult implements Serializable
{
private int n;
private int r;
private int w;
private float timeSinceWrite;
private int numberVersionsStale;
private float consistencyProbability;
private float averageReadLatency;
private float averageWriteLatency;
private long percentileReadLatencyValue;
private float percentileReadLatencyPercentile;
private long percentileWriteLatencyValue;
private float percentileWriteLatencyPercentile;
public PBSPredictionResult(int n,
int r,
int w,
float timeSinceWrite,
int numberVersionsStale,
float consistencyProbability,
float averageReadLatency,
float averageWriteLatency,
long percentileReadLatencyValue,
float percentileReadLatencyPercentile,
long percentileWriteLatencyValue,
float percentileWriteLatencyPercentile) {
this.n = n;
this.r = r;
this.w = w;
this.timeSinceWrite = timeSinceWrite;
this.numberVersionsStale = numberVersionsStale;
this.consistencyProbability = consistencyProbability;
this.averageReadLatency = averageReadLatency;
this.averageWriteLatency = averageWriteLatency;
this.percentileReadLatencyValue = percentileReadLatencyValue;
this.percentileReadLatencyPercentile = percentileReadLatencyPercentile;
this.percentileWriteLatencyValue = percentileWriteLatencyValue;
this.percentileWriteLatencyPercentile = percentileWriteLatencyPercentile;
}
public int getN()
{
return n;
}
public int getR()
{
return r;
}
public int getW()
{
return w;
}
public float getTimeSinceWrite()
{
return timeSinceWrite;
}
public int getNumberVersionsStale()
{
return numberVersionsStale;
}
public float getConsistencyProbability()
{
return consistencyProbability;
}
public float getAverageReadLatency()
{
return averageReadLatency;
}
public float getAverageWriteLatency()
{
return averageWriteLatency;
}
public long getPercentileReadLatencyValue()
{
return percentileReadLatencyValue;
}
public float getPercentileReadLatencyPercentile()
{
return percentileReadLatencyPercentile;
}
public long getPercentileWriteLatencyValue()
{
return percentileWriteLatencyValue;
}
public float getPercentileWriteLatencyPercentile()
{
return percentileWriteLatencyPercentile;
}
}

636
src/java/org/apache/cassandra/service/PBSPredictor.java Normal file
View File

@ -0,0 +1,636 @@
/**
* 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 org.apache.cassandra.net.MessageIn;
import java.lang.management.ManagementFactory;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.primitives.Longs;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Performs latency and consistency predictions as described in
* <a href="http://arxiv.org/pdf/1204.6082.pdf">
* "Probabilistically Bounded Staleness for Practical Partial Quorums"</a>
* by Bailis et al. in VLDB 2012. The predictions are of the form:
* <p/>
* <i>With ReplicationFactor <tt>N</tt>, read consistency level of
* <tt>R</tt>, and write consistency level <tt>W</tt>, after
* <tt>t</tt> seconds, <tt>p</tt>% of reads will return a version
* within <tt>k</tt> versions of the last written; this should result
* in a latency of <tt>L</tt> ms.</i>
* <p/>
* <p/>
* These predictions should be used as a rough guideline for system
* operators. This interface is exposed through nodetool.
* <p/>
* <p/>
* The class accomplishes this by measuring latencies for reads and
* writes, then using Monte Carlo simulation to predict behavior under
* a given N,R, and W based on those latencies.
* <p/>
* <p/>
* We capture four distributions:
* <p/>
* <ul>
* <li>
* <tt>W</tt>: time from when the coordinator sends a mutation to the time
* that a replica begins to serve the new value(s)
* </li>
* <p/>
* <li>
* <tt>A</tt>: time from when a replica accepting a mutation sends an
* acknowledgment to the time the coordinator hears of it
* </li>
* <p/>
* <li>
* <tt>R</tt>: time from when the coordinator sends a read request to the time
* that the replica performs the read
* </li>
* <p/>
* <li>
* <tt>S</tt>: time from when the replica sends a read response to the time
* when the coordinator receives it
* </li>
* </ul>
* <p/>
* <tt>A</tt> and <tt>S</tt> are mostly network-bound, while W and R
* depend on both the network and local processing time.
* <p/>
* <p/>
* <b>Caveats:</b>
* Prediction is only as good as the latencies collected. Accurate
* prediction requires synchronizing clocks between replicas. We
* collect a running sample of latencies, but, if latencies change
* dramatically, predictions will be off.
* <p/>
* <p/>
* The predictions are conservative, or worst-case, meaning we may
* predict more staleness than in practice in the following ways:
* <ul>
* <li>
* We do not account for read repair.
* </li>
* <li>
* We do not account for Merkle tree exchange.
* </li>
* <li>
* Multi-version staleness is particularly conservative.
* </li>
* <li>
* We simulate non-local reads and writes. We assume that the
* coordinating Cassandra node is not itself a replica for a given key.
* </li>
* </ul>
* <p/>
* <p/>
* The predictions are optimistic in the following ways:
* <ul>
* <li>
* We do not predict the impact of node failure.
* </li>
* <li>
* We do not model hinted handoff.
* </li>
* </ul>
*
* @see org.apache.cassandra.thrift.ConsistencyLevel
* @see org.apache.cassandra.locator.AbstractReplicationStrategy
*/
public class PBSPredictor implements PBSPredictorMBean
{
private static final Logger logger = LoggerFactory.getLogger(PBSPredictor.class);
public static final String MBEAN_NAME = "org.apache.cassandra.service:type=PBSPredictor";
private static final boolean DEFAULT_DO_LOG_LATENCIES = false;
private static final int DEFAULT_MAX_LOGGED_LATENCIES = 10000;
private static final int DEFAULT_NUMBER_TRIALS_PREDICTION = 10000;
/*
* We record a fixed size set of WARS latencies for read and
* mutation operations. We store the order in which each
* operation arrived, and use an LRU policy to evict old
* messages.
*
* This information is stored as a mapping from messageIDs to
* latencies.
*/
/*
* Helper class which minimizes the number of HashMaps we maintain.
* For a given messageId, this class maintains the startTime of the message,
* and a queue for send times and reply times.
*
* sendLats corresponds to W and R, while replyLats is used for A and S.
*/
private class MessageLatencyCollection
{
MessageLatencyCollection(Long startTime)
{
this.startTime = startTime;
this.sendLats = new ConcurrentLinkedQueue<Long>();
this.replyLats = new ConcurrentLinkedQueue<Long>();
}
void addSendLat(Long sendLat)
{
sendLats.add(sendLat);
}
void addReplyLat(Long replyLat)
{
replyLats.add(replyLat);
}
Collection<Long> getSendLats()
{
return sendLats;
}
Collection<Long> getReplyLats()
{
return replyLats;
}
Long getStartTime()
{
return startTime;
}
Long startTime;
Collection<Long> sendLats;
Collection<Long> replyLats;
}
// used for LRU replacement
private final Queue<String> writeMessageIds = new LinkedBlockingQueue<String>();
private final Queue<String> readMessageIds = new LinkedBlockingQueue<String>();
private final Map<String, MessageLatencyCollection> messageIdToWriteLats = new ConcurrentHashMap<String, MessageLatencyCollection>();
private final Map<String, MessageLatencyCollection> messageIdToReadLats = new ConcurrentHashMap<String, MessageLatencyCollection>();
private Random random;
private boolean initialized = false;
private boolean logLatencies = DEFAULT_DO_LOG_LATENCIES;
private int maxLoggedLatencies = DEFAULT_MAX_LOGGED_LATENCIES;
private int numberTrialsPrediction = DEFAULT_NUMBER_TRIALS_PREDICTION;
private static final PBSPredictor instance = new PBSPredictor();
public static PBSPredictor instance()
{
return instance;
}
private PBSPredictor()
{
init();
}
public void enableConsistencyPredictionLogging()
{
logLatencies = true;
}
public void disableConsistencyPredictionLogging()
{
logLatencies = false;
}
public boolean isLoggingEnabled()
{
return logLatencies;
}
public void setMaxLoggedLatenciesForConsistencyPrediction(int maxLogged)
{
maxLoggedLatencies = maxLogged;
}
public void setNumberTrialsForConsistencyPrediction(int numTrials)
{
numberTrialsPrediction = numTrials;
}
public void init()
{
if (!initialized)
{
random = new Random();
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(this, new ObjectName(PBSPredictor.MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
initialized = true;
}
}
// used for random sampling from the latencies
private long getRandomElement(List<Long> list)
{
if (list.size() == 0)
throw new RuntimeException("Not enough data for prediction");
return list.get(random.nextInt(list.size()));
}
// used for calculating the average latency of a read or write operation
// given a set of simulated latencies
private float listAverage(List<Long> list)
{
long accum = 0;
for (long value : list)
accum += value;
return (float) accum / list.size();
}
// calculate the percentile entry of a list
private long getPercentile(List<Long> list, float percentile)
{
Collections.sort(list);
return list.get((int) (list.size() * percentile));
}
/*
* For our trials, sample the latency for the (replicaNumber)th
* reply for one of WARS
* if replicaNumber > the number of replicas we have data for
* (say we have data for ReplicationFactor 2 but ask for N=3)
* then we randomly sample from all response times
*/
private long getRandomLatencySample(Map<Integer, List<Long>> samples, int replicaNumber)
{
if (samples.containsKey(replicaNumber))
{
return getRandomElement(samples.get(replicaNumber));
}
return getRandomElement(samples.get(samples.keySet().toArray()[random.nextInt(samples.keySet().size())]));
}
/*
* To perform the prediction, we randomly sample from the
* collected WARS latencies, simulating writes followed by reads
* exactly t milliseconds afterwards. We count the number of
* reordered reads and writes to calculate the probability of
* staleness along with recording operation latencies.
*/
public PBSPredictionResult doPrediction(int n,
int r,
int w,
float timeSinceWrite,
int numberVersionsStale,
float percentileLatency) throws Exception
{
if (r > n)
throw new IllegalArgumentException("r must be less than n");
if (r < 0)
throw new IllegalArgumentException("r must be positive");
if (w > n)
throw new IllegalArgumentException("w must be less than n");
if (w < 0)
throw new IllegalArgumentException("w must be positive");
if (percentileLatency < 0 || percentileLatency > 1)
throw new IllegalArgumentException("percentileLatency must be between 0 and 1 inclusive");
if (numberVersionsStale < 0)
throw new IllegalArgumentException("numberVersionsStale must be positive");
if (!logLatencies)
throw new InvalidRequestException("Latency logging is not enabled");
// get a mapping of {replica number : latency} for each of WARS
Map<Integer, List<Long>> wLatencies = getOrderedWLatencies();
Map<Integer, List<Long>> aLatencies = getOrderedALatencies();
Map<Integer, List<Long>> rLatencies = getOrderedRLatencies();
Map<Integer, List<Long>> sLatencies = getOrderedSLatencies();
if (wLatencies.isEmpty() || aLatencies.isEmpty())
throw new InvalidRequestException("No write latencies have been recorded so far. Run some (non-local) inserts.");
if (rLatencies.isEmpty() || sLatencies.isEmpty())
throw new InvalidRequestException("No read latencies have been recorded so far. Run some (non-local) reads.");
// storage for simulated read and write latencies
ArrayList<Long> readLatencies = new ArrayList<Long>();
ArrayList<Long> writeLatencies = new ArrayList<Long>();
long consistentReads = 0;
// storage for latencies for each replica for a given Monte Carlo trial
// arr[i] will hold the ith replica's latency for one of WARS
ArrayList<Long> trialWLatencies = new ArrayList<Long>();
ArrayList<Long> trialRLatencies = new ArrayList<Long>();
ArrayList<Long> replicaWriteLatencies = new ArrayList<Long>();
ArrayList<Long> replicaReadLatencies = new ArrayList<Long>();
//run repeated trials and observe staleness
for (int i = 0; i < numberTrialsPrediction; ++i)
{
//simulate sending a write to N replicas then sending a
//read to N replicas and record the latencies by randomly
//sampling from gathered latencies
for (int replicaNo = 0; replicaNo < n; ++replicaNo)
{
long trialWLatency = getRandomLatencySample(wLatencies, replicaNo);
long trialALatency = getRandomLatencySample(aLatencies, replicaNo);
trialWLatencies.add(trialWLatency);
replicaWriteLatencies.add(trialWLatency + trialALatency);
}
// reads are only sent to R replicas - so pick R random read and
// response latencies
for (int replicaNo = 0; replicaNo < r; ++replicaNo)
{
long trialRLatency = getRandomLatencySample(rLatencies, replicaNo);
long trialSLatency = getRandomLatencySample(sLatencies, replicaNo);
trialRLatencies.add(trialRLatency);
replicaReadLatencies.add(trialRLatency + trialSLatency);
}
// the write latency for this trial is the time it takes
// for the wth replica to respond (W+A)
Collections.sort(replicaWriteLatencies);
long writeLatency = replicaWriteLatencies.get(w - 1);
writeLatencies.add(writeLatency);
ArrayList<Long> sortedReplicaReadLatencies = new ArrayList<Long>(replicaReadLatencies);
Collections.sort(sortedReplicaReadLatencies);
// the read latency for this trial is the time it takes
// for the rth replica to respond (R+S)
readLatencies.add(sortedReplicaReadLatencies.get(r - 1));
// were all of the read responses reordered?
// for each of the first r messages (the ones the
// coordinator will pick from):
//--if the read message came in after this replica saw the
// write, it will be consistent
//--each read request is sent at time
// writeLatency+timeSinceWrite
for (int responseNumber = 0; responseNumber < r; ++responseNumber)
{
int replicaNumber = replicaReadLatencies.indexOf(sortedReplicaReadLatencies.get(responseNumber));
if (writeLatency + timeSinceWrite + trialRLatencies.get(replicaNumber) >=
trialWLatencies.get(replicaNumber))
{
consistentReads++;
break;
}
// tombstone this replica in the case that we have
// duplicate read latencies
replicaReadLatencies.set(replicaNumber, -1L);
}
// clear storage for the next trial
trialWLatencies.clear();
trialRLatencies.clear();
replicaReadLatencies.clear();
replicaWriteLatencies.clear();
}
float oneVersionConsistencyProbability = (float) consistentReads / numberTrialsPrediction;
// to calculate multi-version staleness, we exponentiate the staleness probability by the number of versions
float consistencyProbability = (float) (1 - Math.pow((double) (1 - oneVersionConsistencyProbability),
numberVersionsStale));
float averageWriteLatency = listAverage(writeLatencies);
float averageReadLatency = listAverage(readLatencies);
long percentileWriteLatency = getPercentile(writeLatencies, percentileLatency);
long percentileReadLatency = getPercentile(readLatencies, percentileLatency);
return new PBSPredictionResult(n,
r,
w,
timeSinceWrite,
numberVersionsStale,
consistencyProbability,
averageReadLatency,
averageWriteLatency,
percentileReadLatency,
percentileLatency,
percentileWriteLatency,
percentileLatency);
}
public void startWriteOperation(String id)
{
if (!logLatencies)
return;
startWriteOperation(id, System.currentTimeMillis());
}
public void startWriteOperation(String id, long startTime)
{
if (!logLatencies)
return;
assert (!messageIdToWriteLats.containsKey(id));
writeMessageIds.add(id);
// LRU replacement of latencies
// the maximum number of entries is sloppy, but that's acceptable for our purposes
if (writeMessageIds.size() > maxLoggedLatencies)
{
String toEvict = writeMessageIds.remove();
messageIdToWriteLats.remove(toEvict);
}
messageIdToWriteLats.put(id, new MessageLatencyCollection(startTime));
}
public void startReadOperation(String id)
{
if (!logLatencies)
return;
startReadOperation(id, System.currentTimeMillis());
}
public void startReadOperation(String id, long startTime)
{
if (!logLatencies)
return;
assert (!messageIdToReadLats.containsKey(id));
readMessageIds.add(id);
// LRU replacement of latencies
// the maximum number of entries is sloppy, but that's acceptable for our purposes
if (readMessageIds.size() > maxLoggedLatencies)
{
String toEvict = readMessageIds.remove();
messageIdToReadLats.remove(toEvict);
}
messageIdToReadLats.put(id, new MessageLatencyCollection(startTime));
}
public void logWriteResponse(String id, long constructionTime)
{
if (!logLatencies)
return;
logWriteResponse(id, constructionTime, System.currentTimeMillis());
}
public void logWriteResponse(String id, long responseCreationTime, long receivedTime)
{
if (!logLatencies)
return;
MessageLatencyCollection writeLatsCollection = messageIdToWriteLats.get(id);
if (writeLatsCollection == null)
{
return;
}
Long startTime = writeLatsCollection.getStartTime();
writeLatsCollection.addSendLat(Math.max(0, responseCreationTime - startTime));
writeLatsCollection.addReplyLat(Math.max(0, receivedTime - responseCreationTime));
}
public void logReadResponse(String id, long constructionTime)
{
if (!logLatencies)
return;
logReadResponse(id, constructionTime, System.currentTimeMillis());
}
public void logReadResponse(String id, long responseCreationTime, long receivedTime)
{
if (!logLatencies)
return;
MessageLatencyCollection readLatsCollection = messageIdToReadLats.get(id);
if (readLatsCollection == null)
{
return;
}
Long startTime = readLatsCollection.getStartTime();
readLatsCollection.addSendLat(Math.max(0, responseCreationTime - startTime));
readLatsCollection.addReplyLat(Math.max(0, receivedTime - responseCreationTime));
}
Map<Integer, List<Long>> getOrderedWLatencies()
{
Collection<Collection<Long>> allWLatencies = new ArrayList<Collection<Long>>();
for (MessageLatencyCollection wlc : messageIdToWriteLats.values())
{
allWLatencies.add(wlc.getSendLats());
}
return getOrderedLatencies(allWLatencies);
}
Map<Integer, List<Long>> getOrderedALatencies()
{
Collection<Collection<Long>> allALatencies = new ArrayList<Collection<Long>>();
for (MessageLatencyCollection wlc : messageIdToWriteLats.values())
allALatencies.add(wlc.getReplyLats());
return getOrderedLatencies(allALatencies);
}
Map<Integer, List<Long>> getOrderedRLatencies()
{
Collection<Collection<Long>> allRLatencies = new ArrayList<Collection<Long>>();
for (MessageLatencyCollection rlc : messageIdToReadLats.values())
{
allRLatencies.add(rlc.getSendLats());
}
return getOrderedLatencies(allRLatencies);
}
Map<Integer, List<Long>> getOrderedSLatencies()
{
Collection<Collection<Long>> allSLatencies = new ArrayList<Collection<Long>>();
for (MessageLatencyCollection rlc : messageIdToReadLats.values())
allSLatencies.add(rlc.getReplyLats());
return getOrderedLatencies(allSLatencies);
}
// Return the collected latencies indexed by response number instead of by messageID
private Map<Integer, List<Long>> getOrderedLatencies(Collection<Collection<Long>> latencyLists)
{
Map<Integer, List<Long>> ret = new HashMap<Integer, List<Long>>();
// N may vary
int maxResponses = 0;
for (Collection<Long> latencies : latencyLists)
{
List<Long> sortedLatencies = new ArrayList<Long>(latencies);
Collections.sort(sortedLatencies);
if (sortedLatencies.size() > maxResponses)
{
for (int i = maxResponses + 1; i <= sortedLatencies.size(); ++i)
{
ret.put(i, new Vector<Long>());
}
maxResponses = sortedLatencies.size();
}
// indexing by 0 is awkward since we're talking about the ith response
for (int i = 1; i <= sortedLatencies.size(); ++i)
{
ret.get(i).add(sortedLatencies.get(i - 1));
}
}
return ret;
}
}

35
src/java/org/apache/cassandra/service/PBSPredictorMBean.java Normal file
View File

@ -0,0 +1,35 @@
/**
* 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;
public interface PBSPredictorMBean
{
public PBSPredictionResult doPrediction(int n,
int r,
int w,
float timeSinceWrite,
int numberVersionsStale,
float percentileLatency) throws Exception;
public void enableConsistencyPredictionLogging();
public void disableConsistencyPredictionLogging();
public void setMaxLoggedLatenciesForConsistencyPrediction(int maxLogged);
public void setNumberTrialsForConsistencyPrediction(int numTrials);
}

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

@ -405,6 +405,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
throw new AssertionError(e);
}
PBSPredictor.instance().init();
if (Boolean.parseBoolean(System.getProperty("cassandra.load_ring_state", "true")))
{
logger.info("Loading persisted ring state");

65
src/java/org/apache/cassandra/tools/NodeCmd.java
View File

@ -33,6 +33,12 @@ import com.google.common.collect.LinkedHashMultimap;
import com.google.common.collect.Maps;
import org.apache.commons.cli.*;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.CacheServiceMBean;
import org.apache.cassandra.service.PBSPredictionResult;
import org.apache.cassandra.service.PBSPredictorMBean;
import org.apache.cassandra.service.StorageProxyMBean;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutorMBean;
import org.apache.cassandra.db.ColumnFamilyStoreMBean;
import org.apache.cassandra.db.Table;
@ -137,7 +143,8 @@ public class NodeCmd
DESCRIBERING,
RANGEKEYSAMPLE,
REBUILD_INDEX,
RESETLOCALSCHEMA
RESETLOCALSCHEMA,
PREDICTCONSISTENCY
}
@ -866,6 +873,48 @@ public class NodeCmd
outs.println(probe.isThriftServerRunning() ? "running" : "not running");
}
public void predictConsistency(Integer replicationFactor,
Integer timeAfterWrite,
Integer numVersions,
Float percentileLatency,
PrintStream output)
{
PBSPredictorMBean predictorMBean = probe.getPBSPredictorMBean();
for(int r = 1; r <= replicationFactor; ++r) {
for(int w = 1; w <= replicationFactor; ++w) {
if(w+r > replicationFactor+1)
continue;
try {
PBSPredictionResult result = predictorMBean.doPrediction(replicationFactor,
r,
w,
timeAfterWrite,
numVersions,
percentileLatency);
if(r == 1 && w == 1) {
output.printf("%dms after a given write, with maximum version staleness of k=%d\n", timeAfterWrite, numVersions);
}
output.printf("N=%d, R=%d, W=%d\n", replicationFactor, r, w);
output.printf("Probability of consistent reads: %f\n", result.getConsistencyProbability());
output.printf("Average read latency: %fms (%.3fth %%ile %dms)\n", result.getAverageReadLatency(),
result.getPercentileReadLatencyPercentile()*100,
result.getPercentileReadLatencyValue());
output.printf("Average write latency: %fms (%.3fth %%ile %dms)\n\n", result.getAverageWriteLatency(),
result.getPercentileWriteLatencyPercentile()*100,
result.getPercentileWriteLatencyValue());
} catch (Exception e) {
System.out.println(e.getMessage());
e.printStackTrace();
return;
}
}
}
}
public static void main(String[] args) throws IOException, InterruptedException, ConfigurationException, ParseException
{
CommandLineParser parser = new PosixParser();
@ -1134,6 +1183,20 @@ public class NodeCmd
nodeCmd.printRangeKeySample(System.out);
break;
case PREDICTCONSISTENCY:
if (arguments.length < 2) { badUse("Requires replication factor and time"); }
int numVersions = 1;
if (arguments.length == 3) { numVersions = Integer.parseInt(arguments[2]); }
float percentileLatency = .999f;
if (arguments.length == 4) { percentileLatency = Float.parseFloat(arguments[3]); }
nodeCmd.predictConsistency(Integer.parseInt(arguments[0]),
Integer.parseInt(arguments[1]),
numVersions,
percentileLatency,
System.out);
break;
default :
throw new RuntimeException("Unreachable code.");
}

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

@ -73,6 +73,7 @@ public class NodeProbe
public MessagingServiceMBean msProxy;
private FailureDetectorMBean fdProxy;
private CacheServiceMBean cacheService;
private PBSPredictorMBean PBSPredictorProxy;
private StorageProxyMBean spProxy;
/**
@ -142,6 +143,8 @@ public class NodeProbe
{
ObjectName name = new ObjectName(ssObjName);
ssProxy = JMX.newMBeanProxy(mbeanServerConn, name, StorageServiceMBean.class);
name = new ObjectName(PBSPredictor.MBEAN_NAME);
PBSPredictorProxy = JMX.newMBeanProxy(mbeanServerConn, name, PBSPredictorMBean.class);
name = new ObjectName(MessagingService.MBEAN_NAME);
msProxy = JMX.newMBeanProxy(mbeanServerConn, name, MessagingServiceMBean.class);
name = new ObjectName(StreamingService.MBEAN_OBJECT_NAME);
@ -714,6 +717,11 @@ public class NodeProbe
return ssProxy.describeRingJMX(keyspaceName);
}
public PBSPredictorMBean getPBSPredictorMBean()
{
return PBSPredictorProxy;
}
public void rebuild(String sourceDc)
{
ssProxy.rebuild(sourceDc);

4
src/resources/org/apache/cassandra/tools/NodeToolHelp.yaml
View File

@ -149,4 +149,6 @@ commands:
- name: getsstables <keyspace> <cf> <key>
help: |
Print the sstable filenames that own the key
- name: predictconsistency <replication_factor> <time> [versions] [latency_percentile]
help: |
Predict latency and consistency "t" ms after writes

114
test/unit/org/apache/cassandra/service/PBSPredictorTest.java Normal file
View File

@ -0,0 +1,114 @@
/*
* 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 org.junit.Test;
import static org.junit.Assert.*;
public class PBSPredictorTest
{
private static PBSPredictor predictor = PBSPredictor.instance();
private void createWriteResponse(long W, long A, String id)
{
predictor.startWriteOperation(id, 0);
predictor.logWriteResponse(id, W, W+A);
}
private void createReadResponse(long R, long S, String id)
{
predictor.startReadOperation(id, 0);
predictor.logReadResponse(id, R, R+S);
}
@Test
public void testDoPrediction()
{
try {
predictor.enableConsistencyPredictionLogging();
predictor.init();
/*
Ensure accuracy given a set of basic latencies
Predictions here match a prior Python implementation
*/
for (int i = 0; i < 10; ++i)
{
createWriteResponse(10, 0, String.format("W%d", i));
createReadResponse(0, 0, String.format("R%d", i));
}
for (int i = 0; i < 10; ++i)
{
createWriteResponse(0, 0, String.format("WS%d", i));
}
// 10ms after write
PBSPredictionResult result = predictor.doPrediction(2,1,1,10.0f,1, 0.99f);
assertEquals(1, result.getConsistencyProbability(), 0);
assertEquals(2.5, result.getAverageWriteLatency(), .5);
// 0ms after write
result = predictor.doPrediction(2,1,1,0f,1, 0.99f);
assertEquals(.75, result.getConsistencyProbability(), 0.05);
// k=5 versions staleness
result = predictor.doPrediction(2,1,1,5.0f,5, 0.99f);
assertEquals(.98, result.getConsistencyProbability(), 0.05);
assertEquals(2.5, result.getAverageWriteLatency(), .5);
for (int i = 0; i < 10; ++i)
{
createWriteResponse(20, 0, String.format("WL%d", i));
}
// 5ms after write
result = predictor.doPrediction(2,1,1,5.0f,1, 0.99f);
assertEquals(.67, result.getConsistencyProbability(), .05);
// N = 5
result = predictor.doPrediction(5,1,1,5.0f,1, 0.99f);
assertEquals(.42, result.getConsistencyProbability(), .05);
assertEquals(1.33, result.getAverageWriteLatency(), .5);
for (int i = 0; i < 10; ++i)
{
createWriteResponse(100, 100, String.format("WVL%d", i));
createReadResponse(100, 100, String.format("RL%d", i));
}
result = predictor.doPrediction(2,1,1,0f,1, 0.99f);
assertEquals(.860, result.getConsistencyProbability(), .05);
assertEquals(26.5, result.getAverageWriteLatency(), 1);
assertEquals(100.33, result.getAverageReadLatency(), 4);
result = predictor.doPrediction(2,2,1,0f,1, 0.99f);
assertEquals(1, result.getConsistencyProbability(), 0);
} catch (Exception e) {
fail(e.getMessage());
}
}
}