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

Merge branch 'cassandra-2.1.0' into cassandra-2.1 

Conflicts:
	CHANGES.txt
	build.xml
	debian/changelog
	tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java
	tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java
This commit is contained in:
Benedict Elliott Smith 2014-09-07 21:23:52 +07:00
parent 9aff086a48 0580fb2b77
commit 77d0c17574
63 changed files with 2000 additions and 755 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -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)

13
NEWS.txt
View File

@ -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/<PLACEHOLDER> 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.

12
debian/changelog vendored
View File

@ -1,3 +1,15 @@
cassandra (2.1.0) unstable; urgency=medium
* New release
-- Sylvain Lebresne <slebresne@apache.org> Sun, 07 Sep 2014 15:21:41 +0200
cassandra (2.1.0~rc7) unstable; urgency=medium
* New RC release
-- Sylvain Lebresne <slebresne@apache.org> Thu, 28 Aug 2014 16:32:12 +0200
cassandra (2.1.0~rc6) unstable; urgency=medium
* New RC release

20
tools/cqlstress-counter-example.yaml
View File

@ -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)

25
tools/cqlstress-example.yaml
View File

@ -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)

20
tools/cqlstress-insanity-example.yaml
View File

@ -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)

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

@ -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);

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

@ -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> work;
volatile boolean stop = false;
final AtomicLong permits;
public FixedWorkQueue(ArrayBlockingQueue<Work> work)
public FixedWorkManager(long permits)
{
this.work = work;
this.permits = new AtomicLong(permits);
}
@Override
public Work poll()
public int takePermits(int count)
{
if (stop)
return null;
return work.poll();
}
@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> 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
}
}
}

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

@ -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
);
}

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

@ -24,15 +24,18 @@ package org.apache.cassandra.stress;
import com.datastax.driver.core.*;
import com.datastax.driver.core.exceptions.AlreadyExistsException;
import com.google.common.base.Function;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.statements.CreateKeyspaceStatement;
import org.apache.cassandra.exceptions.RequestValidationException;
import org.apache.cassandra.stress.generate.Distribution;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.PartitionGenerator;
import org.apache.cassandra.stress.generate.RatioDistributionFactory;
import org.apache.cassandra.stress.generate.SeedManager;
import org.apache.cassandra.stress.generate.values.Booleans;
import org.apache.cassandra.stress.generate.values.Bytes;
import org.apache.cassandra.stress.generate.values.Generator;
@ -88,7 +91,7 @@ public class StressProfile implements Serializable
public String keyspaceName;
public String tableName;
private Map<String, GeneratorConfig> columnConfigs;
private Map<String, String> queries;
private Map<String, StressYaml.QueryDef> queries;
private Map<String, String> insert;
transient volatile TableMetadata tableMetaData;
@ -97,11 +100,11 @@ public class StressProfile implements Serializable
transient volatile BatchStatement.Type batchType;
transient volatile DistributionFactory partitions;
transient volatile RatioDistributionFactory pervisit;
transient volatile RatioDistributionFactory perbatch;
transient volatile RatioDistributionFactory selectchance;
transient volatile PreparedStatement insertStatement;
transient volatile Integer thriftInsertId;
transient volatile Map<String, SchemaQuery.ArgSelect> argSelects;
transient volatile Map<String, PreparedStatement> queryStatements;
transient volatile Map<String, Integer> thriftQueryIds;
@ -242,13 +245,18 @@ public class StressProfile implements Serializable
ThriftClient tclient = settings.getThriftClient();
Map<String, PreparedStatement> stmts = new HashMap<>();
Map<String, Integer> tids = new HashMap<>();
for (Map.Entry<String, String> e : queries.entrySet())
Map<String, SchemaQuery.ArgSelect> args = new HashMap<>();
for (Map.Entry<String, StressYaml.QueryDef> e : queries.entrySet())
{
stmts.put(e.getKey().toLowerCase(), jclient.prepare(e.getValue()));
tids.put(e.getKey().toLowerCase(), tclient.prepare_cql3_query(e.getValue(), Compression.NONE));
stmts.put(e.getKey().toLowerCase(), jclient.prepare(e.getValue().cql));
tids.put(e.getKey().toLowerCase(), tclient.prepare_cql3_query(e.getValue().cql, Compression.NONE));
args.put(e.getKey().toLowerCase(), e.getValue().fields == null
? SchemaQuery.ArgSelect.MULTIROW
: SchemaQuery.ArgSelect.valueOf(e.getValue().fields.toUpperCase()));
}
thriftQueryIds = tids;
queryStatements = stmts;
argSelects = args;
}
catch (TException e)
{
@ -260,7 +268,9 @@ public class StressProfile implements Serializable
// TODO validation
name = name.toLowerCase();
return new SchemaQuery(timer, generator, settings, thriftQueryIds.get(name), queryStatements.get(name), ThriftConversion.fromThrift(settings.command.consistencyLevel), ValidationType.NOT_FAIL);
if (!queryStatements.containsKey(name))
throw new IllegalArgumentException("No query defined with name " + name);
return new SchemaQuery(timer, generator, settings, thriftQueryIds.get(name), queryStatements.get(name), ThriftConversion.fromThrift(settings.command.consistencyLevel), ValidationType.NOT_FAIL, argSelects.get(name));
}
public SchemaInsert getInsert(Timer timer, PartitionGenerator generator, StressSettings settings)
@ -328,18 +338,37 @@ public class StressProfile implements Serializable
insert = new HashMap<>();
lowerCase(insert);
partitions = OptionDistribution.get(!insert.containsKey("partitions") ? "fixed(1)" : insert.remove("partitions"));
pervisit = OptionRatioDistribution.get(!insert.containsKey("pervisit") ? "fixed(1)/1" : insert.remove("pervisit"));
perbatch = OptionRatioDistribution.get(!insert.containsKey("perbatch") ? "fixed(1)/1" : insert.remove("perbatch"));
batchType = !insert.containsKey("batchtype") ? BatchStatement.Type.LOGGED : BatchStatement.Type.valueOf(insert.remove("batchtype"));
partitions = select(settings.insert.batchsize, "partitions", "fixed(1)", insert, OptionDistribution.BUILDER);
selectchance = select(settings.insert.selectRatio, "select", "fixed(1)/1", insert, OptionRatioDistribution.BUILDER);
batchType = settings.insert.batchType != null
? settings.insert.batchType
: !insert.containsKey("batchtype")
? BatchStatement.Type.LOGGED
: BatchStatement.Type.valueOf(insert.remove("batchtype"));
if (!insert.isEmpty())
throw new IllegalArgumentException("Unrecognised insert option(s): " + insert);
Distribution visits = settings.insert.visits.get();
// these min/max are not absolutely accurate if selectchance < 1, but they're close enough to
// guarantee the vast majority of actions occur in these bounds
double minBatchSize = selectchance.get().min() * partitions.get().minValue() * generator.minRowCount * (1d / visits.maxValue());
double maxBatchSize = selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount * (1d / visits.minValue());
System.out.printf("Generating batches with [%d..%d] partitions and [%.0f..%.0f] rows (of [%.0f..%.0f] total rows in the partitions)\n",
partitions.get().minValue(), partitions.get().maxValue(),
minBatchSize, maxBatchSize,
partitions.get().minValue() * generator.minRowCount,
partitions.get().maxValue() * generator.maxRowCount);
if (generator.maxRowCount > 100 * 1000 * 1000)
System.err.printf("WARNING: You have defined a schema that permits very large partitions (%.0f max rows (>100M))\n", generator.maxRowCount);
if (perbatch.get().max() * pervisit.get().max() * partitions.get().maxValue() * generator.maxRowCount > 100000)
if (batchType == BatchStatement.Type.LOGGED && maxBatchSize > 65535)
{
System.err.printf("ERROR: You have defined a workload that generates batches with more than 65k rows (%.0f), but have required the use of LOGGED batches. There is a 65k row limit on a single batch.\n",
selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount);
System.exit(1);
}
if (maxBatchSize > 100000)
System.err.printf("WARNING: You have defined a schema that permits very large batches (%.0f max rows (>100K)). This may OOM this stress client, or the server.\n",
perbatch.get().max() * pervisit.get().max() * partitions.get().maxValue() * generator.maxRowCount);
selectchance.get().max() * partitions.get().maxValue() * generator.maxRowCount);
JavaDriverClient client = settings.getJavaDriverClient();
String query = sb.toString();
@ -356,10 +385,20 @@ public class StressProfile implements Serializable
}
}
return new SchemaInsert(timer, generator, settings, partitions.get(), pervisit.get(), perbatch.get(), thriftInsertId, insertStatement, ThriftConversion.fromThrift(settings.command.consistencyLevel), batchType);
return new SchemaInsert(timer, generator, settings, partitions.get(), selectchance.get(), thriftInsertId, insertStatement, ThriftConversion.fromThrift(settings.command.consistencyLevel), batchType);
}
public PartitionGenerator newGenerator(StressSettings settings)
private static <E> E select(E first, String key, String defValue, Map<String, String> map, Function<String, E> builder)
{
String val = map.remove(key);
if (first != null)
return first;
if (val != null)
return builder.apply(val);
return builder.apply(defValue);
}
public PartitionGenerator newGenerator(StressSettings settings, SeedManager seeds)
{
if (generatorFactory == null)
{
@ -371,7 +410,7 @@ public class StressProfile implements Serializable
}
}
return generatorFactory.newGenerator();
return generatorFactory.newGenerator(settings, seeds);
}
private class GeneratorFactory
@ -393,9 +432,9 @@ public class StressProfile implements Serializable
valueColumns.add(new ColumnInfo(metadata.getName(), metadata.getType(), columnConfigs.get(metadata.getName())));
}
PartitionGenerator newGenerator()
PartitionGenerator newGenerator(StressSettings settings, SeedManager seeds)
{
return new PartitionGenerator(get(partitionKeys), get(clusteringColumns), get(valueColumns));
return new PartitionGenerator(get(partitionKeys), get(clusteringColumns), get(valueColumns), settings.generate.order, seeds);
}
List<Generator> get(List<ColumnInfo> columnInfos)

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

