From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
From: Paul Sauve <paul@technove.co>
Date: Fri, 12 Feb 2021 16:29:41 -0600
Subject: [PATCH] Multithreaded entity tracking

Adds a configuration option to run the tracker on multiple threads.
This will generally be faster than the single threaded approach,
unless you don't have a lot of entities/players.

Airplane
Copyright (C) 2020 Technove LLC

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.

diff --git a/src/main/java/com/tuinity/tuinity/util/maplist/IteratorSafeOrderedReferenceSet.java b/src/main/java/com/tuinity/tuinity/util/maplist/IteratorSafeOrderedReferenceSet.java
index be408aebbccbda46e8aa82ef337574137cfa0096..b9bdce7c7ca8b7b107b0dca165d722321c16229d 100644
--- a/src/main/java/com/tuinity/tuinity/util/maplist/IteratorSafeOrderedReferenceSet.java
+++ b/src/main/java/com/tuinity/tuinity/util/maplist/IteratorSafeOrderedReferenceSet.java
@@ -16,7 +16,7 @@ public final class IteratorSafeOrderedReferenceSet<E> {
 
     /* list impl */
     protected E[] listElements;
-    protected int listSize;
+    protected int listSize; public int getListSize() { return this.listSize; } // Airplane - getter
 
     protected final double maxFragFactor;
 
diff --git a/src/main/java/gg/airplane/AirplaneConfig.java b/src/main/java/gg/airplane/AirplaneConfig.java
index 3a5a65a2f282b00602643470520fd05fba5907fb..e183b6cae23935823fa9e0f767e499a8102b0ad1 100644
--- a/src/main/java/gg/airplane/AirplaneConfig.java
+++ b/src/main/java/gg/airplane/AirplaneConfig.java
@@ -87,4 +87,18 @@ public class AirplaneConfig {
         // todo lookup token (off-thread) and let users know if their token is valid
     }
 
+
+    public static boolean multithreadedEntityTracker = true;
+    public static boolean entityTrackerAsyncPackets = false;
+
+    private static void entityTracker() {
+        manual.setComment("tracker", "Options to improve the performance of the entity tracker");
+
+        multithreadedEntityTracker = manual.get("tracker.multithreaded", multithreadedEntityTracker,
+          "This enables the multithreading of the tracker.");
+        entityTrackerAsyncPackets = manual.get("tracker.unsafe-async-packets", entityTrackerAsyncPackets,
+          "This option can break plugins that assume packets from the",
+          "entity tracker will be sent sync.");
+    }
+
 }
