修复异步状态发布一致性

2026-08-01 18:06:11 +08:00
parent 550dbd3b55
commit 9b26134e46
3 changed files with 272 additions and 29 deletions
@@ -7,6 +7,7 @@ import javafx.beans.property.ReadOnlyObjectWrapper;

 import java.util.Objects;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
@@ -18,11 +19,18 @@ public final class AsyncOperationController<T> {
    private final Object transitionLock = new Object();
    private final OperationStateMachine stateMachine = new OperationStateMachine();
    private final AtomicBoolean cancellationRequested = new AtomicBoolean();
+    private final ConcurrentLinkedQueue<Publication> publicationQueue =
+            new ConcurrentLinkedQueue<>();
+    private final AtomicBoolean drainingPublications = new AtomicBoolean();
    private final ReadOnlyObjectWrapper<OperationSnapshot> snapshot =
            new ReadOnlyObjectWrapper<>(OperationSnapshot.idle());
    private final ReadOnlyBooleanWrapper running = new ReadOnlyBooleanWrapper(false);
+    private boolean operationRunning;
    private long nextGeneration;
    private long activeGeneration;
+    private long nextPublicationSequence;
+    private long appliedGeneration;
+    private long appliedPublicationSequence;

    public AsyncOperationController(Executor executor, Consumer<Runnable> uiDispatcher) {
        this.executor = Objects.requireNonNull(executor, "executor");
@@ -36,42 +44,46 @@ public final class AsyncOperationController<T> {
    public void start(String runningMessage, OperationTask<T> task,
                      Consumer<T> onSuccess, Consumer<Throwable> onFailure) {
        Objects.requireNonNull(task, "task");
-        long generation;
+        StartTransition startTransition;
        synchronized (transitionLock) {
-            if (running.get()) {
+            if (operationRunning) {
                throw new IllegalStateException("Operation is already running");
            }
-            generation = prepareStart(runningMessage);
+            startTransition = prepareStart(runningMessage);
        }
+        dispatchPublication(startTransition.publication());

        executor.execute(() -> {
            OperationContext context = new OperationContext(
                    (message, progress) -> dispatchProgress(
-                            generation, message, progress),
+                            startTransition.generation(), message, progress),
                    cancellationRequested::get);
            try {
                T result = task.execute(context);
                uiDispatcher.accept(() -> completeSuccess(
-                        generation, result, onSuccess));
+                        startTransition.generation(), result, onSuccess));
            } catch (Throwable error) {
                if (error instanceof InterruptedException) {
                    Thread.currentThread().interrupt();
                }
                uiDispatcher.accept(() -> completeFailure(
-                        generation, error, onFailure));
+                        startTransition.generation(), error, onFailure));
            }
        });
    }

    public void cancel() {
+        Publication publication;
        synchronized (transitionLock) {
-            if (!running.get()
+            if (!operationRunning
                    || stateMachine.snapshot().state() != OperationState.RUNNING) {
                return;
            }
            cancellationRequested.set(true);
-            publish(stateMachine.requestCancel());
+            OperationSnapshot cancelling = stateMachine.requestCancel();
+            publication = publication(cancelling);
        }
+        dispatchPublication(publication);
    }

    public OperationSnapshot snapshot() {
@@ -86,52 +98,61 @@ public final class AsyncOperationController<T> {
        return running.getReadOnlyProperty();
    }

-    private long prepareStart(String message) {
+    private StartTransition prepareStart(String message) {
        OperationState current = stateMachine.snapshot().state();
        if (current == OperationState.SUCCEEDED || current == OperationState.FAILED
                || current == OperationState.CANCELLED) {
            stateMachine.reset();
        }
        cancellationRequested.set(false);
-        publish(stateMachine.start(message));
-        running.set(true);
+        OperationSnapshot started = stateMachine.start(message);
+        operationRunning = true;
        activeGeneration = ++nextGeneration;
-        return activeGeneration;
+        return new StartTransition(activeGeneration, publication(started));
    }

    private void dispatchProgress(long generation, String message, double progress) {
-        uiDispatcher.accept(() -> {
+        uiDispatcher.accept(() -> applyProgress(generation, message, progress));
+    }
+
+    private void applyProgress(long generation, String message, double progress) {
+        Publication publication;
        synchronized (transitionLock) {
            if (generation != activeGeneration
                    || stateMachine.snapshot().state() != OperationState.RUNNING) {
                return;
            }
-                publish(stateMachine.progress(message, progress));
+            publication = publication(stateMachine.progress(message, progress));
        }
-        });
+        enqueuePublication(publication);
    }

    private void completeSuccess(long generation, T result, Consumer<T> onSuccess) {
+        Publication publication;
        synchronized (transitionLock) {
            if (generation != activeGeneration) {
                return;
            }
            OperationState state = stateMachine.snapshot().state();
+            OperationSnapshot terminal;
            if (state == OperationState.RUNNING) {
-                publish(stateMachine.succeed("操作完成"));
+                terminal = stateMachine.succeed("操作完成");
            } else if (state == OperationState.CANCELLING) {
-                publish(stateMachine.cancel());
+                terminal = stateMachine.cancel();
            } else {
                return;
            }
-            running.set(false);
+            operationRunning = false;
+            publication = publication(terminal);
        }
+        enqueuePublication(publication);
        onSuccess.accept(result);
    }

    private void completeFailure(long generation, Throwable error,
                                 Consumer<Throwable> onFailure) {
        boolean cancelled;
+        Publication publication;
        synchronized (transitionLock) {
            if (generation != activeGeneration) {
                return;
@@ -145,18 +166,58 @@ public final class AsyncOperationController<T> {
                    && cancellationRequested.get()
                    && (error instanceof CancellationException
                    || error instanceof InterruptedException);
-            publish(cancelled
+            OperationSnapshot terminal = cancelled
                    ? stateMachine.cancel()
-                    : stateMachine.fail(readableMessage(error)));
-            running.set(false);
+                    : stateMachine.fail(readableMessage(error));
+            operationRunning = false;
+            publication = publication(terminal);
        }
+        enqueuePublication(publication);
        if (!cancelled) {
            onFailure.accept(error);
        }
    }

-    private void publish(OperationSnapshot value) {
-        snapshot.set(value);
+    private Publication publication(OperationSnapshot value) {
+        return new Publication(activeGeneration, ++nextPublicationSequence,
+                operationRunning, value);
+    }
+
+    private void dispatchPublication(Publication publication) {
+        uiDispatcher.accept(() -> enqueuePublication(publication));
+    }
+
+    private void enqueuePublication(Publication publication) {
+        publicationQueue.add(publication);
+        drainPublications();
+    }
+
+    private void drainPublications() {
+        if (!drainingPublications.compareAndSet(false, true)) {
+            return;
+        }
+        try {
+            Publication publication;
+            while ((publication = publicationQueue.poll()) != null) {
+                applyPublication(publication);
+            }
+        } finally {
+            drainingPublications.set(false);
+            if (!publicationQueue.isEmpty()) {
+                drainPublications();
+            }
+        }
+    }
+
+    private void applyPublication(Publication publication) {
+        if (publication.sequence() <= appliedPublicationSequence
+                || publication.generation() < appliedGeneration) {
+            return;
+        }
+        appliedGeneration = publication.generation();
+        appliedPublicationSequence = publication.sequence();
+        running.set(publication.running());
+        snapshot.set(publication.snapshot());
    }

    private String readableMessage(Throwable error) {
@@ -165,4 +226,11 @@ public final class AsyncOperationController<T> {
                ? error.getClass().getSimpleName()
                : message;
    }
+
+    private record StartTransition(long generation, Publication publication) {
+    }
+
+    private record Publication(long generation, long sequence, boolean running,
+                               OperationSnapshot snapshot) {
+    }
 }
@@ -94,6 +94,7 @@ class SharedPaneAccessibilityTest {
            }, ignored -> { });
            pendingOperation.get().run();
            uiUpdates.get(0).run();
+            uiUpdates.get(1).run();

            assertTrue(pane.getChildren().stream()
                    .filter(javafx.scene.control.Label.class::isInstance)
@@ -102,6 +103,7 @@ class SharedPaneAccessibilityTest {
                    .anyMatch(text -> text.contains("42%")));

            cancelButton.fire();
+            uiUpdates.get(3).run();
            assertEquals(OperationState.CANCELLING, controller.snapshot().state());
        });
    }
@@ -222,6 +222,12 @@ class AsyncOperationControllerTest {
        }, ignored -> { });
        controller.cancel();

+        ui.runLast();
+        assertEquals(OperationState.CANCELLING, controller.snapshot().state());
+
+        ui.runFirst();
+        assertEquals(OperationState.CANCELLING, controller.snapshot().state());
+
        assertDoesNotThrow(ui::runFirst);
        assertEquals(OperationState.CANCELLING, controller.snapshot().state());

@@ -243,28 +249,34 @@ class AsyncOperationControllerTest {
        ui.runLast();
        assertEquals(OperationState.SUCCEEDED, controller.snapshot().state());
        assertDoesNotThrow(ui::runFirst);
+        assertDoesNotThrow(ui::runFirst);
        assertEquals(OperationState.SUCCEEDED, controller.snapshot().state());
    }

    @Test
    void previousOperationProgressCannotPolluteRestartedOperation() {
        QueuedUiDispatcher ui = new QueuedUiDispatcher();
+        Deque<Runnable> operations = new ArrayDeque<>();
        AsyncOperationController<String> controller =
-                new AsyncOperationController<>(Runnable::run, ui);
+                new AsyncOperationController<>(operations::addLast, ui);

        controller.start("旧任务", context -> {
            context.report("旧进度", 0.75);
            return "old-result";
        }, ignored -> { });
+        operations.removeFirst().run();
        ui.runLast();
        controller.start("新任务", context -> "new-result", ignored -> { });
+        ui.runLast();

+        ui.runFirst();
        ui.runFirst();

        assertEquals(OperationState.RUNNING, controller.snapshot().state());
        assertEquals("新任务", controller.snapshot().message());
        assertEquals(-1, controller.snapshot().progress());

+        operations.removeFirst().run();
        ui.runFirst();
        assertEquals(OperationState.SUCCEEDED, controller.snapshot().state());
    }
@@ -303,8 +315,10 @@ class AsyncOperationControllerTest {
        }

        assertTrue(errors.isEmpty(), () -> "cancel errors: " + errors);
+        ui.runLast();
        assertEquals(OperationState.CANCELLING, controller.snapshot().state());

+        ui.runFirst();
        ui.runFirst();
        assertEquals(OperationState.CANCELLED, controller.snapshot().state());
        assertFalse(controller.runningProperty().get());
@@ -318,6 +332,7 @@ class AsyncOperationControllerTest {
            AsyncOperationController<String> controller =
                    new AsyncOperationController<>(Runnable::run, ui);
            controller.start("运行中", context -> "valid-result", result::set);
+            ui.runFirst();
            CountDownLatch ready = new CountDownLatch(2);
            CountDownLatch release = new CountDownLatch(1);
            List<Throwable> errors = java.util.Collections.synchronizedList(
@@ -344,6 +359,164 @@ class AsyncOperationControllerTest {
        }
    }

+    @Test
+    void runningSnapshotListenerCanImmediatelyCancel() {
+        AtomicReference<Runnable> pendingOperation = new AtomicReference<>();
+        AtomicBoolean runningWasPublishedFirst = new AtomicBoolean();
+        AsyncOperationController<String> controller =
+                new AsyncOperationController<>(pendingOperation::set, Runnable::run);
+        controller.snapshotProperty().addListener((observable, oldValue, value) -> {
+            if (value.state() == OperationState.RUNNING) {
+                runningWasPublishedFirst.set(controller.runningProperty().get());
+                controller.cancel();
+            }
+        });
+
+        controller.start("运行中", context -> "result", ignored -> { });
+
+        assertTrue(runningWasPublishedFirst.get());
+        assertEquals(OperationState.CANCELLING, controller.snapshot().state());
+        pendingOperation.get().run();
+        assertEquals(OperationState.CANCELLED, controller.snapshot().state());
+    }
+
+    @Test
+    void terminalSnapshotListenerCanImmediatelyRestart() {
+        Deque<Runnable> operations = new ArrayDeque<>();
+        AtomicBoolean restarted = new AtomicBoolean();
+        AtomicBoolean terminalSawStopped = new AtomicBoolean();
+        AsyncOperationController<String> controller =
+                new AsyncOperationController<>(operations::addLast, Runnable::run);
+        controller.snapshotProperty().addListener((observable, oldValue, value) -> {
+            if (value.state() == OperationState.SUCCEEDED
+                    && !restarted.get()) {
+                terminalSawStopped.set(!controller.runningProperty().get());
+                controller.start("第二个任务", context -> "second", ignored -> { });
+                restarted.set(true);
+            }
+        });
+
+        controller.start("第一个任务", context -> "first", ignored -> { });
+        operations.removeFirst().run();
+
+        assertTrue(restarted.get());
+        assertTrue(terminalSawStopped.get());
+        assertTrue(controller.runningProperty().get());
+        assertEquals(OperationState.RUNNING, controller.snapshot().state());
+        assertEquals("第二个任务", controller.snapshot().message());
+
+        operations.removeFirst().run();
+        assertEquals(OperationState.SUCCEEDED, controller.snapshot().state());
+    }
+
+    @Test
+    void blockingSnapshotListenerDoesNotHoldTransitionLock() throws Exception {
+        AtomicReference<Runnable> pendingOperation = new AtomicReference<>();
+        AsyncOperationController<String> controller =
+                new AsyncOperationController<>(pendingOperation::set, Runnable::run);
+        CountDownLatch listenerEntered = new CountDownLatch(1);
+        CountDownLatch releaseListener = new CountDownLatch(1);
+        CountDownLatch cancelReturned = new CountDownLatch(1);
+        controller.snapshotProperty().addListener((observable, oldValue, value) -> {
+            if (value.state() == OperationState.RUNNING) {
+                listenerEntered.countDown();
+                try {
+                    releaseListener.await();
+                } catch (InterruptedException error) {
+                    Thread.currentThread().interrupt();
+                    throw new AssertionError(error);
+                }
+            }
+        });
+        Thread starter = new Thread(() -> controller.start(
+                "运行中", context -> "result", ignored -> { }));
+        Thread canceller = new Thread(() -> {
+            controller.cancel();
+            cancelReturned.countDown();
+        });
+
+        try {
+            starter.start();
+            assertTrue(listenerEntered.await(2, TimeUnit.SECONDS));
+            canceller.start();
+
+            assertTrue(cancelReturned.await(1, TimeUnit.SECONDS),
+                    "a listener must not retain the transition lock");
+        } finally {
+            releaseListener.countDown();
+            starter.join(2_000);
+            canceller.join(2_000);
+        }
+
+        assertFalse(starter.isAlive());
+        assertFalse(canceller.isAlive());
+        assertEquals(OperationState.CANCELLING, controller.snapshot().state());
+    }
+
+    @Test
+    void backgroundCancelPublishesOnUiDispatcherThread() throws Exception {
+        ExecutorService uiExecutor = Executors.newSingleThreadExecutor(action -> {
+            Thread thread = new Thread(action);
+            thread.setName("operation-ui-dispatcher");
+            return thread;
+        });
+        try {
+            AtomicReference<Runnable> pendingOperation = new AtomicReference<>();
+            AsyncOperationController<String> controller =
+                    new AsyncOperationController<>(pendingOperation::set, uiExecutor::execute);
+            CountDownLatch runningSeen = new CountDownLatch(1);
+            CountDownLatch cancellingSeen = new CountDownLatch(1);
+            AtomicReference<String> runningThread = new AtomicReference<>();
+            AtomicReference<String> cancellingThread = new AtomicReference<>();
+            controller.snapshotProperty().addListener((observable, oldValue, value) -> {
+                if (value.state() == OperationState.RUNNING) {
+                    runningThread.set(Thread.currentThread().getName());
+                    runningSeen.countDown();
+                } else if (value.state() == OperationState.CANCELLING) {
+                    cancellingThread.set(Thread.currentThread().getName());
+                    cancellingSeen.countDown();
+                }
+            });
+
+            controller.start("运行中", context -> "result", ignored -> { });
+            assertTrue(runningSeen.await(2, TimeUnit.SECONDS));
+            Thread background = new Thread(controller::cancel, "background-canceller");
+            background.start();
+            background.join(2_000);
+
+            assertFalse(background.isAlive());
+            assertTrue(cancellingSeen.await(2, TimeUnit.SECONDS));
+            assertEquals("operation-ui-dispatcher", runningThread.get());
+            assertEquals("operation-ui-dispatcher", cancellingThread.get());
+        } finally {
+            uiExecutor.shutdownNow();
+        }
+    }
+
+    @Test
+    void outOfOrderPublicationsCannotRollBackVisibleState() {
+        QueuedUiDispatcher ui = new QueuedUiDispatcher();
+        AtomicReference<Runnable> pendingOperation = new AtomicReference<>();
+        AsyncOperationController<String> controller =
+                new AsyncOperationController<>(pendingOperation::set, ui);
+
+        controller.start("运行中", context -> "result", ignored -> { });
+        controller.cancel();
+
+        ui.runLast();
+        assertTrue(controller.runningProperty().get());
+        assertEquals(OperationState.CANCELLING, controller.snapshot().state());
+
+        ui.runFirst();
+        assertTrue(controller.runningProperty().get());
+        assertEquals(OperationState.CANCELLING, controller.snapshot().state());
+
+        pendingOperation.get().run();
+        ui.runFirst();
+        assertFalse(controller.runningProperty().get());
+        assertEquals(OperationState.CANCELLED, controller.snapshot().state());
+    }
+
    private static void race(CountDownLatch ready, CountDownLatch release,
                             Runnable action, List<Throwable> errors) {
        ready.countDown();