diff --git a/netty/src/main/java/io/grpc/netty/EpollTcpInfoAccessor.java b/netty/src/main/java/io/grpc/netty/EpollTcpInfoAccessor.java new file mode 100644 index 00000000000..00ed86f2d70 --- /dev/null +++ b/netty/src/main/java/io/grpc/netty/EpollTcpInfoAccessor.java @@ -0,0 +1,142 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed 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 io.grpc.netty; + +import com.google.common.annotations.VisibleForTesting; +import io.netty.channel.Channel; +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; +import java.util.EnumMap; +import java.util.logging.Level; +import java.util.logging.Logger; +import javax.annotation.Nullable; + +/** + * Reflective access to {@code EpollSocketChannel.tcpInfo()}, since grpc-netty does not depend on + * netty-epoll at compile time. + */ +final class EpollTcpInfoAccessor { + private static final Logger log = Logger.getLogger(EpollTcpInfoAccessor.class.getName()); + + @Nullable + static final EpollTcpInfoAccessor INSTANCE = load(); + + enum Field { + STATE("state"), + CA_STATE("caState"), + RETRANSMITS("retransmits"), + PROBES("probes"), + BACKOFF("backoff"), + OPTIONS("options"), + SND_WSCALE("sndWscale"), + RCV_WSCALE("rcvWscale"), + RTO("rto"), + ATO("ato"), + SND_MSS("sndMss"), + RCV_MSS("rcvMss"), + UNACKED("unacked"), + SACKED("sacked"), + LOST("lost"), + RETRANS("retrans"), + FACKETS("fackets"), + LAST_DATA_SENT("lastDataSent"), + LAST_ACK_SENT("lastAckSent"), + LAST_DATA_RECV("lastDataRecv"), + LAST_ACK_RECV("lastAckRecv"), + PMTU("pmtu"), + RCV_SSTHRESH("rcvSsthresh"), + RTT("rtt"), + RTTVAR("rttvar"), + SND_SSTHRESH("sndSsthresh"), + SND_CWND("sndCwnd"), + ADVMSS("advmss"), + REORDERING("reordering"), + TOTAL_RETRANS("totalRetrans"); + + final String methodName; + + Field(String methodName) { + this.methodName = methodName; + } + } + + private final Class channelClass; + private final Constructor infoConstructor; + private final Method tcpInfo; + private final EnumMap getters; + + private EpollTcpInfoAccessor( + Class channelClass, Constructor infoConstructor, Method tcpInfo, + EnumMap getters) { + this.channelClass = channelClass; + this.infoConstructor = infoConstructor; + this.tcpInfo = tcpInfo; + this.getters = getters; + } + + @Nullable + private static EpollTcpInfoAccessor load() { + boolean epollAvailable = false; + try { + Class epollClass = Class.forName("io.netty.channel.epoll.Epoll"); + Method isAvailableMethod = epollClass.getDeclaredMethod("isAvailable"); + epollAvailable = (Boolean) isAvailableMethod.invoke(null); + if (epollAvailable) { + return create( + Class.forName("io.netty.channel.epoll.EpollSocketChannel"), + Class.forName("io.netty.channel.epoll.EpollTcpInfo")); + } + } catch (ReflectiveOperationException e) { + log.log(Level.FINE, "Failed to initialize Epoll tcp_info reflection", e); + } finally { + log.log(Level.FINE, "Epoll available during static init of EpollTcpInfoAccessor:" + + "{0}", epollAvailable); + } + return null; + } + + @VisibleForTesting + static EpollTcpInfoAccessor create(Class channelClass, Class infoClass) + throws ReflectiveOperationException { + EnumMap getters = new EnumMap<>(Field.class); + for (Field field : Field.values()) { + getters.put(field, infoClass.getMethod(field.methodName)); + } + return new EpollTcpInfoAccessor( + channelClass, + infoClass.getDeclaredConstructor(), + channelClass.getMethod("tcpInfo", infoClass), + getters); + } + + boolean isEpollSocketChannel(Channel channel) { + return channelClass.isInstance(channel); + } + + Object newTcpInfo() throws ReflectiveOperationException { + return infoConstructor.newInstance(); + } + + /** Fills {@code info} with the current {@code TCP_INFO} of {@code channel}. */ + void read(Channel channel, Object info) throws ReflectiveOperationException { + tcpInfo.invoke(channel, info); + } + + long get(Object info, Field field) throws ReflectiveOperationException { + return ((Number) getters.get(field).invoke(info)).longValue(); + } +} diff --git a/netty/src/main/java/io/grpc/netty/NettySocketSupport.java b/netty/src/main/java/io/grpc/netty/NettySocketSupport.java index 528a3978ce9..0f12783beec 100644 --- a/netty/src/main/java/io/grpc/netty/NettySocketSupport.java +++ b/netty/src/main/java/io/grpc/netty/NettySocketSupport.java @@ -16,18 +16,25 @@ package io.grpc.netty; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; import io.grpc.InternalChannelz.TcpInfo; +import io.grpc.netty.EpollTcpInfoAccessor.Field; import io.netty.channel.Channel; import java.util.Map; +import java.util.logging.Level; +import java.util.logging.Logger; import javax.annotation.Nullable; /** * An class for getting low level socket info. */ final class NettySocketSupport { - private static volatile Helper instance = new NettySocketHelperImpl(); + private static final Logger log = Logger.getLogger(NettySocketSupport.class.getName()); + + private static volatile Helper instance = + new NettySocketHelperImpl(EpollTcpInfoAccessor.INSTANCE); interface Helper { /** @@ -63,12 +70,67 @@ static void setHelper(Helper helper) { instance = Preconditions.checkNotNull(helper); } - private static final class NettySocketHelperImpl implements Helper { + @VisibleForTesting + static final class NettySocketHelperImpl implements Helper { + @Nullable + private final EpollTcpInfoAccessor epollInfo; + + NettySocketHelperImpl(@Nullable EpollTcpInfoAccessor epollInfo) { + this.epollInfo = epollInfo; + } + @Override + @Nullable public NativeSocketOptions getNativeSocketOptions(Channel ch) { - // TODO(zpencer): if netty-epoll, use reflection to call EpollSocketChannel.tcpInfo() - // And/or if some other low level socket support library is available, call it now. - return null; + if (epollInfo == null || !epollInfo.isEpollSocketChannel(ch)) { + return null; + } + TcpInfo tcpInfo; + try { + tcpInfo = readTcpInfo(epollInfo, ch); + } catch (ReflectiveOperationException e) { + log.log(Level.FINE, "Failed to read Epoll tcp_info", e); + return null; + } + return new NativeSocketOptions(tcpInfo, ImmutableMap.of()); + } + + // Channelz fields are uint32, so narrowing the epoll values to int keeps their bits. + private static TcpInfo readTcpInfo(EpollTcpInfoAccessor epollInfo, Channel ch) + throws ReflectiveOperationException { + Object info = epollInfo.newTcpInfo(); + epollInfo.read(ch, info); + return new TcpInfo.Builder() + .setState((int) epollInfo.get(info, Field.STATE)) + .setCaState((int) epollInfo.get(info, Field.CA_STATE)) + .setRetransmits((int) epollInfo.get(info, Field.RETRANSMITS)) + .setProbes((int) epollInfo.get(info, Field.PROBES)) + .setBackoff((int) epollInfo.get(info, Field.BACKOFF)) + .setOptions((int) epollInfo.get(info, Field.OPTIONS)) + .setSndWscale((int) epollInfo.get(info, Field.SND_WSCALE)) + .setRcvWscale((int) epollInfo.get(info, Field.RCV_WSCALE)) + .setRto((int) epollInfo.get(info, Field.RTO)) + .setAto((int) epollInfo.get(info, Field.ATO)) + .setSndMss((int) epollInfo.get(info, Field.SND_MSS)) + .setRcvMss((int) epollInfo.get(info, Field.RCV_MSS)) + .setUnacked((int) epollInfo.get(info, Field.UNACKED)) + .setSacked((int) epollInfo.get(info, Field.SACKED)) + .setLost((int) epollInfo.get(info, Field.LOST)) + .setRetrans((int) epollInfo.get(info, Field.RETRANS)) + .setFackets((int) epollInfo.get(info, Field.FACKETS)) + .setLastDataSent((int) epollInfo.get(info, Field.LAST_DATA_SENT)) + .setLastAckSent((int) epollInfo.get(info, Field.LAST_ACK_SENT)) + .setLastDataRecv((int) epollInfo.get(info, Field.LAST_DATA_RECV)) + .setLastAckRecv((int) epollInfo.get(info, Field.LAST_ACK_RECV)) + .setPmtu((int) epollInfo.get(info, Field.PMTU)) + .setRcvSsthresh((int) epollInfo.get(info, Field.RCV_SSTHRESH)) + .setRtt((int) epollInfo.get(info, Field.RTT)) + .setRttvar((int) epollInfo.get(info, Field.RTTVAR)) + .setSndSsthresh((int) epollInfo.get(info, Field.SND_SSTHRESH)) + .setSndCwnd((int) epollInfo.get(info, Field.SND_CWND)) + .setAdvmss((int) epollInfo.get(info, Field.ADVMSS)) + .setReordering((int) epollInfo.get(info, Field.REORDERING)) + .build(); } } } diff --git a/netty/src/main/java/io/grpc/netty/TcpMetrics.java b/netty/src/main/java/io/grpc/netty/TcpMetrics.java index 10f3a55014d..34edf1d484a 100644 --- a/netty/src/main/java/io/grpc/netty/TcpMetrics.java +++ b/netty/src/main/java/io/grpc/netty/TcpMetrics.java @@ -21,7 +21,6 @@ import io.grpc.MetricRecorder; import io.netty.channel.Channel; import io.netty.util.concurrent.ScheduledFuture; -import java.lang.reflect.Method; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.Arrays; @@ -38,57 +37,7 @@ final class TcpMetrics { private static final Logger log = Logger.getLogger(TcpMetrics.class.getName()); - static EpollInfo epollInfo = loadEpollInfo(); - - static final class EpollInfo { - final Class channelClass; - final java.lang.reflect.Constructor infoConstructor; - final Method tcpInfo; - final Method totalRetrans; - final Method retransmits; - final Method rtt; - - EpollInfo( - Class channelClass, - java.lang.reflect.Constructor infoConstructor, - Method tcpInfo, - Method totalRetrans, - Method retransmits, - Method rtt) { - this.channelClass = channelClass; - this.infoConstructor = infoConstructor; - this.tcpInfo = tcpInfo; - this.totalRetrans = totalRetrans; - this.retransmits = retransmits; - this.rtt = rtt; - } - } - - static EpollInfo loadEpollInfo() { - boolean epollAvailable = false; - try { - Class epollClass = Class.forName("io.netty.channel.epoll.Epoll"); - Method isAvailableMethod = epollClass.getDeclaredMethod("isAvailable"); - epollAvailable = (Boolean) isAvailableMethod.invoke(null); - if (epollAvailable) { - Class channelClass = Class.forName("io.netty.channel.epoll.EpollSocketChannel"); - Class infoClass = Class.forName("io.netty.channel.epoll.EpollTcpInfo"); - return new EpollInfo( - channelClass, - infoClass.getDeclaredConstructor(), - channelClass.getMethod("tcpInfo", infoClass), - infoClass.getMethod("totalRetrans"), - infoClass.getMethod("retrans"), - infoClass.getMethod("rtt")); - } - } catch (ReflectiveOperationException e) { - log.log(Level.FINE, "Failed to initialize Epoll tcp_info reflection", e); - } finally { - log.log(Level.FINE, "Epoll available during static init of TcpMetrics:" - + "{0}", epollAvailable); - } - return null; - } + static EpollTcpInfoAccessor epollInfo = EpollTcpInfoAccessor.INSTANCE; private static final long RECORD_INTERVAL_MILLIS = TimeUnit.MINUTES.toMillis(5); private final MetricRecorder metricRecorder; @@ -102,7 +51,7 @@ static EpollInfo loadEpollInfo() { Object tcpInfo = null; if (epollInfo != null) { try { - tcpInfo = epollInfo.infoConstructor.newInstance(); + tcpInfo = epollInfo.newTcpInfo(); } catch (ReflectiveOperationException e) { log.log(Level.FINE, "Failed to instantiate EpollTcpInfo", e); } @@ -120,7 +69,7 @@ void channelActive(Channel channel) { } private void scheduleNextReport(final Channel channel, boolean isInitial) { - if (epollInfo == null || !epollInfo.channelClass.isInstance(channel) || !channel.isActive()) { + if (epollInfo == null || !epollInfo.isEpollSocketChannel(channel) || !channel.isActive()) { return; } @@ -147,7 +96,7 @@ void channelInactive(Channel channel) { metricRecorder.addLongUpDownCounter(InternalTcpMetrics.CONNECTION_COUNT_INSTRUMENT, -1, Collections.emptyList(), labelValues); // Final collection on close - if (epollInfo != null && epollInfo.channelClass.isInstance(channel)) { + if (epollInfo != null && epollInfo.isEpollSocketChannel(channel)) { recordTcpInfo(channel, true); } } @@ -157,7 +106,7 @@ void recordTcpInfo(Channel channel) { } private void recordTcpInfo(Channel channel, boolean isClose) { - if (epollInfo == null || !epollInfo.channelClass.isInstance(channel)) { + if (epollInfo == null || !epollInfo.isEpollSocketChannel(channel)) { return; } List labelValues = getLabelValues(channel); @@ -165,10 +114,10 @@ private void recordTcpInfo(Channel channel, boolean isClose) { long retransmits; long rtt; try { - epollInfo.tcpInfo.invoke(channel, tcpInfo); - totalRetrans = (Long) epollInfo.totalRetrans.invoke(tcpInfo); - retransmits = (Long) epollInfo.retransmits.invoke(tcpInfo); - rtt = (Long) epollInfo.rtt.invoke(tcpInfo); + epollInfo.read(channel, tcpInfo); + totalRetrans = epollInfo.get(tcpInfo, EpollTcpInfoAccessor.Field.TOTAL_RETRANS); + retransmits = epollInfo.get(tcpInfo, EpollTcpInfoAccessor.Field.RETRANSMITS); + rtt = epollInfo.get(tcpInfo, EpollTcpInfoAccessor.Field.RTT); } catch (ReflectiveOperationException e) { log.log(Level.FINE, "Error computing TCP metrics", e); return; diff --git a/netty/src/test/java/io/grpc/netty/EpollTcpInfoAccessorTest.java b/netty/src/test/java/io/grpc/netty/EpollTcpInfoAccessorTest.java new file mode 100644 index 00000000000..56440a4d9af --- /dev/null +++ b/netty/src/test/java/io/grpc/netty/EpollTcpInfoAccessorTest.java @@ -0,0 +1,46 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed 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 io.grpc.netty; + +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.channel.epoll.EpollSocketChannel; +import io.netty.channel.epoll.EpollTcpInfo; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class EpollTcpInfoAccessorTest { + + @Test + public void create_resolvesAllFieldsOnNettyEpollClasses() throws Exception { + // Class literals do not trigger epoll's native initialization, so this runs on any platform. + EpollTcpInfoAccessor accessor = + EpollTcpInfoAccessor.create(EpollSocketChannel.class, EpollTcpInfo.class); + + assertThat(accessor.isEpollSocketChannel(new EmbeddedChannel())).isFalse(); + } + + @Test + public void create_missingGetter_throws() { + assertThrows(NoSuchMethodException.class, + () -> EpollTcpInfoAccessor.create(FakeEpollTcpInfo.FakeChannel.class, Object.class)); + } +} diff --git a/netty/src/test/java/io/grpc/netty/FakeEpollTcpInfo.java b/netty/src/test/java/io/grpc/netty/FakeEpollTcpInfo.java new file mode 100644 index 00000000000..3ab18b987d6 --- /dev/null +++ b/netty/src/test/java/io/grpc/netty/FakeEpollTcpInfo.java @@ -0,0 +1,178 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed 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 io.grpc.netty; + +import io.grpc.netty.EpollTcpInfoAccessor.Field; +import io.netty.channel.embedded.EmbeddedChannel; +import java.util.EnumMap; + +/** Mirrors the public API of {@code io.netty.channel.epoll.EpollTcpInfo}. */ +public final class FakeEpollTcpInfo { + private final EnumMap values = new EnumMap<>(Field.class); + + static EpollTcpInfoAccessor newAccessor() throws ReflectiveOperationException { + return EpollTcpInfoAccessor.create(FakeChannel.class, FakeEpollTcpInfo.class); + } + + public void set(Field field, long value) { + values.put(field, value); + } + + public void setValues(long totalRetrans, long retransmits, long rtt) { + set(Field.TOTAL_RETRANS, totalRetrans); + set(Field.RETRANSMITS, retransmits); + set(Field.RTT, rtt); + } + + long get(Field field) { + return values.getOrDefault(field, 0L); + } + + public int state() { + return (int) get(Field.STATE); + } + + public int caState() { + return (int) get(Field.CA_STATE); + } + + public int retransmits() { + return (int) get(Field.RETRANSMITS); + } + + public int probes() { + return (int) get(Field.PROBES); + } + + public int backoff() { + return (int) get(Field.BACKOFF); + } + + public int options() { + return (int) get(Field.OPTIONS); + } + + public int sndWscale() { + return (int) get(Field.SND_WSCALE); + } + + public int rcvWscale() { + return (int) get(Field.RCV_WSCALE); + } + + public long rto() { + return get(Field.RTO); + } + + public long ato() { + return get(Field.ATO); + } + + public long sndMss() { + return get(Field.SND_MSS); + } + + public long rcvMss() { + return get(Field.RCV_MSS); + } + + public long unacked() { + return get(Field.UNACKED); + } + + public long sacked() { + return get(Field.SACKED); + } + + public long lost() { + return get(Field.LOST); + } + + public long retrans() { + return get(Field.RETRANS); + } + + public long fackets() { + return get(Field.FACKETS); + } + + public long lastDataSent() { + return get(Field.LAST_DATA_SENT); + } + + public long lastAckSent() { + return get(Field.LAST_ACK_SENT); + } + + public long lastDataRecv() { + return get(Field.LAST_DATA_RECV); + } + + public long lastAckRecv() { + return get(Field.LAST_ACK_RECV); + } + + public long pmtu() { + return get(Field.PMTU); + } + + public long rcvSsthresh() { + return get(Field.RCV_SSTHRESH); + } + + public long rtt() { + return get(Field.RTT); + } + + public long rttvar() { + return get(Field.RTTVAR); + } + + public long sndSsthresh() { + return get(Field.SND_SSTHRESH); + } + + public long sndCwnd() { + return get(Field.SND_CWND); + } + + public long advmss() { + return get(Field.ADVMSS); + } + + public long reordering() { + return get(Field.REORDERING); + } + + public long totalRetrans() { + return get(Field.TOTAL_RETRANS); + } + + /** Mirrors {@code EpollSocketChannel.tcpInfo(EpollTcpInfo)}, copying from a source info. */ + public static final class FakeChannel extends EmbeddedChannel { + private final FakeEpollTcpInfo source; + + public FakeChannel(FakeEpollTcpInfo source) { + this.source = source; + } + + public void tcpInfo(FakeEpollTcpInfo info) { + info.values.clear(); + info.values.putAll(source.values); + } + } +} diff --git a/netty/src/test/java/io/grpc/netty/NettySocketSupportTest.java b/netty/src/test/java/io/grpc/netty/NettySocketSupportTest.java new file mode 100644 index 00000000000..4ec8b47dd43 --- /dev/null +++ b/netty/src/test/java/io/grpc/netty/NettySocketSupportTest.java @@ -0,0 +1,156 @@ +/* + * Copyright 2026 The gRPC Authors + * + * Licensed 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 io.grpc.netty; + +import static com.google.common.truth.Truth.assertThat; +import static com.google.common.truth.TruthJUnit.assume; + +import io.grpc.InternalChannelz.TcpInfo; +import io.grpc.netty.EpollTcpInfoAccessor.Field; +import io.grpc.netty.NettySocketSupport.NativeSocketOptions; +import io.grpc.netty.NettySocketSupport.NettySocketHelperImpl; +import io.netty.bootstrap.Bootstrap; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelException; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.channel.epoll.EpollIoHandler; +import io.netty.channel.epoll.EpollServerSocketChannel; +import io.netty.channel.epoll.EpollSocketChannel; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class NettySocketSupportTest { + private static final int TCP_ESTABLISHED = 1; + + @Test + public void epollSocketChannel_populatesAllTcpInfoFields() throws Exception { + FakeEpollTcpInfo source = new FakeEpollTcpInfo(); + long value = 1; + for (Field field : Field.values()) { + source.set(field, value++); + } + NettySocketHelperImpl helper = new NettySocketHelperImpl(FakeEpollTcpInfo.newAccessor()); + + NativeSocketOptions options = + helper.getNativeSocketOptions(new FakeEpollTcpInfo.FakeChannel(source)); + + assertThat(options.otherInfo).isEmpty(); + TcpInfo tcpInfo = options.tcpInfo; + assertThat(tcpInfo.state).isEqualTo((int) source.get(Field.STATE)); + assertThat(tcpInfo.caState).isEqualTo((int) source.get(Field.CA_STATE)); + assertThat(tcpInfo.retransmits).isEqualTo((int) source.get(Field.RETRANSMITS)); + assertThat(tcpInfo.probes).isEqualTo((int) source.get(Field.PROBES)); + assertThat(tcpInfo.backoff).isEqualTo((int) source.get(Field.BACKOFF)); + assertThat(tcpInfo.options).isEqualTo((int) source.get(Field.OPTIONS)); + assertThat(tcpInfo.sndWscale).isEqualTo((int) source.get(Field.SND_WSCALE)); + assertThat(tcpInfo.rcvWscale).isEqualTo((int) source.get(Field.RCV_WSCALE)); + assertThat(tcpInfo.rto).isEqualTo((int) source.get(Field.RTO)); + assertThat(tcpInfo.ato).isEqualTo((int) source.get(Field.ATO)); + assertThat(tcpInfo.sndMss).isEqualTo((int) source.get(Field.SND_MSS)); + assertThat(tcpInfo.rcvMss).isEqualTo((int) source.get(Field.RCV_MSS)); + assertThat(tcpInfo.unacked).isEqualTo((int) source.get(Field.UNACKED)); + assertThat(tcpInfo.sacked).isEqualTo((int) source.get(Field.SACKED)); + assertThat(tcpInfo.lost).isEqualTo((int) source.get(Field.LOST)); + assertThat(tcpInfo.retrans).isEqualTo((int) source.get(Field.RETRANS)); + assertThat(tcpInfo.fackets).isEqualTo((int) source.get(Field.FACKETS)); + assertThat(tcpInfo.lastDataSent).isEqualTo((int) source.get(Field.LAST_DATA_SENT)); + assertThat(tcpInfo.lastAckSent).isEqualTo((int) source.get(Field.LAST_ACK_SENT)); + assertThat(tcpInfo.lastDataRecv).isEqualTo((int) source.get(Field.LAST_DATA_RECV)); + assertThat(tcpInfo.lastAckRecv).isEqualTo((int) source.get(Field.LAST_ACK_RECV)); + assertThat(tcpInfo.pmtu).isEqualTo((int) source.get(Field.PMTU)); + assertThat(tcpInfo.rcvSsthresh).isEqualTo((int) source.get(Field.RCV_SSTHRESH)); + assertThat(tcpInfo.rtt).isEqualTo((int) source.get(Field.RTT)); + assertThat(tcpInfo.rttvar).isEqualTo((int) source.get(Field.RTTVAR)); + assertThat(tcpInfo.sndSsthresh).isEqualTo((int) source.get(Field.SND_SSTHRESH)); + assertThat(tcpInfo.sndCwnd).isEqualTo((int) source.get(Field.SND_CWND)); + assertThat(tcpInfo.advmss).isEqualTo((int) source.get(Field.ADVMSS)); + assertThat(tcpInfo.reordering).isEqualTo((int) source.get(Field.REORDERING)); + } + + @Test + public void nonEpollChannel_returnsNull() throws Exception { + NettySocketHelperImpl helper = new NettySocketHelperImpl(FakeEpollTcpInfo.newAccessor()); + + assertThat(helper.getNativeSocketOptions(new EmbeddedChannel())).isNull(); + } + + @Test + public void epollUnavailable_returnsNull() { + NettySocketHelperImpl helper = new NettySocketHelperImpl(null); + + assertThat(helper.getNativeSocketOptions(new EmbeddedChannel())).isNull(); + } + + @Test + public void closedChannel_tcpInfoFails_returnsNull() throws Exception { + NettySocketHelperImpl helper = new NettySocketHelperImpl( + EpollTcpInfoAccessor.create(ClosedFdChannel.class, FakeEpollTcpInfo.class)); + ClosedFdChannel channel = new ClosedFdChannel(); + channel.close(); + + assertThat(helper.getNativeSocketOptions(channel)).isNull(); + } + + @Test + public void realEpollSocketChannel_reportsEstablishedTcpInfo() throws Exception { + assume().that(Utils.isEpollAvailable()).isTrue(); + + EventLoopGroup group = new MultiThreadIoEventLoopGroup(1, EpollIoHandler.newFactory()); + try { + Channel server = new ServerBootstrap() + .group(group) + .channel(EpollServerSocketChannel.class) + .childHandler(new ChannelInboundHandlerAdapter()) + .bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0)) + .sync() + .channel(); + Channel client = new Bootstrap() + .group(group) + .channel(EpollSocketChannel.class) + .handler(new ChannelInboundHandlerAdapter()) + .connect(server.localAddress()) + .sync() + .channel(); + + TcpInfo tcpInfo = Utils.getSocketOptions(client).tcpInfo; + + assertThat(tcpInfo).isNotNull(); + assertThat(tcpInfo.state).isEqualTo(TCP_ESTABLISHED); + assertThat(tcpInfo.sndMss).isGreaterThan(0); + client.close().sync(); + server.close().sync(); + } finally { + group.shutdownGracefully(); + } + } + + public static final class ClosedFdChannel extends EmbeddedChannel { + public void tcpInfo(FakeEpollTcpInfo info) { + if (!isOpen()) { + throw new ChannelException("getsockopt() failed: Bad file descriptor"); + } + } + } +} diff --git a/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java b/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java index f75a98b46df..956fff77cd3 100644 --- a/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java +++ b/netty/src/test/java/io/grpc/netty/TcpMetricsTest.java @@ -57,19 +57,19 @@ public class TcpMetricsTest { @Mock private MetricRecorder metricRecorder; - private ConfigurableFakeWithTcpInfo channel; + private FakeEpollTcpInfo.FakeChannel channel; private TcpMetrics metrics; @Before public void setUp() throws Exception { FakeEpollTcpInfo dummyInfo = new FakeEpollTcpInfo(); - channel = new ConfigurableFakeWithTcpInfo(dummyInfo); + channel = new FakeEpollTcpInfo.FakeChannel(dummyInfo); metrics = new TcpMetrics(metricRecorder); } @After public void tearDown() throws Exception { - TcpMetrics.epollInfo = TcpMetrics.loadEpollInfo(); + TcpMetrics.epollInfo = EpollTcpInfoAccessor.INSTANCE; } @Test @@ -82,48 +82,15 @@ public void metricsInitialization() { assertNotNull(InternalTcpMetrics.MIN_RTT_INSTRUMENT); } - public static class FakeEpollTcpInfo { - long totalRetrans; - long retransmits; - long rtt; - - public void setValues(long totalRetrans, long retransmits, long rtt) { - this.totalRetrans = totalRetrans; - this.retransmits = retransmits; - this.rtt = rtt; - } - - @SuppressWarnings("unused") - public long totalRetrans() { - return totalRetrans; - } - - @SuppressWarnings("unused") - public long retrans() { - return retransmits; - } - - @SuppressWarnings("unused") - public long rtt() { - return rtt; - } - } - @Test public void tracker_recordTcpInfo_reflectionSuccess() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); channel.writeInbound("dummy"); tracker.channelInactive(channel); @@ -139,21 +106,33 @@ public void tracker_recordTcpInfo_reflectionSuccess() throws Exception { eq(0.005), any(), any()); } + @Test + public void tracker_recurringRetransmits_usesRetransmitsNotRetrans() throws Exception { + MetricRecorder recorder = mock(MetricRecorder.class); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); + TcpMetrics tracker = new TcpMetrics(recorder); + + FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); + infoSource.setValues(123, 4, 5000); + infoSource.set(EpollTcpInfoAccessor.Field.RETRANS, 9); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); + + tracker.channelInactive(channel); + + verify(recorder).addLongCounter( + eq(Objects.requireNonNull(InternalTcpMetrics.RECURRING_RETRANSMITS_INSTRUMENT)), + eq(4L), any(), any()); + } + @Test public void tracker_periodicRecord_doesNotRecordRecurringRetransmits() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); tracker.channelActive(channel); @@ -180,18 +159,12 @@ public void tracker_periodicRecord_doesNotRecordRecurringRetransmits() throws Ex @Test public void tracker_channelInactive_recordsRecurringRetransmits_raw_notDelta() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); tracker.channelActive(channel); @@ -207,7 +180,7 @@ public void tracker_channelInactive_recordsRecurringRetransmits_raw_notDelta() t // Let's just create a new channel instance where tcpInfo sets retrans=5. FakeEpollTcpInfo infoSource2 = new FakeEpollTcpInfo(); infoSource2.setValues(130, 5, 5000); - ConfigurableFakeWithTcpInfo channel2 = new ConfigurableFakeWithTcpInfo(infoSource2); + FakeEpollTcpInfo.FakeChannel channel2 = new FakeEpollTcpInfo.FakeChannel(infoSource2); tracker.channelInactive(channel2); @@ -224,18 +197,12 @@ public void tracker_channelInactive_recordsRecurringRetransmits_raw_notDelta() t @Test public void tracker_periodicRecord_reportsDeltaForTotalRetrans() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); tracker.channelActive(channel); @@ -278,18 +245,12 @@ public void tracker_periodicRecord_reportsDeltaForTotalRetrans() throws Exceptio @Test public void tracker_periodicRecord_doesNotReportZeroDeltaForTotalRetrans() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); infoSource.setValues(123, 4, 5000); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); tracker.channelActive(channel); @@ -326,21 +287,6 @@ public void tracker_periodicRecord_doesNotReportZeroDeltaForTotalRetrans() throw eq(0.005), any(), any()); } - public static class ConfigurableFakeWithTcpInfo extends - io.netty.channel.embedded.EmbeddedChannel { - private final FakeEpollTcpInfo infoToCopy; - - public ConfigurableFakeWithTcpInfo(FakeEpollTcpInfo infoToCopy) { - this.infoToCopy = infoToCopy; - } - - public void tcpInfo(FakeEpollTcpInfo info) { - info.totalRetrans = infoToCopy.totalRetrans; - info.retransmits = infoToCopy.retransmits; - info.rtt = infoToCopy.rtt; - } - } - private static class AddressOverrideEmbeddedChannel extends io.netty.channel.embedded.EmbeddedChannel { private final SocketAddress local; @@ -366,17 +312,11 @@ public SocketAddress remoteAddress() { public void tracker_reportsDeltas_correctly() throws Exception { MetricRecorder recorder = mock(MetricRecorder.class); - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); TcpMetrics tracker = new TcpMetrics(recorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); // 10 retransmits total infoSource.setValues(10, 2, 1000); @@ -552,18 +492,12 @@ public void channelInactive_decrementsCount_noEpoll_noError() { @Test public void channelActive_schedulesReportTimer() throws Exception { - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); metrics = new TcpMetrics(metricRecorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); metrics.channelActive(channel); @@ -591,18 +525,12 @@ public void channelActive_schedulesReportTimer() throws Exception { @Test public void channelInactive_cancelsReportTimer() throws Exception { - TcpMetrics.epollInfo = new TcpMetrics.EpollInfo( - ConfigurableFakeWithTcpInfo.class, - FakeEpollTcpInfo.class.getConstructor(), - ConfigurableFakeWithTcpInfo.class.getMethod("tcpInfo", FakeEpollTcpInfo.class), - FakeEpollTcpInfo.class.getMethod("totalRetrans"), - FakeEpollTcpInfo.class.getMethod("retrans"), - FakeEpollTcpInfo.class.getMethod("rtt")); + TcpMetrics.epollInfo = FakeEpollTcpInfo.newAccessor(); metrics = new TcpMetrics(metricRecorder); FakeEpollTcpInfo infoSource = new FakeEpollTcpInfo(); - ConfigurableFakeWithTcpInfo channel = new ConfigurableFakeWithTcpInfo(infoSource); + FakeEpollTcpInfo.FakeChannel channel = new FakeEpollTcpInfo.FakeChannel(infoSource); metrics.channelActive(channel);