@ -30,8 +30,14 @@ public class StressYaml
public String table;
public String table_definition;
public List<Map<String,Object>> columnspec;
public Map<String,String> queries;
public Map<String,String> insert;
public List<Map<String, Object>> columnspec;
public Map<String, QueryDef> queries;
public Map<String, String> insert;
public static class QueryDef
{
public String cql;
public String fields;
}
}

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

@ -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);
}
}

90
tools/stress/src/org/apache/cassandra/stress/generate/DistributionQuantized.java Normal file
View File

@ -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)"});
}
}

116
tools/stress/src/org/apache/cassandra/stress/generate/FasterRandom.java Normal file
View File

@ -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();
}
}

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

@ -23,24 +23,34 @@ package org.apache.cassandra.stress.generate;
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Deque;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.stress.generate.values.Generator;
// a partition is re-used to reduce garbage generation, as is its internal RowIterator
// TODO: we should batch the generation of clustering components so we can bound the time and size necessary to
// generate huge partitions with only a small number of clustering components; i.e. we should generate seeds for batches
// of a single component, and then generate the values within those batches as necessary. this will be difficult with
// generating sorted partitions, and may require generator support (e.g. we may need to support generating prefixes
// that are extended/suffixed to generate each batch, so that we can sort the prefixes)
public class Partition
{
private long idseed;
private Seed seed;
private final Object[] partitionKey;
private final PartitionGenerator generator;
private final RowIterator iterator;
@ -55,31 +65,32 @@ public class Partition
iterator = new SingleRowIterator();
}
void setSeed(long seed)
void setSeed(Seed seed)
{
long idseed = 0;
for (int i = 0 ; i < partitionKey.length ; i++)
{
Generator generator = this.generator.partitionKey.get(i);
// set the partition key seed based on the current work item we're processing
generator.setSeed(seed);
generator.setSeed(seed.seed);
Object key = generator.generate();
partitionKey[i] = key;
// then contribute this value to the data seed
idseed = seed(key, generator.type, idseed);
}
this.seed = seed;
this.idseed = idseed;
}
public RowIterator iterator(double useChance)
public RowIterator iterator(double useChance, boolean isWrite)
{
iterator.reset(useChance, 0);
iterator.reset(useChance, 0, 1, isWrite);
return iterator;
}
public RowIterator iterator(int targetCount)
public RowIterator iterator(int targetCount, boolean isWrite)
{
iterator.reset(Double.NaN, targetCount);
iterator.reset(Double.NaN, targetCount, 1, isWrite);
return iterator;
}
@ -87,12 +98,12 @@ public class Partition
{
boolean done;
void reset(double useChance, int targetCount)
void reset(double useChance, int targetCount, int batches, boolean isWrite)
{
done = false;
}
public Iterable<Row> batch(double ratio)
public Iterable<Row> next()
{
if (done)
return Collections.emptyList();
@ -110,6 +121,12 @@ public class Partition
{
return done;
}
public void markWriteFinished()
{
assert done;
generator.seeds.markFinished(seed);
}
}
public abstract class RowIterator
@ -117,10 +134,10 @@ public class Partition
// we reuse the row object to save garbage
final Row row = new Row(partitionKey, new Object[generator.clusteringComponents.size() + generator.valueComponents.size()]);
public abstract Iterable<Row> batch(double ratio);
abstract void reset(double useChance, int targetCount);
public abstract Iterable<Row> next();
public abstract boolean done();
public abstract void markWriteFinished();
abstract void reset(double useChance, int targetCount, int batches, boolean isWrite);
public Partition partition()
{
@ -128,31 +145,40 @@ public class Partition
}
}
// permits iterating a random subset of the procedurally generated rows in this partition; this is the only mechanism for visiting rows
// permits iterating a random subset of the procedurally generated rows in this partition. this is the only mechanism for visiting rows.
// we maintain a stack of clustering components and their seeds; for each clustering component we visit, we generate all values it takes at that level,
// and then, using the average (total) number of children it takes we randomly choose whether or not we visit its children;
// if we do, we generate all possible values the children can take, and repeat the process. So at any one time we are using space proportional
// if we do, we generate all possible values the immediate children can take, and repeat the process. So at any one time we are using space proportional
// to C.N, where N is the average number of values each clustering component takes, as opposed to N^C total values in the partition.
// TODO : guarantee at least one row is always returned
// TODO : support first/last row, and constraining reads to rows we know are populated
class MultiRowIterator extends RowIterator
{
// probability any single row will be generated in this iteration
double useChance;
double expectedRowCount;
// the current seed in use at any given level; used to save recalculating it for each row, so we only need to recalc
// from prior row
// the seed used to generate the current values for the clustering components at each depth;
// used to save recalculating it for each row, so we only need to recalc from prior row.
final long[] clusteringSeeds = new long[generator.clusteringComponents.size()];
// the components remaining to be visited for each level of the current stack
final Queue<Object>[] clusteringComponents = new ArrayDeque[generator.clusteringComponents.size()];
final Deque<Object>[] clusteringComponents = new ArrayDeque[generator.clusteringComponents.size()];
// we want our chance of selection to be applied uniformly, so we compound the roll we make at each level
// so that we know with what chance we reached there, and we adjust our roll at that level by that amount
double[] chancemodifier = new double[generator.clusteringComponents.size()];
double[] rollmodifier = new double[generator.clusteringComponents.size()];
final double[] chancemodifier = new double[generator.clusteringComponents.size()];
final double[] rollmodifier = new double[generator.clusteringComponents.size()];
// 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<Object> unique = new HashSet<>();
final List<Comparable> tosort = new ArrayList<>();
final Random random = new Random();
MultiRowIterator()
@ -163,126 +189,262 @@ public class Partition
chancemodifier[0] = generator.clusteringChildAverages[0];
}
void reset(double useChance, int targetCount)
// if we're a write, the expected behaviour is that the requested batch count is compounded with the seed's visit
// count to decide how much we should return in one iteration
void reset(double useChance, int targetCount, int batches, boolean isWrite)
{
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<Object> queue, int count, Generator generator)
{
if (count == 1)
{
queue.add(generator.generate());
return;
}
else
switch (Partition.this.generator.order)
{
unique.clear();
for (int i = 0 ; i < count ; i++)
{
Object next = generator.generate();
if (unique.add(next))
queue.add(next);
}
case SORTED:
if (Comparable.class.isAssignableFrom(generator.clazz))
{
tosort.clear();
for (int i = 0 ; i < count ; i++)
tosort.add((Comparable) generator.generate());
Collections.sort(tosort);
for (int i = 0 ; i < count ; i++)
if (i == 0 || tosort.get(i - 1).compareTo(i) < 0)
queue.add(tosort.get(i));
break;
}
case ARBITRARY:
unique.clear();
for (int i = 0 ; i < count ; i++)
{
Object next = generator.generate();
if (unique.add(next))
queue.add(next);
}
break;
case SHUFFLED:
unique.clear();
tosort.clear();
for (int i = 0 ; i < count ; i++)
{
Object next = generator.generate();
if (unique.add(next))
tosort.add(new RandomOrder(next));
}
Collections.sort(tosort);
for (Object o : tosort)
queue.add(((RandomOrder)o).value);
break;
default:
throw new IllegalStateException();
}
}
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<Row> batch(final double ratio)
public Iterable<Row> next()
{
final double continueChance = 1d - (Math.pow(ratio, expectedRowCount * useChance));
final int[] limit = position.clone();
int remainingSize = batchSize;
for (int i = 0 ; i < limit.length && remainingSize > 0 ; i++)
{
limit[i] += remainingSize / generator.clusteringChildAverages[i];
remainingSize %= generator.clusteringChildAverages[i];
}
assert remainingSize == 0;
for (int i = limit.length - 1 ; i > 0 ; i--)
{
if (limit[i] > generator.clusteringChildAverages[i])
{
limit[i - 1] += limit[i] / generator.clusteringChildAverages[i];
limit[i] %= generator.clusteringChildAverages[i];
}
}
for (int i = 0 ; i < limit.length ; i++)
{
if (limit[i] < this.limit[i])
break;
limit[i] = Math.min(limit[i], this.limit[i]);
}
return new Iterable<Row>()
{
public Iterator<Row> iterator()
{
return new Iterator<Row>()
{
boolean hasNext = true;
public boolean hasNext()
{
return hasNext;
if (done())
return false;
for (int i = 0 ; i < position.length ; i++)
if (position[i] < limit[i])
return true;
return false;
}
public Row next()
{
hasNext = advance(continueChance);
advance();
return row;
}
@ -300,26 +462,37 @@ public class Partition
return clusteringComponents[0].isEmpty();
}
public void markWriteFinished()
{
if (done())
generator.seeds.markFinished(seed);
else
generator.seeds.markVisited(seed, position.clone());
}
public Partition partition()
{
return Partition.this;
}
}
public String getKeyAsString()
private static class RandomOrder implements Comparable<RandomOrder>
{
StringBuilder sb = new StringBuilder();
int i = 0;
for (Object key : partitionKey)
final int order = ThreadLocalRandom.current().nextInt();
final Object value;
private RandomOrder(Object value)
{
if (i > 0)
sb.append("|");
AbstractType type = generator.partitionKey.get(i++).type;
sb.append(type.getString(type.decompose(key)));
this.value = value;
}
public int compareTo(RandomOrder that)
{
return Integer.compare(this.order, that.order);
}
return sb.toString();
}
// calculate a new seed based on the combination of a parent seed and the generated child, to generate
// any children of this child
static long seed(Object object, AbstractType type, long seed)
{
if (object instanceof ByteBuffer)
@ -355,6 +528,20 @@ public class Partition
return partitionKey[i];
}
public String getKeyAsString()
{
StringBuilder sb = new StringBuilder();
int i = 0;
for (Object key : partitionKey)
{
if (i > 0)
sb.append("|");
AbstractType type = generator.partitionKey.get(i++).type;
sb.append(type.getString(type.decompose(key)));
}
return sb.toString();
}
// used for thrift smart routing - if it's a multi-part key we don't try to route correctly right now
public ByteBuffer getToken()
{

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

@ -30,18 +30,27 @@ import java.util.NoSuchElementException;
import com.google.common.collect.Iterables;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.generate.values.Generator;
public class PartitionGenerator
{
public static enum Order
{
ARBITRARY, SHUFFLED, SORTED
}
public final double maxRowCount;
public final double minRowCount;
final List<Generator> partitionKey;
final List<Generator> clusteringComponents;
final List<Generator> valueComponents;
final int[] clusteringChildAverages;
private final Map<String, Integer> indexMap;
final Order order;
final SeedManager seeds;
final List<Partition> recyclable = new ArrayList<>();
int partitionsInUse = 0;
@ -51,18 +60,25 @@ public class PartitionGenerator
partitionsInUse = 0;
}
public PartitionGenerator(List<Generator> partitionKey, List<Generator> clusteringComponents, List<Generator> valueComponents)
public PartitionGenerator(List<Generator> partitionKey, List<Generator> clusteringComponents, List<Generator> valueComponents, Order order, SeedManager seeds)
{
this.partitionKey = partitionKey;
this.clusteringComponents = clusteringComponents;
this.valueComponents = valueComponents;
this.order = order;
this.seeds = seeds;
this.clusteringChildAverages = new int[clusteringComponents.size()];
for (int i = clusteringChildAverages.length - 1 ; i >= 0 ; i--)
clusteringChildAverages[i] = (int) (i < (clusteringChildAverages.length - 1) ? clusteringComponents.get(i + 1).clusteringDistribution.average() * clusteringChildAverages[i + 1] : 1);
double maxRowCount = 1d;
double minRowCount = 1d;
for (Generator component : clusteringComponents)
{
maxRowCount *= component.clusteringDistribution.maxValue();
minRowCount *= component.clusteringDistribution.minValue();
}
this.maxRowCount = maxRowCount;
this.minRowCount = minRowCount;
this.indexMap = new HashMap<>();
int i = 0;
for (Generator generator : partitionKey)
@ -72,6 +88,11 @@ public class PartitionGenerator
indexMap.put(generator.name, i++);
}
public boolean permitNulls(int index)
{
return !(index < 0 || index < clusteringComponents.size());
}
public int indexOf(String name)
{
Integer i = indexMap.get(name);
@ -80,11 +101,14 @@ public class PartitionGenerator
return i;
}
public Partition generate(long seed)
public Partition generate(Operation op)
{
if (recyclable.size() <= partitionsInUse || recyclable.get(partitionsInUse) == null)
recyclable.add(new Partition(this));
Seed seed = seeds.next(op);
if (seed == null)
return null;
Partition partition = recyclable.get(partitionsInUse++);
partition.setSeed(seed);
return partition;

5
tools/stress/src/org/apache/cassandra/stress/generate/RatioDistribution.java
View File

@ -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);

67
tools/stress/src/org/apache/cassandra/stress/generate/Seed.java Normal file
View File

@ -0,0 +1,67 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.stress.generate;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import org.apache.cassandra.stress.util.DynamicList;
public class Seed implements Comparable<Seed>
{
public final long seed;
final int visits;
DynamicList.Node poolNode;
volatile int[] position;
volatile State state = State.HELD;
private static final AtomicReferenceFieldUpdater<Seed, Seed.State> stateUpdater = AtomicReferenceFieldUpdater.newUpdater(Seed.class, State.class, "state");
public int compareTo(Seed that)
{
return Long.compare(this.seed, that.seed);
}
static enum State
{
HELD, AVAILABLE
}
Seed(long seed, int visits)
{
this.seed = seed;
this.visits = visits;
}
boolean take()
{
return stateUpdater.compareAndSet(this, State.AVAILABLE, State.HELD);
}
void yield()
{
state = State.AVAILABLE;
}
public int[] position()
{
return position;
}
}

29
tools/stress/src/org/apache/cassandra/stress/generate/SeedGenerator.java
View File

@ -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);
}

249
tools/stress/src/org/apache/cassandra/stress/generate/SeedManager.java Normal file
View File

@ -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<Seed, Seed> managing = new ConcurrentHashMap<>();
final DynamicList<Seed> 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<Seed, Seed> 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<Seed, Seed> 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);
}
}
}
}

