From 6c579a0102fa3e67215fef5d9f8aa97191e3a216 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Fri, 12 Dec 2014 14:09:37 +0000 Subject: [PATCH] cassandra-stress simultaneous inserts over same seed patch by benedict; reviewed by rstupp CASSANDRA-7964 --- CHANGES.txt | 1 + .../apache/cassandra/stress/Operation.java | 86 ++- .../apache/cassandra/stress/StressAction.java | 96 +-- .../cassandra/stress/StressMetrics.java | 1 - .../cassandra/stress/StressProfile.java | 65 +- .../apache/cassandra/stress/StressServer.java | 3 +- .../cassandra/stress/generate/Partition.java | 554 ------------------ .../stress/generate/PartitionGenerator.java | 39 +- .../cassandra/stress/generate/Seed.java | 62 +- .../stress/generate/SeedManager.java | 66 +-- .../stress/generate/values/Bytes.java | 7 +- .../generate/values/GeneratorConfig.java | 9 +- .../stress/generate/values/Strings.java | 2 - .../stress/generate/values/TimeUUIDs.java | 4 +- .../operations/FixedOpDistribution.java | 7 - .../stress/operations/OpDistribution.java | 1 - .../operations/SampledOpDistribution.java | 9 - .../predefined/CqlCounterAdder.java | 5 +- .../predefined/CqlCounterGetter.java | 5 +- .../operations/predefined/CqlInserter.java | 6 +- .../operations/predefined/CqlOperation.java | 32 +- .../operations/predefined/CqlReader.java | 7 +- .../predefined/PredefinedOperation.java | 40 +- .../predefined/ThriftCounterAdder.java | 9 +- .../predefined/ThriftCounterGetter.java | 5 +- .../operations/predefined/ThriftInserter.java | 11 +- .../operations/predefined/ThriftReader.java | 7 +- .../operations/userdefined/SchemaInsert.java | 52 +- .../operations/userdefined/SchemaQuery.java | 55 +- .../userdefined/SchemaStatement.java | 16 +- .../cassandra/stress/settings/Command.java | 5 +- .../settings/OptionAnyProbabilities.java | 8 +- .../stress/settings/OptionDistribution.java | 4 +- .../settings/OptionEnumProbabilities.java | 2 +- .../stress/settings/OptionMulti.java | 7 +- .../settings/OptionRatioDistribution.java | 13 - .../stress/settings/SettingsColumn.java | 12 +- .../stress/settings/SettingsCommand.java | 1 - .../settings/SettingsCommandPreDefined.java | 7 +- .../SettingsCommandPreDefinedMixed.java | 6 +- .../stress/settings/SettingsCommandUser.java | 14 +- .../stress/settings/SettingsErrors.java | 3 - .../stress/settings/SettingsNode.java | 7 +- .../stress/settings/SettingsSchema.java | 1 - .../stress/settings/StressSettings.java | 3 - .../cassandra/stress/util/DynamicList.java | 15 +- .../stress/util/JavaDriverClient.java | 4 +- .../stress/util/SmartThriftClient.java | 3 +- .../apache/cassandra/stress/util/Timer.java | 1 - .../apache/cassandra/stress/util/Timing.java | 1 - .../cassandra/stress/util/TimingInterval.java | 1 - 51 files changed, 327 insertions(+), 1053 deletions(-) delete mode 100644 tools/stress/src/org/apache/cassandra/stress/generate/Partition.java diff --git a/CHANGES.txt b/CHANGES.txt index fa42e8549b..25140a8a91 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 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) diff --git a/tools/stress/src/org/apache/cassandra/stress/Operation.java b/tools/stress/src/org/apache/cassandra/stress/Operation.java index 5560240159..edf3a5426a 100644 --- a/tools/stress/src/org/apache/cassandra/stress/Operation.java +++ b/tools/stress/src/org/apache/cassandra/stress/Operation.java @@ -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 partitions; + private final List partitionCache = new ArrayList<>(); + protected List 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 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 keys = new ArrayList<>(); - for (Partition partition : partitions) + for (PartitionIterator partition : partitions) keys.add(partition.getKeyAsString()); return keys.toString(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java index 68e00041ae..1433742a79 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressAction.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressAction.java @@ -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; - } - - } } diff --git a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java b/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java index 3a4a4a382e..d1cc0d4f6b 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java @@ -31,7 +31,6 @@ import java.util.concurrent.ThreadFactory; 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; diff --git a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java index 76642be12f..1517fcb52f 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java @@ -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 select(E first, String key, String defValue, Map map, Function 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 get(List columnInfos) diff --git a/tools/stress/src/org/apache/cassandra/stress/StressServer.java b/tools/stress/src/org/apache/cassandra/stress/StressServer.java index 3c9e2a669c..a6dfaf4501 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressServer.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressServer.java @@ -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(); diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/Partition.java b/tools/stress/src/org/apache/cassandra/stress/generate/Partition.java deleted file mode 100644 index 66f8c1d71d..0000000000 --- a/tools/stress/src/org/apache/cassandra/stress/generate/Partition.java +++ /dev/null @@ -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 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 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[] 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 unique = new HashSet<>(); - final List 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 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 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() - { - public Iterator iterator() - { - return new Iterator() - { - - 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 - { - 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]); - } - -} diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java b/tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java index 128d2f51b2..9f88068dc9 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java @@ -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 partitionKey; final List clusteringComponents; final List valueComponents; - final int[] clusteringChildAverages; + final int[] clusteringDescendantAverages; + final int[] clusteringComponentAverages; private final Map indexMap; final Order order; - final SeedManager seeds; - final List recyclable = new ArrayList<>(); - int partitionsInUse = 0; - - public void reset() - { - partitionsInUse = 0; - } - - public PartitionGenerator(List partitionKey, List clusteringComponents, List valueComponents, Order order, SeedManager seeds) + public PartitionGenerator(List partitionKey, List clusteringComponents, List 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) diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java b/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java index f427608f31..9e2e65bd57 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java @@ -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 { + 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 stateUpdater = AtomicReferenceFieldUpdater.newUpdater(Seed.class, State.class, "state"); + private static final AtomicIntegerFieldUpdater 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 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 sampleFrom) + { + sampleFrom.remove(poolNode); + } } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java b/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java index dba721dd44..071d888b9d 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java @@ -33,9 +33,12 @@ public class SeedManager final Distribution visits; final Generator writes; final Generator reads; - final ConcurrentHashMap managing = new ConcurrentHashMap<>(); + final ConcurrentHashMap managing = new ConcurrentHashMap<>(); final DynamicList 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 diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java index 358163c10a..3c15c8790d 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java @@ -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 { diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java index 8f7b2eaaf6..3522338b87 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java index 71aaae6019..b58fee28c4 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java index efe4b79c38..7bfabf5056 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java @@ -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 { final Dates dateGen; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java index 3212795f56..533b6308a4 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java @@ -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(); - } - } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java index bcbd0bfeaf..0fc15a67f2 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java @@ -27,6 +27,5 @@ public interface OpDistribution { Operation next(); - public int maxBatchSize(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java index 0bd64c5e98..432e9913a0 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java @@ -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 pair : operations.getPmf()) - max = Math.max(max, (int) pair.getFirst().partitionCount.maxValue()); - return max; - } - public Operation next() { while (remaining == 0) diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java index b7d1ee78b7..456c821828 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java @@ -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 { 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(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java index 94c8faf097..8c1c65c327 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java @@ -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 { - 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 diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java index 622eb14c45..88ee752a24 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java @@ -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 { - 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 diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java index 0264cd1cb0..9cea854ba3 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java @@ -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 extends PredefinedOperation protected abstract String buildQuery(); protected abstract CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List 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 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 - { - - protected CqlRunOpFetchKeys(ClientWrapper client, String query, Object queryId, List 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 { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java index 3a7f75ab81..12cc241201 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java @@ -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 { - 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 diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java index dba2e51939..d5c3edcfee 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java @@ -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; @@ -174,7 +170,7 @@ public abstract class PredefinedOperation extends Operation protected List 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) @@ -186,7 +182,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) { @@ -194,10 +190,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(); } @@ -207,10 +203,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(); } @@ -220,10 +216,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(); } @@ -232,10 +228,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(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java index 4ee42e9f8a..be34a0723a 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java @@ -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(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java index 10c6aabcef..ca81fe9b11 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java @@ -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 diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java index d6adbf92e2..1827c06b98 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java @@ -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() diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java index 276d8c5b41..d77dc6a81d 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java @@ -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 diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java index 8e20ab3780..61237f1cd7 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java @@ -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 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; } } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java index 866f6aba17..72db2f71a2 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java @@ -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 thriftArgs(Partition partition) + List thriftArgs() { switch (argSelect) { case MULTIROW: List 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(); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java index 1f7ed80eb9..75e49b48b4 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java @@ -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 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; } diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/Command.java b/tools/stress/src/org/apache/cassandra/stress/settings/Command.java index 7138cbbaf2..15729460b7 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/Command.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/Command.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionAnyProbabilities.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionAnyProbabilities.java index b685294707..9c2f3673be 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionAnyProbabilities.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionAnyProbabilities.java @@ -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 diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java index ef3dbb1cee..45e832a0ab 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java @@ -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 */ diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionEnumProbabilities.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionEnumProbabilities.java index 71d7a8d915..c218176469 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionEnumProbabilities.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionEnumProbabilities.java @@ -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 extends OptionMulti diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java index 32bfc652cf..6d11012789 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionRatioDistribution.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionRatioDistribution.java index 756536fb85..416f045d61 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionRatioDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionRatioDistribution.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsColumn.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsColumn.java index ecdef29572..bf78bf9277 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsColumn.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsColumn.java @@ -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; /** diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommand.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommand.java index 885054444d..8751dbff37 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommand.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommand.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java index 8583e90de0..ee1958b7ca 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java @@ -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 names = settings.columns.namestrs; List partitionKey = Collections.singletonList(new HexBytes("key", @@ -79,7 +78,7 @@ public class SettingsCommandPreDefined extends SettingsCommand List 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.emptyList(), columns, PartitionGenerator.Order.ARBITRARY, seeds); + return new PartitionGenerator(partitionKey, Collections.emptyList(), columns, PartitionGenerator.Order.ARBITRARY); } public SettingsCommandPreDefined(Command type, Options options) diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java index 6beb55f521..3a1d552f19 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java @@ -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); } }; } diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java index 4e2997fa37..9841433696 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java @@ -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); } }; } diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsErrors.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsErrors.java index 625f8030b1..66daac49b4 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsErrors.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsErrors.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java index 5297da8026..ba1fcb5765 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java index 6e3a02e0f0..342dd0be32 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java @@ -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 diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java b/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java index 815fe9b2a5..219e7cdd46 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java b/tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java index e5aaf44c56..e9d43a36e7 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java +++ b/tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java @@ -41,7 +41,7 @@ public class DynamicList private final int[] size; // TODO: alternate links to save space private final Node[] links; - private final E value; + private E value; private Node(int height, E value) { @@ -96,7 +96,7 @@ public class DynamicList 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 return 1 + Integer.bitCount(ThreadLocalRandom.current().nextInt() & ((1 << (maxHeight - 1)) - 1)); } + public Node 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 append(E value) + public Node append(E value, int maxSize) { Node newTail = new Node<>(randomLevel(), value); lock.writeLock().lock(); try { + if (size >= maxSize) + return null; size++; Node tail = head; @@ -146,6 +153,8 @@ public class DynamicList public void remove(Node node) { lock.writeLock().lock(); + assert node.value != null; + node.value = null; try { size--; diff --git a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java b/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java index 2105179741..295ed25a8b 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java +++ b/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java @@ -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 diff --git a/tools/stress/src/org/apache/cassandra/stress/util/SmartThriftClient.java b/tools/stress/src/org/apache/cassandra/stress/util/SmartThriftClient.java index 6367936bd2..178da64268 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/SmartThriftClient.java +++ b/tools/stress/src/org/apache/cassandra/stress/util/SmartThriftClient.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/util/Timer.java b/tools/stress/src/org/apache/cassandra/stress/util/Timer.java index 7449a0021c..ff625a8554 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/Timer.java +++ b/tools/stress/src/org/apache/cassandra/stress/util/Timer.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/util/Timing.java b/tools/stress/src/org/apache/cassandra/stress/util/Timing.java index f48ce350b4..9464b194f7 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/Timing.java +++ b/tools/stress/src/org/apache/cassandra/stress/util/Timing.java @@ -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; diff --git a/tools/stress/src/org/apache/cassandra/stress/util/TimingInterval.java b/tools/stress/src/org/apache/cassandra/stress/util/TimingInterval.java index 11d313f5de..065ea52e93 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/TimingInterval.java +++ b/tools/stress/src/org/apache/cassandra/stress/util/TimingInterval.java @@ -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