/*
Parallel Programming - 2026/2027
- Skeleton of an implementation of a termination detection algorithm
+ Implementation of a termination detection algorithm (Dijkstra-Scholten)
*/
#include <mpi.h>
#include <pthread.h>
-
#include <stdio.h>
#include <stdbool.h>
#include <unistd.h>
#include "control.h"
#include "util.h"
-// the tag(s) of the control messages
-#define CONTROL_SIGNAL 99 // --> must be distinct from BASIC_MESSAGE = 1
+// Tag for control messages. Distinct from basic algorithm's tags.
+#define CONTROL_SIGNAL 99
-// type for control messages
+// Type for control messages (empty payload, just a signal)
typedef struct {
- int ping;
+ int dummy;
} control_message_t;
// Node operational states
PASSIVE_STATE
} node_state_t;
-
-// Shared Local State (Protected by state_mutex)
+// --- 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 int deficit = 0; // C_i counter
+static int deficit = 0; // C_i counter (unacknowledged messages)
static bool is_initiator = false;
+// --- 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 (state == PASSIVE_STATE && deficit == 0)
+ {
+ if (is_initiator) {
+ // Root is passive and deficit is 0 -> Global Termination!
+ return;
+ }
+
+ if (parent != -1)
+ {
+ control_message_t sig = {0};
+ // Send acknowledgment signal up the tree to our parent
+ MPI_Send(&sig, sizeof(control_message_t), MPI_BYTE, parent,
+ CONTROL_SIGNAL, MPI_COMM_WORLD);
+
+ trace("%d: [CONTROL] Sent tree-signal to parent %d\n", my_id, parent);
+ parent = -1; // Detach from tree
+ }
+ }
+}
/*
Main loop of the termination detection algorithm.
-
- When termination is detected, it must end (returning NULL or another
- appropriate value), which signals the main thread that the basic
- algorithm has terminated.
-
- The arguments of the termination detection algorithm are the ID of
- the process, the number of PROCESSES running the basic algorithm,
- and whether the process is an initiator.
*/
-void *detect_termination(void *_args){
- thread_args_t *args = _args;
- int id = args->id;
- int processes = args->processes;
- bool initiator = args->initiator;
-
- while (true){
- pthread_mutex_lock(&state_mutex);
-
- // CHECK GLOBAL TERMINATION CONDITION (only happens at root)
- if (initiator && state == PASSIVE_STATE && deficit == 0)
- {
- trace("%d: [CONTROL] GLOBAL TERMINATION DETECTED!\n", id);
- pthread_mutex_unlock(&state_mutex);
- break; // Breaks loop, returns NULL, tells main() to shut down
- }
-
- pthread_mutex_unlock(&state_mutex);
-
- // Check network for incoming child signals (non-blocking with MPI_Iprobe)
- int flag = 0;
- MPI_Status status;
- MPI_Iprobe(MPI_ANY_SOURCE, CONTROL_SIGNAL, MPI_COMM_WORLD, &flag, &status);
-
- 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);
-
- pthread_mutex_lock(&state_mutex);
- deficit--;
- trace("%d: [CONTROL] Got signal from %d (New Deficit: %d)\n",
- id, status.MPI_SOURCE, deficit);
-
- try_resolve_tree(id);
- pthread_mutex_unlock(&state_mutex);
- }
- else{
- // Sleep for 1 millisecond to prevent this while(true) loop
- // from pinning the CPU core at 100% usage
- usleep(1000);
- }
- }
-
- return NULL;
+void *detect_termination(void *_args)
+{
+ thread_args_t *args = _args;
+ int id = args->id;
+ is_initiator = args->initiator;
+
+ // (Removed unused 'processes' variable to fix compiler warning)
+
+ while (true)
+ {
+ pthread_mutex_lock(&state_mutex);
+
+ // CHECK GLOBAL TERMINATION CONDITION
+ if (is_initiator && state == PASSIVE_STATE && deficit == 0)
+ {
+ trace("%d: [CONTROL] GLOBAL TERMINATION DETECTED!\n", id);
+ pthread_mutex_unlock(&state_mutex);
+ break; // Breaks loop, tells main() to shut down
+ }
+
+ pthread_mutex_unlock(&state_mutex);
+
+ // Check network for incoming child signals (non-blocking)
+ int flag = 0;
+ MPI_Status status;
+ MPI_Iprobe(MPI_ANY_SOURCE, CONTROL_SIGNAL, MPI_COMM_WORLD, &flag, &status);
+
+ 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);
+
+ 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",
+ id, status.MPI_SOURCE, deficit);
+
+ try_resolve_tree(id);
+ pthread_mutex_unlock(&state_mutex);
+ }
+ else
+ {
+ // Sleep briefly to prevent pinning the CPU core at 100% usage
+ usleep(1000);
+ }
+ }
+
+ return NULL;
}
/*
*/
void control_basic_start_hook(int id)
{
+ pthread_mutex_lock(&state_mutex);
+ state = is_initiator ? ACTIVE_STATE : PASSIVE_STATE;
+ parent = -1;
+ deficit = 0;
+ pthread_mutex_unlock(&state_mutex);
}
/*
*/
void control_become_passive_hook(int id)
{
+ pthread_mutex_lock(&state_mutex);
+ state = PASSIVE_STATE;
+ try_resolve_tree(id); // Check if we can leave the tree right now
+ pthread_mutex_unlock(&state_mutex);
}
/*
*/
void control_become_active_hook(int id)
{
+ pthread_mutex_lock(&state_mutex);
+ state = ACTIVE_STATE;
+ pthread_mutex_unlock(&state_mutex);
}
/*
*/
void control_basic_send_hook(int id, int peer)
{
+ pthread_mutex_lock(&state_mutex);
+ deficit++; // Rule A: Sent unacknowledged work
+ pthread_mutex_unlock(&state_mutex);
}
/*
*/
void control_basic_receive_hook(int id, int peer)
{
-}
\ No newline at end of file
+ pthread_mutex_lock(&state_mutex);
+
+ if (parent == -1)
+ {
+ // Rule B (Case 1): Woken up from passive state
+ state = ACTIVE_STATE;
+ parent = peer; // Adopt sender as parent in the tree
+ trace("%d: [CONTROL] Joined tree under parent %d\n", id, parent);
+ }
+ else
+ {
+ // Rule B (Case 2): Already active! Reject parent change, immediately signal back
+ control_message_t sig = {0};
+ MPI_Send(&sig, sizeof(control_message_t), MPI_BYTE, peer,
+ CONTROL_SIGNAL, MPI_COMM_WORLD);
+ trace("%d: [CONTROL] Rejected parent %d (Already Active)\n", id, peer);
+ }
+
+ pthread_mutex_unlock(&state_mutex);
+}
projects / ue-pp-terminationdetectionalgorithm.git / commitdiff