basic.c

   1 /*
   2   Parallel Programming - 2026/2027
   3
   4   Basic algorithm
   5 */
   6
   7 #include <mpi.h>
   8 #include <pthread.h>
   9
  10 #include <stdio.h>
  11 #include <stdbool.h>
  12 #include <unistd.h>
  13
  14 #include "global.h"
  15 #include "basic.h"
  16
  17 #include "util.h"
  18 #include "control.h"
  19
  20 #define SEND_BASIC_MESSAGE_PROBABILITY  50
  21 #define BECOME_PASSIVE_PROBABILITY      30
  22
  23 #define WORK_TIME_MIN 500               // ms
  24 #define WORK_TIME_MAX 1500              // ms
  25
  26 #define BASIC_MESSAGE 1                 // basic message tag
  27
  28 // a process is either active or passive
  29 typedef enum { ACTIVE, PASSIVE } process_state_t;
  30
  31 // basic messages
  32 typedef struct { int value; } basic_message_t;
  33
  34 static int pid;                         // this process id
  35
  36 static bool is_initiator = false;
  37 static process_state_t state = PASSIVE;
  38
  39 /*
  40   Process with id ID becomes active.
  41 */
  42 static void basic_set_active(int id)
  43 {
  44   state = ACTIVE;
  45
  46   trace("%d: is ACTIVE\n", id);
  47
  48   control_become_active_hook(id);
  49 }
  50
  51 /*
  52   Process with id ID becomes passive.
  53 */
  54 static void basic_set_passive(int id)
  55 {
  56   state = PASSIVE;
  57
  58   trace("%d: is PASSIVE\n", id);
  59
  60   control_become_passive_hook(id);
  61 }
  62
  63 /*
  64   Active process with id ID becomes passive with probability
  65   BECOME_PASSIVE_PROBABILITY.
  66 */
  67 static process_state_t basic_new_state(int id)
  68 {
  69   if (rand_ab(1, 100) <= BECOME_PASSIVE_PROBABILITY)
  70     basic_set_passive(id);
  71
  72   return state;
  73 }
  74
  75 /*
  76   Receive a basic message. The process is blocked until a message is
  77   received.
  78
  79   The contents of the message are stored at address PTR and the id of
  80   the sender process is stored at PEER.
  81 */
  82 static void basic_receive_message(int id, basic_message_t *ptr, int *peer)
  83 {
  84   MPI_Status status;
  85
  86   MPI_Recv(ptr, sizeof(basic_message_t), MPI_BYTE, MPI_ANY_SOURCE,
  87            BASIC_MESSAGE, MPI_COMM_WORLD, &status);
  88
  89   *peer = status.MPI_SOURCE;
  90
  91   control_basic_receive_hook(id, *peer);
  92 }
  93
  94 /*
  95   Receive a basic message, in case there is a pending message
  96   (non-blocking).
  97
  98   If there is a pending message, the contents of the message are
  99   stored at address PTR, the id of the sender process is stored at
 100   PEER and return true.
 101
 102   Otherwise, return false.
 103  */
 104 static bool basic_receive_message_maybe(int id, basic_message_t *ptr, int *peer)
 105 {
 106   int flag;
 107   MPI_Status status;
 108
 109   MPI_Iprobe(MPI_ANY_SOURCE, BASIC_MESSAGE, MPI_COMM_WORLD, &flag, &status);
 110
 111   if (flag == 0)
 112     return false;
 113
 114   MPI_Recv(ptr, sizeof(basic_message_t), MPI_BYTE, MPI_ANY_SOURCE,
 115            BASIC_MESSAGE, MPI_COMM_WORLD, &status);
 116
 117   *peer = status.MPI_SOURCE;
 118
 119   control_basic_receive_hook(id, *peer);
 120
 121   return true;
 122 }
 123
 124 /*
 125   Send a basic message, with probability
 126   SEND_BASIC_MESSAGE_PROBABILITY, to a random process.
 127
 128   PROCESSES is the total number of processes running the basic
 129   algorithm.
 130 */
 131 static void basic_send_message_maybe(int id, int processes)
 132 {
 133   static int msg_no = 0;
 134   int peer;
 135   basic_message_t message;
 136
 137   if (rand_ab(1, 100) > SEND_BASIC_MESSAGE_PROBABILITY)
 138     return;
 139
 140   message.value = ++msg_no;
 141
 142   // choose the process to send the message to
 143   do
 144     peer = rand_ab(0, processes - 1);
 145   while (peer == id);
 146
 147   control_basic_send_hook(id, peer);
 148
 149   MPI_Send(&message, sizeof(basic_message_t), MPI_BYTE, peer, BASIC_MESSAGE,
 150            MPI_COMM_WORLD);
 151
 152   trace("%d: sent %d to %d\n", id, message.value, peer);
 153 }
 154
 155 /*
 156   The basic algorithm:
 157
 158   - An initiator process is initially active, a non-initiator process
 159     is initially passive.
 160
 161   - While a process is active:
 162     - It sends a message to some other process, with probability
 163       SEND_BASIC_MESSAGE_PROBABILITY;
 164     - It becomes passive with probability BECOME_PASSIVE_PROBABILITY.
 165
 166   - A passive process waits to be sent a message and the becomes
 167     active.
 168
 169   The arguments of the basic algorithm are the ID of the process, the
 170   number of PROCESSES running the algorithm, and whether the process
 171   is an initiator.
 172 */
 173 void *basic_algorithm(void *_args)
 174 {
 175   thread_args_t *args = _args;
 176   int id = args->id;
 177   int processes = args->processes;
 178   bool initiator = args->initiator;
 179
 180   basic_message_t message;
 181
 182   pid = id;
 183
 184   is_initiator = initiator;
 185
 186   // tell the control algorithm that the basic algorithm has started;
 187   // allows the synchronisation of both algorithms
 188   control_basic_start_hook(id);
 189
 190   if (!is_initiator)
 191     {
 192       int peer;
 193
 194       // wait for a message
 195       basic_receive_message(id, &message, &peer);
 196
 197       trace("%d: got %d from %d\n", id, message.value, peer);
 198     }
 199
 200   // process becomes active
 201   basic_set_active(id);
 202
 203   while (true)
 204     {
 205       if (state == ACTIVE)
 206         {
 207           // may send a message to another process
 208           basic_send_message_maybe(id, processes);
 209
 210           // may become passive
 211           state = basic_new_state(id);
 212         }
 213
 214       if (state == PASSIVE)
 215         {
 216           int peer;
 217
 218           // wait for an incoming message
 219           basic_receive_message(id, &message, &peer);
 220
 221           trace("%d: got %d from %d\n", id, message.value, peer);
 222
 223           // process becomes active
 224           basic_set_active(id);
 225         }
 226       else
 227         {
 228           int peer;
 229
 230           // see if there's an incoming message
 231           if (basic_receive_message_maybe(id, &message, &peer))
 232             trace("%d: got %d from %d\n", id, message.value, peer);
 233         }
 234
 235       // let some time go by
 236       rand_sleep(WORK_TIME_MIN, WORK_TIME_MAX);
 237     }
 238 }