KeYiMC/patches/server/0028-Multithreaded-tracker.patch

From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
From: peaches94 <peachescu94@gmail.com>
Date: Sat, 2 Jul 2022 00:35:56 -0500
Subject: [PATCH] Multithreaded tracker

Original code by Bloom-host, licensed under GNU General Public License v3.0
You can find the original code on https://github.com/Bloom-host/Petal

Co-authored-by: Paul Sauve <paul@technove.co>
Co-authored-by: Kevin Raneri <kevin.raneri@gmail.com>

based off the airplane multithreaded tracker this patch properly handles
concurrent accesses everywhere, as well as being much simpler to maintain

some things are too unsafe to run off the main thread so we don't attempt to do
that. this multithreaded tracker remains accurate, non-breaking and fast

we also learned from pufferfish core that changes have to be sent ordered
now we do that in an optimized way

diff --git a/src/main/java/host/bloom/tracker/MultithreadedTracker.java b/src/main/java/host/bloom/tracker/MultithreadedTracker.java
new file mode 100644
index 0000000000000000000000000000000000000000..d27b7224ed2bcc63386dc46c33bfb8b272d91f92
--- /dev/null
+++ b/src/main/java/host/bloom/tracker/MultithreadedTracker.java
@@ -0,0 +1,154 @@
+package host.bloom.tracker;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.papermc.paper.util.maplist.IteratorSafeOrderedReferenceSet;
+import io.papermc.paper.world.ChunkEntitySlices;
+import net.minecraft.server.MinecraftServer;
+import net.minecraft.server.level.ChunkMap;
+import net.minecraft.world.entity.Entity;
+import net.minecraft.world.level.chunk.LevelChunk;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class MultithreadedTracker {
+
+    private enum TrackerStage {
+        UPDATE_PLAYERS,
+        SEND_CHANGES
+    }
+
+    private static final int parallelism = Math.max(4, Runtime.getRuntime().availableProcessors());
+    private static final Executor trackerExecutor = Executors.newFixedThreadPool(parallelism, new ThreadFactoryBuilder()
+            .setNameFormat("petal-tracker-%d")
+            .setPriority(Thread.NORM_PRIORITY - 2)
+            .build());
+
+    private final IteratorSafeOrderedReferenceSet<LevelChunk> entityTickingChunks;
+    private final AtomicInteger taskIndex = new AtomicInteger();
+
+    private final ConcurrentLinkedQueue<Runnable> mainThreadTasks;
+    private final AtomicInteger finishedTasks = new AtomicInteger();
+
+    public MultithreadedTracker(IteratorSafeOrderedReferenceSet<LevelChunk> entityTickingChunks, ConcurrentLinkedQueue<Runnable> mainThreadTasks) {
+        this.entityTickingChunks = entityTickingChunks;
+        this.mainThreadTasks = mainThreadTasks;
+    }
+
+    public void tick() {
+        int iterator = this.entityTickingChunks.createRawIterator();
+
+        if (iterator == -1) {
+            return;
+        }
+
+        // start with updating players
+        try {
+            this.taskIndex.set(iterator);
+            this.finishedTasks.set(0);
+
+            for (int i = 0; i < parallelism; i++) {
+                trackerExecutor.execute(this::runUpdatePlayers);
+            }
+
+            while (this.taskIndex.get() < this.entityTickingChunks.getListSize()) {
+                this.runMainThreadTasks();
+                this.handleChunkUpdates(5); // assist
+            }
+
+            while (this.finishedTasks.get() != parallelism) {
+                this.runMainThreadTasks();
+            }
+
+            this.runMainThreadTasks(); // finish any remaining tasks
+        } finally {
+            this.entityTickingChunks.finishRawIterator();
+        }
+
+        // then send changes
+        iterator = this.entityTickingChunks.createRawIterator();
+
+        if (iterator == -1) {
+            return;
+        }
+
+        try {
+            do {
+                LevelChunk chunk = this.entityTickingChunks.rawGet(iterator);
+
+                if (chunk != null) {
+                    this.updateChunkEntities(chunk, TrackerStage.SEND_CHANGES);
+                }
+            } while (++iterator < this.entityTickingChunks.getListSize());
+        } finally {
+            this.entityTickingChunks.finishRawIterator();
+        }
+    }
+
+    private void runMainThreadTasks() {
+        try {
+            Runnable task;
+            while ((task = this.mainThreadTasks.poll()) != null) {
+                task.run();
+            }
+        } catch (Throwable throwable) {
+            MinecraftServer.LOGGER.warn("Tasks failed while ticking track queue", throwable);
+        }
+    }
+
+    private void runUpdatePlayers() {
+        try {
+            while (handleChunkUpdates(10));
+        } finally {
+            this.finishedTasks.incrementAndGet();
+        }
+    }
+
+    private boolean handleChunkUpdates(int tasks) {
+        int index;
+        while ((index = this.taskIndex.getAndAdd(tasks)) < this.entityTickingChunks.getListSize()) {
+            for (int i = index; i < index + tasks && i < this.entityTickingChunks.getListSize(); i++) {
+                LevelChunk chunk = this.entityTickingChunks.rawGet(i);
+                if (chunk != null) {
+                    try {
+                        this.updateChunkEntities(chunk, TrackerStage.UPDATE_PLAYERS);
+                    } catch (Throwable throwable) {
+                        MinecraftServer.LOGGER.warn("Ticking tracker failed", throwable);
+                    }
+
+                }
+            }
+
+            return true;
+        }
+
+        return false;
+    }
+
+    private void updateChunkEntities(LevelChunk chunk, TrackerStage trackerStage) {
+        final ChunkEntitySlices entitySlices = chunk.level.getEntityLookup().getChunk(chunk.locX, chunk.locZ);
+        if (entitySlices == null) {
+            return;
+        }
+
+        final Entity[] rawEntities = entitySlices.entities.getRawData();
+        final ChunkMap chunkMap = chunk.level.chunkSource.chunkMap;
+
+        for (int i = 0; i < rawEntities.length; i++) {
+            Entity entity = rawEntities[i];
+            if (entity != null) {
+                ChunkMap.TrackedEntity entityTracker = chunkMap.entityMap.get(entity.getId());
+                if (entityTracker != null) {
+                    if (trackerStage == TrackerStage.SEND_CHANGES) {
+                        entityTracker.serverEntity.sendChanges();
+                    } else if (trackerStage == TrackerStage.UPDATE_PLAYERS) {
+                        entityTracker.updatePlayers(entityTracker.entity.getPlayersInTrackRange());
+                    }
+                }
+            }
+        }
+    }
+
+}
diff --git a/src/main/java/io/papermc/paper/util/maplist/IteratorSafeOrderedReferenceSet.java b/src/main/java/io/papermc/paper/util/maplist/IteratorSafeOrderedReferenceSet.java
index 0fd814f1d65c111266a2b20f86561839a4cef755..169ac3ad1b1e8e3e1874ada2471e478233c6ada7 100644
--- a/src/main/java/io/papermc/paper/util/maplist/IteratorSafeOrderedReferenceSet.java
+++ b/src/main/java/io/papermc/paper/util/maplist/IteratorSafeOrderedReferenceSet.java
@@ -15,7 +15,7 @@ public final class IteratorSafeOrderedReferenceSet<E> {
 
     /* list impl */
     protected E[] listElements;
-    protected int listSize;
+    protected int listSize; public int getListSize() { return this.listSize; } // petal - expose listSize
 
     protected final double maxFragFactor;
 
diff --git a/src/main/java/io/papermc/paper/world/ChunkEntitySlices.java b/src/main/java/io/papermc/paper/world/ChunkEntitySlices.java
index f597d65d56964297eeeed6c7e77703764178fee0..665c377e2d0d342f4dcc89c4cbdfcc9e4b96e95c 100644
--- a/src/main/java/io/papermc/paper/world/ChunkEntitySlices.java
+++ b/src/main/java/io/papermc/paper/world/ChunkEntitySlices.java
@@ -35,7 +35,7 @@ public final class ChunkEntitySlices {
     protected final EntityCollectionBySection allEntities;
     protected final EntityCollectionBySection hardCollidingEntities;
     protected final Reference2ObjectOpenHashMap<Class<? extends Entity>, EntityCollectionBySection> entitiesByClass;
-    protected final EntityList entities = new EntityList();
+    public final EntityList entities = new EntityList();
 
     public ChunkHolder.FullChunkStatus status;
 
diff --git a/src/main/java/net/minecraft/server/level/ChunkMap.java b/src/main/java/net/minecraft/server/level/ChunkMap.java
index 9976630fad886392057f642e84f919f0b95cc040..c4284894258df0b0bcc2d86c5a6da0047b994191 100644
--- a/src/main/java/net/minecraft/server/level/ChunkMap.java
+++ b/src/main/java/net/minecraft/server/level/ChunkMap.java
@@ -1236,9 +1236,38 @@ public class ChunkMap extends ChunkStorage implements ChunkHolder.PlayerProvider
         entity.tracker = null; // Paper - We're no longer tracked
     }
 
+    // petal start - multithreaded tracker
+    private @Nullable host.bloom.tracker.MultithreadedTracker multithreadedTracker;
+    private final java.util.concurrent.ConcurrentLinkedQueue<Runnable> trackerMainThreadTasks = new java.util.concurrent.ConcurrentLinkedQueue<>();
+    private boolean tracking = false;
+
+    public void runOnTrackerMainThread(final Runnable runnable) {
+        if (this.tracking) {
+            this.trackerMainThreadTasks.add(runnable);
+        } else {
+            runnable.run();
+        }
+    }
+
     // Paper start - optimised tracker
     private final void processTrackQueue() {
+        if (true) {
+            if (this.multithreadedTracker == null) {
+                this.multithreadedTracker = new host.bloom.tracker.MultithreadedTracker(this.level.chunkSource.entityTickingChunks, this.trackerMainThreadTasks);
+            }
+
+            this.tracking = true;
+            try {
+                this.multithreadedTracker.tick();
+            } finally {
+                this.tracking = false;
+            }
+            return;
+        }
+        // petal end
+
         this.level.timings.tracker1.startTiming();
+        //this.level.timings.tracker1.startTiming(); // Purpur
         try {
             for (TrackedEntity tracker : this.entityMap.values()) {
                 // update tracker entry
@@ -1461,11 +1490,11 @@ public class ChunkMap extends ChunkStorage implements ChunkHolder.PlayerProvider
 
     public class TrackedEntity {
 
-        final ServerEntity serverEntity;
-        final Entity entity;
+        public final ServerEntity serverEntity; // petal -> public
+        public final Entity entity; // petal -> public
         private final int range;
         SectionPos lastSectionPos;
-        public final Set<ServerPlayerConnection> seenBy = new ReferenceOpenHashSet<>(); // Paper - optimise map impl
+        public final Set<ServerPlayerConnection> seenBy = it.unimi.dsi.fastutil.objects.ReferenceSets.synchronize(new ReferenceOpenHashSet<>()); // Paper - optimise map impl // petal - sync
 
         public TrackedEntity(Entity entity, int i, int j, boolean flag) {
             this.serverEntity = new ServerEntity(ChunkMap.this.level, entity, j, flag, this::broadcast, this.seenBy); // CraftBukkit
@@ -1477,7 +1506,7 @@ public class ChunkMap extends ChunkStorage implements ChunkHolder.PlayerProvider
         // Paper start - use distance map to optimise tracker
         com.destroystokyo.paper.util.misc.PooledLinkedHashSets.PooledObjectLinkedOpenHashSet<ServerPlayer> lastTrackerCandidates;
 
-        final void updatePlayers(com.destroystokyo.paper.util.misc.PooledLinkedHashSets.PooledObjectLinkedOpenHashSet<ServerPlayer> newTrackerCandidates) {
+        public final void updatePlayers(com.destroystokyo.paper.util.misc.PooledLinkedHashSets.PooledObjectLinkedOpenHashSet<ServerPlayer> newTrackerCandidates) { // petal -> public
             com.destroystokyo.paper.util.misc.PooledLinkedHashSets.PooledObjectLinkedOpenHashSet<ServerPlayer> oldTrackerCandidates = this.lastTrackerCandidates;
             this.lastTrackerCandidates = newTrackerCandidates;
 
@@ -1549,7 +1578,7 @@ public class ChunkMap extends ChunkStorage implements ChunkHolder.PlayerProvider
         }
 
         public void removePlayer(ServerPlayer player) {
-            org.spigotmc.AsyncCatcher.catchOp("player tracker clear"); // Spigot
+            //org.spigotmc.AsyncCatcher.catchOp("player tracker clear"); // Spigot // petal - we can remove async too
             if (this.seenBy.remove(player.connection)) {
                 this.serverEntity.removePairing(player);
             }
@@ -1557,7 +1586,7 @@ public class ChunkMap extends ChunkStorage implements ChunkHolder.PlayerProvider
         }
 
         public void updatePlayer(ServerPlayer player) {
-            org.spigotmc.AsyncCatcher.catchOp("player tracker update"); // Spigot
+            //org.spigotmc.AsyncCatcher.catchOp("player tracker update"); // Spigot // petal - we can update async
             if (player != this.entity) {
                 // Paper start - remove allocation of Vec3D here
                 // Vec3 vec3d = player.position().subtract(this.entity.position());
diff --git a/src/main/java/net/minecraft/server/level/ServerBossEvent.java b/src/main/java/net/minecraft/server/level/ServerBossEvent.java
index ca42c2642a729b90d22b968af7258f3aee72e14b..40261b80d947a6be43465013fae5532197cfe721 100644
--- a/src/main/java/net/minecraft/server/level/ServerBossEvent.java
+++ b/src/main/java/net/minecraft/server/level/ServerBossEvent.java
@@ -13,7 +13,7 @@ import net.minecraft.util.Mth;
 import net.minecraft.world.BossEvent;
 
 public class ServerBossEvent extends BossEvent {
-    private final Set<ServerPlayer> players = Sets.newHashSet();
+    private final Set<ServerPlayer> players = Sets.newConcurrentHashSet(); // petal - players can be removed in async tracking
     private final Set<ServerPlayer> unmodifiablePlayers = Collections.unmodifiableSet(this.players);
     public boolean visible = true;
 
diff --git a/src/main/java/net/minecraft/server/level/ServerEntity.java b/src/main/java/net/minecraft/server/level/ServerEntity.java
index 3441339e1ba5efb0e25c16fa13cb65d2fbdafc42..555336cf6700566e8a99e0f0cd6f0cc41b6c5ba0 100644
--- a/src/main/java/net/minecraft/server/level/ServerEntity.java
+++ b/src/main/java/net/minecraft/server/level/ServerEntity.java
@@ -249,14 +249,18 @@ public class ServerEntity {
 
     public void removePairing(ServerPlayer player) {
         this.entity.stopSeenByPlayer(player);
-        player.connection.send(new ClientboundRemoveEntitiesPacket(new int[]{this.entity.getId()}));
+        // petal start - ensure main thread
+        ((ServerLevel) this.entity.level).chunkSource.chunkMap.runOnTrackerMainThread(() ->
+            player.connection.send(new ClientboundRemoveEntitiesPacket(new int[]{this.entity.getId()}))
+        );
+        // petal end
     }
 
     public void addPairing(ServerPlayer player) {
         ServerGamePacketListenerImpl playerconnection = player.connection;
 
         Objects.requireNonNull(player.connection);
-        this.sendPairingData(playerconnection::send, player); // CraftBukkit - add player
+        ((ServerLevel) this.entity.level).chunkSource.chunkMap.runOnTrackerMainThread(() -> this.sendPairingData(playerconnection::send, player)); // CraftBukkit - add player // petal - main thread
         this.entity.startSeenByPlayer(player);
     }
 
@@ -362,19 +366,30 @@ public class ServerEntity {
         SynchedEntityData datawatcher = this.entity.getEntityData();
 
         if (datawatcher.isDirty()) {
-            this.broadcastAndSend(new ClientboundSetEntityDataPacket(this.entity.getId(), datawatcher, false));
+            // Petal start - sync
+            ((ServerLevel) this.entity.level).chunkSource.chunkMap.runOnTrackerMainThread(() ->
+                this.broadcastAndSend(new ClientboundSetEntityDataPacket(this.entity.getId(), datawatcher, false))
+            );
+            // Petal end
         }
 
         if (this.entity instanceof LivingEntity) {
             Set<AttributeInstance> set = ((LivingEntity) this.entity).getAttributes().getDirtyAttributes();
 
             if (!set.isEmpty()) {
+                // Petal start - sync
+                final var copy = Lists.newArrayList(set);
+                ((ServerLevel) this.entity.level).chunkSource.chunkMap.runOnTrackerMainThread(() -> {
+
                 // CraftBukkit start - Send scaled max health
                 if (this.entity instanceof ServerPlayer) {
-                    ((ServerPlayer) this.entity).getBukkitEntity().injectScaledMaxHealth(set, false);
+                    ((ServerPlayer) this.entity).getBukkitEntity().injectScaledMaxHealth(copy, false);
                 }
                 // CraftBukkit end
-                this.broadcastAndSend(new ClientboundUpdateAttributesPacket(this.entity.getId(), set));
+                this.broadcastAndSend(new ClientboundUpdateAttributesPacket(this.entity.getId(), copy));
+
+                });
+                // Petal end
             }
 
             set.clear();