java - Instrument Netty 的 EventLoop 用于确定性执行计划任务

Question

我想检测 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。

ScheduledFutureTaskHack.java

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 Framework 的作者，一个基于 WebSocket 的网络框架，具有纯 Java 客户端和非阻塞双因素身份验证。它经常使用 Netty。遗憾的是，有许多基于 JMockit 的测试不能可靠地运行，因为执行顺序可能是不确定的(这是每段调度任务的代码的问题)。

Answer 1

伙计们，这太明显了:我重写了每个 schedule* 方法，将 Runnable/Callable 和其他参数堆积在数据中对象，并将其添加到某个队列中。然后我从测试中显式触发任务执行。

因为创建任务的代码是我的，所以我用标记接口(interface)装饰每个任务(恰好是方法引用)。然后测试可以检查它运行预期的任务。