examples/msg/parallel_task/test_ptask.c

   1 #include <stdio.h>
   2 #include "msg/msg.h"            /* Yeah! If you want to use msg, you
   3                                    need to include msg/msg.h */
   4 #include "xbt/sysdep.h"         /* calloc, printf */
   5
   6 /* Create a log channel to have nice outputs. */
   7 #include "xbt/log.h"
   8 #include "xbt/asserts.h"
   9 XBT_LOG_NEW_DEFAULT_CATEGORY(msg_test,
  10                              "Messages specific for this msg example");
  11
  12 int execute(int argc, char *argv[]);
  13 int redistribute(int argc, char *argv[]);
  14 MSG_error_t test_all(const char *platform_file, const char *application_file);
  15
  16 typedef enum {
  17   PORT_22 = 0,
  18   MAX_CHANNEL
  19 } channel_t;
  20
  21
  22 int execute(int argc, char *argv[])
  23 {
  24   char buffer[32];
  25   int i, j;
  26   m_host_t *m_host_list = NULL;
  27   m_task_t task = NULL;
  28   int host_list_size;
  29   double *computation_duration = NULL;
  30   double *communication_table = NULL;
  31   double communication_amount;
  32   double computation_amount;
  33   double execution_time;
  34
  35
  36   host_list_size = argc - 3;
  37   DEBUG1("host_list_size=%d", host_list_size);
  38   m_host_list = calloc(host_list_size, sizeof(m_host_t));
  39   for (i = 1; i <= host_list_size; i++) {
  40     m_host_list[i - 1] = MSG_get_host_by_name(argv[i]);
  41     xbt_assert1(m_host_list[i - 1] != NULL,
  42                 "Unknown host %s. Stopping Now! ", argv[i]);
  43   }
  44
  45   xbt_assert1(sscanf(argv[argc - 2], "%lg", &computation_amount),
  46               "Invalid argument %s\n", argv[argc - 2]);
  47   xbt_assert1(sscanf(argv[argc - 1], "%lg", &communication_amount),
  48               "Invalid argument %s\n", argv[argc - 1]);
  49   computation_duration = (double *) calloc(host_list_size, sizeof(double));
  50   communication_table =
  51     (double *) calloc(host_list_size * host_list_size, sizeof(double));
  52   for (i = 0; i < host_list_size; i++) {
  53     computation_duration[i] = computation_amount / host_list_size;
  54     for (j = 0; j < host_list_size; j++)
  55       communication_table[i * host_list_size + j] =
  56         communication_amount / (host_list_size * host_list_size);
  57   }
  58
  59   sprintf(buffer, "redist#0\n");
  60   task = MSG_parallel_task_create(buffer,
  61                                   host_list_size,
  62                                   m_host_list,
  63                                   computation_duration,
  64                                   communication_table, NULL);
  65
  66   execution_time = MSG_get_clock();
  67   MSG_parallel_task_execute(task);
  68   execution_time = MSG_get_clock() - execution_time;
  69
  70   INFO1("execution_time=%g ", execution_time);
  71
  72   return 0;
  73 }
  74
  75
  76 int redistribute(int argc, char *argv[])
  77 {
  78   char buffer[32];
  79   int i, j;
  80   m_host_t *m_host_list = NULL;
  81   m_task_t task = NULL;
  82   int host_list_size;
  83   double *computation_duration = NULL;
  84   double *communication_table = NULL;
  85   double communication_amount;
  86   double redistribution_time;
  87
  88
  89   host_list_size = argc - 2;
  90   DEBUG1("host_list_size=%d", host_list_size);
  91   m_host_list = calloc(host_list_size, sizeof(m_host_t));
  92   for (i = 1; i <= host_list_size; i++) {
  93     m_host_list[i - 1] = MSG_get_host_by_name(argv[i]);
  94     xbt_assert1(m_host_list[i - 1] != NULL,
  95                 "Unknown host %s. Stopping Now! ", argv[i]);
  96   }
  97
  98   xbt_assert1(sscanf(argv[argc - 1], "%lg", &communication_amount),
  99               "Invalid argument %s\n", argv[argc - 1]);
 100   computation_duration = (double *) calloc(host_list_size, sizeof(double));
 101   communication_table =
 102     (double *) calloc(host_list_size * host_list_size, sizeof(double));
 103   for (i = 0; i < host_list_size; i++) {
 104     for (j = 0; j < host_list_size; j++)
 105       communication_table[i * host_list_size + j] =
 106         communication_amount / (host_list_size * host_list_size);
 107   }
 108
 109   sprintf(buffer, "redist#0\n");
 110   task = MSG_parallel_task_create(buffer,
 111                                   host_list_size,
 112                                   m_host_list,
 113                                   computation_duration,
 114                                   communication_table, NULL);
 115
 116   redistribution_time = MSG_get_clock();
 117   MSG_parallel_task_execute(task);
 118   redistribution_time = MSG_get_clock() - redistribution_time;
 119
 120   INFO1("redistribution_time=%g ", redistribution_time);
 121
 122   return 0;
 123 }
 124
 125
 126 MSG_error_t test_all(const char *platform_file, const char *application_file)
 127 {
 128   MSG_error_t res = MSG_OK;
 129
 130
 131   MSG_config("workstation_model", "ptask_L07");
 132
 133   /*  Simulation setting */
 134   MSG_set_channel_number(MAX_CHANNEL);
 135   MSG_create_environment(platform_file);
 136
 137   /*   Application deployment */
 138   MSG_function_register("execute", execute);
 139   MSG_function_register("redistribute", redistribute);
 140   MSG_launch_application(application_file);
 141
 142   res = MSG_main();
 143
 144   INFO1("Simulation time %g", MSG_get_clock());
 145   return res;
 146 }
 147
 148
 149 int main(int argc, char *argv[])
 150 {
 151   MSG_error_t res = MSG_OK;
 152
 153   MSG_global_init(&argc, argv);
 154   if (argc < 3) {
 155     printf("Usage: %s platform_file deployment_file\n", argv[0]);
 156     exit(1);
 157   }
 158   res = test_all(argv[1], argv[2]);
 159   MSG_clean();
 160
 161   if (res == MSG_OK)
 162     return 0;
 163   else
 164     return 1;
 165 }