async pathfinding: fix path race condition

#519 #589

Closed #604

Author: hayanesuru

leaf-server/minecraft-patches/features/0102-Petal-Async-Pathfinding.patch

@@ -728,18 +728,21 @@ index 6c5696ab4981a6d582d4d0f13c9822bf84a5d9f1..5cd93a95091deb045b24c23d75a35a85
 
      @Override
 diff --git a/net/minecraft/world/level/pathfinder/Path.java b/net/minecraft/world/level/pathfinder/Path.java
-index 5959e1b1772ffbdfb108365171fe37cbf56ef825..68723bebf60bdb8faa243058e8b0d584cb9a2177 100644
+index 5959e1b1772ffbdfb108365171fe37cbf56ef825..701a3158c3e078e558fb1d3e11b6c40c7778561c 100644
 --- a/net/minecraft/world/level/pathfinder/Path.java
 +++ b/net/minecraft/world/level/pathfinder/Path.java
-@@ -11,7 +11,7 @@ import net.minecraft.world.entity.Entity;
+@@ -11,9 +11,9 @@ import net.minecraft.world.entity.Entity;
  import net.minecraft.world.phys.Vec3;
  import org.jspecify.annotations.Nullable;
 
 -public final class Path {
-+public class Path { // Kaiiju - petal - async path processing - not final
++public class Path { // Kaiiju - petal - async path processing - public -> public-f
      public static final StreamCodec<FriendlyByteBuf, Path> STREAM_CODEC = StreamCodec.of((buffer, value) -> value.writeToStream(buffer), Path::createFromStream);
-     public final List<Node> nodes;
+-    public final List<Node> nodes;
++    public List<Node> nodes; // Kaiiju - petal - async path processing - public -> public-f
      private Path.@Nullable DebugData debugData;
+     private int nextNodeIndex;
+     private final BlockPos target;
 @@ -27,6 +27,17 @@ public final class Path {
          this.reached = reached;
      }

leaf-server/src/main/java/org/dreeam/leaf/async/path/AsyncPath.java

@@ -10,20 +10,17 @@
 import org.jspecify.annotations.Nullable;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.function.Supplier;
 
 /**
  * I'll be using this to represent a path that not be processed yet!
  */
-public class AsyncPath extends Path {
+public final class AsyncPath extends Path implements Runnable {
 
-    /**
-     * Instead of three states, only one is actually required
-     * This will update when any thread is done with the path
-     */
-    private volatile boolean ready = false;
+    private boolean ready = false;
 
     /**
      * Runnable waiting for this to be processed
@@ -36,19 +33,9 @@ public class AsyncPath extends Path {
      */
     private final Set<BlockPos> positions;
 
-    /**
-     * The supplier of the real processed path
-     */
-    private final Supplier<Path> pathSupplier;
-
-    /*
-     * Processed values
-     */
+    private @Nullable Supplier<Path> task;
+    private volatile @Nullable Path ret;
 
-    /**
-     * This is a reference to the nodes list in the parent `Path` object
-     */
-    private final List<Node> nodes;
     /**
      * The block we're trying to path to
      * <p>
@@ -72,13 +59,20 @@ public class AsyncPath extends Path {
     public AsyncPath(List<Node> emptyNodeList, Set<BlockPos> positions, Supplier<Path> pathSupplier) {
         super(emptyNodeList, null, false);
 
-        this.nodes = emptyNodeList;
         this.positions = positions;
-        this.pathSupplier = pathSupplier;
+        this.task = pathSupplier;
 
         AsyncPathProcessor.queue(this);
     }
 
+    @Override
+    public void run() {
+        Supplier<Path> task = this.task;
+        if (task != null) {
+            this.ret = task.get();
+        }
+    }
+
     @Override
     public boolean isProcessed() {
         return this.ready;
@@ -92,9 +86,6 @@ public final void schedulePostProcessing(Runnable runnable) {
             runnable.run();
         } else {
             this.postProcessing.offer(runnable);
-            if (this.ready) {
-                this.runAllPostProcessing(true);
-            }
         }
     }
 
@@ -105,34 +96,31 @@ public final void schedulePostProcessing(Runnable runnable) {
      * @return true if we are processing the same positions
      */
     public final boolean hasSameProcessingPositions(final Set<BlockPos> positions) {
-        if (this.positions.size() != positions.size()) {
-            return false;
-        }
-
-        // For single position (common case), do direct comparison
-        if (positions.size() == 1) { // Both have the same size at this point
-            return this.positions.iterator().next().equals(positions.iterator().next());
-        }
-
-        return this.positions.containsAll(positions);
+        return this.positions.equals(positions);
     }
 
     /**
      * Starts processing this path
      * Since this is no longer a synchronized function, checkProcessed is no longer required
      */
-    public final void process() {
-        if (this.ready) return;
-
-        synchronized (this) {
-            if (this.ready) return; // In the worst case, the main thread only waits until any async thread is done and returns immediately
-            final Path bestPath = this.pathSupplier.get();
-            this.nodes.addAll(bestPath.nodes); // We mutate this list to reuse the logic in Path
-            this.target = bestPath.getTarget();
-            this.distToTarget = bestPath.getDistToTarget();
-            this.canReach = bestPath.canReach();
-            this.ready = true;
+    private final void process() {
+        if (this.ready) {
+            return;
         }
+        final Path ret = this.ret;
+        final Supplier<Path> task = this.task;
+        final Path bestPath = ret != null ? ret : Objects.requireNonNull(task).get();
+        complete(bestPath);
+    }
+
+    // not this.ready
+    private final void complete(Path bestPath) {
+        this.nodes = bestPath.nodes;
+        this.target = bestPath.getTarget();
+        this.distToTarget = bestPath.getDistToTarget();
+        this.canReach = bestPath.canReach();
+        this.task = null;
+        this.ready = true;
 
         this.runAllPostProcessing(TickThread.isTickThread());
     }
@@ -176,7 +164,14 @@ public boolean canReach() {
 
     @Override
     public boolean isDone() {
-        return this.ready && super.isDone();
+        boolean ready = this.ready;
+        if (!ready) {
+            Path ret = this.ret;
+            if (ret != null) {
+                complete(ret);
+            }
+        }
+        return ready && super.isDone();
     }
 
     @Override

leaf-server/src/main/java/org/dreeam/leaf/async/path/AsyncPathProcessor.java

@@ -46,8 +46,8 @@ public static void init() {
         }
     }
 
-    protected static CompletableFuture<Void> queue(AsyncPath path) {
-        return CompletableFuture.runAsync(path::process, PATH_PROCESSING_EXECUTOR)
+    protected static CompletableFuture<Void> queue(Runnable path) {
+        return CompletableFuture.runAsync(path, PATH_PROCESSING_EXECUTOR)
             .orTimeout(60L, TimeUnit.SECONDS)
             .exceptionally(throwable -> {
                 if (throwable instanceof TimeoutException e) {