accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From e..@apache.org
Subject accumulo git commit: ACCUMULO-3508 added read-write lock to ZooCache
Date Wed, 14 Oct 2015 19:15:47 GMT
Repository: accumulo
Updated Branches:
  refs/heads/master 10cafac94 -> dd420f631


ACCUMULO-3508 added read-write lock to ZooCache


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

Branch: refs/heads/master
Commit: dd420f6312322d13b814a177fc87fabb0576dc04
Parents: 10cafac
Author: Eric C. Newton <eric.newton@gmail.com>
Authored: Wed Oct 14 15:12:23 2015 -0400
Committer: Eric C. Newton <eric.newton@gmail.com>
Committed: Wed Oct 14 15:13:42 2015 -0400

----------------------------------------------------------------------
 .../accumulo/fate/zookeeper/ZooCache.java       | 325 +++++++++++--------
 1 file changed, 195 insertions(+), 130 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/dd420f63/fate/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java
----------------------------------------------------------------------
diff --git a/fate/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java b/fate/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java
index f043d83..8c9f80d 100644
--- a/fate/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java
+++ b/fate/src/main/java/org/apache/accumulo/fate/zookeeper/ZooCache.java
@@ -28,6 +28,10 @@ import java.util.ConcurrentModificationException;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.LockSupport;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.Code;
@@ -46,14 +50,18 @@ import com.google.common.annotations.VisibleForTesting;
 public class ZooCache {
   private static final Logger log = LoggerFactory.getLogger(ZooCache.class);
 
-  private ZCacheWatcher watcher = new ZCacheWatcher();
-  private Watcher externalWatcher = null;
+  private final ZCacheWatcher watcher = new ZCacheWatcher();
+  private final Watcher externalWatcher;
 
-  private HashMap<String,byte[]> cache;
-  private HashMap<String,Stat> statCache;
-  private HashMap<String,List<String>> childrenCache;
+  private final ReadWriteLock cacheLock = new ReentrantReadWriteLock(false);
+  private final Lock cacheWriteLock = cacheLock.writeLock();
+  private final Lock cacheReadLock = cacheLock.readLock();
 
-  private ZooReader zReader;
+  private final HashMap<String,byte[]> cache;
+  private final HashMap<String,Stat> statCache;
+  private final HashMap<String,List<String>> childrenCache;
+
+  private final ZooReader zReader;
 
   private ZooKeeper getZooKeeper() {
     return zReader.getZooKeeper();
@@ -63,8 +71,9 @@ public class ZooCache {
     @Override
     public void process(WatchedEvent event) {
 
-      if (log.isTraceEnabled())
+      if (log.isTraceEnabled()) {
         log.trace("{}", event);
+      }
 
       switch (event.getType()) {
         case NodeDataChanged:
@@ -89,10 +98,12 @@ public class ZooCache {
               break;
             default:
               log.warn("Unhandled: " + event);
+              break;
           }
           break;
         default:
           log.warn("Unhandled: " + event);
+          break;
       }
 
       if (externalWatcher != null) {
@@ -143,50 +154,50 @@ public class ZooCache {
     this.externalWatcher = watcher;
   }
 
-  private interface ZooRunnable {
+  private abstract class ZooRunnable<T> {
     /**
      * Runs an operation against ZooKeeper, automatically retrying in the face of KeeperExceptions
      */
-    void run(ZooKeeper zooKeeper) throws KeeperException, InterruptedException;
-  }
+    abstract T run(ZooKeeper zooKeeper) throws KeeperException, InterruptedException;
 
-  private synchronized void retry(ZooRunnable op) {
+    public T retry() {
 
-    int sleepTime = 100;
+      int sleepTime = 100;
 
-    while (true) {
+      while (true) {
 
-      ZooKeeper zooKeeper = getZooKeeper();
+        ZooKeeper zooKeeper = getZooKeeper();
 
-      try {
-        op.run(zooKeeper);
-        return;
-
-      } catch (KeeperException e) {
-        final Code code = e.code();
-        if (code == Code.NONODE) {
-          log.error("Looked up non-existent node in cache " + e.getPath(), e);
-        } else if (code == Code.CONNECTIONLOSS || code == Code.OPERATIONTIMEOUT || code ==
Code.SESSIONEXPIRED) {
-          log.warn("Saw (possibly) transient exception communicating with ZooKeeper, will
retry", e);
-        } else {
-          log.warn("Zookeeper error, will retry", e);
+        try {
+          return run(zooKeeper);
+        } catch (KeeperException e) {
+          final Code code = e.code();
+          if (code == Code.NONODE) {
+            log.error("Looked up non-existent node in cache " + e.getPath(), e);
+          } else if (code == Code.CONNECTIONLOSS || code == Code.OPERATIONTIMEOUT || code
== Code.SESSIONEXPIRED) {
+            log.warn("Saw (possibly) transient exception communicating with ZooKeeper, will
retry", e);
+          } else {
+            log.warn("Zookeeper error, will retry", e);
+          }
+        } catch (InterruptedException e) {
+          log.info("Zookeeper error, will retry", e);
+        } catch (ConcurrentModificationException e) {
+          log.debug("Zookeeper was modified, will retry");
         }
-      } catch (InterruptedException e) {
-        log.info("Zookeeper error, will retry", e);
-      } catch (ConcurrentModificationException e) {
-        log.debug("Zookeeper was modified, will retry");
-      }
 
-      try {
-        // do not hold lock while sleeping
-        wait(sleepTime);
-      } catch (InterruptedException e) {
-        log.debug("Wait in retry() was interrupted.", e);
+        try {
+          // do not hold lock while sleeping
+          Thread.sleep(sleepTime);
+        } catch (InterruptedException e) {
+          log.debug("Wait in retry() was interrupted.", e);
+        }
+        LockSupport.parkNanos(sleepTime);
+        if (sleepTime < 10_000) {
+          sleepTime = (int) (sleepTime + sleepTime * Math.random());
+        }
       }
-      if (sleepTime < 10000)
-        sleepTime = (int) (sleepTime + sleepTime * Math.random());
-
     }
+
   }
 
   /**
@@ -198,29 +209,41 @@ public class ZooCache {
    */
   public synchronized List<String> getChildren(final String zPath) {
 
-    ZooRunnable zr = new ZooRunnable() {
+    ZooRunnable<List<String>> zr = new ZooRunnable<List<String>>()
{
 
       @Override
-      public void run(ZooKeeper zooKeeper) throws KeeperException, InterruptedException {
-
-        if (childrenCache.containsKey(zPath))
-          return;
+      public List<String> run(ZooKeeper zooKeeper) throws KeeperException, InterruptedException
{
+        try {
+          cacheReadLock.lock();
+          if (childrenCache.containsKey(zPath)) {
+            return childrenCache.get(zPath);
+          }
+        } finally {
+          cacheReadLock.unlock();
+        }
 
+        cacheWriteLock.lock();
         try {
+          if (childrenCache.containsKey(zPath)) {
+            return childrenCache.get(zPath);
+          }
           List<String> children = zooKeeper.getChildren(zPath, watcher);
           childrenCache.put(zPath, children);
+          return children;
         } catch (KeeperException ke) {
           if (ke.code() != Code.NONODE) {
             throw ke;
           }
+        } finally {
+          cacheWriteLock.unlock();
         }
+        return null;
       }
 
     };
 
-    retry(zr);
+    List<String> children = zr.retry();
 
-    List<String> children = childrenCache.get(zPath);
     if (children == null) {
       return null;
     }
@@ -234,7 +257,7 @@ public class ZooCache {
    *          path to get
    * @return path data, or null if non-existent
    */
-  public synchronized byte[] get(final String zPath) {
+  public byte[] get(final String zPath) {
     return get(zPath, null);
   }
 
@@ -243,94 +266,127 @@ public class ZooCache {
    *
    * @param zPath
    *          path to get
-   * @param stat
+   * @param status
    *          status object to populate
    * @return path data, or null if non-existent
    */
-  public synchronized byte[] get(final String zPath, Stat stat) {
-    ZooRunnable zr = new ZooRunnable() {
+  public byte[] get(final String zPath, final Stat status) {
+    ZooRunnable<byte[]> zr = new ZooRunnable<byte[]>() {
 
       @Override
-      public void run(ZooKeeper zooKeeper) throws KeeperException, InterruptedException {
-
-        if (cache.containsKey(zPath))
-          return;
-
-        // The following call to exists() is important, since we are caching that a node
does not exist. Once the node comes into existence, it will be added to
-        // the cache. But this notification of a node coming into existence will only be
given if exists() was previously called.
-        //
-        // If the call to exists() is bypassed and only getData() is called with a special
case that looks for Code.NONODE in the KeeperException, then
-        // non-existence can not be cached.
-        Stat stat = zooKeeper.exists(zPath, watcher);
-
-        byte[] data = null;
-
-        if (stat == null) {
-          if (log.isTraceEnabled())
-            log.trace("zookeeper did not contain " + zPath);
-        } else {
-          try {
-            data = zooKeeper.getData(zPath, watcher, stat);
-          } catch (KeeperException.BadVersionException e1) {
-            throw new ConcurrentModificationException();
-          } catch (KeeperException.NoNodeException e2) {
-            throw new ConcurrentModificationException();
+      public byte[] run(ZooKeeper zooKeeper) throws KeeperException, InterruptedException
{
+        Stat stat = null;
+        cacheReadLock.lock();
+        try {
+          if (cache.containsKey(zPath)) {
+            stat = statCache.get(zPath);
+            copyStats(status, stat);
+            return cache.get(zPath);
           }
-          if (log.isTraceEnabled())
-            log.trace("zookeeper contained " + zPath + " " + (data == null ? null : new String(data,
UTF_8)));
+        } finally {
+          cacheReadLock.unlock();
         }
-        if (log.isTraceEnabled())
-          log.trace("putting " + zPath + " " + (data == null ? null : new String(data, UTF_8))
+ " in cache");
-        put(zPath, data, stat);
-      }
 
+        /*
+         * The following call to exists() is important, since we are caching that a node
does not exist. Once the node comes into existence, it will be added to
+         * the cache. But this notification of a node coming into existence will only be
given if exists() was previously called.
+         *
+         * If the call to exists() is bypassed and only getData() is called with a special
case that looks for Code.NONODE in the KeeperException, then
+         * non-existence can not be cached.
+         */
+        cacheWriteLock.lock();
+        try {
+          stat = zooKeeper.exists(zPath, watcher);
+          byte[] data = null;
+          if (stat == null) {
+            if (log.isTraceEnabled()) {
+              log.trace("zookeeper did not contain " + zPath);
+            }
+          } else {
+            try {
+              data = zooKeeper.getData(zPath, watcher, stat);
+            } catch (KeeperException.BadVersionException e1) {
+              throw new ConcurrentModificationException();
+            } catch (KeeperException.NoNodeException e2) {
+              throw new ConcurrentModificationException();
+            }
+            if (log.isTraceEnabled()) {
+              log.trace("zookeeper contained " + zPath + " " + (data == null ? null : new
String(data, UTF_8)));
+            }
+          }
+          put(zPath, data, stat);
+          copyStats(status, stat);
+          statCache.put(zPath, stat);
+          return data;
+        } finally {
+          cacheWriteLock.unlock();
+        }
+      }
     };
 
-    retry(zr);
+    return zr.retry();
+  }
 
-    if (stat != null) {
-      Stat cstat = statCache.get(zPath);
-      if (cstat != null) {
-        try {
-          ByteArrayOutputStream baos = new ByteArrayOutputStream();
-          DataOutputStream dos = new DataOutputStream(baos);
-          cstat.write(dos);
-          dos.close();
-
-          ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
-          DataInputStream dis = new DataInputStream(bais);
-          stat.readFields(dis);
-
-          dis.close();
-        } catch (IOException e) {
-          throw new RuntimeException(e);
-        }
+  /**
+   * Helper method to copy stats from the cached stat into userStat
+   *
+   * @param userStat
+   *          user Stat object
+   * @param cachedStat
+   *          cached statistic, that is or will be cached
+   */
+  protected void copyStats(Stat userStat, Stat cachedStat) {
+    if (userStat != null && cachedStat != null) {
+      try {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos);
+        cachedStat.write(dos);
+        dos.close();
+
+        ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
+        DataInputStream dis = new DataInputStream(bais);
+        userStat.readFields(dis);
+
+        dis.close();
+      } catch (IOException e) {
+        throw new RuntimeException(e);
       }
     }
-
-    return cache.get(zPath);
   }
 
-  private synchronized void put(String zPath, byte[] data, Stat stat) {
-    cache.put(zPath, data);
-    statCache.put(zPath, stat);
+  private void put(String zPath, byte[] data, Stat stat) {
+    cacheWriteLock.lock();
+    try {
+      cache.put(zPath, data);
+      statCache.put(zPath, stat);
+    } finally {
+      cacheWriteLock.unlock();
+    }
   }
 
-  private synchronized void remove(String zPath) {
-    if (log.isTraceEnabled())
-      log.trace("removing " + zPath + " from cache");
-    cache.remove(zPath);
-    childrenCache.remove(zPath);
-    statCache.remove(zPath);
+  private void remove(String zPath) {
+    cacheWriteLock.lock();
+    try {
+      cache.remove(zPath);
+      childrenCache.remove(zPath);
+      statCache.remove(zPath);
+    } finally {
+      cacheWriteLock.unlock();
+    }
   }
 
   /**
    * Clears this cache.
    */
   public synchronized void clear() {
-    cache.clear();
-    childrenCache.clear();
-    statCache.clear();
+    cacheWriteLock.lock();
+    try {
+      cache.clear();
+      childrenCache.clear();
+      statCache.clear();
+    } finally {
+      cacheWriteLock.unlock();
+    }
   }
 
   /**
@@ -353,8 +409,13 @@ public class ZooCache {
    * @return true if children are cached
    */
   @VisibleForTesting
-  synchronized boolean childrenCached(String zPath) {
-    return childrenCache.containsKey(zPath);
+  boolean childrenCached(String zPath) {
+    cacheReadLock.lock();
+    try {
+      return childrenCache.containsKey(zPath);
+    } finally {
+      cacheReadLock.unlock();
+    }
   }
 
   /**
@@ -363,24 +424,28 @@ public class ZooCache {
    * @param zPath
    *          path of top node
    */
-  public synchronized void clear(String zPath) {
-
-    for (Iterator<String> i = cache.keySet().iterator(); i.hasNext();) {
-      String path = i.next();
-      if (path.startsWith(zPath))
-        i.remove();
-    }
+  public void clear(String zPath) {
+    cacheWriteLock.lock();
+    try {
+      for (Iterator<String> i = cache.keySet().iterator(); i.hasNext();) {
+        String path = i.next();
+        if (path.startsWith(zPath))
+          i.remove();
+      }
 
-    for (Iterator<String> i = childrenCache.keySet().iterator(); i.hasNext();) {
-      String path = i.next();
-      if (path.startsWith(zPath))
-        i.remove();
-    }
+      for (Iterator<String> i = childrenCache.keySet().iterator(); i.hasNext();) {
+        String path = i.next();
+        if (path.startsWith(zPath))
+          i.remove();
+      }
 
-    for (Iterator<String> i = statCache.keySet().iterator(); i.hasNext();) {
-      String path = i.next();
-      if (path.startsWith(zPath))
-        i.remove();
+      for (Iterator<String> i = statCache.keySet().iterator(); i.hasNext();) {
+        String path = i.next();
+        if (path.startsWith(zPath))
+          i.remove();
+      }
+    } finally {
+      cacheWriteLock.unlock();
     }
   }
 


Mime
View raw message