Add disposeGracefully method to Scheduler

reactor-core/src/jcstress/java/reactor/core/scheduler/RacingDisposeGracefullyStressTest.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright (c) 2022 VMware Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package reactor.core.scheduler;
+
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+abstract class RacingDisposeGracefullyStressTest<T extends Scheduler> {
+
+    final T scheduler;
+    final CountDownLatch arbiterLatch = new CountDownLatch(1);
+
+    {
+        scheduler = createScheduler();
+    }
+
+    abstract T createScheduler();
+    abstract boolean isTerminated();
+
+    void awaitArbiter() {
+        while (true) {
+            try {
+                if (arbiterLatch.await(40, TimeUnit.MILLISECONDS)) {
+                    return;
+                }
+            }
+            catch (InterruptedException ignored) {
+            }
+        }
+    }
+
+    void arbiterStarted() {
+        arbiterLatch.countDown();
+    }
+
+    boolean checkDisposeGracefullyTimesOut() {
+        long start = System.nanoTime();
+        try {
+            scheduler.disposeGracefully(Duration.ofMillis(40)).block();
+        } catch (Exception e) {
+            long duration = System.nanoTime() - start;
+            // Validate that the wait took non-zero time.
+            return (e.getCause() instanceof TimeoutException) && Duration.ofNanos(duration).toMillis() > 30;
+        }
+        return false;
+    }
+
+    boolean validateSchedulerDisposed() {
+        try {
+            scheduler.schedule(() -> {});
+        } catch (RejectedExecutionException e) {
+            scheduler.disposeGracefully(Duration.ofMillis(50)).block();
+            return scheduler.isDisposed() && isTerminated();
+        }
+        return false;
+    }
+}

reactor-core/src/jcstress/java/reactor/core/scheduler/SchedulersStressTest.java

