accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject [3/6] git commit: ACCUMULO-2489 Fixes race condition in TableConfiguration where NPE may occur.
Date Thu, 20 Mar 2014 22:44:28 GMT
ACCUMULO-2489 Fixes race condition in TableConfiguration where NPE may occur.

Make invalidateCache much more efficient by calling clear on ZooCache
instead of creating a new one.


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

Branch: refs/heads/master
Commit: 1d608a81f488c2bf371fc81f83e0022bd2943a36
Parents: 63d5e55
Author: Josh Elser <elserj@apache.org>
Authored: Thu Mar 20 18:08:55 2014 -0400
Committer: Josh Elser <elserj@apache.org>
Committed: Thu Mar 20 18:08:55 2014 -0400

----------------------------------------------------------------------
 .../server/conf/ServerConfiguration.java        |   1 +
 .../server/conf/TableConfiguration.java         |  80 ++++++----
 .../test/TableConfigurationUpdateTest.java      | 152 +++++++++++++++++++
 3 files changed, 200 insertions(+), 33 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/1d608a81/server/src/main/java/org/apache/accumulo/server/conf/ServerConfiguration.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/accumulo/server/conf/ServerConfiguration.java
b/server/src/main/java/org/apache/accumulo/server/conf/ServerConfiguration.java
index b2acd1a..8653274 100644
--- a/server/src/main/java/org/apache/accumulo/server/conf/ServerConfiguration.java
+++ b/server/src/main/java/org/apache/accumulo/server/conf/ServerConfiguration.java
@@ -27,6 +27,7 @@ import org.apache.accumulo.core.conf.ConfigSanityCheck;
 import org.apache.accumulo.core.conf.DefaultConfiguration;
 import org.apache.accumulo.core.conf.SiteConfiguration;
 import org.apache.accumulo.core.data.KeyExtent;
