control.c

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