@Test public void testDifferentKeys() throws Exception { try (CloseableTestServer server1 = createServer(); CloseableTestServer server2 = createServer()) { GrpcChannelKey key1 = GrpcChannelKey.create(sConf).setServerAddress(server1.getConnectAddress()); GrpcChannelKey key2 = GrpcChannelKey.create(sConf).setServerAddress(server2.getConnectAddress()); GrpcConnection conn1 = GrpcConnectionPool.INSTANCE.acquireConnection(key1, sConf); GrpcConnection conn2 = GrpcConnectionPool.INSTANCE.acquireConnection(key2, sConf); assertNotEquals(conn1, conn2); } }
public GrpcConnection acquireConnection(GrpcChannelKey channelKey, AlluxioConfiguration conf) { GrpcConnectionKey connectionKey = getConnectionKey(channelKey, conf); CountingReference<ManagedChannel> connectionRef = mChannels.compute(connectionKey, (key, ref) -> { boolean shutdownExistingConnection = false; int existingRefCount = 0; if (ref != null) { if (waitForConnectionReady(ref.get(), conf)) { LOG.debug("Acquiring an existing connection. ConnectionKey: {}. Ref-count: {}", key, ref.getRefCount()); return ref.reference(); } else { shutdownExistingConnection = true; } } if (shutdownExistingConnection) { existingRefCount = ref.getRefCount(); LOG.debug("Shutting down an existing unhealthy connection. " + "ConnectionKey: {}. Ref-count: {}", key, existingRefCount); shutdownManagedChannel(ref.get(), conf); } LOG.debug("Creating a new managed channel. ConnectionKey: {}. Ref-count:{}", key, existingRefCount); ManagedChannel managedChannel = createManagedChannel(channelKey, conf); return new CountingReference(managedChannel, existingRefCount).reference(); }); return new GrpcConnection(connectionKey, connectionRef.get(), conf); }
GrpcConnectionPool { public GrpcConnection acquireConnection(GrpcChannelKey channelKey, AlluxioConfiguration conf) { GrpcConnectionKey connectionKey = getConnectionKey(channelKey, conf); CountingReference<ManagedChannel> connectionRef = mChannels.compute(connectionKey, (key, ref) -> { boolean shutdownExistingConnection = false; int existingRefCount = 0; if (ref != null) { if (waitForConnectionReady(ref.get(), conf)) { LOG.debug("Acquiring an existing connection. ConnectionKey: {}. Ref-count: {}", key, ref.getRefCount()); return ref.reference(); } else { shutdownExistingConnection = true; } } if (shutdownExistingConnection) { existingRefCount = ref.getRefCount(); LOG.debug("Shutting down an existing unhealthy connection. " + "ConnectionKey: {}. Ref-count: {}", key, existingRefCount); shutdownManagedChannel(ref.get(), conf); } LOG.debug("Creating a new managed channel. ConnectionKey: {}. Ref-count:{}", key, existingRefCount); ManagedChannel managedChannel = createManagedChannel(channelKey, conf); return new CountingReference(managedChannel, existingRefCount).reference(); }); return new GrpcConnection(connectionKey, connectionRef.get(), conf); } }
GrpcConnectionPool { public GrpcConnection acquireConnection(GrpcChannelKey channelKey, AlluxioConfiguration conf) { GrpcConnectionKey connectionKey = getConnectionKey(channelKey, conf); CountingReference<ManagedChannel> connectionRef = mChannels.compute(connectionKey, (key, ref) -> { boolean shutdownExistingConnection = false; int existingRefCount = 0; if (ref != null) { if (waitForConnectionReady(ref.get(), conf)) { LOG.debug("Acquiring an existing connection. ConnectionKey: {}. Ref-count: {}", key, ref.getRefCount()); return ref.reference(); } else { shutdownExistingConnection = true; } } if (shutdownExistingConnection) { existingRefCount = ref.getRefCount(); LOG.debug("Shutting down an existing unhealthy connection. " + "ConnectionKey: {}. Ref-count: {}", key, existingRefCount); shutdownManagedChannel(ref.get(), conf); } LOG.debug("Creating a new managed channel. ConnectionKey: {}. Ref-count:{}", key, existingRefCount); ManagedChannel managedChannel = createManagedChannel(channelKey, conf); return new CountingReference(managedChannel, existingRefCount).reference(); }); return new GrpcConnection(connectionKey, connectionRef.get(), conf); } GrpcConnectionPool(); }
GrpcConnectionPool { public GrpcConnection acquireConnection(GrpcChannelKey channelKey, AlluxioConfiguration conf) { GrpcConnectionKey connectionKey = getConnectionKey(channelKey, conf); CountingReference<ManagedChannel> connectionRef = mChannels.compute(connectionKey, (key, ref) -> { boolean shutdownExistingConnection = false; int existingRefCount = 0; if (ref != null) { if (waitForConnectionReady(ref.get(), conf)) { LOG.debug("Acquiring an existing connection. ConnectionKey: {}. Ref-count: {}", key, ref.getRefCount()); return ref.reference(); } else { shutdownExistingConnection = true; } } if (shutdownExistingConnection) { existingRefCount = ref.getRefCount(); LOG.debug("Shutting down an existing unhealthy connection. " + "ConnectionKey: {}. Ref-count: {}", key, existingRefCount); shutdownManagedChannel(ref.get(), conf); } LOG.debug("Creating a new managed channel. ConnectionKey: {}. Ref-count:{}", key, existingRefCount); ManagedChannel managedChannel = createManagedChannel(channelKey, conf); return new CountingReference(managedChannel, existingRefCount).reference(); }); return new GrpcConnection(connectionKey, connectionRef.get(), conf); } GrpcConnectionPool(); GrpcConnection acquireConnection(GrpcChannelKey channelKey, AlluxioConfiguration conf); void releaseConnection(GrpcConnectionKey connectionKey, AlluxioConfiguration conf); }
GrpcConnectionPool { public GrpcConnection acquireConnection(GrpcChannelKey channelKey, AlluxioConfiguration conf) { GrpcConnectionKey connectionKey = getConnectionKey(channelKey, conf); CountingReference<ManagedChannel> connectionRef = mChannels.compute(connectionKey, (key, ref) -> { boolean shutdownExistingConnection = false; int existingRefCount = 0; if (ref != null) { if (waitForConnectionReady(ref.get(), conf)) { LOG.debug("Acquiring an existing connection. ConnectionKey: {}. Ref-count: {}", key, ref.getRefCount()); return ref.reference(); } else { shutdownExistingConnection = true; } } if (shutdownExistingConnection) { existingRefCount = ref.getRefCount(); LOG.debug("Shutting down an existing unhealthy connection. " + "ConnectionKey: {}. Ref-count: {}", key, existingRefCount); shutdownManagedChannel(ref.get(), conf); } LOG.debug("Creating a new managed channel. ConnectionKey: {}. Ref-count:{}", key, existingRefCount); ManagedChannel managedChannel = createManagedChannel(channelKey, conf); return new CountingReference(managedChannel, existingRefCount).reference(); }); return new GrpcConnection(connectionKey, connectionRef.get(), conf); } GrpcConnectionPool(); GrpcConnection acquireConnection(GrpcChannelKey channelKey, AlluxioConfiguration conf); void releaseConnection(GrpcConnectionKey connectionKey, AlluxioConfiguration conf); static final GrpcConnectionPool INSTANCE; }