From 8d961542d802ffec786d2346ee9f6a037755f04f Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Wed, 9 Nov 2022 16:40:16 +0000 Subject: feat(common): Add common dispatcher interface This change adds a new interface `Dispatcher` that is used throughout OpenDC for scheduling the execution of future tasks. This replaces the `CoroutineContext` and `Clock` that exist on many of the implementations in OpenDC. With this approach, we reduce the dependency on Kotlin Coroutines. --- .../main/java/org/opendc/common/Dispatcher.java | 63 ++++++++++++++++++++++ .../java/org/opendc/common/DispatcherHandle.java | 33 ++++++++++++ 2 files changed, 96 insertions(+) create mode 100644 opendc-common/src/main/java/org/opendc/common/Dispatcher.java create mode 100644 opendc-common/src/main/java/org/opendc/common/DispatcherHandle.java (limited to 'opendc-common/src/main/java/org/opendc/common') diff --git a/opendc-common/src/main/java/org/opendc/common/Dispatcher.java b/opendc-common/src/main/java/org/opendc/common/Dispatcher.java new file mode 100644 index 00000000..8c919311 --- /dev/null +++ b/opendc-common/src/main/java/org/opendc/common/Dispatcher.java @@ -0,0 +1,63 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.common; + +import java.time.InstantSource; + +/** + * A {@link Dispatcher} is used in OpenDC to schedule the execution of future tasks over potentially multiple threads. + */ +public interface Dispatcher { + /** + * Return the time source of the dispatcher as an {@link InstantSource}. + */ + InstantSource getTimeSource(); + + /** + * Schedule the specified {@link Runnable} to run as soon as possible. + * + * @param command The task to execute. + */ + default void schedule(Runnable command) { + schedule(0, command); + } + + /** + * Schedule the specified {@link Runnable} to run after the specified delay. + *

+ * Use this method to eliminate potential allocations in case the task does not need to be cancellable. + * + * @param delayMs The time from now to the delayed execution (in milliseconds). + * @param command The task to execute. + */ + void schedule(long delayMs, Runnable command); + + /** + * Schedule the specified {@link Runnable} to run after the specified delay. + * + * @param delayMs The time from now to the delayed execution (in milliseconds). + * @param command The task to execute. + * @return A {@link DispatcherHandle} representing pending completion of the task. + */ + DispatcherHandle scheduleCancellable(long delayMs, Runnable command); +} diff --git a/opendc-common/src/main/java/org/opendc/common/DispatcherHandle.java b/opendc-common/src/main/java/org/opendc/common/DispatcherHandle.java new file mode 100644 index 00000000..e34e5e11 --- /dev/null +++ b/opendc-common/src/main/java/org/opendc/common/DispatcherHandle.java @@ -0,0 +1,33 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.common; + +/** + * A handle returned by a {@link Dispatcher} representing a scheduled task. + */ +public interface DispatcherHandle { + /** + * Attempt to cancel execution of the task. + */ + void cancel(); +} -- cgit v1.2.3 From c22d744464f91eaa5f1aabee408351e864f36f1d Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Wed, 9 Nov 2022 16:44:25 +0000 Subject: feat(common): Add compatibility with Kotlin coroutines This change adds support for converting a `Dispatcher` implementation into a `CoroutineDispatcher` instance. --- .../java/org/opendc/common/DispatcherProvider.java | 33 ++++++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 opendc-common/src/main/java/org/opendc/common/DispatcherProvider.java (limited to 'opendc-common/src/main/java/org/opendc/common') diff --git a/opendc-common/src/main/java/org/opendc/common/DispatcherProvider.java b/opendc-common/src/main/java/org/opendc/common/DispatcherProvider.java new file mode 100644 index 00000000..2717bd0f --- /dev/null +++ b/opendc-common/src/main/java/org/opendc/common/DispatcherProvider.java @@ -0,0 +1,33 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.common; + +/** + * Interface to expose the {@link Dispatcher} instance used by a class. + */ +public interface DispatcherProvider { + /** + * Return the {@link Dispatcher} associated with this class. + */ + Dispatcher getDispatcher(); +} -- cgit v1.2.3 From fb2672afb2d8236d5291cd028196c99d8e4d47f1 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Wed, 9 Nov 2022 21:59:07 +0000 Subject: refactor: Replace use of CoroutineContext by Dispatcher This change replaces the use of `CoroutineContext` for passing the `SimulationDispatcher` across the different modules of OpenDC by the lightweight `Dispatcher` interface of the OpenDC common module. --- .../main/java/org/opendc/common/util/Pacer.java | 94 ++++++++ .../org/opendc/common/util/TimerScheduler.java | 256 +++++++++++++++++++++ 2 files changed, 350 insertions(+) create mode 100644 opendc-common/src/main/java/org/opendc/common/util/Pacer.java create mode 100644 opendc-common/src/main/java/org/opendc/common/util/TimerScheduler.java (limited to 'opendc-common/src/main/java/org/opendc/common') diff --git a/opendc-common/src/main/java/org/opendc/common/util/Pacer.java b/opendc-common/src/main/java/org/opendc/common/util/Pacer.java new file mode 100644 index 00000000..5b8d8cb0 --- /dev/null +++ b/opendc-common/src/main/java/org/opendc/common/util/Pacer.java @@ -0,0 +1,94 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.common.util; + +import java.util.function.LongConsumer; +import org.opendc.common.Dispatcher; +import org.opendc.common.DispatcherHandle; + +/** + * Helper class to pace the incoming scheduling requests. + */ +public final class Pacer { + private final Dispatcher dispatcher; + private final long quantumMs; + private final LongConsumer process; + + /** + * The current {@link DispatcherHandle} representing the pending scheduling cycle. + */ + private DispatcherHandle handle; + + /** + * Construct a {@link Pacer} instance. + * + * @param dispatcher The {@link Dispatcher} to schedule future invocations. + * @param quantumMs The scheduling quantum in milliseconds. + * @param process The process to invoke for the incoming requests. + */ + public Pacer(Dispatcher dispatcher, long quantumMs, LongConsumer process) { + this.dispatcher = dispatcher; + this.quantumMs = quantumMs; + this.process = process; + } + + /** + * Determine whether a scheduling cycle is pending. + */ + public boolean isPending() { + return handle != null; + } + + /** + * Enqueue a new scheduling cycle. + */ + public void enqueue() { + if (handle != null) { + return; + } + + final Dispatcher dispatcher = this.dispatcher; + long quantumMs = this.quantumMs; + long now = dispatcher.getTimeSource().millis(); + + // We assume that the scheduler runs at a fixed slot every time quantum (e.g t=0, t=60, t=120). + // We calculate here the delay until the next scheduling slot. + long timeUntilNextSlot = quantumMs - (now % quantumMs); + + handle = dispatcher.scheduleCancellable(timeUntilNextSlot, () -> { + process.accept(now + timeUntilNextSlot); + handle = null; + }); + } + + /** + * Cancel the currently pending scheduling cycle. + */ + public void cancel() { + final DispatcherHandle handle = this.handle; + if (handle != null) { + this.handle = null; + handle.cancel(); + } + } +} diff --git a/opendc-common/src/main/java/org/opendc/common/util/TimerScheduler.java b/opendc-common/src/main/java/org/opendc/common/util/TimerScheduler.java new file mode 100644 index 00000000..a85605e9 --- /dev/null +++ b/opendc-common/src/main/java/org/opendc/common/util/TimerScheduler.java @@ -0,0 +1,256 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.common.util; + +import java.util.ArrayDeque; +import java.util.HashMap; +import java.util.PriorityQueue; +import org.jetbrains.annotations.NotNull; +import org.opendc.common.Dispatcher; +import org.opendc.common.DispatcherHandle; + +/** + * A {@link TimerScheduler} facilitates scheduled execution of future tasks. + */ +public final class TimerScheduler { + private final Dispatcher dispatcher; + + /** + * The stack of the invocations to occur in the future. + */ + private final ArrayDeque invocations = new ArrayDeque<>(); + + /** + * A priority queue containing the tasks to be scheduled in the future. + */ + private final PriorityQueue> queue = new PriorityQueue>(); + + /** + * A map that keeps track of the timers. + */ + private final HashMap> timers = new HashMap<>(); + + /** + * Construct a {@link TimerScheduler} instance. + * + * @param dispatcher The {@link Dispatcher} to schedule future invocations. + */ + public TimerScheduler(Dispatcher dispatcher) { + this.dispatcher = dispatcher; + } + + /** + * Start a timer that will invoke the specified [block] after [delay]. + *

+ * Each timer has a key and if a new timer with same key is started the previous is cancelled. + * + * @param key The key of the timer to start. + * @param delay The delay before invoking the block. + * @param block The block to invoke. + */ + public void startSingleTimer(T key, long delay, Runnable block) { + startSingleTimerTo(key, dispatcher.getTimeSource().millis() + delay, block); + } + + /** + * Start a timer that will invoke the specified [block] at [timestamp]. + *

+ * Each timer has a key and if a new timer with same key is started the previous is cancelled. + * + * @param key The key of the timer to start. + * @param timestamp The timestamp at which to invoke the block. + * @param block The block to invoke. + */ + public void startSingleTimerTo(T key, long timestamp, Runnable block) { + long now = dispatcher.getTimeSource().millis(); + final PriorityQueue> queue = this.queue; + final ArrayDeque invocations = this.invocations; + + if (timestamp < now) { + throw new IllegalArgumentException("Timestamp must be in the future"); + } + + timers.compute(key, (k, old) -> { + if (old != null && old.timestamp == timestamp) { + // Fast-path: timer for the same timestamp already exists + old.block = block; + return old; + } else { + // Slow-path: cancel old timer and replace it with new timer + Timer timer = new Timer(key, timestamp, block); + + if (old != null) { + old.isCancelled = true; + } + queue.add(timer); + trySchedule(now, invocations, timestamp); + + return timer; + } + }); + } + + /** + * Check if a timer with a given key is active. + * + * @param key The key to check if active. + * @return `true` if the timer with the specified [key] is active, `false` otherwise. + */ + public boolean isTimerActive(T key) { + return timers.containsKey(key); + } + + /** + * Cancel a timer with a given key. + *

+ * If canceling a timer that was already canceled, or key never was used to start + * a timer this operation will do nothing. + * + * @param key The key of the timer to cancel. + */ + public void cancel(T key) { + final Timer timer = timers.remove(key); + + // Mark the timer as cancelled + if (timer != null) { + timer.isCancelled = true; + } + } + + /** + * Cancel all timers. + */ + public void cancelAll() { + queue.clear(); + timers.clear(); + + // Cancel all pending invocations + for (final Invocation invocation : invocations) { + invocation.cancel(); + } + + invocations.clear(); + } + + /** + * Try to schedule an engine invocation at the specified [target]. + * + * @param now The current virtual timestamp. + * @param target The virtual timestamp at which the engine invocation should happen. + * @param scheduled The queue of scheduled invocations. + */ + private void trySchedule(long now, ArrayDeque scheduled, long target) { + final Invocation head = scheduled.peek(); + + // Only schedule a new scheduler invocation in case the target is earlier than all other pending + // scheduler invocations + if (head == null || target < head.timestamp) { + final DispatcherHandle handle = dispatcher.scheduleCancellable(target - now, this::doRunTimers); + scheduled.addFirst(new Invocation(target, handle)); + } + } + + /** + * This method is invoked when the earliest timer expires. + */ + private void doRunTimers() { + final ArrayDeque invocations = this.invocations; + final Invocation invocation = invocations.remove(); + + final PriorityQueue> queue = this.queue; + final HashMap> timers = this.timers; + long now = invocation.timestamp; + + while (!queue.isEmpty()) { + final Timer timer = queue.peek(); + + long timestamp = timer.timestamp; + boolean isCancelled = timer.isCancelled; + + assert timestamp >= now : "Found task in the past"; + + if (timestamp > now && !isCancelled) { + // Schedule a task for the next event to occur. + trySchedule(now, invocations, timestamp); + break; + } + + queue.poll(); + + if (!isCancelled) { + timers.remove(timer.key); + timer.run(); + } + } + } + + /** + * A task that is scheduled to run in the future. + */ + private static class Timer implements Comparable> { + final T key; + final long timestamp; + Runnable block; + + /** + * A flag to indicate that the task has been cancelled. + */ + boolean isCancelled; + + /** + * Construct a {@link Timer} instance. + */ + public Timer(T key, long timestamp, Runnable block) { + this.key = key; + this.timestamp = timestamp; + this.block = block; + } + + /** + * Run the task. + */ + void run() { + block.run(); + } + + @Override + public int compareTo(@NotNull Timer other) { + return Long.compare(timestamp, other.timestamp); + } + } + + /** + * A future engine invocation. + *

+ * This class is used to keep track of the future engine invocations created using the {@link Dispatcher} instance. + * In case the invocation is not needed anymore, it can be cancelled via [cancel]. + */ + private record Invocation(long timestamp, DispatcherHandle handle) { + /** + * Cancel the engine invocation. + */ + void cancel() { + handle.cancel(); + } + } +} -- cgit v1.2.3