cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bened...@apache.org
Subject [2/5] cassandra git commit: cassandra-stress simultaneous inserts over same seed
Date Fri, 12 Dec 2014 14:10:23 GMT
cassandra-stress simultaneous inserts over same seed

patch by benedict; reviewed by rstupp CASSANDRA-7964


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/6c579a01
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/6c579a01
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/6c579a01

Branch: refs/heads/cassandra-2.1
Commit: 6c579a0102fa3e67215fef5d9f8aa97191e3a216
Parents: cdba5aa
Author: Benedict Elliott Smith <benedict@apache.org>
Authored: Fri Dec 12 14:09:37 2014 +0000
Committer: Benedict Elliott Smith <benedict@apache.org>
Committed: Fri Dec 12 14:09:37 2014 +0000

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../org/apache/cassandra/stress/Operation.java  |  86 ++-
 .../apache/cassandra/stress/StressAction.java   |  96 +---
 .../apache/cassandra/stress/StressMetrics.java  |   1 -
 .../apache/cassandra/stress/StressProfile.java  |  65 +--
 .../apache/cassandra/stress/StressServer.java   |   3 +-
 .../cassandra/stress/generate/Partition.java    | 554 -------------------
 .../stress/generate/PartitionGenerator.java     |  39 +-
 .../apache/cassandra/stress/generate/Seed.java  |  52 +-
 .../cassandra/stress/generate/SeedManager.java  |  66 ++-
 .../cassandra/stress/generate/values/Bytes.java |   7 +-
 .../stress/generate/values/GeneratorConfig.java |   9 +-
 .../stress/generate/values/Strings.java         |   2 -
 .../stress/generate/values/TimeUUIDs.java       |   4 +-
 .../stress/operations/FixedOpDistribution.java  |   7 -
 .../stress/operations/OpDistribution.java       |   1 -
 .../operations/SampledOpDistribution.java       |   9 -
 .../operations/predefined/CqlCounterAdder.java  |   5 +-
 .../operations/predefined/CqlCounterGetter.java |   5 +-
 .../operations/predefined/CqlInserter.java      |   6 +-
 .../operations/predefined/CqlOperation.java     |  32 +-
 .../stress/operations/predefined/CqlReader.java |   7 +-
 .../predefined/PredefinedOperation.java         |  40 +-
 .../predefined/ThriftCounterAdder.java          |   9 +-
 .../predefined/ThriftCounterGetter.java         |   5 +-
 .../operations/predefined/ThriftInserter.java   |  11 +-
 .../operations/predefined/ThriftReader.java     |   7 +-
 .../operations/userdefined/SchemaInsert.java    |  52 +-
 .../operations/userdefined/SchemaQuery.java     |  55 +-
 .../operations/userdefined/SchemaStatement.java |  16 +-
 .../cassandra/stress/settings/Command.java      |   5 +-
 .../stress/settings/OptionAnyProbabilities.java |   8 +-
 .../stress/settings/OptionDistribution.java     |   4 +-
 .../settings/OptionEnumProbabilities.java       |   2 +-
 .../cassandra/stress/settings/OptionMulti.java  |   7 +-
 .../settings/OptionRatioDistribution.java       |  15 +-
 .../stress/settings/SettingsColumn.java         |  12 +-
 .../stress/settings/SettingsCommand.java        |   1 -
 .../settings/SettingsCommandPreDefined.java     |   7 +-
 .../SettingsCommandPreDefinedMixed.java         |   6 +-
 .../stress/settings/SettingsCommandUser.java    |  14 +-
 .../stress/settings/SettingsErrors.java         |   3 -
 .../cassandra/stress/settings/SettingsNode.java |   7 +-
 .../stress/settings/SettingsSchema.java         |   1 -
 .../stress/settings/StressSettings.java         |   3 -
 .../cassandra/stress/util/DynamicList.java      |  15 +-
 .../cassandra/stress/util/JavaDriverClient.java |   4 +-
 .../stress/util/SmartThriftClient.java          |   3 +-
 .../org/apache/cassandra/stress/util/Timer.java |   1 -
 .../apache/cassandra/stress/util/Timing.java    |   1 -
 .../cassandra/stress/util/TimingInterval.java   |   1 -
 51 files changed, 323 insertions(+), 1049 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index fa42e85..25140a8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.1.3
