From 436745ce90211db949d79657f1533598fc4d944b Mon Sep 17 00:00:00 2001
From: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
Date: Tue, 6 May 2025 17:25:30 +0200
Subject: [PATCH 01/13] [FEAT][WIP] Initial work towards committed output

Define code structure and developer contract
Work needs to be done.
Needs testing.

Signed-off-by: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
---
 src/CMakeLists.txt     |  1 +
 src/ROOT-Sim.h         | 28 +++++++++++++++++
 src/core/output.c      | 71 ++++++++++++++++++++++++++++++++++++++++++
 src/core/output.h      | 27 ++++++++++++++++
 src/lp/msg.h           |  3 ++
 src/lp/process.c       | 12 ++-----
 src/mm/msg_allocator.c |  2 ++
 src/mm/msg_allocator.h |  1 +
 8 files changed, 136 insertions(+), 9 deletions(-)
 create mode 100644 src/core/output.c
 create mode 100644 src/core/output.h

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 6bce4eba..0f0fae1d 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -7,6 +7,7 @@ set(rscore_srcs
         core/core.c
         init.c
         core/sync.c
+        core/output.c
         datatypes/msg_queue.c
         distributed/control_msg.c
         gvt/fossil.c
diff --git a/src/ROOT-Sim.h b/src/ROOT-Sim.h
index 8b861e55..b68da469 100644
--- a/src/ROOT-Sim.h
+++ b/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 (*PerformOutput_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,15 @@ 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 be executed only once the event being currently executed is committed
+ *
+ * @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
+ */
+extern void ScheduleOutput(unsigned output_type, const void *output_content, unsigned output_size);
+
 /**
  * @brief Allocates rollbackable memory
  *
@@ -206,6 +232,8 @@ struct simulation_configuration {
 	ProcessEvent_t dispatcher;
 	/// Function pointer to the termination detection function
 	CanEnd_t committed;
+	/// Function pointer to the output handling function
+	PerformOutput_t perform_output;
 };
 
 extern int RootsimInit(const struct simulation_configuration *conf);
diff --git a/src/core/output.c b/src/core/output.c
new file mode 100644
index 00000000..ca12f706
--- /dev/null
+++ b/src/core/output.c
@@ -0,0 +1,71 @@
+#include <ROOT-Sim.h>
+#include <core/core.h>
+#include <core/output.h>
+#include <lp/msg.h>
+
+extern __thread bool silent_processing;
+extern __thread struct lp_msg *current_msg;
+
+void ScheduleOutput(unsigned output_type, const void *output_content, unsigned output_size)
+{
+	if(unlikely(global_config.serial)) {
+		// Chiama direttamente il dispatcher
+		return;
+	}
+
+	if(unlikely(silent_processing))
+		return;
+
+	char *content = mm_alloc(output_size);
+
+	if(__builtin_expect(output_size && !content, 0)) {
+		logger(LOG_FATAL, "Out of memory!");
+		abort(); // TODO: this can be criticized as xmalloc() in gcc. We shall dump partial stats before.
+	}
+	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)); // Alloc array and then init
+		if(__builtin_expect(!outputs, 0)) {
+			logger(LOG_FATAL, "Out of memory!");
+			abort(); // TODO: this can be criticized as xmalloc() in gcc. We shall dump partial stats
+			         // before.
+		}
+		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.perform_output(msg->dest, data.type, data.content, data.size);
+		mm_free(data.content);
+	}
+
+	array_count(*outputs) = 0;
+}
+
+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);
+}
diff --git a/src/core/output.h b/src/core/output.h
new file mode 100644
index 00000000..a67214e7
--- /dev/null
+++ b/src/core/output.h
@@ -0,0 +1,27 @@
+/**
+ * @file coure/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 dyn_array(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);
diff --git a/src/lp/msg.h b/src/lp/msg.h
index 5ef0d035..f17d8f36 100644
--- a/src/lp/msg.h
+++ b/src/lp/msg.h
@@ -11,6 +11,7 @@
 #pragma once
 
 #include <core/core.h>
+#include <core/output.h>
 
 #include <limits.h>
 #include <stdatomic.h>
@@ -81,6 +82,8 @@ struct lp_msg {
 	uint32_t m_type;
 	/// The message payload size
 	uint32_t pl_size;
+	/// Data for committed output
+	output_array_t *outputs;
 	/// The initial part of the payload
 	unsigned char pl[MSG_PAYLOAD_BASE_SIZE];
 	/// The continuation of the payload
diff --git a/src/lp/process.c b/src/lp/process.c
index acc597a8..ecdce80b 100644
--- a/src/lp/process.c
+++ b/src/lp/process.c
@@ -23,12 +23,10 @@
 #include <serial/serial.h>
 
 /// The flag used in ScheduleNewEvent() to keep track of silent execution
-static _Thread_local bool silent_processing = false;
-#ifndef NDEBUG
+__thread bool silent_processing = false;
 /// The currently processed message
-/** This is not necessary for normal operation, but it's useful in debug */
-static _Thread_local struct lp_msg *current_msg;
-#endif
+/** Necessary for committed output */
+__thread struct lp_msg *current_msg;
 
 /**
  * @brief Marks a message as remote.
@@ -117,9 +115,7 @@ void process_lp_init(struct lp_ctx *lp)
 
 	struct lp_msg *msg = msg_allocator_pack(lp - lps, 0, LP_INIT, NULL, 0U);
 	msg->raw_flags = MSG_FLAG_PROCESSED;
-#ifndef NDEBUG
 	current_msg = msg;
-#endif
 	current_lp = lp;
 	common_msg_process(lp, msg);
 	lp->p.bound = 0.0;
@@ -407,9 +403,7 @@ void process_msg(void)
 	if(unlikely(lp->p.bound >= msg->dest_t && msg_is_before(msg, array_peek(lp->p.p_msgs))))
 		handle_straggler_msg(lp, msg);
 
-#ifndef NDEBUG
 	current_msg = msg;
-#endif
 
 	common_msg_process(lp, msg);
 	lp->p.bound = msg->dest_t;
diff --git a/src/mm/msg_allocator.c b/src/mm/msg_allocator.c
index 209f6ea5..cf378bbd 100644
--- a/src/mm/msg_allocator.c
+++ b/src/mm/msg_allocator.c
@@ -70,6 +70,8 @@ struct lp_msg *msg_allocator_alloc(const unsigned payload_size)
  */
 void msg_allocator_free(struct lp_msg *msg)
 {
+	free_msg_outputs(msg->outputs);
+	msg->outputs = NULL;
 	if(likely(msg->pl_size <= MSG_PAYLOAD_BASE_SIZE))
 		array_push(free_list, msg);
 	else
diff --git a/src/mm/msg_allocator.h b/src/mm/msg_allocator.h
index add91269..00f4d5e8 100644
--- a/src/mm/msg_allocator.h
+++ b/src/mm/msg_allocator.h
@@ -44,6 +44,7 @@ static inline struct lp_msg *msg_allocator_pack(lp_id_t receiver, simtime_t time
 	msg->dest = receiver;
 	msg->dest_t = timestamp;
 	msg->m_type = event_type;
+	msg->outputs = NULL;
 
 	if(likely(payload_size))
 		memcpy(msg->pl, payload, payload_size);

From 2a6d5c241fc8b895d56e37c8021c841c403c5e58 Mon Sep 17 00:00:00 2001
From: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
Date: Thu, 8 May 2025 18:57:26 +0200
Subject: [PATCH 02/13] Remove redundant allocation code

Signed-off-by: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
---
 src/core/output.c | 12 +-----------
 1 file changed, 1 insertion(+), 11 deletions(-)

diff --git a/src/core/output.c b/src/core/output.c
index ca12f706..6f6599f7 100644
--- a/src/core/output.c
+++ b/src/core/output.c
@@ -17,23 +17,13 @@ void ScheduleOutput(unsigned output_type, const void *output_content, unsigned o
 		return;
 
 	char *content = mm_alloc(output_size);
-
-	if(__builtin_expect(output_size && !content, 0)) {
-		logger(LOG_FATAL, "Out of memory!");
-		abort(); // TODO: this can be criticized as xmalloc() in gcc. We shall dump partial stats before.
-	}
 	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)); // Alloc array and then init
-		if(__builtin_expect(!outputs, 0)) {
-			logger(LOG_FATAL, "Out of memory!");
-			abort(); // TODO: this can be criticized as xmalloc() in gcc. We shall dump partial stats
-			         // before.
-		}
+		outputs = mm_alloc(sizeof(output_array_t));
 		array_init(*outputs);
 		current_msg->outputs = outputs;
 	}

From c470a826746510cf680a5b4a54fd8cf67d911e34 Mon Sep 17 00:00:00 2001
From: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
Date: Fri, 9 May 2025 19:41:55 +0200
Subject: [PATCH 03/13] Finished committed output

Signed-off-by: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
---
 src/ROOT-Sim.h    |  2 ++
 src/core/output.c | 17 ++++++++++++++++-
 src/core/output.h |  8 ++++++++
 src/gvt/fossil.c  | 18 ++++++++++++++----
 src/lp/process.c  |  8 ++++++--
 5 files changed, 46 insertions(+), 7 deletions(-)

diff --git a/src/ROOT-Sim.h b/src/ROOT-Sim.h
index b68da469..328def04 100644
--- a/src/ROOT-Sim.h
+++ b/src/ROOT-Sim.h
@@ -114,6 +114,8 @@ extern void SetState(void *new_state);
  * @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);
 
diff --git a/src/core/output.c b/src/core/output.c
index 6f6599f7..9340a9ac 100644
--- a/src/core/output.c
+++ b/src/core/output.c
@@ -2,6 +2,7 @@
 #include <core/core.h>
 #include <core/output.h>
 #include <lp/msg.h>
+#include <lp/process.h>
 
 extern __thread bool silent_processing;
 extern __thread struct lp_msg *current_msg;
@@ -43,7 +44,7 @@ void execute_outputs(struct lp_msg *msg)
 		mm_free(data.content);
 	}
 
