accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject svn commit: r1341596 - /accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
Date Tue, 22 May 2012 19:32:51 GMT
Author: kturner
Date: Tue May 22 19:32:51 2012
New Revision: 1341596

URL: http://svn.apache.org/viewvc?rev=1341596&view=rev
Log:
ACCUMULO-348 parallelized adding splits to a table

Modified:
    accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java

Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java?rev=1341596&r1=1341595&r2=1341596&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
(original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperationsImpl.java
Tue May 22 19:32:51 2012
@@ -32,6 +32,11 @@ import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeMap;
 import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.client.AccumuloException;
@@ -301,6 +306,62 @@ public class TableOperationsImpl extends
     }
   }
   
+  private static class SplitEnv {
+    private String tableName;
+    private String tableId;
+    private ExecutorService executor;
+    private CountDownLatch latch;
+    private AtomicReference<Exception> exception;
+    
+    SplitEnv(String tableName, String tableId, ExecutorService executor, CountDownLatch latch,
AtomicReference<Exception> exception) {
+      this.tableName = tableName;
+      this.tableId = tableId;
+      this.executor = executor;
+      this.latch = latch;
+      this.exception = exception;
+    }
+  }
+  
+  private class SplitTask implements Runnable {
+    
+    private List<Text> splits;
+    private SplitEnv env;
+    
+    SplitTask(SplitEnv env, List<Text> splits) {
+      this.env = env;
+      this.splits = splits;
+    }
+    
+    @Override
+    public void run() {
+      try {
+        if (env.exception.get() != null)
+          return;
+        
+        if (splits.size() <= 2) {
+          addSplits(env.tableName, new TreeSet<Text>(splits), env.tableId);
+          for (int i = 0; i < splits.size(); i++)
+            env.latch.countDown();
+          return;
+        }
+        
+        int mid = splits.size() / 2;
+        
+        // split the middle split point to ensure that child task split different tablets
and can therefore
+        // run in parallel
+        addSplits(env.tableName, new TreeSet<Text>(splits.subList(mid, mid + 1)), env.tableId);
+        env.latch.countDown();
+        
+        env.executor.submit(new SplitTask(env, splits.subList(0, mid)));
+        env.executor.submit(new SplitTask(env, splits.subList(mid + 1, splits.size())));
+        
+      } catch (Exception e) {
+        env.exception.compareAndSet(null, e);
+      }
+    }
+    
+  }
+
   /**
    * @param tableName
    *          the name of the table
@@ -315,7 +376,40 @@ public class TableOperationsImpl extends
    */
   public void addSplits(String tableName, SortedSet<Text> partitionKeys) throws TableNotFoundException,
AccumuloException, AccumuloSecurityException {
     String tableId = Tables.getTableId(instance, tableName);
-    addSplits(tableName, partitionKeys, tableId);
+    
+    List<Text> splits = new ArrayList<Text>(partitionKeys);
+    // should be sorted because we copied from a sorted set, but that makes assumptions about
+    // how the copy was done so resort to be sure.
+    Collections.sort(splits);
+
+    CountDownLatch latch = new CountDownLatch(splits.size());
+    AtomicReference<Exception> exception = new AtomicReference<Exception>(null);
+    
+    ExecutorService executor = Executors.newFixedThreadPool(16);
+    try {
+      executor.submit(new SplitTask(new SplitEnv(tableName, tableId, executor, latch, exception),
splits));
+
+      while (!latch.await(100, TimeUnit.MILLISECONDS)) {
+        if (exception.get() != null) {
+          executor.shutdownNow();
+          Exception excep = exception.get();
+          if (excep instanceof TableNotFoundException)
+            throw (TableNotFoundException) excep;
+          else if (excep instanceof AccumuloException)
+            throw (AccumuloException) excep;
+          else if (excep instanceof AccumuloSecurityException)
+            throw (AccumuloSecurityException) excep;
+          else if (excep instanceof RuntimeException)
+            throw (RuntimeException) excep;
+          else
+            throw new RuntimeException(excep);
+        }
+      }
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    } finally {
+      executor.shutdown();
+    }
   }
   
   private void addSplits(String tableName, SortedSet<Text> partitionKeys, String tableId)
throws AccumuloException, AccumuloSecurityException,



Mime
View raw message