/*
Parallel Programming - 2026/2027
-
Implementation of a termination detection algorithm (Dijkstra-Scholten)
*/
#include "control.h"
#include "util.h"
-// Tag for control messages. Distinct from basic algorithm's tags.
+// Tag for control messages. Distinct from basic algorithm's tags (1).
#define CONTROL_SIGNAL 99
// Message types for control routing
PASSIVE_STATE
} node_state_t;
-// --- SHARED LOCAL STATE ---
-// Must be protected by mutex because basic_thread and control_thread access it concurrently
+
+// SHARED LOCAL STATE - Must be protected by mutex because basic_thread and control_thread access it concurrently
static pthread_mutex_t state_mutex = PTHREAD_MUTEX_INITIALIZER;
static node_state_t state = PASSIVE_STATE;
static int parent = -1; // -1 indicates null/no parent
static bool is_initiator = false;
static bool started = false; // RACE CONDITION FIX
-// --- HELPER FUNCTION ---
-// MUST BE DECLARED ABOVE THE OTHER FUNCTIONS!
-// Evaluates if the node can detach from the tree. Caller must hold state_mutex.
-static void try_resolve_tree(int my_id)
-{
- if (started && state == PASSIVE_STATE && deficit == 0)
- {
- if (is_initiator) {
+
+// HELPER FUNCTION - Evaluates if the node can detach from the tree. Caller must hold state_mutex.
+static void try_resolve_tree(int my_id){
+ if (started && state == PASSIVE_STATE && deficit == 0){
+ if(is_initiator){
// Root is passive and deficit is 0 -> Global Termination!
return;
}
- if (parent != -1)
- {
+ if(parent != -1){
control_message_t sig = { MSG_ACK };
// Send acknowledgment signal up the tree to our parent
MPI_Send(&sig, sizeof(control_message_t), MPI_BYTE, parent,
}
}
-/*
- Main loop of the termination detection algorithm.
-*/
-void *detect_termination(void *_args)
-{
+
+// Main loop of the termination detection algorithm.
+void *detect_termination(void *_args){
thread_args_t *args = _args;
int id = args->id;
- int processes = args->processes; // Needed for the KILL broadcast
+ int processes = args->processes; // MPI interface - needed for the KILL broadcast
is_initiator = args->initiator;
- while (true)
- {
+ while(true){
pthread_mutex_lock(&state_mutex);
- // CHECK GLOBAL TERMINATION CONDITION
- // Only evaluate if the basic algorithm has officially started
- if (started && is_initiator && state == PASSIVE_STATE && deficit == 0)
- {
+ // CHECK GLOBAL TERMINATION CONDITION - Only evaluate if the basic algorithm has officially started
+ if(started && is_initiator && state == PASSIVE_STATE && deficit == 0){
trace("%d: [CONTROL] GLOBAL TERMINATION DETECTED!\n", id);
- // Broadcast KILL signal to unblock all other control threads
- control_message_t kill_sig = { MSG_KILL };
- for (int p = 0; p < processes; p++) {
- if (p != id) {
- MPI_Send(&kill_sig, sizeof(control_message_t), MPI_BYTE, p,
- CONTROL_SIGNAL, MPI_COMM_WORLD);
+ // Broadcast KILL signal to unblock all other control threads - use
+ control_message_t kill_sig = {MSG_KILL};
+ for(int p = 0; p < processes; p++){
+ if(p != id){
+ MPI_Send(
+ &kill_sig, // pointer to the control_message_t struct.
+ sizeof(control_message_t), // the size of the message.
+ MPI_BYTE, // sending raw memory bytes.
+ p, // destinnationn, parent node.
+ CONTROL_SIGNAL, // label 99.
+ MPI_COMM_WORLD // the overall group of processes this message is restricted to.
+ );
}
}
pthread_mutex_unlock(&state_mutex);
- // Check network for incoming child signals (non-blocking)
+ // Check network for incoming child signals (non-blocking) - use MPI_Recv to read messages sent over the network by other processes
int flag = 0;
MPI_Status status;
MPI_Iprobe(MPI_ANY_SOURCE, CONTROL_SIGNAL, MPI_COMM_WORLD, &flag, &status);
- if (flag)
- {
+ if(flag){
control_message_t sig;
- MPI_Recv(&sig, sizeof(control_message_t), MPI_BYTE, status.MPI_SOURCE,
- CONTROL_SIGNAL, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-
- if (sig.type == MSG_KILL) {
+ MPI_Recv(
+ &sig, // pointer to the memory space where the incoming message will be stored
+ sizeof(control_message_t), // the maximum size (in bytes) of the data willing to receive.
+ MPI_BYTE, // raw stream of bytes
+ status.MPI_SOURCE, // the ID (rank) of the process you want to receive from MPI_Iprobe.
+ CONTROL_SIGNAL, // label of the message
+ MPI_COMM_WORLD, // the overall group of processes this message belongs to.
+ MPI_STATUS_IGNORE // just to save memory and processing time.
+ );
+
+ if(sig.type == MSG_KILL){
trace("%d: [CONTROL] Received KILL signal. Shutting down.\n", id);
break;
}
- else if (sig.type == MSG_ACK) {
+ else if(sig.type == MSG_ACK){
pthread_mutex_lock(&state_mutex);
deficit--; // Rule D: A child finished its work and reported back
trace("%d: [CONTROL] Got signal from %d (New Deficit: %d)\n",
pthread_mutex_unlock(&state_mutex);
}
}
- else
- {
+ else{
// Sleep briefly to prevent pinning the CPU core at 100% usage
usleep(1000);
}
// A node is only "out of the tree" if parent == -1.
// It can be PASSIVE but still in the tree if it is waiting for children (deficit > 0).
- if (parent == -1 && !is_initiator)
- {
+ if(parent == -1 && !is_initiator){
// Not in the tree -> Join it
parent = peer;
trace("%d: [CONTROL] Joined tree under parent %d\n", id, parent);
- }
- else
- {
+ }else{
// Already in the tree (or is root) -> Reject parent change, immediately signal back
control_message_t sig = { MSG_ACK };
MPI_Send(&sig, sizeof(control_message_t), MPI_BYTE, peer,
projects / ue-pp-terminationdetectionalgorithm.git / blobdiff