grpc
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

core: Make getTransport's fast path lock-free 

This change exposed a pre-existing bug where shutdownNow wasn't called
for decommissionedTransports. The bug is fixed and a test added in this
commit.

Fixes #2120
This commit is contained in:
Eric Anderson 2016-09-12 12:55:03 -07:00 committed by GitHub
parent 3d4ae36074
commit 6907d81109
2 changed files with 21 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
core/src/main/java/io/grpc/internal/ManagedChannelImpl.java
View File

@ -63,10 +63,10 @@ import java.net.URI;
import java.net.URISyntaxException; import java.net.URISyntaxException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet; import java.util.HashSet;
import java.util.List; 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.Executor;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
@ -139,14 +139,16 @@ public final class ManagedChannelImpl extends ManagedChannel implements WithLogI
private LoadBalancer<ClientTransport> loadBalancer; private LoadBalancer<ClientTransport> loadBalancer;
/** /**
* Maps EquivalentAddressGroups to transports for that server. * Maps EquivalentAddressGroups to transports for that server. "lock" must be held when mutating.
*/ */
@GuardedBy("lock") // Even though we set a concurrency level of 1, this is better than Collections.synchronizedMap
private final Map<EquivalentAddressGroup, TransportSet> transports = // because it doesn't need to acquire a lock for reads.
new HashMap<EquivalentAddressGroup, TransportSet>(); private final ConcurrentMap<EquivalentAddressGroup, TransportSet> transports =
new ConcurrentHashMap<EquivalentAddressGroup, TransportSet>(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") @GuardedBy("lock")
private final HashSet<TransportSet> decommissionedTransports = new HashSet<TransportSet>(); private final HashSet<TransportSet> decommissionedTransports = new HashSet<TransportSet>();
@ -425,6 +427,8 @@ public final class ManagedChannelImpl extends ManagedChannel implements WithLogI
maybeTerminateChannel(); maybeTerminateChannel();
if (!terminated) { if (!terminated) {
transportsCopy.addAll(transports.values()); transportsCopy.addAll(transports.values());
transports.clear();
decommissionedTransports.addAll(transportsCopy);
delayedTransportsCopy.addAll(delayedTransports); delayedTransportsCopy.addAll(delayedTransports);
oobTransportsCopy.addAll(oobTransports); oobTransportsCopy.addAll(oobTransports);
} }
@ -472,6 +476,7 @@ public final class ManagedChannelImpl extends ManagedChannel implements WithLogI
List<OobTransportProviderImpl> oobTransportsCopy; List<OobTransportProviderImpl> oobTransportsCopy;
synchronized (lock) { synchronized (lock) {
transportsCopy = new ArrayList<TransportSet>(transports.values()); transportsCopy = new ArrayList<TransportSet>(transports.values());
transportsCopy.addAll(decommissionedTransports);
delayedTransportsCopy = new ArrayList<DelayedClientTransport>(delayedTransports); delayedTransportsCopy = new ArrayList<DelayedClientTransport>(delayedTransports);
oobTransportsCopy = new ArrayList<OobTransportProviderImpl>(oobTransports); oobTransportsCopy = new ArrayList<OobTransportProviderImpl>(oobTransports);
} }
@ -589,7 +594,10 @@ public final class ManagedChannelImpl extends ManagedChannel implements WithLogI
@Override @Override
public ClientTransport getTransport(final EquivalentAddressGroup addressGroup) { public ClientTransport getTransport(final EquivalentAddressGroup addressGroup) {
checkNotNull(addressGroup, "addressGroup"); checkNotNull(addressGroup, "addressGroup");
TransportSet ts; TransportSet ts = transports.get(addressGroup);
if (ts != null) {
return ts.obtainActiveTransport();
}
synchronized (lock) { synchronized (lock) {
if (shutdown) { if (shutdown) {
return SHUTDOWN_TRANSPORT; return SHUTDOWN_TRANSPORT;

5
core/src/test/java/io/grpc/internal/ManagedChannelImplIdlenessTest.java
View File

@ -60,6 +60,7 @@ import io.grpc.MethodDescriptor;
import io.grpc.NameResolver; import io.grpc.NameResolver;
import io.grpc.ResolvedServerInfo; import io.grpc.ResolvedServerInfo;
import io.grpc.ResolvedServerInfoGroup; import io.grpc.ResolvedServerInfoGroup;
import io.grpc.Status;
import io.grpc.StringMarshaller; import io.grpc.StringMarshaller;
import io.grpc.TransportManager.InterimTransport; import io.grpc.TransportManager.InterimTransport;
import io.grpc.TransportManager.OobTransportProvider; import io.grpc.TransportManager.OobTransportProvider;
@ -303,6 +304,10 @@ public class ManagedChannelImplIdlenessTest {
channel.shutdown(); channel.shutdown();
verify(t1.transport).shutdown(); verify(t1.transport).shutdown();
channel.shutdownNow();
verify(t0.transport).shutdownNow(any(Status.class));
verify(t1.transport).shutdownNow(any(Status.class));
t1.listener.transportTerminated(); t1.listener.transportTerminated();
assertFalse(channel.isTerminated()); assertFalse(channel.isTerminated());
t0.listener.transportTerminated(); t0.listener.transportTerminated();