54
tools/stress/src/org/apache/cassandra/stress/generate/SeedRandomGenerator.java
View File

@ -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;
}
}

42
tools/stress/src/org/apache/cassandra/stress/generate/SeedSeriesGenerator.java
View File

@ -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);
}
}

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

@ -26,7 +26,7 @@ public class Booleans extends Generator<Boolean>
{
public Booleans(String name, GeneratorConfig config)
{
super(BooleanType.instance, config, name);
super(BooleanType.instance, config, name, Boolean.class);
}
@Override

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

@ -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<ByteBuffer>
{
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<ByteBuffer>
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));

3
tools/stress/src/org/apache/cassandra/stress/generate/values/Dates.java
View File

@ -30,9 +30,10 @@ public class Dates extends Generator<Date>
{
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()
{

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

@ -26,7 +26,7 @@ public class Doubles extends Generator<Double>
{
public Doubles(String name, GeneratorConfig config)
{
super(DoubleType.instance, config, name);
super(DoubleType.instance, config, name, Double.class);
}
@Override

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

@ -26,7 +26,7 @@ public class Floats extends Generator<Float>
{
public Floats(String name, GeneratorConfig config)
{
super(FloatType.instance, config, name);
super(FloatType.instance, config, name, Float.class);
}
@Override

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

@ -31,15 +31,17 @@ public abstract class Generator<T>
public final String name;
public final AbstractType<T> type;
public final Class<T> clazz;
final long salt;
final Distribution identityDistribution;
final Distribution sizeDistribution;
public final Distribution clusteringDistribution;
public Generator(AbstractType<T> type, GeneratorConfig config, String name)
public Generator(AbstractType<T> type, GeneratorConfig config, String name, Class<T> clazz)
{
this.type = type;
this.name = name;
this.clazz = clazz;
this.salt = config.salt;
this.identityDistribution = config.getIdentityDistribution(defaultIdentityDistribution());
this.sizeDistribution = config.getSizeDistribution(defaultSizeDistribution());

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

@ -31,7 +31,7 @@ public class HexBytes extends Generator<ByteBuffer>
public HexBytes(String name, GeneratorConfig config)
{
super(BytesType.instance, config, name);
super(BytesType.instance, config, name, ByteBuffer.class);
bytes = new byte[(int) sizeDistribution.maxValue()];
}

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

@ -20,8 +20,6 @@
*/
package org.apache.cassandra.stress.generate.values;
import java.util.Random;
import org.apache.cassandra.db.marshal.UTF8Type;
public class HexStrings extends Generator<String>
@ -30,7 +28,7 @@ public class HexStrings extends Generator<String>
public HexStrings(String name, GeneratorConfig config)
{
super(UTF8Type.instance, config, name);
super(UTF8Type.instance, config, name, String.class);
chars = new char[(int) sizeDistribution.maxValue()];
}

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

@ -31,7 +31,7 @@ public class Inets extends Generator<InetAddress>
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];
}

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

@ -27,7 +27,7 @@ public class Integers extends Generator<Integer>
public Integers(String name, GeneratorConfig config)
{
super(Int32Type.instance, config, name);
super(Int32Type.instance, config, name, Integer.class);
}
@Override

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

@ -33,7 +33,7 @@ public class Lists extends Generator<List>
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()];
}

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

@ -26,7 +26,7 @@ public class Longs extends Generator<Long>
{
public Longs(String name, GeneratorConfig config)
{
super(LongType.instance, config, name);
super(LongType.instance, config, name, Long.class);
}
@Override

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

@ -32,7 +32,7 @@ public class Sets extends Generator<Set>
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;
}

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

@ -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<String>
{
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<String>
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);
}
}

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

