Solve inconsistency issues in client gametests

fabric-client-gametest-api-v1/src/client/java/net/fabricmc/fabric/api/client/gametest/v1/package-info.java

@@ -19,8 +19,18 @@
  * {@link net.fabricmc.fabric.api.client.gametest.v1.ClientGameTestContext#waitTick() ClientGameTestContext.waitTick()}.
  *
  * <p>A few changes have been made to how the vanilla game threads run, to make tests more reproducible. Notably, there
- * is exactly one server tick per client tick while a server is running (singleplayer or multiplayer). On singleplayer,
- * packets will always arrive on a consistent tick.
+ * is exactly one server tick per client tick while a server is running (singleplayer or multiplayer). There is also a
+ * limit of one client tick per frame.
+ *
+ * <h1>Network synchronization</h1>
+ *
+ * <p>Network packets are internally tracked and managed so that they are always handled at a consistent time, always
+ * before the next tick. Calling {@code waitTick()} is always enough for a server packet to be handled on the client or
+ * vice versa.
+ *
+ * <p>If your mod interacts with the network code at a low level, such as by directly hooking into the Netty pipeline to
+ * send or handle packets, you may need to disable network synchronization. You can do this by setting the
+ * {@code fabric.client.gametest.disableNetworkSynchronizer} system property.
  *
  * <h1>Default settings</h1>
  * The client gametest API adjusts some default settings, usually for consistency of tests. These settings can always be

fabric-client-gametest-api-v1/src/client/java/net/fabricmc/fabric/impl/client/gametest/NetworkSynchronizer.java

@@ -0,0 +1,210 @@
+/*
+ * Copyright (c) 2016, 2017, 2018, 2019 FabricMC
+ *
+ * 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 net.fabricmc.fabric.impl.client.gametest;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import com.google.common.collect.ConcurrentHashMultiset;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import net.minecraft.util.Unit;
+import net.minecraft.util.thread.ThreadExecutor;
+
+/**
+ * Ensures packets are always handled by the end of the task loop on the receiving thread.
+ *
+ * <h1>Implementation notes</h1>
+ *
+ * <ul>
+ *     <li>A packet can be either "in-flight", which is between the time it is sent and the time it is handled on the
+ *         Netty thread on the receiving side, or it can be queued for handling on the receiving main thread, which it
+ *         between when it is handled on the Netty thread and it is removed from the main thread task queue.
+ *         <ul>
+ *             <li>Some packets are handled directly on the Netty thread and never enter the second stage. The
+ *                 {@code NetworkSynchronizer} is careful not to assume that all packets must be added to the task
+ *                 queue.</li>
+ *         </ul>
+ *     </li>
+ *     <li>Once the packets are tracked in this way, the key change is that the client and server now continue running
+ *         their task loops until there are no in-flight packets and no packets waiting to be handled in the vanilla
+ *         task queues.</li>
+ *     <li>Network synchronization can be disabled via a system property, which is useful for mods which directly
+ *         interface with the Netty pipeline, which would desynchronize the in-flight packet counter.</li>
+ * </ul>
+ */
+public final class NetworkSynchronizer {
+	private static final Logger LOGGER = LoggerFactory.getLogger("fabric-client-gametest-api-v1");
+	public static final boolean DISABLED = System.getProperty("fabric.client.gametest.disableNetworkSynchronizer") != null;
+
+	public static final NetworkSynchronizer CLIENTBOUND = new NetworkSynchronizer();
+	public static final NetworkSynchronizer SERVERBOUND = new NetworkSynchronizer();
+
+	private final ThreadLocal<Unit> isNettyThread = new ThreadLocal<>();
+	private final AtomicInteger inFlightPackets = new AtomicInteger();
+	private final ConcurrentHashMultiset<RunnableBox> mainThreadPacketHandlers = ConcurrentHashMultiset.create();
+	private final Lock morePacketsLock = new ReentrantLock();
+	private final Condition morePacketsCondition = morePacketsLock.newCondition();
+	private final AtomicBoolean invalid = new AtomicBoolean();
+	private boolean isRunningNetworkTasks = false;
+
+	public void preSendPacket() {
+		if (DISABLED) {
+			return;
+		}
+
+		inFlightPackets.incrementAndGet();
+	}
+
+	public void preNettyHandlePacket() {
+		if (DISABLED) {
+			return;
+		}
+
+		isNettyThread.set(Unit.INSTANCE);
+	}
+
+	public void postNettyHandlePacket() {
+		if (DISABLED) {
+			return;
+		}
+
+		int remainingInFlightPackets = inFlightPackets.decrementAndGet();
+
+		if (remainingInFlightPackets < 0) {
+			markInvalid();
+			return;
+		}
+
+		isNettyThread.remove();
+
+		if (remainingInFlightPackets == 0) {
+			signalMorePackets();
+		}
+	}
+
+	public void preTaskAdded(Runnable task) {
+		if (DISABLED) {
+			return;
+		}
+
+		if (isNettyThread.get() != null) {
+			mainThreadPacketHandlers.add(new RunnableBox(task));
+			signalMorePackets();
+		}
+	}
+
+	public void postTaskRun(Runnable task) {
+		if (DISABLED) {
+			return;
+		}
+
+		checkInvalid();
+		mainThreadPacketHandlers.remove(new RunnableBox(task));
+	}
+
+	public void waitForPacketHandlers(ThreadExecutor<?> executor) {
+		if (DISABLED) {
+			return;
+		}
+
+		while (inFlightPackets.get() > 0 || !mainThreadPacketHandlers.isEmpty()) {
+			while (inFlightPackets.get() > 0 && mainThreadPacketHandlers.isEmpty()) {
+				morePacketsLock.lock();
+
+				try {
+					if (!morePacketsCondition.await(10, TimeUnit.SECONDS)) {
+						markInvalid();
+						checkInvalid();
+					}
+				} catch (InterruptedException e) {
+					throw new RuntimeException(e);
+				} finally {
+					morePacketsLock.unlock();
+				}
+			}
+
+			isRunningNetworkTasks = true;
+
+			long startTime = System.nanoTime();
+
+			try {
+				executor.runTasks(() -> {
+					if (System.nanoTime() - startTime > 10_000_000_000L) {
+						markInvalid();
+						checkInvalid();
+					}
+
+					return mainThreadPacketHandlers.isEmpty();
+				});
+			} finally {
+				isRunningNetworkTasks = false;
+			}
+		}
+	}
+
+	public void reset() {
+		inFlightPackets.set(0);
+		mainThreadPacketHandlers.clear();
+		signalMorePackets();
+	}
+
+	public boolean isRunningNetworkTasks() {
+		return isRunningNetworkTasks;
+	}
+
+	private void signalMorePackets() {
+		morePacketsLock.lock();
+		morePacketsCondition.signal();
+		morePacketsLock.unlock();
+	}
+
+	private void markInvalid() {
+		if (!invalid.getAndSet(true)) {
+			LOGGER.error("Detected interfacing with packets at a lower level. Please disable network synchronization by setting the fabric.client.gametest.disableNetworkSynchronizer system property");
+			signalMorePackets();
+		}
+	}
+
+	private void checkInvalid() {
+		if (invalid.get()) {
+			throw new AssertionError("Network synchronizer in invalid state, see earlier log messages");
+		}
+	}
+
+	// Wraps a runnable to always use identity hashCode and equals
+	private record RunnableBox(Runnable runnable) {
+		@Override
+		public boolean equals(Object other) {
+			if (!(other instanceof RunnableBox(Runnable otherRunnable))) {
+				return false;
+			}
+
+			return otherRunnable == this.runnable;
+		}
+
+		@Override
+		public int hashCode() {
+			return System.identityHashCode(this.runnable);
+		}
+	}
+}

fabric-client-gametest-api-v1/src/client/java/net/fabricmc/fabric/impl/client/gametest/ThreadingImpl.java

@@ -49,13 +49,13 @@
  * released while leaving {@linkplain #taskToRun} as {@code null}, which they will interpret to mean they are to
  * continue into {@linkplain #PHASE_TICK}.
  *
- * <p>The reason these phases were chosen are to make client-server interaction in singleplayer as consistent as
- * possible. The task queues are when most packets are handled, and without them being run in sequence it would be
- * unspecified whether a packet would be handled on the current tick until the next one. The server task queue is before
- * the client so that changes on the server appear on the client more readily. The test phase is run after the task
- * queues rather than at the end of the physical tick (i.e. {@code MinecraftClient}'s and {@code MinecraftServer}'s
- * {@code tick} methods), for no particular reason other than to avoid needing a 5th phase, and having a power of 2
- * number of phases is convenient when using {@linkplain Phaser}, as it doesn't break when the phase counter overflows.
+ * <p>The reason these phases were chosen are to make client-server communication as consistent as possible. The task
+ * queues are when most packets are handled, and without them being run in sequence it would be unspecified whether a
+ * packet would be handled on the current tick until the next one. The server task queue is before the client so that
+ * changes on the server appear on the client more readily. The test phase is run after the task queues rather than at
+ * the end of the physical tick (i.e. {@code MinecraftClient}'s and {@code MinecraftServer}'s {@code tick} methods), for
+ * no particular reason other than to avoid needing a 5th phase, and having a power of 2 number of phases is convenient
+ * when using {@linkplain Phaser}, as it doesn't break when the phase counter overflows.
  *
  * <p>Other challenges include that a client or server can be started during {@linkplain #PHASE_TEST} but haven't
  * reached their semaphore code yet meaning they are unable to accept tasks. This is solved by setting a flag to true
@@ -104,7 +104,7 @@ private ThreadingImpl() {
 	private static volatile boolean gameCrashed = false;
 
 	public static void enterPhase(int phase) {
-		while (enablePhases && (PHASER.getPhase() & PHASE_MASK) != phase) {
+		while (enablePhases && getNextPhase() != phase) {
 			PHASER.arriveAndAwaitAdvance();
 		}
 
@@ -113,6 +113,18 @@ public static void enterPhase(int phase) {
 		}
 	}
 
+	public static int getCurrentPhase() {
+		return (getNextPhase() - 1) & PHASE_MASK;
+	}
+
+	private static int getNextPhase() {
+		return PHASER.getPhase() & PHASE_MASK;
+	}
+
+	public static boolean isGameCrashed() {
+		return gameCrashed;
+	}
+
 	public static void setGameCrashed() {
 		enablePhases = false;
 		gameCrashed = true;

fabric-client-gametest-api-v1/src/client/java/net/fabricmc/fabric/mixin/client/gametest/ClientConnectionMixin.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2016, 2017, 2018, 2019 FabricMC
+ *
+ * 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 net.fabricmc.fabric.mixin.client.gametest;
+
+import com.llamalad7.mixinextras.injector.wrapmethod.WrapMethod;
+import com.llamalad7.mixinextras.injector.wrapoperation.Operation;
+import io.netty.channel.ChannelHandlerContext;
+import org.spongepowered.asm.mixin.Final;
+import org.spongepowered.asm.mixin.Mixin;
+import org.spongepowered.asm.mixin.Shadow;
+import org.spongepowered.asm.mixin.injection.At;
+import org.spongepowered.asm.mixin.injection.Inject;
+import org.spongepowered.asm.mixin.injection.callback.CallbackInfo;
+
+import net.minecraft.network.ClientConnection;
+import net.minecraft.network.NetworkSide;
+import net.minecraft.network.packet.Packet;
+
+import net.fabricmc.fabric.impl.client.gametest.NetworkSynchronizer;
+
+@Mixin(ClientConnection.class)
+public class ClientConnectionMixin {
+	@Shadow
+	@Final
+	private NetworkSide side;
+
+	@WrapMethod(method = "channelRead0(Lio/netty/channel/ChannelHandlerContext;Lnet/minecraft/network/packet/Packet;)V")
+	private void onNettyReceivePacket(ChannelHandlerContext context, Packet<?> packet, Operation<Void> original) {
+		NetworkSynchronizer synchronizer = side == NetworkSide.CLIENTBOUND ? NetworkSynchronizer.CLIENTBOUND : NetworkSynchronizer.SERVERBOUND;
+		synchronizer.preNettyHandlePacket();
+
+		try {
+			original.call(context, packet);
+		} finally {
+			synchronizer.postNettyHandlePacket();
+		}
+	}
+
+	@Inject(method = "sendImmediately", at = @At("HEAD"))
+	private void onSendPacket(CallbackInfo ci) {
+		NetworkSynchronizer synchronizer = side == NetworkSide.CLIENTBOUND ? NetworkSynchronizer.SERVERBOUND : NetworkSynchronizer.CLIENTBOUND;
+		synchronizer.preSendPacket();
+	}
+}