@@ -0,0 +1,288 @@
+/*
+ * Copyright (c) 2022 VMware Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package reactor.core.scheduler;
+
+import org.openjdk.jcstress.annotations.Actor;
+import org.openjdk.jcstress.annotations.Arbiter;
+import org.openjdk.jcstress.annotations.Expect;
+import org.openjdk.jcstress.annotations.JCStressTest;
+import org.openjdk.jcstress.annotations.Outcome;
+import org.openjdk.jcstress.annotations.State;
+import org.openjdk.jcstress.infra.results.ZZZ_Result;
+import org.openjdk.jcstress.infra.results.Z_Result;
+
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public abstract class SchedulersStressTest {
+
+	private static void restart(Scheduler scheduler) {
+		scheduler.disposeGracefully(Duration.ofMillis(100)).block(Duration.ofMillis(100));
+		try {
+			scheduler.start();
+		} catch (Exception ignored) {
+		}
+	}
+
+	private static boolean canScheduleTask(Scheduler scheduler) {
+		final CountDownLatch latch = new CountDownLatch(1);
+		if (scheduler.isDisposed()) {
+			return false;
+		}
+		scheduler.schedule(latch::countDown);
+		boolean taskDone = false;
+		try {
+			taskDone = latch.await(100, TimeUnit.MILLISECONDS);
+		} catch (InterruptedException ignored) {
+		}
+		return taskDone;
+	}
+
+	@JCStressTest
+	@Outcome(id = {"true"}, expect = Expect.ACCEPTABLE, desc = "Task scheduled after racing restart")
+	@State
+	public static class SingleSchedulerStartDisposeStressTest {
+
+		final SingleScheduler scheduler = new SingleScheduler(Thread::new);
+		{
+			scheduler.start();
+		}
+
+		@Actor
+		public void restart1() {
+			restart(scheduler);
+		}
+
+		@Actor
+		public void restart2() {
+			restart(scheduler);
+		}
+
+		@Arbiter
+		public void arbiter(Z_Result r) {
+			// At this stage, at least one actor called scheduler.start(),
+			// so we should be able to execute a task.
+			r.r1 = canScheduleTask(scheduler);
+			scheduler.dispose();
+		}
+	}
+
+	@JCStressTest
+	@Outcome(id = {"true"}, expect = Expect.ACCEPTABLE, desc = "Task scheduled after racing restart")
+	@State
+	public static class ParallelSchedulerStartDisposeStressTest {
+
+		final ParallelScheduler scheduler = new ParallelScheduler(4, Thread::new);
+		{
+			scheduler.start();
+		}
+
+		@Actor
+		public void restart1() {
+			restart(scheduler);
+		}
+
+		@Actor
+		public void restart2() {
+			restart(scheduler);
+		}
+
+		@Arbiter
+		public void arbiter(Z_Result r) {
+			// At this stage, at least one actor called scheduler.start(),
+			// so we should be able to execute a task.
+			r.r1 = canScheduleTask(scheduler);
+			scheduler.dispose();
+		}
+	}
+
+	@JCStressTest
+	@Outcome(id = {"true, true, true"}, expect = Expect.ACCEPTABLE,
+			desc = "Both time out, task gets rejected, scheduler disposed eventually")
+	@State
+	public static class SingleSchedulerDisposeGracefullyStressTest extends RacingDisposeGracefullyStressTest<SingleScheduler> {
+
+		@Override
+		SingleScheduler createScheduler() {
+			return new SingleScheduler(Thread::new);
+		}
+
+		@Override
+		boolean isTerminated() {
+			return scheduler.state.currentResource.isTerminated();
+		}
+
+		{
+			scheduler.start();
+			// Schedule a task that disallows graceful closure until the arbiter kicks in
+			// to make sure that actors fail while waiting.
+			scheduler.schedule(this::awaitArbiter);
+		}
+
+		@Actor
+		public void disposeGracefully1(ZZZ_Result r) {
+			r.r1 = checkDisposeGracefullyTimesOut();
+		}
+
+		@Actor
+		public void disposeGracefully2(ZZZ_Result r) {
+			r.r2 = checkDisposeGracefullyTimesOut();
+		}
+
+		@Arbiter
+		public void arbiter(ZZZ_Result r) {
+			// Release the task blocking graceful closure.
+			arbiterStarted();
+			r.r3 = validateSchedulerDisposed();
+		}
+	}
+
+	@JCStressTest
+	@Outcome(id = {"true, true, true"}, expect = Expect.ACCEPTABLE,
+			desc = "Both time out, task gets rejected, scheduler disposed eventually")
+	@State
+	public static class ParallelSchedulerDisposeGracefullyStressTest extends RacingDisposeGracefullyStressTest<ParallelScheduler> {
+
+		@Override
+		ParallelScheduler createScheduler() {
+			return new ParallelScheduler(10, Thread::new);
+		}
+
+		@Override
+		boolean isTerminated() {
+			assert scheduler.state.initialResource != null;
+			for (ScheduledExecutorService executor : scheduler.state.initialResource) {
+				if (!executor.isTerminated()) {
+					return false;
+				}
+			}
+			return true;
+		}
+
+		{
+			scheduler.start();
+			// Schedule a task that disallows graceful closure until the arbiter kicks in
+			// to make sure that actors fail while waiting.
+			scheduler.schedule(this::awaitArbiter);
+		}
+
+		@Actor
+		public void disposeGracefully1(ZZZ_Result r) {
+			r.r1 = checkDisposeGracefullyTimesOut();
+		}
+
+		@Actor
+		public void disposeGracefully2(ZZZ_Result r) {
+			r.r2 = checkDisposeGracefullyTimesOut();
+		}
+
+		@Arbiter
+		public void arbiter(ZZZ_Result r) {
+			// Release the task blocking graceful closure.
+			arbiterStarted();
+			r.r3 = validateSchedulerDisposed();
+		}
+	}
+
+	@JCStressTest
+	@Outcome(id = {"true"}, expect = Expect.ACCEPTABLE, desc = "Task scheduled after racing restart")
+	@State
+	public static class BoundedElasticSchedulerStartDisposeStressTest {
+
+		final BoundedElasticScheduler scheduler = new BoundedElasticScheduler(1, 1, Thread::new, 5);
+		{
+			scheduler.start();
+		}
+
+		@Actor
+		public void restart1() {
+			restart(scheduler);
+		}
+
+		@Actor
+		public void restart2() {
+			restart(scheduler);
+		}
+
+		@Arbiter
+		public void arbiter(Z_Result r) {
+			// At this stage, at least one actor called scheduler.start(),
+			// so we should be able to execute a task.
+			r.r1 = canScheduleTask(scheduler);
+			scheduler.dispose();
+		}
+	}
+
+	@JCStressTest
+	@Outcome(id = {"true, true, true"}, expect = Expect.ACCEPTABLE,
+			desc = "Both time out, task gets rejected, scheduler disposed eventually")
+	@State
+	public static class BoundedElasticSchedulerDisposeGracefullyStressTest
+			extends RacingDisposeGracefullyStressTest<BoundedElasticScheduler> {
+
+		@Override
+		BoundedElasticScheduler createScheduler() {
+			return new BoundedElasticScheduler(4, 4, Thread::new, 5);
+		}
+
+		@Override
+		boolean isTerminated() {
+			for (BoundedElasticScheduler.BoundedState bs  :
+					scheduler.state.currentResource.busyStates.array) {
+				if (!bs.executor.isTerminated()) {
+					return false;
+				}
+			}
+			if (!scheduler.state.initialResource.idleQueue.isEmpty()) {
+				return false;
+			}
+			for (BoundedElasticScheduler.BoundedState bs :
+					scheduler.state.initialResource.busyStates.array) {
+				if (!bs.executor.isTerminated()) {
+					return false;
+				}
+			}
+			return scheduler.state.currentResource.idleQueue.isEmpty();
+		}
+
+		{
+			scheduler.start();
+			// Schedule a task that disallows graceful closure until the arbiter kicks in
+			// to make sure that actors fail while waiting.
+			scheduler.schedule(this::awaitArbiter);
+		}
+
+		@Actor
+		public void disposeGracefully1(ZZZ_Result r) {
+			r.r1 = checkDisposeGracefullyTimesOut();
+		}
+
+		@Actor
+		public void disposeGracefully2(ZZZ_Result r) {
+			r.r2 = checkDisposeGracefullyTimesOut();
+		}
+
+		@Arbiter
+		public void arbiter(ZZZ_Result r) {
+			// Release the task blocking graceful closure.
+			arbiterStarted();
+			r.r3 = validateSchedulerDisposed();
+		}
+	}
+}