Add support for timeouts on receive calls

This change allows processes to set a timeout when waiting for a message
to arrive.

2 files changed, 65 insertions, 1 deletions

diff --git a/opendc-omega/src/main/kotlin/nl/atlarge/opendc/kernel/omega/OmegaSimulation.kt b/opendc-omega/src/main/kotlin/nl/atlarge/opendc/kernel/omega/OmegaSimulation.kt
index ec7701e7..65d9dd60 100644
--- a/opendc-omega/src/main/kotlin/nl/atlarge/opendc/kernel/omega/OmegaSimulation.kt
+++ b/opendc-omega/src/main/kotlin/nl/atlarge/opendc/kernel/omega/OmegaSimulation.kt
@@ -101,8 +101,14 @@ internal class OmegaSimulation(override val kernel: OmegaKernel, override val to
 				// Tick has already occurred
 				logger.warn { "message processed out of order" }
 			}
+			// Remove the message from the queue
 			queue.poll()
 
+			// If the sender has canceled the message, we move on to the next message
+			if (envelope.canceled) {
+				continue
+			}
+
 			val context = registry[envelope.destination] ?: continue
 
 			if (envelope.message !is Interrupt) {
@@ -244,7 +250,9 @@ internal class OmegaSimulation(override val kernel: OmegaKernel, override val to
 		}
 
 		/**
-		 * Retrieves and removes a single message from this channel suspending the caller while the channel is empty.
+		 * Retrieve and removes a single message from the entity's mailbox, suspending the function if the mailbox is empty.
+		 * The execution is resumed after the message has landed in the entity's mailbox after which the message [Envelope]
+		 * is mapped through `block` to generate a processed message.
 		 *
 		 * @param block The block to process the message with.
 		 * @return The processed message.
@@ -255,6 +263,29 @@ internal class OmegaSimulation(override val kernel: OmegaKernel, override val to
 		}
 
 		/**
+		 * Retrieve and removes a single message from the entity's mailbox, suspending the function if the mailbox is empty.
+		 * The execution is resumed after the message has landed in the entity's mailbox or the timeout was reached,
+		 *
+		 * If the message has been received, the message [Envelope] is mapped through `block` to generate a processed
+		 * message. If the timeout was reached, `block` is not called and `null` is returned.
+		 *
+		 * @param timeout The duration to wait before resuming execution.
+		 * @param block The block to process the message with.
+		 * @return The processed message or `null` if the timeout was reached.
+		 */
+		suspend override fun <T> receive(timeout: Duration, block: Envelope<*>.(Any) -> T): T? {
+			val receipt = schedule(Timeout, entity, entity, timeout)
+			val envelope = receiveEnvelope()
+
+			if (envelope.message !is Timeout) {
+				receipt.cancel()
+				return block(envelope, envelope.message)
+			}
+
+			return null
+		}
+
+		/**
 		 * Send the given message to the specified entity.
 		 *
 		 * @param msg The message to send.
diff --git a/opendc-omega/src/main/kotlin/nl/atlarge/opendc/kernel/omega/Timeout.kt b/opendc-omega/src/main/kotlin/nl/atlarge/opendc/kernel/omega/Timeout.kt
new file mode 100644
index 00000000..41dcce71
--- /dev/null
+++ b/opendc-omega/src/main/kotlin/nl/atlarge/opendc/kernel/omega/Timeout.kt
@@ -0,0 +1,33 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2017 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 nl.atlarge.opendc.kernel.omega
+
+/**
+ * An internal message used by the Omega simulation kernel to indicate to a suspended [Process], that a timeout has been
+ * reached and that it should wake up and resume execution.
+ *
+ * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl)
+ */
+object Timeout