diff --git a/CHANGES.txt b/CHANGES.txt index a8fd12914e..18d6872436 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -57,6 +57,9 @@ Merged from 1.2: * 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) * Track added sstable size correctly (CASSANDRA-7239) * (cqlsh) Fix case insensitivity (CASSANDRA-7834) diff --git a/NEWS.txt b/NEWS.txt index db34fd3fef..b464b2107f 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -44,12 +44,19 @@ New features will be put back in L0. - Bootstrapping now ensures that range movements are consistent, meaning the data for the new node is taken from the node that is no - longer a responsible for that range of keys. + longer a responsible for that range of keys. If you want the old behavior (due to a lost node perhaps) you can set the following property (-Dcassandra.consistent.rangemovement=false) - It is now possible to use quoted identifiers in triggers' names. WARNING: if you previously used triggers with capital letters in their names, then you must quote them from now on. + - Improved stress tool (http://goo.gl/OTNqiQ) + - New incremental repair option (http://goo.gl/MjohJp, http://goo.gl/f8jSme) + - Incremental replacement of compacted SSTables (http://goo.gl/JfDBGW) + - The row cache can now cache only the head of partitions (http://goo.gl/6TJPH6) + - Off-heap memtables (http://goo.gl/YT7znJ) + - CQL improvements and additions: User-defined types, tuple types, 2ndary + indexing of collections, ... (http://goo.gl/kQl7GW) Upgrading --------- @@ -66,8 +73,8 @@ Upgrading - Counters implementation has been changed, replaced by a safer one with less caveats, but different performance characteristics. You might have to change your data model to accomodate the new implementation. - (See https://issues.apache.org/jira/browse/CASSANDRA-6504 and the dev - blog post at http://www.datastax.com/dev/blog/ for details). + (See https://issues.apache.org/jira/browse/CASSANDRA-6504 and the + blog post at http://goo.gl/qj8iQl for details). - (per-table) index_interval parameter has been replaced with min_index_interval and max_index_interval paratemeters. index_interval has been deprecated. diff --git a/debian/changelog b/debian/changelog index e467737bd3..f2eccebf48 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,15 @@ +cassandra (2.1.0) unstable; urgency=medium + + * New release + + -- Sylvain Lebresne Sun, 07 Sep 2014 15:21:41 +0200 + +cassandra (2.1.0~rc7) unstable; urgency=medium + + * New RC release + + -- Sylvain Lebresne Thu, 28 Aug 2014 16:32:12 +0200 + cassandra (2.1.0~rc6) unstable; urgency=medium * New RC release diff --git a/tools/cqlstress-counter-example.yaml b/tools/cqlstress-counter-example.yaml index cff14b64a0..f8f70ea8f2 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 diff --git a/tools/cqlstress-example.yaml b/tools/cqlstress-example.yaml index d5c90a2159..4dd5e4a909 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) diff --git a/tools/cqlstress-insanity-example.yaml b/tools/cqlstress-insanity-example.yaml index ef1bb3aa3a..ea4f97f09e 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) diff --git a/tools/stress/src/org/apache/cassandra/stress/Operation.java b/tools/stress/src/org/apache/cassandra/stress/Operation.java index 7831074e4a..5560240159 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); diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java index 2105a7295e..e58bfa1000 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,109 +345,60 @@ 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; - volatile boolean stop = false; + final AtomicLong permits; - public FixedWorkQueue(ArrayBlockingQueue 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(); - } - - @Override - public void stop() - { - stop = true; - } - - 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 = 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 - { - - final AtomicLong offset = new AtomicLong(); - final int batchSize; - volatile boolean stop = false; - - private ContinuousWorkQueue(int batchSize) - { - this.batchSize = batchSize; - } - - @Override - public Work poll() - { - 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; + 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() { @@ -466,5 +406,4 @@ public class StressAction implements Runnable } } - } diff --git a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java b/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java index 7e5c1b6310..a9edfc6f3d 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 ); } diff --git a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java index 4e09775406..de561f39c9 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 columnConfigs; - private Map queries; + private Map queries; private Map 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 argSelects; transient volatile Map queryStatements; transient volatile Map thriftQueryIds; @@ -242,13 +245,18 @@ public class StressProfile implements Serializable ThriftClient tclient = settings.getThriftClient(); Map stmts = new HashMap<>(); Map tids = new HashMap<>(); - for (Map.Entry e : queries.entrySet()) + Map args = new HashMap<>(); + for (Map.Entry 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); } - public PartitionGenerator newGenerator(StressSettings settings) + private static E select(E first, String key, String defValue, Map map, Function 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, 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 get(List columnInfos) diff --git a/tools/stress/src/org/apache/cassandra/stress/StressYaml.java b/tools/stress/src/org/apache/cassandra/stress/StressYaml.java index deea1fbde0..b6efc5e32f 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> columnspec; - public Map queries; - public Map insert; + public List> columnspec; + public Map queries; + public Map insert; + + public static class QueryDef + { + public String cql; + public String fields; + } } 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 13fae0db97..4062b58b10 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); + } + } 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 0000000000..9771134db1 --- /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)"}); + } + +} 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 0000000000..455fec480c --- /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(); + } +} 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 f05e95b5c1..18f5732af8 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 batch(double ratio) + public Iterable 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 batch(double ratio); - abstract void reset(double useChance, int targetCount); - + 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() { @@ -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[] clusteringComponents = new ArrayDeque[generator.clusteringComponents.size()]; + 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 - 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()]; - // reusable set for generating unique clustering components + // 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() @@ -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) { - generator.clusteringComponents.get(0).setSeed(idseed); - int firstComponentCount = (int) generator.clusteringComponents.get(0).clusteringDistribution.next(); - this.expectedRowCount = firstComponentCount * generator.clusteringChildAverages[0]; - if (Double.isNaN(useChance)) - useChance = Math.max(0d, Math.min(1d, targetCount / expectedRowCount)); + 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(); - - 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 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(); } } - private boolean advance(double continueChance) + // seek to the provided position (or the first entry if null) + private void seek(int[] position) { - // we always start at the leaf level + 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; - // 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 batch(final double ratio) + public Iterable 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() { public Iterator iterator() { return new Iterator() { - 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 { - 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() { 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 d05350d320..128d2f51b2 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 partitionKey; final List clusteringComponents; final List valueComponents; final int[] clusteringChildAverages; private final Map indexMap; + final Order order; + final SeedManager seeds; final List recyclable = new ArrayList<>(); int partitionsInUse = 0; @@ -51,18 +60,25 @@ public class PartitionGenerator partitionsInUse = 0; } - public PartitionGenerator(List partitionKey, List clusteringComponents, List valueComponents) + public PartitionGenerator(List partitionKey, List clusteringComponents, List 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; 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 37ad4c5b59..c71945a397 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); 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 0000000000..f427608f31 --- /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 +{ + + public final long seed; + final int visits; + + DynamicList.Node poolNode; + volatile int[] position; + volatile State state = State.HELD; + + private static final AtomicReferenceFieldUpdater 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; + } +} 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 d5792233fa..0000000000 --- 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); - -} diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java b/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java new file mode 100644 index 0000000000..dba721dd44 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java @@ -0,0 +1,249 @@ +/* +* 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.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.atomic.AtomicLong; + +import org.apache.cassandra.stress.Operation; +import org.apache.cassandra.stress.settings.StressSettings; +import org.apache.cassandra.stress.util.DynamicList; + +public class SeedManager +{ + + final Distribution visits; + final Generator writes; + final Generator reads; + final ConcurrentHashMap managing = new ConcurrentHashMap<>(); + final DynamicList sampleFrom; + final Distribution sample; + + public SeedManager(StressSettings settings) + { + Generator writes, reads; + if (settings.generate.sequence != null) + { + long[] seq = settings.generate.sequence; + if (settings.generate.readlookback != null) + { + LookbackableWriteGenerator series = new LookbackableWriteGenerator(seq[0], seq[1], settings.generate.wrap, settings.generate.readlookback.get()); + writes = series; + reads = series.reads; + } + else + { + writes = reads = new SeriesGenerator(seq[0], seq[1], settings.generate.wrap); + } + } + else + { + writes = reads = new RandomGenerator(settings.generate.distribution.get()); + } + 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()); + } + + public Seed next(Operation op) + { + if (!op.isWrite()) + { + Seed seed = reads.next(-1); + if (seed == null) + return null; + Seed managing = this.managing.get(seed); + return managing == null ? seed : managing; + } + + while (true) + { + int index = (int) sample.next(); + Seed seed = sampleFrom.get(index); + if (seed != null && seed.take()) + 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; + } + } + + public void markVisited(Seed seed, int[] position) + { + boolean first = seed.position == null; + seed.position = position; + finishedWriting(seed, first, false); + } + + public void markFinished(Seed seed) + { + 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); + } + + private abstract class Generator + { + abstract Seed next(int visits); + void finishWrite(Seed seed) { } + } + + private class RandomGenerator extends Generator + { + + final Distribution distribution; + + public RandomGenerator(Distribution distribution) + { + this.distribution = distribution; + } + + public Seed next(int visits) + { + return new Seed(distribution.next(), visits); + } + } + + private class SeriesGenerator extends Generator + { + + final long start; + final long totalCount; + final boolean wrap; + final AtomicLong next = new AtomicLong(); + + public SeriesGenerator(long start, long end, boolean wrap) + { + this.wrap = wrap; + if (start > end) + throw new IllegalStateException(); + this.start = start; + this.totalCount = 1 + end - start; + } + + public Seed next(int visits) + { + long next = this.next.getAndIncrement(); + if (!wrap && next >= totalCount) + return null; + return new Seed(start + (next % totalCount), visits); + } + } + + private class LookbackableWriteGenerator extends SeriesGenerator + { + + final AtomicLong writeCount = new AtomicLong(); + final ConcurrentSkipListMap afterMin = new ConcurrentSkipListMap<>(); + final LookbackReadGenerator reads; + + public LookbackableWriteGenerator(long start, long end, boolean wrap, Distribution readLookback) + { + super(start, end, wrap); + this.writeCount.set(0); + reads = new LookbackReadGenerator(readLookback); + } + + public Seed next(int visits) + { + long next = this.next.getAndIncrement(); + if (!wrap && next >= totalCount) + return null; + return new Seed(start + (next % totalCount), visits); + } + + void finishWrite(Seed seed) + { + if (seed.seed <= writeCount.get()) + return; + afterMin.put(seed, seed); + while (true) + { + Map.Entry head = afterMin.firstEntry(); + if (head == null) + return; + long min = this.writeCount.get(); + if (head.getKey().seed <= min) + return; + if (head.getKey().seed == min + 1 && this.writeCount.compareAndSet(min, min + 1)) + { + afterMin.remove(head.getKey()); + continue; + } + return; + } + } + + private class LookbackReadGenerator extends Generator + { + + final Distribution lookback; + + public LookbackReadGenerator(Distribution lookback) + { + this.lookback = lookback; + if (lookback.maxValue() > start + totalCount) + throw new IllegalArgumentException("Invalid lookback distribution; max value is " + lookback.maxValue() + + ", but series only ranges from " + writeCount + " to " + (start + totalCount)); + } + + public Seed next(int visits) + { + long lookback = this.lookback.next(); + long range = writeCount.get(); + long startOffset = range - lookback; + if (startOffset < 0) + { + if (range == totalCount && !wrap) + return null; + startOffset = range == 0 ? 0 : lookback % range; + } + return new Seed(start + startOffset, visits); + } + } + + } + +} diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/SeedRandomGenerator.java b/tools/stress/src/org/apache/cassandra/stress/generate/SeedRandomGenerator.java deleted file mode 100644 index b590ffabab..0000000000 --- a/tools/stress/src/org/apache/cassandra/stress/generate/SeedRandomGenerator.java +++ /dev/null @@ -1,54 +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 class SeedRandomGenerator implements SeedGenerator -{ - - final Distribution distribution; - final Distribution clustering; - - private long next; - private int count; - - public SeedRandomGenerator(Distribution distribution, Distribution clustering) - { - this.distribution = distribution; - this.clustering = clustering; - } - - public long next(long workIndex) - { - if (count == 0) - { - next = distribution.next(); - count = (int) clustering.next(); - } - long result = next; - count--; - if (next == distribution.maxValue()) - next = distribution.minValue(); - else - next++; - return result; - } -} diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/SeedSeriesGenerator.java b/tools/stress/src/org/apache/cassandra/stress/generate/SeedSeriesGenerator.java deleted file mode 100644 index 78a8784cf7..0000000000 --- a/tools/stress/src/org/apache/cassandra/stress/generate/SeedSeriesGenerator.java +++ /dev/null @@ -1,42 +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 class SeedSeriesGenerator implements SeedGenerator -{ - - final long min; - final long count; - - public SeedSeriesGenerator(long min, long max) - { - if (min > max) - throw new IllegalStateException(); - this.min = min; - this.count = 1 + max - min; - } - - public long next(long workIndex) - { - return min + (workIndex % count); - } -} diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Booleans.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Booleans.java index b1d84d6926..21525affbb 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Booleans.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Booleans.java @@ -26,7 +26,7 @@ public class Booleans extends Generator { public Booleans(String name, GeneratorConfig config) { - super(BooleanType.instance, config, name); + super(BooleanType.instance, config, name, Boolean.class); } @Override 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 2a5bddf3c9..358163c10a 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 @@ -21,6 +21,7 @@ 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; @@ -29,11 +30,11 @@ import java.util.Random; public class Bytes extends Generator { private final byte[] bytes; - private final Random rand = new Random(); + private final FasterRandom rand = new FasterRandom(); public Bytes(String name, GeneratorConfig config) { - super(BytesType.instance, config, name); + super(BytesType.instance, config, name, ByteBuffer.class); bytes = new byte[(int) sizeDistribution.maxValue()]; } @@ -45,8 +46,8 @@ public class Bytes extends Generator rand.setSeed(~seed); int size = (int) sizeDistribution.next(); for (int i = 0; i < size; ) - for (int v = rand.nextInt(), - n = Math.min(size - i, Integer.SIZE/Byte.SIZE); + for (long v = rand.nextLong(), + n = Math.min(size - i, Long.SIZE/Byte.SIZE); n-- > 0; v >>= Byte.SIZE) bytes[i++] = (byte)v; return ByteBuffer.wrap(Arrays.copyOf(bytes, size)); diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Dates.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Dates.java index 7d36be2d3d..7350f57e5e 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Dates.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Dates.java @@ -30,9 +30,10 @@ public class Dates extends Generator { public Dates(String name, GeneratorConfig config) { - super(DateType.instance, config, name); + super(DateType.instance, config, name, Date.class); } + // TODO: let the range of values generated advance as stress test progresses @Override public Date generate() { diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Doubles.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Doubles.java index 76e983dec8..0f04eb6871 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Doubles.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Doubles.java @@ -26,7 +26,7 @@ public class Doubles extends Generator { public Doubles(String name, GeneratorConfig config) { - super(DoubleType.instance, config, name); + super(DoubleType.instance, config, name, Double.class); } @Override diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Floats.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Floats.java index 8e23c11222..19f449aa47 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Floats.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Floats.java @@ -26,7 +26,7 @@ public class Floats extends Generator { public Floats(String name, GeneratorConfig config) { - super(FloatType.instance, config, name); + super(FloatType.instance, config, name, Float.class); } @Override diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Generator.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Generator.java index 1040bb3802..00f866afa8 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Generator.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Generator.java @@ -31,15 +31,17 @@ public abstract class Generator public final String name; public final AbstractType type; + public final Class clazz; final long salt; final Distribution identityDistribution; final Distribution sizeDistribution; public final Distribution clusteringDistribution; - public Generator(AbstractType type, GeneratorConfig config, String name) + public Generator(AbstractType type, GeneratorConfig config, String name, Class clazz) { this.type = type; this.name = name; + this.clazz = clazz; this.salt = config.salt; this.identityDistribution = config.getIdentityDistribution(defaultIdentityDistribution()); this.sizeDistribution = config.getSizeDistribution(defaultSizeDistribution()); diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/HexBytes.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/HexBytes.java index db46bacf0e..19f2cc3027 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/HexBytes.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/HexBytes.java @@ -31,7 +31,7 @@ public class HexBytes extends Generator public HexBytes(String name, GeneratorConfig config) { - super(BytesType.instance, config, name); + super(BytesType.instance, config, name, ByteBuffer.class); bytes = new byte[(int) sizeDistribution.maxValue()]; } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/HexStrings.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/HexStrings.java index ce65b8aa64..c811a617d6 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/HexStrings.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/HexStrings.java @@ -20,8 +20,6 @@ */ package org.apache.cassandra.stress.generate.values; -import java.util.Random; - import org.apache.cassandra.db.marshal.UTF8Type; public class HexStrings extends Generator @@ -30,7 +28,7 @@ public class HexStrings extends Generator public HexStrings(String name, GeneratorConfig config) { - super(UTF8Type.instance, config, name); + super(UTF8Type.instance, config, name, String.class); chars = new char[(int) sizeDistribution.maxValue()]; } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Inets.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Inets.java index 334d73c5b5..107daad3bd 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Inets.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Inets.java @@ -31,7 +31,7 @@ public class Inets extends Generator final byte[] buf; public Inets(String name, GeneratorConfig config) { - super(InetAddressType.instance, config, name); + super(InetAddressType.instance, config, name, InetAddress.class); buf = new byte[4]; } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Integers.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Integers.java index 8b9b33a077..e05c615488 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Integers.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Integers.java @@ -27,7 +27,7 @@ public class Integers extends Generator public Integers(String name, GeneratorConfig config) { - super(Int32Type.instance, config, name); + super(Int32Type.instance, config, name, Integer.class); } @Override diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Lists.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Lists.java index d188f7ecfb..6480d7aa55 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Lists.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Lists.java @@ -33,7 +33,7 @@ public class Lists extends Generator public Lists(String name, Generator valueType, GeneratorConfig config) { - super(ListType.getInstance(valueType.type), config, name); + super(ListType.getInstance(valueType.type), config, name, List.class); this.valueType = valueType; buffer = new Object[(int) sizeDistribution.maxValue()]; } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Longs.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Longs.java index 0584ed1ba4..638ecd03e0 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Longs.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Longs.java @@ -26,7 +26,7 @@ public class Longs extends Generator { public Longs(String name, GeneratorConfig config) { - super(LongType.instance, config, name); + super(LongType.instance, config, name, Long.class); } @Override diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Sets.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Sets.java index 48bf293548..82462868d9 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Sets.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Sets.java @@ -32,7 +32,7 @@ public class Sets extends Generator public Sets(String name, Generator valueType, GeneratorConfig config) { - super(SetType.getInstance(valueType.type), config, name); + super(SetType.getInstance(valueType.type), config, name, Set.class); this.valueType = valueType; } 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 e01ff20b36..71aaae6019 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 @@ -23,15 +23,16 @@ 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; public class Strings extends Generator { private final char[] chars; - private final Random rnd = new Random(); + private final FasterRandom rnd = new FasterRandom(); public Strings(String name, GeneratorConfig config) { - super(UTF8Type.instance, config, name); + super(UTF8Type.instance, config, name, String.class); chars = new char[(int) sizeDistribution.maxValue()]; } @@ -42,8 +43,11 @@ public class Strings extends Generator sizeDistribution.setSeed(seed); rnd.setSeed(~seed); int size = (int) sizeDistribution.next(); - for (int i = 0 ; i < size ; i++) - chars[i] = (char) (32 +rnd.nextInt(128-32)); + for (int i = 0; i < size; ) + for (long v = rnd.nextLong(), + n = Math.min(size - i, Long.SIZE/Byte.SIZE); + n-- > 0; v >>= Byte.SIZE) + chars[i++] = (char) (((v & 127) + 32) & 127); return new String(chars, 0, size); } } 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 714959de54..efe4b79c38 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 @@ -33,7 +33,7 @@ public class TimeUUIDs extends Generator public TimeUUIDs(String name, GeneratorConfig config) { - super(TimeUUIDType.instance, config, name); + super(TimeUUIDType.instance, config, name, UUID.class); dateGen = new Dates(name, config); clockSeqAndNode = config.salt; } diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/UUIDs.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/UUIDs.java index e8d6501c3d..faa58c6738 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generate/values/UUIDs.java +++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/UUIDs.java @@ -28,7 +28,7 @@ public class UUIDs extends Generator { public UUIDs(String name, GeneratorConfig config) { - super(UUIDType.instance, config, name); + super(UUIDType.instance, config, name, UUID.class); } @Override 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 f794e75a15..b7d1ee78b7 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 @@ -81,4 +81,9 @@ public class CqlCounterAdder extends CqlOperation { return new CqlRunOpAlwaysSucceed(client, query, queryId, params, key, 1); } + + public boolean isWrite() + { + return true; + } } 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 c422f2b165..622eb14c45 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 @@ -76,4 +76,9 @@ public class CqlInserter extends CqlOperation { return new CqlRunOpAlwaysSucceed(client, query, queryId, params, key, 1); } + + public boolean isWrite() + { + return true; + } } 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 7f6412b3ba..dba2e51939 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 @@ -174,7 +174,7 @@ public abstract class PredefinedOperation extends Operation protected List getColumnValues(ColumnSelection columns) { - Row row = partitions.get(0).iterator(1).batch(1f).iterator().next(); + Row row = partitions.get(0).iterator(1, false).next().iterator().next(); ByteBuffer[] r = new ByteBuffer[columns.count()]; int c = 0; if (columns.indices != null) 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 ee766c36d8..4ee42e9f8a 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 @@ -43,6 +43,11 @@ public class ThriftCounterAdder extends PredefinedOperation this.counteradd = counteradd.get(); } + public boolean isWrite() + { + return true; + } + public void run(final ThriftClient client) throws IOException { List columns = new ArrayList<>(); 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 5c2acfe22f..2500c2ede5 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 @@ -42,6 +42,11 @@ public final class ThriftInserter extends PredefinedOperation super(Command.WRITE, timer, generator, settings); } + public boolean isWrite() + { + return true; + } + public void run(final ThriftClient client) throws IOException { final ByteBuffer key = getKey(); 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 673dafe415..ffa965fcde 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 @@ -45,15 +45,13 @@ public class SchemaInsert extends SchemaStatement { private final BatchStatement.Type batchType; - private final RatioDistribution perVisit; - private final RatioDistribution perBatch; + private final RatioDistribution selectChance; - public SchemaInsert(Timer timer, PartitionGenerator generator, StressSettings settings, Distribution partitionCount, RatioDistribution perVisit, RatioDistribution perBatch, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, BatchStatement.Type batchType) + public SchemaInsert(Timer timer, PartitionGenerator generator, StressSettings settings, Distribution batchSize, RatioDistribution selectChance, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, BatchStatement.Type batchType) { - super(timer, generator, settings, partitionCount, statement, thriftId, cl, ValidationType.NOT_FAIL); + super(timer, generator, settings, batchSize, statement, thriftId, cl, ValidationType.NOT_FAIL); this.batchType = batchType; - this.perVisit = perVisit; - this.perBatch = perBatch; + this.selectChance = selectChance; } private class JavaDriverRun extends Runner @@ -69,43 +67,42 @@ public class SchemaInsert extends SchemaStatement { Partition.RowIterator[] iterators = new Partition.RowIterator[partitions.size()]; for (int i = 0 ; i < iterators.length ; i++) - iterators[i] = partitions.get(i).iterator(perVisit.next()); + iterators[i] = partitions.get(i).iterator(selectChance.next(), true); List stmts = new ArrayList<>(); partitionCount = partitions.size(); - boolean done; - do + for (Partition.RowIterator iterator : iterators) { - done = true; - stmts.clear(); - for (Partition.RowIterator iterator : iterators) - { - if (iterator.done()) - continue; + if (iterator.done()) + continue; - for (Row row : iterator.batch(perBatch.next())) - stmts.add(bindRow(row)); - - done &= iterator.done(); - } - - rowCount += stmts.size(); + 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 + for (int j = 0 ; j < stmts.size() ; j += 65535) + { + List substmts = stmts.subList(j, Math.min(stmts.size(), j + 65535)); Statement stmt; if (stmts.size() == 1) { - stmt = stmts.get(0); + stmt = substmts.get(0); } else { BatchStatement batch = new BatchStatement(batchType); batch.setConsistencyLevel(JavaDriverClient.from(cl)); - batch.addAll(stmts); + batch.addAll(substmts); stmt = batch; } - validate(client.getSession().execute(stmt)); - } while (!done); + validate(client.getSession().execute(stmt)); + } + + for (Partition.RowIterator iterator : iterators) + iterator.markWriteFinished(); return true; } @@ -124,27 +121,23 @@ public class SchemaInsert extends SchemaStatement { Partition.RowIterator[] iterators = new Partition.RowIterator[partitions.size()]; for (int i = 0 ; i < iterators.length ; i++) - iterators[i] = partitions.get(i).iterator(perVisit.next()); + iterators[i] = partitions.get(i).iterator(selectChance.next(), true); partitionCount = partitions.size(); - boolean done; - do + for (Partition.RowIterator iterator : iterators) { - done = true; - for (Partition.RowIterator iterator : iterators) + if (iterator.done()) + continue; + + for (Row row : iterator.next()) { - if (iterator.done()) - continue; - - for (Row row : iterator.batch(perBatch.next())) - { - validate(client.execute_prepared_cql3_query(thriftId, iterator.partition().getToken(), thriftRowArgs(row), settings.command.consistencyLevel)); - rowCount += 1; - } - - done &= iterator.done(); + validate(client.execute_prepared_cql3_query(thriftId, iterator.partition().getToken(), thriftRowArgs(row), settings.command.consistencyLevel)); + rowCount += 1; } - } while (!done); + } + + for (Partition.RowIterator iterator : iterators) + iterator.markWriteFinished(); return true; } @@ -156,6 +149,11 @@ public class SchemaInsert extends SchemaStatement timeWithRetry(new JavaDriverRun(client)); } + public boolean isWrite() + { + return true; + } + @Override public void run(ThriftClient client) throws IOException { 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 a047261889..866f6aba17 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 @@ -22,11 +22,18 @@ package org.apache.cassandra.stress.operations.userdefined; 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.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.settings.StressSettings; import org.apache.cassandra.stress.settings.ValidationType; @@ -39,19 +46,21 @@ import org.apache.cassandra.thrift.ThriftConversion; public class SchemaQuery extends SchemaStatement { - public SchemaQuery(Timer timer, PartitionGenerator generator, StressSettings settings, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, ValidationType validationType) + public static enum ArgSelect + { + MULTIROW, SAMEROW; + //TODO: FIRSTROW, LASTROW + } + + final ArgSelect argSelect; + 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) { super(timer, generator, settings, OptionDistribution.get("fixed(1)").get(), statement, thriftId, cl, validationType); - } - - int execute(JavaDriverClient client) throws Exception - { - return client.getSession().execute(bindRandom(partitions.get(0))).all().size(); - } - - int execute(ThriftClient client) throws Exception - { - return client.execute_prepared_cql3_query(thriftId, partitions.get(0).getToken(), thriftRandomArgs(partitions.get(0)), ThriftConversion.toThrift(cl)).getRowsSize(); + this.argSelect = argSelect; + randomBuffer = new Object[argumentIndex.length][argumentIndex.length]; } private class JavaDriverRun extends Runner @@ -65,7 +74,7 @@ public class SchemaQuery extends SchemaStatement public boolean run() throws Exception { - ResultSet rs = client.getSession().execute(bindRandom(partitions.get(0))); + ResultSet rs = client.getSession().execute(bindArgs(partitions.get(0))); validate(rs); rowCount = rs.all().size(); partitionCount = Math.min(1, rowCount); @@ -84,7 +93,7 @@ public class SchemaQuery extends SchemaStatement public boolean run() throws Exception { - CqlResult rs = client.execute_prepared_cql3_query(thriftId, partitions.get(0).getToken(), thriftRandomArgs(partitions.get(0)), ThriftConversion.toThrift(cl)); + CqlResult rs = client.execute_prepared_cql3_query(thriftId, partitions.get(0).getToken(), thriftArgs(partitions.get(0)), ThriftConversion.toThrift(cl)); validate(rs); rowCount = rs.getRowsSize(); partitionCount = Math.min(1, rowCount); @@ -92,6 +101,64 @@ public class SchemaQuery extends SchemaStatement } } + private int fillRandom(Partition partition) + { + int c = 0; + while (c == 0) + { + 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; + } + } + return c; + } + + BoundStatement bindArgs(Partition partition) + { + switch (argSelect) + { + case MULTIROW: + int c = fillRandom(partition); + for (int i = 0 ; i < argumentIndex.length ; i++) + { + int argIndex = argumentIndex[i]; + bindBuffer[i] = randomBuffer[argIndex < 0 ? 0 : random.nextInt(c)][i]; + } + return statement.bind(bindBuffer); + case SAMEROW: + for (Row row : partition.iterator(1, false).next()) + return bindRow(row); + default: + throw new IllegalStateException(); + } + } + + List thriftArgs(Partition partition) + { + switch (argSelect) + { + case MULTIROW: + List args = new ArrayList<>(); + int c = fillRandom(partition); + 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])); + } + return args; + case SAMEROW: + for (Row row : partition.iterator(1, false).next()) + return thriftRowArgs(row); + default: + throw new IllegalStateException(); + } + } + @Override public void run(JavaDriverClient client) throws IOException { 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 2e0170c087..1f7ed80eb9 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 @@ -24,7 +24,6 @@ package org.apache.cassandra.stress.operations.userdefined; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import java.util.Random; @@ -40,8 +39,6 @@ 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; -import org.apache.cassandra.stress.util.JavaDriverClient; -import org.apache.cassandra.stress.util.ThriftClient; import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.CqlResult; import org.apache.cassandra.transport.SimpleClient; @@ -50,14 +47,12 @@ public abstract class SchemaStatement extends Operation { final PartitionGenerator generator; - private final PreparedStatement statement; + final PreparedStatement statement; final Integer thriftId; final ConsistencyLevel cl; final ValidationType validationType; - private final int[] argumentIndex; - private final Object[] bindBuffer; - private final Object[][] randomBuffer; - private final Random random = new Random(); + final int[] argumentIndex; + final Object[] bindBuffer; public SchemaStatement(Timer timer, PartitionGenerator generator, StressSettings settings, Distribution partitionCount, PreparedStatement statement, Integer thriftId, ConsistencyLevel cl, ValidationType validationType) @@ -70,41 +65,19 @@ public abstract class SchemaStatement extends Operation this.validationType = validationType; argumentIndex = new int[statement.getVariables().size()]; bindBuffer = new Object[argumentIndex.length]; - randomBuffer = new Object[argumentIndex.length][argumentIndex.length]; int i = 0; for (ColumnDefinitions.Definition definition : statement.getVariables()) argumentIndex[i++] = generator.indexOf(definition.getName()); } - private int filLRandom(Partition partition) - { - int c = 0; - for (Row row : partition.iterator(randomBuffer.length).batch(1f)) - { - Object[] randomRow = randomBuffer[c++]; - for (int i = 0 ; i < argumentIndex.length ; i++) - randomRow[i] = row.get(argumentIndex[i]); - if (c >= randomBuffer.length) - break; - } - return c; - } - - BoundStatement bindRandom(Partition partition) - { - int c = filLRandom(partition); - for (int i = 0 ; i < argumentIndex.length ; i++) - { - int argIndex = argumentIndex[i]; - bindBuffer[i] = randomBuffer[argIndex < 0 ? 0 : random.nextInt(c)][i]; - } - return statement.bind(bindBuffer); - } - BoundStatement bindRow(Row row) { for (int i = 0 ; i < argumentIndex.length ; i++) + { bindBuffer[i] = row.get(argumentIndex[i]); + if (bindBuffer[i] == null && !generator.permitNulls(argumentIndex[i])) + throw new IllegalStateException(); + } return statement.bind(bindBuffer); } @@ -116,18 +89,6 @@ public abstract class SchemaStatement extends Operation return args; } - List thriftRandomArgs(Partition partition) - { - List args = new ArrayList<>(); - int c = filLRandom(partition); - 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])); - } - return args; - } - void validate(ResultSet rs) { switch (validationType) diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/CliOption.java b/tools/stress/src/org/apache/cassandra/stress/settings/CliOption.java index 0e8ff1b441..4d7c03994e 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/CliOption.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/CliOption.java @@ -26,7 +26,8 @@ import java.util.Map; public enum CliOption { - KEY("Key details such as size in bytes and value distribution", SettingsKey.helpPrinter()), + POP("Population distribution and intra-partition visit order", SettingsPopulation.helpPrinter()), + INSERT("Insert specific options relating to various methods for batching and splitting partition updates", SettingsInsert.helpPrinter()), COL("Column details such as size and count distribution, data generator, names, comparator and if super columns should be used", SettingsColumn.helpPrinter()), RATE("Thread count, rate limit or automatic mode (default is auto)", SettingsRate.helpPrinter()), MODE("Thrift or CQL with options", SettingsMode.helpPrinter()), 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 70a85aee1e..ef3dbb1cee 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java @@ -25,6 +25,8 @@ import java.util.*; 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; @@ -38,6 +40,14 @@ import org.apache.commons.math3.random.JDKRandomGenerator; public class OptionDistribution extends Option { + public static final Function BUILDER = new Function() + { + public DistributionFactory apply(String s) + { + return get(s); + } + }; + private static final Pattern FULL = Pattern.compile("(~?)([A-Z]+)\\((.+)\\)", Pattern.CASE_INSENSITIVE); private static final Pattern ARGS = Pattern.compile("[^,]+"); @@ -45,12 +55,19 @@ public class OptionDistribution extends Option private String spec; private final String defaultSpec; private final String description; + private final boolean required; public OptionDistribution(String prefix, String defaultSpec, String description) + { + this(prefix, defaultSpec, description, defaultSpec == null); + } + + public OptionDistribution(String prefix, String defaultSpec, String description, boolean required) { this.prefix = prefix; this.defaultSpec = defaultSpec; this.description = description; + this.required = required; } @Override @@ -82,13 +99,13 @@ public class OptionDistribution extends Option public DistributionFactory get() { - return spec != null ? get(spec) : get(defaultSpec); + return spec != null ? get(spec) : defaultSpec != null ? get(defaultSpec) : null; } @Override public boolean happy() { - return spec != null || defaultSpec != null; + return !required || spec != null; } public String longDisplay() @@ -102,12 +119,13 @@ public class OptionDistribution extends Option return Arrays.asList( GroupedOptions.formatMultiLine("EXP(min..max)", "An exponential distribution over the range [min..max]"), GroupedOptions.formatMultiLine("EXTREME(min..max,shape)", "An extreme value (Weibull) distribution over the range [min..max]"), + GroupedOptions.formatMultiLine("QEXTREME(min..max,shape,quantas)", "An extreme value, split into quantas, within which the chance of selection is uniform"), GroupedOptions.formatMultiLine("GAUSSIAN(min..max,stdvrng)", "A gaussian/normal distribution, where mean=(min+max)/2, and stdev is (mean-min)/stdvrng"), GroupedOptions.formatMultiLine("GAUSSIAN(min..max,mean,stdev)", "A gaussian/normal distribution, with explicitly defined mean and stdev"), GroupedOptions.formatMultiLine("UNIFORM(min..max)", "A uniform distribution over the range [min, max]"), GroupedOptions.formatMultiLine("FIXED(val)", "A fixed distribution, always returning the same value"), "Preceding the name with ~ will invert the distribution, e.g. ~exp(1..10) will yield 10 most, instead of least, often", - "Aliases: extr, gauss, normal, norm, weibull" + "Aliases: extr, qextr, gauss, normal, norm, weibull" ); } @@ -128,7 +146,9 @@ public class OptionDistribution extends Option final Map lookup = new HashMap<>(); lookup.put("exp", new ExponentialImpl()); lookup.put("extr", new ExtremeImpl()); - lookup.put("extreme", lookup.get("extreme")); + lookup.put("qextr", new QuantizedExtremeImpl()); + lookup.put("extreme", lookup.get("extr")); + lookup.put("qextreme", lookup.get("qextr")); lookup.put("weibull", lookup.get("weibull")); lookup.put("gaussian", new GaussianImpl()); lookup.put("normal", lookup.get("gaussian")); @@ -245,6 +265,32 @@ public class OptionDistribution extends Option } } + private static final class QuantizedExtremeImpl implements Impl + { + @Override + public DistributionFactory getFactory(List params) + { + if (params.size() != 3) + throw new IllegalArgumentException("Invalid parameter list for quantized extreme (Weibull) distribution: " + params); + try + { + String[] bounds = params.get(0).split("\\.\\.+"); + final long min = parseLong(bounds[0]); + final long max = parseLong(bounds[1]); + final double shape = Double.parseDouble(params.get(1)); + final int quantas = Integer.parseInt(params.get(2)); + WeibullDistribution findBounds = new WeibullDistribution(shape, 1d); + // max probability should be roughly equal to accuracy of (max-min) to ensure all values are visitable, + // over entire range, but this results in overly skewed distribution, so take sqrt + final double scale = (max - min) / findBounds.inverseCumulativeProbability(1d - Math.sqrt(1d/(max-min))); + return new QuantizedExtremeFactory(min, max, shape, scale, quantas); + } catch (Exception _) + { + throw new IllegalArgumentException("Invalid parameter list for quantized extreme (Weibull) distribution: " + params); + } + } + } + private static final class UniformImpl implements Impl { @@ -319,7 +365,7 @@ public class OptionDistribution extends Option } } - private static final class ExtremeFactory implements DistributionFactory + private static class ExtremeFactory implements DistributionFactory { final long min, max; final double shape, scale; @@ -338,6 +384,22 @@ public class OptionDistribution extends Option } } + private static final class QuantizedExtremeFactory extends ExtremeFactory + { + final int quantas; + private QuantizedExtremeFactory(long min, long max, double shape, double scale, int quantas) + { + super(min, max, shape, scale); + this.quantas = quantas; + } + + @Override + public Distribution get() + { + return new DistributionQuantized(new DistributionOffsetApache(new WeibullDistribution(new JDKRandomGenerator(), shape, scale, WeibullDistribution.DEFAULT_INVERSE_ABSOLUTE_ACCURACY), min, max), quantas); + } + } + private static final class GaussianFactory implements DistributionFactory { final long min, max; 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 2459c204b6..aacb616022 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionRatioDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionRatioDistribution.java @@ -29,6 +29,7 @@ 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; @@ -50,16 +51,29 @@ import org.apache.cassandra.stress.generate.RatioDistributionFactory; public class OptionRatioDistribution extends Option { + public static final Function BUILDER = new Function() + { + public RatioDistributionFactory apply(String s) + { + return get(s); + } + }; + private static final Pattern FULL = Pattern.compile("(.*)/([0-9]+[KMB]?)", Pattern.CASE_INSENSITIVE); final OptionDistribution delegate; private double divisor; - - private static final RatioDistribution DEFAULT = new RatioDistribution(new DistributionFixed(1), 1); + final String defaultSpec; public OptionRatioDistribution(String prefix, String defaultSpec, String description) { - delegate = new OptionDistribution(prefix, defaultSpec, description); + this(prefix, defaultSpec, description, defaultSpec != null); + } + + public OptionRatioDistribution(String prefix, String defaultSpec, String description, boolean required) + { + delegate = new OptionDistribution(prefix, null, description, required); + this.defaultSpec = defaultSpec; } @Override @@ -74,7 +88,7 @@ public class OptionRatioDistribution extends Option public static RatioDistributionFactory get(String spec) { - OptionRatioDistribution opt = new OptionRatioDistribution("", "", ""); + OptionRatioDistribution opt = new OptionRatioDistribution("", "", "", true); if (!opt.accept(spec)) throw new IllegalArgumentException(); return opt.get(); @@ -82,7 +96,14 @@ public class OptionRatioDistribution extends Option public RatioDistributionFactory get() { - return !delegate.setByUser() ? new DefaultFactory() : new DelegateFactory(delegate.get(), divisor); + if (delegate.setByUser()) + return new DelegateFactory(delegate.get(), divisor); + if (defaultSpec == null) + return null; + OptionRatioDistribution sub = new OptionRatioDistribution(delegate.prefix, null, null, true); + if (!sub.accept(defaultSpec)) + throw new IllegalStateException("Invalid default spec: " + defaultSpec); + return sub.get(); } @Override @@ -124,15 +145,6 @@ public class OptionRatioDistribution extends Option // factories - private static final class DefaultFactory implements RatioDistributionFactory - { - @Override - public RatioDistribution get() - { - return DEFAULT; - } - } - private static final class DelegateFactory implements RatioDistributionFactory { final DistributionFactory delegate; 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 032f00c3b0..59accb9b8a 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommand.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommand.java @@ -26,6 +26,7 @@ import java.util.Arrays; import java.util.List; import java.util.Map; +import org.apache.cassandra.stress.generate.SeedManager; import org.apache.cassandra.stress.operations.OpDistributionFactory; import org.apache.cassandra.thrift.ConsistencyLevel; @@ -35,8 +36,6 @@ public abstract class SettingsCommand implements Serializable public final Command type; public final long count; - public final int tries; - public final boolean ignoreErrors; public final boolean noWarmup; public final ConsistencyLevel consistencyLevel; public final double targetUncertainty; @@ -56,8 +55,6 @@ public abstract class SettingsCommand implements Serializable public SettingsCommand(Command type, Options options, Count count, Uncertainty uncertainty) { this.type = type; - this.tries = Math.max(1, Integer.parseInt(options.retries.value()) + 1); - this.ignoreErrors = options.ignoreErrors.setByUser(); this.consistencyLevel = ConsistencyLevel.valueOf(options.consistencyLevel.value().toUpperCase()); this.noWarmup = options.noWarmup.setByUser(); if (count != null) @@ -80,11 +77,8 @@ public abstract class SettingsCommand implements Serializable static abstract class Options extends GroupedOptions { - final OptionSimple retries = new OptionSimple("tries=", "[0-9]+", "9", "Number of tries to perform for each operation before failing", false); - final OptionSimple ignoreErrors = new OptionSimple("ignore_errors", "", null, "Do not print/log errors", false); - final OptionSimple noWarmup = new OptionSimple("no_warmup", "", null, "Do not warmup the process", false); + final OptionSimple noWarmup = new OptionSimple("no-warmup", "", null, "Do not warmup the process", false); final OptionSimple consistencyLevel = new OptionSimple("cl=", "ONE|QUORUM|LOCAL_QUORUM|EACH_QUORUM|ALL|ANY", "ONE", "Consistency level to use", false); - final OptionSimple atOnce = new OptionSimple("at-once=", "[0-9]+", "1000", "Number of keys per operation for multiget", false); } static class Count extends Options @@ -93,7 +87,7 @@ public abstract class SettingsCommand implements Serializable @Override public List options() { - return Arrays.asList(count, retries, noWarmup, ignoreErrors, consistencyLevel, atOnce); + return Arrays.asList(count, noWarmup, consistencyLevel); } } @@ -105,7 +99,7 @@ public abstract class SettingsCommand implements Serializable @Override public List options() { - return Arrays.asList(uncertainty, minMeasurements, maxMeasurements, retries, noWarmup, ignoreErrors, consistencyLevel, atOnce); + return Arrays.asList(uncertainty, minMeasurements, maxMeasurements, noWarmup, 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 ac113d17b7..5a8b604d67 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefined.java @@ -27,6 +27,7 @@ import java.util.List; 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.generate.values.Bytes; import org.apache.cassandra.stress.generate.values.Generator; import org.apache.cassandra.stress.generate.values.GeneratorConfig; @@ -42,14 +43,16 @@ public class SettingsCommandPreDefined extends SettingsCommand { public final DistributionFactory add; + public final int keySize; public OpDistributionFactory getFactory(final StressSettings settings) { + final SeedManager seeds = new SeedManager(settings); return new OpDistributionFactory() { public OpDistribution get(Timer timer) { - return new FixedOpDistribution(PredefinedOperation.operation(type, timer, newGenerator(settings), settings, add)); + return new FixedOpDistribution(PredefinedOperation.operation(type, timer, newGenerator(settings, seeds), settings, add)); } public String desc() @@ -64,23 +67,24 @@ public class SettingsCommandPreDefined extends SettingsCommand }; } - PartitionGenerator newGenerator(StressSettings settings) + PartitionGenerator newGenerator(StressSettings settings, SeedManager seeds) { List names = settings.columns.namestrs; List partitionKey = Collections.singletonList(new HexBytes("key", new GeneratorConfig("randomstrkey", null, - OptionDistribution.get("fixed(" + settings.keys.keySize + ")"), null))); + OptionDistribution.get("fixed(" + keySize + ")"), null))); 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); + return new PartitionGenerator(partitionKey, Collections.emptyList(), columns, PartitionGenerator.Order.ARBITRARY, seeds); } public SettingsCommandPreDefined(Command type, Options options) { super(type, options.parent); add = options.add.get(); + keySize = Integer.parseInt(options.keysize.value()); } // Option Declarations @@ -93,6 +97,7 @@ public class SettingsCommandPreDefined extends SettingsCommand this.parent = parent; } final OptionDistribution add = new OptionDistribution("add=", "fixed(1)", "Distribution of value of counter increments"); + final OptionSimple keysize = new OptionSimple("keysize=", "[0-9]+", "10", "Key size in bytes", false); @Override public List 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 49d3c56322..f383ed9aa0 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java @@ -28,6 +28,7 @@ import java.util.Map; import org.apache.cassandra.stress.Operation; 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.operations.OpDistributionFactory; import org.apache.cassandra.stress.operations.SampledOpDistributionFactory; import org.apache.cassandra.stress.operations.predefined.PredefinedOperation; @@ -55,11 +56,8 @@ public class SettingsCommandPreDefinedMixed extends SettingsCommandPreDefined public OpDistributionFactory getFactory(final StressSettings settings) { - final List> mathPairs = new ArrayList<>(); - for (Map.Entry entry: ratios.entrySet()) - mathPairs.add(new Pair(entry.getKey(),entry.getValue())); - - return new SampledOpDistributionFactory(mathPairs, clustering) + final SeedManager seeds = new SeedManager(settings); + return new SampledOpDistributionFactory(ratios, clustering) { protected Operation get(Timer timer, PartitionGenerator generator, Command key) { @@ -68,7 +66,7 @@ public class SettingsCommandPreDefinedMixed extends SettingsCommandPreDefined protected PartitionGenerator newGenerator() { - return SettingsCommandPreDefinedMixed.this.newGenerator(settings); + return SettingsCommandPreDefinedMixed.this.newGenerator(settings, seeds); } }; } 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 08d538e0f4..d0e32b2295 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java @@ -26,10 +26,14 @@ import java.util.ArrayList; 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; import org.apache.cassandra.stress.generate.PartitionGenerator; +import org.apache.cassandra.stress.generate.SeedManager; import org.apache.cassandra.stress.operations.OpDistributionFactory; import org.apache.cassandra.stress.operations.SampledOpDistributionFactory; import org.apache.cassandra.stress.util.Timer; @@ -59,11 +63,8 @@ public class SettingsCommandUser extends SettingsCommand public OpDistributionFactory getFactory(final StressSettings settings) { - final List> mathPairs = new ArrayList<>(); - for (Map.Entry entry: ratios.entrySet()) - mathPairs.add(new Pair(entry.getKey(), entry.getValue())); - - return new SampledOpDistributionFactory(mathPairs, clustering) + final SeedManager seeds = new SeedManager(settings); + return new SampledOpDistributionFactory(ratios, clustering) { protected Operation get(Timer timer, PartitionGenerator generator, String key) { @@ -74,7 +75,7 @@ public class SettingsCommandUser extends SettingsCommand protected PartitionGenerator newGenerator() { - return profile.newGenerator(settings); + return profile.newGenerator(settings, seeds); } }; } @@ -86,19 +87,14 @@ public class SettingsCommandUser extends SettingsCommand { this.parent = parent; } - final OptionDistribution clustering = new OptionDistribution("clustering=", "GAUSSIAN(1..10)", "Distribution clustering runs of operations of the same kind"); + final OptionDistribution clustering = new OptionDistribution("clustering=", "gaussian(1..10)", "Distribution clustering runs of operations of the same kind"); final OptionSimple profile = new OptionSimple("profile=", ".*", null, "Specify the path to a yaml cql3 profile", false); final OptionAnyProbabilities ops = new OptionAnyProbabilities("ops", "Specify the ratios for inserts/queries to perform; e.g. ops(insert=2,=1) will perform 2 inserts for each query1"); @Override public List options() { - final List