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

Merge branch 'cassandra-2.1' into trunk

This commit is contained in:
Benedict Elliott Smith 2014-12-12 14:10:04 +00:00
parent add839d317 6c579a0102
commit 36280a0783
51 changed files with 327 additions and 1053 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -40,6 +40,7 @@
2.1.3
* cassandra-stress simultaneous inserts over same seed (CASSANDRA-7964)
* Reduce cassandra-stress sampling memory requirements (CASSANDRA-7926)
* Ensure memtable flush cannot expire commit log entries from its future (CASSANDRA-8383)
* Make read "defrag" async to reclaim memtables (CASSANDRA-8459)

86
tools/stress/src/org/apache/cassandra/stress/Operation.java
View File

@ -19,13 +19,13 @@ package org.apache.cassandra.stress;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.Partition;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.settings.*;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.settings.SettingsLog;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.JavaDriverClient;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
@ -36,17 +36,42 @@ public abstract class Operation
{
public final StressSettings settings;
public final Timer timer;
public final PartitionGenerator generator;
public final Distribution partitionCount;
protected final DataSpec spec;
protected List<Partition> partitions;
private final List<PartitionIterator> partitionCache = new ArrayList<>();
protected List<PartitionIterator> partitions;
public Operation(Timer timer, PartitionGenerator generator, StressSettings settings, Distribution partitionCount)
public static final class DataSpec
{
public final PartitionGenerator partitionGenerator;
final SeedManager seedManager;
final Distribution partitionCount;
final RatioDistribution useRatio;
final Integer targetCount;
public DataSpec(PartitionGenerator partitionGenerator, SeedManager seedManager, Distribution partitionCount, Integer targetCount)
{
this(partitionGenerator, seedManager, partitionCount, null, targetCount);
}
public DataSpec(PartitionGenerator partitionGenerator, SeedManager seedManager, Distribution partitionCount, RatioDistribution useRatio)
{
this(partitionGenerator, seedManager, partitionCount, useRatio, null);
}
private DataSpec(PartitionGenerator partitionGenerator, SeedManager seedManager, Distribution partitionCount, RatioDistribution useRatio, Integer targetCount)
{
this.partitionGenerator = partitionGenerator;
this.seedManager = seedManager;
this.partitionCount = partitionCount;
this.useRatio = useRatio;
this.targetCount = targetCount;
}
}
public Operation(Timer timer, StressSettings settings, DataSpec spec)
{
this.generator = generator;
this.timer = timer;
this.settings = settings;
this.partitionCount = partitionCount;
this.spec = spec;
}
public static interface RunOp
@ -56,9 +81,42 @@ public abstract class Operation
public int rowCount();
}
protected void setPartitions(List<Partition> partitions)
boolean ready(WorkManager permits, RateLimiter rateLimiter)
{
this.partitions = partitions;
int partitionCount = (int) spec.partitionCount.next();
if (partitionCount <= 0)
return false;
partitionCount = permits.takePermits(partitionCount);
if (partitionCount <= 0)
return false;
int i = 0;
boolean success = true;
for (; i < partitionCount && success ; i++)
{
if (i >= partitionCache.size())
partitionCache.add(PartitionIterator.get(spec.partitionGenerator, spec.seedManager));
success = false;
while (!success)
{
Seed seed = spec.seedManager.next(this);
if (seed == null)
break;
if (spec.useRatio == null)
success = partitionCache.get(i).reset(seed, spec.targetCount, this);
else
success = partitionCache.get(i).reset(seed, spec.useRatio.next(), this);
}
}
partitionCount = i;
if (rateLimiter != null)
rateLimiter.acquire(partitionCount);
partitions = partitionCache.subList(0, partitionCount);
return !partitions.isEmpty();
}
public boolean isWrite()
@ -135,7 +193,7 @@ public abstract class Operation
private String key()
{
List<String> keys = new ArrayList<>();
for (Partition partition : partitions)
for (PartitionIterator partition : partitions)
keys.add(partition.getKeyAsString());
return keys.toString();
}

96
tools/stress/src/org/apache/cassandra/stress/StressAction.java
View File

@ -21,19 +21,16 @@ import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.util.concurrent.RateLimiter;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.stress.generate.Partition;
import org.apache.cassandra.stress.operations.OpDistribution;
import org.apache.cassandra.stress.operations.OpDistributionFactory;
import org.apache.cassandra.stress.settings.*;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.JavaDriverClient;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
@ -180,9 +177,9 @@ public class StressAction implements Runnable
: "until stderr of mean < " + settings.command.targetUncertainty));
final WorkManager workManager;
if (opCount < 0)
workManager = new ContinuousWorkManager();
workManager = new WorkManager.ContinuousWorkManager();
else
workManager = new FixedWorkManager(opCount);
workManager = new WorkManager.FixedWorkManager(opCount);
final StressMetrics metrics = new StressMetrics(output, settings.log.intervalMillis, settings);
@ -285,36 +282,12 @@ public class StressAction implements Runnable
throw new IllegalStateException();
}
int maxBatchSize = operations.maxBatchSize();
Partition[] partitions = new Partition[maxBatchSize];
while (true)
{
// TODO: Operation should be able to ecapsulate much of this behaviour
Operation op = operations.next();
op.generator.reset();
int batchSize = workManager.takePermits(Math.max(1, (int) op.partitionCount.next()));
if (batchSize < 0)
if (!op.ready(workManager, rateLimiter))
break;
if (rateLimiter != null)
rateLimiter.acquire(batchSize);
int partitionCount = 0;
while (partitionCount < batchSize)
{
Partition p = op.generator.generate(op);
if (p == null)
break;
partitions[partitionCount++] = p;
}
if (partitionCount == 0)
break;
op.setPartitions(Arrays.asList(partitions).subList(0, partitionCount));
try
{
switch (settings.mode.api)
@ -358,65 +331,4 @@ public class StressAction implements Runnable
}
private interface WorkManager
{
// -1 indicates consumer should terminate
int takePermits(int count);
// signal all consumers to terminate
void stop();
}
private static final class FixedWorkManager implements WorkManager
{
final AtomicLong permits;
public FixedWorkManager(long permits)
{
this.permits = new AtomicLong(permits);
}
@Override
public int takePermits(int count)
{
while (true)
{
long cur = permits.get();
if (cur == 0)
return -1;
count = (int) Math.min(count, cur);
long next = cur - count;
if (permits.compareAndSet(cur, next))
return count;
}
}
@Override
public void stop()
{
permits.getAndSet(0);
}
}
private static final class ContinuousWorkManager implements WorkManager
{
volatile boolean stop = false;
@Override
public int takePermits(int count)
{
if (stop)
return -1;
return count;
}
@Override
public void stop()
{
stop = true;
}
}
}

1
tools/stress/src/org/apache/cassandra/stress/StressMetrics.java
View File

@ -31,7 +31,6 @@ import java.util.concurrent.ThreadFactory;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.commons.lang3.time.DurationFormatUtils;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.stress.settings.SettingsLog;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.JmxCollector;
import org.apache.cassandra.stress.util.Timing;

65
tools/stress/src/org/apache/cassandra/stress/StressProfile.java
View File