@ -33,7 +33,7 @@ public class TimeUUIDs extends Generator<UUID>
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;
}

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

@ -28,7 +28,7 @@ public class UUIDs extends Generator<UUID>
{
public UUIDs(String name, GeneratorConfig config)
{
super(UUIDType.instance, config, name);
super(UUIDType.instance, config, name, UUID.class);
}
@Override

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

@ -81,4 +81,9 @@ public class CqlCounterAdder extends CqlOperation<Integer>
{
return new CqlRunOpAlwaysSucceed(client, query, queryId, params, key, 1);
}
public boolean isWrite()
{
return true;
}
}

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

@ -76,4 +76,9 @@ public class CqlInserter extends CqlOperation<Integer>
{
return new CqlRunOpAlwaysSucceed(client, query, queryId, params, key, 1);
}
public boolean isWrite()
{
return true;
}
}

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

@ -174,7 +174,7 @@ public abstract class PredefinedOperation extends Operation
protected List<ByteBuffer> 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)

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

@ -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<CounterColumn> columns = new ArrayList<>();

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

@ -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();

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

@ -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<BoundStatement> 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<BoundStatement> 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
{

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

@ -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<ByteBuffer> thriftArgs(Partition partition)
{
switch (argSelect)
{
case MULTIROW:
List<ByteBuffer> 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
{

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

@ -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<ByteBuffer> thriftRandomArgs(Partition partition)
{
List<ByteBuffer> 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)

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

@ -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()),

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

@ -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<String, DistributionFactory> BUILDER = new Function<String, DistributionFactory>()
{
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<String, Impl> 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<String> 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;

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

@ -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<String, RatioDistributionFactory> BUILDER = new Function<String, RatioDistributionFactory>()
{
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;

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

@ -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<? extends Option> 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<? extends Option> options()
{
return Arrays.asList(uncertainty, minMeasurements, maxMeasurements, retries, noWarmup, ignoreErrors, consistencyLevel, atOnce);
return Arrays.asList(uncertainty, minMeasurements, maxMeasurements, noWarmup, consistencyLevel);
}
}

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

@ -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<String> names = settings.columns.namestrs;
List<Generator> partitionKey = Collections.<Generator>singletonList(new HexBytes("key",
new GeneratorConfig("randomstrkey", null,
OptionDistribution.get("fixed(" + settings.keys.keySize + ")"), null)));
OptionDistribution.get("fixed(" + keySize + ")"), null)));
List<Generator> columns = new ArrayList<>();
for (int i = 0 ; i < settings.columns.maxColumnsPerKey ; i++)
columns.add(new Bytes(names.get(i), new GeneratorConfig("randomstr" + names.get(i), null, settings.columns.sizeDistribution, null)));
return new PartitionGenerator(partitionKey, Collections.<Generator>emptyList(), columns);
return new PartitionGenerator(partitionKey, Collections.<Generator>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<? extends Option> options()

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

@ -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<Pair<Command,Double>> mathPairs = new ArrayList<>();
for (Map.Entry entry: ratios.entrySet())
mathPairs.add(new Pair(entry.getKey(),entry.getValue()));
return new SampledOpDistributionFactory<Command>(mathPairs, clustering)
final SeedManager seeds = new SeedManager(settings);
return new SampledOpDistributionFactory<Command>(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);
}
};
}

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

