| /* |
| * 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.geode.cache.client.internal.pooling; |
| |
| import static java.util.concurrent.TimeUnit.NANOSECONDS; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.logging.internal.spi.LogWriterLevel.FINEST; |
| import static org.apache.geode.test.awaitility.GeodeAwaitility.await; |
| import static org.assertj.core.api.Assertions.assertThat; |
| import static org.junit.Assert.fail; |
| |
| import java.io.InputStream; |
| import java.io.OutputStream; |
| import java.net.Socket; |
| import java.nio.ByteBuffer; |
| import java.util.Collections; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.function.BooleanSupplier; |
| |
| import org.junit.After; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.CancelCriterion; |
| import org.apache.geode.cache.CacheClosedException; |
| import org.apache.geode.cache.client.AllConnectionsInUseException; |
| import org.apache.geode.cache.client.NoAvailableServersException; |
| import org.apache.geode.cache.client.internal.ClientUpdater; |
| import org.apache.geode.cache.client.internal.Connection; |
| import org.apache.geode.cache.client.internal.ConnectionFactory; |
| import org.apache.geode.cache.client.internal.ConnectionStats; |
| import org.apache.geode.cache.client.internal.Endpoint; |
| import org.apache.geode.cache.client.internal.EndpointManager; |
| import org.apache.geode.cache.client.internal.EndpointManagerImpl; |
| import org.apache.geode.cache.client.internal.Op; |
| import org.apache.geode.cache.client.internal.QueueManager; |
| import org.apache.geode.cache.client.internal.ServerDenyList; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.DistributedSystem; |
| import org.apache.geode.distributed.internal.ServerLocation; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.cache.PoolStats; |
| import org.apache.geode.internal.cache.tier.sockets.ServerQueueStatus; |
| import org.apache.geode.internal.logging.InternalLogWriter; |
| import org.apache.geode.internal.logging.LocalLogWriter; |
| import org.apache.geode.test.awaitility.GeodeAwaitility; |
| import org.apache.geode.test.dunit.ThreadUtils; |
| import org.apache.geode.test.dunit.WaitCriterion; |
| import org.apache.geode.test.junit.categories.ClientServerTest; |
| |
| @Category({ClientServerTest.class}) |
| public class ConnectionManagerJUnitTest { |
| |
| private static final long TIMEOUT_MILLIS = 30 * 1000; |
| // Some machines do not have a monotonic clock. |
| private static final long ALLOWABLE_ERROR_IN_MILLIS = 100; |
| ConnectionManager manager; |
| private InternalLogWriter logger; |
| protected DummyFactory factory; |
| private DistributedSystem ds; |
| private ScheduledExecutorService background; |
| protected EndpointManager endpointManager; |
| private CancelCriterion cancelCriterion; |
| private PoolStats poolStats; |
| |
| @Before |
| public void setUp() { |
| this.logger = new LocalLogWriter(FINEST.intLevel(), System.out); |
| factory = new DummyFactory(); |
| |
| Properties properties = new Properties(); |
| properties.put(MCAST_PORT, "0"); |
| properties.put(LOCATORS, ""); |
| ds = DistributedSystem.connect(properties); |
| background = Executors.newSingleThreadScheduledExecutor(); |
| poolStats = new PoolStats(ds, "connectionManagerJUnitTest"); |
| endpointManager = new EndpointManagerImpl("pool", ds, ds.getCancelCriterion(), poolStats); |
| cancelCriterion = new CancelCriterion() { |
| |
| @Override |
| public String cancelInProgress() { |
| return null; |
| } |
| |
| @Override |
| public RuntimeException generateCancelledException(Throwable e) { |
| return null; |
| } |
| }; |
| } |
| |
| @After |
| public void tearDown() throws InterruptedException { |
| ds.disconnect(); |
| if (manager != null) { |
| manager.close(false); |
| } |
| background.shutdownNow(); |
| } |
| |
| @Test |
| public void testAddVarianceToInterval() { |
| assertThat(ConnectionManagerImpl.addVarianceToInterval(0)).as("Zero gets zero variance") |
| .isEqualTo(0); |
| assertThat(ConnectionManagerImpl.addVarianceToInterval(300000)) |
| .as("Large value gets +/-10% variance").isNotEqualTo(300000).isGreaterThanOrEqualTo(270000) |
| .isLessThanOrEqualTo(330000); |
| assertThat(ConnectionManagerImpl.addVarianceToInterval(9)).as("Small value gets +/-1 variance") |
| .isNotEqualTo(9).isGreaterThanOrEqualTo(8).isLessThanOrEqualTo(10); |
| } |
| |
| @Test |
| public void testGet() |
| throws InterruptedException, AllConnectionsInUseException, NoAvailableServersException { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 3, 0, -1, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| Connection conn[] = new Connection[4]; |
| |
| conn[0] = manager.borrowConnection(0); |
| Assert.assertEquals(1, factory.creates); |
| |
| manager.returnConnection(conn[0]); |
| conn[0] = manager.borrowConnection(0); |
| Assert.assertEquals(1, factory.creates); |
| conn[1] = manager.borrowConnection(0); |
| manager.returnConnection(conn[0]); |
| manager.returnConnection(conn[1]); |
| Assert.assertEquals(2, factory.creates); |
| |
| conn[0] = manager.borrowConnection(0); |
| conn[1] = manager.borrowConnection(0); |
| conn[2] = manager.borrowConnection(0); |
| Assert.assertEquals(3, factory.creates); |
| |
| try { |
| conn[4] = manager.borrowConnection(10); |
| fail("Should have received an all connections in use exception"); |
| } catch (AllConnectionsInUseException e) { |
| // expected exception |
| } |
| } |
| |
| @Test |
| public void testPrefill() throws InterruptedException { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 10, 2, -1, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| final String descrip = manager.toString(); |
| WaitCriterion ev = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return factory.creates == 2 && factory.destroys == 0; |
| } |
| |
| @Override |
| public String description() { |
| return "waiting for manager " + descrip; |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(ev); |
| } |
| |
| @Test |
| public void testInvalidateConnection() |
| throws InterruptedException, AllConnectionsInUseException, NoAvailableServersException { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 10, 0, 0L, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| Connection conn = manager.borrowConnection(0); |
| Assert.assertEquals(1, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| conn.destroy(); |
| manager.returnConnection(conn); |
| Assert.assertEquals(1, factory.creates); |
| Assert.assertEquals(1, factory.destroys); |
| conn = manager.borrowConnection(0); |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(1, factory.destroys); |
| } |
| |
| @Test |
| public void testInvalidateServer() |
| throws InterruptedException, AllConnectionsInUseException, NoAvailableServersException { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 10, 0, -1, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| ServerLocation server1 = new ServerLocation("localhost", 1); |
| ServerLocation server2 = new ServerLocation("localhost", 2); |
| factory.nextServer = server1; |
| Connection conn1 = manager.borrowConnection(0); |
| Connection conn2 = manager.borrowConnection(0); |
| Connection conn3 = manager.borrowConnection(0); |
| factory.nextServer = server2; |
| Connection conn4 = manager.borrowConnection(0); |
| |
| Assert.assertEquals(4, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| |
| manager.returnConnection(conn2); |
| endpointManager.serverCrashed(conn2.getEndpoint()); |
| Assert.assertEquals(3, factory.destroys); |
| conn1.destroy(); |
| manager.returnConnection(conn1); |
| Assert.assertEquals(3, factory.destroys); |
| manager.returnConnection(conn3); |
| manager.returnConnection(conn4); |
| Assert.assertEquals(3, factory.destroys); |
| |
| manager.borrowConnection(0); |
| Assert.assertEquals(4, factory.creates); |
| Assert.assertEquals(3, factory.destroys); |
| } |
| |
| @Test |
| public void testIdleExpiration() |
| throws InterruptedException, AllConnectionsInUseException, NoAvailableServersException { |
| final long idleTimeoutMillis = 300; |
| manager = |
| new ConnectionManagerImpl("pool", factory, endpointManager, 5, 2, idleTimeoutMillis, -1, |
| logger, 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| { |
| factory.waitWhile(() -> factory.creates < 2); |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| Assert.assertEquals(0, factory.closes); |
| Assert.assertEquals(0, poolStats.getIdleExpire()); |
| // no need to wait; dangerous because it gives connections a chance to expire |
| // //wait for prefill task to finish. |
| // Thread.sleep(100); |
| } |
| |
| Connection conn1 = manager.borrowConnection(500); |
| Connection conn2 = manager.borrowConnection(500); |
| Connection conn3 = manager.borrowConnection(500); |
| Connection conn4 = manager.borrowConnection(500); |
| Connection conn5 = manager.borrowConnection(500); |
| |
| // wait to make sure checked out connections aren't timed out |
| Thread.sleep(idleTimeoutMillis * 2); |
| Assert.assertEquals(5, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| Assert.assertEquals(0, factory.closes); |
| Assert.assertEquals(0, poolStats.getIdleExpire()); |
| |
| { |
| // make sure a connection that has been passivated can idle-expire |
| conn1.passivate(true); |
| |
| long elapsedMillis = factory.waitWhile(() -> factory.destroys < 1); |
| Assert.assertEquals(5, factory.creates); |
| Assert.assertEquals(1, factory.destroys); |
| Assert.assertEquals(1, factory.closes); |
| Assert.assertEquals(1, poolStats.getIdleExpire()); |
| checkIdleTimeout(idleTimeoutMillis, elapsedMillis); |
| } |
| |
| // now return all other connections to pool and verify that just 2 expire |
| manager.returnConnection(conn2); |
| manager.returnConnection(conn3); |
| manager.returnConnection(conn4); |
| manager.returnConnection(conn5); |
| |
| { |
| long elapsedMillis = factory.waitWhile(() -> factory.destroys < 3); |
| Assert.assertEquals(5, factory.creates); |
| Assert.assertEquals(3, factory.destroys); |
| Assert.assertEquals(3, factory.closes); |
| Assert.assertEquals(3, poolStats.getIdleExpire()); |
| checkIdleTimeout(idleTimeoutMillis, elapsedMillis); |
| } |
| |
| // wait to make sure min-connections don't time out |
| Thread.sleep(idleTimeoutMillis * 2); |
| Assert.assertEquals(5, factory.creates); |
| Assert.assertEquals(3, factory.destroys); |
| Assert.assertEquals(3, factory.closes); |
| Assert.assertEquals(3, poolStats.getIdleExpire()); |
| } |
| |
| private void checkIdleTimeout(final long idleTimeoutMillis, long elapsedMillis) { |
| Assert.assertTrue( |
| "Elapsed " + elapsedMillis + " is less than idle timeout " + idleTimeoutMillis, |
| elapsedMillis >= (idleTimeoutMillis - ALLOWABLE_ERROR_IN_MILLIS)); |
| Assert.assertTrue( |
| "Elapsed " + elapsedMillis + " is greater than idle timeout " + idleTimeoutMillis, |
| elapsedMillis <= (idleTimeoutMillis + ALLOWABLE_ERROR_IN_MILLIS)); |
| } |
| |
| @Test |
| public void testBug41516() |
| throws InterruptedException, AllConnectionsInUseException, NoAvailableServersException { |
| final long idleTimeoutMillis = 300; |
| final long BORROW_TIMEOUT_MILLIS = 500; |
| manager = |
| new ConnectionManagerImpl("pool", factory, endpointManager, 2, 1, idleTimeoutMillis, -1, |
| logger, 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| Connection conn1 = manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| Connection conn2 = manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| |
| // Return some connections, let them idle expire |
| manager.returnConnection(conn1); |
| manager.returnConnection(conn2); |
| |
| { |
| long elapsedMillis = factory.waitWhile(() -> factory.destroys < 1); |
| Assert.assertEquals(1, factory.destroys); |
| Assert.assertEquals(1, factory.closes); |
| Assert.assertEquals(1, poolStats.getIdleExpire()); |
| Assert.assertTrue( |
| "Elapsed " + elapsedMillis + " is less than idle timeout " + idleTimeoutMillis, |
| elapsedMillis + ALLOWABLE_ERROR_IN_MILLIS >= idleTimeoutMillis); |
| } |
| |
| // Ok, now get some connections that fill our queue |
| Connection ping1 = |
| manager.borrowConnection(new ServerLocation("localhost", 5), false); |
| Connection ping2 = |
| manager.borrowConnection(new ServerLocation("localhost", 5), false); |
| manager.returnConnection(ping1); |
| manager.returnConnection(ping2); |
| |
| manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| long startNanos = nowNanos(); |
| try { |
| manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| fail("Didn't get an exception"); |
| } catch (AllConnectionsInUseException e) { |
| // expected |
| } |
| long elapsedMillis = elapsedMillis(startNanos); |
| Assert.assertTrue("Elapsed = " + elapsedMillis, |
| elapsedMillis >= BORROW_TIMEOUT_MILLIS - ALLOWABLE_ERROR_IN_MILLIS); |
| } |
| |
| @Test |
| public void testLifetimeExpiration() throws InterruptedException, AllConnectionsInUseException, |
| NoAvailableServersException, Throwable { |
| int lifetimeTimeout = 500; |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 2, 2, -1, lifetimeTimeout, |
| logger, 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| { |
| factory.waitWhile(() -> factory.creates < 2); |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| Assert.assertEquals(0, factory.finds); |
| } |
| |
| // need to start a thread that keeps the connections busy |
| // so that their last access time keeps changing |
| AtomicReference exception = new AtomicReference(); |
| int updaterCount = 2; |
| UpdaterThread[] updaters = new UpdaterThread[updaterCount]; |
| |
| for (int i = 0; i < updaterCount; i++) { |
| updaters[i] = new UpdaterThread(null, exception, i, (lifetimeTimeout / 10) * 2); |
| } |
| |
| for (int i = 0; i < updaterCount; i++) { |
| updaters[i].start(); |
| } |
| |
| { |
| long durationMillis = factory.waitWhile(() -> factory.finds < 2); |
| Assert.assertEquals(2, factory.finds); |
| // server shouldn't have changed so no increase in creates or destroys |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| Assert.assertEquals(0, factory.closes); |
| |
| Assert.assertTrue("took too long to expire lifetime; expected=" + lifetimeTimeout |
| + " but took=" + durationMillis, durationMillis < lifetimeTimeout * 5); |
| } |
| |
| for (int i = 0; i < updaterCount; i++) { |
| ThreadUtils.join(updaters[i], 30 * 1000); |
| } |
| |
| if (exception.get() != null) { |
| throw (Throwable) exception.get(); |
| } |
| |
| for (int i = 0; i < updaterCount; i++) { |
| Assert.assertFalse("Updater [" + i + "] is still running", updaters[i].isAlive()); |
| } |
| } |
| |
| @Test |
| public void testExclusiveConnectionAccess() throws Throwable { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 1, 0, -1, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| AtomicReference exception = new AtomicReference(); |
| AtomicBoolean haveConnection = new AtomicBoolean(); |
| int updaterCount = 10; |
| UpdaterThread[] updaters = new UpdaterThread[updaterCount]; |
| |
| for (int i = 0; i < updaterCount; i++) { |
| updaters[i] = new UpdaterThread(haveConnection, exception, i); |
| } |
| |
| for (int i = 0; i < updaterCount; i++) { |
| updaters[i].start(); |
| } |
| |
| for (int i = 0; i < updaterCount; i++) { |
| ThreadUtils.join(updaters[i], 30 * 1000); |
| } |
| |
| if (exception.get() != null) { |
| throw (Throwable) exception.get(); |
| } |
| |
| for (int i = 0; i < updaterCount; i++) { |
| Assert.assertFalse("Updater [" + i + "] is still running", updaters[i].isAlive()); |
| } |
| } |
| |
| @Test |
| public void testClose() |
| throws AllConnectionsInUseException, NoAvailableServersException, InterruptedException { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 10, 0, -1, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| Connection conn1 = manager.borrowConnection(0); |
| manager.borrowConnection(0); |
| manager.returnConnection(conn1); |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| |
| manager.close(false); |
| |
| Assert.assertEquals(2, factory.closes); |
| Assert.assertEquals(2, factory.destroys); |
| |
| } |
| |
| @Test |
| public void testExchangeConnection() throws Exception { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 2, 0, -1, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| Connection conn1 = manager.borrowConnection(10); |
| Connection conn2 = manager.borrowConnection(10); |
| try { |
| manager.borrowConnection(10); |
| fail("Exepected no servers available"); |
| } catch (AllConnectionsInUseException e) { |
| // expected |
| } |
| |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| Assert.assertEquals(2, manager.getConnectionCount()); |
| |
| Connection conn3 = manager.exchangeConnection(conn1, Collections.emptySet()); |
| |
| Assert.assertEquals(3, factory.creates); |
| Assert.assertEquals(1, factory.destroys); |
| Assert.assertEquals(2, manager.getConnectionCount()); |
| |
| manager.returnConnection(conn2); |
| |
| Assert.assertEquals(3, factory.creates); |
| Assert.assertEquals(1, factory.destroys); |
| Assert.assertEquals(2, manager.getConnectionCount()); |
| |
| Connection conn4 = |
| manager.exchangeConnection(conn3, Collections.singleton(conn3.getServer())); |
| |
| Assert.assertEquals(4, factory.creates); |
| Assert.assertEquals(2, factory.destroys); |
| Assert.assertEquals(2, manager.getConnectionCount()); |
| |
| manager.returnConnection(conn4); |
| } |
| |
| /** |
| * This tests that a deadlock between connection formation and connection pool closing has been |
| * fixed. See GEODE-4615 |
| */ |
| @Test |
| public void testThatMapCloseCausesCacheClosedException() throws Exception { |
| final ConnectionManagerImpl connectionManager = new ConnectionManagerImpl("pool", factory, |
| endpointManager, 2, 0, -1, -1, logger, 60 * 1000, cancelCriterion, poolStats); |
| manager = connectionManager; |
| connectionManager.start(background); |
| final ConnectionManagerImpl.ConnectionMap connectionMap = connectionManager.allConnectionsMap; |
| |
| final int thread1 = 0; |
| final int thread2 = 1; |
| final boolean[] ready = new boolean[2]; |
| Thread thread = new Thread("ConnectionManagerJUnitTest thread") { |
| @Override |
| public void run() { |
| setReady(ready, thread1); |
| waitUntilReady(ready, thread2); |
| connectionMap.close(false); |
| } |
| }; |
| thread.setDaemon(true); |
| thread.start(); |
| try { |
| Connection firstConnection = connectionManager.borrowConnection(0); |
| synchronized (firstConnection) { |
| setReady(ready, thread2); |
| waitUntilReady(ready, thread1); |
| // the other thread will now try to close the connection map but it will block |
| // because this thread has locked one of the connections |
| await().until(() -> connectionMap.closing); |
| try { |
| connectionManager.borrowConnection(0); |
| fail("expected a CacheClosedException"); |
| } catch (CacheClosedException e) { |
| // expected |
| } |
| } |
| } finally { |
| if (thread.isAlive()) { |
| System.out.println("stopping background thread"); |
| thread.interrupt(); |
| thread.join(); |
| } |
| } |
| } |
| |
| private void setReady(boolean[] ready, int index) { |
| System.out.println( |
| Thread.currentThread().getName() + ": setting that thread" + (index + 1) + " is ready"); |
| synchronized (ready) { |
| ready[index] = true; |
| } |
| } |
| |
| private void waitUntilReady(boolean[] ready, int index) { |
| System.out.println( |
| Thread.currentThread().getName() + ": waiting for thread" + (index + 1) + " to be ready"); |
| await().until(() -> { |
| synchronized (ready) { |
| return (ready[index]); |
| } |
| }); |
| } |
| |
| private long nowNanos() { |
| return System.nanoTime(); |
| } |
| |
| private long elapsedNanos(long startNanos) { |
| return nowNanos() - startNanos; |
| } |
| |
| private long elapsedMillis(long startNanos) { |
| return NANOSECONDS.toMillis(elapsedNanos(startNanos)); |
| } |
| |
| @Test |
| public void testBlocking() throws Throwable { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 1, 0, -1, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| final Connection conn1 = manager.borrowConnection(10); |
| |
| long BORROW_TIMEOUT_MILLIS = 300; |
| long startNonos = nowNanos(); |
| try { |
| manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| fail("Should have received no servers available"); |
| } catch (AllConnectionsInUseException expected) { |
| |
| } |
| long elapsedMillis = elapsedMillis(startNonos); |
| Assert.assertTrue( |
| "Should have blocked for " + BORROW_TIMEOUT_MILLIS + " millis for a connection", |
| elapsedMillis >= BORROW_TIMEOUT_MILLIS - ALLOWABLE_ERROR_IN_MILLIS); |
| |
| Thread returnThread = new Thread() { |
| @Override |
| public void run() { |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException e) { |
| fail("interrupted"); |
| } |
| manager.returnConnection(conn1); |
| } |
| }; |
| |
| returnThread.start(); |
| BORROW_TIMEOUT_MILLIS = 5000; |
| startNonos = nowNanos(); |
| Connection conn2 = manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| elapsedMillis = elapsedMillis(startNonos); |
| Assert.assertTrue( |
| "Should have blocked for less than " + BORROW_TIMEOUT_MILLIS + " milliseconds", |
| elapsedMillis < BORROW_TIMEOUT_MILLIS + ALLOWABLE_ERROR_IN_MILLIS); |
| manager.returnConnection(conn2); |
| |
| |
| final Connection conn3 = manager.borrowConnection(10); |
| Thread invalidateThread = new Thread() { |
| @Override |
| public void run() { |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException e) { |
| fail("interrupted"); |
| } |
| conn3.destroy(); |
| manager.returnConnection(conn3); |
| } |
| }; |
| |
| invalidateThread.start(); |
| startNonos = nowNanos(); |
| conn2 = manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| elapsedMillis = elapsedMillis(startNonos); |
| Assert.assertTrue( |
| "Should have blocked for less than " + BORROW_TIMEOUT_MILLIS + " milliseconds", |
| elapsedMillis < BORROW_TIMEOUT_MILLIS + ALLOWABLE_ERROR_IN_MILLIS); |
| manager.returnConnection(conn2); |
| |
| final Connection conn4 = manager.borrowConnection(10); |
| Thread invalidateThread2 = new Thread() { |
| @Override |
| public void run() { |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException e) { |
| fail("interrupted"); |
| } |
| endpointManager.serverCrashed(conn4.getEndpoint()); |
| manager.returnConnection(conn4); |
| } |
| }; |
| |
| invalidateThread2.start(); |
| startNonos = nowNanos(); |
| conn2 = manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| elapsedMillis = elapsedMillis(startNonos); |
| Assert.assertTrue( |
| "Should have blocked for less than " + BORROW_TIMEOUT_MILLIS + " milliseconds", |
| elapsedMillis < BORROW_TIMEOUT_MILLIS + ALLOWABLE_ERROR_IN_MILLIS); |
| manager.returnConnection(conn2); |
| } |
| |
| @Test |
| public void testExplicitServer() throws Exception { |
| manager = new ConnectionManagerImpl("pool", factory, endpointManager, 1, 0, -1, -1, logger, |
| 60 * 1000, cancelCriterion, poolStats); |
| manager.start(background); |
| |
| Connection conn1 = manager.borrowConnection(0); |
| |
| try { |
| manager.borrowConnection(10); |
| fail("Should have received an error"); |
| } catch (AllConnectionsInUseException expected) { |
| // do nothing |
| } |
| |
| Connection conn3 = manager.borrowConnection(new ServerLocation("localhost", -2), false); |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(0, factory.destroys); |
| Assert.assertEquals(0, factory.closes); |
| |
| manager.returnConnection(conn3); |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(1, factory.destroys); |
| Assert.assertEquals(1, factory.closes); |
| |
| manager.returnConnection(conn1); |
| Assert.assertEquals(2, factory.creates); |
| Assert.assertEquals(1, factory.destroys); |
| Assert.assertEquals(1, factory.closes); |
| } |
| |
| private class UpdaterThread extends Thread { |
| |
| private AtomicReference exception; |
| |
| private final AtomicBoolean haveConnection; |
| |
| private int id; |
| private final int iterations; |
| |
| public UpdaterThread(AtomicBoolean haveConnection, AtomicReference exception, int id) { |
| this(haveConnection, exception, id, 10); |
| } |
| |
| public UpdaterThread(AtomicBoolean haveConnection, AtomicReference exception, int id, |
| int iterations) { |
| this.haveConnection = haveConnection; |
| this.exception = exception; |
| this.id = id; |
| this.iterations = iterations; |
| } |
| |
| private Connection borrow(int i) { |
| long startNanos = nowNanos(); |
| long BORROW_TIMEOUT_MILLIS = 2000; |
| Connection conn = manager.borrowConnection(BORROW_TIMEOUT_MILLIS); |
| if (haveConnection != null) { |
| Assert.assertTrue("Updater[" + id + "] loop[" + i + "] Someone else has the connection!", |
| haveConnection.compareAndSet(false, true)); |
| } |
| long elapsedMillis = elapsedMillis(startNanos); |
| Assert.assertTrue("Elapsed time (" + elapsedMillis + ") >= " + BORROW_TIMEOUT_MILLIS, |
| elapsedMillis < BORROW_TIMEOUT_MILLIS + ALLOWABLE_ERROR_IN_MILLIS); |
| return conn; |
| } |
| |
| @Override |
| public void run() { |
| int i = 0; |
| Connection conn = null; |
| try { |
| for (i = 0; i < iterations; i++) { |
| conn = borrow(i); |
| try { |
| Thread.sleep(10); |
| if (haveConnection != null) { |
| Assert.assertTrue( |
| "Updater[" + id + "] loop[" + i + "] Someone else changed the connection flag", |
| haveConnection.compareAndSet(true, false)); |
| } |
| } finally { |
| manager.returnConnection(conn); |
| } |
| } |
| } catch (Throwable t) { |
| this.exception.compareAndSet(null, |
| new Exception("ERROR Updater[" + id + "] loop[" + i + "]", t)); |
| } |
| |
| } |
| } |
| |
| public class DummyFactory implements ConnectionFactory { |
| public ServerLocation nextServer = new ServerLocation("localhost", -1); |
| protected volatile int creates; |
| protected volatile int destroys; |
| protected volatile int closes; |
| protected volatile int finds; |
| |
| /** |
| * Wait as long as "whileCondition" is true. |
| * The wait will timeout after TIMEOUT_MILLIS has elapsed. |
| * |
| * @return the elapsed time in milliseconds. |
| */ |
| public synchronized long waitWhile(BooleanSupplier whileCondition) throws InterruptedException { |
| final long startNanos = nowNanos(); |
| long remainingMillis = TIMEOUT_MILLIS; |
| while (whileCondition.getAsBoolean() && remainingMillis > 0) { |
| wait(remainingMillis); |
| remainingMillis = TIMEOUT_MILLIS - elapsedMillis(startNanos); |
| } |
| return elapsedMillis(startNanos); |
| } |
| |
| @Override |
| public ServerDenyList getDenyList() { |
| return new ServerDenyList(1); |
| } |
| |
| |
| @Override |
| public ServerLocation findBestServer(ServerLocation currentServer, Set excludedServers) { |
| synchronized (this) { |
| finds++; |
| this.notifyAll(); |
| } |
| if (excludedServers != null) { |
| if (excludedServers.contains(nextServer)) { |
| return null; |
| } |
| } |
| return nextServer; |
| } |
| |
| @Override |
| public Connection createClientToServerConnection(Set excluded) { |
| return createClientToServerConnection(nextServer, true); |
| } |
| |
| |
| /* |
| * (non-Javadoc) |
| * |
| * @see |
| * org.apache.geode.cache.client.internal.ConnectionFactory#createClientToServerConnection(org. |
| * apache.geode.distributed.internal.ServerLocation) |
| */ |
| @Override |
| public Connection createClientToServerConnection(final ServerLocation location, |
| boolean forQueue) { |
| synchronized (this) { |
| creates++; |
| this.notifyAll(); |
| } |
| DistributedMember fakeMember = null; |
| fakeMember = new InternalDistributedMember("localhost", 555); |
| final DistributedMember member = fakeMember; |
| |
| return new Connection() { |
| |
| private Endpoint endpoint = endpointManager.referenceEndpoint(location, member); |
| |
| @Override |
| public void destroy() { |
| synchronized (DummyFactory.this) { |
| destroys++; |
| DummyFactory.this.notifyAll(); |
| } |
| } |
| |
| @Override |
| public ServerLocation getServer() { |
| return location; |
| } |
| |
| @Override |
| public ByteBuffer getCommBuffer() { |
| return null; |
| } |
| |
| @Override |
| public Socket getSocket() { |
| return null; |
| } |
| |
| @Override |
| public ConnectionStats getStats() { |
| return null; |
| } |
| |
| @Override |
| public void close(boolean keepAlive) throws Exception { |
| synchronized (DummyFactory.this) { |
| closes++; |
| DummyFactory.this.notifyAll(); |
| } |
| } |
| |
| @Override |
| public Endpoint getEndpoint() { |
| return endpoint; |
| } |
| |
| @Override |
| public ServerQueueStatus getQueueStatus() { |
| return null; |
| } |
| |
| @Override |
| public Object execute(Op op) throws Exception { |
| return op.attempt(this); |
| } |
| |
| @Override |
| public boolean isDestroyed() { |
| return false; |
| } |
| |
| @Override |
| public void emergencyClose() {} |
| |
| @Override |
| public short getWanSiteVersion() { |
| return -1; |
| } |
| |
| @Override |
| public int getDistributedSystemId() { |
| return -1; |
| } |
| |
| @Override |
| public void setWanSiteVersion(short wanSiteVersion) {} |
| |
| @Override |
| public InputStream getInputStream() { |
| return null; |
| } |
| |
| @Override |
| public OutputStream getOutputStream() { |
| return null; |
| } |
| |
| @Override |
| public void setConnectionID(long id) {} |
| |
| @Override |
| public long getConnectionID() { |
| return 0; |
| } |
| }; |
| } |
| |
| @Override |
| public ClientUpdater createServerToClientConnection(Endpoint endpoint, QueueManager manager, |
| boolean isPrimary, ClientUpdater failedUpdater) { |
| return null; |
| } |
| } |
| } |