+ * cassandra-stress simultaneous inserts over same seed (CASSANDRA-7964)
  * Reduce cassandra-stress sampling memory requirements (CASSANDRA-7926)
  * Ensure memtable flush cannot expire commit log entries from its future (CASSANDRA-8383)
  * Make read "defrag" async to reclaim memtables (CASSANDRA-8459)

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

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

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java b/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java
index 3a4a4a3..d1cc0d4 100644
--- a/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java
+++ b/tools/stress/src/org/apache/cassandra/stress/StressMetrics.java
@@ -31,7 +31,6 @@ import java.util.concurrent.ThreadFactory;
 import org.apache.commons.lang3.time.DurationFormatUtils;
 
 import org.apache.cassandra.concurrent.NamedThreadFactory;
-import org.apache.cassandra.stress.settings.SettingsLog;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.JmxCollector;
 import org.apache.cassandra.stress.util.Timing;

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

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/StressServer.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/StressServer.java b/tools/stress/src/org/apache/cassandra/stress/StressServer.java
index 3c9e2a6..a6dfaf4 100644
--- a/tools/stress/src/org/apache/cassandra/stress/StressServer.java
+++ b/tools/stress/src/org/apache/cassandra/stress/StressServer.java
@@ -24,9 +24,10 @@ import java.net.InetAddress;
 import java.net.ServerSocket;
 import java.net.Socket;
 
-import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.commons.cli.*;
 