@ -21,37 +21,26 @@
package org.apache.cassandra.stress;
import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.AlreadyExistsException;
import java.io.IOError;
import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.net.URI;
import java.util.*;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Function;
import com.google.common.util.concurrent.Uninterruptibles;
import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.AlreadyExistsException;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.statements.CreateKeyspaceStatement;
import org.apache.cassandra.exceptions.RequestValidationException;
import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.RatioDistributionFactory;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.generate.values.Booleans;
import org.apache.cassandra.stress.generate.values.Bytes;
import org.apache.cassandra.stress.generate.values.Generator;
import org.apache.cassandra.stress.generate.values.Dates;
import org.apache.cassandra.stress.generate.values.Doubles;
import org.apache.cassandra.stress.generate.values.Floats;
import org.apache.cassandra.stress.generate.values.GeneratorConfig;
import org.apache.cassandra.stress.generate.values.Inets;
import org.apache.cassandra.stress.generate.values.Integers;
import org.apache.cassandra.stress.generate.values.Lists;
import org.apache.cassandra.stress.generate.values.Longs;
import org.apache.cassandra.stress.generate.values.Sets;
import org.apache.cassandra.stress.generate.values.Strings;
import org.apache.cassandra.stress.generate.values.TimeUUIDs;
import org.apache.cassandra.stress.generate.values.UUIDs;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.generate.values.*;
import org.apache.cassandra.stress.operations.userdefined.SchemaInsert;
import org.apache.cassandra.stress.operations.userdefined.SchemaQuery;
import org.apache.cassandra.stress.settings.OptionDistribution;
@ -68,19 +57,6 @@ import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.Constructor;
import org.yaml.snakeyaml.error.YAMLException;
import java.io.*;
import java.net.URI;
import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
public class StressProfile implements Serializable
{
private String keyspaceCql;
@ -247,7 +223,7 @@ public class StressProfile implements Serializable
}
}
public SchemaQuery getQuery(String name, Timer timer, PartitionGenerator generator, StressSettings settings)
public SchemaQuery getQuery(String name, Timer timer, PartitionGenerator generator, SeedManager seeds, StressSettings settings)
{
if (queryStatements == null)
{
@ -286,10 +262,11 @@ public class StressProfile implements Serializable
name = name.toLowerCase();
if (!queryStatements.containsKey(name))
throw new IllegalArgumentException("No query defined with name " + name);
return new SchemaQuery(timer, generator, settings, thriftQueryIds.get(name), queryStatements.get(name), ThriftConversion.fromThrift(settings.command.consistencyLevel), ValidationType.NOT_FAIL, argSelects.get(name));
return new SchemaQuery(timer, settings, generator, seeds, thriftQueryIds.get(name), queryStatements.get(name),
ThriftConversion.fromThrift(settings.command.consistencyLevel), ValidationType.NOT_FAIL, argSelects.get(name));
}
public SchemaInsert getInsert(Timer timer, PartitionGenerator generator, StressSettings settings)
public SchemaInsert getInsert(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
if (insertStatement == null)
{
@ -401,7 +378,7 @@ public class StressProfile implements Serializable
}
}
return new SchemaInsert(timer, generator, settings, partitions.get(), selectchance.get(), thriftInsertId, insertStatement, ThriftConversion.fromThrift(settings.command.consistencyLevel), batchType);
return new SchemaInsert(timer, settings, generator, seedManager, partitions.get(), selectchance.get(), thriftInsertId, insertStatement, ThriftConversion.fromThrift(settings.command.consistencyLevel), batchType);
}
private static <E> E select(E first, String key, String defValue, Map<String, String> map, Function<String, E> builder)
@ -415,7 +392,7 @@ public class StressProfile implements Serializable
return builder.apply(defValue);
}
public PartitionGenerator newGenerator(StressSettings settings, SeedManager seeds)
public PartitionGenerator newGenerator(StressSettings settings)
{
if (generatorFactory == null)
{
@ -427,7 +404,7 @@ public class StressProfile implements Serializable
}
}
return generatorFactory.newGenerator(settings, seeds);
return generatorFactory.newGenerator(settings);
}
private class GeneratorFactory
@ -449,9 +426,9 @@ public class StressProfile implements Serializable
valueColumns.add(new ColumnInfo(metadata.getName(), metadata.getType(), columnConfigs.get(metadata.getName())));
}
PartitionGenerator newGenerator(StressSettings settings, SeedManager seeds)
PartitionGenerator newGenerator(StressSettings settings)
{
return new PartitionGenerator(get(partitionKeys), get(clusteringColumns), get(valueColumns), settings.generate.order, seeds);
return new PartitionGenerator(get(partitionKeys), get(clusteringColumns), get(valueColumns), settings.generate.order);
}
List<Generator> get(List<ColumnInfo> columnInfos)

3
tools/stress/src/org/apache/cassandra/stress/StressServer.java
View File

