diff --git a/src/main/java/com/example/workutils/system/ui/async/AsyncOperationController.java b/src/main/java/com/example/workutils/system/ui/async/AsyncOperationController.java index b201e79..1d50922 100644 --- a/src/main/java/com/example/workutils/system/ui/async/AsyncOperationController.java +++ b/src/main/java/com/example/workutils/system/ui/async/AsyncOperationController.java @@ -15,11 +15,14 @@ public final class AsyncOperationController { private final Executor executor; private final Consumer uiDispatcher; + private final Object transitionLock = new Object(); private final OperationStateMachine stateMachine = new OperationStateMachine(); private final AtomicBoolean cancellationRequested = new AtomicBoolean(); private final ReadOnlyObjectWrapper snapshot = new ReadOnlyObjectWrapper<>(OperationSnapshot.idle()); private final ReadOnlyBooleanWrapper running = new ReadOnlyBooleanWrapper(false); + private long nextGeneration; + private long activeGeneration; public AsyncOperationController(Executor executor, Consumer uiDispatcher) { this.executor = Objects.requireNonNull(executor, "executor"); @@ -33,49 +36,42 @@ public final class AsyncOperationController { public void start(String runningMessage, OperationTask task, Consumer onSuccess, Consumer 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 { 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 { cancellationRequested.set(false); publish(stateMachine.start(message)); running.set(true); + activeGeneration = ++nextGeneration; + return activeGeneration; } - private void dispatchSnapshot(java.util.function.Supplier 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 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 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) { diff --git a/src/main/java/com/example/workutils/system/ui/async/OperationStateMachine.java b/src/main/java/com/example/workutils/system/ui/async/OperationStateMachine.java index 84ff475..835d572 100644 --- a/src/main/java/com/example/workutils/system/ui/async/OperationStateMachine.java +++ b/src/main/java/com/example/workutils/system/ui/async/OperationStateMachine.java @@ -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); } diff --git a/src/test/java/com/example/workutils/system/ui/async/AsyncOperationControllerTest.java b/src/test/java/com/example/workutils/system/ui/async/AsyncOperationControllerTest.java index 80096be..46376c4 100644 --- a/src/test/java/com/example/workutils/system/ui/async/AsyncOperationControllerTest.java +++ b/src/test/java/com/example/workutils/system/ui/async/AsyncOperationControllerTest.java @@ -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 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 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 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 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 errors = java.util.Collections.synchronizedList(new ArrayList<>()); + List 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 result = new AtomicReference<>(); + AsyncOperationController controller = + new AsyncOperationController<>(Runnable::run, ui); + controller.start("运行中", context -> "valid-result", result::set); + CountDownLatch ready = new CountDownLatch(2); + CountDownLatch release = new CountDownLatch(1); + List 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 errors) { + ready.countDown(); + try { + release.await(); + action.run(); + } catch (Throwable error) { + errors.add(error); + } + } + + private static final class QueuedUiDispatcher implements Consumer { + + private final Deque 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(); + } + } } diff --git a/src/test/java/com/example/workutils/system/ui/async/OperationStateMachineTest.java b/src/test/java/com/example/workutils/system/ui/async/OperationStateMachineTest.java index 63e5794..9ea50da 100644 --- a/src/test/java/com/example/workutils/system/ui/async/OperationStateMachineTest.java +++ b/src/test/java/com/example/workutils/system/ui/async/OperationStateMachineTest.java @@ -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("不应成功")); + } }