-	array_count(*outputs) = 0;
+	array_count(*outputs) = 0; // Necessary to avoid double freeing the array elements.
 }
 
 void free_msg_outputs(output_array_t *output_array)
@@ -59,3 +60,17 @@ void free_msg_outputs(output_array_t *output_array)
 	array_fini(*output_array);
 	mm_free(output_array);
 }
+
+void execute_outputs_batch(struct lp_msg **msg_array, array_count_t size)
+{
+	// Since dyn_arrays are unnamed structs, we work with the items element
+	for(array_count_t i = 0; i < size; i++) {
+		struct lp_msg *marked_msg = msg_array[i];
+		struct lp_msg *msg = unmark_msg(marked_msg);
+		if(msg->dest_t > global_config.termination_time) {
+			return;
+		}
+		if(is_msg_past(marked_msg))
+			execute_outputs(msg);
+	}
+}
diff --git a/src/core/output.h b/src/core/output.h
index a67214e7..a7931638 100644
--- a/src/core/output.h
+++ b/src/core/output.h
@@ -25,3 +25,11 @@ typedef dyn_array(struct output_data) output_array_t;
  * @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);
diff --git a/src/gvt/fossil.c b/src/gvt/fossil.c
index 0e545dd2..7cd9e408 100644
--- a/src/gvt/fossil.c
+++ b/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
@@ -48,11 +49,20 @@ void fossil_lp_collect(struct lp_ctx *lp)
 
 	past_i = model_allocator_fossil_lp_collect(&lp->mm_state, past_i + 1);
 
-	array_count_t k = past_i;
-	while(k--) {
+	array_count_t k = 0;
+	while(k < past_i) {
 		struct lp_msg *msg = array_get_at(proc_p->p_msgs, k);
-		if(!is_msg_local_sent(msg))
-			msg_allocator_free(unmark_msg(msg));
+		bool past = is_msg_past(msg);
+		bool local_sent = is_msg_local_sent(msg);
+		msg = unmark_msg(msg);
+
+		if(past)
+			execute_outputs(msg);
+
+		if(!local_sent)
+			msg_allocator_free(msg);
+
+		k++;
 	}
 	array_truncate_first(proc_p->p_msgs, past_i);
 
diff --git a/src/lp/process.c b/src/lp/process.c
index ecdce80b..06152a3f 100644
--- a/src/lp/process.c
+++ b/src/lp/process.c
@@ -138,9 +138,13 @@ void process_lp_fini(struct lp_ctx *lp)
 		if(is_msg_local_sent(msg))
 			continue;
 
-		const bool remote = is_msg_remote(msg);
+		bool remote = is_msg_remote(msg);
+		bool past = is_msg_past(msg);
 		msg = unmark_msg(msg);
-		const uint32_t flags = atomic_load_explicit(&msg->flags, memory_order_relaxed);
+		if(past)
+			execute_outputs(msg);
+
+		uint32_t flags = atomic_load_explicit(&msg->flags, memory_order_relaxed);
 		if(remote || !(flags & MSG_FLAG_ANTI))
 			msg_allocator_free(msg);
 	}

From 26009e11df2e9fbbfb49e7d2192c5ea2ee3d26d2 Mon Sep 17 00:00:00 2001
From: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
Date: Tue, 13 May 2025 13:29:15 +0200
Subject: [PATCH 04/13] FIX properly free committed output data upon message
 rollback

Signed-off-by: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
---
 src/core/output.c | 17 ++++++++++++++++-
 src/core/output.h | 10 ++++++++++
 src/lp/process.c  |  3 ++-
 3 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/src/core/output.c b/src/core/output.c
index 9340a9ac..d67032b1 100644
--- a/src/core/output.c
+++ b/src/core/output.c
@@ -10,7 +10,8 @@ extern __thread struct lp_msg *current_msg;
 void ScheduleOutput(unsigned output_type, const void *output_content, unsigned output_size)
 {
 	if(unlikely(global_config.serial)) {
-		// Chiama direttamente il dispatcher
+		// Call the dispatcher directly
+		// TODO
 		return;
 	}
 
@@ -61,6 +62,20 @@ void free_msg_outputs(output_array_t *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_count(*outputs) = 0; // Necessary to avoid double freeing the array elements.
+}
+
 void execute_outputs_batch(struct lp_msg **msg_array, array_count_t size)
 {
 	// Since dyn_arrays are unnamed structs, we work with the items element
diff --git a/src/core/output.h b/src/core/output.h
index a7931638..f4c82cb6 100644
--- a/src/core/output.h
+++ b/src/core/output.h
@@ -33,3 +33,13 @@ struct lp_msg;
  * @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);
diff --git a/src/lp/process.c b/src/lp/process.c
index 06152a3f..31e77f65 100644
--- a/src/lp/process.c
+++ b/src/lp/process.c
@@ -210,7 +210,8 @@ static inline void send_anti_messages(struct process_ctx *msg_processing, const
 			msg = array_get_at(msg_processing->p_msgs, ++i);
 		}
 
-		const uint32_t f = atomic_fetch_add_explicit(&msg->flags, -MSG_FLAG_PROCESSED, memory_order_relaxed);
+		uint32_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);
 		stats_take(STATS_MSG_ROLLBACK, 1);

From e76ab9419027ed1d45e8709805c5691543683926 Mon Sep 17 00:00:00 2001
From: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
Date: Tue, 13 May 2025 15:16:04 +0200
Subject: [PATCH 05/13] Add file header to output.c

Signed-off-by: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
---
 src/core/output.c | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/src/core/output.c b/src/core/output.c
