control.c

   1 /*
   2   Parallel Programming - 2026/2027
   3   Implementation of a termination detection algorithm (Dijkstra-Scholten)
   4 */
   5
   6 #include <mpi.h>
   7 #include <pthread.h>
   8 #include <stdio.h>
   9 #include <stdbool.h>
  10 #include <unistd.h>
  11
  12 #include "global.h"
  13 #include "control.h"
  14 #include "util.h"
  15
  16 // Tag for control messages. Distinct from basic algorithm's tags (1).
  17 #define CONTROL_SIGNAL 99
  18
  19 // Message types for control routing
  20 typedef enum {
  21     MSG_ACK,
  22     MSG_KILL
  23 } msg_type_t;
  24
  25 // Type for control messages
  26 typedef struct {
  27     msg_type_t type;
  28 } control_message_t;
  29
  30 // Node operational states
  31 typedef enum {
  32     ACTIVE_STATE,
  33     PASSIVE_STATE
  34 } node_state_t;
  35
  36
  37 // SHARED LOCAL STATE - Must be protected by mutex because basic_thread and control_thread access it concurrently
  38 static pthread_mutex_t state_mutex = PTHREAD_MUTEX_INITIALIZER;
  39 static node_state_t    state       = PASSIVE_STATE;
  40 static int             parent      = -1; // -1 indicates null/no parent
  41 static int             deficit     = 0;  // C_i counter (unacknowledged messages)
  42 static bool            is_initiator = false;
  43 static bool            started      = false; // RACE CONDITION FIX
  44
  45
  46 // HELPER FUNCTION - Evaluates if the node can detach from the tree. Caller must hold state_mutex.
  47 static void try_resolve_tree(int my_id){
  48     if (started && state == PASSIVE_STATE && deficit == 0){
  49         if(is_initiator){
  50             // Root is passive and deficit is 0 -> Global Termination!
  51             return;
  52         }
  53
  54         if(parent != -1){
  55             control_message_t sig = { MSG_ACK };
  56             // Send acknowledgment signal up the tree to our parent
  57             MPI_Send(&sig, sizeof(control_message_t), MPI_BYTE, parent,
  58                      CONTROL_SIGNAL, MPI_COMM_WORLD);
  59
  60             trace("%d: [CONTROL] Sent tree-signal to parent %d\n", my_id, parent);
  61             parent = -1; // Detach from tree
  62         }
  63     }
  64 }
  65
  66
  67 //  Main loop of the termination detection algorithm.
  68 void *detect_termination(void *_args){
  69     thread_args_t *args = _args;
  70     int id = args->id;
  71     int processes = args->processes; //  MPI interface - needed for the KILL broadcast
  72     is_initiator = args->initiator;
  73
  74     while(true){
  75         pthread_mutex_lock(&state_mutex);
  76
  77         // CHECK GLOBAL TERMINATION CONDITION - Only evaluate if the basic algorithm has officially started
  78         if(started && is_initiator && state == PASSIVE_STATE && deficit == 0){
  79             trace("%d: [CONTROL] GLOBAL TERMINATION DETECTED!\n", id);
  80
  81             // Broadcast KILL signal to unblock all other control threads - use
  82             control_message_t kill_sig = {MSG_KILL};
  83             for(int p = 0; p < processes; p++){
  84                 if(p != id){
  85                     MPI_Send(
  86                         &kill_sig,                      // pointer to the control_message_t struct.
  87                         sizeof(control_message_t),      // the size of the message.
  88                         MPI_BYTE,                       // sending raw memory bytes.
  89                         p,                              // destinnationn, parent node.
  90                         CONTROL_SIGNAL,                 // label 99.
  91                         MPI_COMM_WORLD                  // the overall group of processes this message is restricted to.
  92                     );
  93                 }
  94             }
  95
  96             pthread_mutex_unlock(&state_mutex);
  97             break; // Breaks loop, tells main() to shut down
  98         }
  99
 100         pthread_mutex_unlock(&state_mutex);
 101
 102         // Check network for incoming child signals (non-blocking) - use MPI_Recv to read messages sent over the network by other processes
 103         int flag = 0;
 104         MPI_Status status;
 105         MPI_Iprobe(MPI_ANY_SOURCE, CONTROL_SIGNAL, MPI_COMM_WORLD, &flag, &status);
 106
 107         if(flag){
 108             control_message_t sig;
 109             MPI_Recv(
 110                 &sig,                                   // pointer to the memory space where the incoming message will be stored
 111                 sizeof(control_message_t),              // the maximum size (in bytes) of the data willing to receive.
 112                 MPI_BYTE,                               // raw stream of bytes
 113                 status.MPI_SOURCE,                      // the ID (rank) of the process you want to receive from MPI_Iprobe.
 114                 CONTROL_SIGNAL,                         // label of the message
 115                 MPI_COMM_WORLD,                         // the overall group of processes this message belongs to.
 116                 MPI_STATUS_IGNORE                       // just to save memory and processing time.
 117             );
 118
 119             if(sig.type == MSG_KILL){
 120                 trace("%d: [CONTROL] Received KILL signal. Shutting down.\n", id);
 121                 break;
 122             }
 123             else if(sig.type == MSG_ACK){
 124                 pthread_mutex_lock(&state_mutex);
 125                 deficit--; // Rule D: A child finished its work and reported back
 126                 trace("%d: [CONTROL] Got signal from %d (New Deficit: %d)\n",
 127                       id, status.MPI_SOURCE, deficit);
 128
 129                 try_resolve_tree(id);
 130                 pthread_mutex_unlock(&state_mutex);
 131             }
 132         }
 133         else{
 134             // Sleep briefly to prevent pinning the CPU core at 100% usage
 135             usleep(1000);
 136         }
 137     }
 138
 139     return NULL;
 140 }
 141
 142 /*
 143   Called at startup of the basic algorithm in process ID.
 144 */
 145 void control_basic_start_hook(int id)
 146 {
 147     pthread_mutex_lock(&state_mutex);
 148     is_initiator = (id == 0);
 149     state   = is_initiator ? ACTIVE_STATE : PASSIVE_STATE;
 150     parent  = -1;
 151     deficit = 0;
 152     started = true; // Signal the control thread that it is safe to check for termination
 153     pthread_mutex_unlock(&state_mutex);
 154 }
 155
 156 /*
 157   Called when the basic algorithm process ID becomes passive.
 158 */
 159 void control_become_passive_hook(int id)
 160 {
 161     pthread_mutex_lock(&state_mutex);
 162     state = PASSIVE_STATE;
 163     try_resolve_tree(id); // Check if we can leave the tree right now
 164     pthread_mutex_unlock(&state_mutex);
 165 }
 166
 167 /*
 168   Called when the basic algorithm process ID becomes active.
 169 */
 170 void control_become_active_hook(int id)
 171 {
 172     pthread_mutex_lock(&state_mutex);
 173     state = ACTIVE_STATE;
 174     pthread_mutex_unlock(&state_mutex);
 175 }
 176
 177 /*
 178   Called when the basic algorithm process ID sends a basic message to
 179   process PEER.
 180 */
 181 void control_basic_send_hook(int id, int peer)
 182 {
 183     pthread_mutex_lock(&state_mutex);
 184     deficit++; // Rule A: Sent unacknowledged work
 185     pthread_mutex_unlock(&state_mutex);
 186 }
 187
 188 /*
 189   Called when the basic algorithm process ID receives a basic message
 190   from process PEER.
 191 */
 192 void control_basic_receive_hook(int id, int peer)
 193 {
 194     pthread_mutex_lock(&state_mutex);
 195
 196     // A node is only "out of the tree" if parent == -1.
 197     // It can be PASSIVE but still in the tree if it is waiting for children (deficit > 0).
 198     if(parent == -1 && !is_initiator){
 199         // Not in the tree -> Join it
 200         parent = peer;
 201         trace("%d: [CONTROL] Joined tree under parent %d\n", id, parent);
 202     }else{
 203         // Already in the tree (or is root) -> Reject parent change, immediately signal back
 204         control_message_t sig = { MSG_ACK };
 205         MPI_Send(&sig, sizeof(control_message_t), MPI_BYTE, peer,
 206                  CONTROL_SIGNAL, MPI_COMM_WORLD);
 207         trace("%d: [CONTROL] Rejected parent %d (Already in tree or Root)\n", id, peer);
 208     }
 209
 210     pthread_mutex_unlock(&state_mutex);
 211 }