diff --git a/src/main/java/gg/airplane/structs/TrackQueue.java b/src/main/java/gg/airplane/structs/TrackQueue.java
new file mode 100644
index 0000000000000000000000000000000000000000..601694aa6d36ce230d84cc6b55231d2fc3af30e9
--- /dev/null
+++ b/src/main/java/gg/airplane/structs/TrackQueue.java
@@ -0,0 +1,84 @@
+package gg.airplane.structs;
+
+import com.tuinity.tuinity.util.maplist.IteratorSafeOrderedReferenceSet;
+import net.minecraft.server.Chunk;
+import net.minecraft.server.MinecraftServer;
+import org.apache.logging.log4j.Level;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Helper class to handle processing a track queue.
+ */
+public class TrackQueue {
+
+    private final IteratorSafeOrderedReferenceSet<Chunk> chunks;
+    private final ForkJoinPool pool = ForkJoinPool.commonPool();
+    private final AtomicInteger taskIndex = new AtomicInteger();
+
+    private final ConcurrentLinkedQueue<Runnable> mainThreadTasks;
+
+    public TrackQueue(IteratorSafeOrderedReferenceSet<Chunk> chunks, ConcurrentLinkedQueue<Runnable> mainThreadTasks) {
+        this.chunks = chunks;
+        this.mainThreadTasks = mainThreadTasks;
+    }
+
+    public void start() {
+        int iterator = this.chunks.createRawIterator();
+        if (iterator == -1) {
+            return;
+        }
+        try {
+            this.taskIndex.set(iterator);
+
+            for (int i = 0; i < this.pool.getParallelism(); i++) {
+                this.pool.execute(this::run);
+            }
+
+            while (this.taskIndex.get() < this.chunks.getListSize()) {
+                this.runMainThreadTasks();
+                this.handleTask(); // assist
+            }
+            this.runMainThreadTasks(); // finish tasks
+        } finally {
+            this.chunks.finishRawIterator();
+        }
+    }
+
+    private void runMainThreadTasks() {
+        Runnable task;
+        while ((task = this.mainThreadTasks.poll()) != null) {
+            try {
+                task.run();
+            } catch (Throwable t) {
+                MinecraftServer.LOGGER.log(Level.WARN, "Tasks failed while ticking track queue", t);
+            }
+        }
+
+    }
+
+    private void run() {
+        while (handleTask());
+    }
+
+    private boolean handleTask() {
+        int index;
+        while ((index = this.taskIndex.getAndIncrement()) < this.chunks.getListSize()) {
+            Chunk chunk = this.chunks.rawGet(index);
+            if (chunk != null) {
+                try {
+                    chunk.entityTracker.run();
+                } catch (Throwable t) {
+                    MinecraftServer.LOGGER.log(Level.WARN, "Ticking tracker failed", t);
+                }
+
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+}
diff --git a/src/main/java/net/minecraft/server/Chunk.java b/src/main/java/net/minecraft/server/Chunk.java
index 0924f6b484468f3cf3c2d405101c0158c12d69e6..1d65d884d1f31a73333e4cf6a9ce30d7863dab9f 100644
--- a/src/main/java/net/minecraft/server/Chunk.java
+++ b/src/main/java/net/minecraft/server/Chunk.java
@@ -65,6 +65,26 @@ public class Chunk implements IChunkAccess {
     }
     // Airplane end
 
+    // Airplane start - entity tracker runnable
+    // prevents needing to allocate new lambda in processTrackQueue
+    public final Runnable entityTracker = new Runnable() {
+        @Override
+        public void run() {
+            Entity[] entities = Chunk.this.entities.getRawData();
+            for (int i = 0, len = Chunk.this.entities.size(); i < len; ++i) {
+                Entity entity = entities[i];
+                if (entity != null) {
+                    PlayerChunkMap.EntityTracker tracker = ((WorldServer) Chunk.this.getWorld()).getChunkProvider().playerChunkMap.trackedEntities.get(entity.getId());
+                    if (tracker != null) {
+                        tracker.updatePlayers(tracker.tracker.getPlayersInTrackRange());
+                        tracker.tickEntry();
+                    }
+                }
+            }
+        }
+    };
+    // Airplane end
+
     public Chunk(World world, ChunkCoordIntPair chunkcoordintpair, BiomeStorage biomestorage) {
         this(world, chunkcoordintpair, biomestorage, ChunkConverter.a, TickListEmpty.b(), TickListEmpty.b(), 0L, (ChunkSection[]) null, (Consumer) null);
     }
diff --git a/src/main/java/net/minecraft/server/EntityPlayer.java b/src/main/java/net/minecraft/server/EntityPlayer.java
index 93bca6a52dd01d9ad524a03d90c7ec79d00d2c56..d6c6a389cf214f2f8cbb343fc8106ab7d0845950 100644
--- a/src/main/java/net/minecraft/server/EntityPlayer.java
+++ b/src/main/java/net/minecraft/server/EntityPlayer.java
@@ -52,7 +52,7 @@ public class EntityPlayer extends EntityHuman implements ICrafting {
     public NetworkManager networkManager; // Paper
     public final MinecraftServer server;
     public final PlayerInteractManager playerInteractManager;
-    public final Deque<Integer> removeQueue = new ArrayDeque<>(); // Paper
+    public final Deque<Integer> removeQueue = new java.util.concurrent.ConcurrentLinkedDeque<>(); // Paper // Airplane concurrent deque
     private final AdvancementDataPlayer advancementDataPlayer;
     private final ServerStatisticManager serverStatisticManager;
     private float lastHealthScored = Float.MIN_VALUE;
diff --git a/src/main/java/net/minecraft/server/EntityTrackerEntry.java b/src/main/java/net/minecraft/server/EntityTrackerEntry.java
index 2f758b3d3eaf5d5781a2072d0f3f7442b73cb9a8..63bcd1b61d0485b064d619ddfa0e201034087772 100644
--- a/src/main/java/net/minecraft/server/EntityTrackerEntry.java
+++ b/src/main/java/net/minecraft/server/EntityTrackerEntry.java
@@ -46,6 +46,10 @@ public class EntityTrackerEntry {
      * Requested in https://github.com/PaperMC/Paper/issues/1537 to allow intercepting packets
      */
     public void sendPlayerPacket(EntityPlayer player, Packet packet) {
+        if (!gg.airplane.AirplaneConfig.entityTrackerAsyncPackets && !org.bukkit.Bukkit.isPrimaryThread()) {
+            this.b.chunkProvider.playerChunkMap.trackerEnsureMain(() -> sendPlayerPacket(player, packet));
+            return;
+        }
         player.playerConnection.sendPacket(packet);
     }
 
@@ -74,7 +78,7 @@ public class EntityTrackerEntry {
 
     public final void tick() { this.a(); } // Paper - OBFHELPER
     public void a() {
-        com.tuinity.tuinity.util.TickThread.softEnsureTickThread("Tracker update"); // Tuinity
+        //com.tuinity.tuinity.util.TickThread.softEnsureTickThread("Tracker update"); // Tuinity // Airplane - allow multithreaded
         List<Entity> list = this.tracker.passengers; // Paper - do not copy list
 
         if (!list.equals(this.p)) {
@@ -88,6 +92,8 @@ public class EntityTrackerEntry {
             ItemStack itemstack = entityitemframe.getItem();
 
             if (this.tickCounter % 10 == 0 && itemstack.getItem() instanceof ItemWorldMap) { // CraftBukkit - Moved this.tickCounter % 10 logic here so item frames do not enter the other blocks
+                // Airplane start - process maps on main
+                this.b.chunkProvider.playerChunkMap.trackerEnsureMain(() -> {
                 WorldMap worldmap = ItemWorldMap.getSavedMap(itemstack, this.b);
                 Iterator iterator = this.trackedPlayers.iterator(); // CraftBukkit
 
@@ -101,6 +107,8 @@ public class EntityTrackerEntry {
                         entityplayer.playerConnection.sendPacket(packet);
                     }
                 }
+                });
+                // Airplane end
             }
 
             this.c();
@@ -245,18 +253,25 @@ public class EntityTrackerEntry {
             // CraftBukkit start - Create PlayerVelocity event
             boolean cancelled = false;
 
-            if (this.tracker instanceof EntityPlayer) {
+            if (this.tracker instanceof EntityPlayer && PlayerVelocityEvent.getHandlerList().getRegisteredListeners().length > 0) { // Airplane - ensure there's listeners
+                // Airplane start - run on main thread
+                this.b.chunkProvider.playerChunkMap.trackerEnsureMain(() -> {
                 Player player = (Player) this.tracker.getBukkitEntity();
                 org.bukkit.util.Vector velocity = player.getVelocity();
 
                 PlayerVelocityEvent event = new PlayerVelocityEvent(player, velocity.clone());
                 this.tracker.world.getServer().getPluginManager().callEvent(event);
 
-                if (event.isCancelled()) {
-                    cancelled = true;
-                } else if (!velocity.equals(event.getVelocity())) {
+                if (!event.isCancelled() && !velocity.equals(event.getVelocity())) {
                     player.setVelocity(event.getVelocity());
                 }
+                if (!event.isCancelled()) {
+                    this.broadcastIncludingSelf(new PacketPlayOutEntityVelocity(this.tracker)); // duplicate from !cancelled below
+                }
+
+                });
+                cancelled = true; // don't broadcast until the event has finished
+                // Airplane end
             }
 
             if (!cancelled) {
@@ -356,7 +371,9 @@ public class EntityTrackerEntry {
             if (!list.isEmpty()) {
                 consumer.accept(new PacketPlayOutEntityEquipment(this.tracker.getId(), list));
             }
+            this.b.chunkProvider.playerChunkMap.trackerEnsureMain(() -> { // Airplane
             ((EntityLiving) this.tracker).updateEquipment(); // CraftBukkit - SPIGOT-3789: sync again immediately after sending
+            }); // Airplane
         }
 
         // CraftBukkit start - Fix for nonsensical head yaw
@@ -434,6 +451,10 @@ public class EntityTrackerEntry {
     // Paper end
 
     private void broadcastIncludingSelf(Packet<?> packet) {
+        if (!gg.airplane.AirplaneConfig.entityTrackerAsyncPackets && !org.bukkit.Bukkit.isPrimaryThread()) {
+            this.b.chunkProvider.playerChunkMap.trackerEnsureMain(() -> broadcastIncludingSelf(packet));
+            return;
+        }
         this.f.accept(packet);
         if (this.tracker instanceof EntityPlayer) {
             ((EntityPlayer) this.tracker).playerConnection.sendPacket(packet);
diff --git a/src/main/java/net/minecraft/server/PlayerChunkMap.java b/src/main/java/net/minecraft/server/PlayerChunkMap.java
index 285e976d4a655fb61e70883ae89f974812be7152..5c3e0cadb02fd6e4d55ae0faa4618d78f0433a52 100644
--- a/src/main/java/net/minecraft/server/PlayerChunkMap.java
+++ b/src/main/java/net/minecraft/server/PlayerChunkMap.java
@@ -722,6 +722,11 @@ public class PlayerChunkMap extends IChunkLoader implements PlayerChunk.d {
         return (PlayerChunk) (this.hasPendingVisibleUpdate ? this.pendingVisibleChunks.get(i) : ((ProtectedVisibleChunksMap)this.visibleChunks).safeGet(i));
         // Paper end
     }
+    // Airplane start - since neither map can be updated during tracker tick, it's safe to allow direct retrieval here
+    private PlayerChunk trackerGetVisibleChunk(long i) {
+        return this.hasPendingVisibleUpdate ? this.pendingVisibleChunks.get(i) : ((ProtectedVisibleChunksMap) this.visibleChunks).safeGet(i);
+    }
+    // Airplane end
 
     protected final IntSupplier getPrioritySupplier(long i) { return c(i); } // Paper - OBFHELPER
     protected IntSupplier c(long i) {
@@ -2124,10 +2129,30 @@ public class PlayerChunkMap extends IChunkLoader implements PlayerChunk.d {
         entity.tracker = null; // Paper - We're no longer tracked
     }
 
+    // Airplane start - tools to ensure main thread
+    private final java.util.concurrent.ConcurrentLinkedQueue<Runnable> trackerMainQueue = new java.util.concurrent.ConcurrentLinkedQueue<>();
+    private gg.airplane.structs.TrackQueue trackQueue;
+
+    void trackerEnsureMain(Runnable runnable) {
+        if (this.world.serverThread == Thread.currentThread()) {
+            runnable.run();
+        } else {
+            this.trackerMainQueue.add(runnable);
+        }
+    }
+    // Airplane end
+
     // Paper start - optimised tracker
     private final void processTrackQueue() {
         this.world.timings.tracker1.startTiming();
         try {
+            // Airplane start - multithreaded tracker
+            if (this.trackQueue == null) this.trackQueue = new gg.airplane.structs.TrackQueue(this.world.getChunkProvider().entityTickingChunks, trackerMainQueue);
+            if (gg.airplane.AirplaneConfig.multithreadedEntityTracker) {
+                this.trackQueue.start();
+                return;
+            }
+            // Airplane end
             com.tuinity.tuinity.util.maplist.IteratorSafeOrderedReferenceSet.Iterator<Chunk> iterator = this.world.getChunkProvider().entityTickingChunks.iterator();
             try {
             while (iterator.hasNext()) {
@@ -2389,7 +2414,7 @@ Sections go from 0..16. Now whenever a section is not empty, it can potentially
     public class EntityTracker {
 
         final EntityTrackerEntry trackerEntry; // Paper - private -> package private
-        private final Entity tracker;
+        protected final Entity tracker; // Airplane - public for chunk
         private final int trackingDistance;
         private SectionPosition e;
         // Paper start
@@ -2408,7 +2433,9 @@ Sections go from 0..16. Now whenever a section is not empty, it can potentially
         // Paper start - use distance map to optimise tracker
         com.destroystokyo.paper.util.misc.PooledLinkedHashSets.PooledObjectLinkedOpenHashSet<EntityPlayer> lastTrackerCandidates;
 
-        final void updatePlayers(com.destroystokyo.paper.util.misc.PooledLinkedHashSets.PooledObjectLinkedOpenHashSet<EntityPlayer> newTrackerCandidates) {
+        synchronized final void tickEntry() { this.trackerEntry.tick(); } // Airplane - move entry tick into sync block
+
+        synchronized final void updatePlayers(com.destroystokyo.paper.util.misc.PooledLinkedHashSets.PooledObjectLinkedOpenHashSet<EntityPlayer> newTrackerCandidates) { // Airplane
             com.destroystokyo.paper.util.misc.PooledLinkedHashSets.PooledObjectLinkedOpenHashSet<EntityPlayer> oldTrackerCandidates = this.lastTrackerCandidates;
             this.lastTrackerCandidates = newTrackerCandidates;
 
@@ -2449,7 +2476,13 @@ Sections go from 0..16. Now whenever a section is not empty, it can potentially
             return this.tracker.getId();
         }
 
-        public void broadcast(Packet<?> packet) {
+        public synchronized void broadcast(Packet<?> packet) { // Airplane - synchronized for tracked player
+            // Airplane start
+            if (!gg.airplane.AirplaneConfig.entityTrackerAsyncPackets && !org.bukkit.Bukkit.isPrimaryThread()) {
+                trackerEnsureMain(() -> broadcast(packet));
+                return;
+            }
+            // Airplane end
             Iterator iterator = this.trackedPlayers.iterator();
 
             while (iterator.hasNext()) {
@@ -2461,6 +2494,12 @@ Sections go from 0..16. Now whenever a section is not empty, it can potentially
         }
 
         public void broadcastIncludingSelf(Packet<?> packet) {
+            // Airplane start
+            if (!gg.airplane.AirplaneConfig.entityTrackerAsyncPackets && !org.bukkit.Bukkit.isPrimaryThread()) {
+                trackerEnsureMain(() -> broadcastIncludingSelf(packet));
+                return;
+            }
+            // Airplane end
             this.broadcast(packet);
             if (this.tracker instanceof EntityPlayer) {
                 ((EntityPlayer) this.tracker).playerConnection.sendPacket(packet);
@@ -2487,8 +2526,8 @@ Sections go from 0..16. Now whenever a section is not empty, it can potentially
 
         }
 
-        public void updatePlayer(EntityPlayer entityplayer) {
-            org.spigotmc.AsyncCatcher.catchOp("player tracker update"); // Spigot
+        public synchronized void updatePlayer(EntityPlayer entityplayer) { // Airplane - sync for access to map
+            //org.spigotmc.AsyncCatcher.catchOp("player tracker update"); // Spigot // Airplane - allow sync for tracker
             if (entityplayer != this.tracker) {
                 // Paper start - remove allocation of Vec3D here
                 //Vec3D vec3d = entityplayer.getPositionVector().d(this.tracker.getPositionVector()); // MC-155077, SPIGOT-5113
@@ -2504,7 +2543,7 @@ Sections go from 0..16. Now whenever a section is not empty, it can potentially
 
                     if (!flag1) {
                         ChunkCoordIntPair chunkcoordintpair = new ChunkCoordIntPair(this.tracker.chunkX, this.tracker.chunkZ);
-                        PlayerChunk playerchunk = PlayerChunkMap.this.getVisibleChunk(chunkcoordintpair.pair());
+                        PlayerChunk playerchunk = PlayerChunkMap.this.trackerGetVisibleChunk(chunkcoordintpair.pair()); // Airplane
 
                         if (playerchunk != null && playerchunk.getSendingChunk() != null) { // Paper - no-tick view distance
                             flag1 = PlayerChunkMap.b(chunkcoordintpair, entityplayer, false) <= PlayerChunkMap.this.viewDistance;
