accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [6/7] git commit: ACCUMULO-3259 Be a little smarter in Bulk.xml to let the backlog of tasks work itself down before adding more.
Date Fri, 24 Oct 2014 20:20:55 GMT
ACCUMULO-3259 Be a little smarter in Bulk.xml to let the backlog of tasks work itself down
before adding more.

If the queue size on the threadpool is greater than 50 times the number of
tservers, wait 30 seconds before adding more tasks. This should still ensure
that we can keep Accumulo sufficiently busy without creating an absurd
number of tasks that the test client will never reasonably work through.


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/1477c130
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/1477c130
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/1477c130

Branch: refs/heads/1.6
Commit: 1477c130143bd2477fb8e29bc22ba4698aae3599
Parents: c4f3fc8
Author: Josh Elser <elserj@apache.org>
Authored: Fri Oct 24 15:01:22 2014 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Fri Oct 24 15:24:47 2014 -0400

----------------------------------------------------------------------
 .../test/randomwalk/bulk/BulkImportTest.java    | 84 ++++++++++++++++++++
 .../test/randomwalk/bulk/BulkMinusOne.java      |  8 +-
 .../test/randomwalk/bulk/BulkPlusOne.java       | 20 ++---
 .../accumulo/test/randomwalk/bulk/BulkTest.java |  8 +-
 .../accumulo/test/randomwalk/bulk/Compact.java  |  6 +-
 .../test/randomwalk/bulk/ConsistencyCheck.java  |  6 +-
 .../accumulo/test/randomwalk/bulk/Merge.java    | 12 +--
 .../test/randomwalk/bulk/SelectiveBulkTest.java | 41 ++++++++++
 .../test/randomwalk/bulk/SelectiveQueueing.java | 48 +++++++++++
 .../accumulo/test/randomwalk/bulk/Split.java    |  6 +-
 10 files changed, 206 insertions(+), 33 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkImportTest.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkImportTest.java