@ -24,9 +24,10 @@ import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.commons.cli.*;
import org.apache.cassandra.stress.settings.StressSettings;
public class StressServer
{
private static final Options availableOptions = new Options();

554
tools/stress/src/org/apache/cassandra/stress/generate/Partition.java
View File

@ -1,554 +0,0 @@
package org.apache.cassandra.stress.generate;
/*
*
* 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.
*
*/
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Deque;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.stress.generate.values.Generator;
// a partition is re-used to reduce garbage generation, as is its internal RowIterator
// TODO: we should batch the generation of clustering components so we can bound the time and size necessary to
// generate huge partitions with only a small number of clustering components; i.e. we should generate seeds for batches
// of a single component, and then generate the values within those batches as necessary. this will be difficult with
// generating sorted partitions, and may require generator support (e.g. we may need to support generating prefixes
// that are extended/suffixed to generate each batch, so that we can sort the prefixes)
public class Partition
{
private long idseed;
private Seed seed;
private final Object[] partitionKey;
private final PartitionGenerator generator;
private final RowIterator iterator;
public Partition(PartitionGenerator generator)
{
this.generator = generator;
this.partitionKey = new Object[generator.partitionKey.size()];
if (generator.clusteringComponents.size() > 0)
iterator = new MultiRowIterator();
else
iterator = new SingleRowIterator();
}
void setSeed(Seed seed)
{
long idseed = 0;
for (int i = 0 ; i < partitionKey.length ; i++)
{
Generator generator = this.generator.partitionKey.get(i);
// set the partition key seed based on the current work item we're processing
generator.setSeed(seed.seed);
Object key = generator.generate();
partitionKey[i] = key;
// then contribute this value to the data seed
idseed = seed(key, generator.type, idseed);
}
this.seed = seed;
this.idseed = idseed;
}
public RowIterator iterator(double useChance, boolean isWrite)
{
iterator.reset(useChance, 0, 1, isWrite);
return iterator;
}
public RowIterator iterator(int targetCount, boolean isWrite)
{
iterator.reset(Double.NaN, targetCount, 1, isWrite);
return iterator;
}
class SingleRowIterator extends RowIterator
{
boolean done;
void reset(double useChance, int targetCount, int batches, boolean isWrite)
{
done = false;
}
public Iterable<Row> next()
{
if (done)
return Collections.emptyList();
for (int i = 0 ; i < row.row.length ; i++)
{
Generator gen = generator.valueComponents.get(i);
gen.setSeed(idseed);
row.row[i] = gen.generate();
}
done = true;
return Collections.singleton(row);
}
public boolean done()
{
return done;
}
public void markWriteFinished()
{
assert done;
generator.seeds.markFinished(seed);
}
}
public abstract class RowIterator
{
// we reuse the row object to save garbage
final Row row = new Row(partitionKey, new Object[generator.clusteringComponents.size() + generator.valueComponents.size()]);
public abstract Iterable<Row> next();
public abstract boolean done();
public abstract void markWriteFinished();
abstract void reset(double useChance, int targetCount, int batches, boolean isWrite);
public Partition partition()
{
return Partition.this;
}
}
// permits iterating a random subset of the procedurally generated rows in this partition. this is the only mechanism for visiting rows.
// we maintain a stack of clustering components and their seeds; for each clustering component we visit, we generate all values it takes at that level,
// and then, using the average (total) number of children it takes we randomly choose whether or not we visit its children;
// if we do, we generate all possible values the immediate children can take, and repeat the process. So at any one time we are using space proportional
// to C.N, where N is the average number of values each clustering component takes, as opposed to N^C total values in the partition.
// TODO : guarantee at least one row is always returned
// TODO : support first/last row, and constraining reads to rows we know are populated
class MultiRowIterator extends RowIterator
{
// probability any single row will be generated in this iteration
double useChance;
// the seed used to generate the current values for the clustering components at each depth;
// used to save recalculating it for each row, so we only need to recalc from prior row.
final long[] clusteringSeeds = new long[generator.clusteringComponents.size()];
// the components remaining to be visited for each level of the current stack
final Deque<Object>[] clusteringComponents = new ArrayDeque[generator.clusteringComponents.size()];
// we want our chance of selection to be applied uniformly, so we compound the roll we make at each level
// so that we know with what chance we reached there, and we adjust our roll at that level by that amount
final double[] chancemodifier = new double[generator.clusteringComponents.size()];
final double[] rollmodifier = new double[generator.clusteringComponents.size()];
// track where in the partition we are, and where we are limited to
final int[] position = new int[generator.clusteringComponents.size()];
final int[] limit = new int[position.length];
int batchSize;
boolean returnedOne;
boolean forceReturnOne;
// reusable collections for generating unique and sorted clustering components
final Set<Object> unique = new HashSet<>();
final List<Comparable> tosort = new ArrayList<>();
final Random random = new Random();
MultiRowIterator()
{
for (int i = 0 ; i < clusteringComponents.length ; i++)
clusteringComponents[i] = new ArrayDeque<>();
rollmodifier[0] = 1f;
chancemodifier[0] = generator.clusteringChildAverages[0];
}
// if we're a write, the expected behaviour is that the requested batch count is compounded with the seed's visit
// count to decide how much we should return in one iteration
void reset(double useChance, int targetCount, int batches, boolean isWrite)
{
if (this.useChance < 1d)
{
// we clear our prior roll-modifiers if the use chance was previously less-than zero
Arrays.fill(rollmodifier, 1d);
Arrays.fill(chancemodifier, 1d);
}
// set the seed for the first clustering component
generator.clusteringComponents.get(0).setSeed(idseed);
int[] position = seed.position;
// calculate how many first clustering components we'll generate, and how many total rows this predicts
int firstComponentCount = (int) generator.clusteringComponents.get(0).clusteringDistribution.next();
int expectedRowCount;
if (!isWrite && position != null)
{
expectedRowCount = 0;
for (int i = 0 ; i < position.length ; i++)
{
expectedRowCount += position[i] * generator.clusteringChildAverages[i];
limit[i] = position[i];
}
}
else
{
expectedRowCount = firstComponentCount * generator.clusteringChildAverages[0];
if (isWrite)
batches *= seed.visits;
Arrays.fill(limit, Integer.MAX_VALUE);
}
batchSize = Math.max(1, expectedRowCount / batches);
if (Double.isNaN(useChance))
useChance = Math.max(0d, Math.min(1d, targetCount / (double) expectedRowCount));
// clear any remnants of the last iteration, wire up our constants, and fill in the first clustering components
this.useChance = useChance;
this.returnedOne = false;
for (Queue<?> q : clusteringComponents)
q.clear();
clusteringSeeds[0] = idseed;
fill(clusteringComponents[0], firstComponentCount, generator.clusteringComponents.get(0));
// seek to our start position
seek(isWrite ? position : null);
}
// generate the clustering components for the provided depth; requires preceding components
// to have been generated and their seeds populated into clusteringSeeds
void fill(int depth)
{
long seed = clusteringSeeds[depth - 1];
Generator gen = generator.clusteringComponents.get(depth);
gen.setSeed(seed);
clusteringSeeds[depth] = seed(clusteringComponents[depth - 1].peek(), generator.clusteringComponents.get(depth - 1).type, seed);
fill(clusteringComponents[depth], (int) gen.clusteringDistribution.next(), gen);
}
// generate the clustering components into the queue
void fill(Queue<Object> queue, int count, Generator generator)
{
if (count == 1)
{
queue.add(generator.generate());
return;
}
switch (Partition.this.generator.order)
{
case SORTED:
if (Comparable.class.isAssignableFrom(generator.clazz))
{
tosort.clear();
for (int i = 0 ; i < count ; i++)
tosort.add((Comparable) generator.generate());
Collections.sort(tosort);
for (int i = 0 ; i < count ; i++)
queue.add(tosort.get(i));
break;
}
else
{
throw new RuntimeException("Generator class is not comparable: "+generator.clazz);
}
case ARBITRARY:
unique.clear();
for (int i = 0 ; i < count ; i++)
{
Object next = generator.generate();
if (unique.add(next))
queue.add(next);
}
break;
case SHUFFLED:
unique.clear();
tosort.clear();
for (int i = 0 ; i < count ; i++)
{
Object next = generator.generate();
if (unique.add(next))
tosort.add(new RandomOrder(next));
}
Collections.sort(tosort);
for (Object o : tosort)
queue.add(((RandomOrder)o).value);
break;
default:
throw new IllegalStateException();
}
}
// seek to the provided position (or the first entry if null)
private void seek(int[] position)
{
if (position == null)
{
this.position[0] = -1;
clusteringComponents[0].addFirst(this);
advance(0);
return;
}
assert position.length == clusteringComponents.length;
for (int i = 0 ; i < position.length ; i++)
{
if (i != 0)
fill(i);
for (int c = position[i] ; c > 0 ; c--)
clusteringComponents[i].poll();
row.row[i] = clusteringComponents[i].peek();
}
System.arraycopy(position, 0, this.position, 0, position.length);
}
// normal method for moving the iterator forward; maintains the row object, and delegates to advance(int)
// to move the iterator to the next item
void advance()
{
// we are always at the leaf level when this method is invoked
// so we calculate the seed for generating the row by combining the seed that generated the clustering components
int depth = clusteringComponents.length - 1;
long parentSeed = clusteringSeeds[depth];
long rowSeed = seed(clusteringComponents[depth].peek(), generator.clusteringComponents.get(depth).type, parentSeed);
// and then fill the row with the _non-clustering_ values for the position we _were_ at, as this is what we'll deliver
for (int i = clusteringSeeds.length ; i < row.row.length ; i++)
{
Generator gen = generator.valueComponents.get(i - clusteringSeeds.length);
gen.setSeed(rowSeed);
row.row[i] = gen.generate();
}
returnedOne = true;
forceReturnOne = false;
// then we advance the leaf level
advance(depth);
}
private void advance(int depth)
{
// advance the leaf component
clusteringComponents[depth].poll();
position[depth]++;
while (true)
{
if (clusteringComponents[depth].isEmpty())
{
// if we've run out of clustering components at this level, ascend
if (depth == 0)
return;
depth--;
clusteringComponents[depth].poll();
position[depth]++;
continue;
}
if (depth == 0 && !returnedOne && clusteringComponents[0].size() == 1)
forceReturnOne = true;
// the chance of descending is the uniform usechance, multiplied by the number of children
// we would on average generate (so if we have a 0.1 use chance, but should generate 10 children
// then we will always descend), multiplied by 1/(compound roll), where (compound roll) is the
// chance with which we reached this depth, i.e. if we already beat 50/50 odds, we double our
// chance of beating this next roll
double thischance = useChance * chancemodifier[depth];
if (forceReturnOne || thischance > 0.999f || thischance >= random.nextDouble())
{
// if we're descending, we fill in our clustering component and increase our depth
row.row[depth] = clusteringComponents[depth].peek();
depth++;
if (depth == clusteringComponents.length)
break;
// if we haven't reached the leaf, we update our probability statistics, fill in all of
// this level's clustering components, and repeat
if (useChance < 1d)
{
rollmodifier[depth] = rollmodifier[depth - 1] / Math.min(1d, thischance);
chancemodifier[depth] = generator.clusteringChildAverages[depth] * rollmodifier[depth];
}
position[depth] = 0;
fill(depth);
continue;
}
// if we don't descend, we remove the clustering suffix we've skipped and continue
clusteringComponents[depth].poll();
position[depth]++;
}
}
public Iterable<Row> next()
{
final int[] limit = position.clone();
int remainingSize = batchSize;
for (int i = 0 ; i < limit.length && remainingSize > 0 ; i++)
{
limit[i] += remainingSize / generator.clusteringChildAverages[i];
remainingSize %= generator.clusteringChildAverages[i];
}
assert remainingSize == 0;
for (int i = limit.length - 1 ; i > 0 ; i--)
{
if (limit[i] > generator.clusteringChildAverages[i])
{
limit[i - 1] += limit[i] / generator.clusteringChildAverages[i];
limit[i] %= generator.clusteringChildAverages[i];
}
}
for (int i = 0 ; i < limit.length ; i++)
{
if (limit[i] < this.limit[i])
break;
limit[i] = Math.min(limit[i], this.limit[i]);
}
return new Iterable<Row>()
{
public Iterator<Row> iterator()
{
return new Iterator<Row>()
{
public boolean hasNext()
{
if (done())
return false;
for (int i = 0 ; i < position.length ; i++)
if (position[i] < limit[i])
return true;
return false;
}
public Row next()
{
advance();
return row;
}
public void remove()
{
throw new UnsupportedOperationException();
}
};
}
};
}
public boolean done()
{
return clusteringComponents[0].isEmpty();
}
public void markWriteFinished()
{
if (done())
generator.seeds.markFinished(seed);
else
generator.seeds.markVisited(seed, position.clone());
}
public Partition partition()
{
return Partition.this;
}
}
private static class RandomOrder implements Comparable<RandomOrder>
{
final int order = ThreadLocalRandom.current().nextInt();
final Object value;
private RandomOrder(Object value)
{
this.value = value;
}
public int compareTo(RandomOrder that)
{
return Integer.compare(this.order, that.order);
}
}
// calculate a new seed based on the combination of a parent seed and the generated child, to generate
// any children of this child
static long seed(Object object, AbstractType type, long seed)
{
if (object instanceof ByteBuffer)
{
ByteBuffer buf = (ByteBuffer) object;
for (int i = buf.position() ; i < buf.limit() ; i++)
seed = (31 * seed) + buf.get(i);
return seed;
}
else if (object instanceof String)
{
String str = (String) object;
for (int i = 0 ; i < str.length() ; i++)
seed = (31 * seed) + str.charAt(i);
return seed;
}
else if (object instanceof Number)
{
return (seed * 31) + ((Number) object).longValue();
}
else if (object instanceof UUID)
{
return seed * 31 + (((UUID) object).getLeastSignificantBits() ^ ((UUID) object).getMostSignificantBits());
}
else
{
return seed(type.decompose(object), BytesType.instance, seed);
}
}
public Object getPartitionKey(int i)
{
return partitionKey[i];
}
public String getKeyAsString()
{
StringBuilder sb = new StringBuilder();
int i = 0;
for (Object key : partitionKey)
{
if (i > 0)
sb.append("|");
AbstractType type = generator.partitionKey.get(i++).type;
sb.append(type.getString(type.decompose(key)));
}
return sb.toString();
}
// used for thrift smart routing - if it's a multi-part key we don't try to route correctly right now
public ByteBuffer getToken()
{
return generator.partitionKey.get(0).type.decompose(partitionKey[0]);
}
}

39
tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java
View File

@ -22,7 +22,6 @@ package org.apache.cassandra.stress.generate;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@ -30,7 +29,6 @@ import java.util.NoSuchElementException;
import com.google.common.collect.Iterables;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.generate.values.Generator;
public class PartitionGenerator
@ -46,30 +44,24 @@ public class PartitionGenerator
final List<Generator> partitionKey;
final List<Generator> clusteringComponents;
final List<Generator> valueComponents;
final int[] clusteringChildAverages;
final int[] clusteringDescendantAverages;
final int[] clusteringComponentAverages;
private final Map<String, Integer> indexMap;
final Order order;
final SeedManager seeds;
final List<Partition> recyclable = new ArrayList<>();
int partitionsInUse = 0;
public void reset()
{
partitionsInUse = 0;
}
public PartitionGenerator(List<Generator> partitionKey, List<Generator> clusteringComponents, List<Generator> valueComponents, Order order, SeedManager seeds)
public PartitionGenerator(List<Generator> partitionKey, List<Generator> clusteringComponents, List<Generator> valueComponents, Order order)
{
this.partitionKey = partitionKey;
this.clusteringComponents = clusteringComponents;
this.valueComponents = valueComponents;
this.order = order;
this.seeds = seeds;
this.clusteringChildAverages = new int[clusteringComponents.size()];
for (int i = clusteringChildAverages.length - 1 ; i >= 0 ; i--)
clusteringChildAverages[i] = (int) (i < (clusteringChildAverages.length - 1) ? clusteringComponents.get(i + 1).clusteringDistribution.average() * clusteringChildAverages[i + 1] : 1);
this.clusteringDescendantAverages = new int[clusteringComponents.size()];
this.clusteringComponentAverages = new int[clusteringComponents.size()];
for (int i = 0 ; i < clusteringComponentAverages.length ; i++)
clusteringComponentAverages[i] = (int) clusteringComponents.get(i).clusteringDistribution.average();
for (int i = clusteringDescendantAverages.length - 1 ; i >= 0 ; i--)
clusteringDescendantAverages[i] = (int) (i < (clusteringDescendantAverages.length - 1) ? clusteringComponentAverages[i + 1] * clusteringDescendantAverages[i + 1] : 1);
double maxRowCount = 1d;
double minRowCount = 1d;
for (Generator component : clusteringComponents)
@ -101,19 +93,6 @@ public class PartitionGenerator
return i;
}
public Partition generate(Operation op)
{
if (recyclable.size() <= partitionsInUse || recyclable.get(partitionsInUse) == null)
recyclable.add(new Partition(this));
Seed seed = seeds.next(op);
if (seed == null)
return null;
Partition partition = recyclable.get(partitionsInUse++);
partition.setSeed(seed);
return partition;
}
public ByteBuffer convert(int c, Object v)
{
if (c < 0)

62
tools/stress/src/org/apache/cassandra/stress/generate/Seed.java
View File

@ -18,50 +18,68 @@
*/
package org.apache.cassandra.stress.generate;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.cassandra.stress.util.DynamicList;
public class Seed implements Comparable<Seed>
{
public final int visits;
public final long seed;
final int visits;
DynamicList.Node poolNode;
volatile int[] position;
volatile State state = State.HELD;
private volatile DynamicList.Node poolNode;
private volatile int position;
private static final AtomicReferenceFieldUpdater<Seed, Seed.State> stateUpdater = AtomicReferenceFieldUpdater.newUpdater(Seed.class, State.class, "state");
private static final AtomicIntegerFieldUpdater<Seed> positionUpdater = AtomicIntegerFieldUpdater.newUpdater(Seed.class, "position");
public int compareTo(Seed that)
{
return Long.compare(this.seed, that.seed);
}
static enum State
{
HELD, AVAILABLE
}
Seed(long seed, int visits)
{
this.seed = seed;
this.visits = visits;
}
boolean take()
{
return stateUpdater.compareAndSet(this, State.AVAILABLE, State.HELD);
}
void yield()
{
state = State.AVAILABLE;
}
public int[] position()
public int position()
{
return position;
}
public int moveForwards(int rowCount)
{
return positionUpdater.getAndAdd(this, rowCount);
}
public int hashCode()
{
return (int) seed;
}
public boolean equals(Object that)
{
return that instanceof Seed && this.seed == ((Seed) that).seed;
}
public boolean save(DynamicList<Seed> sampleFrom, int maxSize)
{
DynamicList.Node poolNode = sampleFrom.append(this, maxSize);
if (poolNode == null)
return false;
this.poolNode = poolNode;
return true;
}
public boolean isSaved()
{
return poolNode != null;
}
public void remove(DynamicList<Seed> sampleFrom)
{
sampleFrom.remove(poolNode);
}
}

66
tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java
View File

@ -33,9 +33,12 @@ public class SeedManager
final Distribution visits;
final Generator writes;
final Generator reads;
final ConcurrentHashMap<Seed, Seed> managing = new ConcurrentHashMap<>();
final ConcurrentHashMap<Long, Seed> managing = new ConcurrentHashMap<>();
final DynamicList<Seed> sampleFrom;
final Distribution sample;
final long sampleOffset;
final int sampleSize;
final boolean updateSampleImmediately;
public SeedManager(StressSettings settings)
{
@ -61,10 +64,15 @@ public class SeedManager
this.visits = settings.insert.visits.get();
this.writes = writes;
this.reads = reads;
this.sample = DistributionInverted.invert(settings.insert.revisit.get());
if (sample.maxValue() > Integer.MAX_VALUE || sample.minValue() < 0)
throw new IllegalArgumentException();
this.sampleFrom = new DynamicList<>((int) sample.maxValue());
Distribution sample = settings.insert.revisit.get();
this.sampleOffset = Math.min(sample.minValue(), sample.maxValue());
long sampleSize = 1 + Math.max(sample.minValue(), sample.maxValue()) - sampleOffset;
if (sampleOffset < 0 || sampleSize > Integer.MAX_VALUE)
throw new IllegalArgumentException("sample range is invalid");
this.sampleFrom = new DynamicList<>((int) sampleSize);
this.sample = DistributionInverted.invert(sample);
this.sampleSize = (int) sampleSize;
this.updateSampleImmediately = visits.average() > 1;
}
public Seed next(Operation op)
@ -80,48 +88,38 @@ public class SeedManager
while (true)
{
int index = (int) sample.next();
int index = (int) (sample.next() - sampleOffset);
Seed seed = sampleFrom.get(index);
if (seed != null && seed.take())
if (seed != null && seed.isSaved())
return seed;
seed = writes.next((int) visits.next());
if (seed == null)
return null;
// seeds are created HELD, so if we insert it successfully we have it exclusively for our write
if (managing.putIfAbsent(seed, seed) == null)
return seed;
if (managing.putIfAbsent(seed.seed, seed) == null)
{
if (!updateSampleImmediately || seed.save(sampleFrom, sampleSize))
return seed;
managing.remove(seed.seed, seed);
}
}
}
public void markVisited(Seed seed, int[] position)
public void markLastWrite(Seed seed, boolean first)
{
boolean first = seed.position == null;
seed.position = position;
finishedWriting(seed, first, false);
// we could have multiple iterators mark the last write simultaneously,
// so we ensure we remove conditionally, and only remove the exact seed we were operating over
// this is important because, to ensure correctness, we do not support calling remove multiple
// times on the same DynamicList.Node
if (managing.remove(seed.seed, seed) && !first)
seed.remove(sampleFrom);
}
public void markFinished(Seed seed)
public void markFirstWrite(Seed seed, boolean last)
{
finishedWriting(seed, seed.position == null, true);
}
void finishedWriting(Seed seed, boolean first, boolean completed)
{
if (!completed)
{
if (first)
seed.poolNode = sampleFrom.append(seed);
seed.yield();
}
else
{
if (!first)
sampleFrom.remove(seed.poolNode);
managing.remove(seed);
}
if (first)
writes.finishWrite(seed);
if (!last && !updateSampleImmediately)
seed.save(sampleFrom, Integer.MAX_VALUE);
writes.finishWrite(seed);
}
private abstract class Generator

7
tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java
View File

@ -20,12 +20,11 @@
*/
package org.apache.cassandra.stress.generate.values;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.stress.generate.FasterRandom;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Random;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.stress.generate.FasterRandom;
public class Bytes extends Generator<ByteBuffer>
{

9
tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java
View File

@ -20,17 +20,14 @@
*/
package org.apache.cassandra.stress.generate.values;
import java.io.Serializable;
import java.nio.ByteBuffer;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.MurmurHash;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.Map;
public class GeneratorConfig implements Serializable
{
public final long salt;

2
tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java
View File

@ -20,8 +20,6 @@
*/
package org.apache.cassandra.stress.generate.values;
import java.util.Random;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.stress.generate.FasterRandom;

4
tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java
View File

@ -21,11 +21,11 @@
package org.apache.cassandra.stress.generate.values;
import java.util.UUID;
import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.utils.UUIDGen;
import java.util.UUID;
public class TimeUUIDs extends Generator<UUID>
{
final Dates dateGen;

7
tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java
View File

@ -25,7 +25,6 @@ import org.apache.cassandra.stress.Operation;
public class FixedOpDistribution implements OpDistribution
{
final Operation operation;
public FixedOpDistribution(Operation operation)
@ -37,10 +36,4 @@ public class FixedOpDistribution implements OpDistribution
{
return operation;
}
public int maxBatchSize()
{
return (int) operation.partitionCount.maxValue();
}
}

1
tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java
View File

@ -27,6 +27,5 @@ public interface OpDistribution
{
Operation next();
public int maxBatchSize();
}

9
tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java
View File

@ -22,7 +22,6 @@ package org.apache.cassandra.stress.operations;
import org.apache.commons.math3.distribution.EnumeratedDistribution;
import org.apache.commons.math3.util.Pair;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.generate.Distribution;
@ -41,14 +40,6 @@ public class SampledOpDistribution implements OpDistribution
this.clustering = clustering;
}
public int maxBatchSize()
{
int max = 1;
for (Pair<Operation, Double> pair : operations.getPmf())
max = Math.max(max, (int) pair.getFirst().partitionCount.maxValue());
return max;
}
public Operation next()
{
while (remaining == 0)

5
tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java
View File

@ -28,6 +28,7 @@ import java.util.List;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.Timer;
@ -36,9 +37,9 @@ public class CqlCounterAdder extends CqlOperation<Integer>
{
final Distribution counteradd;
public CqlCounterAdder(DistributionFactory counteradd, Timer timer, PartitionGenerator generator, StressSettings settings)
public CqlCounterAdder(DistributionFactory counteradd, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(Command.COUNTER_WRITE, timer, generator, settings);
super(Command.COUNTER_WRITE, timer, generator, seedManager, settings);
this.counteradd = counteradd.get();
}

5
tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java
View File

@ -26,6 +26,7 @@ import java.util.Collections;
import java.util.List;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.Timer;
@ -33,9 +34,9 @@ import org.apache.cassandra.stress.util.Timer;
public class CqlCounterGetter extends CqlOperation<Integer>
{
public CqlCounterGetter(Timer timer, PartitionGenerator generator, StressSettings settings)
public CqlCounterGetter(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(Command.COUNTER_READ, timer, generator, settings);
super(Command.COUNTER_READ, timer, generator, seedManager, settings);
}
@Override

6
tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java
View File

@ -26,17 +26,17 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.utils.UUIDGen;
public class CqlInserter extends CqlOperation<Integer>
{
public CqlInserter(Timer timer, PartitionGenerator generator, StressSettings settings)
public CqlInserter(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(Command.WRITE, timer, generator, settings);
super(Command.WRITE, timer, generator, seedManager, settings);
}
@Override

32
tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java
View File

@ -24,13 +24,13 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import com.google.common.base.Function;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.google.common.base.Function;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.StressMetrics;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.ConnectionStyle;
import org.apache.cassandra.stress.settings.CqlVersion;
@ -54,9 +54,9 @@ public abstract class CqlOperation<V> extends PredefinedOperation
protected abstract String buildQuery();
protected abstract CqlRunOp<V> buildRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, ByteBuffer key);
public CqlOperation(Command type, Timer timer, PartitionGenerator generator, StressSettings settings)
public CqlOperation(Command type, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(type, timer, generator, settings);
super(type, timer, generator, seedManager, settings);
if (settings.columns.variableColumnCount)
throw new IllegalStateException("Variable column counts are not implemented for CQL");
}
@ -168,28 +168,6 @@ public abstract class CqlOperation<V> extends PredefinedOperation
}
}
// Requires a custom validate() method, but fetches and stores the keys from the result set for further processing
protected abstract class CqlRunOpFetchKeys extends CqlRunOp<byte[][]>
{
protected CqlRunOpFetchKeys(ClientWrapper client, String query, Object queryId, List<Object> params, ByteBuffer key)
{
super(client, query, queryId, KeysHandler.INSTANCE, params, key);
}
@Override
public int partitionCount()
{
return result.length;
}
@Override
public int rowCount()
{
return result.length;
}
}
protected final class CqlRunOpMatchResults extends CqlRunOp<ByteBuffer[][]>
{

7
tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java
View File

@ -22,23 +22,22 @@ package org.apache.cassandra.stress.operations.predefined;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.utils.ByteBufferUtil;
public class CqlReader extends CqlOperation<ByteBuffer[][]>
{
public CqlReader(Timer timer, PartitionGenerator generator, StressSettings settings)
public CqlReader(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(Command.READ, timer, generator, settings);
super(Command.READ, timer, generator, seedManager, settings);
}
@Override

40
tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java
View File

@ -20,26 +20,17 @@ package org.apache.cassandra.stress.operations.predefined;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumMap;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.StressMetrics;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.DistributionFixed;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.Row;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.CqlVersion;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.utils.ByteBufferUtil;
public abstract class PredefinedOperation extends Operation
{
@ -47,13 +38,18 @@ public abstract class PredefinedOperation extends Operation
private final Distribution columnCount;
private Object cqlCache;
public PredefinedOperation(Command type, Timer timer, PartitionGenerator generator, StressSettings settings)
public PredefinedOperation(Command type, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(timer, generator, settings, new DistributionFixed(1));
super(timer, settings, spec(generator, seedManager));
this.type = type;
this.columnCount = settings.columns.countDistribution.get();
}
private static DataSpec spec(PartitionGenerator generator, SeedManager seedManager)
{
return new DataSpec(generator, seedManager, new DistributionFixed(1), 1);
}
public boolean isCql3()
{
return settings.mode.cqlVersion == CqlVersion.CQL3;
@ -170,7 +166,7 @@ public abstract class PredefinedOperation extends Operation
protected List<ByteBuffer> getColumnValues(ColumnSelection columns)
{
Row row = partitions.get(0).iterator(1, false).next().iterator().next();
Row row = partitions.get(0).next();
ByteBuffer[] r = new ByteBuffer[columns.count()];
int c = 0;
if (columns.indices != null)
@ -182,7 +178,7 @@ public abstract class PredefinedOperation extends Operation
return Arrays.asList(r);
}
public static Operation operation(Command type, Timer timer, PartitionGenerator generator, StressSettings settings, DistributionFactory counteradd)
public static Operation operation(Command type, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings, DistributionFactory counteradd)
{
switch (type)
{
@ -190,10 +186,10 @@ public abstract class PredefinedOperation extends Operation
switch(settings.mode.style)
{
case THRIFT:
return new ThriftReader(timer, generator, settings);
return new ThriftReader(timer, generator, seedManager, settings);
case CQL:
case CQL_PREPARED:
return new CqlReader(timer, generator, settings);
return new CqlReader(timer, generator, seedManager, settings);
default:
throw new UnsupportedOperationException();
}
@ -203,10 +199,10 @@ public abstract class PredefinedOperation extends Operation
switch(settings.mode.style)
{
case THRIFT:
return new ThriftCounterGetter(timer, generator, settings);
return new ThriftCounterGetter(timer, generator, seedManager, settings);
case CQL:
case CQL_PREPARED:
return new CqlCounterGetter(timer, generator, settings);
return new CqlCounterGetter(timer, generator, seedManager, settings);
default:
throw new UnsupportedOperationException();
}
@ -216,10 +212,10 @@ public abstract class PredefinedOperation extends Operation
switch(settings.mode.style)
{
case THRIFT:
return new ThriftInserter(timer, generator, settings);
return new ThriftInserter(timer, generator, seedManager, settings);
case CQL:
case CQL_PREPARED:
return new CqlInserter(timer, generator, settings);
return new CqlInserter(timer, generator, seedManager, settings);
default:
throw new UnsupportedOperationException();
}
@ -228,10 +224,10 @@ public abstract class PredefinedOperation extends Operation
switch(settings.mode.style)
{
case THRIFT:
return new ThriftCounterAdder(counteradd, timer, generator, settings);
return new ThriftCounterAdder(counteradd, timer, generator, seedManager, settings);
case CQL:
case CQL_PREPARED:
return new CqlCounterAdder(counteradd, timer, generator, settings);
return new CqlCounterAdder(counteradd, timer, generator, seedManager, settings);
default:
throw new UnsupportedOperationException();
}

9
tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java
View File

@ -27,19 +27,22 @@ import java.util.Map;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.CounterColumn;
import org.apache.cassandra.thrift.Mutation;
public class ThriftCounterAdder extends PredefinedOperation
{
final Distribution counteradd;
public ThriftCounterAdder(DistributionFactory counteradd, Timer timer, PartitionGenerator generator, StressSettings settings)
public ThriftCounterAdder(DistributionFactory counteradd, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(Command.COUNTER_WRITE, timer, generator, settings);
super(Command.COUNTER_WRITE, timer, generator, seedManager, settings);
this.counteradd = counteradd.get();
}

5
tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java
View File

@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.ThriftClient;
@ -31,9 +32,9 @@ import org.apache.cassandra.thrift.SlicePredicate;
public class ThriftCounterGetter extends PredefinedOperation
{
public ThriftCounterGetter(Timer timer, PartitionGenerator generator, StressSettings settings)
public ThriftCounterGetter(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(Command.COUNTER_READ, timer, generator, settings);
super(Command.COUNTER_READ, timer, generator, seedManager, settings);
}
public void run(final ThriftClient client) throws IOException

11
tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java
View File

@ -24,22 +24,23 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
public final class ThriftInserter extends PredefinedOperation
{
public ThriftInserter(Timer timer, PartitionGenerator generator, StressSettings settings)
public ThriftInserter(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(Command.WRITE, timer, generator, settings);
super(Command.WRITE, timer, generator, seedManager, settings);
}
public boolean isWrite()

7
tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java
View File

@ -22,21 +22,20 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SuperColumn;
public final class ThriftReader extends PredefinedOperation
{
public ThriftReader(Timer timer, PartitionGenerator generator, StressSettings settings)
public ThriftReader(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
super(Command.READ, timer, generator, settings);
super(Command.READ, timer, generator, seedManager, settings);
}
public void run(final ThriftClient client) throws IOException

52
tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java
View File

@ -30,11 +30,7 @@ import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Statement;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.Partition;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.RatioDistribution;
import org.apache.cassandra.stress.generate.Row;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.settings.ValidationType;
import org.apache.cassandra.stress.util.JavaDriverClient;
@ -45,13 +41,16 @@ public class SchemaInsert extends SchemaStatement
{
private final BatchStatement.Type batchType;
private final RatioDistribution selectChance;
public SchemaInsert(Timer timer, PartitionGenerator generator, StressSettings settings, Distribution batchSize, RatioDistribution selectChance, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, BatchStatement.Type batchType)
public SchemaInsert(Timer timer, StressSettings settings, PartitionGenerator generator, SeedManager seedManager, Distribution batchSize, RatioDistribution useRatio, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, BatchStatement.Type batchType)
{
super(timer, generator, settings, batchSize, statement, thriftId, cl, ValidationType.NOT_FAIL);
super(timer, settings, spec(generator, seedManager, batchSize, useRatio), statement, thriftId, cl, ValidationType.NOT_FAIL);
this.batchType = batchType;
this.selectChance = selectChance;
}
private static DataSpec spec(PartitionGenerator generator, SeedManager seedManager, Distribution partitionCount, RatioDistribution useRatio)
{
return new DataSpec(generator, seedManager, partitionCount, useRatio);
}
private class JavaDriverRun extends Runner
@ -65,20 +64,13 @@ public class SchemaInsert extends SchemaStatement
public boolean run() throws Exception
{
Partition.RowIterator[] iterators = new Partition.RowIterator[partitions.size()];
for (int i = 0 ; i < iterators.length ; i++)
iterators[i] = partitions.get(i).iterator(selectChance.next(), true);
List<BoundStatement> stmts = new ArrayList<>();
partitionCount = partitions.size();
for (Partition.RowIterator iterator : iterators)
{
if (iterator.done())
continue;
for (PartitionIterator iterator : partitions)
while (iterator.hasNext())
stmts.add(bindRow(iterator.next()));
for (Row row : iterator.next())
stmts.add(bindRow(row));
}
rowCount += stmts.size();
// 65535 is max number of stmts per batch, so if we have more, we need to manually batch them
@ -107,10 +99,6 @@ public class SchemaInsert extends SchemaStatement
e.printStackTrace();
}
}
for (Partition.RowIterator iterator : iterators)
iterator.markWriteFinished();
return true;
}
}
@ -126,26 +114,14 @@ public class SchemaInsert extends SchemaStatement
public boolean run() throws Exception
{
Partition.RowIterator[] iterators = new Partition.RowIterator[partitions.size()];
for (int i = 0 ; i < iterators.length ; i++)
iterators[i] = partitions.get(i).iterator(selectChance.next(), true);
partitionCount = partitions.size();
for (Partition.RowIterator iterator : iterators)
for (PartitionIterator iterator : partitions)
{
if (iterator.done())
continue;
for (Row row : iterator.next())
while (iterator.hasNext())
{
validate(client.execute_prepared_cql3_query(thriftId, iterator.partition().getToken(), thriftRowArgs(row), settings.command.consistencyLevel));
validate(client.execute_prepared_cql3_query(thriftId, iterator.getToken(), thriftRowArgs(iterator.next()), settings.command.consistencyLevel));
rowCount += 1;
}
}
for (Partition.RowIterator iterator : iterators)
iterator.markWriteFinished();
return true;
}
}

55
tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java
View File

@ -31,10 +31,7 @@ import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.stress.generate.Partition;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.Row;
import org.apache.cassandra.stress.settings.OptionDistribution;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.settings.ValidationType;
import org.apache.cassandra.stress.util.JavaDriverClient;
@ -45,7 +42,6 @@ import org.apache.cassandra.thrift.ThriftConversion;
public class SchemaQuery extends SchemaStatement
{
public static enum ArgSelect
{
MULTIROW, SAMEROW;
@ -56,13 +52,18 @@ public class SchemaQuery extends SchemaStatement
final Object[][] randomBuffer;
final Random random = new Random();
public SchemaQuery(Timer timer, PartitionGenerator generator, StressSettings settings, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, ValidationType validationType, ArgSelect argSelect)
public SchemaQuery(Timer timer, StressSettings settings, PartitionGenerator generator, SeedManager seedManager, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, ValidationType validationType, ArgSelect argSelect)
{
super(timer, generator, settings, OptionDistribution.get("fixed(1)").get(), statement, thriftId, cl, validationType);
super(timer, settings, spec(generator, seedManager, statement.getVariables().size(), argSelect), statement, thriftId, cl, validationType);
this.argSelect = argSelect;
randomBuffer = new Object[argumentIndex.length][argumentIndex.length];
}
static DataSpec spec(PartitionGenerator generator, SeedManager seedManager, int argCount, ArgSelect argSelect)
{
return new DataSpec(generator, seedManager, new DistributionFixed(1), argSelect == ArgSelect.MULTIROW ? argCount : 1);
}
private class JavaDriverRun extends Runner
{
final JavaDriverClient client;
@ -74,7 +75,7 @@ public class SchemaQuery extends SchemaStatement
public boolean run() throws Exception
{
ResultSet rs = client.getSession().execute(bindArgs(partitions.get(0)));
ResultSet rs = client.getSession().execute(bindArgs());
validate(rs);
rowCount = rs.all().size();
partitionCount = Math.min(1, rowCount);
@ -93,7 +94,7 @@ public class SchemaQuery extends SchemaStatement
public boolean run() throws Exception
{
CqlResult rs = client.execute_prepared_cql3_query(thriftId, partitions.get(0).getToken(), thriftArgs(partitions.get(0)), ThriftConversion.toThrift(cl));
CqlResult rs = client.execute_prepared_cql3_query(thriftId, partitions.get(0).getToken(), thriftArgs(), ThriftConversion.toThrift(cl));
validate(rs);
rowCount = rs.getRowsSize();
partitionCount = Math.min(1, rowCount);
@ -101,29 +102,29 @@ public class SchemaQuery extends SchemaStatement
}
}
private int fillRandom(Partition partition)
private int fillRandom()
{
int c = 0;
while (c == 0)
PartitionIterator iterator = partitions.get(0);
while (iterator.hasNext())
{
for (Row row : partition.iterator(randomBuffer.length, false).next())
{
Object[] randomRow = randomBuffer[c++];
for (int i = 0 ; i < argumentIndex.length ; i++)
randomRow[i] = row.get(argumentIndex[i]);
if (c >= randomBuffer.length)
break;
}
Row row = iterator.next();
Object[] randomBufferRow = randomBuffer[c++];
for (int i = 0 ; i < argumentIndex.length ; i++)
randomBufferRow[i] = row.get(argumentIndex[i]);
if (c >= randomBuffer.length)
break;
}
assert c > 0;
return c;
}
BoundStatement bindArgs(Partition partition)
BoundStatement bindArgs()
{
switch (argSelect)
{
case MULTIROW:
int c = fillRandom(partition);
int c = fillRandom();
for (int i = 0 ; i < argumentIndex.length ; i++)
{
int argIndex = argumentIndex[i];
@ -131,29 +132,27 @@ public class SchemaQuery extends SchemaStatement
}
return statement.bind(bindBuffer);
case SAMEROW:
for (Row row : partition.iterator(1, false).next())
return bindRow(row);
return bindRow(partitions.get(0).next());
default:
throw new IllegalStateException();
}
}
List<ByteBuffer> thriftArgs(Partition partition)
List<ByteBuffer> thriftArgs()
{
switch (argSelect)
{
case MULTIROW:
List<ByteBuffer> args = new ArrayList<>();
int c = fillRandom(partition);
int c = fillRandom();
for (int i = 0 ; i < argumentIndex.length ; i++)
{
int argIndex = argumentIndex[i];
args.add(generator.convert(argIndex, randomBuffer[argIndex < 0 ? 0 : random.nextInt(c)][i]));
args.add(spec.partitionGenerator.convert(argIndex, randomBuffer[argIndex < 0 ? 0 : random.nextInt(c)][i]));
}
return args;
case SAMEROW:
for (Row row : partition.iterator(1, false).next())
return thriftRowArgs(row);
return thriftRowArgs(partitions.get(0).next());
default:
throw new IllegalStateException();
}

16
tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java
View File

@ -25,7 +25,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ColumnDefinitions;
@ -33,9 +32,6 @@ import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.Partition;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.Row;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.settings.ValidationType;
@ -46,7 +42,6 @@ import org.apache.cassandra.transport.SimpleClient;
public abstract class SchemaStatement extends Operation
{
final PartitionGenerator generator;
final PreparedStatement statement;
final Integer thriftId;
final ConsistencyLevel cl;
@ -54,11 +49,10 @@ public abstract class SchemaStatement extends Operation
final int[] argumentIndex;
final Object[] bindBuffer;
public SchemaStatement(Timer timer, PartitionGenerator generator, StressSettings settings, Distribution partitionCount,
public SchemaStatement(Timer timer, StressSettings settings, DataSpec spec,
PreparedStatement statement, Integer thriftId, ConsistencyLevel cl, ValidationType validationType)
{
super(timer, generator, settings, partitionCount);
this.generator = generator;
super(timer, settings, spec);
this.statement = statement;
this.thriftId = thriftId;
this.cl = cl;
@ -67,7 +61,7 @@ public abstract class SchemaStatement extends Operation
bindBuffer = new Object[argumentIndex.length];
int i = 0;
for (ColumnDefinitions.Definition definition : statement.getVariables())
argumentIndex[i++] = generator.indexOf(definition.getName());
argumentIndex[i++] = spec.partitionGenerator.indexOf(definition.getName());
}
BoundStatement bindRow(Row row)
@ -75,7 +69,7 @@ public abstract class SchemaStatement extends Operation
for (int i = 0 ; i < argumentIndex.length ; i++)
{
bindBuffer[i] = row.get(argumentIndex[i]);
if (bindBuffer[i] == null && !generator.permitNulls(argumentIndex[i]))
if (bindBuffer[i] == null && !spec.partitionGenerator.permitNulls(argumentIndex[i]))
throw new IllegalStateException();
}
return statement.bind(bindBuffer);
@ -85,7 +79,7 @@ public abstract class SchemaStatement extends Operation
{
List<ByteBuffer> args = new ArrayList<>();
for (int i : argumentIndex)
args.add(generator.convert(i, row.get(i)));
args.add(spec.partitionGenerator.convert(i, row.get(i)));
return args;
}

5
tools/stress/src/org/apache/cassandra/stress/settings/Command.java
View File

@ -21,7 +21,10 @@ package org.apache.cassandra.stress.settings;
*/
import java.util.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.google.common.collect.ImmutableList;

8
tools/stress/src/org/apache/cassandra/stress/settings/OptionAnyProbabilities.java
View File

@ -21,13 +21,7 @@ package org.apache.cassandra.stress.settings;
*/
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.*;
public final class OptionAnyProbabilities extends OptionMulti

4
tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java
View File

@ -26,14 +26,14 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import com.google.common.base.Function;
import org.apache.cassandra.stress.generate.*;
import org.apache.commons.math3.distribution.ExponentialDistribution;
import org.apache.commons.math3.distribution.NormalDistribution;
import org.apache.commons.math3.distribution.UniformRealDistribution;
import org.apache.commons.math3.distribution.WeibullDistribution;
import org.apache.commons.math3.random.JDKRandomGenerator;
import org.apache.cassandra.stress.generate.*;
/**
* For selecting a mathematical distribution
*/

2
tools/stress/src/org/apache/cassandra/stress/settings/OptionEnumProbabilities.java
View File

@ -22,9 +22,9 @@ package org.apache.cassandra.stress.settings;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
public final class OptionEnumProbabilities<T> extends OptionMulti

7
tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java
View File

@ -21,12 +21,7 @@ package org.apache.cassandra.stress.settings;
*/
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

13
tools/stress/src/org/apache/cassandra/stress/settings/OptionRatioDistribution.java
View File

@ -21,27 +21,14 @@ package org.apache.cassandra.stress.settings;
*/
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import com.google.common.base.Function;
import org.apache.commons.math3.distribution.ExponentialDistribution;
import org.apache.commons.math3.distribution.NormalDistribution;
import org.apache.commons.math3.distribution.UniformRealDistribution;
import org.apache.commons.math3.distribution.WeibullDistribution;
import org.apache.commons.math3.random.JDKRandomGenerator;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.DistributionBoundApache;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.DistributionFixed;
import org.apache.cassandra.stress.generate.DistributionInverted;
import org.apache.cassandra.stress.generate.DistributionOffsetApache;
import org.apache.cassandra.stress.generate.RatioDistribution;
import org.apache.cassandra.stress.generate.RatioDistributionFactory;

12
tools/stress/src/org/apache/cassandra/stress/settings/SettingsColumn.java
View File

@ -22,19 +22,17 @@ package org.apache.cassandra.stress.settings;
import java.io.IOException;
import java.io.Serializable;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.*;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.DistributionFixed;
import org.apache.cassandra.utils.ByteBufferUtil;
/**

1
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommand.java
View File

@ -27,7 +27,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.operations.OpDistributionFactory;
import org.apache.cassandra.thrift.ConsistencyLevel;

7
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java
View File

@ -37,7 +37,6 @@ import org.apache.cassandra.stress.operations.FixedOpDistribution;
import org.apache.cassandra.stress.operations.OpDistribution;
import org.apache.cassandra.stress.operations.OpDistributionFactory;
import org.apache.cassandra.stress.operations.predefined.PredefinedOperation;
import org.apache.cassandra.stress.settings.SettingsCommandPreDefinedMixed.Options;
import org.apache.cassandra.stress.util.Timer;
// Settings unique to the mixed command type
@ -54,7 +53,7 @@ public class SettingsCommandPreDefined extends SettingsCommand
{
public OpDistribution get(Timer timer)
{
return new FixedOpDistribution(PredefinedOperation.operation(type, timer, newGenerator(settings, seeds), settings, add));
return new FixedOpDistribution(PredefinedOperation.operation(type, timer, newGenerator(settings), seeds, settings, add));
}
public String desc()
@ -69,7 +68,7 @@ public class SettingsCommandPreDefined extends SettingsCommand
};
}
PartitionGenerator newGenerator(StressSettings settings, SeedManager seeds)
PartitionGenerator newGenerator(StressSettings settings)
{
List<String> names = settings.columns.namestrs;
List<Generator> partitionKey = Collections.<Generator>singletonList(new HexBytes("key",
@ -79,7 +78,7 @@ public class SettingsCommandPreDefined extends SettingsCommand
List<Generator> columns = new ArrayList<>();
for (int i = 0 ; i < settings.columns.maxColumnsPerKey ; i++)
columns.add(new Bytes(names.get(i), new GeneratorConfig("randomstr" + names.get(i), null, settings.columns.sizeDistribution, null)));
return new PartitionGenerator(partitionKey, Collections.<Generator>emptyList(), columns, PartitionGenerator.Order.ARBITRARY, seeds);
return new PartitionGenerator(partitionKey, Collections.<Generator>emptyList(), columns, PartitionGenerator.Order.ARBITRARY);
}
public SettingsCommandPreDefined(Command type, Options options)

6
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java
View File

@ -35,8 +35,6 @@ import org.apache.cassandra.stress.operations.SampledOpDistributionFactory;
import org.apache.cassandra.stress.operations.predefined.PredefinedOperation;
import org.apache.cassandra.stress.util.Timer;
import org.apache.commons.math3.util.Pair;
// Settings unique to the mixed command type
public class SettingsCommandPreDefinedMixed extends SettingsCommandPreDefined
{
@ -62,12 +60,12 @@ public class SettingsCommandPreDefinedMixed extends SettingsCommandPreDefined
{
protected Operation get(Timer timer, PartitionGenerator generator, Command key)
{
return PredefinedOperation.operation(key, timer, generator, settings, add);
return PredefinedOperation.operation(key, timer, generator, seeds, settings, add);
}
protected PartitionGenerator newGenerator()
{
return SettingsCommandPreDefinedMixed.this.newGenerator(settings, seeds);
return SettingsCommandPreDefinedMixed.this.newGenerator(settings);
}
};
}

14
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java
View File

@ -23,16 +23,10 @@ package org.apache.cassandra.stress.settings;
import java.io.File;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import com.google.common.collect.ImmutableList;
import com.datastax.driver.core.BatchStatement;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.StressProfile;
import org.apache.cassandra.stress.generate.DistributionFactory;
@ -42,8 +36,6 @@ import org.apache.cassandra.stress.operations.OpDistributionFactory;
import org.apache.cassandra.stress.operations.SampledOpDistributionFactory;
import org.apache.cassandra.stress.util.Timer;
import org.apache.commons.math3.util.Pair;
// Settings unique to the mixed command type
public class SettingsCommandUser extends SettingsCommand
{
@ -81,13 +73,13 @@ public class SettingsCommandUser extends SettingsCommand
protected Operation get(Timer timer, PartitionGenerator generator, String key)
{
if (key.equalsIgnoreCase("insert"))
return profile.getInsert(timer, generator, settings);
return profile.getQuery(key, timer, generator, settings);
return profile.getInsert(timer, generator, seeds, settings);
return profile.getQuery(key, timer, generator, seeds, settings);
}
protected PartitionGenerator newGenerator()
{
return profile.newGenerator(settings, seeds);
return profile.newGenerator(settings);
}
};
}

3
tools/stress/src/org/apache/cassandra/stress/settings/SettingsErrors.java
View File

@ -21,9 +21,6 @@ package org.apache.cassandra.stress.settings;
*/
import java.io.File;
import java.io.FileNotFoundException;
import java.io.PrintStream;
import java.io.Serializable;
import java.util.Arrays;
import java.util.List;

7
tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java
View File

@ -25,12 +25,7 @@ import java.io.*;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.*;
import com.datastax.driver.core.Host;

1
tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java
View File

@ -25,7 +25,6 @@ import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.utils.ByteBufferUtil;
public class SettingsSchema implements Serializable

3
tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java
View File

@ -25,9 +25,6 @@ import java.io.Serializable;
import java.util.*;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
import com.datastax.driver.core.policies.RoundRobinPolicy;
import com.datastax.driver.core.policies.WhiteListPolicy;
import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.stress.util.JavaDriverClient;
import org.apache.cassandra.stress.util.SimpleThriftClient;

15
tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java
View File

@ -41,7 +41,7 @@ public class DynamicList<E>
private final int[] size;
// TODO: alternate links to save space
private final Node<E>[] links;
private final E value;
private E value;
private Node(int height, E value)
{
@ -96,7 +96,7 @@ public class DynamicList<E>
public DynamicList(int maxExpectedSize)
{
this.maxHeight = 3 + (int) Math.ceil(Math.log(maxExpectedSize) / Math.log(2));
this.maxHeight = 3 + Math.max(0, (int) Math.ceil(Math.log(maxExpectedSize) / Math.log(2)));
head = new Node<>(maxHeight, null);
}
@ -105,15 +105,22 @@ public class DynamicList<E>
return 1 + Integer.bitCount(ThreadLocalRandom.current().nextInt() & ((1 << (maxHeight - 1)) - 1));
}
public Node<E> append(E value)
{
return append(value, Integer.MAX_VALUE);
}
// add the value to the end of the list, and return the associated Node that permits efficient removal
// regardless of its future position in the list from other modifications
public Node<E> append(E value)
public Node<E> append(E value, int maxSize)
{
Node<E> newTail = new Node<>(randomLevel(), value);
lock.writeLock().lock();
try
{
if (size >= maxSize)
return null;
size++;
Node<E> tail = head;
@ -146,6 +153,8 @@ public class DynamicList<E>
public void remove(Node<E> node)
{
lock.writeLock().lock();
assert node.value != null;
node.value = null;
try
{
size--;

4
tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
View File

@ -25,10 +25,10 @@ import javax.net.ssl.SSLContext;
import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
import com.datastax.driver.core.policies.WhiteListPolicy;
import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.security.SSLFactory;
import io.netty.util.internal.logging.InternalLoggerFactory;
import io.netty.util.internal.logging.Slf4JLoggerFactory;
import org.apache.cassandra.config.EncryptionOptions;
import org.apache.cassandra.security.SSLFactory;
import org.apache.cassandra.stress.settings.StressSettings;
public class JavaDriverClient

3
tools/stress/src/org/apache/cassandra/stress/util/SmartThriftClient.java
View File

@ -29,9 +29,10 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.collect.Iterators;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.google.common.collect.Iterators;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.ByteBufferUtil;

1
tools/stress/src/org/apache/cassandra/stress/util/Timer.java
View File

@ -23,7 +23,6 @@ package org.apache.cassandra.stress.util;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;

1
tools/stress/src/org/apache/cassandra/stress/util/Timing.java
View File

@ -24,7 +24,6 @@ package org.apache.cassandra.stress.util;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;

1
tools/stress/src/org/apache/cassandra/stress/util/TimingInterval.java
View File

@ -24,7 +24,6 @@ package org.apache.cassandra.stress.util;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
// represents measurements taken over an interval of time