Add disposeGracefully method to Scheduler

[chemicL]

Currently, all Schedulers forcefully shutdown the underlying ExecutorServices by calling shutdownNow() method. In some scenarios, that is undesired as it does not allow proper cleanup.
Schedulers should allow shutting down by not accepting new work, but giving the currently executing tasks a chance to finish without interruption. A Mono<Void> disposeGracefully() method has been added for that purpose. Upon subscription, it calls shutdown() instead of shutdownNow() and creates a background task that does awaitTermination() to complete the returned Mono. It can be combined with timeout() and retry() operators in realistic scenarios.

[simonbasle]

this is shaping up good, but the review triggers a few thoughts on my part and surface additional corner cases 🤔

[simonbasle]

the semantics of disposeGracefully may vary widely, so I would make the documented contract say that explicitly (eg. "each class implementing this trait should define how subsequent calls behave during the grace period and after it")

[simonbasle]

one limitation I'm thinking of with this API is that most of the time the underlying resource(s) being disposed gracefully will be atomically swapped out. Which means that even if one re-subscribes to the Mono, including with onErrorResume() or retry(), the underlying resources won't be reachable anymore.

thus, there will be no way of composing operators to fall back to a "hard" shutdown once the graceful shutdown is initiated.

I'm thinking this is fine if documented. The recommendation for implementors should probably be to trigger a hard dispose at the end of the gracePeriod THEN propagate a TimeoutException, noting that it only serves as a warning / logging but cannot be recovered.

[chemicL]

We discussed the possible contracts here with @OlegDokuka. The approach with forceful shutdown before propagating the TimeoutException and non-recoverable errors might be confusing to users if they don't read the specific Scheduler documentation, but it has some advantages implementation wise.
Another approach could be to propagate a retry-able error and allow re-initiating the shutdown() + awaitTermination(...) procedure, while also allowing for an explicit final call to explicit shutdownNow() when desired. I'll go back to the original issue and ask for opinion from the user's perspective to guide the design.

[OlegDokuka]

My personal opinion is that once we fix specific behavior (e.g. call shutdownNow() if a timeout or InteruptedException) then we probably will end up with everyone doing scheduler.disposeGracefully(Duration.ofHours(9999999)).subscribe() or then complaining that they did not wont to have shutdownNow called but rather retry later

Another thought on TimoutException - any exception is useless if we can not do anything useful after that. I'm not sure that logging such an event makes any sense. This exception is just a fact that we forced shutdown process so a user just has to take it. Also, taking into account the impl details - all other active subscribers are going to get the same notification but the other late subscriber will not get it, then it is going to be too confusing so even having it documented will not resolve this confusion.

My personal recommendation is to prefer flexibility over fixed behavior. One can always write something like the following to mimic what we can hardcode

scheduler.disposeGracefully(Duration.ofMillis(100))
    .retryWhen(Retry.backoff(5, Duration.ofMillis(50)))
    .onErrorResume(e -> Mono.fromRunnable(scheduler::dispose));

[simonbasle]

@chemicL what do you think of this piece of code to shutdown multiple executors at once, try to await as close to the grace period as possible while still only needing one thread, and finally shutdownNow in case the whole thing takes more than gracePeriod?

	//this supposes that we somehow can get all the executors and swap them with an empty array
	//with more advanced schedulers like BoundedElasticScheduler, we might not get an ExecutorService array
	//but an array of another resource (like BoundedState[]), hence the Function
	static <RES> void shutdownAndAwait(final RES[] resources, Function<RES, ExecutorService> serviceExtractor, Duration gracePeriod, Sinks.Empty<Void> disposeNotifier) {
		for (int i = 0; i < resources.length; i++) {
			ExecutorService service = serviceExtractor.apply(resources[i]);
			service.shutdown();
		}

		//TODO: use a configurable separate pool?
		final ExecutorService service = Executors.newSingleThreadExecutor();
		service.submit(() -> {
			long nanoStart = System.nanoTime();
			long nanoGraceRemaining = gracePeriod.toNanos();

			boolean allAwaited = true;
			//wait for one executor at a time
			int index = 0;
			while (index < resources.length) {
				ExecutorService toAwait = serviceExtractor.apply(resources[index]);;
				//short case: the current executor has already terminated
				if (toAwait.isTerminated()) {
					index++;
					continue;
				}

				//we're inspecting the next executor and giving it nanoGraceRemaining ns to terminate gracefully
				try {
					if (!toAwait.awaitTermination(nanoGraceRemaining, TimeUnit.NANOSECONDS)) {
						//if it didn't terminate gracefully, the whole graceful operation can be considered a failure
						allAwaited = false;
						break;
					}
					else {
						//update the nanoGraceRemaining so that the global operation is within gracePeriod bounds
						long oldStart = nanoStart;
						nanoStart = System.nanoTime();
						nanoGraceRemaining = Math.max(0, nanoStart - oldStart);
						index++;
					}
				}
				catch (InterruptedException e) {
					allAwaited = false;
					break;
				}
			}
			if (allAwaited) {
				disposeNotifier.tryEmitEmpty();
			}
			else {
				for (int i = 0; i < resources.length; i++) {
					ExecutorService executorService = serviceExtractor.apply(resources[i]);
					executorService.shutdownNow();
				}
				disposeNotifier.tryEmitError(new TimeoutException("Scheduler didn't shutdown gracefully in time (" + gracePeriod + "), used shutdownNow"));
			}
		});
	}

[chemicL]

@chemicL what do you think of this piece of code to shutdown multiple executors at once, try to await as close to the grace period as possible while still only needing one thread, and finally shutdownNow in case the whole thing takes more than gracePeriod?
(...)

Yep, that's a great optimization, thanks for the suggestion.

[chemicL]

This is no longer the case. It still could race, but now there's a few more instructions in dispose() so the task gets aborted. IMO it's not a feature worth testing - the test was introduced in 6f3383d but one should not rely on a race to dispose tasks.

[OlegDokuka]

Nice progress overall. Left my comments. Also, there are a set of general polishing points:

1. Let's use plain volatile access (e.g. this.state instead of STATE.get(this)) plain access is the fastest approach and we use it consistently through the codebase. AtomicXXXFieldUpdate#get is an option to get value when plain access is impossible (e.g. shared utils function i.e Operators#requested)
2. Let's make sure imports are not collapsed
3. ShedulerState#terminated is not making useful of the old state except bounded elastic, thus lets avoid terminated(old) and use static final TERMINATED_STATE = new SchedulerState(dead_executor_service, Mono.empty()) where possible ;

[chemicL]

The latest changes include improvements for avoiding looping in the dispose/disposeGracefully/start methods as suggested by @OlegDokuka. This was only possible by revisiting BoundedElasticScheduler's inner workings and avoiding atomic replacing of underlying BoundedStates by a tombstone. Instead, an encapsulating class was introduced for state management of BoundedServices, which in turn simplified the state transitions in BoundedElasticScheduler itself with regards to the generic SchedulerState. As convoluted as it sounds, the latest set of changes also incorporates a fix for a leakage of BoundedState which would not be shutdown if dispose happened while an ExecutorService was being picked.

[OlegDokuka]

looks good overall with few minor comments

[simonbasle]

great job @chemicL ! finally approved and ready to merge 😄

now the only remaining step is to try to summarize the design of the change for the commit message 📖
feel free to ping me if you want me to also review that, or if you need help with merging / forward-merging.

[reactorbot]

@chemicL this PR seems to have been merged on a maintenance branch, please ensure the change is merge-forwarded to intermediate maintenance branches and up to main 🙇