b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkImportTest.java
new file mode 100644
index 0000000..aa44741
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkImportTest.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.test.randomwalk.bulk;
+
+import java.util.Properties;
+
+import org.apache.accumulo.test.randomwalk.State;
+
+/**
+ * If we have a sufficient back-up of imports, let them work off before adding even more
bulk-imports. Imports of PlusOne must always be balanced with imports
+ * of MinusOne.
+ */
+public abstract class BulkImportTest extends BulkTest {
+
+  public static final String SKIPPED_IMPORT = "skipped.import", TRUE = Boolean.TRUE.toString(),
FALSE = Boolean.FALSE.toString();
+
+  @Override
+  public void visit(final State state, Properties props) throws Exception {
+    /**
+     * Each visit() is performed sequentially and then submitted to the threadpool which
will have async execution. As long as we're checking the state and
+     * making decisions about what to do before we submit something to the thread pool, we're
fine.
+     */
+
+    String lastImportSkipped = state.getString(SKIPPED_IMPORT);
+    // We have a marker in the state for the previous insert, we have to balance skipping
BulkPlusOne
+    // with skipping the new BulkMinusOne to make sure that we maintain consistency
+    if (null != lastImportSkipped) {
+      if (!getClass().equals(BulkMinusOne.class)) {
+        throw new IllegalStateException("Should not have a skipped import marker for a class
other than " + BulkMinusOne.class.getName() + " but was "
+            + getClass().getName());
+      }
+
+      if (TRUE.equals(lastImportSkipped)) {
+        log.debug("Last import was skipped, skipping this import to ensure consistency");
+        state.remove(SKIPPED_IMPORT);
+
+        // Wait 30s to balance the skip of a BulkPlusOne/BulkMinusOne pair
+        log.debug("Waiting 30s before continuing");
+        try {
+          Thread.sleep(30 * 1000);
+        } catch (InterruptedException e) {}
+
+        return;
+      } else {
+        // last import was not skipped, remove the marker
+        state.remove(SKIPPED_IMPORT);
+      }
+    }
+
+    if (shouldQueueMoreImports(state)) {
+      super.visit(state, props);
+    } else {
+      log.debug("Not queuing more imports this round because too many are already queued");
+      state.set(SKIPPED_IMPORT, TRUE);
+      // Don't sleep here, let the sleep happen when we skip the next BulkMinusOne
+    }
+  }
+
+  private boolean shouldQueueMoreImports(State state) throws Exception {
+    // Only selectively import when it's BulkPlusOne. If we did a BulkPlusOne,
+    // we must also do a BulkMinusOne to keep the table consistent
+    if (getClass().equals(BulkPlusOne.class)) {
+      // Only queue up more imports if the number of queued tasks already
+      // exceeds the number of tservers by 50x
+      return SelectiveQueueing.shouldQueueOperation(state);
+    }
+
+    return true;
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
index 4ebf23f..1704e49 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkMinusOne.java
@@ -20,14 +20,14 @@ import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.test.randomwalk.State;
 
-public class BulkMinusOne extends BulkTest {
-  
+public class BulkMinusOne extends BulkImportTest {
+
   private static final Value negOne = new Value("-1".getBytes(Constants.UTF8));
-  
+
   @Override
   protected void runLater(State state) throws Exception {
     log.info("Decrementing");
     BulkPlusOne.bulkLoadLots(log, state, negOne);
   }
-  
+
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkPlusOne.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkPlusOne.java
b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkPlusOne.java
index cdfbb36..6d56f13 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkPlusOne.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkPlusOne.java
@@ -38,8 +38,8 @@ import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
 import org.apache.log4j.Logger;
 
-public class BulkPlusOne extends BulkTest {
-  
+public class BulkPlusOne extends BulkImportTest {
+
   public static final int LOTS = 100000;
   public static final int COLS = 10;
   public static final int HEX_SIZE = (int) Math.ceil(Math.log(LOTS) / Math.log(16));
@@ -53,7 +53,7 @@ public class BulkPlusOne extends BulkTest {
   }
   public static final Text MARKER_CF = new Text("marker");
   static final AtomicLong counter = new AtomicLong();
-  
+
   private static final Value ONE = new Value("1".getBytes());
 
   static void bulkLoadLots(Logger log, State state, Value value) throws Exception {
@@ -64,22 +64,22 @@ public class BulkPlusOne extends BulkTest {
     final FileSystem fs = (FileSystem) state.get("fs");
     fs.mkdirs(fail);
     final int parts = rand.nextInt(10) + 1;
-    
+
     TreeSet<Integer> startRows = new TreeSet<Integer>();
     startRows.add(0);
     while (startRows.size() < parts)
       startRows.add(rand.nextInt(LOTS));
-    
+
     List<String> printRows = new ArrayList<String>(startRows.size());
     for (Integer row : startRows)
       printRows.add(String.format(FMT, row));
-    
+
     String markerColumnQualifier = String.format("%07d", counter.incrementAndGet());
     log.debug("preparing bulk files with start rows " + printRows + " last row " + String.format(FMT,
LOTS - 1) + " marker " + markerColumnQualifier);
-    
+
     List<Integer> rows = new ArrayList<Integer>(startRows);
     rows.add(LOTS);
-    
+
     for (int i = 0; i < parts; i++) {
       String fileName = dir + "/" + String.format("part_%d.", i) + RFile.EXTENSION;
       FileSKVWriter f = FileOperations.getInstance().openWriter(fileName, fs, fs.getConf(),
defaultConfiguration);
@@ -105,11 +105,11 @@ public class BulkPlusOne extends BulkTest {
     fs.delete(fail, true);
     log.debug("Finished bulk import, start rows " + printRows + " last row " + String.format(FMT,
LOTS - 1) + " marker " + markerColumnQualifier);
   }
-  
+
   @Override
   protected void runLater(State state) throws Exception {
     log.info("Incrementing");
     bulkLoadLots(log, state, ONE);
   }
-  
+
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkTest.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkTest.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkTest.java
index 4afefd9..b24f61a 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkTest.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/BulkTest.java
@@ -22,7 +22,7 @@ import org.apache.accumulo.test.randomwalk.State;
 import org.apache.accumulo.test.randomwalk.Test;
 
 public abstract class BulkTest extends Test {
-  
+
   @Override
   public void visit(final State state, Properties props) throws Exception {
     Setup.run(state, new Runnable() {
@@ -34,10 +34,10 @@ public abstract class BulkTest extends Test {
           log.error(ex, ex);
         }
       }
-      
+
     });
   }
-  
+
   abstract protected void runLater(State state) throws Exception;
-  
+
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Compact.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Compact.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Compact.java
index 86dae5c..8b17256 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Compact.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Compact.java
@@ -19,8 +19,8 @@ package org.apache.accumulo.test.randomwalk.bulk;
 import org.apache.accumulo.test.randomwalk.State;
 import org.apache.hadoop.io.Text;
 
-public class Compact extends BulkTest {
-  
+public class Compact extends SelectiveBulkTest {
+
   @Override
   protected void runLater(State state) throws Exception {
     final Text[] points = Merge.getRandomTabletRange(state);
@@ -29,5 +29,5 @@ public class Compact extends BulkTest {
     state.getConnector().tableOperations().compact(Setup.getTableName(), points[0], points[1],
false, true);
     log.info("Compaction " + rangeString + " finished");
   }
-  
+
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/ConsistencyCheck.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/ConsistencyCheck.java
b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/ConsistencyCheck.java
index e60f8cf..7e528a7 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/ConsistencyCheck.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/ConsistencyCheck.java
@@ -28,8 +28,8 @@ import org.apache.accumulo.core.security.Authorizations;
 import org.apache.accumulo.test.randomwalk.State;
 import org.apache.hadoop.io.Text;
 
-public class ConsistencyCheck extends BulkTest {
-  
+public class ConsistencyCheck extends SelectiveBulkTest {
+
   @Override
   protected void runLater(State state) throws Exception {
     Random rand = (Random) state.get("rand");
@@ -52,5 +52,5 @@ public class ConsistencyCheck extends BulkTest {
         throw new RuntimeException("Inconsistent value at " + entry.getKey() + " was " +
entry.getValue() + " should be " + v + " first read at " + first);
     }
   }
-  
+
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Merge.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Merge.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Merge.java
index 2dd0345..9d66e0c 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Merge.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Merge.java
@@ -22,8 +22,8 @@ import java.util.Random;
 import org.apache.accumulo.test.randomwalk.State;
 import org.apache.hadoop.io.Text;
 
-public class Merge extends BulkTest {
-  
+public class Merge extends SelectiveBulkTest {
+
   @Override
   protected void runLater(State state) throws Exception {
     Text[] points = getRandomTabletRange(state);
@@ -31,15 +31,15 @@ public class Merge extends BulkTest {
     state.getConnector().tableOperations().merge(Setup.getTableName(), points[0], points[1]);
     log.info("merging " + rangeToString(points) + " complete");
   }
-  
+
   public static String rangeToString(Text[] points) {
     return "(" + (points[0] == null ? "-inf" : points[0]) + " -> " + (points[1] == null
? "+inf" : points[1]) + "]";
   }
-  
+
   public static Text getRandomRow(Random rand) {
     return new Text(String.format(BulkPlusOne.FMT, (rand.nextLong() & 0x7fffffffffffffffl)
% BulkPlusOne.LOTS));
   }
-  
+
   public static Text[] getRandomTabletRange(State state) {
     Random rand = (Random) state.get("rand");
     Text points[] = {getRandomRow(rand), getRandomRow(rand),};
@@ -56,5 +56,5 @@ public class Merge extends BulkTest {
     }
     return points;
   }
-  
+
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/SelectiveBulkTest.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/SelectiveBulkTest.java
b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/SelectiveBulkTest.java
new file mode 100644
index 0000000..ca66775
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/SelectiveBulkTest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.test.randomwalk.bulk;
+
+import java.util.Properties;
+
+import org.apache.accumulo.test.randomwalk.State;
+
+/**
+ * Selectively runs the actual {@link BulkTest} based on the number of active TServers and
the number of queued operations.
+ */
+public abstract class SelectiveBulkTest extends BulkTest {
+
+  @Override
+  public void visit(State state, Properties props) throws Exception {
+    if (SelectiveQueueing.shouldQueueOperation(state)) {
+      super.visit(state, props);
+    } else {
+      log.debug("Skipping queueing of " + getClass().getSimpleName() + " because of excessive
queued tasks already");
+      log.debug("Waiting 30 seconds before continuing");
+      try {
+        Thread.sleep(30 * 1000);
+      } catch (InterruptedException e) {}
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/SelectiveQueueing.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/SelectiveQueueing.java
b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/SelectiveQueueing.java
new file mode 100644
index 0000000..e6dde4d
--- /dev/null
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/SelectiveQueueing.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.test.randomwalk.bulk;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.test.randomwalk.State;
+import org.apache.log4j.Logger;
+
+/**
+ * Chooses whether or not an operation should be queued based on the current thread pool
queue length and the number of available TServers.
+ */
+public class SelectiveQueueing {
+  private static final Logger log = Logger.getLogger(SelectiveQueueing.class);
+
+  public static boolean shouldQueueOperation(State state) throws Exception {
+    final ThreadPoolExecutor pool = (ThreadPoolExecutor) state.get("pool");
+    long queuedThreads = pool.getTaskCount() - pool.getActiveCount() - pool.getCompletedTaskCount();
+    final Connector conn = state.getConnector();
+    int numTservers = conn.instanceOperations().getTabletServers().size();
+
+    if (!shouldQueue(queuedThreads, numTservers)) {
+      log.info("Not queueing because of " + queuedThreads + " outstanding tasks");
+      return false;
+    }
+
+    return true;
+  }
+
+  private static boolean shouldQueue(long queuedThreads, int numTservers) {
+    return queuedThreads < numTservers * 50;
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1477c130/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Split.java
----------------------------------------------------------------------
diff --git a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Split.java b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Split.java
index 157e2ab..7a93321 100644
--- a/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Split.java
+++ b/test/src/main/java/org/apache/accumulo/test/randomwalk/bulk/Split.java
@@ -23,8 +23,8 @@ import java.util.TreeSet;
 import org.apache.accumulo.test.randomwalk.State;
 import org.apache.hadoop.io.Text;
 
-public class Split extends BulkTest {
-  
+public class Split extends SelectiveBulkTest {
+
   @Override
   protected void runLater(State state) throws Exception {
     SortedSet<Text> splits = new TreeSet<Text>();
@@ -36,5 +36,5 @@ public class Split extends BulkTest {
     state.getConnector().tableOperations().addSplits(Setup.getTableName(), splits);
     log.info("split for " + splits + " finished");
   }
-  
+
 }


Mime
View raw message