修复异步取消并发与延迟事件

2026-08-01 17:44:56 +08:00
parent 7867304e07
commit 550dbd3b55
4 changed files with 267 additions and 31 deletions
@@ -15,11 +15,14 @@ public final class AsyncOperationController<T> {

    private final Executor executor;
    private final Consumer<Runnable> uiDispatcher;
+    private final Object transitionLock = new Object();
    private final OperationStateMachine stateMachine = new OperationStateMachine();
    private final AtomicBoolean cancellationRequested = new AtomicBoolean();
    private final ReadOnlyObjectWrapper<OperationSnapshot> snapshot =
            new ReadOnlyObjectWrapper<>(OperationSnapshot.idle());
    private final ReadOnlyBooleanWrapper running = new ReadOnlyBooleanWrapper(false);
+    private long nextGeneration;
+    private long activeGeneration;

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

        executor.execute(() -> {
            OperationContext context = new OperationContext(
-                    (message, progress) -> dispatchSnapshot(
-                            () -> stateMachine.progress(message, progress)),
+                    (message, progress) -> dispatchProgress(
+                            generation, message, progress),
                    cancellationRequested::get);
            try {
                T result = task.execute(context);
-                uiDispatcher.accept(() -> {
-                    publish(cancellationRequested.get()
-                            ? stateMachine.cancel()
-                            : stateMachine.succeed("操作完成"));
-                    running.set(false);
-                    onSuccess.accept(result);
-                });
+                uiDispatcher.accept(() -> completeSuccess(
+                        generation, result, onSuccess));
            } catch (Throwable error) {
                if (error instanceof InterruptedException) {
                    Thread.currentThread().interrupt();
                }
-                uiDispatcher.accept(() -> {
-                    boolean cancelled = cancellationRequested.get()
-                            && (error instanceof CancellationException
-                            || error instanceof InterruptedException);
-                    publish(cancelled
-                            ? stateMachine.cancel()
-                            : stateMachine.fail(readableMessage(error)));
-                    running.set(false);
-                    if (!cancelled) {
-                        onFailure.accept(error);
-                    }
-                });
+                uiDispatcher.accept(() -> completeFailure(
+                        generation, error, onFailure));
            }
        });
    }

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

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

