--- /dev/null
+# replace with the file containing your algorithm
+CONTROL_ALGORITHM = control-skeleton.c
+
+# add extra flags here
+XCFLAGS =
+
+CC = mpicc
+CFLAGS = -std=gnu11 -Wall -Wno-parentheses -g3 $(XWARN) $(XCFLAGS)
+
+#LDLIBS = -lrt
+
+XWARN = # -Wextra -Wconversion
+
+
+C_FILES = termination.c basic.c util.c $(CONTROL_ALGORITHM)
+O_FILES = $(addsuffix .o, $(basename $(C_FILES)))
+
+H_FILES = global.h util.h basic.h control.h
+
+termination : $(O_FILES)
+
+
+# dependencies
+
+$(O_FILES) : $(H_FILES)
+
+$(addsuffix .o, $(basename $(CONTROL_ALGORITHM))) : control.h
+
+
+clean :
+ @$(RM) $(O_FILES) termination
--- /dev/null
+/*
+ Parallel Programming - 2026/2027
+
+ Basic algorithm
+*/
+
+#include <mpi.h>
+#include <pthread.h>
+
+#include <stdio.h>
+#include <stdbool.h>
+#include <unistd.h>
+
+#include "global.h"
+#include "basic.h"
+
+#include "util.h"
+#include "control.h"
+
+#define SEND_BASIC_MESSAGE_PROBABILITY 50
+#define BECOME_PASSIVE_PROBABILITY 30
+
+#define WORK_TIME_MIN 500 // ms
+#define WORK_TIME_MAX 1500 // ms
+
+#define BASIC_MESSAGE 1 // basic message tag
+
+// a process is either active or passive
+typedef enum { ACTIVE, PASSIVE } process_state_t;
+
+// basic messages
+typedef struct { int value; } basic_message_t;
+
+static int pid; // this process id
+
+static bool is_initiator = false;
+static process_state_t state = PASSIVE;
+
+/*
+ Process with id ID becomes active.
+*/
+static void basic_set_active(int id)
+{
+ state = ACTIVE;
+
+ trace("%d: is ACTIVE\n", id);
+
+ control_become_active_hook(id);
+}
+
+/*
+ Process with id ID becomes passive.
+*/
+static void basic_set_passive(int id)
+{
+ state = PASSIVE;
+
+ trace("%d: is PASSIVE\n", id);
+
+ control_become_passive_hook(id);
+}
+
+/*
+ Active process with id ID becomes passive with probability
+ BECOME_PASSIVE_PROBABILITY.
+*/
+static process_state_t basic_new_state(int id)
+{
+ if (rand_ab(1, 100) <= BECOME_PASSIVE_PROBABILITY)
+ basic_set_passive(id);
+
+ return state;
+}
+
+/*
+ Receive a basic message. The process is blocked until a message is
+ received.
+
+ The contents of the message are stored at address PTR and the id of
+ the sender process is stored at PEER.
+*/
+static void basic_receive_message(int id, basic_message_t *ptr, int *peer)
+{
+ MPI_Status status;
+
+ MPI_Recv(ptr, sizeof(basic_message_t), MPI_BYTE, MPI_ANY_SOURCE,
+ BASIC_MESSAGE, MPI_COMM_WORLD, &status);
+
+ *peer = status.MPI_SOURCE;
+
+ control_basic_receive_hook(id, *peer);
+}
+
+/*
+ Receive a basic message, in case there is a pending message
+ (non-blocking).
+
+ If there is a pending message, the contents of the message are
+ stored at address PTR, the id of the sender process is stored at
+ PEER and return true.
+
+ Otherwise, return false.
+ */
+static bool basic_receive_message_maybe(int id, basic_message_t *ptr, int *peer)
+{
+ int flag;
+ MPI_Status status;
+
+ MPI_Iprobe(MPI_ANY_SOURCE, BASIC_MESSAGE, MPI_COMM_WORLD, &flag, &status);
+
+ if (flag == 0)
+ return false;
+
+ MPI_Recv(ptr, sizeof(basic_message_t), MPI_BYTE, MPI_ANY_SOURCE,
+ BASIC_MESSAGE, MPI_COMM_WORLD, &status);
+
+ *peer = status.MPI_SOURCE;
+
+ control_basic_receive_hook(id, *peer);
+
+ return true;
+}
+
+/*
+ Send a basic message, with probability
+ SEND_BASIC_MESSAGE_PROBABILITY, to a random process.
+
+ PROCESSES is the total number of processes running the basic
+ algorithm.
+*/
+static void basic_send_message_maybe(int id, int processes)
+{
+ static int msg_no = 0;
+ int peer;
+ basic_message_t message;
+
+ if (rand_ab(1, 100) > SEND_BASIC_MESSAGE_PROBABILITY)
+ return;
+
+ message.value = ++msg_no;
+
+ // choose the process to send the message to
+ do
+ peer = rand_ab(0, processes - 1);
+ while (peer == id);
+
+ control_basic_send_hook(id, peer);
+
+ MPI_Send(&message, sizeof(basic_message_t), MPI_BYTE, peer, BASIC_MESSAGE,
+ MPI_COMM_WORLD);
+
+ trace("%d: sent %d to %d\n", id, message.value, peer);
+}
+
+/*
+ The basic algorithm:
+
+ - An initiator process is initially active, a non-initiator process
+ is initially passive.
+
+ - While a process is active:
+ - It sends a message to some other process, with probability
+ SEND_BASIC_MESSAGE_PROBABILITY;
+ - It becomes passive with probability BECOME_PASSIVE_PROBABILITY.
+
+ - A passive process waits to be sent a message and the becomes
+ active.
+
+ The arguments of the basic algorithm are the ID of the process, the
+ number of PROCESSES running the algorithm, and whether the process
+ is an initiator.
+*/
+void *basic_algorithm(void *_args)
+{
+ thread_args_t *args = _args;
+ int id = args->id;
+ int processes = args->processes;
+ bool initiator = args->initiator;
+
+ basic_message_t message;
+
+ pid = id;
+
+ is_initiator = initiator;
+
+ // tell the control algorithm that the basic algorithm has started;
+ // allows the synchronisation of both algorithms
+ control_basic_start_hook(id);
+
+ if (!is_initiator)
+ {
+ int peer;
+
+ // wait for a message
+ basic_receive_message(id, &message, &peer);
+
+ trace("%d: got %d from %d\n", id, message.value, peer);
+ }
+
+ // process becomes active
+ basic_set_active(id);
+
+ while (true)
+ {
+ if (state == ACTIVE)
+ {
+ // may send a message to another process
+ basic_send_message_maybe(id, processes);
+
+ // may become passive
+ state = basic_new_state(id);
+ }
+
+ if (state == PASSIVE)
+ {
+ int peer;
+
+ // wait for an incoming message
+ basic_receive_message(id, &message, &peer);
+
+ trace("%d: got %d from %d\n", id, message.value, peer);
+
+ // process becomes active
+ basic_set_active(id);
+ }
+ else
+ {
+ int peer;
+
+ // see if there's an incoming message
+ if (basic_receive_message_maybe(id, &message, &peer))
+ trace("%d: got %d from %d\n", id, message.value, peer);
+ }
+
+ // let some time go by
+ rand_sleep(WORK_TIME_MIN, WORK_TIME_MAX);
+ }
+}
--- /dev/null
+/*
+ Parallel Programming - 2026/2027
+
+ Basic algorithm interface
+*/
+
+void *basic_algorithm(void *);
--- /dev/null
+/*
+ Parallel Programming - 2026/2027
+
+ Skeleton of an implementation of a termination detection algorithm
+*/
+
+#include <mpi.h>
+#include <pthread.h>
+
+#include <stdio.h>
+#include <stdbool.h>
+#include <unistd.h>
+
+#include "global.h"
+#include "control.h"
+#include "util.h"
+
+// the tag(s) of the control messages
+//#define ...
+
+// type for control messages
+//typedef struct {
+// ...
+//} control_message_t;
+
+// local variables
+//static ...
+
+/*
+ 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;
+
+ // 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);
+
+ return NULL;
+}
+
+/*
+ Called at startup of the basic algorithm in process ID.
+*/
+void control_basic_start_hook(int id)
+{
+}
+
+/*
+ Called when the basic algorithm process ID becomes passive.
+*/
+void control_become_passive_hook(int id)
+{
+}
+
+/*
+ Called when the basic algorithm process ID becomes active.
+*/
+void control_become_active_hook(int id)
+{
+}
+
+/*
+ Called when the basic algorithm process ID sends a basic message to
+ process PEER.
+*/
+void control_basic_send_hook(int id, int peer)
+{
+}
+
+/*
+ Called when the basic algorithm process ID receives a basic message
+ from process PEER.
+*/
+void control_basic_receive_hook(int id, int peer)
+{
+}
--- /dev/null
+/*
+ Parallel Programming - 2026/2027
+
+ Control algorithm interface
+*/
+
+void *detect_termination(void *);
+
+void control_basic_start_hook(int);
+
+void control_become_passive_hook(int);
+void control_become_active_hook(int);
+
+void control_basic_send_hook(int, int);
+void control_basic_receive_hook(int, int);
--- /dev/null
+/*
+ Parallel Programming - 2026/2027
+
+ Common variables, constants and types
+*/
+
+#include <stdbool.h>
+
+typedef struct {
+ int id, processes;
+ bool initiator;
+} thread_args_t;
--- /dev/null
+/*
+ Parallel Programming - 2026/2027
+
+ The main program
+*/
+
+#include <mpi.h>
+#include <pthread.h>
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <string.h>
+
+#include "global.h"
+#include "basic.h"
+#include "control.h"
+#include "util.h"
+
+/*
+ Main program:
+
+ - Initialise the MPI environment.
+ - Process command line arguments.
+ - Start the basic algorithm thread.
+ - Start the control algorithm (termination detection) thread.
+ - Wait for the control algorithm to terminate, signalling that the
+ terminatation of the basic algorithm has been detected.
+ - Cancel the basic algorithm thread.
+*/
+int main(int argc, char *argv[])
+{
+ // process data
+ int rank, processes;
+
+ int mpi_threads_support;
+
+ // initialise MPI framework
+ MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &mpi_threads_support);
+
+ MPI_Comm_rank(MPI_COMM_WORLD, &rank); // get process id
+ MPI_Comm_size(MPI_COMM_WORLD, &processes); // get number of processes
+
+ if (mpi_threads_support < MPI_THREAD_MULTIPLE && rank == 0)
+ warn("%s: warning: threads not fully supported\n", argv[0]);
+
+ if (processes < 2)
+ warn("%s: warning: there must be at least 2 processes\n", argv[0]);
+
+ // parse arguments
+ int arg = 1;
+
+ while (arg < argc)
+ {
+ // parse one argument
+ if (strcmp(argv[arg], "--quiet") == 0)
+ {
+ // silence debugging messages
+ QUIET = 1;
+ }
+ else if (strcmp(argv[arg], "-r") == 0)
+ {
+ // re-seed random number generator
+ srand(time(0));
+ }
+ else
+ warn("%s: unknown option `%s'\n", argv[0]);
+
+ arg++;
+ }
+
+ // give all the processes a chance to start
+ MPI_Barrier(MPI_COMM_WORLD);
+
+ // the program
+ if (rank == 0)
+ printf("Starting basic algorithm\n");
+
+ // arguments for the basic and the control algorithms
+ thread_args_t threads_args = { rank, processes, rank == 0 };
+
+ // start basic algorithm thread
+ pthread_t basic_thread;
+
+ if (pthread_create(&basic_thread, NULL, basic_algorithm, &threads_args) != 0)
+ {
+ perror("pthread_create (basic)");
+
+ return 1;
+ }
+
+ // start control thread
+ pthread_t control_thread;
+
+ if (pthread_create(&control_thread, NULL, detect_termination, &threads_args) != 0)
+ {
+ perror("pthread_create (control)");
+
+ return 2;
+ }
+
+ // wait for control thread to end
+ pthread_join(control_thread, NULL);
+
+ // stop (terminated) basic algorithm thread
+ pthread_cancel(basic_thread);
+ pthread_join(basic_thread, NULL);
+
+ // wrap up
+ MPI_Finalize();
+
+ if (rank == 0)
+ printf("Terminating\n");
+
+ return 0;
+}
--- /dev/null
+/*
+ Parallel Programming - 2026/2027
+
+ Utility functions
+*/
+
+#include <stdio.h>
+#include <stdarg.h>
+#include <stdlib.h>
+#include <time.h>
+
+
+// controls the output of debugging messages, set to 1 to have no
+// messages
+int QUIET = 0;
+
+
+// returns a random integer in [A,B]
+int rand_ab(int a, int b)
+{
+ return a + rand() % (b - a + 1);
+}
+
+// sleep between MIN and MAX ms
+void rand_sleep(int min, int max)
+{
+ int ms = rand_ab(min, max);
+ struct timespec ts = { ms / 1000, (ms % 1000) * 1000 };
+
+ nanosleep(&ts, NULL);
+}
+
+// output a debugging message to stderr
+void trace(char *format, ...)
+{
+ va_list args;
+
+ if (QUIET)
+ return;
+
+ va_start (args, format);
+ vfprintf(stderr, format, args);
+ va_end (args);
+}
+
+// unconditionally output a message to stderr
+void warn(char *format, ...)
+{
+ va_list args;
+
+ va_start (args, format);
+ vfprintf(stderr, format, args);
+ va_end (args);
+}
--- /dev/null
+/*
+ Parallel Programming - 2026/2027
+
+ Utility functions header file
+*/
+
+extern int QUIET;
+
+int rand_ab(int, int);
+void rand_sleep(int, int);
+
+void trace(char *, ...);
+void warn(char *, ...);
projects / ue-pp-terminationdetectionalgorithm.git / commitdiff