From 9b26134e469733815347ce52020f020788fde5ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E5=AE=B6=E8=AE=AD?= Date: Sat, 1 Aug 2026 18:06:11 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BC=82=E6=AD=A5=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E5=8F=91=E5=B8=83=E4=B8=80=E8=87=B4=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ui/async/AsyncOperationController.java | 124 ++++++++++--- .../ui/SharedPaneAccessibilityTest.java | 2 + .../async/AsyncOperationControllerTest.java | 175 +++++++++++++++++- 3 files changed, 272 insertions(+), 29 deletions(-) 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 1d50922..2f0c1ce 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 @@ -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 { private final Object transitionLock = new Object(); private final OperationStateMachine stateMachine = new OperationStateMachine(); private final AtomicBoolean cancellationRequested = new AtomicBoolean(); + private final ConcurrentLinkedQueue publicationQueue = + new ConcurrentLinkedQueue<>(); + private final AtomicBoolean drainingPublications = new AtomicBoolean(); private final ReadOnlyObjectWrapper 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 uiDispatcher) { this.executor = Objects.requireNonNull(executor, "executor"); @@ -36,42 +44,46 @@ public final class AsyncOperationController { public void start(String runningMessage, OperationTask task, Consumer onSuccess, Consumer 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 { 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(() -> { - synchronized (transitionLock) { - if (generation != activeGeneration - || stateMachine.snapshot().state() != OperationState.RUNNING) { - return; - } - publish(stateMachine.progress(message, progress)); + 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; } - }); + publication = publication(stateMachine.progress(message, progress)); + } + enqueuePublication(publication); } private void completeSuccess(long generation, T result, Consumer 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 onFailure) { boolean cancelled; + Publication publication; synchronized (transitionLock) { if (generation != activeGeneration) { return; @@ -145,18 +166,58 @@ public final class AsyncOperationController { && 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 { ? error.getClass().getSimpleName() : message; } + + private record StartTransition(long generation, Publication publication) { + } + + private record Publication(long generation, long sequence, boolean running, + OperationSnapshot snapshot) { + } } diff --git a/src/test/java/com/example/workutils/system/ui/SharedPaneAccessibilityTest.java b/src/test/java/com/example/workutils/system/ui/SharedPaneAccessibilityTest.java index 0e02392..18c7bf0 100644 --- a/src/test/java/com/example/workutils/system/ui/SharedPaneAccessibilityTest.java +++ b/src/test/java/com/example/workutils/system/ui/SharedPaneAccessibilityTest.java @@ -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()); }); } 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 46376c4..1a1c40d 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 @@ -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 operations = new ArrayDeque<>(); AsyncOperationController 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 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 errors = java.util.Collections.synchronizedList( @@ -344,6 +359,164 @@ class AsyncOperationControllerTest { } } + @Test + void runningSnapshotListenerCanImmediatelyCancel() { + AtomicReference pendingOperation = new AtomicReference<>(); + AtomicBoolean runningWasPublishedFirst = new AtomicBoolean(); + AsyncOperationController 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 operations = new ArrayDeque<>(); + AtomicBoolean restarted = new AtomicBoolean(); + AtomicBoolean terminalSawStopped = new AtomicBoolean(); + AsyncOperationController 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 pendingOperation = new AtomicReference<>(); + AsyncOperationController 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 pendingOperation = new AtomicReference<>(); + AsyncOperationController controller = + new AsyncOperationController<>(pendingOperation::set, uiExecutor::execute); + CountDownLatch runningSeen = new CountDownLatch(1); + CountDownLatch cancellingSeen = new CountDownLatch(1); + AtomicReference runningThread = new AtomicReference<>(); + AtomicReference 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 pendingOperation = new AtomicReference<>(); + AsyncOperationController 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 errors) { ready.countDown();