-    private void prepareStart(String message) {
+    private long prepareStart(String message) {
        OperationState current = stateMachine.snapshot().state();
        if (current == OperationState.SUCCEEDED || current == OperationState.FAILED
                || current == OperationState.CANCELLED) {
@@ -99,10 +95,64 @@ public final class AsyncOperationController<T> {
        cancellationRequested.set(false);
        publish(stateMachine.start(message));
        running.set(true);
+        activeGeneration = ++nextGeneration;
+        return activeGeneration;
    }

-    private void dispatchSnapshot(java.util.function.Supplier<OperationSnapshot> transition) {
-        uiDispatcher.accept(() -> publish(transition.get()));
+    private void dispatchProgress(long generation, String message, double progress) {
+        uiDispatcher.accept(() -> {
+            synchronized (transitionLock) {
+                if (generation != activeGeneration
+                        || stateMachine.snapshot().state() != OperationState.RUNNING) {
+                    return;
+                }
+                publish(stateMachine.progress(message, progress));
+            }
+        });
+    }
+
+    private void completeSuccess(long generation, T result, Consumer<T> onSuccess) {
+        synchronized (transitionLock) {
+            if (generation != activeGeneration) {
+                return;
+            }
+            OperationState state = stateMachine.snapshot().state();
+            if (state == OperationState.RUNNING) {
+                publish(stateMachine.succeed("操作完成"));
+            } else if (state == OperationState.CANCELLING) {
+                publish(stateMachine.cancel());
+            } else {
+                return;
+            }
+            running.set(false);
+        }
+        onSuccess.accept(result);
+    }
+
+    private void completeFailure(long generation, Throwable error,
+                                 Consumer<Throwable> onFailure) {
+        boolean cancelled;
+        synchronized (transitionLock) {
+            if (generation != activeGeneration) {
+                return;
+            }
+            OperationState state = stateMachine.snapshot().state();
+            if (state != OperationState.RUNNING
+                    && state != OperationState.CANCELLING) {
+                return;
+            }
+            cancelled = state == OperationState.CANCELLING
+                    && cancellationRequested.get()
+                    && (error instanceof CancellationException
+                    || error instanceof InterruptedException);
+            publish(cancelled
+                    ? stateMachine.cancel()
+                    : stateMachine.fail(readableMessage(error)));
+            running.set(false);
+        }
+        if (!cancelled) {
+            onFailure.accept(error);
+        }
    }

    private void publish(OperationSnapshot value) {
@@ -26,7 +26,7 @@ public final class OperationStateMachine {
    }

    public synchronized OperationSnapshot succeed(String message) {
-        requireState(OperationState.RUNNING, OperationState.CANCELLING);
+        requireState(OperationState.RUNNING);
        return update(OperationState.SUCCEEDED, message, 1);
    }

@@ -3,6 +3,10 @@ package com.example.workutils.system.ui.async;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;

+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutorService;
@@ -10,7 +14,9 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;

+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -203,4 +209,175 @@ class AsyncOperationControllerTest {
            Thread.interrupted();
        }
    }
+
+    @Test
+    void queuedProgressAfterCancellationIsDiscarded() {
+        QueuedUiDispatcher ui = new QueuedUiDispatcher();
+        AsyncOperationController<String> controller =
+                new AsyncOperationController<>(Runnable::run, ui);
+
+        controller.start("运行中", context -> {
+            context.report("延迟进度", 0.5);
+            return "result";
+        }, ignored -> { });
+        controller.cancel();
+
+        assertDoesNotThrow(ui::runFirst);
+        assertEquals(OperationState.CANCELLING, controller.snapshot().state());
+
+        ui.runFirst();
+        assertEquals(OperationState.CANCELLED, controller.snapshot().state());
+    }
+
+    @Test
+    void queuedProgressAfterCompletionIsDiscarded() {
+        QueuedUiDispatcher ui = new QueuedUiDispatcher();
+        AsyncOperationController<String> controller =
+                new AsyncOperationController<>(Runnable::run, ui);
+
+        controller.start("运行中", context -> {
+            context.report("延迟进度", 0.5);
+            return "result";
+        }, ignored -> { });
+
+        ui.runLast();
+        assertEquals(OperationState.SUCCEEDED, controller.snapshot().state());
+        assertDoesNotThrow(ui::runFirst);
+        assertEquals(OperationState.SUCCEEDED, controller.snapshot().state());
+    }
+
+    @Test
+    void previousOperationProgressCannotPolluteRestartedOperation() {
+        QueuedUiDispatcher ui = new QueuedUiDispatcher();
+        AsyncOperationController<String> controller =
+                new AsyncOperationController<>(Runnable::run, ui);
+
+        controller.start("旧任务", context -> {
+            context.report("旧进度", 0.75);
+            return "old-result";
+        }, ignored -> { });
+        ui.runLast();
+        controller.start("新任务", context -> "new-result", ignored -> { });
+
+        ui.runFirst();
+
+        assertEquals(OperationState.RUNNING, controller.snapshot().state());
+        assertEquals("新任务", controller.snapshot().message());
+        assertEquals(-1, controller.snapshot().progress());
+
+        ui.runFirst();
+        assertEquals(OperationState.SUCCEEDED, controller.snapshot().state());
+    }
+
+    @Test
+    void repeatedConcurrentCancellationIsIdempotent() throws Exception {
+        QueuedUiDispatcher ui = new QueuedUiDispatcher();
+        AsyncOperationController<String> controller =
+                new AsyncOperationController<>(Runnable::run, ui);
+        controller.start("运行中", context -> "result", ignored -> { });
+        int callerCount = 4;
+        CountDownLatch ready = new CountDownLatch(callerCount);
+        CountDownLatch release = new CountDownLatch(1);
+        List<Throwable> errors = java.util.Collections.synchronizedList(new ArrayList<>());
+        List<Thread> callers = new ArrayList<>();
+
+        for (int index = 0; index < callerCount; index++) {
+            Thread caller = new Thread(() -> {
+                ready.countDown();
+                try {
+                    release.await();
+                    controller.cancel();
+                } catch (Throwable error) {
+                    errors.add(error);
+                }
+            });
+            callers.add(caller);
+            caller.start();
+        }
+
+        assertTrue(ready.await(2, TimeUnit.SECONDS));
+        release.countDown();
+        for (Thread caller : callers) {
+            caller.join(2_000);
+            assertFalse(caller.isAlive());
+        }
+
+        assertTrue(errors.isEmpty(), () -> "cancel errors: " + errors);
+        assertEquals(OperationState.CANCELLING, controller.snapshot().state());
+
+        ui.runFirst();
+        assertEquals(OperationState.CANCELLED, controller.snapshot().state());
+        assertFalse(controller.runningProperty().get());
+    }
+
+    @Test
+    void cancellationAndQueuedCompletionSerializeToOneTerminalState() throws Exception {
+        for (int attempt = 0; attempt < 50; attempt++) {
+            QueuedUiDispatcher ui = new QueuedUiDispatcher();
+            AtomicReference<String> result = new AtomicReference<>();
+            AsyncOperationController<String> controller =
+                    new AsyncOperationController<>(Runnable::run, ui);
+            controller.start("运行中", context -> "valid-result", result::set);
+            CountDownLatch ready = new CountDownLatch(2);
+            CountDownLatch release = new CountDownLatch(1);
+            List<Throwable> errors = java.util.Collections.synchronizedList(
+                    new ArrayList<>());
+            Thread canceller = new Thread(() -> race(ready, release,
+                    controller::cancel, errors));
+            Thread completer = new Thread(() -> race(ready, release,
+                    ui::runFirst, errors));
+            canceller.start();
+            completer.start();
+
+            assertTrue(ready.await(2, TimeUnit.SECONDS));
+            release.countDown();
+            canceller.join(2_000);
+            completer.join(2_000);
+
+            assertFalse(canceller.isAlive());
+            assertFalse(completer.isAlive());
+            assertTrue(errors.isEmpty(), () -> "transition errors: " + errors);
+            assertTrue(controller.snapshot().state() == OperationState.SUCCEEDED
+                    || controller.snapshot().state() == OperationState.CANCELLED);
+            assertEquals("valid-result", result.get());
+            assertFalse(controller.runningProperty().get());
+        }
+    }
+
+    private static void race(CountDownLatch ready, CountDownLatch release,
+                             Runnable action, List<Throwable> errors) {
+        ready.countDown();
+        try {
+            release.await();
+            action.run();
+        } catch (Throwable error) {
+            errors.add(error);
+        }
+    }
+
+    private static final class QueuedUiDispatcher implements Consumer<Runnable> {
+
+        private final Deque<Runnable> pending = new ArrayDeque<>();
+
+        @Override
+        public synchronized void accept(Runnable action) {
+            pending.addLast(action);
+        }
+
+        private void runFirst() {
+            Runnable action;
+            synchronized (this) {
+                action = pending.removeFirst();
+            }
+            action.run();
+        }
+
+        private void runLast() {
+            Runnable action;
+            synchronized (this) {
+                action = pending.removeLast();
+            }
+            action.run();
+        }
+    }
 }
@@ -76,4 +76,13 @@ class OperationStateMachineTest {

        assertThrows(IllegalStateException.class, machine::cancel);
    }
+
+    @Test
+    void cancellingOperationCannotSucceed() {
+        OperationStateMachine machine = new OperationStateMachine();
+        machine.start("运行");
+        machine.requestCancel();
+
+        assertThrows(IllegalStateException.class, () -> machine.succeed("不应成功"));
+    }
 }