X-Git-Url: https://vgcfreebox.myrthtech.pt/gitweb/ue-pp-terminationdetectionalgorithm.git/blobdiff_plain/df5d391823a04906ddacef09ad616a5025fd1b61..e817d5df83a20d84bdd0e5946c6ecea37e55a5a3:/control.c?ds=sidebyside diff --git a/control.c b/control.c index 3bcb8c8..67a4df8 100644 --- a/control.c +++ b/control.c @@ -1,6 +1,5 @@ /* Parallel Programming - 2026/2027 - Implementation of a termination detection algorithm (Dijkstra-Scholten) */ @@ -14,7 +13,7 @@ #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 @@ -34,8 +33,8 @@ typedef enum { 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 @@ -43,20 +42,16 @@ static int deficit = 0; // C_i counter (unacknowledged messages 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, @@ -68,32 +63,33 @@ static void try_resolve_tree(int my_id) } } -/* - 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. + ); } } @@ -103,22 +99,28 @@ void *detect_termination(void *_args) 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", @@ -128,8 +130,7 @@ void *detect_termination(void *_args) pthread_mutex_unlock(&state_mutex); } } - else - { + else{ // Sleep briefly to prevent pinning the CPU core at 100% usage usleep(1000); } @@ -194,14 +195,11 @@ void control_basic_receive_hook(int id, int peer) // 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,