Copied: manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java (from r1539618, manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZookeeperLockManager.java) URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java?p2=manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java&p1=manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZookeeperLockManager.java&r1=1539618&r2=1540056&rev=1540056&view=diff ============================================================================== --- manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZookeeperLockManager.java (original) +++ manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java Fri Nov 8 14:44:19 2013 @@ -29,51 +29,48 @@ import java.io.*; /** The lock manager manages locks across all threads and JVMs and cluster members, using Zookeeper. * There should be no more than ONE instance of this class per thread!!! The factory should enforce this. -* Note well: This class extends LockManager because zookeeper implementations of some features (e.g. -* critical sections) do not need to be overridden. */ -public class ZookeeperLockManager extends LockManager implements ILockManager +public class ZooKeeperLockManager extends BaseLockManager implements ILockManager { public static final String _rcsid = "@(#)$Id: LockManager.java 988245 2010-08-23 18:39:35Z kwright $"; protected final static String zookeeperConnectStringParameter = "org.apache.manifoldcf.zookeeper.connectstring"; protected final static String zookeeperSessionTimeoutParameter = "org.apache.manifoldcf.zookeeper.sessiontimeout"; - - // Zookeeper construction values - protected final String connectString; - protected final int sessionTimeout; - // One zookeeper client per thread - protected ZooKeeper zookeeper = null; - protected ZookeeperWatcher zookeeperWatcher = null; + private final static String CONFIGURATION_PATH = "org.apache.manifoldcf.configuration"; + private final static String RESOURCE_PATH_PREFIX = "org.apache.manifoldcf.resources/"; + private final static String FLAG_PATH_PREFIX = "org.apache.manifoldcf.flags/"; + + // ZooKeeper connection pool + protected static Integer connectionPoolLock = new Integer(0); + protected static ZooKeeperConnectionPool pool = null; + protected static Integer zookeeperPoolLocker = new Integer(0); + protected static LockPool myZooKeeperLocks = null; /** Constructor */ - public ZookeeperLockManager() - throws ManifoldCFException - { - connectString = ManifoldCF.getStringProperty(zookeeperConnectStringParameter,null); - if (connectString != null) - throw new ManifoldCFException("Zookeeper lock manager requires a valid "+zookeeperConnectStringParameter+" property"); - sessionTimeout = ManifoldCF.getIntProperty(zookeeperSessionTimeoutParameter,300000); - } - - public ZooKeeper getSession() + public ZooKeeperLockManager() throws ManifoldCFException { - if (zookeeper == null) + synchronized (connectionPoolLock) { - try + if (pool == null) { - ZookeeperWatcher zookeeperWatcher = new ZookeeperWatcher(); - zookeeper = new ZooKeeper(connectString, sessionTimeout, zookeeperWatcher); - this.zookeeperWatcher = zookeeperWatcher; + // Initialize the ZooKeeper connection pool + String connectString = ManifoldCF.getStringProperty(zookeeperConnectStringParameter,null); + if (connectString != null) + throw new ManifoldCFException("Zookeeper lock manager requires a valid "+zookeeperConnectStringParameter+" property"); + int sessionTimeout = ManifoldCF.getIntProperty(zookeeperSessionTimeoutParameter,300000); + ManifoldCF.addShutdownHook(new ZooKeeperShutdown()); + pool = new ZooKeeperConnectionPool(connectString, sessionTimeout); } - catch (IOException e) + } + synchronized (zookeeperPoolLocker) + { + if (myZooKeeperLocks == null) { - throw new ManifoldCFException("Zookeeper initialization error: "+e.getMessage(),e); + myZooKeeperLocks = new LockPool(new ZooKeeperLockObjectFactory(pool)); } } - return zookeeper; } /** Get the current shared configuration. This configuration is available in common among all nodes, @@ -85,8 +82,27 @@ public class ZookeeperLockManager extend public ManifoldCFConfiguration getSharedConfiguration() throws ManifoldCFException { - // MHL - return null; + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + // Read as a byte array, then parse + byte[] configurationData = connection.readData(CONFIGURATION_PATH); + if (configurationData != null) + return new ManifoldCFConfiguration(new ByteArrayInputStream(configurationData)); + else + return new ManifoldCFConfiguration(); + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } } /** Raise a flag. Use this method to assert a condition, or send a global signal. The flag will be reset when the @@ -97,7 +113,22 @@ public class ZookeeperLockManager extend public void setGlobalFlag(String flagName) throws ManifoldCFException { - // MHL + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.setGlobalFlag(FLAG_PATH_PREFIX + flagName); + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } } /** Clear a flag. Use this method to clear a condition, or retract a global signal. @@ -107,7 +138,22 @@ public class ZookeeperLockManager extend public void clearGlobalFlag(String flagName) throws ManifoldCFException { - // MHL + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.clearGlobalFlag(FLAG_PATH_PREFIX + flagName); + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } } /** Check the condition of a specified flag. @@ -118,8 +164,22 @@ public class ZookeeperLockManager extend public boolean checkGlobalFlag(String flagName) throws ManifoldCFException { - // MHL - return false; + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + return connection.checkGlobalFlag(FLAG_PATH_PREFIX + flagName); + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } } /** Read data from a shared data resource. Use this method to read any existing data, or get a null back if there is no such resource. @@ -131,8 +191,22 @@ public class ZookeeperLockManager extend public byte[] readData(String resourceName) throws ManifoldCFException { - // MHL - return null; + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + return connection.readData(RESOURCE_PATH_PREFIX + resourceName); + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } } /** Write data to a shared data resource. Use this method to write a body of data into a shared resource. @@ -144,270 +218,69 @@ public class ZookeeperLockManager extend public void writeData(String resourceName, byte[] data) throws ManifoldCFException { - // MHL - } - - /** Enter a write locked code area (i.e., block out both readers and other writers). - * Write locks permit only ONE thread to be in the named section, across JVM's - * as well. In order to guarantee this, the current thread may wait until all other - * threads have left the section. - *@param lockKey is the name of the lock. - */ - @Override - public void enterWriteLock(String lockKey) - throws ManifoldCFException - { - // From Zookeeper recipes - - ZooKeeper zk = getSession(); - - /* - 1. Call create( ) to create a node with pathname "guid-/write-". This is the lock node - spoken of later in the protocol. Make sure to set both sequence and ephemeral flags. - If a recoverable error occurs calling create() the client should call getChildren() and - check for a node containing the guid used in the path name. This handles the case - (noted above) of the create() succeeding on the server but the server crashing before - returning the name of the new node. - */ - - /* - 2. Call getChildren( ) on the lock node without setting the watch flag - this is important, - as it avoids the herd effect. - */ - - /* - 3. If there are no children with a lower sequence number than the node created in step - 1, the client has the lock and the client exits the protocol. - */ - - /* - 4. Call exists( ), with watch flag set, on the node with the pathname that has the next - lowest sequence number. - */ - - /* 5. If exists( ) returns false, goto step 2. Otherwise, wait for a notification for the - pathname from the previous step before going to step 2. - */ - - // MHL - } - - /** Enter a write locked code area (i.e., block out both readers and other writers), - * but do not wait if the lock cannot be obtained. - * Write locks permit only ONE thread to be in the named section, across JVM's - * as well. In order to guarantee this, an exception (LockException) will be - * thrown if the lock condition cannot be met immediately. - *@param lockKey is the name of the lock. - */ - @Override - public void enterWriteLockNoWait(String lockKey) - throws ManifoldCFException, LockException - { - // MHL - } - - /** Leave a write locked code area. Use this method to exit a write-locked section. The lockKey - * parameter must correspond to the key used for the enter method. - * @param lockKey is the name of the lock. - */ - @Override - public void leaveWriteLock(String lockKey) - throws ManifoldCFException - { - // From Zookeeper recipes - - /* - Delete the node we created in step 1 above. - */ - - // MHL - } - - /** Enter a non-exclusive write-locked area (blocking out all readers, but letting in other "writers"). - * This kind of lock is designed to be used in conjunction with read locks. It is used typically in - * a situation where the read lock represents a query and the non-exclusive write lock represents a modification - * to an individual item that might affect the query, but where multiple modifications do not individually - * interfere with one another (use of another, standard, write lock per item can guarantee this). - * This method works across JVMs, and may wait if the required lock cannot be immediately obtained. - *@param lockKey is the name of the lock. - */ - @Override - public void enterNonExWriteLock(String lockKey) - throws ManifoldCFException - { - // MHL - } - - /** Enter a non-exclusive write-locked area (blocking out all readers, but letting in other "writers"). - * This kind of lock is designed to be used in conjunction with read locks. It is used typically in - * a situation where the read lock represents a query and the non-exclusive write lock represents a modification - * to an individual item that might affect the query, but where multiple modifications do not individually - * interfere with one another (use of another, standard, write lock per item can guarantee this). - * This method works across JVMs, and will throw LockException if the lock condition cannot be immediately met. - *@param lockKey is the name of the lock. - */ - @Override - public void enterNonExWriteLockNoWait(String lockKey) - throws ManifoldCFException, LockException - { - // MHL - } - - /** Leave a non-exclusive write locked code area. Use this method to exit a non-ex-write-locked section. - * The lockKey - * parameter must correspond to the key used for the enter method. - *@param lockKey is the name of the lock. - */ - @Override - public void leaveNonExWriteLock(String lockKey) - throws ManifoldCFException - { - // MHL - } - - /** Enter a read-only locked area (i.e., block ONLY if there's a writer). This kind of lock - * permits multiple threads inside the same code area, but only if there is no "writer" in the - * same section at the same time. - * This method works across JVMs, and may wait if the required lock cannot be immediately obtained. - *@param lockKey is the name of the lock. - */ - @Override - public void enterReadLock(String lockKey) - throws ManifoldCFException - { - - // From Zookeeper recipes - - /* - 1. Call create( ) to create a node with pathname "guid-/read-". This is the lock node use later in the - protocol. Make sure to set both the sequence and ephemeral flags. - If a recoverable error occurs calling create() the client should call getChildren() and - check for a node containing the guid used in the path name. This handles the case - (noted above) of the create() succeeding on the server but the server crashing before - returning the name of the new node. - */ - - /* - 2. Call getChildren( ) on the lock node without setting the watch flag - this is important, as it - avoids the herd effect. - */ - - /* - 3. If there are no children with a pathname starting with "write-" and having a lower - sequence number than the node created in step 1, the client has the lock and can exit the protocol. - */ - - /* - 4. Otherwise, call exists( ), with watch flag, set on the node in lock directory with pathname - staring with "write-" having the next lowest sequence number. - */ - - /* - 5. If exists( ) returns false, goto step 2. - */ - - /* - 6. Otherwise, wait for a notification for the pathname from the previous step before going to step 2 - */ - - // MHL - } - - /** Enter a read-only locked area (i.e., block ONLY if there's a writer). This kind of lock - * permits multiple threads inside the same code area, but only if there is no "writer" in the - * same section at the same time. - * This method works across JVMs, and will throw LockException if the required lock cannot be immediately met. - *@param lockKey is the name of the lock. - */ - @Override - public void enterReadLockNoWait(String lockKey) - throws ManifoldCFException, LockException - { - // MHL - } - - /** Leave a read-locked code area. Use this method to exit a read-locked section. The lockKey - * parameter must correspond to the key used for the enter method. - *@param lockKey is the name of the lock. - */ - @Override - public void leaveReadLock(String lockKey) - throws ManifoldCFException - { - // From Zookeeper recipes - - /* - Delete the node we created in step 1 above. - */ - - // MHL - } - - /** Enter multiple locks simultaneously. Use this method if a series or set of locks needs to be - * thrown for an operation to take place. This operation will avoid deadlock if all the locks are - * thrown at the start of the area using this method. - * This method works cross-JVM, and will wait if the required locks are not available. - *@param readLocks is an array of read lock names, or null if there are no read locks desired. - *@param nonExWriteLocks is an array of non-ex write lock names, or null if none desired. - *@param writeLocks is an array of write lock names, or null if there are none desired. - */ - @Override - public void enterLocks(String[] readLocks, String[] nonExWriteLocks, String[] writeLocks) - throws ManifoldCFException - { - // MHL - } - - /** Enter multiple locks simultaneously. Use this method if a series or set of locks needs to be - * thrown for an operation to take place. This operation will avoid deadlock if all the locks are - * thrown at the start of the area using this method. - * This method works cross-JVM, and will throw LockException if the required locks are not available. - *@param readLocks is an array of read lock names, or null if there are no read locks desired. - *@param nonExWriteLocks is an array of non-ex write lock names, or null if none desired. - *@param writeLocks is an array of write lock names, or null if there are none desired. - */ - @Override - public void enterLocksNoWait(String[] readLocks, String[] nonExWriteLocks, String[] writeLocks) - throws ManifoldCFException, LockException - { - // MHL + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.writeData(RESOURCE_PATH_PREFIX + resourceName, data); + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } } - /** Leave multiple locks. Use this method to leave a section started with enterLocks() or - * enterLocksNoWait(). The parameters must correspond to those passed to the enter method. - *@param readLocks is an array of read lock names, or null if there are no read locks desired. - *@param nonExWriteLocks is an array of non-ex write lock names, or null if none desired. - *@param writeLocks is an array of write lock names, or null if there are none desired. + /** Override this method to change the nature of global locks. */ @Override - public void leaveLocks(String[] readLocks, String[] nonExWriteLocks, String[] writeLocks) - throws ManifoldCFException + protected LockPool getGlobalLockPool() { - // MHL + return myZooKeeperLocks; } - /** Clear all outstanding locks in the system. - * This is a very dangerous method to use (obviously)... + /** Shutdown the connection pool. */ - @Override - public void clearLocks() + protected static void shutdownPool() throws ManifoldCFException { - // MHL + synchronized (connectionPoolLock) + { + if (pool != null) + { + try + { + pool.closeAll(); + pool = null; + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } + } + } } - - /** Watcher class for zookeeper, so we get notified about zookeeper events. */ - protected static class ZookeeperWatcher implements Watcher + + protected static class ZooKeeperShutdown implements IShutdownHook { - public ZookeeperWatcher() + public ZooKeeperShutdown() { - // MHL } - public void process(WatchedEvent event) + /** Do the requisite cleanup. + */ + @Override + public void doCleanup() + throws ManifoldCFException { - // MHL + shutdownPool(); } } + }
Added: manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObject.java URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObject.java?rev=1540056&view=auto ============================================================================== --- manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObject.java (added) +++ manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObject.java Fri Nov 8 14:44:19 2013 @@ -0,0 +1,137 @@ +/* $Id$ */ + +/** +* 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.manifoldcf.core.lockmanager; + +import org.apache.manifoldcf.core.interfaces.*; +import org.apache.manifoldcf.core.system.ManifoldCF; +import org.apache.manifoldcf.core.system.Logging; +import java.io.*; + +/** One instance of this object exists for each lock on each JVM! +* This is the ZooKeeper version of the lock. +*/ +public class ZooKeeperLockObject extends LockObject +{ + public static final String _rcsid = "@(#)$Id$"; + + private final static String LOCK_PATH_PREFIX = "org.apache.manifoldcf.locks/"; + + private final ZooKeeperConnectionPool pool; + private final String lockPath; + + public ZooKeeperLockObject(LockPool lockPool, Object lockKey, ZooKeeperConnectionPool pool) + { + super(lockPool,lockKey); + this.pool = pool; + this.lockPath = LOCK_PATH_PREFIX + lockKey.toString(); + } + + @Override + protected void obtainGlobalWriteLock() + throws ManifoldCFException, LockException, InterruptedException + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.obtainGlobalWriteLock(lockPath); + } + finally + { + pool.release(connection); + } + } + + @Override + protected void clearGlobalWriteLock() + throws ManifoldCFException, LockException, InterruptedException + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.clearGlobalWriteLock(lockPath); + } + finally + { + pool.release(connection); + } + } + + @Override + protected void obtainGlobalNonExWriteLock() + throws ManifoldCFException, LockException, InterruptedException + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.obtainGlobalNonExWriteLock(lockPath); + } + finally + { + pool.release(connection); + } + } + + @Override + protected void clearGlobalNonExWriteLock() + throws ManifoldCFException, LockException, InterruptedException + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.clearGlobalNonExWriteLock(lockPath); + } + finally + { + pool.release(connection); + } + } + + @Override + protected void obtainGlobalReadLock() + throws ManifoldCFException, LockException, InterruptedException + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.obtainGlobalReadLock(lockPath); + } + finally + { + pool.release(connection); + } + } + + @Override + protected void clearGlobalReadLock() + throws ManifoldCFException, LockException, InterruptedException + { + ZooKeeperConnection connection = pool.grab(); + try + { + connection.clearGlobalReadLock(lockPath); + } + finally + { + pool.release(connection); + } + } + + +} + Propchange: manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObject.java ------------------------------------------------------------------------------ svn:eol-style = native Propchange: manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObject.java ------------------------------------------------------------------------------ svn:keywords = Id Added: manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObjectFactory.java URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObjectFactory.java?rev=1540056&view=auto ============================================================================== --- manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObjectFactory.java (added) +++ manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObjectFactory.java Fri Nov 8 14:44:19 2013 @@ -0,0 +1,45 @@ +/* $Id$ */ + +/** +* 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.manifoldcf.core.lockmanager; + +import org.apache.manifoldcf.core.interfaces.*; +import org.apache.manifoldcf.core.system.ManifoldCF; +import org.apache.manifoldcf.core.system.Logging; +import java.io.*; + +/** Base factory for zookeeper lock objects. +*/ +public class ZooKeeperLockObjectFactory extends LockObjectFactory +{ + public static final String _rcsid = "@(#)$Id$"; + + protected final ZooKeeperConnectionPool pool; + + public ZooKeeperLockObjectFactory(ZooKeeperConnectionPool pool) + { + this.pool = pool; + } + + @Override + public LockObject newLockObject(LockPool lockPool, Object lockKey) + { + return new ZooKeeperLockObject(lockPool, lockKey, pool); + } +} + Propchange: manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObjectFactory.java ------------------------------------------------------------------------------ svn:eol-style = native Propchange: manifoldcf/branches/CONNECTORS-13/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockObjectFactory.java ------------------------------------------------------------------------------ svn:keywords = Id
