[WIP] Retractable notifications

src/CMakeLists.txt

@@ -17,6 +17,7 @@ set(rscore_srcs
         log/stats.c
         lp/lp.c
         lp/process.c
+        lp/retractable.c
         mm/checkpoint/autonomic.c
         mm/checkpoint/full.c
         mm/checkpoint/incremental.c

src/ROOT-Sim.h

@@ -71,7 +71,7 @@ typedef bool (*CanEnd_t)(lp_id_t me, const void *snapshot);
 /// These event types are automatically scheduled to the model according to the following logic:
 /// - `LP_INIT`: Represents the initialization event for a logical process.
 /// - `LP_FINI`: Represents the finalization event for a logical process.
-enum rootsim_event { LP_INIT = 65534, LP_FINI };
+enum rootsim_event { LP_RETRACTABLE = 0, LP_INIT = 65534, LP_FINI };
 
 /**
  * @brief API to inject a new event in the simulation
@@ -155,6 +155,29 @@ extern void rs_free(void *ptr);
  */
 extern void *rs_realloc(void *ptr, size_t req_size);
 
+/**
+ * @brief Enable retractable notifications for the calling Logical Process (LP)
+ *
+ * This function allows an LP to enable retractable notifications — special events
+ * that the LP can self-schedule and efficiently modify or cancel by updating their timestamp.
+ *
+ * @param retractable_ts_pointer
+ *   Pointer to a simulation time variable stored in rollback-able memory. The LP updates this
+ *   variable to control when its retractable notification should trigger.
+ *
+ * Usage details:
+ * - The LP should update the value pointed to by retractable_ts_pointer whenever its
+ *   future notification time changes.
+ * - Setting this timestamp to SIMTIME_MAX temporarily disables the retractable notification.
+ * - Retractable notifications are logically ordered after normal events with the same timestamp,
+ *   giving normal messages priority.
+ * - Setting the timestamp to a value earlier than the current simulation time is a logical error.
+ *
+ * Retractable notifications help avoid excessive scheduling of ignored events by allowing
+ * the LP to efficiently adjust or cancel its own future notifications based on dynamic state changes.
+ */
+extern void rs_retractable_enable(simtime_t *retractable_ts_pointer);
+
 /**
  * @brief Logging levels used by the simulation kernel.
  *

src/datatypes/retractable_heap.h

@@ -0,0 +1,147 @@
+/**
+ * @file datatypes/heap.h
+ *
+ * @brief Heap datatype
+ *
+ * A very simple binary heap implemented on top of our dynamic array
+ *
+ * SPDX-FileCopyrightText: 2008-2021 HPDCS Group <rootsim@googlegroups.com>
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
+#pragma once
+
+#include <datatypes/array.h>
+
+#define HEAP_K 4
+
+#define retractable_heap_declare(type) dyn_array(type)
+
+#define retractable_heap_parent_i(i) (((i) - 1) / HEAP_K)
+#define retractable_heap_child_i(i) ((i) * HEAP_K + 1)
+
+/**
+ * @brief Initializes an empty heap
+ * @param self the heap to initialize
+ */
+#define retractable_heap_init(self) array_init(self)
+
+/**
+ * @brief Finalizes an heap
+ * @param self the heap to finalize
+ *
+ * The user is responsible for cleaning up the possibly contained items.
+ */
+#define retractable_heap_fini(self) array_fini(self)
+
+#define retractable_heap_is_empty(self) array_is_empty(self)
+#define retractable_heap_min(self) (*(__typeof__(*array_items(self)) *const)array_items(self))
+
+/**
+ * @brief Inserts an element into the heap
+ * @param self the heap target of the insertion
+ * @param cmp_f a comparing function f(a, b) which returns true iff a < b
+ * @param elem the element to insert
+ * @returns the position of the inserted element in the underlying array
+ *
+ * For correct operation of the heap you need to always pass the same @a cmp_f,
+ * both for insertion and extraction
+ */
+#define retractable_heap_insert(self, cmp_f, upd_f, elem)                                                              \
+	__extension__({                                                                                                \
+		array_expand(self);                                                                                    \
+		__typeof__(array_count(self)) k = array_count(self)++;                                                 \
+		__typeof__(array_items(self)) items = array_items(self);                                               \
+		while(k && cmp_f(elem, items[retractable_heap_parent_i(k)])) {                                         \
+			items[k] = items[retractable_heap_parent_i(k)];                                                \
+			upd_f(items[k], k);                                                                            \
+			k = retractable_heap_parent_i(k);                                                              \
+		}                                                                                                      \
+		items[k] = elem;                                                                                       \
+		upd_f(items[k], k);                                                                                    \
+		k;                                                                                                     \
+	})
+
+/**
+ * @brief Extracts an element from the heap
+ * @param self the heap from where to extract the element
+ * @param cmp_f a comparing function f(a, b) which returns true iff a < b
+ * @returns the extracted element
+ *
+ * For correct operation of the heap you need to always pass the same @a cmp_f
+ * both for insertion and extraction
+ */
+#define retractable_heap_extract(self, cmp_f, upd_f)                                                                   \
+	__extension__({                                                                                                \
+		__typeof__(array_items(self)) items = array_items(self);                                               \
+		__typeof__(*array_items(self)) reth = items[0], lasth = array_pop(self);                               \
+		__typeof__(array_count(self)) p = 0U, i = 1U, mh;                                                      \
+__bubble_up__:                                                                                                         \
+		mh = min(i + HEAP_K, array_count(self));                                                               \
+		while(i < mh) {                                                                                        \
+			if(cmp_f(items[i], lasth)) {                                                                   \
+				__typeof__(array_count(self)) j = i + 1;                                               \
+				while(j < mh) {                                                                        \
+					if(cmp_f(items[j], items[i]))                                                  \
+						i = j;                                                                 \
+					++j;                                                                           \
+				}                                                                                      \
+				items[p] = items[i];                                                                   \
+				upd_f(items[p], p);                                                                    \
+				p = i;                                                                                 \
+				i = retractable_heap_child_i(i);                                                       \
+				goto __bubble_up__;                                                                    \
+			}                                                                                              \
+			++i;                                                                                           \
+		}                                                                                                      \
+		items[p] = lasth;                                                                                      \
+		upd_f(items[p], p);                                                                                    \
+		reth;                                                                                                  \
+	})
+
+/**
+ * @brief Moves an element in the heap according to new priority
+ * @param self the heap in which to move the element
+ * @param cmp_f a comparing function f(a, b) which returns true iff a < b
+ *
+ * For correct operation of the heap you need to always pass the same @a cmp_f
+ * both for insertion, extraction and priority change
+ */
+#define retractable_heap_priority_increased(self, cmp_f, upd_f, elem, pos)                                             \
+	__extension__({                                                                                                \
+		__typeof__(array_items(self)) items = array_items(self);                                               \
+		__typeof__(array_count(self)) p = (pos), k = retractable_heap_parent_i(p);                             \
+		while(p && cmp_f((elem), items[k])) {                                                                  \
+			items[p] = items[k];                                                                           \
+			upd_f(items[p], p);                                                                            \
+			p = k;                                                                                         \
+			k = retractable_heap_parent_i(p);                                                              \
+		}                                                                                                      \
+		items[p] = (elem);                                                                                     \
+		upd_f(items[p], p);                                                                                    \
+	})
+
+#define retractable_heap_priority_decreased(self, cmp_f, upd_f, elem, pos)                                             \
+	__extension__({                                                                                                \
+		__typeof__(array_items(self)) items = array_items(self);                                               \
+		__typeof__(array_count(self)) p = (pos), k = retractable_heap_child_i(p), mh;                          \
+		mh = min(k + HEAP_K, array_count(self));                                                               \
+		while(k < mh) {                                                                                        \
+			if(!cmp_f(items[k], (elem))) {                                                                 \
+				++k;                                                                                   \
+				continue;                                                                              \
+			}                                                                                              \
+			__typeof__(array_count(self)) j = k + 1;                                                       \
+			while(j < mh) {                                                                                \
+				if(cmp_f(items[j], items[k]))                                                          \
+					k = j;                                                                         \
+				++j;                                                                                   \
+			}                                                                                              \
+			items[p] = items[k];                                                                           \
+			upd_f(items[p], p);                                                                            \
+			p = k;                                                                                         \
+			k = retractable_heap_child_i(k);                                                               \
+			mh = min(k + HEAP_K, array_count(self));                                                       \
+		}                                                                                                      \
+		items[p] = (elem);                                                                                     \
+		upd_f(items[p], p);                                                                                    \
+	})

src/lp/lp.h

@@ -23,6 +23,10 @@ struct lp_ctx {
 	simtime_t termination_t;
 	/// The pointer set by the model with the SetState() API call
 	void *state_pointer;
+	/// Retractable timestamp pointer set with RetractableEnable(), needs to be in rollback-able memory
+	simtime_t *retractable_ts_pointer;
+	/// Position of the LP entry in the retractable heap
+	array_count_t retractable_entry_pos;
 	/// The housekeeping epoch number
 	unsigned fossil_epoch;
 	/// The automatic checkpointing interval selection data

src/lp/retractable.c

@@ -0,0 +1,77 @@
+#include <lp/retractable.h>
+
+#include <datatypes/retractable_heap.h>
+#include <mm/msg_allocator.h>
+
+#include <assert.h>
+
+#define rq_elem_is_before(a, b) ((a).t < (b).t)
+#define rq_elem_update(rq, i) ((rq).lp->retractable_entry_pos = (i))
+
+struct rq_elem {
+	simtime_t t;
+	struct lp_ctx *lp;
+};
+
+static _Thread_local retractable_heap_declare(struct rq_elem) retractable_queue;
+static _Thread_local array_count_t bogus_pos;
+
+void retractable_init(void)
+{
+	retractable_heap_init(retractable_queue);
+	// bogus entry to make sure retractable_queue is always non-empty
+	struct lp_ctx *bogus = (struct lp_ctx *)((char *)&bogus_pos - offsetof(struct lp_ctx, retractable_entry_pos));
+	const struct rq_elem rq = {.t = SIMTIME_MAX, .lp = bogus};
+	retractable_heap_insert(retractable_queue, rq_elem_is_before, rq_elem_update, rq);
+}
+
+void retractable_lp_init(struct lp_ctx *lp)
+{
+	lp->retractable_ts_pointer = NULL;
+}
+
+void rs_retractable_enable(simtime_t *retractable_ts_pointer)
+{
+	assert(!current_lp->retractable_ts_pointer);
+	current_lp->retractable_ts_pointer = retractable_ts_pointer;
+	const struct rq_elem rq = {.t = SIMTIME_MAX, .lp = current_lp};
+	retractable_heap_insert(retractable_queue, rq_elem_is_before, rq_elem_update, rq);
+}
+
+void retractable_fini(void)
+{
+	retractable_heap_fini(retractable_queue);
+}
+
+void retractable_reschedule(const struct lp_ctx *lp)
+{
+	if(!lp->retractable_ts_pointer)
+		return;
+
+	simtime_t t = *lp->retractable_ts_pointer;
+	array_count_t pos = lp->retractable_entry_pos;
+	struct rq_elem rq = array_get_at(retractable_queue, pos);
+	if(t == rq.t)
+		return;
+
+	bool lowered = t > rq.t;
+	rq.t = t;
+
+	if(lowered)
+		retractable_heap_priority_decreased(retractable_queue, rq_elem_is_before, rq_elem_update, rq, pos);
+	else
+		retractable_heap_priority_increased(retractable_queue, rq_elem_is_before, rq_elem_update, rq, pos);
+}
+
+struct lp_msg *retractable_extract(void)
+{
+	struct rq_elem rq = retractable_heap_min(retractable_queue);
+	struct lp_msg *ret = msg_allocator_pack(rq.lp - lps, rq.t, LP_RETRACTABLE, NULL, 0);
+	ret->raw_flags = 0;
+	return ret;
+}
+
+bool retractable_is_before(const simtime_t normal_t)
+{
+	return retractable_heap_min(retractable_queue).t < normal_t;
+}

src/lp/retractable.h

@@ -0,0 +1,13 @@
+#pragma once
+
+#include <lp/lp.h>
+#include <lp/msg.h>
+
+#define is_retractable(msg) ((msg)->m_type == LP_RETRACTABLE)
+
+extern void retractable_init(void);
+extern void retractable_lp_init(struct lp_ctx *lp_ctx);
+extern void retractable_fini(void);
+extern void retractable_reschedule(const struct lp_ctx *lp_ctx);
+extern struct lp_msg *retractable_extract(void);
+extern bool retractable_is_before(simtime_t normal_t);

test/CMakeLists.txt

@@ -16,6 +16,10 @@ test_program_link_libraries(load rscore)
 
 # Test data structures and subsystems
 test_program(bitmap datatypes/bitmap.c)
+test_program(msg_queue datatypes/msg_queue.c)
+test_program_link_libraries(msg_queue rscore)
+test_program(retractable_heap datatypes/retractable_heap.c)
+test_program_link_libraries(retractable_heap rscore)
 test_program(mm mm/buddy.c mm/buddy_hard.c mm/parallel.c mm/main.c mock.c)
 target_include_directories(test_mm PRIVATE .)
 test_program_link_libraries(mm rscore)

test/datatypes/msg_queue.c

@@ -0,0 +1,82 @@
+/**
+ * @file test/tests/datatypes/msg_queue_test.c
+ *
+ * @brief Test: parallel message queue
+ *
+ * SPDX-FileCopyrightText: 2008-2023 HPDCS Group <rootsim@googlegroups.com>
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
+#include <test.h>
+
+#include <datatypes/msg_queue.h>
+#include <lp/lp.h> // to initialize lps, which is used in msg_queue.c
+
+#include <stdatomic.h>
+#include <stdlib.h>
+
+#define N_THREADS 0
+#define THREAD_REPS 100000
+
+static atomic_uint msg_missing = 0;
+static atomic_uint barrier_init = 0, barrier_insert = 0;
+
+static int msg_queue_test_global_init(_unused void *_)
+{
+	global_config.n_threads = test_thread_cores_count();
+	lps = malloc(sizeof(*lps) * global_config.n_threads);
+	n_lps_node = global_config.n_threads;
+	lid_node_first = 0;
+	msg_queue_global_init();
+	return 0;
+}
+
+static int msg_queue_test_global_fini(_unused void *_)
+{
+	msg_queue_global_fini();
+	free(lps);
+	return (int)(atomic_load_explicit(&msg_missing, memory_order_relaxed));
+}
+
+static int msg_queue_test(_unused void *_)
+{
+	rid = test_parallel_thread_id();
+	msg_queue_init();
+
+	atomic_fetch_add_explicit(&barrier_init, 1U, memory_order_relaxed);
+	while(atomic_load_explicit(&barrier_init, memory_order_relaxed) != test_thread_cores_count())
+		;
+
+	for(unsigned i = THREAD_REPS; i--;) {
+		struct lp_msg *msg = malloc(sizeof(*msg));
+		memset(msg, 0, sizeof(*msg));
+		msg->dest = test_random_range(N_THREADS);
+		msg->dest_t = (double)test_random_range(THREAD_REPS);
+		msg_queue_insert(msg);
+		atomic_fetch_add_explicit(&msg_missing, 1U, memory_order_relaxed);
+	}
+
+	atomic_fetch_add_explicit(&barrier_insert, 1U, memory_order_relaxed);
+	while(atomic_load_explicit(&barrier_insert, memory_order_relaxed) != test_thread_cores_count())
+		;
+
+	int ret = 0;
+	simtime_t last_time = 0.0;
+	struct lp_msg *msg;
+	while((msg = msg_queue_extract())) {
+		if(msg->dest_t < last_time)
+			--ret;
+		last_time = msg->dest_t;
+		free(msg);
+		atomic_fetch_sub_explicit(&msg_missing, 1U, memory_order_relaxed);
+	}
+
+	msg_queue_fini();
+	return ret;
+}
+
+int main(void)
+{
+	test("Message queue test: initialization", msg_queue_test_global_init, NULL);
+	test_parallel("Message queue test: extractions", msg_queue_test, NULL, N_THREADS);
+	test("Message queue test: finalization", msg_queue_test_global_fini, NULL);
+}