Implement committed output

src/CMakeLists.txt

@@ -5,13 +5,14 @@ set(rscore_srcs
         arch/mem.c
         arch/thread.c
         core/core.c
-        init.c
         core/sync.c
+        core/output.c
         datatypes/msg_queue.c
         distributed/control_msg.c
         gvt/fossil.c
         gvt/gvt.c
         gvt/termination.c
+        init.c
         log/file.c
         log/log.c
         log/stats.c

src/ROOT-Sim.h

@@ -66,6 +66,23 @@ typedef void (*ProcessEvent_t)(lp_id_t me, simtime_t now, unsigned event_type, c
  */
 typedef bool (*CanEnd_t)(lp_id_t me, const void *snapshot);
 
+/**
+ * @brief Perform output operations.
+ * @param me The logical process ID of the LP performing the output
+ * @param output_type Numerical output type
+ * @param output_content The output content
+ * @param output_size The size (in bytes) of the output content
+ *
+ * This function is called by the simulation kernel whenever an event is committed, during the handling of which
+ * an output operation had been scheduled by the simulation model using the ScheduleOutput() function.
+ * This function will receive the same data that was passed to the ScheduleOutput() function.
+ *
+ * @warning The memory pointed by output_content is not a copy, and is owned by the simulation kernel. It will be
+ * freed by the simulation kernel after the function returns.
+ * @warning The function shall not perform any memory-managed allocation (e.g. using rs_malloc).
+ */
+typedef void (*OutputCallback_t)(lp_id_t me, unsigned output_type, const void *output_content, unsigned output_size);
+
 /// @brief Internal event types used by the simulation kernel.
 ///
 /// These event types are automatically scheduled to the model according to the following logic:
@@ -91,6 +108,19 @@ extern void ScheduleNewEvent(lp_id_t receiver, simtime_t timestamp, unsigned eve
 
 extern void SetState(void *new_state);
 
+/**
+ * @brief API to schedule a new output event to execute once the event being currently executed is committed
+ *
+ * This function makes a copy of the data in @p output_content , which stays owned by the caller.
+ *
+ * @param output_type Numerical output type to be passed to the output handling function
+ * @param output_content The output content
+ * @param output_size The size (in bytes) of the output content
+ *
+ * @warning This function shall not be called during the handling of LP_FINI events, as it will produce no output.
+ */
+extern void ScheduleOutput(unsigned output_type, const void *output_content, unsigned output_size);
+
 /**
  * @brief Allocates rollbackable memory
  *
@@ -206,6 +236,8 @@ struct simulation_configuration {
 	ProcessEvent_t dispatcher;
 	/// Function pointer to the termination detection function
 	CanEnd_t committed;
+	/// Function pointer to the output handling function
+	OutputCallback_t output_callback;
 };
 
 extern int RootsimInit(const struct simulation_configuration *conf);

src/core/output.c

@@ -0,0 +1,88 @@
+/**
+ * @file core/output.c
+ *
+ * @brief Committed output management functions
+ *
+ * This module implements the facilities for committed output
+ *
+ * SPDX-FileCopyrightText: 2008-2023 HPDCS Group <rootsim@googlegroups.com>
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
+#include <ROOT-Sim.h>
+#include <core/core.h>
+#include <core/output.h>
+#include <lp/common.h>
+#include <lp/msg.h>
+#include <lp/process.h>
+
+void ScheduleOutput(unsigned output_type, const void *output_content, unsigned output_size)
+{
+	if(unlikely(silent_processing))
+		return;
+
+	if(unlikely(global_config.synchronization == SERIAL)) {
+		global_config.output_callback(current_msg->dest, output_type, output_content, output_size);
+		return;
+	}
+
+	char *content = mm_alloc(output_size);
+	memcpy(content, output_content, output_size);
+
+	struct output_data data = {.type = output_type, .content = content, .size = output_size};
+
+	output_array_t *outputs = current_msg->outputs;
+	if(!outputs) {
+		outputs = mm_alloc(sizeof(output_array_t));
+		array_init(*outputs);
+		current_msg->outputs = outputs;
+	}
+
+	array_push(*outputs, data);
+}
+
+void execute_outputs(struct lp_msg *msg)
+{
+	output_array_t *outputs = msg->outputs;
+	if(!outputs)
+		return;
+
+	for(array_count_t i = 0; i < array_count(*outputs); ++i) {
+		struct output_data data = array_get_at(*outputs, i);
+		global_config.output_callback(msg->dest, data.type, data.content, data.size);
+		mm_free(data.content);
+	}
+
+	/* Use array_clear() rather than array_fini(): the message buffer remains alive in the
+	 * system and will eventually reach free_msg_outputs(), which performs the final
+	 * array_fini() + mm_free(). When called right before msg_allocator_free() (e.g. from
+	 * fossil_lp_collect()), the array_clear() is redundant but harmless. */
+	array_clear(*outputs);
+}
+
+void free_msg_outputs(output_array_t *output_array)
+{
+	if(!output_array)
+		return;
+
+	for(array_count_t i = 0; i < array_count(*output_array); ++i) {
+		struct output_data data = array_get_at(*output_array, i);
+		mm_free(data.content);
+	}
+
+	array_fini(*output_array);
+	mm_free(output_array);
+}
+
+void committed_output_on_rollback(struct lp_msg *msg)
+{
+	output_array_t *outputs = msg->outputs;
+	if(!outputs)
+		return;
+
+	for(array_count_t i = 0; i < array_count(*outputs); ++i) {
+		struct output_data data = array_get_at(*outputs, i);
+		mm_free(data.content);
+	}
+
+	array_clear(*outputs);
+}

src/core/output.h

@@ -0,0 +1,45 @@
+/**
+ * @file core/output.h
+ *
+ * @brief Committed output management functions
+ *
+ * Committed output management functions
+ *
+ * SPDX-FileCopyrightText: 2008-2023 HPDCS Group <rootsim@googlegroups.com>
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
+#pragma once
+
+#include <datatypes/array.h>
+
+struct output_data {
+	unsigned type;
+	void *content;
+	unsigned size;
+};
+
+typedef array_declare(struct output_data) output_array_t;
+
+/**
+ * @brief Free the outputs stored for later from a message
+ * @param output_array the output_data from the message
+ */
+void free_msg_outputs(output_array_t *output_array);
+
+struct lp_msg;
+
+/**
+ * @brief Invoke the output callback on all outputs stored in the message.
+ * @param msg the message to execute outputs from
+ */
+void execute_outputs(struct lp_msg *msg);
+
+/**
+ * @brief Handle rolling-back of a message
+ * @param msg the rolled-back
+ *
+ * This function is called when a message is rolled back. It frees the output data
+ * stored in the message, as they are no longer valid.
+ * The output array is not freed, as it can still be used in the future.
+ */
+void committed_output_on_rollback(struct lp_msg *msg);

src/gvt/fossil.c

@@ -10,6 +10,7 @@
 #include <gvt/fossil.h>
 
 #include <mm/msg_allocator.h>
+#include <core/output.h>
 
 _Thread_local unsigned fossil_epoch_current;
 /// The value of the last GVT, kept here for easier fossil collection operations
@@ -50,12 +51,14 @@ void fossil_lp_collect(struct lp_ctx *lp)
 
 	past_i = model_allocator_fossil_lp_collect(&lp->mm_state, past_i + 1);
 
-	for(array_count_t k = past_i; k;) {
-		struct pes_entry e = array_get_at(proc_ctx->pes, --k);
+	for(array_count_t k = 0; k < past_i; ++k) {
+		struct pes_entry e = array_get_at(proc_ctx->pes, k);
+
 		if(pes_entry_is_sent_local(e))
 			continue;
 
 		struct lp_msg *m = pes_entry_msg(e);
+		execute_outputs(m);
 		msg_allocator_free(m);
 	}
 	array_truncate_first(proc_ctx->pes, past_i);

src/lp/common.c

@@ -1,21 +1,19 @@
 /**
-* @file lp/common.c
-*
-* @brief Common LP and message functionalities
-*
-* SPDX-FileCopyrightText: 2008-2025 HPCS Group <rootsim@googlegroups.com>
-* SPDX-License-Identifier: GPL-3.0-only
-*/
+ * @file lp/common.c
+ *
+ * @brief Common LP and message functionalities
+ *
+ * SPDX-FileCopyrightText: 2008-2025 HPCS Group <rootsim@googlegroups.com>
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
 #include <lp/common.h>
 
 #include <serial/serial.h>
 
-#ifndef NDEBUG
-_Thread_local const struct lp_msg *current_msg;
-#endif
+_Thread_local struct lp_msg *current_msg;
 
-void ScheduleNewEvent(const lp_id_t receiver, const simtime_t timestamp, const unsigned event_type,
-    const void *payload, const unsigned payload_size)
+void ScheduleNewEvent(const lp_id_t receiver, const simtime_t timestamp, const unsigned event_type, const void *payload,
+    const unsigned payload_size)
 {
 #ifndef NDEBUG
 	if(unlikely(event_type >= LP_INIT)) {

src/lp/common.h

@@ -15,11 +15,8 @@
 #include <log/stats.h>
 #include <mm/msg_allocator.h>
 
-#ifndef NDEBUG
 /// The currently processed message
-/** This is not necessary for normal operation, but it's useful in debug */
-extern __thread const struct lp_msg *current_msg;
-#endif
+extern _Thread_local struct lp_msg *current_msg;
 
 /**
  * @brief Process a message for a given LP (Logical Process)
@@ -30,16 +27,12 @@ extern __thread const struct lp_msg *current_msg;
  * @param lp A pointer to the LP associated with the message.
  * @param msg A pointer to the message to be processed.
  */
-static inline void common_msg_process(const struct lp_ctx *lp, const struct lp_msg *msg)
+static inline void common_msg_process(const struct lp_ctx *lp, struct lp_msg *msg)
 {
 	timer_uint t = timer_hr_new();
-#ifndef NDEBUG
 	current_msg = msg;
-#endif
 	global_config.dispatcher(msg->dest, msg->dest_t, msg->m_type, msg->pl, msg->pl_size, lp->state_pointer);
-#ifndef NDEBUG
 	current_msg = NULL;
-#endif
 	stats_take(STATS_MSG_PROCESSED_TIME, timer_hr_value(t));
 	stats_take(STATS_MSG_PROCESSED, 1);
 }

src/lp/msg.h

@@ -11,6 +11,7 @@
 #pragma once
 
 #include <core/core.h>
+#include <core/output.h>
 
 #include <limits.h>
 #include <stdatomic.h>
@@ -59,6 +60,8 @@
 struct lp_msg {
 	/// The next element in the message list (used in the message queue)
 	struct lp_msg *next;
+	/// Data for committed output — kept in the preamble so it is not transmitted over MPI
+	output_array_t *outputs;
 	/// The id of the recipient LP
 	lp_id_t dest;
 	/// The intended destination logical time of this message

src/lp/process.c

@@ -21,10 +21,9 @@
 #include <lp/lp.h>
 #include <mm/checkpoint/checkpoint.h>
 #include <mm/msg_allocator.h>
-#include <serial/serial.h>
 
 /// The flag used in ScheduleNewEvent() to keep track of silent execution
-static _Thread_local bool silent_processing = false;
+_Thread_local bool silent_processing = false;
 
 /**
  * @brief Schedule a new event. Parallel (Time Warp) version.
@@ -96,6 +95,8 @@ void process_lp_fini(struct lp_ctx *lp)
 		if(pes_entry_is_sent_local(e))
 			continue;
 
+		execute_outputs(pes_entry_msg(e));
+
 		if(pes_entry_is_sent_remote(e) ||
 		    !(atomic_load_explicit(&pes_entry_msg_received(e)->flags, memory_order_relaxed) & MSG_FLAG_ANTI))
 			msg_allocator_free(pes_entry_msg(e));
@@ -166,6 +167,7 @@ static inline void send_anti_messages(struct process_ctx *msg_processing, const
 
 		struct lp_msg *msg = pes_entry_msg_received(e);
 		const uint64_t f = atomic_fetch_add_explicit(&msg->flags, -MSG_FLAG_PROCESSED, memory_order_relaxed);
+		committed_output_on_rollback(msg);
 		if(!(f & MSG_FLAG_ANTI))
 			msg_queue_insert_self(msg);
 

src/lp/process.h

@@ -97,6 +97,8 @@ struct process_ctx {
 
 struct lp_ctx; // forward declaration
 
+extern _Thread_local bool silent_processing;
+
 extern void process_lp_init(struct lp_ctx *lp);
 extern void process_lp_fini(struct lp_ctx *lp);
 

src/mm/msg_allocator.c

@@ -67,6 +67,7 @@ struct lp_msg *msg_allocator_alloc(const unsigned payload_size)
 		ret = array_pop(free_list);
 	}
 	ret->pl_size = payload_size;
+	ret->outputs = NULL;
 	return ret;
 }
 
@@ -76,6 +77,7 @@ struct lp_msg *msg_allocator_alloc(const unsigned payload_size)
  */
 void msg_allocator_free(struct lp_msg *msg)
 {
+	free_msg_outputs(msg->outputs);
 	if(likely(msg->pl_size <= MSG_PAYLOAD_BASE_SIZE))
 		array_push(free_list, msg);
 	else

src/serial/serial.c

@@ -82,7 +82,7 @@ static int serial_simulation_run(void)
 	lp_id_t to_terminate = global_config.lps;
 
 	while(likely(!heap_is_empty(queue))) {
-		const struct lp_msg *msg = heap_min(queue);
+		struct lp_msg *msg = heap_min(queue);
 		struct lp_ctx *lp = &lps[msg->dest];
 		current_lp = lp;
 

test/CMakeLists.txt

@@ -45,3 +45,5 @@ test_program(correctness_parallel integration/correctness/timewarp.c integration
 test_program_link_libraries(correctness_parallel rscore)
 test_program(phold integration/phold.c)
 test_program_link_libraries(phold rscore)
+test_program(committed_output core/output.c)
+test_program_link_libraries(committed_output rscore)