control-skeleton.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 // Type for control messages (empty payload, just a signal)
  21 typedef struct {
  22     int dummy;
  23 } control_message_t;
  24
  25 // Node operational states
  26 typedef enum {
  27     ACTIVE_STATE,
  28     PASSIVE_STATE
  29 } node_state_t;
  30
  31 // --- SHARED LOCAL STATE ---
  32 // Must be protected by mutex because basic_thread and control_thread access it concurrently
  33 static pthread_mutex_t state_mutex = PTHREAD_MUTEX_INITIALIZER;
  34 static node_state_t    state       = PASSIVE_STATE;
  35 static int             parent      = -1; // -1 indicates null/no parent
  36 static int             deficit     = 0;  // C_i counter (unacknowledged messages)
  37 static bool            initiator = false;
  38
  39 // --- HELPER FUNCTION ---
  40 // MUST BE DECLARED ABOVE THE OTHER FUNCTIONS!
  41 // Evaluates if the node can detach from the tree. Caller must hold state_mutex.
  42 static void try_resolve_tree(int my_id)
  43 {
  44     if (state == PASSIVE_STATE && deficit == 0)
  45     {
  46         if (initiator) {
  47             // Root is passive and deficit is 0 -> Global Termination!
  48             return;
  49         }
  50
  51         if (parent != -1)
  52         {
  53             control_message_t sig = {0};
  54             // Send acknowledgment signal up the tree to our parent
  55             MPI_Send(&sig, sizeof(control_message_t), MPI_BYTE, parent,
  56                      CONTROL_SIGNAL, MPI_COMM_WORLD);
  57
  58             trace("%d: [CONTROL] Sent tree-signal to parent %d\n", my_id, parent);
  59             parent = -1; // Detach from tree
  60         }
  61     }
  62 }
  63
  64 /*
  65   Main loop of the termination detection algorithm.
  66 */
  67 void *detect_termination(void *_args)
  68 {
  69     thread_args_t *args = _args;
  70     int id = args->id;
  71     initiator = args->initiator;
  72
  73     // (Removed unused 'processes' variable to fix compiler warning)
  74
  75     while (true)
  76     {
  77         pthread_mutex_lock(&state_mutex);
  78
  79         // CHECK GLOBAL TERMINATION CONDITION
  80         if (initiator && state == PASSIVE_STATE && deficit == 0)
  81         {
  82             trace("%d: [CONTROL] GLOBAL TERMINATION DETECTED!\n", id);
  83             pthread_mutex_unlock(&state_mutex);
  84             break; // Breaks loop, tells main() to shut down
  85         }
  86
  87         pthread_mutex_unlock(&state_mutex);
  88
  89         // Check network for incoming child signals (non-blocking)
  90         int flag = 0;
  91         MPI_Status status;
  92         MPI_Iprobe(MPI_ANY_SOURCE, CONTROL_SIGNAL, MPI_COMM_WORLD, &flag, &status);
  93
  94         if (flag)
  95         {
  96             control_message_t sig;
  97             MPI_Recv(&sig, sizeof(control_message_t), MPI_BYTE, status.MPI_SOURCE,
  98                      CONTROL_SIGNAL, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
  99
 100             pthread_mutex_lock(&state_mutex);
 101             deficit--; // Rule D: A child finished its work and reported back
 102             trace("%d: [CONTROL] Got signal from %d (New Deficit: %d)\n",
 103                   id, status.MPI_SOURCE, deficit);
 104
 105             try_resolve_tree(id);
 106             pthread_mutex_unlock(&state_mutex);
 107         }
 108         else
 109         {
 110             // Sleep briefly to prevent pinning the CPU core at 100% usage
 111             usleep(1000);
 112         }
 113     }
 114
 115     return NULL;
 116 }
 117
 118 /*
 119   Called at startup of the basic algorithm in process ID.
 120 */
 121 void control_basic_start_hook(int id)
 122 {
 123     pthread_mutex_lock(&state_mutex);
 124     state   = initiator ? ACTIVE_STATE : PASSIVE_STATE;
 125     parent  = -1;
 126     deficit = 0;
 127     pthread_mutex_unlock(&state_mutex);
 128 }
 129
 130 /*
 131   Called when the basic algorithm process ID becomes passive.
 132 */
 133 void control_become_passive_hook(int id)
 134 {
 135     pthread_mutex_lock(&state_mutex);
 136     state = PASSIVE_STATE;
 137     try_resolve_tree(id); // Check if we can leave the tree right now
 138     pthread_mutex_unlock(&state_mutex);
 139 }
 140
 141 /*
 142   Called when the basic algorithm process ID becomes active.
 143 */
 144 void control_become_active_hook(int id)
 145 {
 146     pthread_mutex_lock(&state_mutex);
 147     state = ACTIVE_STATE;
 148     pthread_mutex_unlock(&state_mutex);
 149 }
 150
 151 /*
 152   Called when the basic algorithm process ID sends a basic message to
 153   process PEER.
 154 */
 155 void control_basic_send_hook(int id, int peer)
 156 {
 157     pthread_mutex_lock(&state_mutex);
 158     deficit++; // Rule A: Sent unacknowledged work
 159     pthread_mutex_unlock(&state_mutex);
 160 }
 161
 162 /*
 163   Called when the basic algorithm process ID receives a basic message
 164   from process PEER.
 165 */
 166 void control_basic_receive_hook(int id, int peer)
 167 {
 168     pthread_mutex_lock(&state_mutex);
 169
 170     if (parent == -1)
 171     {
 172         // Rule B (Case 1): Woken up from passive state
 173         state  = ACTIVE_STATE;
 174         parent = peer; // Adopt sender as parent in the tree
 175         trace("%d: [CONTROL] Joined tree under parent %d\n", id, parent);
 176     }
 177     else
 178     {
 179         // Rule B (Case 2): Already active! Reject parent change, immediately signal back
 180         control_message_t sig = {0};
 181         MPI_Send(&sig, sizeof(control_message_t), MPI_BYTE, peer,
 182                  CONTROL_SIGNAL, MPI_COMM_WORLD);
 183         trace("%d: [CONTROL] Rejected parent %d (Already Active)\n", id, peer);
 184     }
 185
 186     pthread_mutex_unlock(&state_mutex);
 187 }