InterProcessSemaphore { public Lease acquire() throws Exception { String path = internals.attemptLock(-1, null, null); return makeLease(path); } InterProcessSemaphore(CuratorFramework client, String path, int maxLeases);  InterProcessSemaphore(CuratorFramework client, String path, SharedCountReader count); private  InterProcessSemaphore(CuratorFramework client, String path, int maxLeases, SharedCountReader count); void returnAll(Collection<Lease> leases); void returnLease(Lease lease); Lease acquire(); Collection<Lease> acquire(int qty); Lease acquire(long time, TimeUnit unit); Collection<Lease> acquire(int qty, long time, TimeUnit unit);  }
@Test public void testThreadedLeaseIncrease() throws Exception { final Timing timing = new Timing(); CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); try { client.start(); final SharedCount count = new SharedCount(client, "/foo/count", 1); count.start(); final InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", count); ExecutorService service = Executors.newCachedThreadPool(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); Future<Object> future1 = service.submit ( new Callable<Object>() { @Override public Object call() throws Exception { Lease lease = semaphore.acquire(timing.seconds(), TimeUnit.SECONDS); Assert.assertNotNull(lease); latch1.countDown(); lease = semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS); Assert.assertNotNull(lease); latch2.countDown(); return null; } } ); Future<Object> future2 = service.submit ( new Callable<Object>() { @Override public Object call() throws Exception { Assert.assertTrue(latch1.await(timing.forWaiting().seconds(), TimeUnit.SECONDS)); timing.sleepABit(); Assert.assertTrue(count.trySetCount(2)); timing.sleepABit(); Assert.assertTrue(latch2.await(0, TimeUnit.SECONDS)); return null; } } ); future1.get(); future2.get(); count.close(); } finally { TestCleanState.closeAndTestClean(client); } }
@Test public void testClientClose() throws Exception { final Timing timing = new Timing(); CuratorFramework client1 = null; CuratorFramework client2 = null; InterProcessSemaphoreV2 semaphore1; InterProcessSemaphoreV2 semaphore2; try { client1 = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); client2 = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); client1.start(); client2.start(); semaphore1 = new InterProcessSemaphoreV2(client1, "/test", 1); semaphore2 = new InterProcessSemaphoreV2(client2, "/test", 1); Lease lease = semaphore2.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS); Assert.assertNotNull(lease); lease.close(); lease = semaphore1.acquire(10, TimeUnit.SECONDS); Assert.assertNotNull(lease); client1.close(); client1 = null; Assert.assertNotNull(semaphore2.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS)); } finally { TestCleanState.closeAndTestClean(client1); TestCleanState.closeAndTestClean(client2); } }
@Test public void testRelease1AtATime() throws Exception { final Timing timing = new Timing(); final int CLIENT_QTY = 10; final int MAX = CLIENT_QTY / 2; final AtomicInteger maxLeases = new AtomicInteger(0); final AtomicInteger activeQty = new AtomicInteger(0); final AtomicInteger uses = new AtomicInteger(0); List<Future<Object>> futures = Lists.newArrayList(); ExecutorService service = Executors.newFixedThreadPool(CLIENT_QTY); for ( int i = 0; i < CLIENT_QTY; ++i ) { Future<Object> f = service.submit ( new Callable<Object>() { @Override public Object call() throws Exception { CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); client.start(); try { InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", MAX); Lease lease = semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS); Assert.assertNotNull(lease); uses.incrementAndGet(); try { synchronized(maxLeases) { int qty = activeQty.incrementAndGet(); if ( qty > maxLeases.get() ) { maxLeases.set(qty); } } timing.sleepABit(); } finally { activeQty.decrementAndGet(); lease.close(); } } finally { TestCleanState.closeAndTestClean(client); } return null; } } ); futures.add(f); } for ( Future<Object> f : futures ) { f.get(); } Assert.assertEquals(uses.get(), CLIENT_QTY); Assert.assertEquals(maxLeases.get(), MAX); }
@Test public void testThreads() throws Exception { final int THREAD_QTY = 10; Timing timing = new Timing(); CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); client.start(); try { final InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", 1); ExecutorService service = Executors.newFixedThreadPool(THREAD_QTY); for ( int i = 0; i < THREAD_QTY; ++i ) { service.submit ( new Callable<Object>() { @Override public Object call() throws Exception { Lease lease = semaphore.acquire(); try { Thread.sleep(1); } finally { lease.close(); } return null; } } ); } service.shutdown(); Assert.assertTrue(service.awaitTermination(10, TimeUnit.SECONDS)); } finally { TestCleanState.closeAndTestClean(client); } }
@Test public void testSimple() throws Exception { Timing timing = new Timing(); CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); client.start(); try { InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", 1); Assert.assertNotNull(semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS)); Assert.assertNull(semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS)); } finally { TestCleanState.closeAndTestClean(client); } }
@Test public void testSimple2() throws Exception { final int MAX_LEASES = 3; Timing timing = new Timing(); List<Lease> leases = Lists.newArrayList(); CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); client.start(); try { for ( int i = 0; i < MAX_LEASES; ++i ) { InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", MAX_LEASES); Lease lease = semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS); Assert.assertNotNull(lease); leases.add(lease); } InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", MAX_LEASES); Lease lease = semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS); Assert.assertNull(lease); leases.remove(0).close(); Assert.assertNotNull(semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS)); } finally { for ( Lease l : leases ) { CloseableUtils.closeQuietly(l); } TestCleanState.closeAndTestClean(client); } }
@Test public void testGetParticipantNodes() throws Exception { final int LEASES = 3; Timing timing = new Timing(); CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); List<Lease> leases = Lists.newArrayList(); client.start(); try { InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", LEASES); for ( int i = 0; i < LEASES; ++i ) { leases.add(semaphore.acquire()); } Assert.assertEquals(semaphore.getParticipantNodes().size(), LEASES); } finally { for ( Lease l : leases ) { CloseableUtils.closeQuietly(l); } TestCleanState.closeAndTestClean(client); } }
@Test public void testNoOrphanedNodes() throws Exception { final Timing timing = new Timing(); final ExecutorService executor = Executors.newFixedThreadPool(1); CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); client.start(); try { final InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", 1); Lease lease = semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS); Assert.assertNotNull(lease); final List<String> childNodes = client.getChildren().forPath("/test/leases"); Assert.assertEquals(childNodes.size(), 1); final CountDownLatch nodeCreatedLatch = new CountDownLatch(1); client.getChildren().usingWatcher(new CuratorWatcher() { @Override public void process(WatchedEvent event) throws Exception { if ( event.getType() == Watcher.Event.EventType.NodeCreated ) { nodeCreatedLatch.countDown(); } } }).forPath("/test/leases"); final Future<Lease> leaseFuture = executor.submit(new Callable<Lease>() { @Override public Lease call() throws Exception { return semaphore.acquire(timing.forWaiting().multiple(2).seconds(), TimeUnit.SECONDS); } }); timing.awaitLatch(nodeCreatedLatch); String newNode = null; for ( String c : client.getChildren().forPath("/test/leases") ) { if ( !childNodes.contains(c) ) { newNode = c; } } Assert.assertNotNull(newNode); client.delete().forPath("/test/leases/" + newNode); lease.close(); lease = leaseFuture.get(); Assert.assertNotNull(lease); lease.close(); Assert.assertEquals(client.getChildren().forPath("/test/leases").size(), 0); Assert.assertNotNull(semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS)); } finally { executor.shutdownNow(); TestCleanState.closeAndTestClean(client); } }