Fix Astyanax locking to be re-entrant as Hector locks were and possible UG code expects it to be. Remove all reference of Hector locks, changing its existing test to work with Astyanax.
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/973f5a70 Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/973f5a70 Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/973f5a70 Branch: refs/heads/USERGRID-1047 Commit: 973f5a70c9cab0fdd5006684d3def8e18f962b51 Parents: d1c1d01 Author: Michael Russo <[email protected]> Authored: Thu Jan 14 15:30:52 2016 -0800 Committer: Michael Russo <[email protected]> Committed: Thu Jan 14 15:30:52 2016 -0800 ---------------------------------------------------------------------- .../locking/cassandra/AstyanaxLockImpl.java | 31 +++- .../cassandra/AstyanaxLockManagerImpl.java | 4 +- .../locking/cassandra/HectorLockImpl.java | 86 --------- .../cassandra/HectorLockManagerImpl.java | 142 -------------- .../usergrid/locking/AstyanaxLockManagerIT.java | 164 ++++++++++++++++ .../locking/SingleNodeLockTestSingleNode.java | 163 ++++++++++++++++ .../locking/cassandra/HectorLockManagerIT.java | 185 ------------------- .../SingleNodeLockTestSingleNode.java | 170 ----------------- .../test/resources/testApplicationContext.xml | 4 - 9 files changed, 352 insertions(+), 597 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockImpl.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockImpl.java b/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockImpl.java index 3c12d8a..1399d16 100644 --- a/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockImpl.java +++ b/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockImpl.java @@ -19,24 +19,24 @@ package org.apache.usergrid.locking.cassandra; import com.netflix.astyanax.recipes.locks.ColumnPrefixDistributedRowLock; +import com.netflix.astyanax.retry.RetryPolicy; +import com.netflix.astyanax.retry.RunOnce; import org.apache.usergrid.locking.Lock; import org.apache.usergrid.locking.exception.UGLockException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; public class AstyanaxLockImpl implements Lock { - private static final Logger logger = LoggerFactory.getLogger( AstyanaxLockImpl.class ); + private AtomicInteger count = new AtomicInteger(); + private ColumnPrefixDistributedRowLock lock; - - ColumnPrefixDistributedRowLock lock; - - public AstyanaxLockImpl(//AstyanaxLockManagerImpl lockManager, - ColumnPrefixDistributedRowLock lock) { + public AstyanaxLockImpl( ColumnPrefixDistributedRowLock lock ) { this.lock = lock; @@ -45,21 +45,27 @@ public class AstyanaxLockImpl implements Lock { @Override public boolean tryLock( long timeout, TimeUnit time ) throws UGLockException { - lock.withTtl( (int) timeout, time); try { + lock.acquire(); - return true; + count.incrementAndGet(); + } catch (Exception e) { return false; } + + return true; } @Override public void lock() throws UGLockException { try { + lock.acquire(); + count.incrementAndGet(); + } catch (Exception e) { throw new UGLockException("Unable to acquire lock with id: " + lock.getLockId()); } @@ -68,8 +74,15 @@ public class AstyanaxLockImpl implements Lock { @Override public void unlock() throws UGLockException { + // all re-entrant locks to be used and only release them all when the count is 0 + int current = count.decrementAndGet(); + try { - lock.release(); + + if ( current == 0 ) { + lock.release(); + } + } catch (Exception e) { throw new UGLockException("Unable to release lock with id: " + lock.getLockId()); } http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockManagerImpl.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockManagerImpl.java b/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockManagerImpl.java index ece942f..1786d72 100644 --- a/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockManagerImpl.java +++ b/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/AstyanaxLockManagerImpl.java @@ -38,6 +38,7 @@ import org.slf4j.LoggerFactory; import java.util.HashMap; import java.util.Map; import java.util.UUID; +import java.util.concurrent.TimeUnit; public class AstyanaxLockManagerImpl implements LockManager { @@ -72,7 +73,8 @@ public class AstyanaxLockManagerImpl implements LockManager { String lockPath = LockPathBuilder.buildPath( applicationId, path ); ColumnPrefixDistributedRowLock<String> lock = - new ColumnPrefixDistributedRowLock<>(keyspace, columnFamily, lockPath); + new ColumnPrefixDistributedRowLock<>(keyspace, columnFamily, lockPath) + .expireLockAfter( Integer.MAX_VALUE, TimeUnit.MILLISECONDS); return new AstyanaxLockImpl( lock ); http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/HectorLockImpl.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/HectorLockImpl.java b/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/HectorLockImpl.java deleted file mode 100644 index 57b9b5c..0000000 --- a/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/HectorLockImpl.java +++ /dev/null @@ -1,86 +0,0 @@ -/* - * 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.usergrid.locking.cassandra; - - -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -import org.apache.usergrid.locking.Lock; -import org.apache.usergrid.locking.exception.UGLockException; - -import me.prettyprint.hector.api.locking.HLock; -import me.prettyprint.hector.api.locking.HLockManager; -import me.prettyprint.hector.api.locking.HLockTimeoutException; - - -/** @author tnine */ -public class HectorLockImpl implements Lock { - - private HLock lock; - private HLockManager lm; - private AtomicInteger count = new AtomicInteger(); - - - /** - * - */ - public HectorLockImpl( HLock lock, HLockManager lm ) { - this.lock = lock; - this.lm = lm; - } - - - /* (non-Javadoc) - * @see org.apache.usergrid.locking.Lock#acquire(long, java.util.concurrent.TimeUnit) - */ - @Override - public boolean tryLock( long timeout, TimeUnit time ) throws UGLockException { - try { - lm.acquire( this.lock, time.toMillis( timeout ) ); - count.incrementAndGet(); - } - catch ( HLockTimeoutException hlte ) { - return false; - } - - return true; - } - - - /* (non-Javadoc) - * @see org.apache.usergrid.locking.Lock#lock() - */ - @Override - public void lock() throws UGLockException { - lm.acquire( lock ); - count.incrementAndGet(); - } - - - /* (non-Javadoc) - * @see org.apache.usergrid.locking.Lock#release() - */ - @Override - public void unlock() throws UGLockException { - int current = count.decrementAndGet(); - - if ( current == 0 ) { - lm.release( this.lock ); - } - } -} http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/HectorLockManagerImpl.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/HectorLockManagerImpl.java b/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/HectorLockManagerImpl.java deleted file mode 100644 index ae7878a..0000000 --- a/stack/core/src/main/java/org/apache/usergrid/locking/cassandra/HectorLockManagerImpl.java +++ /dev/null @@ -1,142 +0,0 @@ -/* - * 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.usergrid.locking.cassandra; - - -import java.util.UUID; - -import javax.annotation.PostConstruct; - -import me.prettyprint.cassandra.locking.HLockManagerImpl; -import me.prettyprint.hector.api.Cluster; -import me.prettyprint.hector.api.ConsistencyLevelPolicy; -import me.prettyprint.hector.api.locking.HLockManager; -import me.prettyprint.hector.api.locking.HLockManagerConfigurator; - -import org.springframework.util.Assert; -import org.apache.usergrid.locking.Lock; -import org.apache.usergrid.locking.LockManager; -import org.apache.usergrid.locking.LockPathBuilder; - - -/** - * Uses the hector based locking implementation to obtain locks - * - * @author tnine - */ -public class HectorLockManagerImpl implements LockManager { - private int replicationFactor = 1; - private int numberOfLockObserverThreads = 1; - private long lockTtl = 2000; - private String keyspaceName; - private Cluster cluster; - private HLockManager lm; - private ConsistencyLevelPolicy consistencyLevelPolicy; - - - /** - * - */ - public HectorLockManagerImpl() { - } - - - @PostConstruct - public void init() { - HLockManagerConfigurator hlc = new HLockManagerConfigurator(); - hlc.setReplicationFactor( replicationFactor ); - hlc.setKeyspaceName( keyspaceName ); - hlc.setNumberOfLockObserverThreads( numberOfLockObserverThreads ); - hlc.setLocksTTLInMillis( lockTtl ); - lm = new HLockManagerImpl( cluster, hlc ); - if ( consistencyLevelPolicy != null ) { - lm.getKeyspace().setConsistencyLevelPolicy(consistencyLevelPolicy); - } - // if consistencyLevelPolicy == null, use hector's default, which is QuorumAll, no need to explicitly set - lm.init(); - } - - - /* - * (non-Javadoc) - * - * @see org.apache.usergrid.locking.LockManager#createLock(java.util.UUID, - * java.lang.String[]) - */ - @Override - public Lock createLock( UUID applicationId, String... path ) { - - String lockPath = LockPathBuilder.buildPath( applicationId, path ); - - return new HectorLockImpl( lm.createLock( lockPath ), lm ); - } - - - /** - * Note that in a real environment this MUST be an odd number. Locks are read and written at QUORUM. RF >= 3 is - * preferred for failure tolerance and replication. Defaults to 1 - * - * @param replicationFactor the replicationFactor to set - */ - public void setReplicationFactor( int replicationFactor ) { - - Assert.isTrue( numberOfLockObserverThreads % 2 != 0, "You must specify an odd number for replication factor" ); - - this.replicationFactor = replicationFactor; - } - - - /** - * Set the number of threads the lock heartbeat executor uses. Must accommodate the total number of locks that may - * exist in the system. Locks are always renewed at the ttl/2 time. - * - * @param numberOfLockObserverThreads the numberOfLockObserverThreads to set - */ - public void setNumberOfLockObserverThreads( int numberOfLockObserverThreads ) { - this.numberOfLockObserverThreads = numberOfLockObserverThreads; - } - - - /** - * The amount of time a lock must not be renewed before it times out. Set in milliseconds. 2000 is the default - * - * @param lockTtl the lockTtl to set - */ - public void setLockTtl( long lockTtl ) { - this.lockTtl = lockTtl; - } - - - /** @param keyspaceName the keyspaceName to set */ - public void setKeyspaceName( String keyspaceName ) { - this.keyspaceName = keyspaceName; - } - - - /** @param cluster the cluster to set */ - public void setCluster( Cluster cluster ) { - this.cluster = cluster; - } - - - /** - * @param consistencyLevelPolicy the consistencyLevelPolicy to set - */ - public void setConsistencyLevelPolicy(ConsistencyLevelPolicy consistencyLevelPolicy) { - this.consistencyLevelPolicy = consistencyLevelPolicy; - } -} http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/core/src/test/java/org/apache/usergrid/locking/AstyanaxLockManagerIT.java ---------------------------------------------------------------------- diff --git a/stack/core/src/test/java/org/apache/usergrid/locking/AstyanaxLockManagerIT.java b/stack/core/src/test/java/org/apache/usergrid/locking/AstyanaxLockManagerIT.java new file mode 100644 index 0000000..e95c0d5 --- /dev/null +++ b/stack/core/src/test/java/org/apache/usergrid/locking/AstyanaxLockManagerIT.java @@ -0,0 +1,164 @@ +/* + * 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.usergrid.locking; + + +import org.apache.usergrid.AbstractCoreIT; +import org.apache.usergrid.locking.exception.UGLockException; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.UUID; +import java.util.concurrent.*; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + + +public class AstyanaxLockManagerIT extends AbstractCoreIT { + private static final Logger logger = LoggerFactory.getLogger( AstyanaxLockManagerIT.class ); + + + private static LockManager lockManager; + private static ExecutorService pool; + + + @BeforeClass + public static void setup() throws Exception { + + lockManager = setup.getInjector().getInstance(LockManager.class); + } + + + + @BeforeClass + public static void start() { + // Create a different thread to lock the same node, that is held by the main thread. + pool = Executors.newFixedThreadPool( 1 ); + } + + + @AfterClass + public static void tearDown() throws Exception { + pool.shutdownNow(); + } + + + /** Locks a path and launches a thread which also locks the same path. */ + @Test + public void testLock() throws InterruptedException, ExecutionException, UGLockException { + final UUID application = UUID.randomUUID(); + final UUID entity = UUID.randomUUID(); + + logger.info( "Locking:" + application.toString() + "/" + entity.toString() ); + + // Lock a node twice to test re-entrancy and validate. + Lock lock = lockManager.createLock( application, entity.toString() ); + lock.lock(); + lock.lock(); + + boolean wasLocked = lockInDifferentThread( application, entity ); + Assert.assertEquals( false, wasLocked ); + + // Unlock once + lock.unlock(); + + // Try from the thread expecting to fail since we still hold one re-entrant lock. + wasLocked = lockInDifferentThread( application, entity ); + assertFalse( wasLocked ); + + // Unlock completely + logger.info( "Releasing lock:" + application.toString() + "/" + entity.toString() ); + lock.unlock(); + + // Try to effectively get the lock from the thread since the current one has + // already released it. + wasLocked = lockInDifferentThread( application, entity ); + Assert.assertEquals( true, wasLocked ); + } + + + /** Locks a couple of times and try to clean up. Later oin another thread successfully acquire the lock */ + @Test + public void testLock2() throws InterruptedException, ExecutionException, UGLockException { + final UUID application = UUID.randomUUID(); + final UUID entity = UUID.randomUUID(); + final UUID entity2 = UUID.randomUUID(); + + logger.info( "Locking:" + application.toString() + "/" + entity.toString() ); + + // Acquire to locks. One of them twice. + Lock lock = lockManager.createLock( application, entity.toString() ); + lock.lock(); + lock.lock(); + + Lock second = lockManager.createLock( application, entity2.toString() ); + second.lock(); + + // Cleanup the locks for main thread + logger.info( "Cleaning up locks for current thread..." ); + lock.unlock(); + lock.unlock(); + + second.unlock(); + + boolean locked = lockInDifferentThread( application, entity ); + assertTrue( locked ); + + locked = lockInDifferentThread( application, entity2 ); + assertTrue( locked ); + } + + + /** Acquires a lock in a different thread. */ + private boolean lockInDifferentThread( final UUID application, final UUID entity ) { + Callable<Boolean> callable = new Callable<Boolean>() { + @Override + public Boolean call() throws Exception { + Lock lock = lockManager.createLock( application, entity.toString() ); + + // False here means that the lock WAS NOT ACQUIRED. And that is + // what we expect. + boolean locked = lock.tryLock( 0, TimeUnit.MILLISECONDS ); + + // shouldn't lock, so unlock to avoid polluting future tests + if ( locked ) { + lock.unlock(); + } + + return locked; + } + }; + + Future<Boolean> status = pool.submit( callable ); + + boolean wasLocked = true; + + try { + wasLocked = status.get( 2, TimeUnit.SECONDS ); + } + catch ( Exception e ) { + wasLocked = false; + } + + return wasLocked; + } +} http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/core/src/test/java/org/apache/usergrid/locking/SingleNodeLockTestSingleNode.java ---------------------------------------------------------------------- diff --git a/stack/core/src/test/java/org/apache/usergrid/locking/SingleNodeLockTestSingleNode.java b/stack/core/src/test/java/org/apache/usergrid/locking/SingleNodeLockTestSingleNode.java new file mode 100644 index 0000000..ba7b48d --- /dev/null +++ b/stack/core/src/test/java/org/apache/usergrid/locking/SingleNodeLockTestSingleNode.java @@ -0,0 +1,163 @@ +/* + * 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.usergrid.locking; + + +import org.apache.usergrid.locking.exception.UGLockException; +import org.apache.usergrid.locking.singlenode.SingleNodeLockManagerImpl; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.UUID; +import java.util.concurrent.*; + +import static org.junit.Assert.assertTrue; + + +public class SingleNodeLockTestSingleNode { + + private static final Logger logger = LoggerFactory.getLogger( SingleNodeLockTestSingleNode.class ); + + private LockManager manager; + + private ExecutorService pool; + + + @Before + public void setUp() throws Exception { + + manager = new SingleNodeLockManagerImpl(); + + // Create a different thread to lock the same node, that is held by the main + // thread. + pool = Executors.newFixedThreadPool( 1 ); + } + + + @After + public void tearDown() throws Exception { + pool.shutdownNow(); + } + + + /** Locks a path and launches a thread which also locks the same path. */ + @Test + public void testLock() throws InterruptedException, ExecutionException, UGLockException { + + final UUID application = UUID.randomUUID(); + final UUID entity = UUID.randomUUID(); + + logger.info( "Locking:" + application.toString() + "/" + entity.toString() ); + + // Lock a node twice to test reentrancy and validate. + Lock lock = manager.createLock( application, entity.toString() ); + lock.lock(); + lock.lock(); + + boolean wasLocked = lockInDifferentThread( application, entity ); + Assert.assertEquals( false, wasLocked ); + + // Unlock once + lock.unlock(); + + // Try from the thread expecting to fail since we still hold one reentrant + // lock. + wasLocked = lockInDifferentThread( application, entity ); + Assert.assertEquals( false, wasLocked ); + + // Unlock completely + logger.info( "Releasing lock:" + application.toString() + "/" + entity.toString() ); + lock.unlock(); + + // Try to effectively get the lock from the thread since the current one has + // already released it. + wasLocked = lockInDifferentThread( application, entity ); + Assert.assertEquals( true, wasLocked ); + } + + + /** Locks a couple of times and try to clean up. Later oin another thread successfully acquire the lock */ + @Test + public void testLock2() throws InterruptedException, ExecutionException, UGLockException { + + final UUID application = UUID.randomUUID(); + final UUID entity = UUID.randomUUID(); + final UUID entity2 = UUID.randomUUID(); + + logger.info( "Locking:" + application.toString() + "/" + entity.toString() ); + + // Acquire to locks. One of them twice. + Lock lock = manager.createLock( application, entity.toString() ); + lock.lock(); + lock.lock(); + + Lock second = manager.createLock( application, entity2.toString() ); + second.lock(); + + // Cleanup the locks for main thread + logger.info( "Cleaning up locks for current thread..." ); + lock.unlock(); + lock.unlock(); + + second.unlock(); + + boolean locked = lockInDifferentThread( application, entity ); + assertTrue( locked ); + + locked = lockInDifferentThread( application, entity2 ); + assertTrue( locked ); + } + + + /** Acquires a lock in a different thread. */ + private boolean lockInDifferentThread( final UUID application, final UUID entity ) { + Future<Boolean> status = pool.submit( new Callable<Boolean>() { + + @Override + public Boolean call() throws Exception { + + Lock lock = manager.createLock( application, entity.toString() ); + + // False here means that the lock WAS NOT ACQUIRED. And that is + // what we expect. + + boolean locked = lock.tryLock( 0, TimeUnit.MILLISECONDS ); + + // shouldn't lock, so unlock to avoid polluting future tests + if ( locked ) { + lock.unlock(); + } + + return locked; + } + } ); + + boolean wasLocked = true; + try { + wasLocked = status.get( 2, TimeUnit.SECONDS ); + } + catch ( Exception e ) { + wasLocked = false; + } + + return wasLocked; + } +} http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/core/src/test/java/org/apache/usergrid/locking/cassandra/HectorLockManagerIT.java ---------------------------------------------------------------------- diff --git a/stack/core/src/test/java/org/apache/usergrid/locking/cassandra/HectorLockManagerIT.java b/stack/core/src/test/java/org/apache/usergrid/locking/cassandra/HectorLockManagerIT.java deleted file mode 100644 index 34b2154..0000000 --- a/stack/core/src/test/java/org/apache/usergrid/locking/cassandra/HectorLockManagerIT.java +++ /dev/null @@ -1,185 +0,0 @@ -/* - * 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.usergrid.locking.cassandra; - - -import java.util.UUID; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; - -import org.junit.AfterClass; -import org.junit.Assert; -import org.junit.BeforeClass; -import org.junit.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.usergrid.AbstractCoreIT; -import org.apache.usergrid.locking.Lock; -import org.apache.usergrid.locking.LockManager; -import org.apache.usergrid.locking.exception.UGLockException; - -import me.prettyprint.cassandra.model.ConfigurableConsistencyLevel; -import me.prettyprint.hector.api.ConsistencyLevelPolicy; -import me.prettyprint.hector.api.HConsistencyLevel; - -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; - - -public class HectorLockManagerIT extends AbstractCoreIT { - private static final Logger logger = LoggerFactory.getLogger( HectorLockManagerIT.class ); - - private static LockManager manager; - private static ExecutorService pool; - - - @BeforeClass - public static void setup() throws Exception { - HectorLockManagerImpl hlockManager = new HectorLockManagerImpl(); - hlockManager.setCluster( setup.getCassSvc().getCluster() ); - hlockManager.setKeyspaceName( "Locks_Test" ); - hlockManager.setLockTtl( 2000 ); - hlockManager.setNumberOfLockObserverThreads( 1 ); - hlockManager.setReplicationFactor( 1 ); - ConsistencyLevelPolicy consistencyLevel = new ConfigurableConsistencyLevel(); - ((ConfigurableConsistencyLevel) consistencyLevel).setDefaultReadConsistencyLevel(HConsistencyLevel.ONE); - ((ConfigurableConsistencyLevel) consistencyLevel).setDefaultWriteConsistencyLevel(HConsistencyLevel.ONE); - hlockManager.setConsistencyLevelPolicy(consistencyLevel); - hlockManager.init(); - - manager = hlockManager; - } - - - @BeforeClass - public static void start() { - // Create a different thread to lock the same node, that is held by the main thread. - pool = Executors.newFixedThreadPool( 1 ); - } - - - @AfterClass - public static void tearDown() throws Exception { - pool.shutdownNow(); - } - - - /** Locks a path and launches a thread which also locks the same path. */ - @Test - public void testLock() throws InterruptedException, ExecutionException, UGLockException { - final UUID application = UUID.randomUUID(); - final UUID entity = UUID.randomUUID(); - - logger.info( "Locking:" + application.toString() + "/" + entity.toString() ); - - // Lock a node twice to test re-entrancy and validate. - Lock lock = manager.createLock( application, entity.toString() ); - lock.lock(); - lock.lock(); - - boolean wasLocked = lockInDifferentThread( application, entity ); - Assert.assertEquals( false, wasLocked ); - - // Unlock once - lock.unlock(); - - // Try from the thread expecting to fail since we still hold one re-entrant lock. - wasLocked = lockInDifferentThread( application, entity ); - assertFalse( wasLocked ); - - // Unlock completely - logger.info( "Releasing lock:" + application.toString() + "/" + entity.toString() ); - lock.unlock(); - - // Try to effectively get the lock from the thread since the current one has - // already released it. - wasLocked = lockInDifferentThread( application, entity ); - Assert.assertEquals( true, wasLocked ); - } - - - /** Locks a couple of times and try to clean up. Later oin another thread successfully acquire the lock */ - @Test - public void testLock2() throws InterruptedException, ExecutionException, UGLockException { - final UUID application = UUID.randomUUID(); - final UUID entity = UUID.randomUUID(); - final UUID entity2 = UUID.randomUUID(); - - logger.info( "Locking:" + application.toString() + "/" + entity.toString() ); - - // Acquire to locks. One of them twice. - Lock lock = manager.createLock( application, entity.toString() ); - lock.lock(); - lock.lock(); - - Lock second = manager.createLock( application, entity2.toString() ); - second.lock(); - - // Cleanup the locks for main thread - logger.info( "Cleaning up locks for current thread..." ); - lock.unlock(); - lock.unlock(); - - second.unlock(); - - boolean locked = lockInDifferentThread( application, entity ); - assertTrue( locked ); - - locked = lockInDifferentThread( application, entity2 ); - assertTrue( locked ); - } - - - /** Acquires a lock in a different thread. */ - private boolean lockInDifferentThread( final UUID application, final UUID entity ) { - Callable<Boolean> callable = new Callable<Boolean>() { - @Override - public Boolean call() throws Exception { - Lock lock = manager.createLock( application, entity.toString() ); - - // False here means that the lock WAS NOT ACQUIRED. And that is - // what we expect. - boolean locked = lock.tryLock( 0, TimeUnit.MILLISECONDS ); - - // shouldn't lock, so unlock to avoid polluting future tests - if ( locked ) { - lock.unlock(); - } - - return locked; - } - }; - - Future<Boolean> status = pool.submit( callable ); - - boolean wasLocked = true; - - try { - wasLocked = status.get( 2, TimeUnit.SECONDS ); - } - catch ( Exception e ) { - wasLocked = false; - } - - return wasLocked; - } -} http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/core/src/test/java/org/apache/usergrid/locking/singlenode/SingleNodeLockTestSingleNode.java ---------------------------------------------------------------------- diff --git a/stack/core/src/test/java/org/apache/usergrid/locking/singlenode/SingleNodeLockTestSingleNode.java b/stack/core/src/test/java/org/apache/usergrid/locking/singlenode/SingleNodeLockTestSingleNode.java deleted file mode 100644 index db1defd..0000000 --- a/stack/core/src/test/java/org/apache/usergrid/locking/singlenode/SingleNodeLockTestSingleNode.java +++ /dev/null @@ -1,170 +0,0 @@ -/* - * 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.usergrid.locking.singlenode; - - -import java.util.UUID; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; - -import org.junit.After; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.usergrid.locking.Lock; -import org.apache.usergrid.locking.LockManager; -import org.apache.usergrid.locking.exception.UGLockException; - -import static org.junit.Assert.assertTrue; - - -public class SingleNodeLockTestSingleNode { - - private static final Logger logger = LoggerFactory.getLogger( SingleNodeLockTestSingleNode.class ); - - private LockManager manager; - - private ExecutorService pool; - - - @Before - public void setUp() throws Exception { - - manager = new SingleNodeLockManagerImpl(); - - // Create a different thread to lock the same node, that is held by the main - // thread. - pool = Executors.newFixedThreadPool( 1 ); - } - - - @After - public void tearDown() throws Exception { - pool.shutdownNow(); - } - - - /** Locks a path and launches a thread which also locks the same path. */ - @Test - public void testLock() throws InterruptedException, ExecutionException, UGLockException { - - final UUID application = UUID.randomUUID(); - final UUID entity = UUID.randomUUID(); - - logger.info( "Locking:" + application.toString() + "/" + entity.toString() ); - - // Lock a node twice to test reentrancy and validate. - Lock lock = manager.createLock( application, entity.toString() ); - lock.lock(); - lock.lock(); - - boolean wasLocked = lockInDifferentThread( application, entity ); - Assert.assertEquals( false, wasLocked ); - - // Unlock once - lock.unlock(); - - // Try from the thread expecting to fail since we still hold one reentrant - // lock. - wasLocked = lockInDifferentThread( application, entity ); - Assert.assertEquals( false, wasLocked ); - - // Unlock completely - logger.info( "Releasing lock:" + application.toString() + "/" + entity.toString() ); - lock.unlock(); - - // Try to effectively get the lock from the thread since the current one has - // already released it. - wasLocked = lockInDifferentThread( application, entity ); - Assert.assertEquals( true, wasLocked ); - } - - - /** Locks a couple of times and try to clean up. Later oin another thread successfully acquire the lock */ - @Test - public void testLock2() throws InterruptedException, ExecutionException, UGLockException { - - final UUID application = UUID.randomUUID(); - final UUID entity = UUID.randomUUID(); - final UUID entity2 = UUID.randomUUID(); - - logger.info( "Locking:" + application.toString() + "/" + entity.toString() ); - - // Acquire to locks. One of them twice. - Lock lock = manager.createLock( application, entity.toString() ); - lock.lock(); - lock.lock(); - - Lock second = manager.createLock( application, entity2.toString() ); - second.lock(); - - // Cleanup the locks for main thread - logger.info( "Cleaning up locks for current thread..." ); - lock.unlock(); - lock.unlock(); - - second.unlock(); - - boolean locked = lockInDifferentThread( application, entity ); - assertTrue( locked ); - - locked = lockInDifferentThread( application, entity2 ); - assertTrue( locked ); - } - - - /** Acquires a lock in a different thread. */ - private boolean lockInDifferentThread( final UUID application, final UUID entity ) { - Future<Boolean> status = pool.submit( new Callable<Boolean>() { - - @Override - public Boolean call() throws Exception { - - Lock lock = manager.createLock( application, entity.toString() ); - - // False here means that the lock WAS NOT ACQUIRED. And that is - // what we expect. - - boolean locked = lock.tryLock( 0, TimeUnit.MILLISECONDS ); - - // shouldn't lock, so unlock to avoid polluting future tests - if ( locked ) { - lock.unlock(); - } - - return locked; - } - } ); - - boolean wasLocked = true; - try { - wasLocked = status.get( 2, TimeUnit.SECONDS ); - } - catch ( Exception e ) { - wasLocked = false; - } - - return wasLocked; - } -} http://git-wip-us.apache.org/repos/asf/usergrid/blob/973f5a70/stack/websocket/src/test/resources/testApplicationContext.xml ---------------------------------------------------------------------- diff --git a/stack/websocket/src/test/resources/testApplicationContext.xml b/stack/websocket/src/test/resources/testApplicationContext.xml index 8363021..d92f4fb 100644 --- a/stack/websocket/src/test/resources/testApplicationContext.xml +++ b/stack/websocket/src/test/resources/testApplicationContext.xml @@ -102,10 +102,6 @@ <constructor-arg ref="cassandraCredentials" /> </bean> - <!-- - <bean name="lockManager" class="org.apache.usergrid.locking.cassandra.HectorLockManagerImpl" /> - --> - <bean id="injector" class="org.apache.usergrid.corepersistence.GuiceFactory"> </bean>
