Improve stress workload realism patch by benedict; reviewed by tjake for CASSANDRA-7519
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/0580fb2b Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/0580fb2b Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/0580fb2b Branch: refs/heads/cassandra-2.1.0 Commit: 0580fb2b7707beaa69019a73a6c53d86fe088a0a Parents: c6a2c65 Author: Benedict Elliott Smith <bened...@apache.org> Authored: Sun Sep 7 21:18:53 2014 +0700 Committer: Benedict Elliott Smith <bened...@apache.org> Committed: Sun Sep 7 21:19:58 2014 +0700 ---------------------------------------------------------------------- CHANGES.txt | 2 +- tools/cqlstress-counter-example.yaml | 20 +- tools/cqlstress-example.yaml | 25 +- tools/cqlstress-insanity-example.yaml | 20 +- .../org/apache/cassandra/stress/Operation.java | 9 +- .../apache/cassandra/stress/StressAction.java | 169 +++------- .../apache/cassandra/stress/StressMetrics.java | 26 +- .../apache/cassandra/stress/StressProfile.java | 75 ++++- .../org/apache/cassandra/stress/StressYaml.java | 12 +- .../stress/generate/DistributionInverted.java | 7 + .../stress/generate/DistributionQuantized.java | 90 +++++ .../cassandra/stress/generate/FasterRandom.java | 116 +++++++ .../cassandra/stress/generate/Partition.java | 327 +++++++++++++++---- .../stress/generate/PartitionGenerator.java | 28 +- .../stress/generate/RatioDistribution.java | 5 + .../apache/cassandra/stress/generate/Seed.java | 67 ++++ .../stress/generate/SeedGenerator.java | 29 -- .../cassandra/stress/generate/SeedManager.java | 249 ++++++++++++++ .../stress/generate/SeedRandomGenerator.java | 54 --- .../stress/generate/SeedSeriesGenerator.java | 42 --- .../stress/generate/values/Booleans.java | 2 +- .../cassandra/stress/generate/values/Bytes.java | 9 +- .../cassandra/stress/generate/values/Dates.java | 3 +- .../stress/generate/values/Doubles.java | 2 +- .../stress/generate/values/Floats.java | 2 +- .../stress/generate/values/Generator.java | 4 +- .../stress/generate/values/HexBytes.java | 2 +- .../stress/generate/values/HexStrings.java | 4 +- .../cassandra/stress/generate/values/Inets.java | 2 +- .../stress/generate/values/Integers.java | 2 +- .../cassandra/stress/generate/values/Lists.java | 2 +- .../cassandra/stress/generate/values/Longs.java | 2 +- .../cassandra/stress/generate/values/Sets.java | 2 +- .../stress/generate/values/Strings.java | 12 +- .../stress/generate/values/TimeUUIDs.java | 2 +- .../cassandra/stress/generate/values/UUIDs.java | 2 +- .../operations/predefined/CqlCounterAdder.java | 5 + .../operations/predefined/CqlInserter.java | 5 + .../predefined/PredefinedOperation.java | 2 +- .../predefined/ThriftCounterAdder.java | 5 + .../operations/predefined/ThriftInserter.java | 5 + .../operations/userdefined/SchemaInsert.java | 80 +++-- .../operations/userdefined/SchemaQuery.java | 87 ++++- .../operations/userdefined/SchemaStatement.java | 53 +-- .../cassandra/stress/settings/CliOption.java | 3 +- .../stress/settings/OptionDistribution.java | 72 +++- .../settings/OptionRatioDistribution.java | 40 ++- .../stress/settings/SettingsCommand.java | 14 +- .../settings/SettingsCommandPreDefined.java | 13 +- .../SettingsCommandPreDefinedMixed.java | 4 +- .../stress/settings/SettingsCommandUser.java | 16 +- .../stress/settings/SettingsErrors.java | 92 ++++++ .../stress/settings/SettingsInsert.java | 103 ++++++ .../cassandra/stress/settings/SettingsKey.java | 153 --------- .../stress/settings/SettingsPopulation.java | 176 ++++++++++ .../stress/settings/SettingsSchema.java | 17 +- .../stress/settings/StressSettings.java | 23 +- .../cassandra/stress/util/DynamicList.java | 259 +++++++++++++++ .../org/apache/cassandra/stress/util/Timer.java | 7 +- .../apache/cassandra/stress/util/Timing.java | 13 +- .../cassandra/stress/util/TimingInterval.java | 6 +- 61 files changed, 1955 insertions(+), 724 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 46836bf..e42d9c4 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -5,7 +5,7 @@ * cqlsh: DESCRIBE support for frozen UDTs, tuples (CASSANDRA-7863) * Avoid exposing internal classes over JMX (CASSANDRA-7879) * Add null check for keys when freezing collection (CASSANDRA-7869) - + * Improve stress workload realism (CASSANDRA-7519) 2.1.0-rc7 * Add frozen keyword and require UDT to be frozen (CASSANDRA-7857) http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/cqlstress-counter-example.yaml ---------------------------------------------------------------------- diff --git a/tools/cqlstress-counter-example.yaml b/tools/cqlstress-counter-example.yaml index cff14b6..f8f70ea 100644 --- a/tools/cqlstress-counter-example.yaml +++ b/tools/cqlstress-counter-example.yaml @@ -62,19 +62,17 @@ columnspec: population: fixed(1) insert: - partitions: fixed(1) # number of unique partitions to update in a single operation - # if perbatch < 1, multiple batches will be used but all partitions will - # occur in all batches (unless already finished); only the row counts will vary - pervisit: fixed(1)/1 # ratio of rows each partition should update in a single visit to the partition, - # as a proportion of the total possible for the partition - perbatch: fixed(1)/1 # number of rows each partition should update in a single batch statement, - # as a proportion of the proportion we are inserting this visit - # (i.e. compounds with (and capped by) pervisit) - batchtype: UNLOGGED # type of batch to use + partitions: fixed(1) # number of unique partitions to update in a single operation + # if batchcount > 1, multiple batches will be used but all partitions will + # occur in all batches (unless they finish early); only the row counts will vary + batchtype: LOGGED # type of batch to use + select: fixed(1)/1 # uniform chance any single generated CQL row will be visited in a partition; + # generated for each partition independently, each time we visit it # # A list of queries you wish to run against the schema # queries: - simple1: select * from counttest where name = ? - + simple1: + cql: select * from counttest where name = ? + fields: samerow # samerow or multirow (select arguments from the same row, or randomly from all rows in the partition) \ No newline at end of file http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/cqlstress-example.yaml ---------------------------------------------------------------------- diff --git a/tools/cqlstress-example.yaml b/tools/cqlstress-example.yaml index d5c90a2..4dd5e4a 100644 --- a/tools/cqlstress-example.yaml +++ b/tools/cqlstress-example.yaml @@ -69,25 +69,26 @@ columnspec: size: uniform(1..10) population: uniform(1..1M) # the range of unique values to select for the field (default is 100Billion) - name: date - cluster: uniform(1..4) + cluster: uniform(20..40) - name: lval population: gaussian(1..1000) cluster: uniform(1..4) insert: - partitions: uniform(1..50) # number of unique partitions to update in a single operation - # if perbatch < 1, multiple batches will be used but all partitions will - # occur in all batches (unless already finished); only the row counts will vary - pervisit: uniform(1..10)/10 # ratio of rows each partition should update in a single visit to the partition, - # as a proportion of the total possible for the partition - perbatch: ~exp(1..3)/4 # number of rows each partition should update in a single batch statement, - # as a proportion of the proportion we are inserting this visit - # (i.e. compounds with (and capped by) pervisit) - batchtype: UNLOGGED # type of batch to use + partitions: uniform(1..50) # number of unique partitions to update in a single operation + # if batchcount > 1, multiple batches will be used but all partitions will + # occur in all batches (unless they finish early); only the row counts will vary + batchtype: LOGGED # type of batch to use + select: uniform(1..10)/10 # uniform chance any single generated CQL row will be visited in a partition; + # generated for each partition independently, each time we visit it # # A list of queries you wish to run against the schema # queries: - simple1: select * from typestest where name = ? and choice = ? LIMIT 100 - range1: select * from typestest where name = ? and choice = ? and date >= ? LIMIT 100 + simple1: + cql: select * from typestest where name = ? and choice = ? LIMIT 100 + fields: samerow # samerow or multirow (select arguments from the same row, or randomly from all rows in the partition) + range1: + cql: select * from typestest where name = ? and choice = ? and date >= ? LIMIT 100 + fields: multirow # samerow or multirow (select arguments from the same row, or randomly from all rows in the partition) http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/cqlstress-insanity-example.yaml ---------------------------------------------------------------------- diff --git a/tools/cqlstress-insanity-example.yaml b/tools/cqlstress-insanity-example.yaml index ef1bb3a..ea4f97f 100644 --- a/tools/cqlstress-insanity-example.yaml +++ b/tools/cqlstress-insanity-example.yaml @@ -74,19 +74,17 @@ columnspec: insert: - partitions: fixed(1) # number of unique partitions to update in a single operation - # if perbatch < 1, multiple batches will be used but all partitions will - # occur in all batches (unless already finished); only the row counts will vary - pervisit: uniform(1..10)/10 # ratio of rows each partition should update in a single visit to the partition, - # as a proportion of the total possible for the partition - perbatch: fixed(1)/1 # number of rows each partition should update in a single batch statement, - # as a proportion of the proportion we are inserting this visit - # (i.e. compounds with (and capped by) pervisit) - batchtype: UNLOGGED # type of batch to use + partitions: fixed(1) # number of unique partitions to update in a single operation + # if batchcount > 1, multiple batches will be used but all partitions will + # occur in all batches (unless they finish early); only the row counts will vary + batchtype: LOGGED # type of batch to use + select: fixed(1)/1 # uniform chance any single generated CQL row will be visited in a partition; + # generated for each partition independently, each time we visit it # # A list of queries you wish to run against the schema # queries: - simple1: select * from insanitytest where name = ? and choice = ? LIMIT 100 - + simple1: + cql: select * from insanitytest where name = ? and choice = ? LIMIT 100 + fields: samerow # samerow or multirow (select arguments from the same row, or randomly from all rows in the partition) http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/Operation.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/Operation.java b/tools/stress/src/org/apache/cassandra/stress/Operation.java index 7831074..5560240 100644 --- a/tools/stress/src/org/apache/cassandra/stress/Operation.java +++ b/tools/stress/src/org/apache/cassandra/stress/Operation.java @@ -61,6 +61,11 @@ public abstract class Operation this.partitions = partitions; } + public boolean isWrite() + { + return false; + } + /** * Run operation * @param client Cassandra Thrift client connection @@ -84,7 +89,7 @@ public abstract class Operation String exceptionMessage = null; int tries = 0; - for (; tries < settings.command.tries; tries++) + for (; tries < settings.errors.tries; tries++) { try { @@ -144,7 +149,7 @@ public abstract class Operation protected void error(String message) throws IOException { - if (!settings.command.ignoreErrors) + if (!settings.errors.ignore) throw new IOException(message); else if (settings.log.level.compareTo(SettingsLog.Level.MINIMAL) > 0) System.err.println(message); http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/StressAction.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java index 2105a72..e58bfa1 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressAction.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressAction.java @@ -23,7 +23,6 @@ import java.io.PrintStream; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; @@ -32,7 +31,6 @@ 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.generate.SeedGenerator; import org.apache.cassandra.stress.operations.OpDistribution; import org.apache.cassandra.stress.operations.OpDistributionFactory; import org.apache.cassandra.stress.settings.*; @@ -58,6 +56,7 @@ public class StressAction implements Runnable // creating keyspace and column families settings.maybeCreateKeyspaces(); + // TODO: warmup should if (!settings.command.noWarmup) warmup(settings.command.getFactory(settings)); @@ -155,8 +154,8 @@ public class StressAction implements Runnable double improvement = 0; for (int i = results.size() - count ; i < results.size() ; i++) { - double prev = results.get(i - 1).getTiming().getHistory().realOpRate(); - double cur = results.get(i).getTiming().getHistory().realOpRate(); + double prev = results.get(i - 1).getTiming().getHistory().opRate(); + double cur = results.get(i).getTiming().getHistory().opRate(); improvement += (cur - prev) / prev; } return improvement / count; @@ -169,11 +168,11 @@ public class StressAction implements Runnable operations.desc(), threadCount, opCount > 0 ? " for " + opCount + " iterations" : "until stderr of mean < " + settings.command.targetUncertainty)); - final WorkQueue workQueue; + final WorkManager workManager; if (opCount < 0) - workQueue = new ContinuousWorkQueue(50); + workManager = new ContinuousWorkManager(); else - workQueue = FixedWorkQueue.build(opCount); + workManager = new FixedWorkManager(opCount); RateLimiter rateLimiter = null; // TODO : move this to a new queue wrapper that gates progress based on a poisson (or configurable) distribution @@ -185,7 +184,7 @@ public class StressAction implements Runnable final CountDownLatch done = new CountDownLatch(threadCount); final Consumer[] consumers = new Consumer[threadCount]; for (int i = 0; i < threadCount; i++) - consumers[i] = new Consumer(operations, done, workQueue, metrics, rateLimiter); + consumers[i] = new Consumer(operations, done, workManager, metrics, rateLimiter); // starting worker threadCount for (int i = 0; i < threadCount; i++) @@ -201,14 +200,15 @@ public class StressAction implements Runnable settings.command.minimumUncertaintyMeasurements, settings.command.maximumUncertaintyMeasurements); } catch (InterruptedException e) { } - workQueue.stop(); + workManager.stop(); } try { done.await(); metrics.stop(); - } catch (InterruptedException e) {} + } + catch (InterruptedException e) {} if (metrics.wasCancelled()) return null; @@ -231,20 +231,18 @@ public class StressAction implements Runnable private final OpDistribution operations; private final StressMetrics metrics; private final Timer timer; - private final SeedGenerator seedGenerator; private final RateLimiter rateLimiter; private volatile boolean success = true; - private final WorkQueue workQueue; + private final WorkManager workManager; private final CountDownLatch done; - public Consumer(OpDistributionFactory operations, CountDownLatch done, WorkQueue workQueue, StressMetrics metrics, RateLimiter rateLimiter) + public Consumer(OpDistributionFactory operations, CountDownLatch done, WorkManager workManager, StressMetrics metrics, RateLimiter rateLimiter) { this.done = done; this.rateLimiter = rateLimiter; - this.workQueue = workQueue; + this.workManager = workManager; this.metrics = metrics; this.timer = metrics.getTiming().newTimer(); - this.seedGenerator = settings.keys.newSeedGenerator(); this.operations = operations.get(timer); } @@ -275,42 +273,33 @@ public class StressAction implements Runnable } int maxBatchSize = operations.maxBatchSize(); - Work work = workQueue.poll(); Partition[] partitions = new Partition[maxBatchSize]; - int workDone = 0; - while (work != null) + while (true) { + // TODO: Operation should be able to ecapsulate much of this behaviour Operation op = operations.next(); op.generator.reset(); - int batchSize = Math.max(1, (int) op.partitionCount.next()); - int partitionCount = 0; + int batchSize = workManager.takePermits(Math.max(1, (int) op.partitionCount.next())); + if (batchSize < 0) + break; + + if (rateLimiter != null) + rateLimiter.acquire(batchSize); + + int partitionCount = 0; while (partitionCount < batchSize) { - int count = Math.min((work.count - workDone), batchSize - partitionCount); - for (int i = 0 ; i < count ; i++) - { - long seed = seedGenerator.next(work.offset + workDone + i); - partitions[partitionCount + i] = op.generator.generate(seed); - } - workDone += count; - partitionCount += count; - if (workDone == work.count) - { - workDone = 0; - work = workQueue.poll(); - if (work == null) - { - if (partitionCount == 0) - return; - break; - } - if (rateLimiter != null) - rateLimiter.acquire(work.count); - } + 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 @@ -340,7 +329,7 @@ public class StressAction implements Runnable e.printStackTrace(output); success = false; - workQueue.stop(); + workManager.stop(); metrics.cancel(); return; } @@ -356,107 +345,58 @@ public class StressAction implements Runnable } - private interface WorkQueue + private interface WorkManager { - // null indicates consumer should terminate - Work poll(); + // -1 indicates consumer should terminate + int takePermits(int count); // signal all consumers to terminate void stop(); } - private static final class Work - { - // index of operations - final long offset; - - // how many operations to perform - final int count; - - public Work(long offset, int count) - { - this.offset = offset; - this.count = count; - } - } - - private static final class FixedWorkQueue implements WorkQueue + private static final class FixedWorkManager implements WorkManager { - final ArrayBlockingQueue<Work> work; - volatile boolean stop = false; + final AtomicLong permits; - public FixedWorkQueue(ArrayBlockingQueue<Work> work) + public FixedWorkManager(long permits) { - this.work = work; + this.permits = new AtomicLong(permits); } @Override - public Work poll() + public int takePermits(int count) { - if (stop) - return null; - return work.poll(); + 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() { - stop = true; + permits.getAndSet(0); } - - static FixedWorkQueue build(long operations) - { - // target splitting into around 50-500k items, with a minimum size of 20 - if (operations > Integer.MAX_VALUE * (1L << 19)) - throw new IllegalStateException("Cannot currently support more than approx 2^50 operations for one stress run. This is a LOT."); - int batchSize = (int) (operations / (1 << 19)); - if (batchSize < 20) - batchSize = 20; - ArrayBlockingQueue<Work> work = new ArrayBlockingQueue<>( - (int) ((operations / batchSize) - + (operations % batchSize == 0 ? 0 : 1)) - ); - long offset = 0; - while (offset < operations) - { - work.add(new Work(offset, (int) Math.min(batchSize, operations - offset))); - offset += batchSize; - } - return new FixedWorkQueue(work); - } - } - private static final class ContinuousWorkQueue implements WorkQueue + private static final class ContinuousWorkManager implements WorkManager { - final AtomicLong offset = new AtomicLong(); - final int batchSize; volatile boolean stop = false; - private ContinuousWorkQueue(int batchSize) - { - this.batchSize = batchSize; - } - @Override - public Work poll() + public int takePermits(int count) { if (stop) - return null; - return new Work(nextOffset(), batchSize); - } - - private long nextOffset() - { - final int inc = batchSize; - while (true) - { - final long cur = offset.get(); - if (offset.compareAndSet(cur, cur + inc)) - return cur; - } + return -1; + return count; } @Override @@ -466,5 +406,4 @@ public class StressAction implements Runnable } } - } http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java b/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java index 7e5c1b6..a9edfc6 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java @@ -42,7 +42,7 @@ public class StressMetrics private final Thread thread; private volatile boolean stop = false; private volatile boolean cancelled = false; - private final Uncertainty opRateUncertainty = new Uncertainty(); + private final Uncertainty rowRateUncertainty = new Uncertainty(); private final CountDownLatch stopped = new CountDownLatch(1); private final Timing timing = new Timing(); @@ -68,6 +68,7 @@ public class StressMetrics Thread.sleep(logIntervalMillis); else Thread.sleep(sleep); + update(); } catch (InterruptedException e) { @@ -86,6 +87,7 @@ public class StressMetrics } finally { + rowRateUncertainty.wakeAll(); stopped.countDown(); } } @@ -99,7 +101,7 @@ public class StressMetrics public void waitUntilConverges(double targetUncertainty, int minMeasurements, int maxMeasurements) throws InterruptedException { - opRateUncertainty.await(targetUncertainty, minMeasurements, maxMeasurements); + rowRateUncertainty.await(targetUncertainty, minMeasurements, maxMeasurements); } public void cancel() @@ -107,7 +109,7 @@ public class StressMetrics cancelled = true; stop = true; thread.interrupt(); - opRateUncertainty.wakeAll(); + rowRateUncertainty.wakeAll(); } public void stop() throws InterruptedException @@ -120,8 +122,11 @@ public class StressMetrics private void update() throws InterruptedException { TimingInterval interval = timing.snapInterval(); - printRow("", interval, timing.getHistory(), opRateUncertainty, output); - opRateUncertainty.update(interval.adjustedOpRate()); + if (interval.partitionCount != 0) + printRow("", interval, timing.getHistory(), rowRateUncertainty, output); + rowRateUncertainty.update(interval.adjustedRowRate()); + if (timing.done()) + stop = true; } @@ -132,14 +137,15 @@ public class StressMetrics private static void printHeader(String prefix, PrintStream output) { - output.println(prefix + String.format(HEADFORMAT, "partitions","op/s", "pk/s", "row/s","mean","med",".95",".99",".999","max","time","stderr")); + output.println(prefix + String.format(HEADFORMAT, "total ops","adj row/s","op/s","pk/s","row/s","mean","med",".95",".99",".999","max","time","stderr")); } private static void printRow(String prefix, TimingInterval interval, TimingInterval total, Uncertainty opRateUncertainty, PrintStream output) { output.println(prefix + String.format(ROWFORMAT, - total.partitionCount, - interval.realOpRate(), + total.operationCount, + interval.adjustedRowRate(), + interval.opRate(), interval.partitionRate(), interval.rowRate(), interval.meanLatency(), @@ -157,7 +163,7 @@ public class StressMetrics output.println("\n"); output.println("Results:"); TimingInterval history = timing.getHistory(); - output.println(String.format("op rate : %.0f", history.realOpRate())); + output.println(String.format("op rate : %.0f", history.opRate())); output.println(String.format("partition rate : %.0f", history.partitionRate())); output.println(String.format("row rate : %.0f", history.rowRate())); output.println(String.format("latency mean : %.1f", history.meanLatency())); @@ -181,7 +187,7 @@ public class StressMetrics printRow(String.format(formatstr, ids.get(i)), summarise.get(i).timing.getHistory(), summarise.get(i).timing.getHistory(), - summarise.get(i).opRateUncertainty, + summarise.get(i).rowRateUncertainty, out ); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java index 4e09775..de561f3 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java @@ -24,15 +24,18 @@ package org.apache.cassandra.stress; import com.datastax.driver.core.*; import com.datastax.driver.core.exceptions.AlreadyExistsException; +import com.google.common.base.Function; import com.google.common.util.concurrent.Uninterruptibles; 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.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; @@ -88,7 +91,7 @@ public class StressProfile implements Serializable public String keyspaceName; public String tableName; private Map<String, GeneratorConfig> columnConfigs; - private Map<String, String> queries; + private Map<String, StressYaml.QueryDef> queries; private Map<String, String> insert; transient volatile TableMetadata tableMetaData; @@ -97,11 +100,11 @@ public class StressProfile implements Serializable transient volatile BatchStatement.Type batchType; transient volatile DistributionFactory partitions; - transient volatile RatioDistributionFactory pervisit; - transient volatile RatioDistributionFactory perbatch; + transient volatile RatioDistributionFactory selectchance; transient volatile PreparedStatement insertStatement; transient volatile Integer thriftInsertId; + transient volatile Map<String, SchemaQuery.ArgSelect> argSelects; transient volatile Map<String, PreparedStatement> queryStatements; transient volatile Map<String, Integer> thriftQueryIds; @@ -242,13 +245,18 @@ public class StressProfile implements Serializable ThriftClient tclient = settings.getThriftClient(); Map<String, PreparedStatement> stmts = new HashMap<>(); Map<String, Integer> tids = new HashMap<>(); - for (Map.Entry<String, String> e : queries.entrySet()) + Map<String, SchemaQuery.ArgSelect> args = new HashMap<>(); + for (Map.Entry<String, StressYaml.QueryDef> e : queries.entrySet()) { - stmts.put(e.getKey().toLowerCase(), jclient.prepare(e.getValue())); - tids.put(e.getKey().toLowerCase(), tclient.prepare_cql3_query(e.getValue(), Compression.NONE)); + stmts.put(e.getKey().toLowerCase(), jclient.prepare(e.getValue().cql)); + tids.put(e.getKey().toLowerCase(), tclient.prepare_cql3_query(e.getValue().cql, Compression.NONE)); + args.put(e.getKey().toLowerCase(), e.getValue().fields == null + ? SchemaQuery.ArgSelect.MULTIROW + : SchemaQuery.ArgSelect.valueOf(e.getValue().fields.toUpperCase())); } thriftQueryIds = tids; queryStatements = stmts; + argSelects = args; } catch (TException e) { @@ -260,7 +268,9 @@ public class StressProfile implements Serializable // TODO validation name = name.toLowerCase(); - return new SchemaQuery(timer, generator, settings, thriftQueryIds.get(name), queryStatements.get(name), ThriftConversion.fromThrift(settings.command.consistencyLevel), ValidationType.NOT_FAIL); + 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)); } public SchemaInsert getInsert(Timer timer, PartitionGenerator generator, StressSettings settings) @@ -328,18 +338,37 @@ public class StressProfile implements Serializable insert = new HashMap<>(); lowerCase(insert); - partitions = OptionDistribution.get(!insert.containsKey("partitions") ? "fixed(1)" : insert.remove("partitions")); - pervisit = OptionRatioDistribution.get(!insert.containsKey("pervisit") ? "fixed(1)/1" : insert.remove("pervisit")); - perbatch = OptionRatioDistribution.get(!insert.containsKey("perbatch") ? "fixed(1)/1" : insert.remove("perbatch")); - batchType = !insert.containsKey("batchtype") ? BatchStatement.Type.LOGGED : BatchStatement.Type.valueOf(insert.remove("batchtype")); + partitions = select(settings.insert.batchsize, "partitions", "fixed(1)", insert, OptionDistribution.BUILDER); + selectchance = select(settings.insert.selectRatio, "select", "fixed(1)/1", insert, OptionRatioDistribution.BUILDER); + batchType = settings.insert.batchType != null + ? settings.insert.batchType + : !insert.containsKey("batchtype") + ? BatchStatement.Type.LOGGED + : BatchStatement.Type.valueOf(insert.remove("batchtype")); if (!insert.isEmpty()) throw new IllegalArgumentException("Unrecognised insert option(s): " + insert); + Distribution visits = settings.insert.visits.get(); + // these min/max are not absolutely accurate if selectchance < 1, but they're close enough to + // guarantee the vast majority of actions occur in these bounds + double minBatchSize = selectchance.get().min() * partitions.get().minValue() * generator.minRowCount * (1d / visits.maxValue()); + double maxBatchSize = selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount * (1d / visits.minValue()); + System.out.printf("Generating batches with [%d..%d] partitions and [%.0f..%.0f] rows (of [%.0f..%.0f] total rows in the partitions)\n", + partitions.get().minValue(), partitions.get().maxValue(), + minBatchSize, maxBatchSize, + partitions.get().minValue() * generator.minRowCount, + partitions.get().maxValue() * generator.maxRowCount); if (generator.maxRowCount > 100 * 1000 * 1000) System.err.printf("WARNING: You have defined a schema that permits very large partitions (%.0f max rows (>100M))\n", generator.maxRowCount); - if (perbatch.get().max() * pervisit.get().max() * partitions.get().maxValue() * generator.maxRowCount > 100000) + if (batchType == BatchStatement.Type.LOGGED && maxBatchSize > 65535) + { + System.err.printf("ERROR: You have defined a workload that generates batches with more than 65k rows (%.0f), but have required the use of LOGGED batches. There is a 65k row limit on a single batch.\n", + selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount); + System.exit(1); + } + if (maxBatchSize > 100000) System.err.printf("WARNING: You have defined a schema that permits very large batches (%.0f max rows (>100K)). This may OOM this stress client, or the server.\n", - perbatch.get().max() * pervisit.get().max() * partitions.get().maxValue() * generator.maxRowCount); + selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount); JavaDriverClient client = settings.getJavaDriverClient(); String query = sb.toString(); @@ -356,10 +385,20 @@ public class StressProfile implements Serializable } } - return new SchemaInsert(timer, generator, settings, partitions.get(), pervisit.get(), perbatch.get(), thriftInsertId, insertStatement, ThriftConversion.fromThrift(settings.command.consistencyLevel), batchType); + return new SchemaInsert(timer, generator, settings, partitions.get(), selectchance.get(), thriftInsertId, insertStatement, ThriftConversion.fromThrift(settings.command.consistencyLevel), batchType); + } + + private static <E> E select(E first, String key, String defValue, Map<String, String> map, Function<String, E> builder) + { + String val = map.remove(key); + if (first != null) + return first; + if (val != null) + return builder.apply(val); + return builder.apply(defValue); } - public PartitionGenerator newGenerator(StressSettings settings) + public PartitionGenerator newGenerator(StressSettings settings, SeedManager seeds) { if (generatorFactory == null) { @@ -371,7 +410,7 @@ public class StressProfile implements Serializable } } - return generatorFactory.newGenerator(); + return generatorFactory.newGenerator(settings, seeds); } private class GeneratorFactory @@ -393,9 +432,9 @@ public class StressProfile implements Serializable valueColumns.add(new ColumnInfo(metadata.getName(), metadata.getType(), columnConfigs.get(metadata.getName()))); } - PartitionGenerator newGenerator() + PartitionGenerator newGenerator(StressSettings settings, SeedManager seeds) { - return new PartitionGenerator(get(partitionKeys), get(clusteringColumns), get(valueColumns)); + return new PartitionGenerator(get(partitionKeys), get(clusteringColumns), get(valueColumns), settings.generate.order, seeds); } List<Generator> get(List<ColumnInfo> columnInfos) http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/StressYaml.java b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java index deea1fb..b6efc5e 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressYaml.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java @@ -30,8 +30,14 @@ public class StressYaml public String table; public String table_definition; - public List<Map<String,Object>> columnspec; - public Map<String,String> queries; - public Map<String,String> insert; + public List<Map<String, Object>> columnspec; + public Map<String, QueryDef> queries; + public Map<String, String> insert; + + public static class QueryDef + { + public String cql; + public String fields; + } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/generate/DistributionInverted.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/DistributionInverted.java b/tools/stress/src/org/apache/cassandra/stress/generate/DistributionInverted.java index 13fae0d..4062b58 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/DistributionInverted.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/DistributionInverted.java @@ -55,4 +55,11 @@ public class DistributionInverted extends Distribution wrapped.setSeed(seed); } + public static Distribution invert(Distribution distribution) + { + if (distribution instanceof DistributionInverted) + return ((DistributionInverted) distribution).wrapped; + return new DistributionInverted(distribution); + } + } http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/generate/DistributionQuantized.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/DistributionQuantized.java b/tools/stress/src/org/apache/cassandra/stress/generate/DistributionQuantized.java new file mode 100644 index 0000000..9771134 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/generate/DistributionQuantized.java @@ -0,0 +1,90 @@ +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.util.Arrays; +import java.util.Random; + +import org.apache.cassandra.stress.Stress; + +public class DistributionQuantized extends Distribution +{ + + final Distribution delegate; + final long[] bounds; + final Random random = new Random(); + + public DistributionQuantized(Distribution delegate, int quantas) + { + this.delegate = delegate; + this.bounds = new long[quantas + 1]; + bounds[0] = delegate.minValue(); + bounds[quantas] = delegate.maxValue() + 1; + for (int i = 1 ; i < quantas ; i++) + bounds[i] = delegate.inverseCumProb(i / (double) quantas); + } + + @Override + public long next() + { + int quanta = quanta(delegate.next()); + return bounds[quanta] + (long) (random.nextDouble() * ((bounds[quanta + 1] - bounds[quanta]))); + } + + public double nextDouble() + { + throw new UnsupportedOperationException(); + } + + @Override + public long inverseCumProb(double cumProb) + { + long val = delegate.inverseCumProb(cumProb); + int quanta = quanta(val); + if (quanta < 0) + return bounds[0]; + if (quanta >= bounds.length - 1) + return bounds[bounds.length - 1] - 1; + cumProb -= (quanta / ((double) bounds.length - 1)); + cumProb *= (double) bounds.length - 1; + return bounds[quanta] + (long) (cumProb * (bounds[quanta + 1] - bounds[quanta])); + } + + int quanta(long val) + { + int i = Arrays.binarySearch(bounds, val); + if (i < 0) + return -2 -i; + return i - 1; + } + + public void setSeed(long seed) + { + delegate.setSeed(seed); + } + + public static void main(String[] args) throws Exception + { + Stress.main(new String[] { "print", "dist=qextreme(1..1M,2,2)"}); + } + +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java b/tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java new file mode 100644 index 0000000..455fec4 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java @@ -0,0 +1,116 @@ +/* +* Licensed to the Apache Software Foundation (ASF) under one +* or more contributor license agreements. See the NOTICE file +* distributed with this work for additional information +* regarding copyright ownership. The ASF licenses this file +* to you under the Apache License, Version 2.0 (the +* "License"); you may not use this file except in compliance +* with the License. You may obtain a copy of the License at +* +* http://www.apache.org/licenses/LICENSE-2.0 +* +* Unless required by applicable law or agreed to in writing, +* software distributed under the License is distributed on an +* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +* KIND, either express or implied. See the License for the +* specific language governing permissions and limitations +* under the License. +*/ +package org.apache.cassandra.stress.generate; + +import java.util.Random; + +import org.apache.commons.math3.random.RandomGenerator; + +// based on http://en.wikipedia.org/wiki/Xorshift, but periodically we reseed with our stronger random generator +// note it is also non-atomically updated, so expects to be used by a single thread +public class FasterRandom implements RandomGenerator +{ + final Random random = new Random(); + + private long seed; + private int reseed; + + public void setSeed(int seed) + { + setSeed((long) seed); + } + + public void setSeed(int[] ints) + { + if (ints.length > 1) + setSeed (((long) ints[0] << 32) | ints[1]); + else + setSeed(ints[0]); + } + + public void setSeed(long seed) + { + this.seed = seed; + rollover(); + } + + private void rollover() + { + this.reseed = 0; + random.setSeed(seed); + seed = random.nextLong(); + } + + public void nextBytes(byte[] bytes) + { + int i = 0; + while (i < bytes.length) + { + long next = nextLong(); + while (i < bytes.length) + { + bytes[i++] = (byte) (next & 0xFF); + next >>>= 8; + } + } + } + + public int nextInt() + { + return (int) nextLong(); + } + + public int nextInt(int i) + { + return Math.abs((int) nextLong() % i); + } + + public long nextLong() + { + if (++this.reseed == 32) + rollover(); + + long seed = this.seed; + seed ^= seed >> 12; + seed ^= seed << 25; + seed ^= seed >> 27; + this.seed = seed; + return seed * 2685821657736338717L; + } + + public boolean nextBoolean() + { + return ((int) nextLong() & 1) == 1; + } + + public float nextFloat() + { + return Float.intBitsToFloat((int) nextLong()); + } + + public double nextDouble() + { + return Double.longBitsToDouble(nextLong()); + } + + public double nextGaussian() + { + return random.nextGaussian(); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/generate/Partition.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/Partition.java b/tools/stress/src/org/apache/cassandra/stress/generate/Partition.java index f05e95b..18f5732 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/Partition.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/Partition.java @@ -23,24 +23,34 @@ package org.apache.cassandra.stress.generate; 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; @@ -55,31 +65,32 @@ public class Partition iterator = new SingleRowIterator(); } - void setSeed(long seed) + 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); + 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) + public RowIterator iterator(double useChance, boolean isWrite) { - iterator.reset(useChance, 0); + iterator.reset(useChance, 0, 1, isWrite); return iterator; } - public RowIterator iterator(int targetCount) + public RowIterator iterator(int targetCount, boolean isWrite) { - iterator.reset(Double.NaN, targetCount); + iterator.reset(Double.NaN, targetCount, 1, isWrite); return iterator; } @@ -87,12 +98,12 @@ public class Partition { boolean done; - void reset(double useChance, int targetCount) + void reset(double useChance, int targetCount, int batches, boolean isWrite) { done = false; } - public Iterable<Row> batch(double ratio) + public Iterable<Row> next() { if (done) return Collections.emptyList(); @@ -110,6 +121,12 @@ public class Partition { return done; } + + public void markWriteFinished() + { + assert done; + generator.seeds.markFinished(seed); + } } public abstract class RowIterator @@ -117,10 +134,10 @@ public class Partition // we reuse the row object to save garbage final Row row = new Row(partitionKey, new Object[generator.clusteringComponents.size() + generator.valueComponents.size()]); - public abstract Iterable<Row> batch(double ratio); - abstract void reset(double useChance, int targetCount); - + public abstract Iterable<Row> next(); public abstract boolean done(); + public abstract void markWriteFinished(); + abstract void reset(double useChance, int targetCount, int batches, boolean isWrite); public Partition partition() { @@ -128,31 +145,40 @@ public class Partition } } - // permits iterating a random subset of the procedurally generated rows in this partition; this is the only mechanism for visiting rows + // 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 children can take, and repeat the process. So at any one time we are using space proportional + // 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; - double expectedRowCount; - // the current seed in use at any given level; used to save recalculating it for each row, so we only need to recalc - // from prior row + // 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 Queue<Object>[] clusteringComponents = new ArrayDeque[generator.clusteringComponents.size()]; + final Deque<Object>[] clusteringComponents = new ArrayDeque[generator.clusteringComponents.size()]; // we want our chance of selection to be applied uniformly, so we compound the roll we make at each level // so that we know with what chance we reached there, and we adjust our roll at that level by that amount - double[] chancemodifier = new double[generator.clusteringComponents.size()]; - double[] rollmodifier = new double[generator.clusteringComponents.size()]; + 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 set for generating unique clustering components + // reusable collections for generating unique and sorted clustering components final Set<Object> unique = new HashSet<>(); + final List<Comparable> tosort = new ArrayList<>(); final Random random = new Random(); MultiRowIterator() @@ -163,126 +189,262 @@ public class Partition chancemodifier[0] = generator.clusteringChildAverages[0]; } - void reset(double useChance, int targetCount) + // 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(); - this.expectedRowCount = firstComponentCount * generator.clusteringChildAverages[0]; + 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 / expectedRowCount)); + 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(); - - this.useChance = useChance; clusteringSeeds[0] = idseed; - clusteringComponents[0].add(this); fill(clusteringComponents[0], firstComponentCount, generator.clusteringComponents.get(0)); - advance(0, 1f); + + // seek to our start position + seek(isWrite ? position : null); } - void fill(int component) + // 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[component - 1]; - Generator gen = generator.clusteringComponents.get(component); + long seed = clusteringSeeds[depth - 1]; + Generator gen = generator.clusteringComponents.get(depth); gen.setSeed(seed); - clusteringSeeds[component] = seed(clusteringComponents[component - 1].peek(), generator.clusteringComponents.get(component - 1).type, seed); - fill(clusteringComponents[component], (int) gen.clusteringDistribution.next(), gen); + clusteringSeeds[depth] = seed(clusteringComponents[depth - 1].peek(), generator.clusteringComponents.get(depth - 1).type, seed); + fill(clusteringComponents[depth], (int) gen.clusteringDistribution.next(), gen); } + // generate the clustering components into the queue void fill(Queue<Object> queue, int count, Generator generator) { if (count == 1) { queue.add(generator.generate()); + return; } - else + + switch (Partition.this.generator.order) { - unique.clear(); - for (int i = 0 ; i < count ; i++) - { - Object next = generator.generate(); - if (unique.add(next)) - queue.add(next); - } + 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++) + if (i == 0 || tosort.get(i - 1).compareTo(i) < 0) + queue.add(tosort.get(i)); + break; + } + 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); } - private boolean advance(double continueChance) + // 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 always start at the leaf level + // 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; - // fill the row with the position we *were* at (unless pre-start) + 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); - long seed = clusteringSeeds[depth]; - seed = seed(clusteringComponents[depth].peek(), generator.clusteringComponents.get(depth).type, seed); - gen.setSeed(seed); + gen.setSeed(rowSeed); row.row[i] = gen.generate(); } - clusteringComponents[depth].poll(); + returnedOne = true; + forceReturnOne = false; - return advance(depth, continueChance); + // then we advance the leaf level + advance(depth); } - private boolean advance(int depth, double continueChance) + 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 false; + return; depth--; clusteringComponents[depth].poll(); + position[depth]++; continue; } - // the chance of descending is the uniform use chance, multiplied by the number of children + 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 (thischance > 0.999f || thischance >= random.nextDouble()) + 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; - rollmodifier[depth] = rollmodifier[depth - 1] / Math.min(1d, thischance); - chancemodifier[depth] = generator.clusteringChildAverages[depth] * rollmodifier[depth]; + // 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]++; } - - return continueChance >= 1.0d || continueChance >= random.nextDouble(); } - public Iterable<Row> batch(final double ratio) + public Iterable<Row> next() { - final double continueChance = 1d - (Math.pow(ratio, expectedRowCount * useChance)); + final int[] limit = position.clone(); + int remainingSize = batchSize; + for (int i = 0 ; i < limit.length && remainingSize > 0 ; i++) + { + limit[i] += remainingSize / generator.clusteringChildAverages[i]; + remainingSize %= generator.clusteringChildAverages[i]; + } + assert remainingSize == 0; + for (int i = limit.length - 1 ; i > 0 ; i--) + { + if (limit[i] > generator.clusteringChildAverages[i]) + { + limit[i - 1] += limit[i] / generator.clusteringChildAverages[i]; + limit[i] %= generator.clusteringChildAverages[i]; + } + } + for (int i = 0 ; i < limit.length ; i++) + { + if (limit[i] < this.limit[i]) + break; + limit[i] = Math.min(limit[i], this.limit[i]); + } return new Iterable<Row>() { public Iterator<Row> iterator() { return new Iterator<Row>() { - boolean hasNext = true; + public boolean hasNext() { - return 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() { - hasNext = advance(continueChance); + advance(); return row; } @@ -300,26 +462,37 @@ public class Partition 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; } } - public String getKeyAsString() + private static class RandomOrder implements Comparable<RandomOrder> { - StringBuilder sb = new StringBuilder(); - int i = 0; - for (Object key : partitionKey) + final int order = ThreadLocalRandom.current().nextInt(); + final Object value; + private RandomOrder(Object value) { - if (i > 0) - sb.append("|"); - AbstractType type = generator.partitionKey.get(i++).type; - sb.append(type.getString(type.decompose(key))); + this.value = value; + } + + public int compareTo(RandomOrder that) + { + return Integer.compare(this.order, that.order); } - return sb.toString(); } + // 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) @@ -355,6 +528,20 @@ public class Partition 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() { http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java ---------------------------------------------------------------------- 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 d05350d..128d2f5 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java @@ -30,18 +30,27 @@ 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 { + public static enum Order + { + ARBITRARY, SHUFFLED, SORTED + } + public final double maxRowCount; + public final double minRowCount; final List<Generator> partitionKey; final List<Generator> clusteringComponents; final List<Generator> valueComponents; final int[] clusteringChildAverages; private final Map<String, Integer> indexMap; + final Order order; + final SeedManager seeds; final List<Partition> recyclable = new ArrayList<>(); int partitionsInUse = 0; @@ -51,18 +60,25 @@ public class PartitionGenerator partitionsInUse = 0; } - public PartitionGenerator(List<Generator> partitionKey, List<Generator> clusteringComponents, List<Generator> valueComponents) + public PartitionGenerator(List<Generator> partitionKey, List<Generator> clusteringComponents, List<Generator> valueComponents, Order order, SeedManager seeds) { 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); double maxRowCount = 1d; + double minRowCount = 1d; for (Generator component : clusteringComponents) + { maxRowCount *= component.clusteringDistribution.maxValue(); + minRowCount *= component.clusteringDistribution.minValue(); + } this.maxRowCount = maxRowCount; + this.minRowCount = minRowCount; this.indexMap = new HashMap<>(); int i = 0; for (Generator generator : partitionKey) @@ -72,6 +88,11 @@ public class PartitionGenerator indexMap.put(generator.name, i++); } + public boolean permitNulls(int index) + { + return !(index < 0 || index < clusteringComponents.size()); + } + public int indexOf(String name) { Integer i = indexMap.get(name); @@ -80,11 +101,14 @@ public class PartitionGenerator return i; } - public Partition generate(long seed) + 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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistribution.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistribution.java b/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistribution.java index 37ad4c5..c71945a 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/RatioDistribution.java @@ -39,6 +39,11 @@ public class RatioDistribution return Math.max(0f, Math.min(1f, distribution.nextDouble() / divisor)); } + public double min() + { + return Math.min(1d, distribution.minValue() / divisor); + } + public double max() { return Math.min(1d, distribution.maxValue() / divisor); http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java b/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java new file mode 100644 index 0000000..f427608 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/generate/Seed.java @@ -0,0 +1,67 @@ +/* +* Licensed to the Apache Software Foundation (ASF) under one +* or more contributor license agreements. See the NOTICE file +* distributed with this work for additional information +* regarding copyright ownership. The ASF licenses this file +* to you under the Apache License, Version 2.0 (the +* "License"); you may not use this file except in compliance +* with the License. You may obtain a copy of the License at +* +* http://www.apache.org/licenses/LICENSE-2.0 +* +* Unless required by applicable law or agreed to in writing, +* software distributed under the License is distributed on an +* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +* KIND, either express or implied. See the License for the +* specific language governing permissions and limitations +* under the License. +*/ +package org.apache.cassandra.stress.generate; + +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; + +import org.apache.cassandra.stress.util.DynamicList; + +public class Seed implements Comparable<Seed> +{ + + public final long seed; + final int visits; + + DynamicList.Node poolNode; + volatile int[] position; + volatile State state = State.HELD; + + private static final AtomicReferenceFieldUpdater<Seed, Seed.State> stateUpdater = AtomicReferenceFieldUpdater.newUpdater(Seed.class, State.class, "state"); + + 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() + { + return position; + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/0580fb2b/tools/stress/src/org/apache/cassandra/stress/generate/SeedGenerator.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/SeedGenerator.java b/tools/stress/src/org/apache/cassandra/stress/generate/SeedGenerator.java deleted file mode 100644 index d579223..0000000 --- a/tools/stress/src/org/apache/cassandra/stress/generate/SeedGenerator.java +++ /dev/null @@ -1,29 +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. - * - */ - - -public interface SeedGenerator -{ - - long next(long workIndex); - -}
