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