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(); -}