From 8c73731cb63f0751f1c30df728207ad38dc61141 Mon Sep 17 00:00:00 2001 From: Andrea Piccione Date: Thu, 3 Jul 2025 10:57:50 +0200 Subject: [PATCH] Retractable notifications, re-enabled the test of the message queue This commit introduces the new retractable notifications API. It still is not wired in the simulator logic, hence it is not working yet. Also we are missing a test for the whole feature. This will be completed soon. --- src/CMakeLists.txt | 1 + src/ROOT-Sim.h | 25 ++++- src/datatypes/retractable_heap.h | 147 ++++++++++++++++++++++++++++++ src/lp/lp.h | 4 + src/lp/retractable.c | 77 ++++++++++++++++ src/lp/retractable.h | 13 +++ test/CMakeLists.txt | 4 + test/datatypes/msg_queue.c | 82 +++++++++++++++++ test/datatypes/retractable_heap.c | 112 +++++++++++++++++++++++ test/lp/retractable.c | 3 + test/old_tests/msg_queue.c | 101 -------------------- 11 files changed, 467 insertions(+), 102 deletions(-) create mode 100644 src/datatypes/retractable_heap.h create mode 100644 src/lp/retractable.c create mode 100644 src/lp/retractable.h create mode 100644 test/datatypes/msg_queue.c create mode 100644 test/datatypes/retractable_heap.c create mode 100644 test/lp/retractable.c delete mode 100644 test/old_tests/msg_queue.c diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 6bce4eba..6b7e6ad7 100644 --- a/src/CMakeLists.txt +++ b/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 diff --git a/src/ROOT-Sim.h b/src/ROOT-Sim.h index 8b861e55..7edb2a05 100644 --- a/src/ROOT-Sim.h +++ b/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. * diff --git a/src/datatypes/retractable_heap.h b/src/datatypes/retractable_heap.h new file mode 100644 index 00000000..17db0974 --- /dev/null +++ b/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 + * SPDX-License-Identifier: GPL-3.0-only + */ +#pragma once + +#include + +#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); \ + }) diff --git a/src/lp/lp.h b/src/lp/lp.h index 8c156c5c..b73f8836 100644 --- a/src/lp/lp.h +++ b/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 diff --git a/src/lp/retractable.c b/src/lp/retractable.c new file mode 100644 index 00000000..71cc619b --- /dev/null +++ b/src/lp/retractable.c @@ -0,0 +1,77 @@ +#include + +#include +#include + +#include + +#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; +} diff --git a/src/lp/retractable.h b/src/lp/retractable.h new file mode 100644 index 00000000..5204369f --- /dev/null +++ b/src/lp/retractable.h @@ -0,0 +1,13 @@ +#pragma once + +#include +#include + +#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); diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 9ac3ad43..24b0a294 100644 --- a/test/CMakeLists.txt +++ b/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) diff --git a/test/datatypes/msg_queue.c b/test/datatypes/msg_queue.c new file mode 100644 index 00000000..96e308a9 --- /dev/null +++ b/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 + * SPDX-License-Identifier: GPL-3.0-only + */ +#include + +#include +#include // to initialize lps, which is used in msg_queue.c + +#include +#include + +#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); +} diff --git a/test/datatypes/retractable_heap.c b/test/datatypes/retractable_heap.c new file mode 100644 index 00000000..5d9d2452 --- /dev/null +++ b/test/datatypes/retractable_heap.c @@ -0,0 +1,112 @@ +/** + * @file test/datatypes/retractable_heap.c + * + * @brief Test: bitmap datatype + * + * SPDX-FileCopyrightText: 2008-2025 HPCS Group + * SPDX-License-Identifier: GPL-3.0-only + */ +#include + +#include + +#include + +#define test_cmp_f(a, b) ((a).val < (b).val) +#define test_upd_f(e, p) ((e).pos = p) + +#define NUM_ELEMENTS 16384 +#define NUM_PRIORITY_CHANGES 256 + +struct test_elem { + int val; + array_count_t pos; +}; + +static retractable_heap_declare(struct test_elem) test_heap; + +bool check_heap_property(void) +{ + array_count_t n = array_count(test_heap); + struct test_elem *items = array_items(test_heap); + for(array_count_t i = 0; i < n; ++i) { + array_count_t c = retractable_heap_child_i(i); + for(int k = 0; k < HEAP_K; k++) { + if(c + k < n && !test_cmp_f(items[i], items[c + k]) && items[i].val != items[c + k].val) + return false; + } + } + return true; +} + +static int retractable_heap_init_test(_unused void *unused) +{ + retractable_heap_init(test_heap); + for(int i = 0; i < NUM_ELEMENTS; ++i) { + struct test_elem e = (struct test_elem){.val = (int)test_random_range(INT_MAX / 2)}; + retractable_heap_insert(test_heap, test_cmp_f, test_upd_f, e); + test_assert(check_heap_property()); + } + return 0; +} + +static int retractable_heap_priority_increase_test(_unused void *unused) +{ + for(int i = 0; i < NUM_PRIORITY_CHANGES; ++i) { + test_assert(array_count(test_heap)); + + array_count_t idx = test_random_range(array_count(test_heap) - 1); + struct test_elem e = array_get_at(test_heap, idx); + + int delta = (int)test_random_range(64); + if(e.val - delta < 0) + continue; + e.val -= delta; + + retractable_heap_priority_increased(test_heap, test_cmp_f, test_upd_f, e, idx); + test_assert(check_heap_property()); + } + return 0; +} + +static int retractable_heap_priority_decrease_test(_unused void *unused) +{ + for(int i = 0; i < NUM_PRIORITY_CHANGES; ++i) { + test_assert(array_count(test_heap)); + + array_count_t idx = test_random_range(array_count(test_heap) - 1); + struct test_elem e = array_get_at(test_heap, idx); + + int delta = (int)test_random_range(128); + e.val += delta; + + retractable_heap_priority_decreased(test_heap, test_cmp_f, test_upd_f, e, idx); + test_assert(check_heap_property()); + } + return 0; +} + +static int retractable_heap_fini_test(_unused void *unused) +{ + int min_val = retractable_heap_min(test_heap).val, n_elems = 0; + while(!retractable_heap_is_empty(test_heap)) { + int current_min = retractable_heap_extract(test_heap, test_cmp_f, test_upd_f).val; + test_assert(current_min >= min_val); + min_val = current_min; + test_assert(check_heap_property()); + ++n_elems; + } + test_assert(n_elems == NUM_ELEMENTS); + + retractable_heap_fini(test_heap); + return 0; +} + +int main(void) +{ + test("Testing retractable heap init", retractable_heap_init_test, NULL); + test("Testing retractable heap priority increase", retractable_heap_priority_increase_test, NULL); + test("Testing retractable heap priority decrease", retractable_heap_priority_decrease_test, NULL); + test("Testing retractable heap fini", retractable_heap_fini_test, NULL); + return 0; +} diff --git a/test/lp/retractable.c b/test/lp/retractable.c new file mode 100644 index 00000000..3379b5ee --- /dev/null +++ b/test/lp/retractable.c @@ -0,0 +1,3 @@ +// +// Created by apiccione on 7/2/25. +// diff --git a/test/old_tests/msg_queue.c b/test/old_tests/msg_queue.c deleted file mode 100644 index 18182d77..00000000 --- a/test/old_tests/msg_queue.c +++ /dev/null @@ -1,101 +0,0 @@ -/** - * @file test/tests/datatypes/msg_queue_test.c - * - * @brief Test: parallel message queue - * - * SPDX-FileCopyrightText: 2008-2025 HPCS Group - * SPDX-License-Identifier: GPL-3.0-only - */ -#include -#include -#include - -#include "test.h" - -#include "datatypes/msg_queue.h" - -#define N_THREADS 2 -#define THREAD_REPS 100000 -#define N_LPS_PER_NODE 64 - -static atomic_uint msg_missing = 0; -static atomic_uint msg_to_free = N_THREADS; - -static void msg_allocator_free(struct lp_msg *msg) -{ - atomic_fetch_sub_explicit(&msg_to_free, 1U, memory_order_relaxed); - free(msg); -} - -static test_ret_t msg_queue_test_init(__unused void *_) -{ - msg_queue_init(); - return 0; -} - -static test_ret_t msg_queue_test_fini(__unused void *_) -{ - msg_queue_global_fini(); - return (test_ret_t)(atomic_load(&msg_missing) | atomic_load(&msg_to_free)); -} - -static test_ret_t msg_queue_populate(__unused void *_) -{ - msg_queue_init(); - - unsigned i = THREAD_REPS; - while(i--) { - struct lp_msg *msg = malloc(sizeof(*msg)); - memset(msg, 0, sizeof(*msg)); - msg->dest_t = (double)RANDOM(THREAD_REPS); - msg->dest = RANDOM(N_LPS_PER_NODE); - msg_queue_insert(msg); - atomic_fetch_add_explicit(&msg_missing, 1U, memory_order_relaxed); - } - - return (msg_missing != THREAD_REPS * N_THREADS); -} - -static test_ret_t msg_queue_empty(__unused void *_) -{ - int ret = 0; - struct lp_msg *msg; - simtime_t last_time = 0.0; - - 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); - } - - // to test msg cleanup - msg = malloc(sizeof(*msg)); - memset(msg, 0, sizeof(*msg)); - msg_queue_insert(msg); - - // test_thread_barrier(); - - msg_queue_fini(); - return ret; -} - - -void foo() {} - -int main(void) -{ - init(N_THREADS); - - n_lps_node = N_LPS_PER_NODE; - - msg_queue_global_init(); - - parallel_test("Initializing message queue", msg_queue_test_init, NULL); - parallel_test("Populating threads message queues", msg_queue_populate, NULL); - parallel_test("Extracting messages from queues", msg_queue_empty, NULL); - test("Finalizing message queue", msg_queue_test_fini, NULL); - - finish(); -}