001/**
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements.  See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership.  The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License.  You may obtain a copy of the License at
010 *
011 *     http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing, software
014 * distributed under the License is distributed on an "AS IS" BASIS,
015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
016 * See the License for the specific language governing permissions and
017 * limitations under the License.
018 */
019package org.apache.hadoop.hbase.rest;
020
021import java.io.DataInput;
022import java.io.DataOutput;
023import java.io.IOException;
024import java.io.PrintStream;
025import java.lang.reflect.Constructor;
026import java.text.SimpleDateFormat;
027import java.util.ArrayList;
028import java.util.Arrays;
029import java.util.Date;
030import java.util.List;
031import java.util.Map;
032import java.util.Random;
033import java.util.TreeMap;
034import java.util.regex.Matcher;
035import java.util.regex.Pattern;
036import org.apache.hadoop.conf.Configuration;
037import org.apache.hadoop.conf.Configured;
038import org.apache.hadoop.fs.FSDataInputStream;
039import org.apache.hadoop.fs.FileStatus;
040import org.apache.hadoop.fs.FileSystem;
041import org.apache.hadoop.fs.Path;
042import org.apache.hadoop.hbase.ArrayBackedTag;
043import org.apache.hadoop.hbase.CompareOperator;
044import org.apache.hadoop.hbase.HBaseConfiguration;
045import org.apache.hadoop.hbase.HConstants;
046import org.apache.hadoop.hbase.KeyValue;
047import org.apache.hadoop.hbase.TableName;
048import org.apache.hadoop.hbase.Tag;
049import org.apache.hadoop.hbase.client.BufferedMutator;
050import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
051import org.apache.hadoop.hbase.client.Connection;
052import org.apache.hadoop.hbase.client.ConnectionFactory;
053import org.apache.hadoop.hbase.client.Durability;
054import org.apache.hadoop.hbase.client.Get;
055import org.apache.hadoop.hbase.client.Put;
056import org.apache.hadoop.hbase.client.Result;
057import org.apache.hadoop.hbase.client.ResultScanner;
058import org.apache.hadoop.hbase.client.Scan;
059import org.apache.hadoop.hbase.client.Table;
060import org.apache.hadoop.hbase.client.TableDescriptor;
061import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
062import org.apache.hadoop.hbase.filter.BinaryComparator;
063import org.apache.hadoop.hbase.filter.Filter;
064import org.apache.hadoop.hbase.filter.PageFilter;
065import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
066import org.apache.hadoop.hbase.filter.WhileMatchFilter;
067import org.apache.hadoop.hbase.io.compress.Compression;
068import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
069import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
070import org.apache.hadoop.hbase.rest.client.Client;
071import org.apache.hadoop.hbase.rest.client.Cluster;
072import org.apache.hadoop.hbase.rest.client.RemoteAdmin;
073import org.apache.hadoop.hbase.util.ByteArrayHashKey;
074import org.apache.hadoop.hbase.util.Bytes;
075import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
076import org.apache.hadoop.hbase.util.Hash;
077import org.apache.hadoop.hbase.util.MurmurHash;
078import org.apache.hadoop.hbase.util.Pair;
079import org.apache.hadoop.io.LongWritable;
080import org.apache.hadoop.io.NullWritable;
081import org.apache.hadoop.io.Text;
082import org.apache.hadoop.io.Writable;
083import org.apache.hadoop.mapreduce.InputSplit;
084import org.apache.hadoop.mapreduce.Job;
085import org.apache.hadoop.mapreduce.JobContext;
086import org.apache.hadoop.mapreduce.Mapper;
087import org.apache.hadoop.mapreduce.RecordReader;
088import org.apache.hadoop.mapreduce.TaskAttemptContext;
089import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
090import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
091import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
092import org.apache.hadoop.util.LineReader;
093import org.apache.hadoop.util.Tool;
094import org.apache.hadoop.util.ToolRunner;
095import org.slf4j.Logger;
096import org.slf4j.LoggerFactory;
097
098/**
099 * Script used evaluating Stargate performance and scalability.  Runs a SG
100 * client that steps through one of a set of hardcoded tests or 'experiments'
101 * (e.g. a random reads test, a random writes test, etc.). Pass on the
102 * command-line which test to run and how many clients are participating in
103 * this experiment. Run <code>java PerformanceEvaluation --help</code> to
104 * obtain usage.
105 *
106 * <p>This class sets up and runs the evaluation programs described in
107 * Section 7, <i>Performance Evaluation</i>, of the <a
108 * href="http://labs.google.com/papers/bigtable.html">Bigtable</a>
109 * paper, pages 8-10.
110 *
111 * <p>If number of clients > 1, we start up a MapReduce job. Each map task
112 * runs an individual client. Each client does about 1GB of data.
113 */
114public class PerformanceEvaluation extends Configured implements Tool {
115  protected static final Logger LOG =
116      LoggerFactory.getLogger(PerformanceEvaluation.class);
117
118  private static final int DEFAULT_ROW_PREFIX_LENGTH = 16;
119  private static final int ROW_LENGTH = 1000;
120  private static final int TAG_LENGTH = 256;
121  private static final int ONE_GB = 1024 * 1024 * 1000;
122  private static final int ROWS_PER_GB = ONE_GB / ROW_LENGTH;
123
124  public static final TableName TABLE_NAME = TableName.valueOf("TestTable");
125  public static final byte[] FAMILY_NAME = Bytes.toBytes("info");
126  public static final byte[] QUALIFIER_NAME = Bytes.toBytes("data");
127  private TableName tableName = TABLE_NAME;
128
129  protected TableDescriptor TABLE_DESCRIPTOR;
130  protected Map<String, CmdDescriptor> commands = new TreeMap<>();
131  protected static Cluster cluster = new Cluster();
132
133  volatile Configuration conf;
134  private boolean nomapred = false;
135  private int N = 1;
136  private int R = ROWS_PER_GB;
137  private Compression.Algorithm compression = Compression.Algorithm.NONE;
138  private DataBlockEncoding blockEncoding = DataBlockEncoding.NONE;
139  private boolean flushCommits = true;
140  private boolean writeToWAL = true;
141  private boolean inMemoryCF = false;
142  private int presplitRegions = 0;
143  private boolean useTags = false;
144  private int noOfTags = 1;
145  private Connection connection;
146
147  private static final Path PERF_EVAL_DIR = new Path("performance_evaluation");
148
149  /**
150   * Regex to parse lines in input file passed to mapreduce task.
151   */
152  public static final Pattern LINE_PATTERN =
153      Pattern.compile("tableName=(\\w+),\\s+" +
154          "startRow=(\\d+),\\s+" +
155          "perClientRunRows=(\\d+),\\s+" +
156          "totalRows=(\\d+),\\s+" +
157          "clients=(\\d+),\\s+" +
158          "flushCommits=(\\w+),\\s+" +
159          "writeToWAL=(\\w+),\\s+" +
160          "useTags=(\\w+),\\s+" +
161          "noOfTags=(\\d+)");
162
163  /**
164   * Enum for map metrics.  Keep it out here rather than inside in the Map
165   * inner-class so we can find associated properties.
166   */
167  protected enum Counter {
168    /** elapsed time */
169    ELAPSED_TIME,
170    /** number of rows */
171    ROWS
172  }
173
174  /**
175   * Constructor
176   * @param c Configuration object
177   */
178  public PerformanceEvaluation(final Configuration c) {
179    this.conf = c;
180
181    addCommandDescriptor(RandomReadTest.class, "randomRead",
182        "Run random read test");
183    addCommandDescriptor(RandomSeekScanTest.class, "randomSeekScan",
184        "Run random seek and scan 100 test");
185    addCommandDescriptor(RandomScanWithRange10Test.class, "scanRange10",
186        "Run random seek scan with both start and stop row (max 10 rows)");
187    addCommandDescriptor(RandomScanWithRange100Test.class, "scanRange100",
188        "Run random seek scan with both start and stop row (max 100 rows)");
189    addCommandDescriptor(RandomScanWithRange1000Test.class, "scanRange1000",
190        "Run random seek scan with both start and stop row (max 1000 rows)");
191    addCommandDescriptor(RandomScanWithRange10000Test.class, "scanRange10000",
192        "Run random seek scan with both start and stop row (max 10000 rows)");
193    addCommandDescriptor(RandomWriteTest.class, "randomWrite",
194        "Run random write test");
195    addCommandDescriptor(SequentialReadTest.class, "sequentialRead",
196        "Run sequential read test");
197    addCommandDescriptor(SequentialWriteTest.class, "sequentialWrite",
198        "Run sequential write test");
199    addCommandDescriptor(ScanTest.class, "scan",
200        "Run scan test (read every row)");
201    addCommandDescriptor(FilteredScanTest.class, "filterScan",
202        "Run scan test using a filter to find a specific row based " +
203        "on it's value (make sure to use --rows=20)");
204  }
205
206  protected void addCommandDescriptor(Class<? extends Test> cmdClass,
207      String name, String description) {
208    CmdDescriptor cmdDescriptor = new CmdDescriptor(cmdClass, name, description);
209    commands.put(name, cmdDescriptor);
210  }
211
212  /**
213   * Implementations can have their status set.
214   */
215  interface Status {
216    /**
217     * Sets status
218     * @param msg status message
219     * @throws IOException if setting the status fails
220     */
221    void setStatus(final String msg) throws IOException;
222  }
223
224  /**
225   *  This class works as the InputSplit of Performance Evaluation
226   *  MapReduce InputFormat, and the Record Value of RecordReader.
227   *  Each map task will only read one record from a PeInputSplit,
228   *  the record value is the PeInputSplit itself.
229   */
230  public static class PeInputSplit extends InputSplit implements Writable {
231    private TableName tableName;
232    private int startRow;
233    private int rows;
234    private int totalRows;
235    private int clients;
236    private boolean flushCommits;
237    private boolean writeToWAL;
238    private boolean useTags;
239    private int noOfTags;
240
241    public PeInputSplit(TableName tableName, int startRow, int rows, int totalRows, int clients,
242        boolean flushCommits, boolean writeToWAL, boolean useTags, int noOfTags) {
243      this.tableName = tableName;
244      this.startRow = startRow;
245      this.rows = rows;
246      this.totalRows = totalRows;
247      this.clients = clients;
248      this.flushCommits = flushCommits;
249      this.writeToWAL = writeToWAL;
250      this.useTags = useTags;
251      this.noOfTags = noOfTags;
252    }
253
254    @Override
255    public void readFields(DataInput in) throws IOException {
256      int tableNameLen = in.readInt();
257      byte[] name = new byte[tableNameLen];
258      in.readFully(name);
259      this.tableName = TableName.valueOf(name);
260      this.startRow = in.readInt();
261      this.rows = in.readInt();
262      this.totalRows = in.readInt();
263      this.clients = in.readInt();
264      this.flushCommits = in.readBoolean();
265      this.writeToWAL = in.readBoolean();
266      this.useTags = in.readBoolean();
267      this.noOfTags = in.readInt();
268    }
269
270    @Override
271    public void write(DataOutput out) throws IOException {
272      byte[] name = this.tableName.toBytes();
273      out.writeInt(name.length);
274      out.write(name);
275      out.writeInt(startRow);
276      out.writeInt(rows);
277      out.writeInt(totalRows);
278      out.writeInt(clients);
279      out.writeBoolean(flushCommits);
280      out.writeBoolean(writeToWAL);
281      out.writeBoolean(useTags);
282      out.writeInt(noOfTags);
283    }
284
285    @Override
286    public long getLength() {
287      return 0;
288    }
289
290    @Override
291    public String[] getLocations() {
292      return new String[0];
293    }
294
295    public int getStartRow() {
296      return startRow;
297    }
298
299    public TableName getTableName() {
300      return tableName;
301    }
302
303    public int getRows() {
304      return rows;
305    }
306
307    public int getTotalRows() {
308      return totalRows;
309    }
310
311    public boolean isFlushCommits() {
312      return flushCommits;
313    }
314
315    public boolean isWriteToWAL() {
316      return writeToWAL;
317    }
318
319    public boolean isUseTags() {
320      return useTags;
321    }
322
323    public int getNoOfTags() {
324      return noOfTags;
325    }
326  }
327
328  /**
329   *  InputFormat of Performance Evaluation MapReduce job.
330   *  It extends from FileInputFormat, want to use it's methods such as setInputPaths().
331   */
332  public static class PeInputFormat extends FileInputFormat<NullWritable, PeInputSplit> {
333    @Override
334    public List<InputSplit> getSplits(JobContext job) throws IOException {
335      // generate splits
336      List<InputSplit> splitList = new ArrayList<>();
337
338      for (FileStatus file : listStatus(job)) {
339        if (file.isDirectory()) {
340          continue;
341        }
342        Path path = file.getPath();
343        FileSystem fs = path.getFileSystem(job.getConfiguration());
344        FSDataInputStream fileIn = fs.open(path);
345        LineReader in = new LineReader(fileIn, job.getConfiguration());
346        int lineLen;
347        while (true) {
348          Text lineText = new Text();
349          lineLen = in.readLine(lineText);
350          if (lineLen <= 0) {
351            break;
352          }
353          Matcher m = LINE_PATTERN.matcher(lineText.toString());
354          if ((m != null) && m.matches()) {
355            TableName tableName = TableName.valueOf(m.group(1));
356            int startRow = Integer.parseInt(m.group(2));
357            int rows = Integer.parseInt(m.group(3));
358            int totalRows = Integer.parseInt(m.group(4));
359            int clients = Integer.parseInt(m.group(5));
360            boolean flushCommits = Boolean.parseBoolean(m.group(6));
361            boolean writeToWAL = Boolean.parseBoolean(m.group(7));
362            boolean useTags = Boolean.parseBoolean(m.group(8));
363            int noOfTags = Integer.parseInt(m.group(9));
364
365            LOG.debug("tableName=" + tableName +
366                      " split["+ splitList.size() + "] " +
367                      " startRow=" + startRow +
368                      " rows=" + rows +
369                      " totalRows=" + totalRows +
370                      " clients=" + clients +
371                      " flushCommits=" + flushCommits +
372                      " writeToWAL=" + writeToWAL +
373                      " useTags=" + useTags +
374                      " noOfTags=" + noOfTags);
375
376            PeInputSplit newSplit =
377              new PeInputSplit(tableName, startRow, rows, totalRows, clients,
378                  flushCommits, writeToWAL, useTags, noOfTags);
379            splitList.add(newSplit);
380          }
381        }
382        in.close();
383      }
384
385      LOG.info("Total # of splits: " + splitList.size());
386      return splitList;
387    }
388
389    @Override
390    public RecordReader<NullWritable, PeInputSplit> createRecordReader(InputSplit split,
391        TaskAttemptContext context) {
392      return new PeRecordReader();
393    }
394
395    public static class PeRecordReader extends RecordReader<NullWritable, PeInputSplit> {
396      private boolean readOver = false;
397      private PeInputSplit split = null;
398      private NullWritable key = null;
399      private PeInputSplit value = null;
400
401      @Override
402      public void initialize(InputSplit split, TaskAttemptContext context) {
403        this.readOver = false;
404        this.split = (PeInputSplit)split;
405      }
406
407      @Override
408      public boolean nextKeyValue() {
409        if (readOver) {
410          return false;
411        }
412
413        key = NullWritable.get();
414        value = split;
415
416        readOver = true;
417        return true;
418      }
419
420      @Override
421      public NullWritable getCurrentKey() {
422        return key;
423      }
424
425      @Override
426      public PeInputSplit getCurrentValue() {
427        return value;
428      }
429
430      @Override
431      public float getProgress() {
432        if (readOver) {
433          return 1.0f;
434        } else {
435          return 0.0f;
436        }
437      }
438
439      @Override
440      public void close() {
441        // do nothing
442      }
443    }
444  }
445
446  /**
447   * MapReduce job that runs a performance evaluation client in each map task.
448   */
449  public static class EvaluationMapTask
450      extends Mapper<NullWritable, PeInputSplit, LongWritable, LongWritable> {
451
452    /** configuration parameter name that contains the command */
453    public final static String CMD_KEY = "EvaluationMapTask.command";
454    /** configuration parameter name that contains the PE impl */
455    public static final String PE_KEY = "EvaluationMapTask.performanceEvalImpl";
456
457    private Class<? extends Test> cmd;
458    private PerformanceEvaluation pe;
459
460    @Override
461    protected void setup(Context context) {
462      this.cmd = forName(context.getConfiguration().get(CMD_KEY), Test.class);
463
464      // this is required so that extensions of PE are instantiated within the
465      // map reduce task...
466      Class<? extends PerformanceEvaluation> peClass =
467          forName(context.getConfiguration().get(PE_KEY), PerformanceEvaluation.class);
468      try {
469        this.pe = peClass.getConstructor(Configuration.class)
470            .newInstance(context.getConfiguration());
471      } catch (Exception e) {
472        throw new IllegalStateException("Could not instantiate PE instance", e);
473      }
474    }
475
476    private <Type> Class<? extends Type> forName(String className, Class<Type> type) {
477      Class<? extends Type> clazz;
478      try {
479        clazz = Class.forName(className).asSubclass(type);
480      } catch (ClassNotFoundException e) {
481        throw new IllegalStateException("Could not find class for name: " + className, e);
482      }
483      return clazz;
484    }
485
486    @Override
487    protected void map(NullWritable key, PeInputSplit value, final Context context)
488        throws IOException, InterruptedException {
489      Status status = context::setStatus;
490
491      // Evaluation task
492      pe.tableName = value.getTableName();
493      long elapsedTime = this.pe.runOneClient(this.cmd, value.getStartRow(),
494        value.getRows(), value.getTotalRows(),
495        value.isFlushCommits(), value.isWriteToWAL(),
496        value.isUseTags(), value.getNoOfTags(),
497        ConnectionFactory.createConnection(context.getConfiguration()), status);
498      // Collect how much time the thing took. Report as map output and
499      // to the ELAPSED_TIME counter.
500      context.getCounter(Counter.ELAPSED_TIME).increment(elapsedTime);
501      context.getCounter(Counter.ROWS).increment(value.rows);
502      context.write(new LongWritable(value.startRow), new LongWritable(elapsedTime));
503      context.progress();
504    }
505  }
506
507  /**
508   * If table does not already exist, create.
509   * @param admin Client to use checking.
510   * @return True if we created the table.
511   * @throws IOException if an operation on the table fails
512   */
513  private boolean checkTable(RemoteAdmin admin) throws IOException {
514    TableDescriptor tableDescriptor = getDescriptor();
515    if (this.presplitRegions > 0) {
516      // presplit requested
517      if (admin.isTableAvailable(tableDescriptor.getTableName().getName())) {
518        admin.deleteTable(tableDescriptor.getTableName().getName());
519      }
520
521      byte[][] splits = getSplits();
522      for (int i=0; i < splits.length; i++) {
523        LOG.debug(" split " + i + ": " + Bytes.toStringBinary(splits[i]));
524      }
525      admin.createTable(tableDescriptor);
526      LOG.info("Table created with " + this.presplitRegions + " splits");
527    } else {
528      boolean tableExists = admin.isTableAvailable(tableDescriptor.getTableName().getName());
529      if (!tableExists) {
530        admin.createTable(tableDescriptor);
531        LOG.info("Table " + tableDescriptor + " created");
532      }
533    }
534
535    return admin.isTableAvailable(tableDescriptor.getTableName().getName());
536  }
537
538  protected TableDescriptor getDescriptor() {
539    if (TABLE_DESCRIPTOR == null) {
540      TABLE_DESCRIPTOR =
541        TableDescriptorBuilder.newBuilder(tableName)
542          .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(FAMILY_NAME)
543            .setDataBlockEncoding(blockEncoding).setCompressionType(compression)
544            .setInMemory(inMemoryCF).build())
545          .build();
546    }
547    return TABLE_DESCRIPTOR;
548  }
549
550  /**
551   * Generates splits based on total number of rows and specified split regions
552   *
553   * @return splits : array of byte []
554   */
555  protected  byte[][] getSplits() {
556    if (this.presplitRegions == 0) {
557      return new byte[0][];
558    }
559
560    int numSplitPoints = presplitRegions - 1;
561    byte[][] splits = new byte[numSplitPoints][];
562    int jump = this.R  / this.presplitRegions;
563    for (int i = 0; i < numSplitPoints; i++) {
564      int rowkey = jump * (1 + i);
565      splits[i] = format(rowkey);
566    }
567    return splits;
568  }
569
570  /**
571   * We're to run multiple clients concurrently.  Setup a mapreduce job.  Run
572   * one map per client.  Then run a single reduce to sum the elapsed times.
573   * @param cmd Command to run.
574   */
575  private void runNIsMoreThanOne(final Class<? extends Test> cmd)
576      throws IOException, InterruptedException, ClassNotFoundException {
577    RemoteAdmin remoteAdmin = new RemoteAdmin(new Client(cluster), getConf());
578    checkTable(remoteAdmin);
579    if (nomapred) {
580      doMultipleClients(cmd);
581    } else {
582      doMapReduce(cmd);
583    }
584  }
585
586  /**
587   * Run all clients in this vm each to its own thread.
588   * @param cmd Command to run
589   * @throws IOException if creating a connection fails
590   */
591  private void doMultipleClients(final Class<? extends Test> cmd) throws IOException {
592    final List<Thread> threads = new ArrayList<>(this.N);
593    final long[] timings = new long[this.N];
594    final int perClientRows = R/N;
595    final TableName tableName = this.tableName;
596    final DataBlockEncoding encoding = this.blockEncoding;
597    final boolean flushCommits = this.flushCommits;
598    final Compression.Algorithm compression = this.compression;
599    final boolean writeToWal = this.writeToWAL;
600    final int preSplitRegions = this.presplitRegions;
601    final boolean useTags = this.useTags;
602    final int numTags = this.noOfTags;
603    final Connection connection = ConnectionFactory.createConnection(getConf());
604    for (int i = 0; i < this.N; i++) {
605      final int index = i;
606      Thread t = new Thread("TestClient-" + i) {
607        @Override
608        public void run() {
609          super.run();
610          PerformanceEvaluation pe = new PerformanceEvaluation(getConf());
611          pe.tableName = tableName;
612          pe.blockEncoding = encoding;
613          pe.flushCommits = flushCommits;
614          pe.compression = compression;
615          pe.writeToWAL = writeToWal;
616          pe.presplitRegions = preSplitRegions;
617          pe.N = N;
618          pe.connection = connection;
619          pe.useTags = useTags;
620          pe.noOfTags = numTags;
621          try {
622            long elapsedTime = pe.runOneClient(cmd, index * perClientRows,
623                perClientRows, R,
624                 flushCommits, writeToWAL, useTags, noOfTags, connection,
625              msg -> LOG.info("client-" + getName() + " " + msg));
626            timings[index] = elapsedTime;
627            LOG.info("Finished " + getName() + " in " + elapsedTime +
628              "ms writing " + perClientRows + " rows");
629          } catch (IOException e) {
630            throw new RuntimeException(e);
631          }
632        }
633      };
634      threads.add(t);
635    }
636    for (Thread t : threads) {
637      t.start();
638    }
639    for (Thread t : threads) {
640      while (t.isAlive()) {
641        try {
642          t.join();
643        } catch (InterruptedException e) {
644          LOG.debug("Interrupted, continuing" + e.toString());
645        }
646      }
647    }
648    final String test = cmd.getSimpleName();
649    LOG.info("[" + test + "] Summary of timings (ms): "
650             + Arrays.toString(timings));
651    Arrays.sort(timings);
652    long total = 0;
653    for (int i = 0; i < this.N; i++) {
654      total += timings[i];
655    }
656    LOG.info("[" + test + "]"
657             + "\tMin: " + timings[0] + "ms"
658             + "\tMax: " + timings[this.N - 1] + "ms"
659             + "\tAvg: " + (total / this.N) + "ms");
660  }
661
662  /**
663   * Run a mapreduce job.  Run as many maps as asked-for clients.
664   * Before we start up the job, write out an input file with instruction
665   * per client regards which row they are to start on.
666   * @param cmd Command to run.
667   */
668  private void doMapReduce(final Class<? extends Test> cmd)
669      throws IOException, InterruptedException, ClassNotFoundException {
670    Configuration conf = getConf();
671    Path inputDir = writeInputFile(conf);
672    conf.set(EvaluationMapTask.CMD_KEY, cmd.getName());
673    conf.set(EvaluationMapTask.PE_KEY, getClass().getName());
674    Job job = Job.getInstance(conf);
675    job.setJarByClass(PerformanceEvaluation.class);
676    job.setJobName("HBase Performance Evaluation");
677
678    job.setInputFormatClass(PeInputFormat.class);
679    PeInputFormat.setInputPaths(job, inputDir);
680
681    job.setOutputKeyClass(LongWritable.class);
682    job.setOutputValueClass(LongWritable.class);
683
684    job.setMapperClass(EvaluationMapTask.class);
685    job.setReducerClass(LongSumReducer.class);
686    job.setNumReduceTasks(1);
687
688    job.setOutputFormatClass(TextOutputFormat.class);
689    TextOutputFormat.setOutputPath(job, new Path(inputDir.getParent(), "outputs"));
690    TableMapReduceUtil.addDependencyJars(job);
691    TableMapReduceUtil.initCredentials(job);
692    job.waitForCompletion(true);
693  }
694
695  /**
696   * Write input file of offsets-per-client for the mapreduce job.
697   * @param c Configuration
698   * @return Directory that contains file written.
699   * @throws IOException if creating the directory or the file fails
700   */
701  private Path writeInputFile(final Configuration c) throws IOException {
702    SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMddHHmmss");
703    Path jobdir = new Path(PERF_EVAL_DIR, formatter.format(new Date()));
704    Path inputDir = new Path(jobdir, "inputs");
705
706    FileSystem fs = FileSystem.get(c);
707    fs.mkdirs(inputDir);
708    Path inputFile = new Path(inputDir, "input.txt");
709    // Make input random.
710    try (PrintStream out = new PrintStream(fs.create(inputFile))) {
711      Map<Integer, String> m = new TreeMap<>();
712      Hash h = MurmurHash.getInstance();
713      int perClientRows = (this.R / this.N);
714      for (int i = 0; i < 10; i++) {
715        for (int j = 0; j < N; j++) {
716          StringBuilder s = new StringBuilder();
717          s.append("tableName=").append(tableName);
718          s.append(", startRow=").append((j * perClientRows) + (i * (perClientRows / 10)));
719          s.append(", perClientRunRows=").append(perClientRows / 10);
720          s.append(", totalRows=").append(R);
721          s.append(", clients=").append(N);
722          s.append(", flushCommits=").append(flushCommits);
723          s.append(", writeToWAL=").append(writeToWAL);
724          s.append(", useTags=").append(useTags);
725          s.append(", noOfTags=").append(noOfTags);
726
727          byte[] b = Bytes.toBytes(s.toString());
728          int hash = h.hash(new ByteArrayHashKey(b, 0, b.length), -1);
729          m.put(hash, s.toString());
730        }
731      }
732      for (Map.Entry<Integer, String> e : m.entrySet()) {
733        out.println(e.getValue());
734      }
735    }
736    return inputDir;
737  }
738
739  /**
740   * Describes a command.
741   */
742  static class CmdDescriptor {
743    private Class<? extends Test> cmdClass;
744    private String name;
745    private String description;
746
747    CmdDescriptor(Class<? extends Test> cmdClass, String name, String description) {
748      this.cmdClass = cmdClass;
749      this.name = name;
750      this.description = description;
751    }
752
753    public Class<? extends Test> getCmdClass() {
754      return cmdClass;
755    }
756
757    public String getName() {
758      return name;
759    }
760
761    public String getDescription() {
762      return description;
763    }
764  }
765
766  /**
767   * Wraps up options passed to {@link org.apache.hadoop.hbase.PerformanceEvaluation} tests
768   * This makes the reflection logic a little easier to understand...
769   */
770  static class TestOptions {
771    private int startRow;
772    private int perClientRunRows;
773    private int totalRows;
774    private TableName tableName;
775    private boolean flushCommits;
776    private boolean writeToWAL;
777    private boolean useTags;
778    private int noOfTags;
779    private Connection connection;
780
781    TestOptions(int startRow, int perClientRunRows, int totalRows, TableName tableName,
782        boolean flushCommits, boolean writeToWAL, boolean useTags,
783        int noOfTags, Connection connection) {
784      this.startRow = startRow;
785      this.perClientRunRows = perClientRunRows;
786      this.totalRows = totalRows;
787      this.tableName = tableName;
788      this.flushCommits = flushCommits;
789      this.writeToWAL = writeToWAL;
790      this.useTags = useTags;
791      this.noOfTags = noOfTags;
792      this.connection = connection;
793    }
794
795    public int getStartRow() {
796      return startRow;
797    }
798
799    public int getPerClientRunRows() {
800      return perClientRunRows;
801    }
802
803    public int getTotalRows() {
804      return totalRows;
805    }
806
807    public TableName getTableName() {
808      return tableName;
809    }
810
811    public boolean isFlushCommits() {
812      return flushCommits;
813    }
814
815    public boolean isWriteToWAL() {
816      return writeToWAL;
817    }
818
819    public Connection getConnection() {
820      return connection;
821    }
822
823    public boolean isUseTags() {
824      return this.useTags;
825    }
826
827    public int getNumTags() {
828      return this.noOfTags;
829    }
830  }
831
832  /*
833   * A test.
834   * Subclass to particularize what happens per row.
835   */
836  static abstract class Test {
837    // Below is make it so when Tests are all running in the one
838    // jvm, that they each have a differently seeded Random.
839    private static final Random randomSeed =
840      new Random(EnvironmentEdgeManager.currentTime());
841    private static long nextRandomSeed() {
842      return randomSeed.nextLong();
843    }
844    protected final Random rand = new Random(nextRandomSeed());
845
846    protected final int startRow;
847    protected final int perClientRunRows;
848    protected final int totalRows;
849    private final Status status;
850    protected TableName tableName;
851    protected volatile Configuration conf;
852    protected boolean writeToWAL;
853    protected boolean useTags;
854    protected int noOfTags;
855    protected Connection connection;
856
857    /**
858     * Note that all subclasses of this class must provide a public contructor
859     * that has the exact same list of arguments.
860     */
861    Test(final Configuration conf, final TestOptions options, final Status status) {
862      super();
863      this.startRow = options.getStartRow();
864      this.perClientRunRows = options.getPerClientRunRows();
865      this.totalRows = options.getTotalRows();
866      this.status = status;
867      this.tableName = options.getTableName();
868      this.conf = conf;
869      this.writeToWAL = options.isWriteToWAL();
870      this.useTags = options.isUseTags();
871      this.noOfTags = options.getNumTags();
872      this.connection = options.getConnection();
873    }
874
875    protected String generateStatus(final int sr, final int i, final int lr) {
876      return sr + "/" + i + "/" + lr;
877    }
878
879    protected int getReportingPeriod() {
880      int period = this.perClientRunRows / 10;
881      return period == 0? this.perClientRunRows: period;
882    }
883
884    abstract void testTakedown()  throws IOException;
885
886    /**
887     * Run test
888     * @return Elapsed time.
889     * @throws IOException if something in the test fails
890     */
891    long test() throws IOException {
892      testSetup();
893      LOG.info("Timed test starting in thread " + Thread.currentThread().getName());
894      final long startTime = System.nanoTime();
895      try {
896        testTimed();
897      } finally {
898        testTakedown();
899      }
900      return (System.nanoTime() - startTime) / 1000000;
901    }
902
903    abstract void testSetup() throws IOException;
904
905    /**
906     * Provides an extension point for tests that don't want a per row invocation.
907     */
908    void testTimed() throws IOException {
909      int lastRow = this.startRow + this.perClientRunRows;
910      // Report on completion of 1/10th of total.
911      for (int i = this.startRow; i < lastRow; i++) {
912        testRow(i);
913        if (status != null && i > 0 && (i % getReportingPeriod()) == 0) {
914          status.setStatus(generateStatus(this.startRow, i, lastRow));
915        }
916      }
917    }
918
919    /**
920     * Test for individual row.
921     * @param i Row index.
922     */
923    abstract void testRow(final int i) throws IOException;
924  }
925
926  static abstract class TableTest extends Test {
927    protected Table table;
928
929    public TableTest(Configuration conf, TestOptions options, Status status) {
930      super(conf, options, status);
931    }
932
933    @Override
934    void testSetup() throws IOException {
935      this.table = connection.getTable(tableName);
936    }
937
938    @Override
939    void testTakedown() throws IOException {
940      table.close();
941    }
942  }
943
944  static abstract class BufferedMutatorTest extends Test {
945    protected BufferedMutator mutator;
946    protected boolean flushCommits;
947
948    public BufferedMutatorTest(Configuration conf, TestOptions options, Status status) {
949      super(conf, options, status);
950      this.flushCommits = options.isFlushCommits();
951    }
952
953    @Override
954    void testSetup() throws IOException {
955      this.mutator = connection.getBufferedMutator(tableName);
956    }
957
958    @Override
959    void testTakedown()  throws IOException {
960      if (flushCommits) {
961        this.mutator.flush();
962      }
963      mutator.close();
964    }
965  }
966
967  static class RandomSeekScanTest extends TableTest {
968    RandomSeekScanTest(Configuration conf, TestOptions options, Status status) {
969      super(conf, options, status);
970    }
971
972    @Override
973    void testRow(final int i) throws IOException {
974      Scan scan = new Scan().withStartRow(getRandomRow(this.rand, this.totalRows));
975      scan.addColumn(FAMILY_NAME, QUALIFIER_NAME);
976      scan.setFilter(new WhileMatchFilter(new PageFilter(120)));
977      ResultScanner s = this.table.getScanner(scan);
978      s.close();
979    }
980
981    @Override
982    protected int getReportingPeriod() {
983      int period = this.perClientRunRows / 100;
984      return period == 0? this.perClientRunRows: period;
985    }
986  }
987
988  @SuppressWarnings("unused")
989  static abstract class RandomScanWithRangeTest extends TableTest {
990    RandomScanWithRangeTest(Configuration conf, TestOptions options, Status status) {
991      super(conf, options, status);
992    }
993
994    @Override
995    void testRow(final int i) throws IOException {
996      Pair<byte[], byte[]> startAndStopRow = getStartAndStopRow();
997      Scan scan = new Scan().withStartRow(startAndStopRow.getFirst())
998        .withStopRow(startAndStopRow.getSecond());
999      scan.addColumn(FAMILY_NAME, QUALIFIER_NAME);
1000      ResultScanner s = this.table.getScanner(scan);
1001      int count = 0;
1002      for (Result rr = null; (rr = s.next()) != null;) {
1003        count++;
1004      }
1005
1006      if (i % 100 == 0) {
1007        LOG.info(String.format("Scan for key range %s - %s returned %s rows",
1008            Bytes.toString(startAndStopRow.getFirst()),
1009            Bytes.toString(startAndStopRow.getSecond()), count));
1010      }
1011
1012      s.close();
1013    }
1014
1015    protected abstract Pair<byte[], byte[]> getStartAndStopRow();
1016
1017    protected Pair<byte[], byte[]> generateStartAndStopRows(int maxRange) {
1018      int start = this.rand.nextInt(Integer.MAX_VALUE) % totalRows;
1019      int stop = start + maxRange;
1020      return new Pair<>(format(start), format(stop));
1021    }
1022
1023    @Override
1024    protected int getReportingPeriod() {
1025      int period = this.perClientRunRows / 100;
1026      return period == 0? this.perClientRunRows: period;
1027    }
1028  }
1029
1030  static class RandomScanWithRange10Test extends RandomScanWithRangeTest {
1031    RandomScanWithRange10Test(Configuration conf, TestOptions options, Status status) {
1032      super(conf, options, status);
1033    }
1034
1035    @Override
1036    protected Pair<byte[], byte[]> getStartAndStopRow() {
1037      return generateStartAndStopRows(10);
1038    }
1039  }
1040
1041  static class RandomScanWithRange100Test extends RandomScanWithRangeTest {
1042    RandomScanWithRange100Test(Configuration conf, TestOptions options, Status status) {
1043      super(conf, options, status);
1044    }
1045
1046    @Override
1047    protected Pair<byte[], byte[]> getStartAndStopRow() {
1048      return generateStartAndStopRows(100);
1049    }
1050  }
1051
1052  static class RandomScanWithRange1000Test extends RandomScanWithRangeTest {
1053    RandomScanWithRange1000Test(Configuration conf, TestOptions options, Status status) {
1054      super(conf, options, status);
1055    }
1056
1057    @Override
1058    protected Pair<byte[], byte[]> getStartAndStopRow() {
1059      return generateStartAndStopRows(1000);
1060    }
1061  }
1062
1063  static class RandomScanWithRange10000Test extends RandomScanWithRangeTest {
1064    RandomScanWithRange10000Test(Configuration conf, TestOptions options, Status status) {
1065      super(conf, options, status);
1066    }
1067
1068    @Override
1069    protected Pair<byte[], byte[]> getStartAndStopRow() {
1070      return generateStartAndStopRows(10000);
1071    }
1072  }
1073
1074  static class RandomReadTest extends TableTest {
1075    RandomReadTest(Configuration conf, TestOptions options, Status status) {
1076      super(conf, options, status);
1077    }
1078
1079    @Override
1080    void testRow(final int i) throws IOException {
1081      Get get = new Get(getRandomRow(this.rand, this.totalRows));
1082      get.addColumn(FAMILY_NAME, QUALIFIER_NAME);
1083      this.table.get(get);
1084    }
1085
1086    @Override
1087    protected int getReportingPeriod() {
1088      int period = this.perClientRunRows / 100;
1089      return period == 0? this.perClientRunRows: period;
1090    }
1091  }
1092
1093  static class RandomWriteTest extends BufferedMutatorTest {
1094    RandomWriteTest(Configuration conf, TestOptions options, Status status) {
1095      super(conf, options, status);
1096    }
1097
1098    @Override
1099    void testRow(final int i) throws IOException {
1100      byte[] row = getRandomRow(this.rand, this.totalRows);
1101      Put put = new Put(row);
1102      byte[] value = generateData(this.rand, ROW_LENGTH);
1103      if (useTags) {
1104        byte[] tag = generateData(this.rand, TAG_LENGTH);
1105        Tag[] tags = new Tag[noOfTags];
1106        for (int n = 0; n < noOfTags; n++) {
1107          Tag t = new ArrayBackedTag((byte) n, tag);
1108          tags[n] = t;
1109        }
1110        KeyValue kv = new KeyValue(row, FAMILY_NAME, QUALIFIER_NAME, HConstants.LATEST_TIMESTAMP,
1111            value, tags);
1112        put.add(kv);
1113      } else {
1114        put.addColumn(FAMILY_NAME, QUALIFIER_NAME, value);
1115      }
1116      put.setDurability(writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1117      mutator.mutate(put);
1118    }
1119  }
1120
1121  static class ScanTest extends TableTest {
1122    private ResultScanner testScanner;
1123
1124    ScanTest(Configuration conf, TestOptions options, Status status) {
1125      super(conf, options, status);
1126    }
1127
1128    @Override
1129    void testTakedown() throws IOException {
1130      if (this.testScanner != null) {
1131        this.testScanner.close();
1132      }
1133      super.testTakedown();
1134    }
1135
1136    @Override
1137    void testRow(final int i) throws IOException {
1138      if (this.testScanner == null) {
1139        Scan scan = new Scan().withStartRow(format(this.startRow));
1140        scan.addColumn(FAMILY_NAME, QUALIFIER_NAME);
1141        this.testScanner = table.getScanner(scan);
1142      }
1143      testScanner.next();
1144    }
1145  }
1146
1147  static class SequentialReadTest extends TableTest {
1148    SequentialReadTest(Configuration conf, TestOptions options, Status status) {
1149      super(conf, options, status);
1150    }
1151
1152    @Override
1153    void testRow(final int i) throws IOException {
1154      Get get = new Get(format(i));
1155      get.addColumn(FAMILY_NAME, QUALIFIER_NAME);
1156      table.get(get);
1157    }
1158  }
1159
1160  static class SequentialWriteTest extends BufferedMutatorTest {
1161    SequentialWriteTest(Configuration conf, TestOptions options, Status status) {
1162      super(conf, options, status);
1163    }
1164
1165    @Override
1166    void testRow(final int i) throws IOException {
1167      byte[] row = format(i);
1168      Put put = new Put(row);
1169      byte[] value = generateData(this.rand, ROW_LENGTH);
1170      if (useTags) {
1171        byte[] tag = generateData(this.rand, TAG_LENGTH);
1172        Tag[] tags = new Tag[noOfTags];
1173        for (int n = 0; n < noOfTags; n++) {
1174          Tag t = new ArrayBackedTag((byte) n, tag);
1175          tags[n] = t;
1176        }
1177        KeyValue kv = new KeyValue(row, FAMILY_NAME, QUALIFIER_NAME, HConstants.LATEST_TIMESTAMP,
1178            value, tags);
1179        put.add(kv);
1180      } else {
1181        put.addColumn(FAMILY_NAME, QUALIFIER_NAME, value);
1182      }
1183      put.setDurability(writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1184      mutator.mutate(put);
1185    }
1186  }
1187
1188  static class FilteredScanTest extends TableTest {
1189    protected static final Logger LOG = LoggerFactory.getLogger(FilteredScanTest.class.getName());
1190
1191    FilteredScanTest(Configuration conf, TestOptions options, Status status) {
1192      super(conf, options, status);
1193    }
1194
1195    @Override
1196    void testRow(int i) throws IOException {
1197      byte[] value = generateValue(this.rand);
1198      Scan scan = constructScan(value);
1199      try (ResultScanner scanner = this.table.getScanner(scan)) {
1200        while (scanner.next() != null) {
1201        }
1202      }
1203    }
1204
1205    protected Scan constructScan(byte[] valuePrefix) {
1206      Filter filter = new SingleColumnValueFilter(
1207          FAMILY_NAME, QUALIFIER_NAME, CompareOperator.EQUAL,
1208          new BinaryComparator(valuePrefix)
1209      );
1210      Scan scan = new Scan();
1211      scan.addColumn(FAMILY_NAME, QUALIFIER_NAME);
1212      scan.setFilter(filter);
1213      return scan;
1214    }
1215  }
1216
1217  /**
1218   * Format passed integer.
1219   * @param number the integer to format
1220   * @return Returns zero-prefixed 10-byte wide decimal version of passed number (Does absolute in
1221   *    case number is negative).
1222   */
1223  public static byte [] format(final int number) {
1224    byte[] b = new byte[DEFAULT_ROW_PREFIX_LENGTH + 10];
1225    int d = Math.abs(number);
1226    for (int i = b.length - 1; i >= 0; i--) {
1227      b[i] = (byte)((d % 10) + '0');
1228      d /= 10;
1229    }
1230    return b;
1231  }
1232
1233  public static byte[] generateData(final Random r, int length) {
1234    byte[] b = new byte [length];
1235    int i;
1236
1237    for (i = 0; i < (length-8); i += 8) {
1238      b[i] = (byte) (65 + r.nextInt(26));
1239      b[i+1] = b[i];
1240      b[i+2] = b[i];
1241      b[i+3] = b[i];
1242      b[i+4] = b[i];
1243      b[i+5] = b[i];
1244      b[i+6] = b[i];
1245      b[i+7] = b[i];
1246    }
1247
1248    byte a = (byte) (65 + r.nextInt(26));
1249    for (; i < length; i++) {
1250      b[i] = a;
1251    }
1252    return b;
1253  }
1254
1255  public static byte[] generateValue(final Random r) {
1256    byte [] b = new byte [ROW_LENGTH];
1257    r.nextBytes(b);
1258    return b;
1259  }
1260
1261  static byte[] getRandomRow(final Random random, final int totalRows) {
1262    return format(random.nextInt(Integer.MAX_VALUE) % totalRows);
1263  }
1264
1265  long runOneClient(final Class<? extends Test> cmd, final int startRow,
1266      final int perClientRunRows, final int totalRows,
1267      boolean flushCommits, boolean writeToWAL, boolean useTags, int noOfTags,
1268      Connection connection, final Status status) throws IOException {
1269    status.setStatus("Start " + cmd + " at offset " + startRow + " for " +
1270      perClientRunRows + " rows");
1271    long totalElapsedTime;
1272
1273    TestOptions options = new TestOptions(startRow, perClientRunRows,
1274      totalRows, tableName, flushCommits, writeToWAL, useTags, noOfTags, connection);
1275    final Test t;
1276    try {
1277      Constructor<? extends Test> constructor = cmd.getDeclaredConstructor(
1278          Configuration.class, TestOptions.class, Status.class);
1279      t = constructor.newInstance(this.conf, options, status);
1280    } catch (NoSuchMethodException e) {
1281      throw new IllegalArgumentException("Invalid command class: " +
1282          cmd.getName() + ".  It does not provide a constructor as described by" +
1283          "the javadoc comment.  Available constructors are: " +
1284          Arrays.toString(cmd.getConstructors()));
1285    } catch (Exception e) {
1286      throw new IllegalStateException("Failed to construct command class", e);
1287    }
1288    totalElapsedTime = t.test();
1289
1290    status.setStatus("Finished " + cmd + " in " + totalElapsedTime +
1291      "ms at offset " + startRow + " for " + perClientRunRows + " rows");
1292    return totalElapsedTime;
1293  }
1294
1295  private void runNIsOne(final Class<? extends Test> cmd) {
1296    Status status = LOG::info;
1297
1298    RemoteAdmin admin;
1299    try {
1300      Client client = new Client(cluster);
1301      admin = new RemoteAdmin(client, getConf());
1302      checkTable(admin);
1303      runOneClient(cmd, 0, this.R, this.R, this.flushCommits, this.writeToWAL,
1304        this.useTags, this.noOfTags, this.connection, status);
1305    } catch (Exception e) {
1306      LOG.error("Failed", e);
1307    }
1308  }
1309
1310  private void runTest(final Class<? extends Test> cmd)
1311      throws IOException, InterruptedException, ClassNotFoundException {
1312    if (N == 1) {
1313      // If there is only one client and one HRegionServer, we assume nothing
1314      // has been set up at all.
1315      runNIsOne(cmd);
1316    } else {
1317      // Else, run
1318      runNIsMoreThanOne(cmd);
1319    }
1320  }
1321
1322  protected void printUsage() {
1323    printUsage(null);
1324  }
1325
1326  protected void printUsage(final String message) {
1327    if (message != null && message.length() > 0) {
1328      System.err.println(message);
1329    }
1330    System.err.println("Usage: java " + this.getClass().getName() + " \\");
1331    System.err.println("  [--nomapred] [--rows=ROWS] [--table=NAME] \\");
1332    System.err.println("  [--compress=TYPE] [--blockEncoding=TYPE] " +
1333      "[-D<property=value>]* <command> <nclients>");
1334    System.err.println();
1335    System.err.println("General Options:");
1336    System.err.println(" nomapred        Run multiple clients using threads " +
1337      "(rather than use mapreduce)");
1338    System.err.println(" rows            Rows each client runs. Default: One million");
1339    System.err.println();
1340    System.err.println("Table Creation / Write Tests:");
1341    System.err.println(" table           Alternate table name. Default: 'TestTable'");
1342    System.err.println(" compress        Compression type to use (GZ, LZO, ...). Default: 'NONE'");
1343    System.err.println(" flushCommits    Used to determine if the test should flush the table. " +
1344      "Default: false");
1345    System.err.println(" writeToWAL      Set writeToWAL on puts. Default: True");
1346    System.err.println(" presplit        Create presplit table. Recommended for accurate perf " +
1347      "analysis (see guide).  Default: disabled");
1348    System.err.println(" usetags         Writes tags along with KVs.  Use with HFile V3. " +
1349      "Default : false");
1350    System.err.println(" numoftags        Specify the no of tags that would be needed. " +
1351      "This works only if usetags is true.");
1352    System.err.println();
1353    System.err.println("Read Tests:");
1354    System.err.println(" inmemory        Tries to keep the HFiles of the CF inmemory as far as " +
1355      "possible.  Not guaranteed that reads are always served from inmemory.  Default: false");
1356    System.err.println();
1357    System.err.println(" Note: -D properties will be applied to the conf used. ");
1358    System.err.println("  For example: ");
1359    System.err.println("   -Dmapreduce.output.fileoutputformat.compress=true");
1360    System.err.println("   -Dmapreduce.task.timeout=60000");
1361    System.err.println();
1362    System.err.println("Command:");
1363    for (CmdDescriptor command : commands.values()) {
1364      System.err.println(String.format(" %-15s %s", command.getName(), command.getDescription()));
1365    }
1366    System.err.println();
1367    System.err.println("Args:");
1368    System.err.println(" nclients      Integer. Required. Total number of " +
1369      "clients (and HRegionServers)");
1370    System.err.println("               running: 1 <= value <= 500");
1371    System.err.println("Examples:");
1372    System.err.println(" To run a single evaluation client:");
1373    System.err.println(" $ hbase " + this.getClass().getName()
1374        + " sequentialWrite 1");
1375  }
1376
1377  private void getArgs(final int start, final String[] args) {
1378    if (start + 1 > args.length) {
1379      throw new IllegalArgumentException("must supply the number of clients");
1380    }
1381    N = Integer.parseInt(args[start]);
1382    if (N < 1) {
1383      throw new IllegalArgumentException("Number of clients must be > 1");
1384    }
1385    // Set total number of rows to write.
1386    R = R * N;
1387  }
1388
1389  @Override
1390  public int run(String[] args) throws Exception {
1391    // Process command-line args. TODO: Better cmd-line processing
1392    // (but hopefully something not as painful as cli options).
1393    int errCode = -1;
1394    if (args.length < 1) {
1395      printUsage();
1396      return errCode;
1397    }
1398
1399    try {
1400      for (int i = 0; i < args.length; i++) {
1401        String cmd = args[i];
1402        if (cmd.equals("-h") || cmd.startsWith("--h")) {
1403          printUsage();
1404          errCode = 0;
1405          break;
1406        }
1407
1408        final String nmr = "--nomapred";
1409        if (cmd.startsWith(nmr)) {
1410          nomapred = true;
1411          continue;
1412        }
1413
1414        final String rows = "--rows=";
1415        if (cmd.startsWith(rows)) {
1416          R = Integer.parseInt(cmd.substring(rows.length()));
1417          continue;
1418        }
1419
1420        final String table = "--table=";
1421        if (cmd.startsWith(table)) {
1422          this.tableName = TableName.valueOf(cmd.substring(table.length()));
1423          continue;
1424        }
1425
1426        final String compress = "--compress=";
1427        if (cmd.startsWith(compress)) {
1428          this.compression = Compression.Algorithm.valueOf(cmd.substring(compress.length()));
1429          continue;
1430        }
1431
1432        final String blockEncoding = "--blockEncoding=";
1433        if (cmd.startsWith(blockEncoding)) {
1434          this.blockEncoding = DataBlockEncoding.valueOf(cmd.substring(blockEncoding.length()));
1435          continue;
1436        }
1437
1438        final String flushCommits = "--flushCommits=";
1439        if (cmd.startsWith(flushCommits)) {
1440          this.flushCommits = Boolean.parseBoolean(cmd.substring(flushCommits.length()));
1441          continue;
1442        }
1443
1444        final String writeToWAL = "--writeToWAL=";
1445        if (cmd.startsWith(writeToWAL)) {
1446          this.writeToWAL = Boolean.parseBoolean(cmd.substring(writeToWAL.length()));
1447          continue;
1448        }
1449
1450        final String presplit = "--presplit=";
1451        if (cmd.startsWith(presplit)) {
1452          this.presplitRegions = Integer.parseInt(cmd.substring(presplit.length()));
1453          continue;
1454        }
1455
1456        final String inMemory = "--inmemory=";
1457        if (cmd.startsWith(inMemory)) {
1458          this.inMemoryCF = Boolean.parseBoolean(cmd.substring(inMemory.length()));
1459          continue;
1460        }
1461
1462        this.connection = ConnectionFactory.createConnection(getConf());
1463
1464        final String useTags = "--usetags=";
1465        if (cmd.startsWith(useTags)) {
1466          this.useTags = Boolean.parseBoolean(cmd.substring(useTags.length()));
1467          continue;
1468        }
1469
1470        final String noOfTags = "--nooftags=";
1471        if (cmd.startsWith(noOfTags)) {
1472          this.noOfTags = Integer.parseInt(cmd.substring(noOfTags.length()));
1473          continue;
1474        }
1475
1476        final String host = "--host=";
1477        if (cmd.startsWith(host)) {
1478          cluster.add(cmd.substring(host.length()));
1479          continue;
1480        }
1481
1482        Class<? extends Test> cmdClass = determineCommandClass(cmd);
1483        if (cmdClass != null) {
1484          getArgs(i + 1, args);
1485          if (cluster.isEmpty()) {
1486            String s = conf.get("stargate.hostname", "localhost");
1487            if (s.contains(":")) {
1488              cluster.add(s);
1489            } else {
1490              cluster.add(s, conf.getInt("stargate.port", 8080));
1491            }
1492          }
1493          runTest(cmdClass);
1494          errCode = 0;
1495          break;
1496        }
1497
1498        printUsage();
1499        break;
1500      }
1501    } catch (Exception e) {
1502      LOG.error("Failed", e);
1503    }
1504
1505    return errCode;
1506  }
1507
1508  private Class<? extends Test> determineCommandClass(String cmd) {
1509    CmdDescriptor descriptor = commands.get(cmd);
1510    return descriptor != null ? descriptor.getCmdClass() : null;
1511  }
1512
1513  public static void main(final String[] args) throws Exception {
1514    int res = ToolRunner.run(new PerformanceEvaluation(HBaseConfiguration.create()), args);
1515    System.exit(res);
1516  }
1517}