+import org.apache.cassandra.stress.settings.StressSettings;
+
 public class StressServer
 {
     private static final Options availableOptions = new Options();

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

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

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

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

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java
index 358163c..3c15c87 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Bytes.java
@@ -20,12 +20,11 @@
  */
 package org.apache.cassandra.stress.generate.values;
 
-import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.stress.generate.FasterRandom;
-
 import java.nio.ByteBuffer;
 import java.util.Arrays;
-import java.util.Random;
+
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.stress.generate.FasterRandom;
 
 public class Bytes extends Generator<ByteBuffer>
 {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java
index 8f7b2ea..3522338 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/GeneratorConfig.java
@@ -20,17 +20,14 @@
  */
 package org.apache.cassandra.stress.generate.values;
 
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+
 import org.apache.cassandra.stress.generate.Distribution;
 import org.apache.cassandra.stress.generate.DistributionFactory;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.MurmurHash;
 
-
-import java.io.Serializable;
-import java.nio.ByteBuffer;
-import java.util.Collections;
-import java.util.Map;
-
 public class GeneratorConfig implements Serializable
 {
     public final long salt;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java
index 71aaae6..b58fee2 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/Strings.java
@@ -20,8 +20,6 @@
  */
 package org.apache.cassandra.stress.generate.values;
 
-import java.util.Random;
-
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.stress.generate.FasterRandom;
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java b/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java
index efe4b79..7bfabf5 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generate/values/TimeUUIDs.java
@@ -21,11 +21,11 @@
 package org.apache.cassandra.stress.generate.values;
 
 
+import java.util.UUID;
+
 import org.apache.cassandra.db.marshal.TimeUUIDType;
 import org.apache.cassandra.utils.UUIDGen;
 
-import java.util.UUID;
-
 public class TimeUUIDs extends Generator<UUID>
 {
     final Dates dateGen;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java
index 3212795..533b630 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/FixedOpDistribution.java
@@ -25,7 +25,6 @@ import org.apache.cassandra.stress.Operation;
 
 public class FixedOpDistribution implements OpDistribution
 {
-
     final Operation operation;
 
     public FixedOpDistribution(Operation operation)
@@ -37,10 +36,4 @@ public class FixedOpDistribution implements OpDistribution
     {
         return operation;
     }
-
-    public int maxBatchSize()
-    {
-        return (int) operation.partitionCount.maxValue();
-    }
-
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java
index bcbd0bf..0fc15a6 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/OpDistribution.java
@@ -27,6 +27,5 @@ public interface OpDistribution
 {
 
     Operation next();
-    public int maxBatchSize();
 
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java
index 0bd64c5..432e991 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistribution.java
@@ -22,7 +22,6 @@ package org.apache.cassandra.stress.operations;
 
 
 import org.apache.commons.math3.distribution.EnumeratedDistribution;
-import org.apache.commons.math3.util.Pair;
 
 import org.apache.cassandra.stress.Operation;
 import org.apache.cassandra.stress.generate.Distribution;
@@ -41,14 +40,6 @@ public class SampledOpDistribution implements OpDistribution
         this.clustering = clustering;
     }
 
-    public int maxBatchSize()
-    {
-        int max = 1;
-        for (Pair<Operation, Double> pair : operations.getPmf())
-            max = Math.max(max, (int) pair.getFirst().partitionCount.maxValue());
-        return max;
-    }
-
     public Operation next()
     {
         while (remaining == 0)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java
index b7d1ee7..456c821 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java
@@ -28,6 +28,7 @@ import java.util.List;
 import org.apache.cassandra.stress.generate.Distribution;
 import org.apache.cassandra.stress.generate.DistributionFactory;
 import org.apache.cassandra.stress.generate.PartitionGenerator;
+import org.apache.cassandra.stress.generate.SeedManager;
 import org.apache.cassandra.stress.settings.Command;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.Timer;
@@ -36,9 +37,9 @@ public class CqlCounterAdder extends CqlOperation<Integer>
 {
 
     final Distribution counteradd;
-    public CqlCounterAdder(DistributionFactory counteradd, Timer timer, PartitionGenerator generator, StressSettings settings)
+    public CqlCounterAdder(DistributionFactory counteradd, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
     {
-        super(Command.COUNTER_WRITE, timer, generator, settings);
+        super(Command.COUNTER_WRITE, timer, generator, seedManager, settings);
         this.counteradd = counteradd.get();
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java
index 94c8faf..8c1c65c 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java
@@ -26,6 +26,7 @@ import java.util.Collections;
 import java.util.List;
 
 import org.apache.cassandra.stress.generate.PartitionGenerator;
+import org.apache.cassandra.stress.generate.SeedManager;
 import org.apache.cassandra.stress.settings.Command;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.Timer;
@@ -33,9 +34,9 @@ import org.apache.cassandra.stress.util.Timer;
 public class CqlCounterGetter extends CqlOperation<Integer>
 {
 
-    public CqlCounterGetter(Timer timer, PartitionGenerator generator, StressSettings settings)
+    public CqlCounterGetter(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
     {
-        super(Command.COUNTER_READ, timer, generator, settings);
+        super(Command.COUNTER_READ, timer, generator, seedManager, settings);
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java
index 622eb14..88ee752 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java
@@ -26,17 +26,17 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.cassandra.stress.generate.PartitionGenerator;
+import org.apache.cassandra.stress.generate.SeedManager;
 import org.apache.cassandra.stress.settings.Command;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.Timer;
-import org.apache.cassandra.utils.UUIDGen;
 
 public class CqlInserter extends CqlOperation<Integer>
 {
 
-    public CqlInserter(Timer timer, PartitionGenerator generator, StressSettings settings)
+    public CqlInserter(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
     {
-        super(Command.WRITE, timer, generator, settings);
+        super(Command.WRITE, timer, generator, seedManager, settings);
     }
 
     @Override

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

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java
index 3a7f75a..12cc241 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java
@@ -22,23 +22,22 @@ package org.apache.cassandra.stress.operations.predefined;
 
 
 import java.nio.ByteBuffer;
-import java.nio.charset.CharacterCodingException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
 import org.apache.cassandra.stress.generate.PartitionGenerator;
+import org.apache.cassandra.stress.generate.SeedManager;
 import org.apache.cassandra.stress.settings.Command;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.Timer;
-import org.apache.cassandra.utils.ByteBufferUtil;
 
 public class CqlReader extends CqlOperation<ByteBuffer[][]>
 {
 
-    public CqlReader(Timer timer, PartitionGenerator generator, StressSettings settings)
+    public CqlReader(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
     {
-        super(Command.READ, timer, generator, settings);
+        super(Command.READ, timer, generator, seedManager, settings);
     }
 
     @Override

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

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java
index 4ee42e9..be34a07 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterAdder.java
@@ -27,19 +27,22 @@ import java.util.Map;
 import org.apache.cassandra.stress.generate.Distribution;
 import org.apache.cassandra.stress.generate.DistributionFactory;
 import org.apache.cassandra.stress.generate.PartitionGenerator;
+import org.apache.cassandra.stress.generate.SeedManager;
 import org.apache.cassandra.stress.settings.Command;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.ThriftClient;
 import org.apache.cassandra.stress.util.Timer;
-import org.apache.cassandra.thrift.*;
+import org.apache.cassandra.thrift.ColumnOrSuperColumn;
+import org.apache.cassandra.thrift.CounterColumn;
+import org.apache.cassandra.thrift.Mutation;
 
 public class ThriftCounterAdder extends PredefinedOperation
 {
 
     final Distribution counteradd;
-    public ThriftCounterAdder(DistributionFactory counteradd, Timer timer, PartitionGenerator generator, StressSettings settings)
+    public ThriftCounterAdder(DistributionFactory counteradd, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
     {
-        super(Command.COUNTER_WRITE, timer, generator, settings);
+        super(Command.COUNTER_WRITE, timer, generator, seedManager, settings);
         this.counteradd = counteradd.get();
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java
index 10c6aab..ca81fe9 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftCounterGetter.java
@@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
 import java.util.List;
 
 import org.apache.cassandra.stress.generate.PartitionGenerator;
+import org.apache.cassandra.stress.generate.SeedManager;
 import org.apache.cassandra.stress.settings.Command;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.ThriftClient;
@@ -31,9 +32,9 @@ import org.apache.cassandra.thrift.SlicePredicate;
 
 public class ThriftCounterGetter extends PredefinedOperation
 {
-    public ThriftCounterGetter(Timer timer, PartitionGenerator generator, StressSettings settings)
+    public ThriftCounterGetter(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
     {
-        super(Command.COUNTER_READ, timer, generator, settings);
+        super(Command.COUNTER_READ, timer, generator, seedManager, settings);
     }
 
     public void run(final ThriftClient client) throws IOException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java
index d6adbf9..1827c06 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftInserter.java
@@ -24,22 +24,23 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.cassandra.db.marshal.TimeUUIDType;
 import org.apache.cassandra.stress.generate.PartitionGenerator;
+import org.apache.cassandra.stress.generate.SeedManager;
 import org.apache.cassandra.stress.settings.Command;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.ThriftClient;
 import org.apache.cassandra.stress.util.Timer;
-import org.apache.cassandra.thrift.*;
+import org.apache.cassandra.thrift.Column;
+import org.apache.cassandra.thrift.ColumnOrSuperColumn;
+import org.apache.cassandra.thrift.Mutation;
 import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.UUIDGen;
 
 public final class ThriftInserter extends PredefinedOperation
 {
 
-    public ThriftInserter(Timer timer, PartitionGenerator generator, StressSettings settings)
+    public ThriftInserter(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
     {
-        super(Command.WRITE, timer, generator, settings);
+        super(Command.WRITE, timer, generator, seedManager, settings);
     }
 
     public boolean isWrite()

http://git-wip-us.apache.org/repos/asf/cassandra/blob/6c579a01/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java
index 276d8c5..d77dc6a 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/ThriftReader.java
@@ -22,21 +22,20 @@ import java.nio.ByteBuffer;
 import java.util.List;
 
 import org.apache.cassandra.stress.generate.PartitionGenerator;
+import org.apache.cassandra.stress.generate.SeedManager;
 import org.apache.cassandra.stress.settings.Command;
 import org.apache.cassandra.stress.settings.StressSettings;
 import org.apache.cassandra.stress.util.ThriftClient;
 import org.apache.cassandra.stress.util.Timer;
 import org.apache.cassandra.thrift.ColumnOrSuperColumn;
 import org.apache.cassandra.thrift.ColumnParent;
-import org.apache.cassandra.thrift.SlicePredicate;
-import org.apache.cassandra.thrift.SuperColumn;
 
 public final class ThriftReader extends PredefinedOperation
 {
 
-    public ThriftReader(Timer timer, PartitionGenerator generator, StressSettings settings)
+    public ThriftReader(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
     {
-        super(Command.READ, timer, generator, settings);
+        super(Command.READ, timer, generator, seedManager, settings);
     }
 
     public void run(final ThriftClient client) throws IOException


Mime
View raw message