+import org.apache.accumulo.server.client.HdfsZooInstance;
 
 public class ServerConfiguration {
   

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1d608a81/server/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
b/server/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
index 59ff1f7..7a3d6e4 100644
--- a/server/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
+++ b/server/src/main/java/org/apache/accumulo/server/conf/TableConfiguration.java
@@ -37,37 +37,46 @@ import org.apache.log4j.Logger;
 
 public class TableConfiguration extends AccumuloConfiguration {
   private static final Logger log = Logger.getLogger(TableConfiguration.class);
-  
+
   // Need volatile keyword to ensure double-checked locking works as intended
   private static volatile ZooCache tablePropCache = null;
+  private static final Object initLock = new Object();
+
   private final String instanceId;
+  private final Instance instance;
   private final AccumuloConfiguration parent;
-  
+
   private String table = null;
   private Set<ConfigurationObserver> observers;
-  
+
   public TableConfiguration(String instanceId, String table, AccumuloConfiguration parent)
{
+    this(instanceId, HdfsZooInstance.getInstance(), table, parent);
+  }
+
+  public TableConfiguration(String instanceId, Instance instance, String table, AccumuloConfiguration
parent) {
     this.instanceId = instanceId;
+    this.instance = instance;
     this.table = table;
     this.parent = parent;
-    
+
     this.observers = Collections.synchronizedSet(new HashSet<ConfigurationObserver>());
   }
-  
-  /**
-   * @deprecated not for client use
-   */
-  @Deprecated
-  private static ZooCache getTablePropCache() {
-    Instance inst = HdfsZooInstance.getInstance();
-    if (tablePropCache == null)
-      synchronized (TableConfiguration.class) {
-        if (tablePropCache == null)
-          tablePropCache = new ZooCache(inst.getZooKeepers(), inst.getZooKeepersSessionTimeOut(),
new TableConfWatcher(inst));
+
+  private void initializeZooCache() {
+    synchronized (initLock) {
+      if (null == tablePropCache) {
+        tablePropCache = new ZooCache(instance.getZooKeepers(), instance.getZooKeepersSessionTimeOut(),
new TableConfWatcher(instance));
       }
+    }
+  }
+
+  private ZooCache getTablePropCache() {
+    if (null == tablePropCache) {
+      initializeZooCache();
+    }
     return tablePropCache;
   }
-  
+
   public void addObserver(ConfigurationObserver co) {
     if (table == null) {
       String err = "Attempt to add observer for non-table configuration";
@@ -77,7 +86,7 @@ public class TableConfiguration extends AccumuloConfiguration {
     iterator();
     observers.add(co);
   }
-  
+
   public void removeObserver(ConfigurationObserver configObserver) {
     if (table == null) {
       String err = "Attempt to remove observer for non-table configuration";
@@ -86,29 +95,29 @@ public class TableConfiguration extends AccumuloConfiguration {
     }
     observers.remove(configObserver);
   }
-  
+
   public void expireAllObservers() {
     Collection<ConfigurationObserver> copy = Collections.unmodifiableCollection(observers);
     for (ConfigurationObserver co : copy)
       co.sessionExpired();
   }
-  
+
   public void propertyChanged(String key) {
     Collection<ConfigurationObserver> copy = Collections.unmodifiableCollection(observers);
     for (ConfigurationObserver co : copy)
       co.propertyChanged(key);
   }
-  
+
   public void propertiesChanged(String key) {
     Collection<ConfigurationObserver> copy = Collections.unmodifiableCollection(observers);
     for (ConfigurationObserver co : copy)
       co.propertiesChanged();
   }
-  
+
   public String get(Property property) {
     String key = property.getKey();
     String value = get(key);
-    
+
     if (value == null || !property.getType().isValidFormat(value)) {
       if (value != null)
         log.error("Using default value for " + key + " due to improperly formatted " + property.getType()
+ ": " + value);
@@ -116,23 +125,24 @@ public class TableConfiguration extends AccumuloConfiguration {
     }
     return value;
   }
-  
+
   private String get(String key) {
     String zPath = ZooUtil.getRoot(instanceId) + Constants.ZTABLES + "/" + table + Constants.ZTABLE_CONF
+ "/" + key;
+
     byte[] v = getTablePropCache().get(zPath);
     String value = null;
     if (v != null)
       value = new String(v, Constants.UTF8);
     return value;
   }
-  
+
   @Override
   public Iterator<Entry<String,String>> iterator() {
     TreeMap<String,String> entries = new TreeMap<String,String>();
-    
+
     for (Entry<String,String> parentEntry : parent)
       entries.put(parentEntry.getKey(), parentEntry.getValue());
-    
+
     List<String> children = getTablePropCache().getChildren(ZooUtil.getRoot(instanceId)
+ Constants.ZTABLES + "/" + table + Constants.ZTABLE_CONF);
     if (children != null) {
       for (String child : children) {
@@ -141,10 +151,10 @@ public class TableConfiguration extends AccumuloConfiguration {
           entries.put(child, value);
       }
     }
-    
+
     return entries.entrySet().iterator();
   }
-  
+
   public String getTableId() {
     return table;
   }
@@ -152,11 +162,15 @@ public class TableConfiguration extends AccumuloConfiguration {
   @Override
   public void invalidateCache() {
     if (null != tablePropCache) {
-      synchronized (TableConfiguration.class) {
-        if (null != tablePropCache) {
-          tablePropCache = null;
-        }
-      }
+      tablePropCache.clear();
     }
+    // Else, if the cache is null, we could lock and double-check
+    // to see if it happened to be created so we could invalidate it
+    // but I don't see much benefit coming from that extra check.
+  }
+  
+  @Override
+  public String toString() {
+    return this.getClass().getSimpleName();
   }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/1d608a81/test/src/test/java/org/apache/accumulo/test/TableConfigurationUpdateTest.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/TableConfigurationUpdateTest.java
b/test/src/test/java/org/apache/accumulo/test/TableConfigurationUpdateTest.java
new file mode 100644
index 0000000..30da268
--- /dev/null
+++ b/test/src/test/java/org/apache/accumulo/test/TableConfigurationUpdateTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.DefaultConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.minicluster.MiniAccumuloCluster;
+import org.apache.accumulo.server.conf.TableConfiguration;
+import org.apache.log4j.Logger;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TableConfigurationUpdateTest {
+  private static final Logger log = Logger.getLogger(TableConfigurationUpdateTest.class);
+
+  public static TemporaryFolder folder = new TemporaryFolder();
+  private MiniAccumuloCluster accumulo;
+  private String secret = "secret";
+
+  @Before
+  public void setUp() throws Exception {
+    folder.create();
+    accumulo = new MiniAccumuloCluster(folder.getRoot(), secret);
+    accumulo.start();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    accumulo.stop();
+    folder.delete();
+  }
+
+  @Test
+  public void test() throws Exception {
+    ZooKeeperInstance inst = new ZooKeeperInstance(accumulo.getInstanceName(), accumulo.getZooKeepers(),
60 * 1000);
+    Connector conn = inst.getConnector("root", new PasswordToken(secret));
+
+    String table = "foo";
+    conn.tableOperations().create(table);
+
+    final DefaultConfiguration defaultConf = AccumuloConfiguration.getDefaultConfiguration();
+
+    // Cache invalidates 25% of the time
+    int randomMax = 4;
+    // Number of threads
+    int numThreads = 2;
+    // Number of iterations per thread
+    int iterations = 100000;
+    AccumuloConfiguration tableConf = new TableConfiguration(inst.getInstanceID(), inst,
table, defaultConf);
+    
+    long start = System.currentTimeMillis();
+    ExecutorService svc = Executors.newFixedThreadPool(numThreads);
+    CountDownLatch countDown = new CountDownLatch(numThreads);
+    ArrayList<Future<Exception>> futures = new ArrayList<Future<Exception>>(numThreads);
+
+    for (int i = 0; i < numThreads; i++) {
+      futures.add(svc.submit(new TableConfRunner(randomMax, iterations, tableConf, countDown)));
+    }
+
+    svc.shutdown();
+    Assert.assertTrue(svc.awaitTermination(60, TimeUnit.MINUTES));
+
+    for (Future<Exception> fut : futures) {
+      Exception e = fut.get();
+      if (null != e) {
+        Assert.fail("Thread failed with exception " + e);
+      }
+    }
+
+    long end = System.currentTimeMillis();
+    log.debug(tableConf + " with " + iterations + " iterations and " + numThreads + " threads
and cache invalidates "
+        + ((1. / randomMax) * 100.) + "% took " + (end - start) / 1000 + " second(s)");
+  }
+
+  public static class TableConfRunner implements Callable<Exception> {
+    private static final Property prop = Property.TABLE_SPLIT_THRESHOLD;
+    private AccumuloConfiguration tableConf;
+    private CountDownLatch countDown;
+    private int iterations, randMax;
+
+    public TableConfRunner(int randMax, int iterations, AccumuloConfiguration tableConf,
CountDownLatch countDown) {
+      this.randMax = randMax;
+      this.iterations = iterations;
+      this.tableConf = tableConf;
+      this.countDown = countDown;
+    }
+
+    @Override
+    public Exception call() {
+      Random r = new Random();
+      countDown.countDown();
+      try {
+        countDown.await();
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        return e;
+      }
+
+      String t = Thread.currentThread().getName() + " ";
+      try {
+        for (int i = 0; i < iterations; i++) {
+          // if (i % 10000 == 0) {
+          // log.info(t + " " + i);
+          // }
+          int choice = r.nextInt(randMax);
+          if (choice < 1) {
+            tableConf.invalidateCache();
+          } else {
+            tableConf.get(prop);
+          }
+        }
+      } catch (Exception e) {
+        log.error(t, e);
+        return e;
+      }
+
+      return null;
+    }
+
+  }
+
+}


Mime
View raw message