diff --git a/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java b/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java index 7f14da2450..f9425eef0e 100644 --- a/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java +++ b/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java @@ -63,10 +63,10 @@ import java.net.URI; import java.net.URISyntaxException; import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; import java.util.HashSet; import java.util.List; -import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; @@ -139,14 +139,16 @@ public final class ManagedChannelImpl extends ManagedChannel implements WithLogI private LoadBalancer loadBalancer; /** - * Maps EquivalentAddressGroups to transports for that server. + * Maps EquivalentAddressGroups to transports for that server. "lock" must be held when mutating. */ - @GuardedBy("lock") - private final Map transports = - new HashMap(); + // Even though we set a concurrency level of 1, this is better than Collections.synchronizedMap + // because it doesn't need to acquire a lock for reads. + private final ConcurrentMap transports = + new ConcurrentHashMap(16, .75f, 1); /** - * TransportSets that are shutdown (but not yet terminated) due to channel idleness. + * TransportSets that are shutdown (but not yet terminated) due to channel idleness or channel + * shut down. */ @GuardedBy("lock") private final HashSet decommissionedTransports = new HashSet(); @@ -425,6 +427,8 @@ public final class ManagedChannelImpl extends ManagedChannel implements WithLogI maybeTerminateChannel(); if (!terminated) { transportsCopy.addAll(transports.values()); + transports.clear(); + decommissionedTransports.addAll(transportsCopy); delayedTransportsCopy.addAll(delayedTransports); oobTransportsCopy.addAll(oobTransports); } @@ -472,6 +476,7 @@ public final class ManagedChannelImpl extends ManagedChannel implements WithLogI List oobTransportsCopy; synchronized (lock) { transportsCopy = new ArrayList(transports.values()); + transportsCopy.addAll(decommissionedTransports); delayedTransportsCopy = new ArrayList(delayedTransports); oobTransportsCopy = new ArrayList(oobTransports); } @@ -589,7 +594,10 @@ public final class ManagedChannelImpl extends ManagedChannel implements WithLogI @Override public ClientTransport getTransport(final EquivalentAddressGroup addressGroup) { checkNotNull(addressGroup, "addressGroup"); - TransportSet ts; + TransportSet ts = transports.get(addressGroup); + if (ts != null) { + return ts.obtainActiveTransport(); + } synchronized (lock) { if (shutdown) { return SHUTDOWN_TRANSPORT; diff --git a/core/src/test/java/io/grpc/internal/ManagedChannelImplIdlenessTest.java b/core/src/test/java/io/grpc/internal/ManagedChannelImplIdlenessTest.java index 9518fd32c6..ff9235990a 100644 --- a/core/src/test/java/io/grpc/internal/ManagedChannelImplIdlenessTest.java +++ b/core/src/test/java/io/grpc/internal/ManagedChannelImplIdlenessTest.java @@ -60,6 +60,7 @@ import io.grpc.MethodDescriptor; import io.grpc.NameResolver; import io.grpc.ResolvedServerInfo; import io.grpc.ResolvedServerInfoGroup; +import io.grpc.Status; import io.grpc.StringMarshaller; import io.grpc.TransportManager.InterimTransport; import io.grpc.TransportManager.OobTransportProvider; @@ -303,6 +304,10 @@ public class ManagedChannelImplIdlenessTest { channel.shutdown(); verify(t1.transport).shutdown(); + channel.shutdownNow(); + verify(t0.transport).shutdownNow(any(Status.class)); + verify(t1.transport).shutdownNow(any(Status.class)); + t1.listener.transportTerminated(); assertFalse(channel.isTerminated()); t0.listener.transportTerminated();