View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase;
20  
21  import static org.codehaus.jackson.map.SerializationConfig.Feature.SORT_PROPERTIES_ALPHABETICALLY;
22  
23  import java.io.IOException;
24  import java.io.PrintStream;
25  import java.lang.reflect.Constructor;
26  import java.math.BigDecimal;
27  import java.math.MathContext;
28  import java.text.DecimalFormat;
29  import java.text.SimpleDateFormat;
30  import java.util.ArrayList;
31  import java.util.Arrays;
32  import java.util.Date;
33  import java.util.LinkedList;
34  import java.util.Locale;
35  import java.util.Map;
36  import java.util.Queue;
37  import java.util.Random;
38  import java.util.TreeMap;
39  import java.util.NoSuchElementException;
40  import java.util.concurrent.Callable;
41  import java.util.concurrent.ExecutionException;
42  import java.util.concurrent.ExecutorService;
43  import java.util.concurrent.Executors;
44  import java.util.concurrent.Future;
45  
46  import com.google.common.base.Objects;
47  import com.google.common.util.concurrent.ThreadFactoryBuilder;
48  
49  import org.apache.commons.lang.StringUtils;
50  import org.apache.commons.logging.Log;
51  import org.apache.commons.logging.LogFactory;
52  import org.apache.hadoop.conf.Configuration;
53  import org.apache.hadoop.conf.Configured;
54  import org.apache.hadoop.fs.FileSystem;
55  import org.apache.hadoop.fs.Path;
56  import org.apache.hadoop.hbase.classification.InterfaceAudience;
57  import org.apache.hadoop.hbase.client.Admin;
58  import org.apache.hadoop.hbase.client.Append;
59  import org.apache.hadoop.hbase.client.BufferedMutator;
60  import org.apache.hadoop.hbase.client.BufferedMutatorParams;
61  import org.apache.hadoop.hbase.client.Connection;
62  import org.apache.hadoop.hbase.client.ConnectionFactory;
63  import org.apache.hadoop.hbase.client.Consistency;
64  import org.apache.hadoop.hbase.client.Delete;
65  import org.apache.hadoop.hbase.client.Durability;
66  import org.apache.hadoop.hbase.client.Get;
67  import org.apache.hadoop.hbase.client.Increment;
68  import org.apache.hadoop.hbase.client.Put;
69  import org.apache.hadoop.hbase.client.Result;
70  import org.apache.hadoop.hbase.client.ResultScanner;
71  import org.apache.hadoop.hbase.client.RowMutations;
72  import org.apache.hadoop.hbase.client.Scan;
73  import org.apache.hadoop.hbase.client.Table;
74  import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
75  import org.apache.hadoop.hbase.filter.BinaryComparator;
76  import org.apache.hadoop.hbase.filter.CompareFilter;
77  import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
78  import org.apache.hadoop.hbase.filter.Filter;
79  import org.apache.hadoop.hbase.filter.FilterAllFilter;
80  import org.apache.hadoop.hbase.filter.FilterList;
81  import org.apache.hadoop.hbase.filter.PageFilter;
82  import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
83  import org.apache.hadoop.hbase.filter.WhileMatchFilter;
84  import org.apache.hadoop.hbase.io.compress.Compression;
85  import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
86  import org.apache.hadoop.hbase.io.hfile.RandomDistribution;
87  import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
88  import org.apache.hadoop.hbase.regionserver.BloomType;
89  import org.apache.hadoop.hbase.trace.HBaseHTraceConfiguration;
90  import org.apache.hadoop.hbase.trace.SpanReceiverHost;
91  import org.apache.hadoop.hbase.util.*;
92  import org.apache.hadoop.io.LongWritable;
93  import org.apache.hadoop.io.Text;
94  import org.apache.hadoop.mapreduce.Job;
95  import org.apache.hadoop.mapreduce.Mapper;
96  import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat;
97  import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
98  import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
99  import org.apache.hadoop.util.Tool;
100 import org.apache.hadoop.util.ToolRunner;
101 import org.codehaus.jackson.map.ObjectMapper;
102 
103 import com.yammer.metrics.core.Histogram;
104 import com.yammer.metrics.stats.UniformSample;
105 
106 import org.apache.htrace.Sampler;
107 import org.apache.htrace.Trace;
108 import org.apache.htrace.TraceScope;
109 import org.apache.htrace.impl.ProbabilitySampler;
110 
111 /**
112  * Script used evaluating HBase performance and scalability.  Runs a HBase
113  * client that steps through one of a set of hardcoded tests or 'experiments'
114  * (e.g. a random reads test, a random writes test, etc.). Pass on the
115  * command-line which test to run and how many clients are participating in
116  * this experiment. Run {@code PerformanceEvaluation --help} to obtain usage.
117  *
118  * <p>This class sets up and runs the evaluation programs described in
119  * Section 7, <i>Performance Evaluation</i>, of the <a
120  * href="http://labs.google.com/papers/bigtable.html">Bigtable</a>
121  * paper, pages 8-10.
122  *
123  * <p>By default, runs as a mapreduce job where each mapper runs a single test
124  * client. Can also run as a non-mapreduce, multithreaded application by
125  * specifying {@code --nomapred}. Each client does about 1GB of data, unless
126  * specified otherwise.
127  */
128 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
129 public class PerformanceEvaluation extends Configured implements Tool {
130   private static final Log LOG = LogFactory.getLog(PerformanceEvaluation.class.getName());
131   private static final ObjectMapper MAPPER = new ObjectMapper();
132   static {
133     MAPPER.configure(SORT_PROPERTIES_ALPHABETICALLY, true);
134   }
135 
136   public static final String TABLE_NAME = "TestTable";
137   public static final String FAMILY_NAME_BASE = "info";
138   public static final byte[] FAMILY_ZERO = Bytes.toBytes("info0");
139   public static final byte[] COLUMN_ZERO = Bytes.toBytes("" + 0);
140   public static final int DEFAULT_VALUE_LENGTH = 1000;
141   public static final int ROW_LENGTH = 26;
142 
143   private static final int ONE_GB = 1024 * 1024 * 1000;
144   private static final int DEFAULT_ROWS_PER_GB = ONE_GB / DEFAULT_VALUE_LENGTH;
145   // TODO : should we make this configurable
146   private static final int TAG_LENGTH = 256;
147   private static final DecimalFormat FMT = new DecimalFormat("0.##");
148   private static final MathContext CXT = MathContext.DECIMAL64;
149   private static final BigDecimal MS_PER_SEC = BigDecimal.valueOf(1000);
150   private static final BigDecimal BYTES_PER_MB = BigDecimal.valueOf(1024 * 1024);
151   private static final TestOptions DEFAULT_OPTS = new TestOptions();
152 
153   private static Map<String, CmdDescriptor> COMMANDS = new TreeMap<String, CmdDescriptor>();
154   private static final Path PERF_EVAL_DIR = new Path("performance_evaluation");
155 
156   static {
157     addCommandDescriptor(RandomReadTest.class, "randomRead",
158       "Run random read test");
159     addCommandDescriptor(RandomSeekScanTest.class, "randomSeekScan",
160       "Run random seek and scan 100 test");
161     addCommandDescriptor(RandomScanWithRange10Test.class, "scanRange10",
162       "Run random seek scan with both start and stop row (max 10 rows)");
163     addCommandDescriptor(RandomScanWithRange100Test.class, "scanRange100",
164       "Run random seek scan with both start and stop row (max 100 rows)");
165     addCommandDescriptor(RandomScanWithRange1000Test.class, "scanRange1000",
166       "Run random seek scan with both start and stop row (max 1000 rows)");
167     addCommandDescriptor(RandomScanWithRange10000Test.class, "scanRange10000",
168       "Run random seek scan with both start and stop row (max 10000 rows)");
169     addCommandDescriptor(RandomWriteTest.class, "randomWrite",
170       "Run random write test");
171     addCommandDescriptor(SequentialReadTest.class, "sequentialRead",
172       "Run sequential read test");
173     addCommandDescriptor(SequentialWriteTest.class, "sequentialWrite",
174       "Run sequential write test");
175     addCommandDescriptor(ScanTest.class, "scan",
176       "Run scan test (read every row)");
177     addCommandDescriptor(FilteredScanTest.class, "filterScan",
178       "Run scan test using a filter to find a specific row based on it's value " +
179       "(make sure to use --rows=20)");
180     addCommandDescriptor(IncrementTest.class, "increment",
181       "Increment on each row; clients overlap on keyspace so some concurrent operations");
182     addCommandDescriptor(AppendTest.class, "append",
183       "Append on each row; clients overlap on keyspace so some concurrent operations");
184     addCommandDescriptor(CheckAndMutateTest.class, "checkAndMutate",
185       "CheckAndMutate on each row; clients overlap on keyspace so some concurrent operations");
186     addCommandDescriptor(CheckAndPutTest.class, "checkAndPut",
187       "CheckAndPut on each row; clients overlap on keyspace so some concurrent operations");
188     addCommandDescriptor(CheckAndDeleteTest.class, "checkAndDelete",
189       "CheckAndDelete on each row; clients overlap on keyspace so some concurrent operations");
190   }
191 
192   /**
193    * Enum for map metrics.  Keep it out here rather than inside in the Map
194    * inner-class so we can find associated properties.
195    */
196   protected static enum Counter {
197     /** elapsed time */
198     ELAPSED_TIME,
199     /** number of rows */
200     ROWS
201   }
202 
203   protected static class RunResult implements Comparable<RunResult> {
204     public RunResult(long duration, Histogram hist) {
205       this.duration = duration;
206       this.hist = hist;
207     }
208 
209     public final long duration;
210     public final Histogram hist;
211 
212     @Override
213     public String toString() {
214       return Long.toString(duration);
215     }
216 
217     @Override public int compareTo(RunResult o) {
218       return Long.compare(this.duration, o.duration);
219     }
220   }
221 
222   /**
223    * Constructor
224    * @param conf Configuration object
225    */
226   public PerformanceEvaluation(final Configuration conf) {
227     super(conf);
228   }
229 
230   protected static void addCommandDescriptor(Class<? extends Test> cmdClass,
231       String name, String description) {
232     CmdDescriptor cmdDescriptor = new CmdDescriptor(cmdClass, name, description);
233     COMMANDS.put(name, cmdDescriptor);
234   }
235 
236   /**
237    * Implementations can have their status set.
238    */
239   interface Status {
240     /**
241      * Sets status
242      * @param msg status message
243      * @throws IOException
244      */
245     void setStatus(final String msg) throws IOException;
246   }
247 
248   /**
249    * MapReduce job that runs a performance evaluation client in each map task.
250    */
251   public static class EvaluationMapTask
252       extends Mapper<LongWritable, Text, LongWritable, LongWritable> {
253 
254     /** configuration parameter name that contains the command */
255     public final static String CMD_KEY = "EvaluationMapTask.command";
256     /** configuration parameter name that contains the PE impl */
257     public static final String PE_KEY = "EvaluationMapTask.performanceEvalImpl";
258 
259     private Class<? extends Test> cmd;
260 
261     @Override
262     protected void setup(Context context) throws IOException, InterruptedException {
263       this.cmd = forName(context.getConfiguration().get(CMD_KEY), Test.class);
264 
265       // this is required so that extensions of PE are instantiated within the
266       // map reduce task...
267       Class<? extends PerformanceEvaluation> peClass =
268           forName(context.getConfiguration().get(PE_KEY), PerformanceEvaluation.class);
269       try {
270         peClass.getConstructor(Configuration.class).newInstance(context.getConfiguration());
271       } catch (Exception e) {
272         throw new IllegalStateException("Could not instantiate PE instance", e);
273       }
274     }
275 
276     private <Type> Class<? extends Type> forName(String className, Class<Type> type) {
277       try {
278         return Class.forName(className).asSubclass(type);
279       } catch (ClassNotFoundException e) {
280         throw new IllegalStateException("Could not find class for name: " + className, e);
281       }
282     }
283 
284     @Override
285     protected void map(LongWritable key, Text value, final Context context)
286            throws IOException, InterruptedException {
287 
288       Status status = new Status() {
289         @Override
290         public void setStatus(String msg) {
291            context.setStatus(msg);
292         }
293       };
294 
295       ObjectMapper mapper = new ObjectMapper();
296       TestOptions opts = mapper.readValue(value.toString(), TestOptions.class);
297       Configuration conf = HBaseConfiguration.create(context.getConfiguration());
298       final Connection con = ConnectionFactory.createConnection(conf);
299 
300       // Evaluation task
301       RunResult result = PerformanceEvaluation.runOneClient(this.cmd, conf, con, opts, status);
302       // Collect how much time the thing took. Report as map output and
303       // to the ELAPSED_TIME counter.
304       context.getCounter(Counter.ELAPSED_TIME).increment(result.duration);
305       context.getCounter(Counter.ROWS).increment(opts.perClientRunRows);
306       context.write(new LongWritable(opts.startRow), new LongWritable(result.duration));
307       context.progress();
308     }
309   }
310 
311   /*
312    * If table does not already exist, create. Also create a table when
313    * {@code opts.presplitRegions} is specified or when the existing table's
314    * region replica count doesn't match {@code opts.replicas}.
315    */
316   static boolean checkTable(Admin admin, TestOptions opts) throws IOException {
317     TableName tableName = TableName.valueOf(opts.tableName);
318     boolean needsDelete = false, exists = admin.tableExists(tableName);
319     boolean isReadCmd = opts.cmdName.toLowerCase(Locale.ROOT).contains("read")
320       || opts.cmdName.toLowerCase(Locale.ROOT).contains("scan");
321     if (!exists && isReadCmd) {
322       throw new IllegalStateException(
323         "Must specify an existing table for read commands. Run a write command first.");
324     }
325     HTableDescriptor desc =
326       exists ? admin.getTableDescriptor(TableName.valueOf(opts.tableName)) : null;
327     byte[][] splits = getSplits(opts);
328 
329     // recreate the table when user has requested presplit or when existing
330     // {RegionSplitPolicy,replica count} does not match requested.
331     if ((exists && opts.presplitRegions != DEFAULT_OPTS.presplitRegions)
332       || (!isReadCmd && desc != null &&
333           !StringUtils.equals(desc.getRegionSplitPolicyClassName(), opts.splitPolicy))
334       || (!isReadCmd && desc != null && desc.getRegionReplication() != opts.replicas)) {
335       needsDelete = true;
336       // wait, why did it delete my table?!?
337       LOG.debug(Objects.toStringHelper("needsDelete")
338         .add("needsDelete", needsDelete)
339         .add("isReadCmd", isReadCmd)
340         .add("exists", exists)
341         .add("desc", desc)
342         .add("presplit", opts.presplitRegions)
343         .add("splitPolicy", opts.splitPolicy)
344         .add("replicas", opts.replicas));
345     }
346 
347     // remove an existing table
348     if (needsDelete) {
349       if (admin.isTableEnabled(tableName)) {
350         admin.disableTable(tableName);
351       }
352       admin.deleteTable(tableName);
353     }
354 
355     // table creation is necessary
356     if (!exists || needsDelete) {
357       desc = getTableDescriptor(opts);
358       if (splits != null) {
359         if (LOG.isDebugEnabled()) {
360           for (int i = 0; i < splits.length; i++) {
361             LOG.debug(" split " + i + ": " + Bytes.toStringBinary(splits[i]));
362           }
363         }
364       }
365       admin.createTable(desc, splits);
366       LOG.info("Table " + desc + " created");
367     }
368     return admin.tableExists(tableName);
369   }
370 
371   /**
372    * Create an HTableDescriptor from provided TestOptions.
373    */
374   protected static HTableDescriptor getTableDescriptor(TestOptions opts) {
375     HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(opts.tableName));
376     for (int family = 0; family < opts.families; family++) {
377       byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
378       HColumnDescriptor familyDesc = new HColumnDescriptor(familyName);
379       familyDesc.setDataBlockEncoding(opts.blockEncoding);
380       familyDesc.setCompressionType(opts.compression);
381       familyDesc.setBloomFilterType(opts.bloomType);
382       familyDesc.setBlocksize(opts.blockSize);
383       if (opts.inMemoryCF) {
384         familyDesc.setInMemory(true);
385       }
386       tableDesc.addFamily(familyDesc);
387     }
388     if (opts.replicas != DEFAULT_OPTS.replicas) {
389       tableDesc.setRegionReplication(opts.replicas);
390     }
391     if (opts.splitPolicy != null && !opts.splitPolicy.equals(DEFAULT_OPTS.splitPolicy)) {
392       tableDesc.setRegionSplitPolicyClassName(opts.splitPolicy);
393     }
394     return tableDesc;
395   }
396 
397   /**
398    * generates splits based on total number of rows and specified split regions
399    */
400   protected static byte[][] getSplits(TestOptions opts) {
401     if (opts.presplitRegions == DEFAULT_OPTS.presplitRegions)
402       return null;
403 
404     int numSplitPoints = opts.presplitRegions - 1;
405     byte[][] splits = new byte[numSplitPoints][];
406     int jump = opts.totalRows / opts.presplitRegions;
407     for (int i = 0; i < numSplitPoints; i++) {
408       int rowkey = jump * (1 + i);
409       splits[i] = format(rowkey);
410     }
411     return splits;
412   }
413 
414   /*
415    * Run all clients in this vm each to its own thread.
416    */
417   static RunResult[] doLocalClients(final TestOptions opts, final Configuration conf)
418       throws IOException, InterruptedException {
419     final Class<? extends Test> cmd = determineCommandClass(opts.cmdName);
420     assert cmd != null;
421     @SuppressWarnings("unchecked")
422     Future<RunResult>[] threads = new Future[opts.numClientThreads];
423     RunResult[] results = new RunResult[opts.numClientThreads];
424     ExecutorService pool = Executors.newFixedThreadPool(opts.numClientThreads,
425       new ThreadFactoryBuilder().setNameFormat("TestClient-%s").build());
426     final Connection con = ConnectionFactory.createConnection(conf);
427     for (int i = 0; i < threads.length; i++) {
428       final int index = i;
429       threads[i] = pool.submit(new Callable<RunResult>() {
430         @Override
431         public RunResult call() throws Exception {
432           TestOptions threadOpts = new TestOptions(opts);
433           if (threadOpts.startRow == 0) threadOpts.startRow = index * threadOpts.perClientRunRows;
434           RunResult run = runOneClient(cmd, conf, con, threadOpts, new Status() {
435             @Override
436             public void setStatus(final String msg) throws IOException {
437               LOG.info(msg);
438             }
439           });
440           LOG.info("Finished " + Thread.currentThread().getName() + " in " + run.duration +
441             "ms over " + threadOpts.perClientRunRows + " rows");
442           return run;
443         }
444       });
445     }
446     pool.shutdown();
447 
448     for (int i = 0; i < threads.length; i++) {
449       try {
450         results[i] = threads[i].get();
451       } catch (ExecutionException e) {
452         throw new IOException(e.getCause());
453       }
454     }
455     final String test = cmd.getSimpleName();
456     LOG.info("[" + test + "] Summary of timings (ms): "
457              + Arrays.toString(results));
458     Arrays.sort(results);
459     long total = 0;
460     for (RunResult result : results) {
461       total += result.duration;
462     }
463     LOG.info("[" + test + "]"
464       + "\tMin: " + results[0] + "ms"
465       + "\tMax: " + results[results.length - 1] + "ms"
466       + "\tAvg: " + (total / results.length) + "ms");
467 
468     con.close();
469 
470     return results;
471   }
472 
473   /*
474    * Run a mapreduce job.  Run as many maps as asked-for clients.
475    * Before we start up the job, write out an input file with instruction
476    * per client regards which row they are to start on.
477    * @param cmd Command to run.
478    * @throws IOException
479    */
480   static Job doMapReduce(TestOptions opts, final Configuration conf)
481       throws IOException, InterruptedException, ClassNotFoundException {
482     final Class<? extends Test> cmd = determineCommandClass(opts.cmdName);
483     assert cmd != null;
484     Path inputDir = writeInputFile(conf, opts);
485     conf.set(EvaluationMapTask.CMD_KEY, cmd.getName());
486     conf.set(EvaluationMapTask.PE_KEY, PerformanceEvaluation.class.getName());
487     Job job = Job.getInstance(conf);
488     job.setJarByClass(PerformanceEvaluation.class);
489     job.setJobName("HBase Performance Evaluation - " + opts.cmdName);
490 
491     job.setInputFormatClass(NLineInputFormat.class);
492     NLineInputFormat.setInputPaths(job, inputDir);
493     // this is default, but be explicit about it just in case.
494     NLineInputFormat.setNumLinesPerSplit(job, 1);
495 
496     job.setOutputKeyClass(LongWritable.class);
497     job.setOutputValueClass(LongWritable.class);
498 
499     job.setMapperClass(EvaluationMapTask.class);
500     job.setReducerClass(LongSumReducer.class);
501 
502     job.setNumReduceTasks(1);
503 
504     job.setOutputFormatClass(TextOutputFormat.class);
505     TextOutputFormat.setOutputPath(job, new Path(inputDir.getParent(), "outputs"));
506 
507     TableMapReduceUtil.addDependencyJars(job);
508     TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
509       Histogram.class,     // yammer metrics
510       ObjectMapper.class); // jackson-mapper-asl
511 
512     TableMapReduceUtil.initCredentials(job);
513 
514     job.waitForCompletion(true);
515     return job;
516   }
517 
518   /*
519    * Write input file of offsets-per-client for the mapreduce job.
520    * @param c Configuration
521    * @return Directory that contains file written.
522    * @throws IOException
523    */
524   private static Path writeInputFile(final Configuration c, final TestOptions opts) throws IOException {
525     SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMddHHmmss");
526     Path jobdir = new Path(PERF_EVAL_DIR, formatter.format(new Date()));
527     Path inputDir = new Path(jobdir, "inputs");
528 
529     FileSystem fs = FileSystem.get(c);
530     fs.mkdirs(inputDir);
531 
532     Path inputFile = new Path(inputDir, "input.txt");
533     PrintStream out = new PrintStream(fs.create(inputFile));
534     // Make input random.
535     Map<Integer, String> m = new TreeMap<Integer, String>();
536     Hash h = MurmurHash.getInstance();
537     int perClientRows = (opts.totalRows / opts.numClientThreads);
538     try {
539       for (int i = 0; i < 10; i++) {
540         for (int j = 0; j < opts.numClientThreads; j++) {
541           TestOptions next = new TestOptions(opts);
542           next.startRow = (j * perClientRows) + (i * (perClientRows/10));
543           next.perClientRunRows = perClientRows / 10;
544           String s = MAPPER.writeValueAsString(next);
545           LOG.info("maptask input=" + s);
546           int hash = h.hash(Bytes.toBytes(s));
547           m.put(hash, s);
548         }
549       }
550       for (Map.Entry<Integer, String> e: m.entrySet()) {
551         out.println(e.getValue());
552       }
553     } finally {
554       out.close();
555     }
556     return inputDir;
557   }
558 
559   /**
560    * Describes a command.
561    */
562   static class CmdDescriptor {
563     private Class<? extends Test> cmdClass;
564     private String name;
565     private String description;
566 
567     CmdDescriptor(Class<? extends Test> cmdClass, String name, String description) {
568       this.cmdClass = cmdClass;
569       this.name = name;
570       this.description = description;
571     }
572 
573     public Class<? extends Test> getCmdClass() {
574       return cmdClass;
575     }
576 
577     public String getName() {
578       return name;
579     }
580 
581     public String getDescription() {
582       return description;
583     }
584   }
585 
586   /**
587    * Wraps up options passed to {@link org.apache.hadoop.hbase.PerformanceEvaluation}.
588    * This makes tracking all these arguments a little easier.
589    * NOTE: ADDING AN OPTION, you need to add a data member, a getter/setter (to make JSON
590    * serialization of this TestOptions class behave), and you need to add to the clone constructor
591    * below copying your new option from the 'that' to the 'this'.  Look for 'clone' below.
592    */
593   static class TestOptions {
594     String cmdName = null;
595     boolean nomapred = false;
596     boolean filterAll = false;
597     int startRow = 0;
598     float size = 1.0f;
599     int perClientRunRows = DEFAULT_ROWS_PER_GB;
600     int numClientThreads = 1;
601     int totalRows = DEFAULT_ROWS_PER_GB;
602     float sampleRate = 1.0f;
603     double traceRate = 0.0;
604     String tableName = TABLE_NAME;
605     boolean flushCommits = true;
606     boolean writeToWAL = true;
607     boolean autoFlush = false;
608     boolean oneCon = false;
609     boolean useTags = false;
610     int noOfTags = 1;
611     boolean reportLatency = false;
612     int multiGet = 0;
613     int randomSleep = 0;
614     boolean inMemoryCF = false;
615     int presplitRegions = 0;
616     int replicas = HTableDescriptor.DEFAULT_REGION_REPLICATION;
617     String splitPolicy = null;
618     Compression.Algorithm compression = Compression.Algorithm.NONE;
619     BloomType bloomType = BloomType.ROW;
620     int blockSize = HConstants.DEFAULT_BLOCKSIZE;
621     DataBlockEncoding blockEncoding = DataBlockEncoding.NONE;
622     boolean valueRandom = false;
623     boolean valueZipf = false;
624     int valueSize = DEFAULT_VALUE_LENGTH;
625     int period = (this.perClientRunRows / 10) == 0? perClientRunRows: perClientRunRows / 10;
626     int columns = 1;
627     int families = 1;
628     int caching = 30;
629     boolean addColumns = true;
630     long bufferSize = 2l * 1024l * 1024l;
631 
632     public TestOptions() {}
633 
634     /**
635      * Clone constructor.
636      * @param that Object to copy from.
637      */
638     public TestOptions(TestOptions that) {
639       this.cmdName = that.cmdName;
640       this.nomapred = that.nomapred;
641       this.startRow = that.startRow;
642       this.size = that.size;
643       this.perClientRunRows = that.perClientRunRows;
644       this.numClientThreads = that.numClientThreads;
645       this.totalRows = that.totalRows;
646       this.sampleRate = that.sampleRate;
647       this.traceRate = that.traceRate;
648       this.tableName = that.tableName;
649       this.flushCommits = that.flushCommits;
650       this.writeToWAL = that.writeToWAL;
651       this.autoFlush = that.autoFlush;
652       this.oneCon = that.oneCon;
653       this.useTags = that.useTags;
654       this.noOfTags = that.noOfTags;
655       this.reportLatency = that.reportLatency;
656       this.multiGet = that.multiGet;
657       this.inMemoryCF = that.inMemoryCF;
658       this.presplitRegions = that.presplitRegions;
659       this.replicas = that.replicas;
660       this.splitPolicy = that.splitPolicy;
661       this.compression = that.compression;
662       this.blockEncoding = that.blockEncoding;
663       this.filterAll = that.filterAll;
664       this.bloomType = that.bloomType;
665       this.blockSize = that.blockSize;
666       this.valueRandom = that.valueRandom;
667       this.valueZipf = that.valueZipf;
668       this.valueSize = that.valueSize;
669       this.period = that.period;
670       this.randomSleep = that.randomSleep;
671       this.addColumns = that.addColumns;
672       this.columns = that.columns;
673       this.families = that.families;
674       this.caching = that.caching;
675       this.bufferSize = that.bufferSize;
676     }
677 
678     public int getCaching() {
679       return this.caching;
680     }
681 
682     public void setCaching(final int caching) {
683       this.caching = caching;
684     }
685 
686     public int getColumns() {
687       return this.columns;
688     }
689 
690     public void setColumns(final int columns) {
691       this.columns = columns;
692     }
693 
694     public int getFamilies() {
695       return this.families;
696     }
697 
698     public void setFamilies(final int families) {
699       this.families = families;
700     }
701 
702     public boolean isValueZipf() {
703       return valueZipf;
704     }
705 
706     public void setValueZipf(boolean valueZipf) {
707       this.valueZipf = valueZipf;
708     }
709 
710     public String getCmdName() {
711       return cmdName;
712     }
713 
714     public void setCmdName(String cmdName) {
715       this.cmdName = cmdName;
716     }
717 
718     public int getRandomSleep() {
719       return randomSleep;
720     }
721 
722     public void setRandomSleep(int randomSleep) {
723       this.randomSleep = randomSleep;
724     }
725 
726     public int getReplicas() {
727       return replicas;
728     }
729 
730     public void setReplicas(int replicas) {
731       this.replicas = replicas;
732     }
733 
734     public String getSplitPolicy() {
735       return splitPolicy;
736     }
737 
738     public void setSplitPolicy(String splitPolicy) {
739       this.splitPolicy = splitPolicy;
740     }
741 
742     public void setNomapred(boolean nomapred) {
743       this.nomapred = nomapred;
744     }
745 
746     public void setFilterAll(boolean filterAll) {
747       this.filterAll = filterAll;
748     }
749 
750     public void setStartRow(int startRow) {
751       this.startRow = startRow;
752     }
753 
754     public void setSize(float size) {
755       this.size = size;
756     }
757 
758     public void setPerClientRunRows(int perClientRunRows) {
759       this.perClientRunRows = perClientRunRows;
760     }
761 
762     public void setNumClientThreads(int numClientThreads) {
763       this.numClientThreads = numClientThreads;
764     }
765 
766     public void setTotalRows(int totalRows) {
767       this.totalRows = totalRows;
768     }
769 
770     public void setSampleRate(float sampleRate) {
771       this.sampleRate = sampleRate;
772     }
773 
774     public void setTraceRate(double traceRate) {
775       this.traceRate = traceRate;
776     }
777 
778     public void setTableName(String tableName) {
779       this.tableName = tableName;
780     }
781 
782     public void setFlushCommits(boolean flushCommits) {
783       this.flushCommits = flushCommits;
784     }
785 
786     public void setWriteToWAL(boolean writeToWAL) {
787       this.writeToWAL = writeToWAL;
788     }
789 
790     public void setAutoFlush(boolean autoFlush) {
791       this.autoFlush = autoFlush;
792     }
793 
794     public void setOneCon(boolean oneCon) {
795       this.oneCon = oneCon;
796     }
797 
798     public void setUseTags(boolean useTags) {
799       this.useTags = useTags;
800     }
801 
802     public void setNoOfTags(int noOfTags) {
803       this.noOfTags = noOfTags;
804     }
805 
806     public void setReportLatency(boolean reportLatency) {
807       this.reportLatency = reportLatency;
808     }
809 
810     public void setMultiGet(int multiGet) {
811       this.multiGet = multiGet;
812     }
813 
814     public void setInMemoryCF(boolean inMemoryCF) {
815       this.inMemoryCF = inMemoryCF;
816     }
817 
818     public void setPresplitRegions(int presplitRegions) {
819       this.presplitRegions = presplitRegions;
820     }
821 
822     public void setCompression(Compression.Algorithm compression) {
823       this.compression = compression;
824     }
825 
826     public void setBloomType(BloomType bloomType) {
827       this.bloomType = bloomType;
828     }
829 
830     public void setBlockSize(int blockSize) {
831       this.blockSize = blockSize;
832     }
833 
834     public void setBlockEncoding(DataBlockEncoding blockEncoding) {
835       this.blockEncoding = blockEncoding;
836     }
837 
838     public void setValueRandom(boolean valueRandom) {
839       this.valueRandom = valueRandom;
840     }
841 
842     public void setValueSize(int valueSize) {
843       this.valueSize = valueSize;
844     }
845 
846     public void setBufferSize(long bufferSize) {
847       this.bufferSize = bufferSize;
848     }
849 
850     public long getBufferSize() {
851       return this.bufferSize;
852     }
853 
854     public void setPeriod(int period) {
855       this.period = period;
856     }
857 
858     public boolean isNomapred() {
859       return nomapred;
860     }
861 
862     public boolean isFilterAll() {
863       return filterAll;
864     }
865 
866     public int getStartRow() {
867       return startRow;
868     }
869 
870     public float getSize() {
871       return size;
872     }
873 
874     public int getPerClientRunRows() {
875       return perClientRunRows;
876     }
877 
878     public int getNumClientThreads() {
879       return numClientThreads;
880     }
881 
882     public int getTotalRows() {
883       return totalRows;
884     }
885 
886     public float getSampleRate() {
887       return sampleRate;
888     }
889 
890     public double getTraceRate() {
891       return traceRate;
892     }
893 
894     public String getTableName() {
895       return tableName;
896     }
897 
898     public boolean isFlushCommits() {
899       return flushCommits;
900     }
901 
902     public boolean isWriteToWAL() {
903       return writeToWAL;
904     }
905 
906     public boolean isAutoFlush() {
907       return autoFlush;
908     }
909 
910     public boolean isUseTags() {
911       return useTags;
912     }
913 
914     public int getNoOfTags() {
915       return noOfTags;
916     }
917 
918     public boolean isReportLatency() {
919       return reportLatency;
920     }
921 
922     public int getMultiGet() {
923       return multiGet;
924     }
925 
926     public boolean isInMemoryCF() {
927       return inMemoryCF;
928     }
929 
930     public int getPresplitRegions() {
931       return presplitRegions;
932     }
933 
934     public Compression.Algorithm getCompression() {
935       return compression;
936     }
937 
938     public DataBlockEncoding getBlockEncoding() {
939       return blockEncoding;
940     }
941 
942     public boolean isValueRandom() {
943       return valueRandom;
944     }
945 
946     public int getValueSize() {
947       return valueSize;
948     }
949 
950     public int getPeriod() {
951       return period;
952     }
953 
954     public BloomType getBloomType() {
955       return bloomType;
956     }
957 
958     public int getBlockSize() {
959       return blockSize;
960     }
961 
962     public boolean isOneCon() {
963       return oneCon;
964     }
965 
966     public boolean getAddColumns() {
967       return addColumns;
968     }
969 
970     public void setAddColumns(boolean addColumns) {
971       this.addColumns = addColumns;
972     }
973   }
974 
975   /*
976    * A test.
977    * Subclass to particularize what happens per row.
978    */
979   static abstract class Test {
980     // Below is make it so when Tests are all running in the one
981     // jvm, that they each have a differently seeded Random.
982     private static final Random randomSeed = new Random(System.currentTimeMillis());
983 
984     private static long nextRandomSeed() {
985       return randomSeed.nextLong();
986     }
987     private final int everyN;
988 
989     protected final Random rand = new Random(nextRandomSeed());
990     protected final Configuration conf;
991     protected final TestOptions opts;
992 
993     private final Status status;
994     private final Sampler<?> traceSampler;
995     private final SpanReceiverHost receiverHost;
996     protected Connection connection;
997 
998     private String testName;
999     private Histogram latencyHistogram;
1000     private Histogram valueSizeHistogram;
1001     private Histogram rpcCallsHistogram;
1002     private Histogram remoteRpcCallsHistogram;
1003     private Histogram millisBetweenNextHistogram;
1004     private Histogram regionsScannedHistogram;
1005     private Histogram bytesInResultsHistogram;
1006     private Histogram bytesInRemoteResultsHistogram;
1007     private RandomDistribution.Zipf zipf;
1008 
1009     /**
1010      * Note that all subclasses of this class must provide a public constructor
1011      * that has the exact same list of arguments.
1012      */
1013     Test(final Connection con, final TestOptions options, final Status status) {
1014       this.connection = con;
1015       this.conf = con == null ? HBaseConfiguration.create() : this.connection.getConfiguration();
1016       this.opts = options;
1017       this.status = status;
1018       this.testName = this.getClass().getSimpleName();
1019       receiverHost = SpanReceiverHost.getInstance(conf);
1020       if (options.traceRate >= 1.0) {
1021         this.traceSampler = Sampler.ALWAYS;
1022       } else if (options.traceRate > 0.0) {
1023         conf.setDouble("hbase.sampler.fraction", options.traceRate);
1024         this.traceSampler = new ProbabilitySampler(new HBaseHTraceConfiguration(conf));
1025       } else {
1026         this.traceSampler = Sampler.NEVER;
1027       }
1028       everyN = (int) (opts.totalRows / (opts.totalRows * opts.sampleRate));
1029       if (options.isValueZipf()) {
1030         this.zipf = new RandomDistribution.Zipf(this.rand, 1, options.getValueSize(), 1.1);
1031       }
1032       LOG.info("Sampling 1 every " + everyN + " out of " + opts.perClientRunRows + " total rows.");
1033     }
1034 
1035     int getValueLength(final Random r) {
1036       if (this.opts.isValueRandom()) {
1037         return r.nextInt(opts.valueSize);
1038       } else if (this.opts.isValueZipf()) {
1039         return Math.abs(this.zipf.nextInt());
1040       } else {
1041         return opts.valueSize;
1042       }
1043     }
1044 
1045     void updateValueSize(final Result [] rs) throws IOException {
1046       if (rs == null || !isRandomValueSize()) return;
1047       for (Result r: rs) updateValueSize(r);
1048     }
1049 
1050     void updateValueSize(final Result r) throws IOException {
1051       if (r == null || !isRandomValueSize()) return;
1052       int size = 0;
1053       for (CellScanner scanner = r.cellScanner(); scanner.advance();) {
1054         size += scanner.current().getValueLength();
1055       }
1056       updateValueSize(size);
1057     }
1058 
1059     void updateValueSize(final int valueSize) {
1060       if (!isRandomValueSize()) return;
1061       this.valueSizeHistogram.update(valueSize);
1062     }
1063 
1064     void updateScanMetrics(final ScanMetrics metrics) {
1065       if (metrics == null) return;
1066       Map<String,Long> metricsMap = metrics.getMetricsMap();
1067       Long rpcCalls = metricsMap.get(ScanMetrics.RPC_CALLS_METRIC_NAME);
1068       if (rpcCalls != null) {
1069         this.rpcCallsHistogram.update(rpcCalls.longValue());
1070       }
1071       Long remoteRpcCalls = metricsMap.get(ScanMetrics.REMOTE_RPC_CALLS_METRIC_NAME);
1072       if (remoteRpcCalls != null) {
1073         this.remoteRpcCallsHistogram.update(remoteRpcCalls.longValue());
1074       }
1075       Long millisBetweenNext = metricsMap.get(ScanMetrics.MILLIS_BETWEEN_NEXTS_METRIC_NAME);
1076       if (millisBetweenNext != null) {
1077         this.millisBetweenNextHistogram.update(millisBetweenNext.longValue());
1078       }
1079       Long regionsScanned = metricsMap.get(ScanMetrics.REGIONS_SCANNED_METRIC_NAME);
1080       if (regionsScanned != null) {
1081         this.regionsScannedHistogram.update(regionsScanned.longValue());
1082       }
1083       Long bytesInResults = metricsMap.get(ScanMetrics.BYTES_IN_RESULTS_METRIC_NAME);
1084       if (bytesInResults != null && bytesInResults.longValue() > 0) {
1085         this.bytesInResultsHistogram.update(bytesInResults.longValue());
1086       }
1087       Long bytesInRemoteResults = metricsMap.get(ScanMetrics.BYTES_IN_REMOTE_RESULTS_METRIC_NAME);
1088       if (bytesInRemoteResults != null && bytesInRemoteResults.longValue() > 0) {
1089         this.bytesInRemoteResultsHistogram.update(bytesInRemoteResults.longValue());
1090       }
1091     }
1092 
1093     String generateStatus(final int sr, final int i, final int lr) {
1094       return sr + "/" + i + "/" + lr + ", latency " + getShortLatencyReport() +
1095         (!isRandomValueSize()? "": ", value size " + getShortValueSizeReport());
1096     }
1097 
1098     boolean isRandomValueSize() {
1099       return opts.valueRandom;
1100     }
1101 
1102     protected int getReportingPeriod() {
1103       return opts.period;
1104     }
1105 
1106     /**
1107      * Populated by testTakedown. Only implemented by RandomReadTest at the moment.
1108      */
1109     public Histogram getLatencyHistogram() {
1110       return latencyHistogram;
1111     }
1112 
1113     void testSetup() throws IOException {
1114       // test metrics
1115       latencyHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1116       valueSizeHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1117       // scan metrics
1118       rpcCallsHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1119       remoteRpcCallsHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1120       millisBetweenNextHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1121       regionsScannedHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1122       bytesInResultsHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1123       bytesInRemoteResultsHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1124 
1125       if (!opts.oneCon) {
1126         this.connection = ConnectionFactory.createConnection(conf);
1127       }
1128 
1129       onStartup();
1130     }
1131 
1132     abstract void onStartup() throws IOException;
1133 
1134     void testTakedown() throws IOException {
1135       onTakedown();
1136       // Print all stats for this thread continuously.
1137       // Synchronize on Test.class so different threads don't intermingle the
1138       // output. We can't use 'this' here because each thread has its own instance of Test class.
1139       synchronized (Test.class) {
1140         status.setStatus("Test : " + testName + ", Thread : " + Thread.currentThread().getName());
1141         status.setStatus("Latency (us) : " + YammerHistogramUtils.getHistogramReport(
1142             latencyHistogram));
1143         status.setStatus("Num measures (latency) : " + latencyHistogram.count());
1144         status.setStatus(YammerHistogramUtils.getPrettyHistogramReport(latencyHistogram));
1145         status.setStatus("ValueSize (bytes) : "
1146             + YammerHistogramUtils.getHistogramReport(valueSizeHistogram));
1147         status.setStatus("Num measures (ValueSize): " + valueSizeHistogram.count());
1148         status.setStatus(YammerHistogramUtils.getPrettyHistogramReport(valueSizeHistogram));
1149         if (rpcCallsHistogram.count() > 0) {
1150           status.setStatus("rpcCalls (count): " +
1151             YammerHistogramUtils.getHistogramReport(rpcCallsHistogram));
1152         }
1153         if (remoteRpcCallsHistogram.count() > 0) {
1154           status.setStatus("remoteRpcCalls (count): " +
1155             YammerHistogramUtils.getHistogramReport(remoteRpcCallsHistogram));
1156         }
1157         if (millisBetweenNextHistogram.count() > 0) {
1158           status.setStatus("millisBetweenNext (latency): " +
1159             YammerHistogramUtils.getHistogramReport(millisBetweenNextHistogram));
1160         }
1161         if (regionsScannedHistogram.count() > 0) {
1162           status.setStatus("regionsScanned (count): " +
1163             YammerHistogramUtils.getHistogramReport(regionsScannedHistogram));
1164         }
1165         if (bytesInResultsHistogram.count() > 0) {
1166           status.setStatus("bytesInResults (size): " +
1167             YammerHistogramUtils.getHistogramReport(bytesInResultsHistogram));
1168         }
1169         if (bytesInRemoteResultsHistogram.count() > 0) {
1170           status.setStatus("bytesInRemoteResults (size): " +
1171             YammerHistogramUtils.getHistogramReport(bytesInRemoteResultsHistogram));
1172         }
1173       }
1174       if (!opts.oneCon) {
1175         connection.close();
1176       }
1177       receiverHost.closeReceivers();
1178     }
1179 
1180     abstract void onTakedown() throws IOException;
1181 
1182     /*
1183      * Run test
1184      * @return Elapsed time.
1185      * @throws IOException
1186      */
1187     long test() throws IOException, InterruptedException {
1188       testSetup();
1189       LOG.info("Timed test starting in thread " + Thread.currentThread().getName());
1190       final long startTime = System.nanoTime();
1191       try {
1192         testTimed();
1193       } finally {
1194         testTakedown();
1195       }
1196       return (System.nanoTime() - startTime) / 1000000;
1197     }
1198 
1199     int getStartRow() {
1200       return opts.startRow;
1201     }
1202 
1203     int getLastRow() {
1204       return getStartRow() + opts.perClientRunRows;
1205     }
1206 
1207     /**
1208      * Provides an extension point for tests that don't want a per row invocation.
1209      */
1210     void testTimed() throws IOException, InterruptedException {
1211       int startRow = getStartRow();
1212       int lastRow = getLastRow();
1213       // Report on completion of 1/10th of total.
1214       for (int i = startRow; i < lastRow; i++) {
1215         if (i % everyN != 0) continue;
1216         long startTime = System.nanoTime();
1217         TraceScope scope = Trace.startSpan("test row", traceSampler);
1218         try {
1219           testRow(i);
1220         } finally {
1221           scope.close();
1222         }
1223         // If multiget is enabled, say set to 10, testRow() returns immediately first 9 times
1224         // and sends the actual get request in the 10th iteration. We should only set latency
1225         // when actual request is sent because otherwise it turns out to be 0.
1226         if (opts.multiGet == 0 || (i - startRow + 1) % opts.multiGet == 0) {
1227           latencyHistogram.update((System.nanoTime() - startTime) / 1000);
1228         }
1229         if (status != null && i > 0 && (i % getReportingPeriod()) == 0) {
1230           status.setStatus(generateStatus(startRow, i, lastRow));
1231         }
1232       }
1233     }
1234 
1235     /**
1236      * @return Subset of the histograms' calculation.
1237      */
1238     public String getShortLatencyReport() {
1239       return YammerHistogramUtils.getShortHistogramReport(this.latencyHistogram);
1240     }
1241 
1242     /**
1243      * @return Subset of the histograms' calculation.
1244      */
1245     public String getShortValueSizeReport() {
1246       return YammerHistogramUtils.getShortHistogramReport(this.valueSizeHistogram);
1247     }
1248 
1249     /*
1250     * Test for individual row.
1251     * @param i Row index.
1252     */
1253     abstract void testRow(final int i) throws IOException, InterruptedException;
1254   }
1255 
1256   static abstract class TableTest extends Test {
1257     protected Table table;
1258 
1259     TableTest(Connection con, TestOptions options, Status status) {
1260       super(con, options, status);
1261     }
1262 
1263     @Override
1264     void onStartup() throws IOException {
1265       this.table = connection.getTable(TableName.valueOf(opts.tableName));
1266     }
1267 
1268     @Override
1269     void onTakedown() throws IOException {
1270       table.close();
1271     }
1272   }
1273 
1274   static abstract class BufferedMutatorTest extends Test {
1275     protected BufferedMutator mutator;
1276 
1277     BufferedMutatorTest(Connection con, TestOptions options, Status status) {
1278       super(con, options, status);
1279     }
1280 
1281     @Override
1282     void onStartup() throws IOException {
1283       BufferedMutatorParams p = new BufferedMutatorParams(TableName.valueOf(opts.tableName));
1284       p.writeBufferSize(opts.bufferSize);
1285       this.mutator = connection.getBufferedMutator(p);
1286     }
1287 
1288     @Override
1289     void onTakedown() throws IOException {
1290       mutator.close();
1291     }
1292   }
1293 
1294   static class RandomSeekScanTest extends TableTest {
1295     RandomSeekScanTest(Connection con, TestOptions options, Status status) {
1296       super(con, options, status);
1297     }
1298 
1299     @Override
1300     void testRow(final int i) throws IOException {
1301       Scan scan = new Scan(getRandomRow(this.rand, opts.totalRows));
1302       FilterList list = new FilterList();
1303       for (int family = 0; family < opts.families; family++) {
1304         byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1305         if (opts.addColumns) {
1306           for (int column = 0; column < opts.columns; column++) {
1307             byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1308             scan.addColumn(familyName, qualifier);
1309           }
1310         } else {
1311           scan.addFamily(familyName);
1312         }
1313       }
1314       if (opts.filterAll) {
1315         list.addFilter(new FilterAllFilter());
1316       }
1317       list.addFilter(new WhileMatchFilter(new PageFilter(120)));
1318       scan.setFilter(list);
1319       scan.setCaching(opts.caching);
1320       scan.setScanMetricsEnabled(true);
1321       ResultScanner s = this.table.getScanner(scan);
1322       try {
1323         for (Result rr; (rr = s.next()) != null;) {
1324           updateValueSize(rr);
1325         }
1326       } finally {
1327         if (s != null) {
1328           updateScanMetrics(scan.getScanMetrics());
1329           s.close();
1330         }
1331       }
1332     }
1333 
1334     @Override
1335     protected int getReportingPeriod() {
1336       int period = opts.perClientRunRows / 100;
1337       return period == 0 ? opts.perClientRunRows : period;
1338     }
1339 
1340   }
1341 
1342   static abstract class RandomScanWithRangeTest extends TableTest {
1343     RandomScanWithRangeTest(Connection con, TestOptions options, Status status) {
1344       super(con, options, status);
1345     }
1346 
1347     @Override
1348     void testRow(final int i) throws IOException {
1349       Pair<byte[], byte[]> startAndStopRow = getStartAndStopRow();
1350       Scan scan = new Scan(startAndStopRow.getFirst(), startAndStopRow.getSecond());
1351       for (int family = 0; family < opts.families; family++) {
1352         byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1353         if (opts.addColumns) {
1354           for (int column = 0; column < opts.columns; column++) {
1355             byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1356             scan.addColumn(familyName, qualifier);
1357           }
1358         } else {
1359           scan.addFamily(familyName);
1360         }
1361       }
1362       if (opts.filterAll) {
1363         scan.setFilter(new FilterAllFilter());
1364       }
1365       scan.setCaching(opts.caching);
1366       scan.setScanMetricsEnabled(true);
1367       Result r = null;
1368       int count = 0;
1369       ResultScanner s = this.table.getScanner(scan);
1370       try {
1371         for (; (r = s.next()) != null;) {
1372           updateValueSize(r);
1373           count++;
1374         }
1375         if (i % 100 == 0) {
1376           LOG.info(String.format("Scan for key range %s - %s returned %s rows",
1377             Bytes.toString(startAndStopRow.getFirst()),
1378             Bytes.toString(startAndStopRow.getSecond()), count));
1379         }
1380       } finally {
1381         if (s != null) {
1382           updateScanMetrics(scan.getScanMetrics());
1383           s.close();
1384         }
1385       }
1386     }
1387 
1388     protected abstract Pair<byte[],byte[]> getStartAndStopRow();
1389 
1390     protected Pair<byte[], byte[]> generateStartAndStopRows(int maxRange) {
1391       int start = this.rand.nextInt(Integer.MAX_VALUE) % opts.totalRows;
1392       int stop = start + maxRange;
1393       return new Pair<byte[],byte[]>(format(start), format(stop));
1394     }
1395 
1396     @Override
1397     protected int getReportingPeriod() {
1398       int period = opts.perClientRunRows / 100;
1399       return period == 0? opts.perClientRunRows: period;
1400     }
1401   }
1402 
1403   static class RandomScanWithRange10Test extends RandomScanWithRangeTest {
1404     RandomScanWithRange10Test(Connection con, TestOptions options, Status status) {
1405       super(con, options, status);
1406     }
1407 
1408     @Override
1409     protected Pair<byte[], byte[]> getStartAndStopRow() {
1410       return generateStartAndStopRows(10);
1411     }
1412   }
1413 
1414   static class RandomScanWithRange100Test extends RandomScanWithRangeTest {
1415     RandomScanWithRange100Test(Connection con, TestOptions options, Status status) {
1416       super(con, options, status);
1417     }
1418 
1419     @Override
1420     protected Pair<byte[], byte[]> getStartAndStopRow() {
1421       return generateStartAndStopRows(100);
1422     }
1423   }
1424 
1425   static class RandomScanWithRange1000Test extends RandomScanWithRangeTest {
1426     RandomScanWithRange1000Test(Connection con, TestOptions options, Status status) {
1427       super(con, options, status);
1428     }
1429 
1430     @Override
1431     protected Pair<byte[], byte[]> getStartAndStopRow() {
1432       return generateStartAndStopRows(1000);
1433     }
1434   }
1435 
1436   static class RandomScanWithRange10000Test extends RandomScanWithRangeTest {
1437     RandomScanWithRange10000Test(Connection con, TestOptions options, Status status) {
1438       super(con, options, status);
1439     }
1440 
1441     @Override
1442     protected Pair<byte[], byte[]> getStartAndStopRow() {
1443       return generateStartAndStopRows(10000);
1444     }
1445   }
1446 
1447   static class RandomReadTest extends TableTest {
1448     private final Consistency consistency;
1449     private ArrayList<Get> gets;
1450     private Random rd = new Random();
1451 
1452     RandomReadTest(Connection con, TestOptions options, Status status) {
1453       super(con, options, status);
1454       consistency = options.replicas == DEFAULT_OPTS.replicas ? null : Consistency.TIMELINE;
1455       if (opts.multiGet > 0) {
1456         LOG.info("MultiGet enabled. Sending GETs in batches of " + opts.multiGet + ".");
1457         this.gets = new ArrayList<Get>(opts.multiGet);
1458       }
1459     }
1460 
1461     @Override
1462     void testRow(final int i) throws IOException, InterruptedException {
1463       if (opts.randomSleep > 0) {
1464         Thread.sleep(rd.nextInt(opts.randomSleep));
1465       }
1466       Get get = new Get(getRandomRow(this.rand, opts.totalRows));
1467       for (int family = 0; family < opts.families; family++) {
1468         byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1469         if (opts.addColumns) {
1470           for (int column = 0; column < opts.columns; column++) {
1471             byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1472             get.addColumn(familyName, qualifier);
1473           }
1474         } else {
1475           get.addFamily(familyName);
1476         }
1477       }
1478       if (opts.filterAll) {
1479         get.setFilter(new FilterAllFilter());
1480       }
1481       get.setConsistency(consistency);
1482       if (LOG.isTraceEnabled()) LOG.trace(get.toString());
1483       if (opts.multiGet > 0) {
1484         this.gets.add(get);
1485         if (this.gets.size() == opts.multiGet) {
1486           Result [] rs = this.table.get(this.gets);
1487           updateValueSize(rs);
1488           this.gets.clear();
1489         }
1490       } else {
1491         updateValueSize(this.table.get(get));
1492       }
1493     }
1494 
1495     @Override
1496     protected int getReportingPeriod() {
1497       int period = opts.perClientRunRows / 10;
1498       return period == 0 ? opts.perClientRunRows : period;
1499     }
1500 
1501     @Override
1502     protected void testTakedown() throws IOException {
1503       if (this.gets != null && this.gets.size() > 0) {
1504         this.table.get(gets);
1505         this.gets.clear();
1506       }
1507       super.testTakedown();
1508     }
1509   }
1510 
1511   static class RandomWriteTest extends BufferedMutatorTest {
1512     RandomWriteTest(Connection con, TestOptions options, Status status) {
1513       super(con, options, status);
1514     }
1515 
1516     @Override
1517     void testRow(final int i) throws IOException {
1518       byte[] row = getRandomRow(this.rand, opts.totalRows);
1519       Put put = new Put(row);
1520       for (int family = 0; family < opts.families; family++) {
1521         byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1522         for (int column = 0; column < opts.columns; column++) {
1523           byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1524           byte[] value = generateData(this.rand, getValueLength(this.rand));
1525           if (opts.useTags) {
1526             byte[] tag = generateData(this.rand, TAG_LENGTH);
1527             Tag[] tags = new Tag[opts.noOfTags];
1528             for (int n = 0; n < opts.noOfTags; n++) {
1529               Tag t = new Tag((byte) n, tag);
1530               tags[n] = t;
1531             }
1532             KeyValue kv = new KeyValue(row, familyName, qualifier, HConstants.LATEST_TIMESTAMP,
1533               value, tags);
1534             put.add(kv);
1535             updateValueSize(kv.getValueLength());
1536           } else {
1537             put.add(familyName, qualifier, value);
1538             updateValueSize(value.length);
1539           }
1540         }
1541       }
1542       put.setDurability(opts.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1543       mutator.mutate(put);
1544       if (opts.autoFlush) {
1545         mutator.flush();
1546       }
1547     }
1548   }
1549 
1550   static class ScanTest extends TableTest {
1551     private Scan scan;
1552     private ResultScanner testScanner;
1553 
1554     ScanTest(Connection con, TestOptions options, Status status) {
1555       super(con, options, status);
1556     }
1557 
1558     @Override
1559     void testTakedown() throws IOException {
1560       if (this.testScanner != null) {
1561         this.testScanner.close();
1562       }
1563       super.testTakedown();
1564     }
1565 
1566 
1567     @Override
1568     void testRow(final int i) throws IOException {
1569       if (this.testScanner == null) {
1570         scan = new Scan(format(opts.startRow));
1571         for (int family = 0; family < opts.families; family++) {
1572           byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1573           if (opts.addColumns) {
1574             for (int column = 0; column < opts.columns; column++) {
1575               byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1576               scan.addColumn(familyName, qualifier);
1577             }
1578           } else {
1579             scan.addFamily(familyName);
1580           }
1581         }
1582         if (opts.filterAll) {
1583           scan.setFilter(new FilterAllFilter());
1584         }
1585         scan.setCaching(opts.caching);
1586         scan.setScanMetricsEnabled(true);
1587         this.testScanner = table.getScanner(scan);
1588       }
1589       try {
1590         Result r = testScanner.next();
1591         updateValueSize(r);
1592       } finally {
1593         updateScanMetrics(scan.getScanMetrics());
1594       }
1595     }
1596   }
1597 
1598   /**
1599    * Base class for operations that are CAS-like; that read a value and then set it based off what
1600    * they read. In this category is increment, append, checkAndPut, etc.
1601    *
1602    * <p>These operations also want some concurrency going on. Usually when these tests run, they
1603    * operate in their own part of the key range. In CASTest, we will have them all overlap on the
1604    * same key space. We do this with our getStartRow and getLastRow overrides.
1605    */
1606   static abstract class CASTableTest extends TableTest {
1607     private final byte [] qualifier;
1608     CASTableTest(Connection con, TestOptions options, Status status) {
1609       super(con, options, status);
1610       qualifier = Bytes.toBytes(this.getClass().getSimpleName());
1611     }
1612 
1613     byte [] getQualifier() {
1614       return this.qualifier;
1615     }
1616 
1617     @Override
1618     int getStartRow() {
1619       return 0;
1620     }
1621 
1622     @Override
1623     int getLastRow() {
1624       return opts.perClientRunRows;
1625     }
1626   }
1627 
1628   static class IncrementTest extends CASTableTest {
1629     IncrementTest(Connection con, TestOptions options, Status status) {
1630       super(con, options, status);
1631     }
1632 
1633     @Override
1634     void testRow(final int i) throws IOException {
1635       Increment increment = new Increment(format(i));
1636       // unlike checkAndXXX tests, which make most sense to do on a single value,
1637       // if multiple families are specified for an increment test we assume it is
1638       // meant to raise the work factor
1639       for (int family = 0; family < opts.families; family++) {
1640         byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1641         increment.addColumn(familyName, getQualifier(), 1l);
1642       }
1643       updateValueSize(this.table.increment(increment));
1644     }
1645   }
1646 
1647   static class AppendTest extends CASTableTest {
1648     AppendTest(Connection con, TestOptions options, Status status) {
1649       super(con, options, status);
1650     }
1651 
1652     @Override
1653     void testRow(final int i) throws IOException {
1654       byte [] bytes = format(i);
1655       Append append = new Append(bytes);
1656       // unlike checkAndXXX tests, which make most sense to do on a single value,
1657       // if multiple families are specified for an increment test we assume it is
1658       // meant to raise the work factor
1659       for (int family = 0; family < opts.families; family++) {
1660         byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1661         append.add(familyName, getQualifier(), bytes);
1662       }
1663       updateValueSize(this.table.append(append));
1664     }
1665   }
1666 
1667   static class CheckAndMutateTest extends CASTableTest {
1668     CheckAndMutateTest(Connection con, TestOptions options, Status status) {
1669       super(con, options, status);
1670     }
1671 
1672     @Override
1673     void testRow(final int i) throws IOException {
1674       final byte [] bytes = format(i);
1675       // checkAndXXX tests operate on only a single value
1676       // Put a known value so when we go to check it, it is there.
1677       Put put = new Put(bytes);
1678       put.addColumn(FAMILY_ZERO, getQualifier(), bytes);
1679       this.table.put(put);
1680       RowMutations mutations = new RowMutations(bytes);
1681       mutations.add(put);
1682       this.table.checkAndMutate(bytes, FAMILY_ZERO, getQualifier(), CompareOp.EQUAL, bytes,
1683           mutations);
1684     }
1685   }
1686 
1687   static class CheckAndPutTest extends CASTableTest {
1688     CheckAndPutTest(Connection con, TestOptions options, Status status) {
1689       super(con, options, status);
1690     }
1691 
1692     @Override
1693     void testRow(final int i) throws IOException {
1694       final byte [] bytes = format(i);
1695       // checkAndXXX tests operate on only a single value
1696       // Put a known value so when we go to check it, it is there.
1697       Put put = new Put(bytes);
1698       put.addColumn(FAMILY_ZERO, getQualifier(), bytes);
1699       this.table.put(put);
1700       this.table.checkAndPut(bytes, FAMILY_ZERO, getQualifier(), CompareOp.EQUAL, bytes, put);
1701     }
1702   }
1703 
1704   static class CheckAndDeleteTest extends CASTableTest {
1705     CheckAndDeleteTest(Connection con, TestOptions options, Status status) {
1706       super(con, options, status);
1707     }
1708 
1709     @Override
1710     void testRow(final int i) throws IOException {
1711       final byte [] bytes = format(i);
1712       // checkAndXXX tests operate on only a single value
1713       // Put a known value so when we go to check it, it is there.
1714       Put put = new Put(bytes);
1715       put.addColumn(FAMILY_ZERO, getQualifier(), bytes);
1716       this.table.put(put);
1717       Delete delete = new Delete(put.getRow());
1718       delete.addColumn(FAMILY_ZERO, getQualifier());
1719       this.table.checkAndDelete(bytes, FAMILY_ZERO, getQualifier(), CompareOp.EQUAL, bytes, delete);
1720     }
1721   }
1722 
1723   static class SequentialReadTest extends TableTest {
1724     SequentialReadTest(Connection con, TestOptions options, Status status) {
1725       super(con, options, status);
1726     }
1727 
1728     @Override
1729     void testRow(final int i) throws IOException {
1730       Get get = new Get(format(i));
1731       for (int family = 0; family < opts.families; family++) {
1732         byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1733         if (opts.addColumns) {
1734           for (int column = 0; column < opts.columns; column++) {
1735             byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1736             get.addColumn(familyName, qualifier);
1737           }
1738         } else {
1739           get.addFamily(familyName);
1740         }
1741       }
1742       if (opts.filterAll) {
1743         get.setFilter(new FilterAllFilter());
1744       }
1745       updateValueSize(table.get(get));
1746     }
1747   }
1748 
1749   static class SequentialWriteTest extends BufferedMutatorTest {
1750     SequentialWriteTest(Connection con, TestOptions options, Status status) {
1751       super(con, options, status);
1752     }
1753 
1754     @Override
1755     void testRow(final int i) throws IOException {
1756       byte[] row = format(i);
1757       Put put = new Put(row);
1758       for (int family = 0; family < opts.families; family++) {
1759         byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1760         for (int column = 0; column < opts.columns; column++) {
1761           byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1762           byte[] value = generateData(this.rand, getValueLength(this.rand));
1763           if (opts.useTags) {
1764             byte[] tag = generateData(this.rand, TAG_LENGTH);
1765             Tag[] tags = new Tag[opts.noOfTags];
1766             for (int n = 0; n < opts.noOfTags; n++) {
1767               Tag t = new Tag((byte) n, tag);
1768               tags[n] = t;
1769             }
1770             KeyValue kv = new KeyValue(row, familyName, qualifier, HConstants.LATEST_TIMESTAMP,
1771               value, tags);
1772             put.add(kv);
1773             updateValueSize(kv.getValueLength());
1774           } else {
1775             put.add(familyName, qualifier, value);
1776             updateValueSize(value.length);
1777           }
1778         }
1779       }
1780       put.setDurability(opts.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1781       mutator.mutate(put);
1782       if (opts.autoFlush) {
1783         mutator.flush();
1784       }
1785     }
1786   }
1787 
1788   static class FilteredScanTest extends TableTest {
1789     protected static final Log LOG = LogFactory.getLog(FilteredScanTest.class.getName());
1790 
1791     FilteredScanTest(Connection con, TestOptions options, Status status) {
1792       super(con, options, status);
1793       if (opts.perClientRunRows == DEFAULT_ROWS_PER_GB) {
1794         LOG.warn("Option \"rows\" unspecified. Using default value " + DEFAULT_ROWS_PER_GB
1795             + ". This could take a very long time.");
1796       }
1797     }
1798 
1799     @Override
1800     void testRow(int i) throws IOException {
1801       byte[] value = generateData(this.rand, getValueLength(this.rand));
1802       Scan scan = constructScan(value);
1803       ResultScanner scanner = null;
1804       try {
1805         scanner = this.table.getScanner(scan);
1806         for (Result r = null; (r = scanner.next()) != null;) {
1807           updateValueSize(r);
1808         }
1809       } finally {
1810         if (scanner != null) {
1811           updateScanMetrics(scan.getScanMetrics());
1812           scanner.close();
1813         }
1814       }
1815     }
1816 
1817     protected Scan constructScan(byte[] valuePrefix) throws IOException {
1818       FilterList list = new FilterList();
1819       Filter filter = new SingleColumnValueFilter(FAMILY_ZERO, COLUMN_ZERO,
1820         CompareFilter.CompareOp.EQUAL, new BinaryComparator(valuePrefix));
1821       list.addFilter(filter);
1822       if (opts.filterAll) {
1823         list.addFilter(new FilterAllFilter());
1824       }
1825       Scan scan = new Scan();
1826       if (opts.addColumns) {
1827         for (int column = 0; column < opts.columns; column++) {
1828           byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1829           scan.addColumn(FAMILY_ZERO, qualifier);
1830         }
1831       } else {
1832         scan.addFamily(FAMILY_ZERO);
1833       }
1834       scan.setFilter(list);
1835       scan.setCaching(opts.caching);
1836       scan.setScanMetricsEnabled(true);
1837       return scan;
1838     }
1839   }
1840 
1841   /**
1842    * Compute a throughput rate in MB/s.
1843    * @param rows Number of records consumed.
1844    * @param timeMs Time taken in milliseconds.
1845    * @return String value with label, ie '123.76 MB/s'
1846    */
1847   private static String calculateMbps(int rows, long timeMs, final int valueSize, int families,
1848       int columns) {
1849     BigDecimal rowSize = BigDecimal.valueOf(ROW_LENGTH +
1850       ((valueSize + (FAMILY_NAME_BASE.length()+1) + COLUMN_ZERO.length) * columns) * families);
1851     BigDecimal mbps = BigDecimal.valueOf(rows).multiply(rowSize, CXT)
1852       .divide(BigDecimal.valueOf(timeMs), CXT).multiply(MS_PER_SEC, CXT)
1853       .divide(BYTES_PER_MB, CXT);
1854     return FMT.format(mbps) + " MB/s";
1855   }
1856 
1857   /*
1858    * Format passed integer.
1859    * @param number
1860    * @return Returns zero-prefixed ROW_LENGTH-byte wide decimal version of passed
1861    * number (Does absolute in case number is negative).
1862    */
1863   public static byte [] format(final int number) {
1864     byte [] b = new byte[ROW_LENGTH];
1865     int d = Math.abs(number);
1866     for (int i = b.length - 1; i >= 0; i--) {
1867       b[i] = (byte)((d % 10) + '0');
1868       d /= 10;
1869     }
1870     return b;
1871   }
1872 
1873   /*
1874    * This method takes some time and is done inline uploading data.  For
1875    * example, doing the mapfile test, generation of the key and value
1876    * consumes about 30% of CPU time.
1877    * @return Generated random value to insert into a table cell.
1878    */
1879   public static byte[] generateData(final Random r, int length) {
1880     byte [] b = new byte [length];
1881     int i;
1882 
1883     for(i = 0; i < (length-8); i += 8) {
1884       b[i] = (byte) (65 + r.nextInt(26));
1885       b[i+1] = b[i];
1886       b[i+2] = b[i];
1887       b[i+3] = b[i];
1888       b[i+4] = b[i];
1889       b[i+5] = b[i];
1890       b[i+6] = b[i];
1891       b[i+7] = b[i];
1892     }
1893 
1894     byte a = (byte) (65 + r.nextInt(26));
1895     for(; i < length; i++) {
1896       b[i] = a;
1897     }
1898     return b;
1899   }
1900 
1901   /**
1902    * @deprecated Use {@link #generateData(java.util.Random, int)} instead.
1903    * @return Generated random value to insert into a table cell.
1904    */
1905   @Deprecated
1906   public static byte[] generateValue(final Random r) {
1907     return generateData(r, DEFAULT_VALUE_LENGTH);
1908   }
1909 
1910   static byte [] getRandomRow(final Random random, final int totalRows) {
1911     return format(random.nextInt(Integer.MAX_VALUE) % totalRows);
1912   }
1913 
1914   static RunResult runOneClient(final Class<? extends Test> cmd, Configuration conf, Connection con,
1915                            TestOptions opts, final Status status)
1916       throws IOException, InterruptedException {
1917     status.setStatus("Start " + cmd + " at offset " + opts.startRow + " for " +
1918       opts.perClientRunRows + " rows");
1919     long totalElapsedTime;
1920 
1921     final Test t;
1922     try {
1923       Constructor<? extends Test> constructor =
1924         cmd.getDeclaredConstructor(Connection.class, TestOptions.class, Status.class);
1925       t = constructor.newInstance(con, opts, status);
1926     } catch (NoSuchMethodException e) {
1927       throw new IllegalArgumentException("Invalid command class: " +
1928           cmd.getName() + ".  It does not provide a constructor as described by " +
1929           "the javadoc comment.  Available constructors are: " +
1930           Arrays.toString(cmd.getConstructors()));
1931     } catch (Exception e) {
1932       throw new IllegalStateException("Failed to construct command class", e);
1933     }
1934     totalElapsedTime = t.test();
1935 
1936     status.setStatus("Finished " + cmd + " in " + totalElapsedTime +
1937       "ms at offset " + opts.startRow + " for " + opts.perClientRunRows + " rows" +
1938       " (" + calculateMbps((int)(opts.perClientRunRows * opts.sampleRate), totalElapsedTime,
1939           getAverageValueLength(opts), opts.families, opts.columns) + ")");
1940 
1941     return new RunResult(totalElapsedTime, t.getLatencyHistogram());
1942   }
1943 
1944   private static int getAverageValueLength(final TestOptions opts) {
1945     return opts.valueRandom? opts.valueSize/2: opts.valueSize;
1946   }
1947 
1948   private void runTest(final Class<? extends Test> cmd, TestOptions opts) throws IOException,
1949       InterruptedException, ClassNotFoundException {
1950     // Log the configuration we're going to run with. Uses JSON mapper because lazy. It'll do
1951     // the TestOptions introspection for us and dump the output in a readable format.
1952     LOG.info(cmd.getSimpleName() + " test run options=" + MAPPER.writeValueAsString(opts));
1953     try(Connection conn = ConnectionFactory.createConnection(getConf());
1954         Admin admin = conn.getAdmin()) {
1955       checkTable(admin, opts);
1956     }
1957     if (opts.nomapred) {
1958       doLocalClients(opts, getConf());
1959     } else {
1960       doMapReduce(opts, getConf());
1961     }
1962   }
1963 
1964   protected void printUsage() {
1965     printUsage(this.getClass().getName(), null);
1966   }
1967 
1968   protected static void printUsage(final String message) {
1969     printUsage(PerformanceEvaluation.class.getName(), message);
1970   }
1971 
1972   protected static void printUsageAndExit(final String message, final int exitCode) {
1973     printUsage(message);
1974     System.exit(exitCode);
1975   }
1976 
1977   protected static void printUsage(final String className, final String message) {
1978     if (message != null && message.length() > 0) {
1979       System.err.println(message);
1980     }
1981     System.err.println("Usage: java " + className + " \\");
1982     System.err.println("  <OPTIONS> [-D<property=value>]* <command> <nclients>");
1983     System.err.println();
1984     System.err.println("Options:");
1985     System.err.println(" nomapred        Run multiple clients using threads " +
1986         "(rather than use mapreduce)");
1987     System.err.println(" rows            Rows each client runs. Default: " +
1988         DEFAULT_OPTS.getPerClientRunRows());
1989     System.err.println(" size            Total size in GiB. Mutually exclusive with --rows. " +
1990         "Default: 1.0.");
1991     System.err.println(" sampleRate      Execute test on a sample of total " +
1992         "rows. Only supported by randomRead. Default: 1.0");
1993     System.err.println(" traceRate       Enable HTrace spans. Initiate tracing every N rows. " +
1994         "Default: 0");
1995     System.err.println(" table           Alternate table name. Default: 'TestTable'");
1996     System.err.println(" multiGet        If >0, when doing RandomRead, perform multiple gets " +
1997         "instead of single gets. Default: 0");
1998     System.err.println(" compress        Compression type to use (GZ, LZO, ...). Default: 'NONE'");
1999     System.err.println(" flushCommits    Used to determine if the test should flush the table. " +
2000         "Default: false");
2001     System.err.println(" writeToWAL      Set writeToWAL on puts. Default: True");
2002     System.err.println(" autoFlush       Set autoFlush on htable. Default: False");
2003     System.err.println(" oneCon          all the threads share the same connection. Default: False");
2004     System.err.println(" presplit        Create presplit table. If a table with same name exists,"
2005         + " it'll be deleted and recreated (instead of verifying count of its existing regions). "
2006         + "Recommended for accurate perf analysis (see guide). Default: disabled");
2007     System.err.println(" inmemory        Tries to keep the HFiles of the CF " +
2008         "inmemory as far as possible. Not guaranteed that reads are always served " +
2009         "from memory.  Default: false");
2010     System.err.println(" usetags         Writes tags along with KVs. Use with HFile V3. " +
2011         "Default: false");
2012     System.err.println(" numoftags       Specify the no of tags that would be needed. " +
2013         "This works only if usetags is true. Default: " + DEFAULT_OPTS.noOfTags);
2014     System.err.println(" filterAll       Helps to filter out all the rows on the server side" +
2015         " there by not returning any thing back to the client.  Helps to check the server side" +
2016         " performance.  Uses FilterAllFilter internally. ");
2017     System.err.println(" latency         Set to report operation latencies. Default: False");
2018     System.err.println(" bloomFilter      Bloom filter type, one of " + Arrays.toString(BloomType.values()));
2019     System.err.println(" blockEncoding   Block encoding to use. Value should be one of "
2020         + Arrays.toString(DataBlockEncoding.values()) + ". Default: NONE");
2021     System.err.println(" valueSize       Pass value size to use: Default: " +
2022         DEFAULT_OPTS.getValueSize());
2023     System.err.println(" valueRandom     Set if we should vary value size between 0 and " +
2024         "'valueSize'; set on read for stats on size: Default: Not set.");
2025     System.err.println(" valueZipf       Set if we should vary value size between 0 and " +
2026         "'valueSize' in zipf form: Default: Not set.");
2027     System.err.println(" period          Report every 'period' rows: " +
2028         "Default: opts.perClientRunRows / 10 = " + DEFAULT_OPTS.getPerClientRunRows()/10);
2029     System.err.println(" multiGet        Batch gets together into groups of N. Only supported " +
2030         "by randomRead. Default: disabled");
2031     System.err.println(" blockSize       Blocksize to use when writing out hfiles. ");
2032     System.err.println(" addColumns      Adds columns to scans/gets explicitly. Default: true");
2033     System.err.println(" replicas        Enable region replica testing. Defaults: 1.");
2034     System.err.println(" splitPolicy     Specify a custom RegionSplitPolicy for the table.");
2035     System.err.println(" randomSleep     Do a random sleep before each get between 0 and entered value. Defaults: 0");
2036     System.err.println(" columns         Columns to write per row. Default: 1");
2037     System.err.println(" families        Specify number of column families for the table. Default: 1");
2038     System.err.println(" caching         Scan caching to use. Default: 30");
2039     System.err.println(" bufferSize      Set the value of client side buffering. Default: 2MB");
2040     System.err.println();
2041     System.err.println(" Note: -D properties will be applied to the conf used. ");
2042     System.err.println("  For example: ");
2043     System.err.println("   -Dmapreduce.output.fileoutputformat.compress=true");
2044     System.err.println("   -Dmapreduce.task.timeout=60000");
2045     System.err.println();
2046     System.err.println("Command:");
2047     for (CmdDescriptor command : COMMANDS.values()) {
2048       System.err.println(String.format(" %-15s %s", command.getName(), command.getDescription()));
2049     }
2050     System.err.println();
2051     System.err.println("Args:");
2052     System.err.println(" nclients        Integer. Required. Total number of clients "
2053         + "(and HRegionServers) running. 1 <= value <= 500");
2054     System.err.println("Examples:");
2055     System.err.println(" To run a single client doing the default 1M sequentialWrites:");
2056     System.err.println(" $ bin/hbase " + className + " sequentialWrite 1");
2057     System.err.println(" To run 10 clients doing increments over ten rows:");
2058     System.err.println(" $ bin/hbase " + className + " --rows=10 --nomapred increment 10");
2059   }
2060 
2061   /**
2062    * Parse options passed in via an arguments array. Assumes that array has been split
2063    * on white-space and placed into a {@code Queue}. Any unknown arguments will remain
2064    * in the queue at the conclusion of this method call. It's up to the caller to deal
2065    * with these unrecognized arguments.
2066    */
2067   static TestOptions parseOpts(Queue<String> args) {
2068     TestOptions opts = new TestOptions();
2069 
2070     String cmd = null;
2071     while ((cmd = args.poll()) != null) {
2072       if (cmd.equals("-h") || cmd.startsWith("--h")) {
2073         // place item back onto queue so that caller knows parsing was incomplete
2074         args.add(cmd);
2075         break;
2076       }
2077 
2078       final String nmr = "--nomapred";
2079       if (cmd.startsWith(nmr)) {
2080         opts.nomapred = true;
2081         continue;
2082       }
2083 
2084       final String rows = "--rows=";
2085       if (cmd.startsWith(rows)) {
2086         opts.perClientRunRows = Integer.parseInt(cmd.substring(rows.length()));
2087         continue;
2088       }
2089 
2090       final String sampleRate = "--sampleRate=";
2091       if (cmd.startsWith(sampleRate)) {
2092         opts.sampleRate = Float.parseFloat(cmd.substring(sampleRate.length()));
2093         continue;
2094       }
2095 
2096       final String table = "--table=";
2097       if (cmd.startsWith(table)) {
2098         opts.tableName = cmd.substring(table.length());
2099         continue;
2100       }
2101 
2102       final String startRow = "--startRow=";
2103       if (cmd.startsWith(startRow)) {
2104         opts.startRow = Integer.parseInt(cmd.substring(startRow.length()));
2105         continue;
2106       }
2107 
2108       final String compress = "--compress=";
2109       if (cmd.startsWith(compress)) {
2110         opts.compression = Compression.Algorithm.valueOf(cmd.substring(compress.length()));
2111         continue;
2112       }
2113 
2114       final String traceRate = "--traceRate=";
2115       if (cmd.startsWith(traceRate)) {
2116         opts.traceRate = Double.parseDouble(cmd.substring(traceRate.length()));
2117         continue;
2118       }
2119 
2120       final String blockEncoding = "--blockEncoding=";
2121       if (cmd.startsWith(blockEncoding)) {
2122         opts.blockEncoding = DataBlockEncoding.valueOf(cmd.substring(blockEncoding.length()));
2123         continue;
2124       }
2125 
2126       final String flushCommits = "--flushCommits=";
2127       if (cmd.startsWith(flushCommits)) {
2128         opts.flushCommits = Boolean.parseBoolean(cmd.substring(flushCommits.length()));
2129         continue;
2130       }
2131 
2132       final String writeToWAL = "--writeToWAL=";
2133       if (cmd.startsWith(writeToWAL)) {
2134         opts.writeToWAL = Boolean.parseBoolean(cmd.substring(writeToWAL.length()));
2135         continue;
2136       }
2137 
2138       final String presplit = "--presplit=";
2139       if (cmd.startsWith(presplit)) {
2140         opts.presplitRegions = Integer.parseInt(cmd.substring(presplit.length()));
2141         continue;
2142       }
2143 
2144       final String inMemory = "--inmemory=";
2145       if (cmd.startsWith(inMemory)) {
2146         opts.inMemoryCF = Boolean.parseBoolean(cmd.substring(inMemory.length()));
2147         continue;
2148       }
2149 
2150       final String autoFlush = "--autoFlush=";
2151       if (cmd.startsWith(autoFlush)) {
2152         opts.autoFlush = Boolean.parseBoolean(cmd.substring(autoFlush.length()));
2153         continue;
2154       }
2155 
2156       final String onceCon = "--oneCon=";
2157       if (cmd.startsWith(onceCon)) {
2158         opts.oneCon = Boolean.parseBoolean(cmd.substring(onceCon.length()));
2159         continue;
2160       }
2161 
2162       final String latency = "--latency";
2163       if (cmd.startsWith(latency)) {
2164         opts.reportLatency = true;
2165         continue;
2166       }
2167 
2168       final String multiGet = "--multiGet=";
2169       if (cmd.startsWith(multiGet)) {
2170         opts.multiGet = Integer.parseInt(cmd.substring(multiGet.length()));
2171         continue;
2172       }
2173 
2174       final String useTags = "--usetags=";
2175       if (cmd.startsWith(useTags)) {
2176         opts.useTags = Boolean.parseBoolean(cmd.substring(useTags.length()));
2177         continue;
2178       }
2179 
2180       final String noOfTags = "--numoftags=";
2181       if (cmd.startsWith(noOfTags)) {
2182         opts.noOfTags = Integer.parseInt(cmd.substring(noOfTags.length()));
2183         continue;
2184       }
2185 
2186       final String replicas = "--replicas=";
2187       if (cmd.startsWith(replicas)) {
2188         opts.replicas = Integer.parseInt(cmd.substring(replicas.length()));
2189         continue;
2190       }
2191 
2192       final String filterOutAll = "--filterAll";
2193       if (cmd.startsWith(filterOutAll)) {
2194         opts.filterAll = true;
2195         continue;
2196       }
2197 
2198       final String size = "--size=";
2199       if (cmd.startsWith(size)) {
2200         opts.size = Float.parseFloat(cmd.substring(size.length()));
2201         continue;
2202       }
2203 
2204       final String splitPolicy = "--splitPolicy=";
2205       if (cmd.startsWith(splitPolicy)) {
2206         opts.splitPolicy = cmd.substring(splitPolicy.length());
2207         continue;
2208       }
2209 
2210       final String randomSleep = "--randomSleep=";
2211       if (cmd.startsWith(randomSleep)) {
2212         opts.randomSleep = Integer.parseInt(cmd.substring(randomSleep.length()));
2213         continue;
2214       }
2215 
2216       final String bloomFilter = "--bloomFilter=";
2217       if (cmd.startsWith(bloomFilter)) {
2218         opts.bloomType = BloomType.valueOf(cmd.substring(bloomFilter.length()));
2219         continue;
2220       }
2221 
2222       final String blockSize = "--blockSize=";
2223       if(cmd.startsWith(blockSize) ) {
2224         opts.blockSize = Integer.parseInt(cmd.substring(blockSize.length()));
2225       }
2226 
2227       final String valueSize = "--valueSize=";
2228       if (cmd.startsWith(valueSize)) {
2229         opts.valueSize = Integer.parseInt(cmd.substring(valueSize.length()));
2230         continue;
2231       }
2232 
2233       final String valueRandom = "--valueRandom";
2234       if (cmd.startsWith(valueRandom)) {
2235         opts.valueRandom = true;
2236         if (opts.valueZipf) {
2237           throw new IllegalStateException("Either valueZipf or valueRandom but not both");
2238         }
2239         continue;
2240       }
2241 
2242       final String valueZipf = "--valueZipf";
2243       if (cmd.startsWith(valueZipf)) {
2244         opts.valueZipf = true;
2245         if (opts.valueRandom) {
2246           throw new IllegalStateException("Either valueZipf or valueRandom but not both");
2247         }
2248         continue;
2249       }
2250 
2251       final String period = "--period=";
2252       if (cmd.startsWith(period)) {
2253         opts.period = Integer.parseInt(cmd.substring(period.length()));
2254         continue;
2255       }
2256 
2257       final String addColumns = "--addColumns=";
2258       if (cmd.startsWith(addColumns)) {
2259         opts.addColumns = Boolean.parseBoolean(cmd.substring(addColumns.length()));
2260         continue;
2261       }
2262 
2263       final String columns = "--columns=";
2264       if (cmd.startsWith(columns)) {
2265         opts.columns = Integer.parseInt(cmd.substring(columns.length()));
2266         continue;
2267       }
2268 
2269       final String families = "--families=";
2270       if (cmd.startsWith(families)) {
2271         opts.families = Integer.parseInt(cmd.substring(families.length()));
2272         continue;
2273       }
2274 
2275       final String caching = "--caching=";
2276       if (cmd.startsWith(caching)) {
2277         opts.caching = Integer.parseInt(cmd.substring(caching.length()));
2278         continue;
2279       }
2280 
2281       final String bufferSize = "--bufferSize=";
2282       if (cmd.startsWith(bufferSize)) {
2283         opts.bufferSize = Long.parseLong(cmd.substring(bufferSize.length()));
2284         continue;
2285       }
2286 
2287       if (isCommandClass(cmd)) {
2288         opts.cmdName = cmd;
2289         try {
2290           opts.numClientThreads = Integer.parseInt(args.remove());
2291         } catch (NoSuchElementException | NumberFormatException e) {
2292           throw new IllegalArgumentException("Command " + cmd + " does not have threads number", e);
2293         }
2294         int rowsPerGB = getRowsPerGB(opts);
2295         if (opts.size != DEFAULT_OPTS.size &&
2296             opts.perClientRunRows != DEFAULT_OPTS.perClientRunRows) {
2297           throw new IllegalArgumentException(rows + " and " + size + " are mutually exclusive arguments.");
2298         }
2299         if (opts.size != DEFAULT_OPTS.size) {
2300           // total size in GB specified
2301           opts.totalRows = (int) opts.size * rowsPerGB;
2302           opts.perClientRunRows = opts.totalRows / opts.numClientThreads;
2303         } else {
2304           opts.totalRows = opts.perClientRunRows * opts.numClientThreads;
2305           opts.size = opts.totalRows / rowsPerGB;
2306         }
2307         break;
2308       } else {
2309         printUsageAndExit("ERROR: Unrecognized option/command: " + cmd, -1);
2310       }
2311 
2312       // Not matching any option or command.
2313       System.err.println("Error: Wrong option or command: " + cmd);
2314       args.add(cmd);
2315       break;
2316     }
2317     return opts;
2318   }
2319 
2320   static int getRowsPerGB(final TestOptions opts) {
2321     return ONE_GB / ((opts.valueRandom? opts.valueSize/2: opts.valueSize) * opts.getColumns() *
2322         opts.getFamilies());
2323   }
2324 
2325   @Override
2326   public int run(String[] args) throws Exception {
2327     // Process command-line args. TODO: Better cmd-line processing
2328     // (but hopefully something not as painful as cli options).
2329     int errCode = -1;
2330     if (args.length < 1) {
2331       printUsage();
2332       return errCode;
2333     }
2334 
2335     try {
2336       LinkedList<String> argv = new LinkedList<String>();
2337       argv.addAll(Arrays.asList(args));
2338       TestOptions opts = parseOpts(argv);
2339 
2340       // args remaining, print help and exit
2341       if (!argv.isEmpty()) {
2342         errCode = 0;
2343         printUsage();
2344         return errCode;
2345       }
2346 
2347       // must run at least 1 client
2348       if (opts.numClientThreads <= 0) {
2349         throw new IllegalArgumentException("Number of clients must be > 0");
2350       }
2351 
2352       // cmdName should not be null, print help and exit
2353       if (opts.cmdName == null) {
2354         printUsage();
2355         return errCode;
2356       }
2357 
2358       Class<? extends Test> cmdClass = determineCommandClass(opts.cmdName);
2359       if (cmdClass != null) {
2360         runTest(cmdClass, opts);
2361         errCode = 0;
2362       }
2363 
2364     } catch (Exception e) {
2365       e.printStackTrace();
2366     }
2367 
2368     return errCode;
2369   }
2370 
2371   private static boolean isCommandClass(String cmd) {
2372     return COMMANDS.containsKey(cmd);
2373   }
2374 
2375   private static Class<? extends Test> determineCommandClass(String cmd) {
2376     CmdDescriptor descriptor = COMMANDS.get(cmd);
2377     return descriptor != null ? descriptor.getCmdClass() : null;
2378   }
2379 
2380   public static void main(final String[] args) throws Exception {
2381     int res = ToolRunner.run(new PerformanceEvaluation(HBaseConfiguration.create()), args);
2382     System.exit(res);
2383   }
2384 }