HBASE-10419 Add multiget support to PerformanceEvaluation

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1569826 13f79535-47bb-0310-9956-ffa450edef68

hbase-server/src/test/java/org/apache/hadoop/hbase/PerformanceEvaluation.java

@@ -137,6 +137,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
   private int presplitRegions = 0;
   private boolean useTags = false;
   private int noOfTags = 1;
+  private int multiGet = 0;
   private HConnection connection;
 
   private static final Path PERF_EVAL_DIR = new Path("performance_evaluation");
@@ -153,7 +154,8 @@ public class PerformanceEvaluation extends Configured implements Tool {
         "writeToWAL=(\\w+),\\s+" +
         "useTags=(\\w+),\\s+" +
         "noOfTags=(\\d+),\\s+" +
-        "reportLatency=(\\w+)");
+        "reportLatency=(\\w+),\\s+" +
+        "multiGet=(\\d+)");
 
   /**
    * Enum for map metrics.  Keep it out here rather than inside in the Map
@@ -235,12 +237,13 @@ public class PerformanceEvaluation extends Configured implements Tool {
     private boolean useTags = false;
     private int noOfTags = 0;
     private boolean reportLatency = false;
+    private int multiGet = 0;
 
     public PeInputSplit() {}
 
     public PeInputSplit(TableName tableName, int startRow, int rows, int totalRows,
         float sampleRate, int clients, boolean flushCommits, boolean writeToWAL,
-        boolean useTags, int noOfTags, boolean reportLatency) {
+        boolean useTags, int noOfTags, boolean reportLatency, int multiGet) {
       this.tableName = tableName;
       this.startRow = startRow;
       this.rows = rows;
@@ -252,6 +255,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
       this.useTags = useTags;
       this.noOfTags = noOfTags;
       this.reportLatency = reportLatency;
+      this.multiGet = multiGet;
     }
 
     @Override
@@ -271,6 +275,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
       this.useTags = in.readBoolean();
       this.noOfTags = in.readInt();
       this.reportLatency = in.readBoolean();
+      this.multiGet = in.readInt();
     }
 
     @Override
@@ -288,6 +293,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
       out.writeBoolean(useTags);
       out.writeInt(noOfTags);
       out.writeBoolean(reportLatency);
+      out.writeInt(multiGet);
     }
 
     @Override
@@ -343,6 +349,10 @@ public class PerformanceEvaluation extends Configured implements Tool {
     public boolean isReportLatency() {
       return reportLatency;
     }
+
+    public int getMultiGet() {
+      return multiGet;
+    }
   }
 
   /**
@@ -384,6 +394,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
             boolean useTags = Boolean.parseBoolean(m.group(9));
             int noOfTags = Integer.parseInt(m.group(10));
             boolean reportLatency = Boolean.parseBoolean(m.group(11));
+            int multiGet = Integer.parseInt(m.group(12));
 
             LOG.debug("tableName=" + tableName +
                       " split["+ splitList.size() + "] " +
@@ -396,11 +407,12 @@ public class PerformanceEvaluation extends Configured implements Tool {
                       " writeToWAL=" + writeToWAL +
                       " useTags=" + useTags +
                       " noOfTags=" + noOfTags +
-                      " reportLatency=" + reportLatency);
+                      " reportLatency=" + reportLatency +
+                      " multiGet=" + multiGet);
 
             PeInputSplit newSplit =
               new PeInputSplit(tableName, startRow, rows, totalRows, sampleRate, clients,
-                flushCommits, writeToWAL, useTags, noOfTags, reportLatency);
+                flushCommits, writeToWAL, useTags, noOfTags, reportLatency, multiGet);
             splitList.add(newSplit);
           }
         }
@@ -523,7 +535,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
       long elapsedTime = this.pe.runOneClient(this.cmd, value.getStartRow(),
           value.getRows(), value.getTotalRows(), value.getSampleRate(),
           value.isFlushCommits(), value.isWriteToWAL(), value.isUseTags(),
-          value.getNoOfTags(), value.isReportLatency(),
+          value.getNoOfTags(), value.isReportLatency(), value.getMultiGet(),
           HConnectionManager.createConnection(context.getConfiguration()), status);
       // Collect how much time the thing took. Report as map output and
       // to the ELAPSED_TIME counter.
@@ -634,6 +646,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
     final int preSplitRegions = this.presplitRegions;
     final boolean useTags = this.useTags;
     final int numTags = this.noOfTags;
+    final int multiGet = this.multiGet;
     final HConnection connection = HConnectionManager.createConnection(getConf());
     for (int i = 0; i < this.N; i++) {
       final int index = i;
@@ -654,10 +667,11 @@ public class PerformanceEvaluation extends Configured implements Tool {
           pe.connection = connection;
           pe.useTags = useTags;
           pe.noOfTags = numTags;
+          pe.multiGet = multiGet;
           try {
             long elapsedTime = pe.runOneClient(cmd, index * perClientRows,
                perClientRows, R, sampleRate, flushCommits, writeToWal, useTags,
-               noOfTags, reportLatency, connection, new Status() {
+               noOfTags, reportLatency, multiGet, connection, new Status() {
                   public void setStatus(final String msg) throws IOException {
                     LOG.info("client-" + getName() + " " + msg);
                   }
@@ -772,7 +786,8 @@ public class PerformanceEvaluation extends Configured implements Tool {
           ", writeToWAL=" + this.writeToWAL +
           ", useTags=" + this.useTags +
           ", noOfTags=" + this.noOfTags +
-          ", reportLatency=" + this.reportLatency;
+          ", reportLatency=" + this.reportLatency +
+          ", multiGet=" + this.multiGet;
           int hash = h.hash(Bytes.toBytes(s));
           m.put(hash, s);
         }
@@ -829,13 +844,15 @@ public class PerformanceEvaluation extends Configured implements Tool {
     private boolean useTags = false;
     private int noOfTags = 0;
     private boolean reportLatency;
+    private int multiGet = 0;
     private HConnection connection;
 
     TestOptions() {}
 
     TestOptions(int startRow, int perClientRunRows, int totalRows, float sampleRate,
         int numClientThreads, TableName tableName, boolean flushCommits, boolean writeToWAL,
-        boolean useTags, int noOfTags, boolean reportLatency, HConnection connection) {
+        boolean useTags, int noOfTags, boolean reportLatency, int multiGet,
+        HConnection connection) {
       this.startRow = startRow;
       this.perClientRunRows = perClientRunRows;
       this.totalRows = totalRows;
@@ -847,6 +864,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
       this.useTags = useTags;
       this.noOfTags = noOfTags;
       this.reportLatency = reportLatency;
+      this.multiGet = multiGet;
       this.connection = connection;
     }
 
@@ -886,6 +904,10 @@ public class PerformanceEvaluation extends Configured implements Tool {
       return reportLatency;
     }
 
+    public int getMultiGet() {
+      return multiGet;
+    }
+
     public HConnection getConnection() {
       return connection;
     }
@@ -1122,6 +1144,8 @@ public class PerformanceEvaluation extends Configured implements Tool {
     private final int everyN;
     private final boolean reportLatency;
     private final double[] times;
+    private final int multiGet;
+    private ArrayList<Get> gets;
     int idx = 0;
 
     RandomReadTest(Configuration conf, TestOptions options, Status status) {
@@ -1129,10 +1153,15 @@ public class PerformanceEvaluation extends Configured implements Tool {
       everyN = (int) (this.totalRows / (this.totalRows * this.sampleRate));
       LOG.info("Sampling 1 every " + everyN + " out of " + perClientRunRows + " total rows.");
       this.reportLatency = options.isReportLatency();
+      this.multiGet = options.getMultiGet();
+      if (this.multiGet > 0) {
+        LOG.info("MultiGet enabled. Sending GETs in batches of " + this.multiGet + ".");
+        this.gets = new ArrayList<Get>(this.multiGet);
+      }
       if (this.reportLatency) {
-        times = new double[(int) Math.ceil(this.perClientRunRows * this.sampleRate)];
+        this.times = new double[(int) Math.ceil(this.perClientRunRows * this.sampleRate / Math.max(1, this.multiGet))];
       } else {
-        times = null;
+        this.times = null;
       }
     }
 
@@ -1141,10 +1170,22 @@ public class PerformanceEvaluation extends Configured implements Tool {
       if (i % everyN == 0) {
         Get get = new Get(getRandomRow(this.rand, this.totalRows));
         get.addColumn(FAMILY_NAME, QUALIFIER_NAME);
+        if (this.multiGet > 0) {
+          this.gets.add(get);
+          if (this.gets.size() == this.multiGet) {
+            long start = System.nanoTime();
+            this.table.get(this.gets);
+            if (this.reportLatency) {
+              times[idx++] = (System.nanoTime() - start) / 1e6;
+            }
+            this.gets.clear();
+          }
+        } else {
           long start = System.nanoTime();
           this.table.get(get);
           if (this.reportLatency) {
-          times[idx++] = (float) ((System.nanoTime() - start) / 1000000.0);
+            times[idx++] = (System.nanoTime() - start) / 1e6;
+          }
         }
       }
     }
@@ -1157,6 +1198,10 @@ public class PerformanceEvaluation extends Configured implements Tool {
 
     @Override
     protected void testTakedown() throws IOException {
+      if (this.gets != null && this.gets.size() > 0) {
+        this.table.get(gets);
+        this.gets.clear();
+      }
       super.testTakedown();
       if (this.reportLatency) {
         Arrays.sort(times);
@@ -1379,7 +1424,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
   long runOneClient(final Class<? extends Test> cmd, final int startRow,
       final int perClientRunRows, final int totalRows, final float sampleRate,
       boolean flushCommits, boolean writeToWAL, boolean useTags, int noOfTags,
-      boolean reportLatency, HConnection connection, final Status status)
+      boolean reportLatency, int multiGet, HConnection connection, final Status status)
   throws IOException {
     status.setStatus("Start " + cmd + " at offset " + startRow + " for " +
       perClientRunRows + " rows");
@@ -1387,7 +1432,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
 
     TestOptions options = new TestOptions(startRow, perClientRunRows,
       totalRows, sampleRate, N, tableName, flushCommits, writeToWAL, useTags, noOfTags,
-      reportLatency, connection);
+      reportLatency, multiGet, connection);
     final Test t;
     try {
       Constructor<? extends Test> constructor = cmd.getDeclaredConstructor(
@@ -1421,7 +1466,8 @@ public class PerformanceEvaluation extends Configured implements Tool {
       admin = new HBaseAdmin(getConf());
       checkTable(admin);
       runOneClient(cmd, 0, this.R, this.R, this.sampleRate, this.flushCommits,
-      this.writeToWAL, this.useTags, this.noOfTags, this.reportLatency, this.connection, status);
+      this.writeToWAL, this.useTags, this.noOfTags, this.reportLatency, this.multiGet,
+        this.connection, status);
     } catch (Exception e) {
       LOG.error("Failed", e);
     } finally {
@@ -1599,6 +1645,12 @@ public class PerformanceEvaluation extends Configured implements Tool {
           continue;
         }
 
+        final String multiGet = "--multiGet=";
+        if (cmd.startsWith(multiGet)) {
+          this.multiGet = Integer.parseInt(cmd.substring(multiGet.length()));
+          continue;
+        }
+
         this.connection = HConnectionManager.createConnection(getConf());
         
         final String useTags = "--usetags=";