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