@ -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<Pair<String,Double>> mathPairs = new ArrayList<>();
for (Map.Entry entry: ratios.entrySet())
mathPairs.add(new Pair(entry.getKey(), entry.getValue()));
return new SampledOpDistributionFactory<String>(mathPairs, clustering)
final SeedManager seeds = new SeedManager(settings);
return new SampledOpDistributionFactory<String>(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,<query1>=1) will perform 2 inserts for each query1");
@Override
public List<? extends Option> options()
{
final List<Option> options = new ArrayList<>();
options.add(clustering);
options.add(ops);
options.add(profile);
options.addAll(parent.options());
return options;
return ImmutableList.<Option>builder().add(ops, clustering, profile).addAll(parent.options()).build();
}
}

92
tools/stress/src/org/apache/cassandra/stress/settings/SettingsErrors.java Normal file
View File

@ -0,0 +1,92 @@
package org.apache.cassandra.stress.settings;
/*
*
* 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.io.File;
import java.io.FileNotFoundException;
import java.io.PrintStream;
import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
public class SettingsErrors implements Serializable
{
public final boolean ignore;
public final int tries;
public SettingsErrors(Options options)
{
ignore = options.ignore.setByUser();
this.tries = Math.max(1, Integer.parseInt(options.retries.value()) + 1);
}
// Option Declarations
public static final class Options extends GroupedOptions
{
final OptionSimple retries = new OptionSimple("retries=", "[0-9]+", "9", "Number of tries to perform for each operation before failing", false);
final OptionSimple ignore = new OptionSimple("ignore", "", null, "Do not fail on errors", false);
@Override
public List<? extends Option> options()
{
return Arrays.asList(retries, ignore);
}
}
// CLI Utility Methods
public static SettingsErrors get(Map<String, String[]> clArgs)
{
String[] params = clArgs.remove("-errors");
if (params == null)
return new SettingsErrors(new Options());
GroupedOptions options = GroupedOptions.select(params, new Options());
if (options == null)
{
printHelp();
System.out.println("Invalid -errors options provided, see output for valid options");
System.exit(1);
}
return new SettingsErrors((Options) options);
}
public static void printHelp()
{
GroupedOptions.printOptions(System.out, "-errors", new Options());
}
public static Runnable helpPrinter()
{
return new Runnable()
{
@Override
public void run()
{
printHelp();
}
};
}
}

103
tools/stress/src/org/apache/cassandra/stress/settings/SettingsInsert.java Normal file
View File

@ -0,0 +1,103 @@
package org.apache.cassandra.stress.settings;
/*
*
* 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.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import com.datastax.driver.core.BatchStatement;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.RatioDistributionFactory;
public class SettingsInsert implements Serializable
{
public final DistributionFactory revisit;
public final DistributionFactory visits;
public final DistributionFactory batchsize;
public final RatioDistributionFactory selectRatio;
public final BatchStatement.Type batchType;
private SettingsInsert(InsertOptions options)
{
this.visits= options.visits.get();
this.revisit = options.revisit.get();
this.batchsize = options.partitions.get();
this.selectRatio = options.selectRatio.get();
this.batchType = !options.batchType.setByUser() ? null : BatchStatement.Type.valueOf(options.batchType.value());
}
// Option Declarations
private static class InsertOptions extends GroupedOptions
{
final OptionDistribution visits = new OptionDistribution("visits=", "fixed(1)", "The target number of inserts to split a partition into; if more than one, the partition will be placed in the revisit set");
final OptionDistribution revisit = new OptionDistribution("revisit=", "uniform(1..1M)", "The distribution with which we revisit partial writes (see visits); implicitly defines size of revisit collection");
final OptionDistribution partitions = new OptionDistribution("partitions=", null, "The number of partitions to update in a single batch", false);
final OptionSimple batchType = new OptionSimple("batchtype=", "unlogged|logged|counter", null, "Specify the type of batch statement (LOGGED, UNLOGGED or COUNTER)", false);
final OptionRatioDistribution selectRatio = new OptionRatioDistribution("select-ratio=", null, "The uniform probability of visiting any CQL row in the generated partition", false);
@Override
public List<? extends Option> options()
{
return Arrays.asList(revisit, visits, partitions, batchType, selectRatio);
}
}
// CLI Utility Methods
public static SettingsInsert get(Map<String, String[]> clArgs)
{
String[] params = clArgs.remove("-insert");
if (params == null)
return new SettingsInsert(new InsertOptions());
InsertOptions options = GroupedOptions.select(params, new InsertOptions());
if (options == null)
{
printHelp();
System.out.println("Invalid -insert options provided, see output for valid options");
System.exit(1);
}
return new SettingsInsert(options);
}
public static void printHelp()
{
GroupedOptions.printOptions(System.out, "-insert", new InsertOptions());
}
public static Runnable helpPrinter()
{
return new Runnable()
{
@Override
public void run()
{
printHelp();
}
};
}
}

153
tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
View File

@ -1,153 +0,0 @@
package org.apache.cassandra.stress.settings;
/*
*
* 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.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.SeedGenerator;
import org.apache.cassandra.stress.generate.SeedRandomGenerator;
import org.apache.cassandra.stress.generate.SeedSeriesGenerator;
// Settings for key generation
public class SettingsKey implements Serializable
{
final int keySize;
private final DistributionFactory distribution;
private final DistributionFactory clustering;
private final long[] range;
public SettingsKey(DistributionOptions options)
{
this.keySize = Integer.parseInt(options.size.value());
this.distribution = options.dist.get();
this.clustering = options.clustering.get();
this.range = null;
}
public SettingsKey(PopulateOptions options)
{
this.keySize = Integer.parseInt(options.size.value());
this.distribution = null;
this.clustering = null;
String[] bounds = options.populate.value().split("\\.\\.+");
this.range = new long[] { OptionDistribution.parseLong(bounds[0]), OptionDistribution.parseLong(bounds[1]) };
}
// Option Declarations
private static final class DistributionOptions extends GroupedOptions
{
final OptionDistribution dist;
final OptionDistribution clustering = new OptionDistribution("cluster=", "fixed(1)", "Keys are clustered in adjacent value runs of this size");
final OptionSimple size = new OptionSimple("size=", "[0-9]+", "10", "Key size in bytes", false);
public DistributionOptions(String defaultLimit)
{
dist = new OptionDistribution("dist=", "GAUSSIAN(1.." + defaultLimit + ")", "Keys are selected from this distribution");
}
@Override
public List<? extends Option> options()
{
return Arrays.asList(dist, size, clustering);
}
}
private static final class PopulateOptions extends GroupedOptions
{
final OptionSimple populate;
final OptionSimple size = new OptionSimple("size=", "[0-9]+", "10", "Key size in bytes", false);
public PopulateOptions(String defaultLimit)
{
populate = new OptionSimple("populate=", "[0-9]+\\.\\.+[0-9]+[MBK]?",
"1.." + defaultLimit,
"Populate all keys in sequence", true);
}
@Override
public List<? extends Option> options()
{
return Arrays.asList(populate, size);
}
}
public SeedGenerator newSeedGenerator()
{
return range == null ? new SeedRandomGenerator(distribution.get(), clustering.get()) : new SeedSeriesGenerator(range[0], range[1]);
}
// CLI Utility Methods
public static SettingsKey get(Map<String, String[]> clArgs, SettingsCommand command)
{
// set default size to number of commands requested, unless set to err convergence, then use 1M
String defaultLimit = command.count <= 0 ? "1000000" : Long.toString(command.count);
String[] params = clArgs.remove("-key");
if (params == null)
{
// return defaults:
switch(command.type)
{
case WRITE:
case COUNTER_WRITE:
return new SettingsKey(new PopulateOptions(defaultLimit));
default:
return new SettingsKey(new DistributionOptions(defaultLimit));
}
}
GroupedOptions options = GroupedOptions.select(params, new PopulateOptions(defaultLimit), new DistributionOptions(defaultLimit));
if (options == null)
{
printHelp();
System.out.println("Invalid -key options provided, see output for valid options");
System.exit(1);
}
return options instanceof PopulateOptions ?
new SettingsKey((PopulateOptions) options) :
new SettingsKey((DistributionOptions) options);
}
public static void printHelp()
{
GroupedOptions.printOptions(System.out, "-key", new PopulateOptions("N"), new DistributionOptions("N"));
}
public static Runnable helpPrinter()
{
return new Runnable()
{
@Override
public void run()
{
printHelp();
}
};
}
}

176
tools/stress/src/org/apache/cassandra/stress/settings/SettingsPopulation.java Normal file
View File

@ -0,0 +1,176 @@
package org.apache.cassandra.stress.settings;
/*
*
* 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.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import com.google.common.collect.ImmutableList;
import org.apache.cassandra.stress.generate.DistributionFactory;
import org.apache.cassandra.stress.generate.PartitionGenerator;
public class SettingsPopulation implements Serializable
{
public final DistributionFactory distribution;
public final DistributionFactory readlookback;
public final PartitionGenerator.Order order;
public final boolean wrap;
public final long[] sequence;
public static enum GenerateOrder
{
ARBITRARY, SHUFFLED, SORTED
}
private SettingsPopulation(GenerateOptions options, DistributionOptions dist, SequentialOptions pop)
{
this.order = !options.contents.setByUser() ? PartitionGenerator.Order.ARBITRARY : PartitionGenerator.Order.valueOf(options.contents.value().toUpperCase());
if (dist != null)
{
this.distribution = dist.seed.get();
this.sequence = null;
this.readlookback = null;
this.wrap = false;
}
else
{
this.distribution = null;
String[] bounds = pop.populate.value().split("\\.\\.+");
this.sequence = new long[] { OptionDistribution.parseLong(bounds[0]), OptionDistribution.parseLong(bounds[1]) };
this.readlookback = pop.lookback.get();
this.wrap = !pop.nowrap.setByUser();
}
}
public SettingsPopulation(DistributionOptions options)
{
this(options, options, null);
}
public SettingsPopulation(SequentialOptions options)
{
this(options, null, options);
}
// Option Declarations
private static class GenerateOptions extends GroupedOptions
{
final OptionSimple contents = new OptionSimple("contents=", "(sorted|shuffled)", null, "SORTED or SHUFFLED (intra-)partition order; if not specified, will be consistent but arbitrary order", false);
@Override
public List<? extends Option> options()
{
return Arrays.asList(contents);
}
}
private static final class DistributionOptions extends GenerateOptions
{
final OptionDistribution seed;
public DistributionOptions(String defaultLimit)
{
seed = new OptionDistribution("dist=", "gaussian(1.." + defaultLimit + ")", "Seeds are selected from this distribution");
}
@Override
public List<? extends Option> options()
{
return ImmutableList.<Option>builder().add(seed).addAll(super.options()).build();
}
}
private static final class SequentialOptions extends GenerateOptions
{
final OptionSimple populate;
final OptionDistribution lookback = new OptionDistribution("read-lookback=", "fixed(1)", "Select read seeds from the recently visited write seeds");
final OptionSimple nowrap = new OptionSimple("no-wrap", "", null, "Terminate the stress test once all seeds in the range have been visited", false);
public SequentialOptions(String defaultLimit)
{
populate = new OptionSimple("seq=", "[0-9]+\\.\\.+[0-9]+[MBK]?",
"1.." + defaultLimit,
"Generate all seeds in sequence", true);
}
@Override
public List<? extends Option> options()
{
return ImmutableList.<Option>builder().add(populate, nowrap, lookback).addAll(super.options()).build();
}
}
// CLI Utility Methods
public static SettingsPopulation get(Map<String, String[]> clArgs, SettingsCommand command)
{
// set default size to number of commands requested, unless set to err convergence, then use 1M
String defaultLimit = command.count <= 0 ? "1000000" : Long.toString(command.count);
String[] params = clArgs.remove("-pop");
if (params == null)
{
// return defaults:
switch(command.type)
{
case WRITE:
case COUNTER_WRITE:
return new SettingsPopulation(new SequentialOptions(defaultLimit));
default:
return new SettingsPopulation(new DistributionOptions(defaultLimit));
}
}
GroupedOptions options = GroupedOptions.select(params, new SequentialOptions(defaultLimit), new DistributionOptions(defaultLimit));
if (options == null)
{
printHelp();
System.out.println("Invalid -pop options provided, see output for valid options");
System.exit(1);
}
return options instanceof SequentialOptions ?
new SettingsPopulation((SequentialOptions) options) :
new SettingsPopulation((DistributionOptions) options);
}
public static void printHelp()
{
GroupedOptions.printOptions(System.out, "-pop", new SequentialOptions("N"), new DistributionOptions("N"));
}
public static Runnable helpPrinter()
{
return new Runnable()
{
@Override
public void run()
{
printHelp();
}
};
}
}

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

@ -44,12 +44,7 @@ public class SettingsSchema implements Serializable
public SettingsSchema(Options options, SettingsCommand command)
{
if (command instanceof SettingsCommandUser)
{
if (options.compaction.setByUser() || options.keyspace.setByUser() || options.compression.setByUser() || options.replication.setByUser())
throw new IllegalArgumentException("Cannot provide command line schema settings if a user profile is provided");
keyspace = ((SettingsCommandUser) command).profile.keyspaceName;
}
else
keyspace = options.keyspace.value();
@ -62,14 +57,7 @@ public class SettingsSchema implements Serializable
public void createKeySpaces(StressSettings settings)
{
if (!(settings.command instanceof SettingsCommandUser))
{
createKeySpacesThrift(settings);
}
else
{
((SettingsCommandUser) settings.command).profile.maybeCreateSchema(settings);
}
createKeySpacesThrift(settings);
}
@ -189,6 +177,9 @@ public class SettingsSchema implements Serializable
if (params == null)
return new SettingsSchema(new Options(), command);
if (command instanceof SettingsCommandUser)
throw new IllegalArgumentException("-schema can only be provided with predefined operations insert, read, etc.; the 'user' command requires a schema yaml instead");
GroupedOptions options = GroupedOptions.select(params, new Options());
if (options == null)
{

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

@ -40,8 +40,10 @@ public class StressSettings implements Serializable
{
public final SettingsCommand command;
public final SettingsRate rate;
public final SettingsKey keys;
public final SettingsPopulation generate;
public final SettingsInsert insert;
public final SettingsColumn columns;
public final SettingsErrors errors;
public final SettingsLog log;
public final SettingsMode mode;
public final SettingsNode node;
@ -50,12 +52,14 @@ public class StressSettings implements Serializable
public final SettingsPort port;
public final String sendToDaemon;
public StressSettings(SettingsCommand command, SettingsRate rate, SettingsKey keys, SettingsColumn columns, SettingsLog log, SettingsMode mode, SettingsNode node, SettingsSchema schema, SettingsTransport transport, SettingsPort port, String sendToDaemon)
public StressSettings(SettingsCommand command, SettingsRate rate, SettingsPopulation generate, SettingsInsert insert, SettingsColumn columns, SettingsErrors errors, SettingsLog log, SettingsMode mode, SettingsNode node, SettingsSchema schema, SettingsTransport transport, SettingsPort port, String sendToDaemon)
{
this.command = command;
this.rate = rate;
this.keys = keys;
this.insert = insert;
this.generate = generate;
this.columns = columns;
this.errors = errors;
this.log = log;
this.mode = mode;
this.node = node;
@ -129,7 +133,7 @@ public class StressSettings implements Serializable
}
catch (Exception e)
{
throw new RuntimeException(e.getMessage());
throw new RuntimeException(e);
}
return client;
@ -189,9 +193,10 @@ public class StressSettings implements Serializable
public void maybeCreateKeyspaces()
{
if (command.type == Command.WRITE || command.type == Command.COUNTER_WRITE || command.type == Command.USER)
if (command.type == Command.WRITE || command.type == Command.COUNTER_WRITE)
schema.createKeySpaces(this);
else if (command.type == Command.USER)
((SettingsCommandUser) command).profile.maybeCreateSchema(this);
}
public static StressSettings parse(String[] args)
@ -221,8 +226,10 @@ public class StressSettings implements Serializable
String sendToDaemon = SettingsMisc.getSendToDaemon(clArgs);
SettingsPort port = SettingsPort.get(clArgs);
SettingsRate rate = SettingsRate.get(clArgs, command);
SettingsKey keys = SettingsKey.get(clArgs, command);
SettingsPopulation generate = SettingsPopulation.get(clArgs, command);
SettingsInsert insert = SettingsInsert.get(clArgs);
SettingsColumn columns = SettingsColumn.get(clArgs);
SettingsErrors errors = SettingsErrors.get(clArgs);
SettingsLog log = SettingsLog.get(clArgs);
SettingsMode mode = SettingsMode.get(clArgs);
SettingsNode node = SettingsNode.get(clArgs);
@ -244,7 +251,7 @@ public class StressSettings implements Serializable
}
System.exit(1);
}
return new StressSettings(command, rate, keys, columns, log, mode, node, schema, transport, port, sendToDaemon);
return new StressSettings(command, rate, generate, insert, columns, errors, log, mode, node, schema, transport, port, sendToDaemon);
}
private static Map<String, String[]> parseMap(String[] args)

259
tools/stress/src/org/apache/cassandra/stress/util/DynamicList.java Normal file
View File

@ -0,0 +1,259 @@
package org.apache.cassandra.stress.util;
import java.util.Arrays;
import java.util.HashMap;
import java.util.TreeSet;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.cassandra.stress.generate.FasterRandom;
// simple thread-unsafe skiplist that permits indexing/removal by position, insertion at the end
// (though easily extended to insertion at any position, not necessary here)
// we use it for sampling items by position for visiting writes in the pool of pending writes
public class DynamicList<E>
{
// represents a value and an index simultaneously; each node maintains a list
// of next pointers for each height in the skip-list this node participates in
// (a contiguous range from [0..height))
public static class Node<E>
{
// stores the size of each descendant
private final int[] size;
// TODO: alternate links to save space
private final Node<E>[] links;
private final E value;
private Node(int height, E value)
{
this.value = value;
links = new Node[height * 2];
size = new int[height];
Arrays.fill(size, 1);
}
private int height()
{
return size.length;
}
private Node<E> next(int i)
{
return links[i * 2];
}
private Node<E> prev(int i)
{
return links[1 + i * 2];
}
private void setNext(int i, Node<E> next)
{
links[i * 2] = next;
}
private void setPrev(int i, Node<E> prev)
{
links[1 + i * 2] = prev;
}
private Node parent(int parentHeight)
{
Node prev = this;
while (true)
{
int height = prev.height();
if (parentHeight < height)
return prev;
prev = prev.prev(height - 1);
}
}
}
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final int maxHeight;
private final Node<E> head;
private int size;
public DynamicList(int maxExpectedSize)
{
this.maxHeight = 3 + (int) Math.ceil(Math.log(maxExpectedSize) / Math.log(2));
head = new Node<>(maxHeight, null);
}
private int randomLevel()
{
return 1 + Integer.bitCount(ThreadLocalRandom.current().nextInt() & ((1 << (maxHeight - 1)) - 1));
}
// add the value to the end of the list, and return the associated Node that permits efficient removal
// regardless of its future position in the list from other modifications
public Node<E> append(E value)
{
Node<E> newTail = new Node<>(randomLevel(), value);
lock.writeLock().lock();
try
{
size++;
Node<E> tail = head;
for (int i = maxHeight - 1 ; i >= newTail.height() ; i--)
{
Node<E> next;
while ((next = tail.next(i)) != null)
tail = next;
tail.size[i]++;
}
for (int i = newTail.height() - 1 ; i >= 0 ; i--)
{
Node<E> next;
while ((next = tail.next(i)) != null)
tail = next;
tail.setNext(i, newTail);
newTail.setPrev(i, tail);
}
return newTail;
}
finally
{
lock.writeLock().unlock();
}
}
// remove the provided node and its associated value from the list
public void remove(Node<E> node)
{
lock.writeLock().lock();
try
{
size--;
// go up through each level in the skip list, unlinking this node; this entails
// simply linking each neighbour to each other, and appending the size of the
// current level owned by this node's index to the preceding neighbour (since
// ownership is defined as any node that you must visit through the index,
// removal of ourselves from a level means the preceding index entry is the
// entry point to all of the removed node's descendants)
for (int i = 0 ; i < node.height() ; i++)
{
Node<E> prev = node.prev(i);
Node<E> next = node.next(i);
assert prev != null;
prev.setNext(i, next);
if (next != null)
next.setPrev(i, prev);
prev.size[i] += node.size[i] - 1;
}
// then go up the levels, removing 1 from the size at each height above ours
for (int i = node.height() ; i < maxHeight ; i++)
{
// if we're at our height limit, we backtrack at our top level until we
// hit a neighbour with a greater height
while (i == node.height())
node = node.prev(i - 1);
node.size[i]--;
}
}
finally
{
lock.writeLock().unlock();
}
}
// retrieve the item at the provided index, or return null if the index is past the end of the list
public E get(int index)
{
lock.readLock().lock();
try
{
if (index >= size)
return null;
index++;
int c = 0;
Node<E> finger = head;
for (int i = maxHeight - 1 ; i >= 0 ; i--)
{
while (c + finger.size[i] <= index)
{
c += finger.size[i];
finger = finger.next(i);
}
}
assert c == index;
return finger.value;
}
finally
{
lock.readLock().unlock();
}
}
// some quick and dirty tests to confirm the skiplist works as intended
// don't create a separate unit test - tools tree doesn't currently warrant them
private boolean isWellFormed()
{
for (int i = 0 ; i < maxHeight ; i++)
{
int c = 0;
for (Node node = head ; node != null ; node = node.next(i))
{
if (node.prev(i) != null && node.prev(i).next(i) != node)
return false;
if (node.next(i) != null && node.next(i).prev(i) != node)
return false;
c += node.size[i];
if (i + 1 < maxHeight && node.parent(i + 1).next(i + 1) == node.next(i))
{
if (node.parent(i + 1).size[i + 1] != c)
return false;
c = 0;
}
}
if (i == maxHeight - 1 && c != size + 1)
return false;
}
return true;
}
public static void main(String[] args)
{
DynamicList<Integer> list = new DynamicList<>(20);
TreeSet<Integer> canon = new TreeSet<>();
HashMap<Integer, Node> nodes = new HashMap<>();
int c = 0;
for (int i = 0 ; i < 100000 ; i++)
{
nodes.put(c, list.append(c));
canon.add(c);
c++;
}
FasterRandom rand = new FasterRandom();
assert list.isWellFormed();
for (int loop = 0 ; loop < 100 ; loop++)
{
System.out.println(loop);
for (int i = 0 ; i < 100000 ; i++)
{
int index = rand.nextInt(100000);
Integer seed = list.get(index);
// assert canon.headSet(seed, false).size() == index;
list.remove(nodes.remove(seed));
canon.remove(seed);
nodes.put(c, list.append(c));
canon.add(c);
c++;
}
assert list.isWellFormed();
}
}
}

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

@ -30,7 +30,7 @@ import java.util.concurrent.CountDownLatch;
public final class Timer
{
private static final int SAMPLE_SIZE_SHIFT = 10;
private static final int SAMPLE_SIZE_SHIFT = 14;
private static final int SAMPLE_SIZE_MASK = (1 << SAMPLE_SIZE_SHIFT) - 1;
private final Random rnd = new Random();
@ -66,6 +66,11 @@ public final class Timer
return 1 + (index >>> SAMPLE_SIZE_SHIFT);
}
public boolean running()
{
return finalReport == null;
}
public void stop(long partitionCount, long rowCount)
{
maybeReport();

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

@ -40,6 +40,7 @@ public class Timing
private final CopyOnWriteArrayList<Timer> timers = new CopyOnWriteArrayList<>();
private volatile TimingInterval history;
private final Random rnd = new Random();
private boolean done;
// TIMING
@ -57,11 +58,16 @@ public class Timing
if (!ready.await(2L, TimeUnit.MINUTES))
throw new RuntimeException("Timed out waiting for a timer thread - seems one got stuck");
boolean done = true;
// reports have been filled in by timer threadCount, so merge
List<TimingInterval> intervals = new ArrayList<>();
for (Timer timer : timers)
{
intervals.add(timer.report);
done &= !timer.running();
}
this.done = done;
return TimingInterval.merge(rnd, intervals, Integer.MAX_VALUE, history.endNanos());
}
@ -78,10 +84,15 @@ public class Timing
history = new TimingInterval(System.nanoTime());
}
public boolean done()
{
return done;
}
public TimingInterval snapInterval() throws InterruptedException
{
final TimingInterval interval = snapInterval(rnd);
history = TimingInterval.merge(rnd, Arrays.asList(interval, history), 50000, history.startNanos());
history = TimingInterval.merge(rnd, Arrays.asList(interval, history), 200000, history.startNanos());
return interval;
}

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

@ -97,14 +97,14 @@ public final class TimingInterval
}
public double realOpRate()
public double opRate()
{
return operationCount / ((end - start) * 0.000000001d);
}
public double adjustedOpRate()
public double adjustedRowRate()
{
return operationCount / ((end - (start + pauseLength)) * 0.000000001d);
return rowCount / ((end - (start + pauseLength)) * 0.000000001d);
}
public double partitionRate()