index d67032b1..7e310406 100644
--- a/src/core/output.c
+++ b/src/core/output.c
@@ -1,3 +1,13 @@
+/**
+ * @file 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>

From 445a153bc67bf3778fbe37ab782d7a0315e7a1c9 Mon Sep 17 00:00:00 2001
From: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
Date: Tue, 13 May 2025 15:41:21 +0200
Subject: [PATCH 06/13] Sequential simulation executes committed output
 immediately

Plus changed some docstring wording

Signed-off-by: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
---
 src/ROOT-Sim.h    | 6 ++++--
 src/core/output.c | 3 +--
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/src/ROOT-Sim.h b/src/ROOT-Sim.h
index 328def04..0b6f517b 100644
--- a/src/ROOT-Sim.h
+++ b/src/ROOT-Sim.h
@@ -109,12 +109,14 @@ 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 be executed only once the event being currently executed is committed
+ * @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);
diff --git a/src/core/output.c b/src/core/output.c
index 7e310406..8844bc89 100644
--- a/src/core/output.c
+++ b/src/core/output.c
@@ -20,8 +20,7 @@ extern __thread struct lp_msg *current_msg;
 void ScheduleOutput(unsigned output_type, const void *output_content, unsigned output_size)
 {
 	if(unlikely(global_config.serial)) {
-		// Call the dispatcher directly
-		// TODO
+		global_config.perform_output(current_msg->dest, output_type, output_content, output_size);
 		return;
 	}
 

From 3deddfdd60da2a49c2e4ecf24036107841284a34 Mon Sep 17 00:00:00 2001
From: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
Date: Fri, 23 May 2025 15:39:03 +0200
Subject: [PATCH 07/13] Test for committed output

Added test for committed output in presence of stragglers. Requires 2 threads.

Signed-off-by: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
---
 test/CMakeLists.txt |   2 +
 test/core/output.c  | 278 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 280 insertions(+)
 create mode 100644 test/core/output.c

diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index 9ac3ad43..410727cc 100644
--- a/test/CMakeLists.txt
+++ b/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)
diff --git a/test/core/output.c b/test/core/output.c
new file mode 100644
index 00000000..4767b2b3
--- /dev/null
+++ b/test/core/output.c
@@ -0,0 +1,278 @@
+/**
+ * @file test/core/output.c
+ *
+ * @brief Test: committed output in presence of stragglers
+ *
+ * This test checks that the committed output is triggered correctly, even in
+ * the presence of stragglers. It uses two LPs, where one waits for the other to
+ * finish before sending a straggler event.
+ *
+ * SPDX-FileCopyrightText: 2008-2025 HPDCS Group <rootsim@googlegroups.com>
+ * SPDX-License-Identifier: GPL-3.0-only
+ */
+#include <ROOT-Sim.h>
+#include <test.h>
+
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdatomic.h>
+#include <string.h>
+#include <assert.h>
+
+#define NUM_LPS 2
+#define NUM_THREADS 2
+
+#define EVENT 1
+#define STRAGGLER_EVENT 2
+
+struct lp_state {
+	unsigned long count;
+};
+
+#define OUTPUT_TYPE 0
+#define OUTPUT_FROM_STRAGGLER 1
+
+#define OUT_SLOTS 100
+#define OUT_SZ 1024
+
+struct output_data {
+	lp_id_t id;
+	unsigned long count;
+};
+
+atomic_bool lp1_turn = false;
+
+char lp_outs[2][OUT_SLOTS][OUT_SZ];
+size_t lp_outs_count[2] = {0, 0};
+
+void ProcessEvent(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s);
+bool CanEnd(lp_id_t me, const void *snapshot);
+void PerformOutput(lp_id_t me, unsigned output_type, const void *output_content, unsigned output_size);
+
+struct simulation_configuration conf = {
+    .lps = NUM_LPS,
+    .n_threads = NUM_THREADS,
+    .termination_time = 25,
+    .gvt_period = 10,
+    .log_level = LOG_INFO,
+    .stats_file = "output_test",
+    .ckpt_interval = 0,
+    .core_binding = true,
+    .serial = false,
+    .dispatcher = ProcessEvent,
+    .committed = CanEnd,
+    .perform_output = PerformOutput,
+};
+
+#define lp0_max_count 20
+simtime_t lp0_times[lp0_max_count] = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20};
+#define lp1_max_count 10
+simtime_t lp1_times[lp1_max_count] = {1.5, 3.5, 5.5, 7.5, 9.5, 11.5, 13.5, 15.5, 17.5, 19.5};
+simtime_t lp1_send_delay = 0.2;
+
+void Handler0(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
+{
+	(void)now;
+	(void)content;
+	(void)size;
+
+	struct lp_state *state = (struct lp_state *)s;
+
+	if(state->count >= lp0_max_count || event_type == LP_FINI) {
+		atomic_store(&lp1_turn, true);
+		return;
+	}
+
+	while(atomic_load(&lp1_turn)) {
+		// Wait for lp1 to schedule the straggler
+	}
+
+	if(event_type == EVENT) {
+		ScheduleOutput(OUTPUT_TYPE, &(struct output_data){.id = me, .count = state->count},
+		    sizeof(struct output_data));
+		// printf("[Base event] ID%lu, ct%lu\n", me, state->count);
+		ScheduleNewEvent(0, lp0_times[state->count], EVENT, NULL, 0);
+		state->count++;
+
+	} else if(event_type == STRAGGLER_EVENT) {
+		// printf("[Straggler received] t%.1lf, ID%lu, ct%lu\n", now, me, state->count);
+		ScheduleOutput(OUTPUT_FROM_STRAGGLER, &(struct output_data){.id = me, .count = state->count},
+		    sizeof(struct output_data));
+	}
+}
+
+void Handler1(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
+{
+	(void)content;
+	(void)size;
+
+	struct lp_state *state = (struct lp_state *)s;
+
+	if(state->count >= lp1_max_count || event_type == LP_FINI) {
+		return;
+	}
+
+	while(!atomic_load(&lp1_turn)) {
+		// Wait for lp0 to finish
+	}
+
+	if(event_type == EVENT) {
+		ScheduleNewEvent(1, lp1_times[state->count], EVENT, NULL, 0);
+		ScheduleNewEvent(0, now + lp1_send_delay, STRAGGLER_EVENT, NULL, 0);
+
+		// printf("[Straggler generator] s%.1lf->d%.1lf, ID%lu, ct%lu\n", now, now + lp1_send_delay, me,
+		// state->count);
+		ScheduleOutput(OUTPUT_TYPE, &(struct output_data){.id = me, .count = state->count},
+		    sizeof(struct output_data));
+		state->count++;
+	}
+
+	atomic_store(&lp1_turn, false);
+}
+
+void ProcessEvent(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
+{
+	struct lp_state *state = (struct lp_state *)s;
+
+	if(event_type != LP_FINI && now > conf.termination_time)
+		return;
+
+	if(event_type == LP_INIT) {
+		state = rs_malloc(sizeof(*state));
+		if(state == NULL)
+			abort();
+		SetState(state);
+		state->count = 0;
+
+		ScheduleNewEvent(me, 0, EVENT, NULL, 0);
+		return;
+	}
+
+	if(me == 0) {
+		Handler0(me, now, event_type, content, size, state);
+	} else if(me == 1) {
+		Handler1(me, now, event_type, content, size, state);
+	} else {
+		fprintf(stderr, "Unknown LP ID\n");
+		abort();
+	}
+}
+
+bool CanEnd(lp_id_t me, const void *snapshot)
+{
+	(void)me;
+	(void)snapshot;
+	return false;
+}
+
+void PerformOutput(lp_id_t me, unsigned output_type, const void *output_content, unsigned output_size)
+{
+	struct output_data *data = NULL;
+	lp_id_t id;
+	unsigned long count;
+	switch(output_type) {
+		case OUTPUT_TYPE:
+			data = (struct output_data *)output_content;
+			id = data->id;
+			count = data->count;
+			// printf("[Normal output] %lu,%lu,%lu\n", me, id, count);
+			snprintf(lp_outs[me][lp_outs_count[me]], OUT_SZ, "N%lu,%lu,%lu", me, id, count);
+			lp_outs_count[me]++;
+			break;
+
+		case OUTPUT_FROM_STRAGGLER:
+			data = (struct output_data *)output_content;
+			id = data->id;
+			count = data->count;
+			// printf("[Straggler output] %lu,%lu,%lu\n", me, id, count);
+			snprintf(lp_outs[me][lp_outs_count[me]], OUT_SZ, "S%lu,%lu,%lu", me, id, count);
+			lp_outs_count[me]++;
+			break;
+
+		default:
+			fprintf(stderr, "Unknown output type %u, with size %u\n", output_type, output_size);
+			break;
+	}
+}
+
+bool check_output(const char *output, const char *expected, size_t expected_len, char *file, int line)
+{
+	if(strncmp(output, expected, expected_len)) {
+		fprintf(stderr, "%s:%d: error: Output did not match expected. Actual: \"%s\", Expected:\"%s\".\n", file,
+		    line, output, expected);
+		test_fail();
+	}
+	return true;
+}
+
+#define chk_out(output, expected, expected_len) check_output(output, expected, expected_len, __FILE__, __LINE__)
+
+int perform_exec()
+{
+	RootsimInit(&conf);
+
+	for(size_t lp = 0; lp < 2; lp++) {
+		for(size_t i = 0; i < OUT_SLOTS; i++) {
+			memset(lp_outs[lp][i], 0, OUT_SZ);
+		}
+	}
+
+	RootsimRun();
+
+	// printf("OUTPUTS\n");
+	// for(size_t lp = 0; lp < 2; lp++) {
+	// 	for(size_t i = 0; i < OUT_SLOTS; i++) {
+	// 		printf("%s\n", lp_outs[lp][i]);
+	// 	}
+	// }
+
+	chk_out(lp_outs[0][0], "N0,0,0", 7);
+	chk_out(lp_outs[0][1], "S0,0,1", 7);
+	chk_out(lp_outs[0][2], "N0,0,1", 7);
+	chk_out(lp_outs[0][3], "S0,0,2", 7);
+	chk_out(lp_outs[0][4], "N0,0,2", 7);
+	chk_out(lp_outs[0][5], "N0,0,3", 7);
+	chk_out(lp_outs[0][6], "S0,0,4", 7);
+	chk_out(lp_outs[0][7], "N0,0,4", 7);
+	chk_out(lp_outs[0][8], "N0,0,5", 7);
+	chk_out(lp_outs[0][9], "S0,0,6", 7);
+	chk_out(lp_outs[0][10], "N0,0,6", 7);
+	chk_out(lp_outs[0][11], "N0,0,7", 7);
+	chk_out(lp_outs[0][12], "S0,0,8", 7);
+	chk_out(lp_outs[0][13], "N0,0,8", 7);
+	chk_out(lp_outs[0][14], "N0,0,9", 7);
+	chk_out(lp_outs[0][15], "S0,0,10", 8);
+	chk_out(lp_outs[0][16], "N0,0,10", 8);
+	chk_out(lp_outs[0][17], "N0,0,11", 8);
+	chk_out(lp_outs[0][18], "S0,0,12", 8);
+	chk_out(lp_outs[0][19], "N0,0,12", 8);
+	chk_out(lp_outs[0][20], "N0,0,13", 8);
+	chk_out(lp_outs[0][21], "S0,0,14", 8);
+	chk_out(lp_outs[0][22], "N0,0,14", 8);
+	chk_out(lp_outs[0][23], "N0,0,15", 8);
+	chk_out(lp_outs[0][24], "S0,0,16", 8);
+	chk_out(lp_outs[0][25], "N0,0,16", 8);
+	chk_out(lp_outs[0][26], "N0,0,17", 8);
+	chk_out(lp_outs[0][27], "S0,0,18", 8);
+	chk_out(lp_outs[0][28], "N0,0,18", 8);
+	chk_out(lp_outs[0][29], "N0,0,19", 8);
+
+	chk_out(lp_outs[1][0], "N1,1,0", 7);
+	chk_out(lp_outs[1][1], "N1,1,1", 7);
+	chk_out(lp_outs[1][2], "N1,1,2", 7);
+	chk_out(lp_outs[1][3], "N1,1,3", 7);
+	chk_out(lp_outs[1][4], "N1,1,4", 7);
+	chk_out(lp_outs[1][5], "N1,1,5", 7);
+	chk_out(lp_outs[1][6], "N1,1,6", 7);
+	chk_out(lp_outs[1][7], "N1,1,7", 7);
+	chk_out(lp_outs[1][8], "N1,1,8", 7);
+	chk_out(lp_outs[1][9], "N1,1,9", 7);
+
+	return 0;
+}
+
+int main(void)
+{
+	test("Testing committed output", perform_exec, NULL);
+}

From ac67e689f5bb8310dd7a8adaf923fccb45662d29 Mon Sep 17 00:00:00 2001
From: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
Date: Thu, 26 Feb 2026 01:07:17 +0100
Subject: [PATCH 08/13] Test committed output - set
 simulation_configuration.synchronization

Signed-off-by: AdrianoPi <24545691+AdrianoPi@users.noreply.github.com>
---
 test/core/output.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/core/output.c b/test/core/output.c
index 4767b2b3..bb2e7c59 100644
--- a/test/core/output.c
+++ b/test/core/output.c
@@ -59,7 +59,7 @@ struct simulation_configuration conf = {
     .stats_file = "output_test",
     .ckpt_interval = 0,
     .core_binding = true,
-    .serial = false,
+    .synchronization = TIME_WARP,
     .dispatcher = ProcessEvent,
     .committed = CanEnd,
     .perform_output = PerformOutput,

From 9747cc2726ffe4c1b62287b9f5f1b9261e3ae74a Mon Sep 17 00:00:00 2001
From: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
Date: Fri, 26 Jun 2026 10:37:42 +0200
Subject: [PATCH 09/13] Don't emit output if not supposed to

This commit avoids accessing output buffers if not necessary. Also, output buffers are initialized upon message creation.

Signed-off-by: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
---
 src/gvt/fossil.c | 2 +-
 src/lp/common.h  | 1 +
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/gvt/fossil.c b/src/gvt/fossil.c
index 628b6789..f4637736 100644
--- a/src/gvt/fossil.c
+++ b/src/gvt/fossil.c
@@ -53,12 +53,12 @@ void fossil_lp_collect(struct lp_ctx *lp)
 
 	for(array_count_t k = past_i; k;) {
 		struct pes_entry e = array_get_at(proc_ctx->pes, --k);
-		execute_outputs(pes_entry_msg(e));
 
 		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);
diff --git a/src/lp/common.h b/src/lp/common.h
index 9ac32987..f2a72cf1 100644
--- a/src/lp/common.h
+++ b/src/lp/common.h
@@ -58,6 +58,7 @@ static inline struct lp_msg *common_msg_pack(const lp_id_t receiver, const simti
 	msg->dest = receiver;
 	msg->dest_t = timestamp;
 	msg->m_type = event_type;
+	msg->outputs = NULL;
 
 	if(likely(payload_size))
 		memcpy(msg->pl, payload, payload_size);

From 4bdc96f6958feddbb596a29b96644df199eaffd6 Mon Sep 17 00:00:00 2001
From: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
Date: Sun, 28 Jun 2026 10:42:59 +0200
Subject: [PATCH 10/13] Fix bugs in committed output and improve its test

Three bugs are fixed in the committed output implementation:

- msg_allocator_alloc() did not initialize msg->outputs to NULL. Since
  mm_alloc() wraps malloc(), freshly allocated messages could contain a
  garbage pointer in that field. All functions guarding on if(!outputs)
  would then skip the NULL check and dereference the garbage value,
  leading to a crash or memory corruption.

- The outputs field in struct lp_msg was placed between pl_size and
  pl[], inside the MPI-transmitted region. When a message was sent to a
  remote node the local pointer value was serialized, and
  the receiving side would overwrite the field with that garbage value
  via MPI_Mrecv. The field is moved into the message preamble (before
  dest) so it is never transmitted.

- fossil_lp_collect() called execute_outputs() on every PES entry,
  including PES_ENTRY_SENT_LOCAL ones. Sent messages do not carry
  outputs; only received messages do. The call is moved after the
  sent-local guard and the already-extracted m pointer is reused to
  avoid a redundant pes_entry_msg() call.

Two additional correctness issues are also addressed in output.c:

- The silent_processing guard in ScheduleOutput() was checked after the
  serial fast-path. This meant that calling ScheduleOutput() during
  LP_FINI in serial mode would incorrectly fire the output callback,
  contrary to what the API documentation states. The guard is now
  checked first.

- The serial mode detection used the deprecated global_config.serial
  boolean instead of global_config.synchronization == SERIAL, leading to lose all
  output: ScheduleOutput() would buffer entries on current_msg->outputs
  that execute_outputs() never flushes in serial mode.

The committed_output test is extended with:

- A serial mode test case, verifying that the output callback fires
  immediately and in causal order.
- A total output count assertion in the Time Warp test that detects any
  phantom outputs from events that were rolled back but whose outputs
  were not correctly suppressed by committed_output_on_rollback().
- Per-invocation output_size verification in the output callback.
- Bounds checking on the output buffer before writing to it.

Signed-off-by: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
---
 src/core/output.c      |  19 +-
 src/core/output.h      |   2 +-
 src/gvt/fossil.c       |   2 +-
 src/lp/msg.h           |   4 +-
 src/mm/msg_allocator.c |   1 +
 test/core/output.c     | 453 ++++++++++++++++++++++++++++-------------
 6 files changed, 327 insertions(+), 154 deletions(-)

diff --git a/src/core/output.c b/src/core/output.c
index 472d5aee..be965e6f 100644
--- a/src/core/output.c
+++ b/src/core/output.c
@@ -1,5 +1,5 @@
 /**
- * @file output.c
+ * @file core/output.c
  *
  * @brief Committed output management functions
  *
@@ -17,13 +17,13 @@
 
 void ScheduleOutput(unsigned output_type, const void *output_content, unsigned output_size)
 {
-	if(unlikely(global_config.serial)) {
-		global_config.output_callback(current_msg->dest, output_type, output_content, output_size);
+	if(unlikely(silent_processing))
 		return;
-	}
 
-	if(unlikely(silent_processing))
+	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);
@@ -52,7 +52,11 @@ void execute_outputs(struct lp_msg *msg)
 		mm_free(data.content);
 	}
 
-	array_count(*outputs) = 0; // Necessary to avoid double freeing the array elements.
+	/* 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)
@@ -80,6 +84,5 @@ void committed_output_on_rollback(struct lp_msg *msg)
 		mm_free(data.content);
 	}
 
-	array_count(*outputs) = 0; // Necessary to avoid double freeing the array elements.
+	array_clear(*outputs);
 }
-
diff --git a/src/core/output.h b/src/core/output.h
index 5faa77c3..5c6fa97c 100644
--- a/src/core/output.h
+++ b/src/core/output.h
@@ -1,5 +1,5 @@
 /**
- * @file coure/output.h
+ * @file core/output.h
  *
  * @brief Committed output management functions
  *
diff --git a/src/gvt/fossil.c b/src/gvt/fossil.c
index 628b6789..f4637736 100644
--- a/src/gvt/fossil.c
+++ b/src/gvt/fossil.c
@@ -53,12 +53,12 @@ void fossil_lp_collect(struct lp_ctx *lp)
 
 	for(array_count_t k = past_i; k;) {
 		struct pes_entry e = array_get_at(proc_ctx->pes, --k);
-		execute_outputs(pes_entry_msg(e));
 
 		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);
diff --git a/src/lp/msg.h b/src/lp/msg.h
index 0e26a6fc..9bdcd1fe 100644
--- a/src/lp/msg.h
+++ b/src/lp/msg.h
@@ -60,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
@@ -80,8 +82,6 @@ struct lp_msg {
 	uint32_t m_type;
 	/// The message payload size
 	uint32_t pl_size;
-	/// Data for committed output
-	output_array_t *outputs;
 	/// The initial part of the payload
 	unsigned char pl[MSG_PAYLOAD_BASE_SIZE];
 	/// The continuation of the payload
diff --git a/src/mm/msg_allocator.c b/src/mm/msg_allocator.c
index fc5c97b6..e3d1fe96 100644
--- a/src/mm/msg_allocator.c
+++ b/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;
 }
 
diff --git a/test/core/output.c b/test/core/output.c
index f854624f..7b77986f 100644
--- a/test/core/output.c
+++ b/test/core/output.c
@@ -1,11 +1,22 @@
 /**
  * @file test/core/output.c
  *
- * @brief Test: committed output in presence of stragglers
+ * @brief Test: committed output correctness
  *
- * This test checks that the committed output is triggered correctly, even in
- * the presence of stragglers. It uses two LPs, where one waits for the other to
- * finish before sending a straggler event.
+ * This file contains two test cases for the committed output mechanism:
+ *
+ * 1. **Time Warp (parallel) test**: Uses two LPs where LP 1 busy-waits until
+ *    LP 0 has processed an event, then sends a straggler that causes LP 0 to
+ *    roll back. The busy-waiting plus core_binding=true force a deterministic
+ *    interleaving, yielding a fixed and verifiable committed output sequence.
+ *    Both the exact output content (per-entry) and the total output count are
+ *    checked; the latter catches any phantom outputs from rolled-back events.
+ *
+ * 2. **Serial test**: Verifies that ScheduleOutput() fires the output callback
+ *    immediately and in causal order during a fully sequential simulation.
+ *
+ * @note Run with AddressSanitizer (-fsanitize=address) to detect memory leaks
+ *       from the output content buffers.
  *
  * SPDX-FileCopyrightText: 2008-2025 HPDCS Group <rootsim@googlegroups.com>
  * SPDX-License-Identifier: GPL-3.0-only
@@ -20,58 +31,91 @@
 #include <string.h>
 #include <assert.h>
 
-#define NUM_LPS 2
-#define NUM_THREADS 2
+/* =========================================================================
+ * Shared event/output type constants and payload struct
+ * ========================================================================= */
 
 #define EVENT 1
 #define STRAGGLER_EVENT 2
 
-struct lp_state {
-	unsigned long count;
-};
-
 #define OUTPUT_TYPE 0
 #define OUTPUT_FROM_STRAGGLER 1
 
 #define OUT_SLOTS 100
-#define OUT_SZ 1024
+#define OUT_SZ 64
 
 struct output_data {
 	lp_id_t id;
 	unsigned long count;
 };
 
-atomic_bool lp1_turn = false;
-
-char lp_outs[2][OUT_SLOTS][OUT_SZ];
-size_t lp_outs_count[2] = {0, 0};
+/* =========================================================================
+ * Time Warp (parallel) test
+ * ========================================================================= */
 
-void ProcessEvent(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s);
-bool CanEnd(lp_id_t me, const void *snapshot);
-void PerformOutput(lp_id_t me, unsigned output_type, const void *output_content, unsigned output_size);
+#define TW_NUM_LPS 2
+#define TW_NUM_THREADS 2
 
-struct simulation_configuration conf = {
-    .lps = NUM_LPS,
-    .n_threads = NUM_THREADS,
-    .termination_time = 25,
-    .gvt_period = 10,
-    .log_level = LOG_INFO,
-    .stats_file = "output_test",
-    .ckpt_interval = 0,
-    .core_binding = true,
-    .synchronization = TIME_WARP,
-    .dispatcher = ProcessEvent,
-    .committed = CanEnd,
-    .output_callback = PerformOutput,
+struct lp_state {
+	unsigned long count;
 };
 
 #define lp0_max_count 20
-simtime_t lp0_times[lp0_max_count] = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20};
+static simtime_t lp0_times[lp0_max_count] = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20};
 #define lp1_max_count 10
-simtime_t lp1_times[lp1_max_count] = {1.5, 3.5, 5.5, 7.5, 9.5, 11.5, 13.5, 15.5, 17.5, 19.5};
-simtime_t lp1_send_delay = 0.2;
+static simtime_t lp1_times[lp1_max_count] = {1.5, 3.5, 5.5, 7.5, 9.5, 11.5, 13.5, 15.5, 17.5, 19.5};
+static simtime_t lp1_send_delay = 0.2;
+
+/* Atomic flag used to coordinate the deterministic interleaving between the
+ * two LPs. lp1_turn==true means LP 0 is paused and LP 1 may proceed. */
+static atomic_bool lp1_turn = false;
 
-void Handler0(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
+/* Output storage for the Time Warp test */
+static char tw_lp_outs[TW_NUM_LPS][OUT_SLOTS][OUT_SZ];
+static size_t tw_lp_outs_count[TW_NUM_LPS];
+/* Total number of callback invocations, used to detect phantom rollback outputs */
+static atomic_size_t tw_total_output_count;
+
+static void TW_PerformOutput(lp_id_t me, unsigned output_type, const void *output_content, unsigned output_size)
+{
+	/* Verify the size is always exactly what was scheduled */
+	if(output_size != sizeof(struct output_data)) {
+		fprintf(stderr, "PerformOutput: unexpected output_size %u (expected %zu)\n", output_size,
+		    sizeof(struct output_data));
+		test_fail();
+	}
+
+	/* Bounds guard: unexpected extra outputs would silently corrupt memory otherwise */
+	size_t idx = tw_lp_outs_count[me];
+	if(idx >= OUT_SLOTS) {
+		fprintf(stderr, "PerformOutput: too many outputs for LP %llu (>= %d)\n", (unsigned long long)me,
+		    OUT_SLOTS);
+		test_fail();
+	}
+
+	const struct output_data *data = output_content;
+	lp_id_t id = data->id;
+	unsigned long count = data->count;
+
+	switch(output_type) {
+		case OUTPUT_TYPE:
+			snprintf(tw_lp_outs[me][idx], OUT_SZ, "N%llu,%llu,%lu", (unsigned long long)me,
+			    (unsigned long long)id, count);
+			break;
+		case OUTPUT_FROM_STRAGGLER:
+			snprintf(tw_lp_outs[me][idx], OUT_SZ, "S%llu,%llu,%lu", (unsigned long long)me,
+			    (unsigned long long)id, count);
+			break;
+		default:
+			fprintf(stderr, "PerformOutput: unknown output_type %u\n", output_type);
+			test_fail();
+	}
+
+	tw_lp_outs_count[me]++;
+	atomic_fetch_add_explicit(&tw_total_output_count, 1, memory_order_relaxed);
+}
+
+static void TW_Handler0(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
 {
 	(void)now;
 	(void)content;
@@ -85,44 +129,37 @@ void Handler0(lp_id_t me, simtime_t now, unsigned event_type, const void *conten
 	}
 
 	while(atomic_load(&lp1_turn)) {
-		// Wait for lp1 to schedule the straggler
+		/* Spin until LP 1 has scheduled the straggler for this round */
 	}
 
 	if(event_type == EVENT) {
 		ScheduleOutput(OUTPUT_TYPE, &(struct output_data){.id = me, .count = state->count},
 		    sizeof(struct output_data));
-		// printf("[Base event] ID%lu, ct%lu\n", me, state->count);
 		ScheduleNewEvent(0, lp0_times[state->count], EVENT, NULL, 0);
 		state->count++;
-
 	} else if(event_type == STRAGGLER_EVENT) {
-		// printf("[Straggler received] t%.1lf, ID%lu, ct%lu\n", now, me, state->count);
 		ScheduleOutput(OUTPUT_FROM_STRAGGLER, &(struct output_data){.id = me, .count = state->count},
 		    sizeof(struct output_data));
 	}
 }
 
-void Handler1(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
+static void TW_Handler1(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
 {
 	(void)content;
 	(void)size;
 
 	struct lp_state *state = s;
 
-	if(state->count >= lp1_max_count || event_type == LP_FINI) {
+	if(state->count >= lp1_max_count || event_type == LP_FINI)
 		return;
-	}
 
 	while(!atomic_load(&lp1_turn)) {
-		// Wait for lp0 to finish
+		/* Spin until LP 0 has finished its current event */
 	}
 
 	if(event_type == EVENT) {
 		ScheduleNewEvent(1, lp1_times[state->count], EVENT, NULL, 0);
 		ScheduleNewEvent(0, now + lp1_send_delay, STRAGGLER_EVENT, NULL, 0);
-
-		// printf("[Straggler generator] s%.1lf->d%.1lf, ID%lu, ct%lu\n", now, now + lp1_send_delay, me,
-		// state->count);
 		ScheduleOutput(OUTPUT_TYPE, &(struct output_data){.id = me, .count = state->count},
 		    sizeof(struct output_data));
 		state->count++;
@@ -131,11 +168,11 @@ void Handler1(lp_id_t me, simtime_t now, unsigned event_type, const void *conten
 	atomic_store(&lp1_turn, false);
 }
 
-void ProcessEvent(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
+static void TW_ProcessEvent(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size, void *s)
 {
 	struct lp_state *state = s;
 
-	if(event_type != LP_FINI && now > conf.termination_time)
+	if(event_type != LP_FINI && now > 25.0)
 		return;
 
 	if(event_type == LP_INIT) {
@@ -144,137 +181,269 @@ void ProcessEvent(lp_id_t me, simtime_t now, unsigned event_type, const void *co
 			abort();
 		SetState(state);
 		state->count = 0;
-
 		ScheduleNewEvent(me, 0, EVENT, NULL, 0);
 		return;
 	}
 
-	if(me == 0) {
-		Handler0(me, now, event_type, content, size, state);
-	} else if(me == 1) {
-		Handler1(me, now, event_type, content, size, state);
-	} else {
-		fprintf(stderr, "Unknown LP ID\n");
+	if(me == 0)
+		TW_Handler0(me, now, event_type, content, size, state);
+	else if(me == 1)
+		TW_Handler1(me, now, event_type, content, size, state);
+	else {
+		fprintf(stderr, "TW_ProcessEvent: unknown LP ID %llu\n", (unsigned long long)me);
 		abort();
 	}
 }
 
-bool CanEnd(lp_id_t me, const void *snapshot)
+static bool TW_CanEnd(lp_id_t me, const void *snapshot)
 {
 	(void)me;
 	(void)snapshot;
 	return false;
 }
 
-void PerformOutput(lp_id_t me, unsigned output_type, const void *output_content, unsigned output_size)
+static struct simulation_configuration tw_conf = {
+    .lps = TW_NUM_LPS,
+    .n_threads = TW_NUM_THREADS,
+    .termination_time = 25,
+    .gvt_period = 10,
+    .log_level = LOG_WARN,
+    .stats_file = "output_test_tw",
+    .ckpt_interval = 0,
+    .core_binding = true,
+    .synchronization = TIME_WARP,
+    .dispatcher = TW_ProcessEvent,
+    .committed = TW_CanEnd,
+    .output_callback = TW_PerformOutput,
+};
+
+static bool chk_out(const char *actual, const char *expected, const char *file, int line)
 {
-	struct output_data *data = NULL;
-	lp_id_t id;
-	unsigned long count;
-	switch(output_type) {
-		case OUTPUT_TYPE:
-			data = (struct output_data *)output_content;
-			id = data->id;
-			count = data->count;
-			// printf("[Normal output] %lu,%lu,%lu\n", me, id, count);
-			snprintf(lp_outs[me][lp_outs_count[me]], OUT_SZ, "N%llu,%llu,%lu", me, id, count);
-			lp_outs_count[me]++;
-			break;
+	if(strcmp(actual, expected) != 0) {
+		fprintf(stderr, "%s:%d: output mismatch: got \"%s\", expected \"%s\"\n", file, line, actual, expected);
+		test_fail();
+	}
+	return true;
+}
 
-		case OUTPUT_FROM_STRAGGLER:
-			data = (struct output_data *)output_content;
-			id = data->id;
-			count = data->count;
-			// printf("[Straggler output] %lu,%lu,%lu\n", me, id, count);
-			snprintf(lp_outs[me][lp_outs_count[me]], OUT_SZ, "S%llu,%llu,%lu", me, id, count);
-			lp_outs_count[me]++;
-			break;
+#define CHK_OUT(actual, expected) chk_out(actual, expected, __FILE__, __LINE__)
 
-		default:
-			fprintf(stderr, "Unknown output type %u, with size %u\n", output_type, output_size);
-			break;
+static int perform_tw_exec(void *arg)
+{
+	(void)arg;
+
+	/* Reset shared state between runs */
+	atomic_store(&lp1_turn, false);
+	memset(tw_lp_outs, 0, sizeof(tw_lp_outs));
+	memset(tw_lp_outs_count, 0, sizeof(tw_lp_outs_count));
+	atomic_store(&tw_total_output_count, 0);
+
+	RootsimInit(&tw_conf);
+	RootsimRun();
+
+	/* --- Verify exact committed output sequence for LP 0 ---
+	 *
+	 * The busy-wait synchronization plus core_binding=true force a deterministic
+	 * execution order. Each round: LP 0 processes an EVENT at time T (output N),
+	 * then LP 1 sends a STRAGGLER_EVENT at time T+0.2 to LP 0. This causes LP 0
+	 * to roll back and process the STRAGGLER_EVENT first (output S), then replay
+	 * its EVENT (output N again). Some rounds don't trigger a straggler due to
+	 * the interleaving; those appear as consecutive N outputs. */
+	CHK_OUT(tw_lp_outs[0][0], "N0,0,0");
+	CHK_OUT(tw_lp_outs[0][1], "S0,0,1");
+	CHK_OUT(tw_lp_outs[0][2], "N0,0,1");
+	CHK_OUT(tw_lp_outs[0][3], "S0,0,2");
+	CHK_OUT(tw_lp_outs[0][4], "N0,0,2");
+	CHK_OUT(tw_lp_outs[0][5], "N0,0,3");
+	CHK_OUT(tw_lp_outs[0][6], "S0,0,4");
+	CHK_OUT(tw_lp_outs[0][7], "N0,0,4");
+	CHK_OUT(tw_lp_outs[0][8], "N0,0,5");
+	CHK_OUT(tw_lp_outs[0][9], "S0,0,6");
+	CHK_OUT(tw_lp_outs[0][10], "N0,0,6");
+	CHK_OUT(tw_lp_outs[0][11], "N0,0,7");
+	CHK_OUT(tw_lp_outs[0][12], "S0,0,8");
+	CHK_OUT(tw_lp_outs[0][13], "N0,0,8");
+	CHK_OUT(tw_lp_outs[0][14], "N0,0,9");
+	CHK_OUT(tw_lp_outs[0][15], "S0,0,10");
+	CHK_OUT(tw_lp_outs[0][16], "N0,0,10");
+	CHK_OUT(tw_lp_outs[0][17], "N0,0,11");
+	CHK_OUT(tw_lp_outs[0][18], "S0,0,12");
+	CHK_OUT(tw_lp_outs[0][19], "N0,0,12");
+	CHK_OUT(tw_lp_outs[0][20], "N0,0,13");
+	CHK_OUT(tw_lp_outs[0][21], "S0,0,14");
+	CHK_OUT(tw_lp_outs[0][22], "N0,0,14");
+	CHK_OUT(tw_lp_outs[0][23], "N0,0,15");
+	CHK_OUT(tw_lp_outs[0][24], "S0,0,16");
+	CHK_OUT(tw_lp_outs[0][25], "N0,0,16");
+	CHK_OUT(tw_lp_outs[0][26], "N0,0,17");
+	CHK_OUT(tw_lp_outs[0][27], "S0,0,18");
+	CHK_OUT(tw_lp_outs[0][28], "N0,0,18");
+	CHK_OUT(tw_lp_outs[0][29], "N0,0,19");
+
+	/* --- Verify exact committed output sequence for LP 1 --- */
+	CHK_OUT(tw_lp_outs[1][0], "N1,1,0");
+	CHK_OUT(tw_lp_outs[1][1], "N1,1,1");
+	CHK_OUT(tw_lp_outs[1][2], "N1,1,2");
+	CHK_OUT(tw_lp_outs[1][3], "N1,1,3");
+	CHK_OUT(tw_lp_outs[1][4], "N1,1,4");
+	CHK_OUT(tw_lp_outs[1][5], "N1,1,5");
+	CHK_OUT(tw_lp_outs[1][6], "N1,1,6");
+	CHK_OUT(tw_lp_outs[1][7], "N1,1,7");
+	CHK_OUT(tw_lp_outs[1][8], "N1,1,8");
+	CHK_OUT(tw_lp_outs[1][9], "N1,1,9");
+
+	/* --- Verify total count: detects phantom outputs from rolled-back events ---
+	 *
+	 * LP 0: 20 N-outputs + 10 S-outputs = 30
+	 * LP 1: 10 N-outputs
+	 * Total expected: 40
+	 *
+	 * If committed_output_on_rollback() fails to suppress outputs, rolled-back
+	 * EVENT handlers on LP 0 would fire extra N-outputs, inflating this count. */
+	size_t total = atomic_load(&tw_total_output_count);
+	if(total != 40) {
+		fprintf(stderr, "Total output count mismatch: got %zu, expected 40\n", total);
+		test_fail();
+	}
+
+	/* Also verify the per-LP counts individually */
+	if(tw_lp_outs_count[0] != 30) {
+		fprintf(stderr, "LP 0 output count mismatch: got %zu, expected 30\n", tw_lp_outs_count[0]);
+		test_fail();
+	}
+	if(tw_lp_outs_count[1] != 10) {
+		fprintf(stderr, "LP 1 output count mismatch: got %zu, expected 10\n", tw_lp_outs_count[1]);
+		test_fail();
 	}
+
+	return 0;
 }
 
-bool check_output(const char *output, const char *expected, size_t expected_len, char *file, int line)
+/* =========================================================================
+ * Serial test
+ * ========================================================================= */
+
+#define SERIAL_NUM_LPS 4
+#define SERIAL_EVENTS_PER_LP 5
+
+/* Output storage for the serial test */
+static char serial_outs[SERIAL_NUM_LPS][SERIAL_EVENTS_PER_LP][OUT_SZ];
+static size_t serial_outs_count[SERIAL_NUM_LPS];
+
+static void Serial_PerformOutput(lp_id_t me, unsigned output_type, const void *output_content, unsigned output_size)
 {
-	if(strncmp(output, expected, expected_len)) {
-		fprintf(stderr, "%s:%d: error: Output did not match expected. Actual: \"%s\", Expected:\"%s\".\n", file,
-		    line, output, expected);
+	if(output_type != OUTPUT_TYPE) {
+		fprintf(stderr, "Serial_PerformOutput: unexpected output_type %u\n", output_type);
 		test_fail();
 	}
-	return true;
+	if(output_size != sizeof(struct output_data)) {
+		fprintf(stderr, "Serial_PerformOutput: unexpected output_size %u (expected %zu)\n", output_size,
+		    sizeof(struct output_data));
+		test_fail();
+	}
+
+	size_t idx = serial_outs_count[me];
+	if(idx >= SERIAL_EVENTS_PER_LP) {
+		fprintf(stderr, "Serial_PerformOutput: too many outputs for LP %llu\n", (unsigned long long)me);
+		test_fail();
+	}
+
+	const struct output_data *data = output_content;
+	snprintf(serial_outs[me][idx], OUT_SZ, "N%llu,%llu,%lu", (unsigned long long)me, (unsigned long long)data->id,
+	    data->count);
+	serial_outs_count[me]++;
 }
 
-#define chk_out(output, expected, expected_len) check_output(output, expected, expected_len, __FILE__, __LINE__)
+static void Serial_ProcessEvent(lp_id_t me, simtime_t now, unsigned event_type, const void *content, unsigned size,
+    void *s)
+{
+	(void)now;
+	(void)content;
+	(void)size;
 
+	struct lp_state *state = s;
 
-int perform_exec(void *arg)
+	if(event_type == LP_INIT) {
+		state = rs_malloc(sizeof(*state));
+		if(state == NULL)
+			abort();
+		SetState(state);
+		state->count = 0;
+		ScheduleNewEvent(me, 1.0, EVENT, NULL, 0);
+		return;
+	}
+
+	if(event_type == LP_FINI)
+		return;
+
+	if(event_type == EVENT && state->count < SERIAL_EVENTS_PER_LP) {
+		ScheduleOutput(OUTPUT_TYPE, &(struct output_data){.id = me, .count = state->count},
+		    sizeof(struct output_data));
+		state->count++;
+		if(state->count < SERIAL_EVENTS_PER_LP)
+			ScheduleNewEvent(me, now + 1.0, EVENT, NULL, 0);
+	}
+}
+
+static bool Serial_CanEnd(lp_id_t me, const void *snapshot)
+{
+	(void)snapshot;
+	const struct lp_state *state = snapshot;
+	(void)state;
+	(void)me;
+	return false;
+}
+
+static struct simulation_configuration serial_conf = {
+    .lps = SERIAL_NUM_LPS,
+    .n_threads = 1,
+    .termination_time = 100,
+    .log_level = LOG_WARN,
+    .stats_file = "output_test_serial",
+    .synchronization = SERIAL,
+    .dispatcher = Serial_ProcessEvent,
+    .committed = Serial_CanEnd,
+    .output_callback = Serial_PerformOutput,
+};
+
+static int perform_serial_exec(void *arg)
 {
 	(void)arg;
-	RootsimInit(&conf);
 
-	for(size_t lp = 0; lp < 2; lp++) {
-		for(size_t i = 0; i < OUT_SLOTS; i++) {
-			memset(lp_outs[lp][i], 0, OUT_SZ);
-		}
-	}
+	memset(serial_outs, 0, sizeof(serial_outs));
+	memset(serial_outs_count, 0, sizeof(serial_outs_count));
 
+	RootsimInit(&serial_conf);
 	RootsimRun();
 
-	// printf("OUTPUTS\n");
-	// for(size_t lp = 0; lp < 2; lp++) {
-	// 	for(size_t i = 0; i < OUT_SLOTS; i++) {
-	// 		printf("%s\n", lp_outs[lp][i]);
-	// 	}
-	// }
-
-	chk_out(lp_outs[0][0], "N0,0,0", 7);
-	chk_out(lp_outs[0][1], "S0,0,1", 7);
-	chk_out(lp_outs[0][2], "N0,0,1", 7);
-	chk_out(lp_outs[0][3], "S0,0,2", 7);
-	chk_out(lp_outs[0][4], "N0,0,2", 7);
-	chk_out(lp_outs[0][5], "N0,0,3", 7);
-	chk_out(lp_outs[0][6], "S0,0,4", 7);
-	chk_out(lp_outs[0][7], "N0,0,4", 7);
-	chk_out(lp_outs[0][8], "N0,0,5", 7);
-	chk_out(lp_outs[0][9], "S0,0,6", 7);
-	chk_out(lp_outs[0][10], "N0,0,6", 7);
-	chk_out(lp_outs[0][11], "N0,0,7", 7);
-	chk_out(lp_outs[0][12], "S0,0,8", 7);
-	chk_out(lp_outs[0][13], "N0,0,8", 7);
-	chk_out(lp_outs[0][14], "N0,0,9", 7);
-	chk_out(lp_outs[0][15], "S0,0,10", 8);
-	chk_out(lp_outs[0][16], "N0,0,10", 8);
-	chk_out(lp_outs[0][17], "N0,0,11", 8);
-	chk_out(lp_outs[0][18], "S0,0,12", 8);
-	chk_out(lp_outs[0][19], "N0,0,12", 8);
-	chk_out(lp_outs[0][20], "N0,0,13", 8);
-	chk_out(lp_outs[0][21], "S0,0,14", 8);
-	chk_out(lp_outs[0][22], "N0,0,14", 8);
-	chk_out(lp_outs[0][23], "N0,0,15", 8);
-	chk_out(lp_outs[0][24], "S0,0,16", 8);
-	chk_out(lp_outs[0][25], "N0,0,16", 8);
-	chk_out(lp_outs[0][26], "N0,0,17", 8);
-	chk_out(lp_outs[0][27], "S0,0,18", 8);
-	chk_out(lp_outs[0][28], "N0,0,18", 8);
-	chk_out(lp_outs[0][29], "N0,0,19", 8);
-
-	chk_out(lp_outs[1][0], "N1,1,0", 7);
-	chk_out(lp_outs[1][1], "N1,1,1", 7);
-	chk_out(lp_outs[1][2], "N1,1,2", 7);
-	chk_out(lp_outs[1][3], "N1,1,3", 7);
-	chk_out(lp_outs[1][4], "N1,1,4", 7);
-	chk_out(lp_outs[1][5], "N1,1,5", 7);
-	chk_out(lp_outs[1][6], "N1,1,6", 7);
-	chk_out(lp_outs[1][7], "N1,1,7", 7);
-	chk_out(lp_outs[1][8], "N1,1,8", 7);
-	chk_out(lp_outs[1][9], "N1,1,9", 7);
+	/* Each LP should have produced exactly SERIAL_EVENTS_PER_LP outputs,
+	 * one per EVENT, in causal order (count 0,1,2,...). */
+	for(lp_id_t lp = 0; lp < SERIAL_NUM_LPS; lp++) {
+		if(serial_outs_count[lp] != SERIAL_EVENTS_PER_LP) {
+			fprintf(stderr, "Serial: LP %llu produced %zu outputs, expected %d\n", (unsigned long long)lp,
+			    serial_outs_count[lp], SERIAL_EVENTS_PER_LP);
+			test_fail();
+		}
+		for(unsigned i = 0; i < SERIAL_EVENTS_PER_LP; i++) {
+			char expected[OUT_SZ];
+			snprintf(expected, OUT_SZ, "N%llu,%llu,%u", (unsigned long long)lp, (unsigned long long)lp, i);
+			if(strcmp(serial_outs[lp][i], expected) != 0) {
+				fprintf(stderr, "Serial: LP %llu output[%u] = \"%s\", expected \"%s\"\n",
+				    (unsigned long long)lp, i, serial_outs[lp][i], expected);
+				test_fail();
+			}
+		}
+	}
 
 	return 0;
 }
 
+/* =========================================================================
+ * Entry point
+ * ========================================================================= */
+
 int main(void)
 {
-	test("Testing committed output", perform_exec, NULL);
+	test("Testing committed output — serial mode", perform_serial_exec, NULL);
+	test("Testing committed output — Time Warp with stragglers", perform_tw_exec, NULL);
 }

From bf73561902a8c16c5c4df884481fd125ed60537e Mon Sep 17 00:00:00 2001
From: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
Date: Sun, 28 Jun 2026 10:55:19 +0200
Subject: [PATCH 11/13] Remove an unnecessary store from the critical path

In the fixing process, I started adding a store that is now
unnecessary given the other fixes. Remove it.

Signed-off-by: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
---
 src/mm/msg_allocator.c | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/mm/msg_allocator.c b/src/mm/msg_allocator.c
index e3d1fe96..b6fa8b9d 100644
--- a/src/mm/msg_allocator.c
+++ b/src/mm/msg_allocator.c
@@ -78,7 +78,6 @@ struct lp_msg *msg_allocator_alloc(const unsigned payload_size)
 void msg_allocator_free(struct lp_msg *msg)
 {
 	free_msg_outputs(msg->outputs);
-	msg->outputs = NULL;
 	if(likely(msg->pl_size <= MSG_PAYLOAD_BASE_SIZE))
 		array_push(free_list, msg);
 	else

From 115633146b6d1f982c8133496e6e23062bcdb082 Mon Sep 17 00:00:00 2001
From: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
Date: Wed, 1 Jul 2026 15:16:11 +0200
Subject: [PATCH 12/13] Remove another unnecessary store from common_msg_pack

In the fixing process, I started adding a store that is now
unnecessary given the other fixes. Remove it.

Signed-off-by: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
---
 src/lp/common.h | 1 -
 1 file changed, 1 deletion(-)

diff --git a/src/lp/common.h b/src/lp/common.h
index f2a72cf1..9ac32987 100644
--- a/src/lp/common.h
+++ b/src/lp/common.h
@@ -58,7 +58,6 @@ static inline struct lp_msg *common_msg_pack(const lp_id_t receiver, const simti
 	msg->dest = receiver;
 	msg->dest_t = timestamp;
 	msg->m_type = event_type;
-	msg->outputs = NULL;
 
 	if(likely(payload_size))
 		memcpy(msg->pl, payload, payload_size);

From d90bbaf356af85fa21243caa303346157ab64284 Mon Sep 17 00:00:00 2001
From: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
Date: Thu, 2 Jul 2026 11:50:21 +0200
Subject: [PATCH 13/13] Ensure output messages are ordered per-LP

Reversing the order in which the output messages are materialized upon
fossil collection ensures a per-LP order.

There is no global order yet, as fossil_lp_collect() is executed lazily
and possibly concurrently by multiple workers.

Signed-off-by: Alessandro Pellegrini <a.pellegrini@ing.uniroma2.it>
---
 src/gvt/fossil.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/gvt/fossil.c b/src/gvt/fossil.c
index f4637736..dd5fae34 100644
--- a/src/gvt/fossil.c
+++ b/src/gvt/fossil.c
@@ -51,8 +51,8 @@ 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;