detect_termination function mechanism, shared local vars protected by mutex

diff --git a/control-skeleton.c b/control-skeleton.c
index 464b8cc0b78de4e732ed745d0174ce1440d84a76..ae25b2fcf5dfe44e64dbab1a5e6a9a5084ec2524 100644 (file)
--- a/control-skeleton.c
+++ b/control-skeleton.c
@@ -16,15 +16,27 @@
 #include "util.h"
 
 // the tag(s) of the control messages
 #include "util.h"
 
 // the tag(s) of the control messages

-//#define ...
+#define CONTROL_SIGNAL 99   // --> must be distinct from BASIC_MESSAGE = 1

 
 // type for control messages
 
 // type for control messages

-//typedef struct {
-//  ...
-//} control_message_t;
+typedef struct {
+    int ping;
+} control_message_t;
+
+// Node operational states
+typedef enum {
+    ACTIVE_STATE,
+    PASSIVE_STATE
+} node_state_t;
+
+
+// Shared Local State (Protected by state_mutex)
+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 bool            is_initiator = false;

 
 

-// local variables
-//static ...

 
 /*
   Main loop of the termination detection algorithm.
 
 /*
   Main loop of the termination detection algorithm.


@@ -37,18 +49,49 @@
   the process, the number of PROCESSES running the basic algorithm,
   and whether the process is an initiator.
 */
   the process, the number of PROCESSES running the basic algorithm,
   and whether the process is an initiator.
 */

-void *detect_termination(void *_args)
-{
+void *detect_termination(void *_args){

   thread_args_t *args = _args;
   int id = args->id;
   int processes = args->processes;
   bool initiator = args->initiator;
 
   thread_args_t *args = _args;
   int id = args->id;
   int processes = args->processes;
   bool initiator = args->initiator;
 

-  // THE FOLLOWING TWO LINES OF CODE MUST BE REPLACED BY THE CODE OF
-  // THE TERMINATION DETECTION ALGORITHM
-
-  // let the basic algorithm run for 10 seconds
-  sleep(10);
+  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;
 }
 
   return NULL;
 }


@@ -88,4 +131,4 @@ void control_basic_send_hook(int id, int peer)
 */
 void control_basic_receive_hook(int id, int peer)
 {
 */
 void control_basic_receive_hook(int id, int peer)
 {

-}
+}
\ No newline at end of file

diff --git a/documents/DistributedSystems-TerminationDetectionDijkstraScholten.pdf b/documents/DistributedSystems-TerminationDetectionDijkstraScholten.pdf
new file mode 100644 (file)
index 0000000..cc1847c
Binary files /dev/null and b/documents/DistributedSystems-TerminationDetectionDijkstraScholten.pdf differ

diff --git a/util.c b/util.c
index 65e145edeeb99038286a2309462f657f07f6d0cd..f7fdbfaa401399985412734514a69e62d496e811 100644 (file)
--- a/util.c
+++ b/util.c
@@ -52,3 +52,29 @@ void warn(char *format, ...)
   vfprintf(stderr, format, args);
   va_end (args);
 }
   vfprintf(stderr, format, args);
   va_end (args);
 }

+
+/*
+  Evaluates Rule C & D of Dijkstra-Scholten. Caller must already 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
+            
+        }
+    }
+}
\ No newline at end of file