我想仪器化Netty的EventLoop以实现以下目标:
- 按确定顺序运行任务。 - 获取调度任务的截止时间。 - 快进虚拟时钟,触发已到期任务的执行。
我知道EmbeddedChannel并在一些测试中使用它。但我想要的是介于单元测试和集成测试之间的东西,可以忽略一些角落案例。断开连接重连和ping场景经常使用计划任务。我可以通过巨大的延迟添加确定性,但我不希望我的测试等待几秒钟或更长时间。因此,仪器化Netty的EventLoop似乎是解决方案。
我已经编写了代码,至少对我来说没有问题。
- 我修改ScheduledFutureTask#nanoTime以返回我的值。 - 我派生NioEventLoopGroup,以获取任务截止时间。 - 我更改ScheduledFutureTask#nanoTime返回的值。
很好的地方是Netty代码只依赖于ScheduledFutureTask#nanoTime返回的值(很棒的设计!),因此这是一个非常有限的更改。我使用ByteBuddy避免复制Netty代码,但这不重要。
像InstrumentedNioEventLoopGroupTest这样的非常简单的测试仅在调度1个任务时会失败,因为AbstractScheduledEventExecutor#pollScheduledTask(long)具有空队列。
我发现每个NioEventLoop都有自己的任务队列,并且可能不会发生队列轮询,因为NioEventLoopGroup等待Selector发出信号,这是有道理的。因此,我将NioEventLoopGroup的线程计数增加到2。我还尝试将ioRatio设置为1并安排更多任务,但结果并没有改善。使用调试器,似乎我的任务始终“掉入”未轮询的任务队列中。
有什么想法可以使这个工作?我正在使用Netty 4.1.24.Final。
- 按确定顺序运行任务。 - 获取调度任务的截止时间。 - 快进虚拟时钟,触发已到期任务的执行。
我知道EmbeddedChannel并在一些测试中使用它。但我想要的是介于单元测试和集成测试之间的东西,可以忽略一些角落案例。断开连接重连和ping场景经常使用计划任务。我可以通过巨大的延迟添加确定性,但我不希望我的测试等待几秒钟或更长时间。因此,仪器化Netty的EventLoop似乎是解决方案。
我已经编写了代码,至少对我来说没有问题。
- 我修改ScheduledFutureTask#nanoTime以返回我的值。 - 我派生NioEventLoopGroup,以获取任务截止时间。 - 我更改ScheduledFutureTask#nanoTime返回的值。
很好的地方是Netty代码只依赖于ScheduledFutureTask#nanoTime返回的值(很棒的设计!),因此这是一个非常有限的更改。我使用ByteBuddy避免复制Netty代码,但这不重要。
像InstrumentedNioEventLoopGroupTest这样的非常简单的测试仅在调度1个任务时会失败,因为AbstractScheduledEventExecutor#pollScheduledTask(long)具有空队列。
我发现每个NioEventLoop都有自己的任务队列,并且可能不会发生队列轮询,因为NioEventLoopGroup等待Selector发出信号,这是有道理的。因此,我将NioEventLoopGroup的线程计数增加到2。我还尝试将ioRatio设置为1并安排更多任务,但结果并没有改善。使用调试器,似乎我的任务始终“掉入”未轮询的任务队列中。
有什么想法可以使这个工作?我正在使用Netty 4.1.24.Final。
package com.otcdlink.chiron.integration.harness;
import com.otcdlink.chiron.toolbox.ToStringTools;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.agent.ByteBuddyAgent;
import net.bytebuddy.dynamic.loading.ClassReloadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.concurrent.Future;
import java.util.function.LongSupplier;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static net.bytebuddy.matcher.ElementMatchers.isPackagePrivate;
import static net.bytebuddy.matcher.ElementMatchers.isStatic;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.returns;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
/**
*
* Got the delegation working with the help of
* https://www.infoq.com/articles/Easily-Create-Java-Agents-with-ByteBuddy
*/
final class ScheduledFutureTaskHack {
private static final Logger LOGGER = LoggerFactory.getLogger( ScheduledFutureTaskHack.class ) ;
private static final Class< ? > SCHEDULEDFUTURETASK_CLASS ;
private static final Method SCHEDULEDFUTURETASK_NANOTIME_METHOD ;
private static final Method SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD ;
private static final Field SCHEDULEDFUTURETASK_DEADLINENANOS_FIELD ;
private static final Field SCHEDULEDFUTURETASK_STARTTIME_FIELD ;
static {
try {
SCHEDULEDFUTURETASK_CLASS = Class.forName( "io.netty.util.concurrent.ScheduledFutureTask" ) ;
SCHEDULEDFUTURETASK_NANOTIME_METHOD =
SCHEDULEDFUTURETASK_CLASS.getDeclaredMethod( "nanoTime" ) ;
SCHEDULEDFUTURETASK_NANOTIME_METHOD.setAccessible( true ) ;
SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD =
SCHEDULEDFUTURETASK_CLASS.getDeclaredMethod( "deadlineNanos") ;
SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD.setAccessible( true ) ;
SCHEDULEDFUTURETASK_DEADLINENANOS_FIELD =
SCHEDULEDFUTURETASK_CLASS.getDeclaredField( "deadlineNanos" ) ;
SCHEDULEDFUTURETASK_DEADLINENANOS_FIELD.setAccessible( true ) ;
SCHEDULEDFUTURETASK_STARTTIME_FIELD =
SCHEDULEDFUTURETASK_CLASS.getDeclaredField( "START_TIME" ) ;
SCHEDULEDFUTURETASK_STARTTIME_FIELD.setAccessible( true ) ;
} catch( ClassNotFoundException | NoSuchMethodException | NoSuchFieldException e ) {
throw new Error( e ) ;
}
}
/**
* Everything is this class must be visible from the redefined class.
*/
@SuppressWarnings( "unused" )
public static final class StaticMethodDelegate {
/**
* Calls to {@link io.netty.util.concurrent.ScheduledFutureTask#nanoTime()} are redirected
* to this method.
* Sadly we can't use parameter annotated with {@link @This} or something giving a hint
* about the call context. It looks like a consequence of JVMTI reload not supporting method
* addition (adding a parameter would imply creating a new method).
*/
public static long nanoTime() {
final long supplied = longSupplier.getAsLong() ;
LOGGER.debug( "Called " + StaticMethodDelegate.class.getSimpleName() + "#nanoTime(), " +
"returns " + supplied + "." ) ;
return supplied ;
}
}
private static LongSupplier longSupplier = null ;
static void install( final LongSupplier longSupplier ) {
install( longSupplier, true ) ;
}
/**
*
* @param longSupplier
* @param suppliedNanosRelativeToClassloadingTime if {@code true}, supplied nanoseconds are
* relative to {@link io.netty.util.concurrent.ScheduledFutureTask#START_TIME}.
* Original behavior of the hacked method is to substract
* {@link io.netty.util.concurrent.ScheduledFutureTask#START_TIME} from value returned
* by {@link System#nanoTime()} (probably to make number more readable and reduce the risk
* of an overflow). During tests we prefer to not care about start time so there is this
* option to add it automatically.
*/
static void install(
final LongSupplier longSupplier,
final boolean suppliedNanosRelativeToClassloadingTime
) {
checkState( ScheduledFutureTaskHack.longSupplier == null ) ;
if( suppliedNanosRelativeToClassloadingTime ) {
final long startTime = START_TIME ;
LOGGER.debug(
"Installing with value of " +
SCHEDULEDFUTURETASK_STARTTIME_FIELD.toGenericString() +
" = " + startTime + " automatically added to the values supplied."
) ;
class AdjustedLongSupplier implements LongSupplier {
@Override
public long getAsLong() {
return longSupplier.getAsLong() + startTime ;
}
@Override
public String toString() {
return ToStringTools.getNiceClassName( this ) + "{startTime=" + startTime + "}" ;
}
}
ScheduledFutureTaskHack.longSupplier = new AdjustedLongSupplier() ;
} else {
ScheduledFutureTaskHack.longSupplier = checkNotNull( longSupplier ) ;
}
ByteBuddyAgent.install() ;
LOGGER.info( "Successfully installed ByteBuddy Agent." ) ;
redefineClass() ;
LOGGER.info( "Successfully redefined static method implementation." ) ;
}
private static void redefineClass() {
new ByteBuddy()
.redefine( SCHEDULEDFUTURETASK_CLASS )
.method( named( "nanoTime" )
.and( isStatic() )
.and( isPackagePrivate() )
.and( takesArguments( 0 ) )
.and( returns( long.class ) )
)
.intercept( MethodDelegation.to( StaticMethodDelegate.class ) )
.make()
.load( ScheduledFutureTaskHack.class.getClassLoader(), ClassReloadingStrategy.fromInstalledAgent() )
;
}
/**
* Invokes method replacing {@link io.netty.util.concurrent.ScheduledFutureTask#nanoTime()}.
*/
public static long invokeNanoTime() {
try {
return ( long ) SCHEDULEDFUTURETASK_NANOTIME_METHOD.invoke( null ) ;
} catch( IllegalAccessException | InvocationTargetException e ) {
throw new Error( e ) ;
}
}
/**
* The {@link io.netty.util.concurrent.ScheduledFutureTask#deadlineNanos()} method returns
* the value made from {@link System#nanoTime()},
* minus {@link io.netty.util.concurrent.ScheduledFutureTask#START_TIME},
* plus the delay before executing the task.
*/
public static Long invokeDeadlineNanos( final Future future ) {
try {
if( SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD.getDeclaringClass()
.isAssignableFrom( future.getClass() )
) {
return ( long ) SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD.invoke( future ) ;
} else {
return null ;
}
} catch( IllegalAccessException | InvocationTargetException e ) {
throw new Error(
"Could not access method " + SCHEDULEDFUTURETASK_DEADLINENANOS_METHOD + " in " + future,
e
) ;
}
}
private static long readStartTime() {
try {
return ( long ) SCHEDULEDFUTURETASK_STARTTIME_FIELD.get( null ) ;
} catch( IllegalAccessException e ) {
throw new Error(
"Could not access static field " + SCHEDULEDFUTURETASK_STARTTIME_FIELD,
e
) ;
}
}
public static final long START_TIME = readStartTime() ;
}
ScheduledFutureTaskHackTest.java
package com.otcdlink.chiron.integration.harness;
import com.otcdlink.chiron.toolbox.ToStringTools;
import com.otcdlink.chiron.toolbox.netty.NettyTools;
import io.netty.channel.nio.NioEventLoopGroup;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static org.assertj.core.api.Assertions.assertThat;
public class ScheduledFutureTaskHackTest {
@Test
public void fastForward() throws InterruptedException {
final AtomicLong nanotimeHolder = new AtomicLong( 0 ) ;
ScheduledFutureTaskHack.install( nanotimeHolder::get ) ;
final long startTime = hackedNanoTime() ;
final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup() ;
final Semaphore scheduledTaskCompleted = new Semaphore( 0 ) ;
nioEventLoopGroup.schedule(
() -> {
scheduledTaskCompleted.release() ;
LOGGER.info( "Scheduled task completed." ) ;
},
1,
TimeUnit.HOURS
) ;
LOGGER.info( "Scheduled task for in 1 hour, now fast-forwarding Netty's clock ..." ) ;
// Test fails when disabling fast-forward below.
nanotimeHolder.set( startTime + TimeUnit.HOURS.toNanos( 1 ) + 1 ) ;
Thread.sleep( 1000 ) ;
hackedNanoTime() ;
// Amazingly Netty detected clock change and ran the task!
assertThat( scheduledTaskCompleted.tryAcquire( 1, TimeUnit.SECONDS ) )
.describedAs( "Scheduled task should have completed within 1 second" )
.isTrue()
;
}
// =======
// Fixture
// =======
private static final Logger LOGGER = LoggerFactory.getLogger(
ScheduledFutureTaskHackTest.class ) ;
static {
NettyTools.forceNettyClassesToLoad() ;
}
private static long hackedNanoTime() {
final long nanoTime = ScheduledFutureTaskHack.invokeNanoTime() ;
LOGGER.info(
ToStringTools.getNiceName( ScheduledFutureTaskHack.StaticMethodDelegate.class ) +
"#nanoTime(): " + nanoTime + "."
) ;
return nanoTime ;
}
}
InstrumentedNioEventLoopGroup.java
package com.otcdlink.chiron.integration.harness;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.ScheduledFuture;
import javax.annotation.Nonnull;
import java.time.Instant;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import static com.google.common.base.Preconditions.checkNotNull;
class InstrumentedNioEventLoopGroup extends NioEventLoopGroup {
/**
* Consume the value obtained from
* {@link io.netty.util.concurrent.ScheduledFutureTask#deadlineNanos()}.
* This is hardly mappable to an exact {@link Instant} (even if the Java flavor retains
* nanoseconds) but this is enough to compare with {@link System#nanoTime()}.
*/
private final Consumer< Long > scheduledTaskMomentConsumer ;
public InstrumentedNioEventLoopGroup(
final ThreadFactory threadFactory,
final Consumer< Long > scheduledTaskMomentConsumer
) {
// Need 2 threads because one will block on Socket Selector if there is no IO,
// so we add one to poll Tasks.
super( 2, threadFactory ) ;
this.scheduledTaskMomentConsumer = checkNotNull( scheduledTaskMomentConsumer ) ;
}
private < FUTURE extends Future > FUTURE recordDeadlineNanos( final FUTURE future ) {
final Long deadlineNanos = ScheduledFutureTaskHack.invokeDeadlineNanos( future ) ;
if( deadlineNanos != null ) {
scheduledTaskMomentConsumer.accept( deadlineNanos ) ;
}
return future ;
}
@Nonnull
@Override
public Future< ? > submit( final Runnable task ) {
return recordDeadlineNanos( super.submit( task ) ) ;
}
@Nonnull
@Override
public < T > Future< T > submit(
final Runnable task,
final T result
) {
return recordDeadlineNanos( super.submit( task, result ) ) ;
}
@Nonnull
@Override
public < T > Future< T > submit( final Callable< T > task ) {
return recordDeadlineNanos( super.submit( task ) ) ;
}
@Nonnull
@Override
public ScheduledFuture< ? > schedule(
final Runnable command,
final long delay,
final TimeUnit unit
) {
return recordDeadlineNanos( super.schedule( command, delay, unit ) ) ;
}
@Nonnull
@Override
public < V > ScheduledFuture< V > schedule(
final Callable< V > callable,
final long delay,
final TimeUnit unit
) {
return recordDeadlineNanos( super.schedule( callable, delay, unit ) ) ;
}
@Nonnull
@Override
public ScheduledFuture< ? > scheduleAtFixedRate(
final Runnable command,
final long initialDelay,
final long period,
final TimeUnit unit
) {
return recordDeadlineNanos(
super.scheduleAtFixedRate( command, initialDelay, period, unit ) ) ;
}
@Nonnull
@Override
public ScheduledFuture< ? > scheduleWithFixedDelay(
final Runnable command,
final long initialDelay,
final long delay,
final TimeUnit unit
) {
return recordDeadlineNanos(
super.scheduleWithFixedDelay( command, initialDelay, delay, unit ) ) ;
}
}
InstrumentedNioEventLoopGroupTest.java
package com.otcdlink.chiron.integration.harness;
import com.otcdlink.chiron.toolbox.concurrent.ExecutorTools;
import com.otcdlink.chiron.toolbox.netty.NettyTools;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.ScheduledFuture;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static org.assertj.core.api.Assertions.assertThat;
public class InstrumentedNioEventLoopGroupTest {
@Test
public void recordAndAdjust() throws InterruptedException {
final int delay = 10 ;
final TimeUnit timeUnit = TimeUnit.SECONDS ;
final AtomicLong nanoInstantSupplier = new AtomicLong() ;
ScheduledFutureTaskHack.install( nanoInstantSupplier::get ) ;
final List< Long > taskDeadlineRecorder = Collections.synchronizedList( new ArrayList<>() ) ;
final InstrumentedNioEventLoopGroup executor = new InstrumentedNioEventLoopGroup(
ExecutorTools.newThreadFactory( "executor" ), taskDeadlineRecorder::add ) ;
executor.setIoRatio( 1 ) ; // Silly but worth trying to see what can get wrong.
final Semaphore doneSemaphore = new Semaphore( 0 ) ;
final ScheduledFuture< ? > scheduledFuture1 =
executor.schedule( ( Runnable ) doneSemaphore::release, delay, timeUnit ) ;
LOGGER.info( "Scheduled " + scheduledFuture1 + "." ) ;
assertThat( taskDeadlineRecorder ).hasSize( 1 ) ;
final Long nanoTime = taskDeadlineRecorder.get( 0 ) - ScheduledFutureTaskHack.START_TIME ;
LOGGER.info( "Recorded " + nanoTime + " as nanoTime deadline for next task." ) ;
assertThat( nanoTime ).isEqualTo( timeUnit.toNanos( delay ) ) ;
final long pastDeadline = nanoTime + 1 ;
nanoInstantSupplier.set( pastDeadline ) ;
LOGGER.info(
"Did set nanoTime to " + pastDeadline + ", past to Task's deadline. " +
"Invocation of hacked nanoTime() returns " +
ScheduledFutureTaskHack.invokeNanoTime() + "."
) ;
LOGGER.info( "Now waiting for task completion ..." ) ;
assertThat( doneSemaphore.tryAcquire( 3, TimeUnit.SECONDS ) ).isTrue() ;
}
/**
* Fails when ran after {@link #recordAndAdjust()} because JUnit doesn't reload classes for
* each method inside a test class.
*/
@Test
public void noInstrumentation() throws InterruptedException {
final NioEventLoopGroup executor =
new NioEventLoopGroup( 1, ExecutorTools.newThreadFactory( "executor" ) ) ;
final Semaphore doneSemaphore = new Semaphore( 0 ) ;
executor.submit( () -> LOGGER.info( "Plain submission works!" ) ) ;
final ScheduledFuture< ? > scheduledFuture =
executor.schedule( ( Runnable ) doneSemaphore::release, 1, TimeUnit.SECONDS ) ;
LOGGER.info( "Scheduled " + scheduledFuture + "." ) ;
assertThat( doneSemaphore.tryAcquire( 3, TimeUnit.SECONDS ) ).isTrue() ;
}
// =======
// Fixture
// =======
private static final Logger LOGGER =
LoggerFactory.getLogger( InstrumentedNioEventLoopGroupTest.class ) ;
static {
NettyTools.forceNettyClassesToLoad() ;
}
}
我是Chiron框架的作者,这是一个基于WebSocket的网络框架,具有纯Java客户端和非阻塞式双因素身份验证。它广泛使用Netty。不幸的是,有许多基于JMockit的测试无法可靠地运行,因为执行顺序可能是不确定的(这是所有安排任务的代码都存在的问题)。