[#412] Add async packet listener without packet reordering

orebfuscator-plugin/pom.xml

@@ -67,6 +67,12 @@
 			<version>${dependency.netty.version}</version>
 			<scope>provided</scope>
 		</dependency>
+		<dependency>
+			<groupId>io.netty</groupId>
+			<artifactId>netty-transport</artifactId>
+			<version>${dependency.netty.version}</version>
+			<scope>provided</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.spigotmc</groupId>
 			<artifactId>spigot-api</artifactId>

orebfuscator-plugin/src/main/java/net/imprex/orebfuscator/Orebfuscator.java

@@ -13,6 +13,7 @@
 import net.imprex.orebfuscator.api.OrebfuscatorService;
 import net.imprex.orebfuscator.cache.ObfuscationCache;
 import net.imprex.orebfuscator.config.OrebfuscatorConfig;
+import net.imprex.orebfuscator.injector.OrebfuscatorInjectorManager;
 import net.imprex.orebfuscator.obfuscation.ObfuscationSystem;
 import net.imprex.orebfuscator.player.OrebfuscatorPlayerMap;
 import net.imprex.orebfuscator.proximity.ProximityDirectorThread;
@@ -33,6 +34,7 @@ public class Orebfuscator extends JavaPlugin implements Listener {
 	private UpdateSystem updateSystem;
 	private ObfuscationCache obfuscationCache;
 	private ObfuscationSystem obfuscationSystem;
+	private OrebfuscatorInjectorManager injectorManager;
 	private ProximityDirectorThread proximityThread;
 	private ProximityPacketListener proximityPacketListener;
 
@@ -87,6 +89,7 @@ public void onEnable() {
 
 			// Load packet listener
 			this.obfuscationSystem.registerChunkListener();
+			this.injectorManager = new OrebfuscatorInjectorManager(this);
 
 			// Store formatted config
 			this.config.store();
@@ -122,6 +125,10 @@ public void onDisable() {
 			this.proximityThread.close();
 		}
 
+		if (this.injectorManager != null) {
+			this.injectorManager.close();
+		}
+
 		OrebfuscatorCompatibility.close();
 		OrebfuscatorNms.close();
 

orebfuscator-plugin/src/main/java/net/imprex/orebfuscator/OrebfuscatorStatistics.java

@@ -1,13 +1,18 @@
 package net.imprex.orebfuscator;
 
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.function.DoubleSupplier;
 import java.util.function.LongSupplier;
 
 import com.google.gson.JsonObject;
 
 public class OrebfuscatorStatistics {
 
+    private static final long MICRO_SCALE  = 1000L;
+    private static final long MILLI_SCALE  = 1000L * MICRO_SCALE;
+    private static final long SECOND_SCALE = 1000L * MILLI_SCALE;
+
 	private static String formatPrecent(double percent) {
 		return String.format("%.2f%%", percent * 100);
 	}
@@ -24,28 +29,41 @@ private static String formatBytes(long bytes) {
 		}
 	}
 
-	private final AtomicLong cacheHitCountMemory = new AtomicLong(0);
-	private final AtomicLong cacheHitCountDisk = new AtomicLong(0);
-	private final AtomicLong cacheMissCount = new AtomicLong(0);
-	private final AtomicLong cacheEstimatedSize = new AtomicLong(0);
+	private static String formatNanos(double nanos) {
+		if (nanos > SECOND_SCALE) {
+			return String.format("%.2f s", nanos / SECOND_SCALE);
+		} else if (nanos > MILLI_SCALE) {
+			return String.format("%.2f ms", nanos / MILLI_SCALE);
+		} else if (nanos > MICRO_SCALE) {
+			return String.format("%.2f µs", nanos / MICRO_SCALE);
+		} else {
+			return String.format("%d ns", nanos);
+		}
+	}
+
+	private final LongAdder cacheHitCountMemory = new LongAdder();
+	private final LongAdder cacheHitCountDisk = new LongAdder();
+	private final LongAdder cacheMissCount = new LongAdder();
+	private final LongAdder cacheEstimatedSize = new LongAdder();
 	private LongSupplier memoryCacheSize = () -> 0;
 	private LongSupplier diskCacheQueueLength = () -> 0;
 	private LongSupplier obfuscationQueueLength = () -> 0;
+	private DoubleSupplier averagePacketDelay = () -> 0;
 
 	public void onCacheHitMemory() {
-		this.cacheHitCountMemory.incrementAndGet();
+		this.cacheHitCountMemory.increment();
 	}
 
 	public void onCacheHitDisk() {
-		this.cacheHitCountDisk.incrementAndGet();
+		this.cacheHitCountDisk.increment();
 	}
 
 	public void onCacheMiss() {
-		this.cacheMissCount.incrementAndGet();
+		this.cacheMissCount.increment();
 	}
 
 	public void onCacheSizeChange(int delta) {
-		this.cacheEstimatedSize.addAndGet(delta);
+		this.cacheEstimatedSize.add(delta);
 	}
 
 	public void setMemoryCacheSizeSupplier(LongSupplier supplier) {
@@ -60,15 +78,20 @@ public void setObfuscationQueueLengthSupplier(LongSupplier supplier) {
 		this.obfuscationQueueLength = Objects.requireNonNull(supplier);
 	}
 
+	public void setAveragePacketDelay(DoubleSupplier supplier) {
+		this.averagePacketDelay = Objects.requireNonNull(supplier);
+	}
+
 	@Override
 	public String toString() {
-		long cacheHitCountMemory = this.cacheHitCountMemory.get();
-		long cacheHitCountDisk = this.cacheHitCountDisk.get();
-		long cacheMissCount = this.cacheMissCount.get();
-		long cacheEstimatedSize = this.cacheEstimatedSize.get();
+		long cacheHitCountMemory = this.cacheHitCountMemory.sum();
+		long cacheHitCountDisk = this.cacheHitCountDisk.sum();
+		long cacheMissCount = this.cacheMissCount.sum();
+		long cacheEstimatedSize = this.cacheEstimatedSize.sum();
 		long memoryCacheSize = this.memoryCacheSize.getAsLong();
 		long diskCacheQueueLength = this.diskCacheQueueLength.getAsLong();
 		long obfuscationQueueLength = this.obfuscationQueueLength.getAsLong();
+		double averagePacketDelay = this.averagePacketDelay.getAsDouble();
 
 		double totalCacheRequest = (double) (cacheHitCountMemory + cacheHitCountDisk + cacheMissCount);
 
@@ -93,20 +116,22 @@ public String toString() {
 		builder.append(" - memoryCacheEntries: ").append(memoryCacheSize).append('\n');
 		builder.append(" - diskCacheQueueLength: ").append(diskCacheQueueLength).append('\n');
 		builder.append(" - obfuscationQueueLength: ").append(obfuscationQueueLength).append('\n');
+		builder.append(" - averagePacketDelay: ").append(formatNanos(averagePacketDelay)).append('\n');
 
 		return builder.toString();
 	}
 
 	public JsonObject toJson() {
 		JsonObject object = new JsonObject();
 
-		object.addProperty("cacheHitCountMemory", this.cacheHitCountMemory.get());
-		object.addProperty("cacheHitCountDisk", this.cacheHitCountDisk.get());
-		object.addProperty("cacheMissCount", this.cacheMissCount.get());
-		object.addProperty("cacheEstimatedSize", this.cacheEstimatedSize.get());
+		object.addProperty("cacheHitCountMemory", this.cacheHitCountMemory.sum());
+		object.addProperty("cacheHitCountDisk", this.cacheHitCountDisk.sum());
+		object.addProperty("cacheMissCount", this.cacheMissCount.sum());
+		object.addProperty("cacheEstimatedSize", this.cacheEstimatedSize.sum());
 		object.addProperty("memoryCacheSize", this.memoryCacheSize.getAsLong());
 		object.addProperty("diskCacheQueueLength", this.diskCacheQueueLength.getAsLong());
 		object.addProperty("obfuscationQueueLength", this.obfuscationQueueLength.getAsLong());
+		object.addProperty("averagePacketDelayNano", this.averagePacketDelay.getAsDouble());
 
 		return object;
 	}

orebfuscator-plugin/src/main/java/net/imprex/orebfuscator/injector/AsyncOutboundPacketHandler.java

@@ -0,0 +1,141 @@
+package net.imprex.orebfuscator.injector;
+
+import java.util.ArrayDeque;
+import java.util.ConcurrentModificationException;
+import java.util.Queue;
+
+import com.comphenix.protocol.PacketType;
+import com.comphenix.protocol.PacketType.Protocol;
+import com.comphenix.protocol.injector.packet.PacketRegistry;
+import com.comphenix.protocol.utility.MinecraftReflection;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.concurrent.Promise;
+import net.imprex.orebfuscator.util.OFCLogger;
+
+public class AsyncOutboundPacketHandler extends ChannelOutboundHandlerAdapter {
+
+	private final OrebfuscatorInjector injector;
+	private final Queue<PendingWrite> pendingWrites = new ArrayDeque<>();
+
+	private ChannelHandlerContext context;
+
+	public AsyncOutboundPacketHandler(OrebfuscatorInjector injector) {
+		this.injector = injector;
+	}
+
+	@Override
+	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+		this.context = ctx;
+	}
+
+	@Override
+	public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+		PacketType packetType = getPacketType(msg);
+		if (packetType != null && this.injector.hasOutboundAsyncListener(packetType)) {
+			// process packet async if we have any listeners
+			Promise<Object> task = ctx.channel().eventLoop().newPromise();
+			this.injector.processOutboundAsync(packetType, msg, task);
+			this.pendingWrites.offer(new PendingWrite(msg, promise, task));
+
+			// we can just call flush on completion as netty calls the listener on the
+			// channel promise (channel) event-loop
+			task.addListener(future -> this.flushWriteQueue());
+
+		} else if (this.pendingWrites.isEmpty()) {
+			// write if we don't wait on any previous message
+			ctx.write(msg, promise);
+		} else {
+			// we also need to delay any other tasks as the en-/decoder is configured by
+			// a runnable that is written to the channels pipeline
+			this.pendingWrites.offer(new PendingWrite(msg, promise, null));
+		}
+	}
+
+	private PacketType getPacketType(Object msg) {
+		if (!MinecraftReflection.isPacketClass(msg)) {
+			return null;
+		}
+
+		PacketType.Protocol protocol = this.injector.getOutboundProtocol();
+		if (protocol == Protocol.UNKNOWN) {
+			OFCLogger.debug("skipping unknown outbound protocol for " + msg.getClass());
+			return null;
+		}
+
+		PacketType packetType = PacketRegistry.getPacketType(protocol, msg.getClass());
+		if (packetType == null) {
+			OFCLogger.debug("skipping unknown outbound packet type for " + msg.getClass());
+			return null;
+		}
+
+		return packetType;
+	}
+
+	private void flushWriteQueue() {
+		if (!this.context.executor().inEventLoop()) {
+			this.context.executor().execute(this::flushWriteQueue);
+			return;
+		}
+
+		while (!this.pendingWrites.isEmpty()) {
+			PendingWrite head = this.pendingWrites.peek();
+			if (!head.isDone()) {
+				return;
+			}
+
+			if (this.pendingWrites.poll() != head) {
+				// paranoia check; this should never happen
+				throw new ConcurrentModificationException();
+			}
+
+			head.write();
+		}
+	}
+
+	private class PendingWrite {
+
+		private final long timestamp = System.nanoTime();
+
+		private Object message;
+		private final ChannelPromise promise;
+		private final Promise<Object> task;
+
+		public PendingWrite(Object message, ChannelPromise promise, Promise<Object> task) {
+			this.message = message;
+			this.promise = promise;
+			this.task = task;
+		}
+
+		public boolean isDone() {
+			return task == null || task.isDone();
+		}
+
+		public void write() {
+			if (task != null) {
+				// packet got cancel; don't write anything
+				if (task.isCancelled()) {
+					return;
+				} else if (task.cause() != null) {
+					OFCLogger.error("An unknown error occurred while processing outbound packet async: "
+							+ this.message.getClass(), task.cause());
+				} else {
+					Object message = task.getNow();
+					if (message != null) {
+						this.message = message;
+					} else {
+						OFCLogger.warn("Async packet processing returned NULL, that shouldn't happen" + task);
+					}
+				}
+			}
+
+			injector.logPacketDelay(System.nanoTime() - timestamp);
+
+			if (!context.isRemoved() && context.channel().isOpen()) {
+				context.write(this.message, this.promise);
+			}
+		}
+	}
+}