@Test public void testGroupSize() throws Exception { int streamingGroupSize = sConf.getInt(PropertyKey.USER_NETWORK_STREAMING_MAX_CONNECTIONS); int acquireCount = streamingGroupSize * 100; try (CloseableTestServer server = createServer()) { List<GrpcChannelKey> keys = new ArrayList(acquireCount); for (int i = 0; i < acquireCount; i++) { keys.add(GrpcChannelKey.create(sConf).setNetworkGroup(GrpcNetworkGroup.STREAMING) .setServerAddress(server.getConnectAddress())); } List<GrpcConnection> connections = keys.stream().map((key) -> GrpcConnectionPool.INSTANCE.acquireConnection(key, sConf)) .collect(Collectors.toList()); Assert.assertEquals(streamingGroupSize, connections.stream().distinct().collect(Collectors.toList()).size()); } }
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; }