src/msg/msg_actions.c

   1 /* Copyright (c) 2009, 2010. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include "simgrid_config.h"     /* getline */
   8 #include "msg/private.h"
   9 #include "xbt/str.h"
  10 #include "xbt/dynar.h"
  11
  12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_action, msg,
  13                                 "MSG actions for trace driven simulation");
  14
  15 static xbt_dict_t action_funs;
  16 static xbt_dict_t action_queues;
  17
  18 /* To split the file if a unique one is given (specific variable for the other case live in runner()) */
  19 static FILE *action_fp = NULL;
  20 static char *action_line = NULL;
  21 static size_t action_len = 0;
  22
  23 static xbt_dynar_t action_get_action(char *name);
  24
  25 /** \ingroup msg_actions
  26  * \brief Registers a function to handle a kind of action
  27  *
  28  * Registers a function to handle a kind of action
  29  * This table is then used by #MSG_action_trace_run
  30  *
  31  * The argument of the function is the line describing the action, splitted on spaces with xbt_str_split_quoted()
  32  *
  33  * \param name the reference name of the action.
  34  * \param code the function; prototype given by the type: void...(xbt_dynar_t action)
  35  */
  36 void MSG_action_register(const char *action_name, msg_action_fun function)
  37 {
  38   xbt_dict_set(action_funs, action_name, function, NULL);
  39 }
  40
  41 /** \ingroup msg_actions
  42  * \brief Unregisters a function, which handled a kind of action
  43  *
  44  * \param name the reference name of the action.
  45  */
  46 void MSG_action_unregister(const char *action_name)
  47 {
  48   xbt_dict_remove(action_funs, action_name);
  49 }
  50
  51 static int MSG_action_runner(int argc, char *argv[])
  52 {
  53   xbt_dynar_t evt = NULL;
  54   char *line = NULL;
  55   size_t line_len = 0;
  56   FILE *fp = NULL;
  57   char *comment = NULL;
  58   char *evtname = NULL;
  59   ssize_t read;
  60   if (action_fp) {              // A unique trace file
  61
  62     while ((evt = action_get_action(argv[0]))) {
  63       msg_action_fun function =
  64           xbt_dict_get(action_funs, xbt_dynar_get_as(evt, 1, char *));
  65       (*function) (evt);
  66       xbt_dynar_free(&evt);
  67     }
  68   } else {                      // Should have got my trace file in argument
  69     xbt_assert1(argc >= 2,
  70                 "No '%s' agent function provided, no simulation-wide trace file provided to MSG_action_trace_run(), "
  71                 "and no process-wide trace file provided in deployment file. Aborting.",
  72                 argv[0]
  73         );
  74
  75     fp = fopen(argv[1], "r");
  76     xbt_assert2(fp != NULL, "Cannot open %s: %s", argv[1],
  77                 strerror(errno));
  78
  79     // Read lines and execute them until I reach the end of file
  80     while ((read = getline(&line, &line_len, fp)) != -1) {
  81       // cleanup and split the string I just read
  82       comment = strchr(line, '#');
  83       if (comment != NULL)
  84         *comment = '\0';
  85       xbt_str_trim(line, NULL);
  86       if (line[0] == '\0')
  87         continue;
  88       evt = xbt_str_split_quoted(line);
  89
  90       evtname = xbt_dynar_get_as(evt, 0, char *);
  91       if (!strcmp(argv[0], evtname)) {
  92         msg_action_fun function =
  93             xbt_dict_get(action_funs, xbt_dynar_get_as(evt, 1, char *));
  94         (*function) (evt);
  95       } else {
  96         WARN1("Ignore trace element not for me: %s",
  97               xbt_str_join(evt, " "));
  98       }
  99       xbt_dynar_free(&evt);
 100     }
 101
 102   }
 103   return 0;
 104 }
 105
 106 void _MSG_action_init()
 107 {
 108   action_funs = xbt_dict_new();
 109   action_queues = xbt_dict_new();
 110   MSG_function_register_default(MSG_action_runner);
 111 }
 112
 113 void _MSG_action_exit()
 114 {
 115   xbt_dict_free(&action_queues);
 116   xbt_dict_free(&action_funs);
 117 }
 118
 119
 120 static xbt_dynar_t action_get_action(char *name)
 121 {
 122   ssize_t read;
 123   xbt_dynar_t evt = NULL;
 124   char *evtname = NULL;
 125
 126   xbt_dynar_t myqueue = xbt_dict_get_or_null(action_queues, name);
 127   if (myqueue == NULL || xbt_dynar_length(myqueue) == 0) {      // nothing stored for me. Read the file further
 128
 129     if (action_fp == NULL) {    // File closed now. There's nothing more to read. I'm out of here
 130       goto todo_done;
 131     }
 132     // Read lines until I reach something for me (which breaks in loop body)
 133     // or end of file reached
 134     while ((read = getline(&action_line, &action_len, action_fp)) != -1) {
 135       // cleanup and split the string I just read
 136       char *comment = strchr(action_line, '#');
 137       if (comment != NULL)
 138         *comment = '\0';
 139       xbt_str_trim(action_line, NULL);
 140       if (action_line[0] == '\0')
 141         continue;
 142       evt = xbt_str_split_quoted(action_line);
 143
 144       // if it's for me, I'm done
 145       evtname = xbt_dynar_get_as(evt, 0, char *);
 146       if (!strcmp(name, evtname)) {
 147         return evt;
 148       } else {
 149         // Else, I have to store it for the relevant colleague
 150         xbt_dynar_t otherqueue =
 151             xbt_dict_get_or_null(action_queues, evtname);
 152         if (otherqueue == NULL) {       // Damn. Create the queue of that guy
 153           otherqueue =
 154               xbt_dynar_new(sizeof(xbt_dynar_t), xbt_dynar_free_voidp);
 155           xbt_dict_set(action_queues, evtname, otherqueue, NULL);
 156         }
 157         xbt_dynar_push(otherqueue, &evt);
 158       }
 159     }
 160     goto todo_done;             // end of file reached while searching in vain for more work
 161   } else {
 162     // Get something from my queue and return it
 163     xbt_dynar_shift(myqueue, &evt);
 164     return evt;
 165   }
 166
 167
 168   // I did all my actions for me in the file (either I closed the file, or a colleague did)
 169   // Let's cleanup before leaving
 170 todo_done:
 171   if (myqueue != NULL) {
 172     xbt_dynar_free(&myqueue);
 173     xbt_dict_remove(action_queues, name);
 174   }
 175   return NULL;
 176 }
 177
 178 /** \ingroup msg_actions
 179  * \brief A trace loader
 180  *
 181  *  If path!=NULL, load a trace file containing actions, and execute them.
 182  *  Else, assume that each process gets the path in its deployment file
 183  */
 184 MSG_error_t MSG_action_trace_run(char *path)
 185 {
 186   MSG_error_t res;
 187   char *name;
 188   xbt_dynar_t todo;
 189   xbt_dict_cursor_t cursor;
 190
 191   if (path) {
 192     action_fp = fopen(path, "r");
 193     xbt_assert2(action_fp != NULL, "Cannot open %s: %s", path,
 194                 strerror(errno));
 195   }
 196   res = MSG_main();
 197
 198   if (xbt_dict_size(action_queues)) {
 199     WARN0
 200         ("Not all actions got consumed. If the simulation ended successfully (without deadlock), you may want to add new processes to your deployment file.");
 201
 202
 203     xbt_dict_foreach(action_queues, cursor, name, todo) {
 204       WARN2("Still %lu actions for %s", xbt_dynar_length(todo), name);
 205     }
 206   }
 207
 208   if (action_line)
 209     free(action_line);
 210   if (path)
 211     fclose(action_fp);
 212   xbt_dict_free(&action_queues);
 213   action_queues = xbt_dict_new();
 214
 215   return res;
 216 }