src/msg/msg_vm.c

   1 /* Copyright (c) 2012. The SimGrid Team. All rights reserved.               */
   2
   3 /* This program is free software; you can redistribute it and/or modify it
   4  * under the terms of the license (GNU LGPL) which comes with this package. */
   5
   6 // QUESTIONS:
   7 // 1./ check how and where a new VM is added to the list of the hosts
   8 // 2./ Diff between SIMIX_Actions and SURF_Actions
   9 // => SIMIX_actions : point synchro entre processus de niveau (theoretically speaking I do not have to create such SIMIX_ACTION
  10 // =>  Surf_Actions
  11
  12 // TODO
  13 //      MSG_TRACE can be revisited in order to use  the host
  14 //      To implement a mixed model between workstation and vm_workstation,
  15 //     please give a look at surf_model_private_t model_private at SURF Level and to the share resource functions
  16 //     double (*share_resources) (double now);
  17 //      For the action into the vm workstation model, we should be able to leverage the usual one (and if needed, look at
  18 //              the workstation model.
  19
  20 #include "msg_private.h"
  21 #include "xbt/sysdep.h"
  22 #include "xbt/log.h"
  23
  24 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_vm, msg,
  25                                 "Cloud-oriented parts of the MSG API");
  26
  27
  28 /* **** ******** GENERAL ********* **** */
  29
  30 /** \ingroup m_vm_management
  31  * \brief Returns the value of a given vm property
  32  *
  33  * \param vm a vm
  34  * \param name a property name
  35  * \return value of a property (or NULL if property not set)
  36  */
  37
  38 const char *MSG_vm_get_property_value(msg_vm_t vm, const char *name)
  39 {
  40   return MSG_host_get_property_value(vm, name);
  41 }
  42
  43 /** \ingroup m_vm_management
  44  * \brief Returns a xbt_dict_t consisting of the list of properties assigned to this host
  45  *
  46  * \param vm a vm
  47  * \return a dict containing the properties
  48  */
  49 xbt_dict_t MSG_vm_get_properties(msg_vm_t vm)
  50 {
  51   xbt_assert((vm != NULL), "Invalid parameters (vm is NULL)");
  52
  53   return (simcall_host_get_properties(vm));
  54 }
  55
  56 /** \ingroup m_host_management
  57  * \brief Change the value of a given host property
  58  *
  59  * \param host a host
  60  * \param name a property name
  61  * \param value what to change the property to
  62  * \param free_ctn the freeing function to use to kill the value on need
  63  */
  64 void MSG_vm_set_property_value(msg_vm_t vm, const char *name, void *value, void_f_pvoid_t free_ctn)
  65 {
  66   xbt_dict_set(MSG_host_get_properties(vm), name, value, free_ctn);
  67 }
  68
  69 /** \ingroup msg_vm_management
  70  * \brief Finds a msg_vm_t using its name.
  71  *
  72  * This is a name directory service
  73  * \param name the name of a vm.
  74  * \return the corresponding vm
  75  *
  76  * Please note that a VM is a specific host. Hence, you should give a different name
  77  * for each VM/PM.
  78  */
  79
  80 msg_vm_t MSG_vm_get_by_name(const char *name)
  81 {
  82         return MSG_get_host_by_name(name);
  83 }
  84
  85 /** \ingroup m_vm_management
  86  *
  87  * \brief Return the name of the #msg_host_t.
  88  *
  89  * This functions checks whether \a host is a valid pointer or not and return
  90    its name.
  91  */
  92 const char *MSG_vm_get_name(msg_vm_t vm)
  93 {
  94   return MSG_host_get_name(vm);
  95 }
  96
  97
  98 /* **** Check state of a VM **** */
  99 static inline int __MSG_vm_is_state(msg_vm_t vm, e_surf_vm_state_t state)
 100 {
 101   return simcall_vm_get_state(vm) == state;
 102 }
 103
 104 /** @brief Returns whether the given VM has just reated, not running.
 105  *  @ingroup msg_VMs
 106  */
 107 int MSG_vm_is_created(msg_vm_t vm)
 108 {
 109   return __MSG_vm_is_state(vm, SURF_VM_STATE_CREATED);
 110 }
 111
 112 /** @brief Returns whether the given VM is currently running
 113  *  @ingroup msg_VMs
 114  */
 115 int MSG_vm_is_running(msg_vm_t vm)
 116 {
 117   return __MSG_vm_is_state(vm, SURF_VM_STATE_RUNNING);
 118 }
 119
 120 /** @brief Returns whether the given VM is currently migrating
 121  *  @ingroup msg_VMs
 122  */
 123 int MSG_vm_is_migrating(msg_vm_t vm)
 124 {
 125   return __MSG_vm_is_state(vm, SURF_VM_STATE_MIGRATING);
 126 }
 127
 128 /** @brief Returns whether the given VM is currently suspended, not running.
 129  *  @ingroup msg_VMs
 130  */
 131 int MSG_vm_is_suspended(msg_vm_t vm)
 132 {
 133   return __MSG_vm_is_state(vm, SURF_VM_STATE_SUSPENDED);
 134 }
 135
 136 /** @brief Returns whether the given VM is being saved (FIXME: live saving or not?).
 137  *  @ingroup msg_VMs
 138  */
 139 int MSG_vm_is_saving(msg_vm_t vm)
 140 {
 141   return __MSG_vm_is_state(vm, SURF_VM_STATE_SAVING);
 142 }
 143
 144 /** @brief Returns whether the given VM has been saved, not running.
 145  *  @ingroup msg_VMs
 146  */
 147 int MSG_vm_is_saved(msg_vm_t vm)
 148 {
 149   return __MSG_vm_is_state(vm, SURF_VM_STATE_SAVED);
 150 }
 151
 152 /** @brief Returns whether the given VM is being restored, not running.
 153  *  @ingroup msg_VMs
 154  */
 155 int MSG_vm_is_restoring(msg_vm_t vm)
 156 {
 157   return __MSG_vm_is_state(vm, SURF_VM_STATE_RESTORING);
 158 }
 159
 160
 161
 162 /* ------------------------------------------------------------------------- */
 163 /* ------------------------------------------------------------------------- */
 164
 165 /* **** ******** MSG vm actions ********* **** */
 166
 167 /** @brief Create a new VM with specified parameters.
 168  *  @ingroup msg_VMs*
 169  *  All parameters are in MBytes
 170  *
 171  */
 172 msg_vm_t MSG_vm_create(msg_host_t ind_pm, const char *name, int ncpus, int ramsize,
 173                                              int net_cap, char *disk_path, int disksize,
 174                                                  int mig_netspeed, int dp_intensity)
 175 {
 176         /* For the moment, intensity_rate is the percentage against the migration bandwidth */
 177         double host_speed = MSG_get_host_speed(ind_pm);
 178         double update_speed = ((double)dp_intensity/100) * mig_netspeed;
 179
 180         msg_vm_t vm = MSG_vm_create_core(ind_pm, name);
 181         s_ws_params_t params;
 182         memset(&params, 0, sizeof(params));
 183         params.ramsize = 1L * 1024 * 1024 * ramsize;
 184         //params.overcommit = 0;
 185         params.devsize = 0;
 186         params.skip_stage2 = 0;
 187         params.max_downtime = 0.03;
 188         params.dp_rate = (update_speed * 1L * 1024 * 1024 ) / host_speed;
 189         params.dp_cap = params.ramsize / 0.9; // working set memory is 90%
 190         params.mig_speed = 1L * 1024 * 1024 * mig_netspeed; // mig_speed
 191
 192    //XBT_INFO("dp rate %f migspeed : %f intensity mem : %d, updatespeed %f, hostspeed %f",params.dp_rate, params.mig_speed, dp_intensity, update_speed, host_speed);
 193         simcall_host_set_params(vm, &params);
 194
 195         return vm;
 196 }
 197
 198
 199 /** @brief Create a new VM object. The VM is not yet started. The resource of the VM is allocated upon MSG_vm_start().
 200  *  @ingroup msg_VMs*
 201  *
 202  * A VM is treated as a host. The name of the VM must be unique among all hosts.
 203  */
 204 msg_vm_t MSG_vm_create_core(msg_host_t ind_pm, const char *name)
 205 {
 206   /* make sure the VM of the same name does not exit */
 207   {
 208     void *ind_host_tmp = xbt_lib_get_elm_or_null(host_lib, name);
 209     if (ind_host_tmp) {
 210       XBT_ERROR("host %s already exits", name);
 211       return NULL;
 212     }
 213   }
 214
 215   /* Note: ind_vm and vm_workstation point to the same elm object. */
 216   msg_vm_t ind_vm = NULL;
 217   void *ind_vm_workstation =  NULL;
 218
 219   /* Ask the SIMIX layer to create the surf vm resource */
 220   ind_vm_workstation = simcall_vm_create(name, ind_pm);
 221   ind_vm = (msg_vm_t) __MSG_host_create(ind_vm_workstation);
 222
 223   XBT_DEBUG("A new VM (%s) has been created", name);
 224
 225   #ifdef HAVE_TRACING
 226   TRACE_msg_vm_create(name, ind_pm);
 227   #endif
 228
 229   return ind_vm;
 230 }
 231
 232 /** @brief Destroy a VM. Destroy the VM object from the simulation.
 233  *  @ingroup msg_VMs
 234  */
 235 void MSG_vm_destroy(msg_vm_t vm)
 236 {
 237   /* First, terminate all processes on the VM if necessary */
 238   if (MSG_vm_is_running(vm))
 239       simcall_vm_shutdown(vm);
 240
 241   if (!MSG_vm_is_created(vm)) {
 242     XBT_CRITICAL("shutdown the given VM before destroying it");
 243     DIE_IMPOSSIBLE;
 244   }
 245
 246   /* Then, destroy the VM object */
 247   simcall_vm_destroy(vm);
 248
 249   __MSG_host_destroy(vm);
 250
 251   #ifdef HAVE_TRACING
 252   TRACE_msg_vm_end(vm);
 253   #endif
 254 }
 255
 256
 257 /** @brief Start a vm (i.e., boot the guest operating system)
 258  *  @ingroup msg_VMs
 259  *
 260  *  If the VM cannot be started, an exception is generated.
 261  *
 262  */
 263 void MSG_vm_start(msg_vm_t vm)
 264 {
 265   simcall_vm_start(vm);
 266
 267   #ifdef HAVE_TRACING
 268   TRACE_msg_vm_start(vm);
 269   #endif
 270 }
 271
 272
 273
 274 /** @brief Immediately kills all processes within the given VM. Any memory that they allocated will be leaked.
 275  *  @ingroup msg_VMs
 276  *
 277  * FIXME: No extra delay occurs. If you want to simulate this too, you want to
 278  * use a #MSG_process_sleep() or something. I'm not quite sure.
 279  */
 280 void MSG_vm_shutdown(msg_vm_t vm)
 281 {
 282   /* msg_vm_t equals to msg_host_t */
 283   simcall_vm_shutdown(vm);
 284
 285   // #ifdef HAVE_TRACING
 286   // TRACE_msg_vm_(vm);
 287   // #endif
 288 }
 289
 290
 291
 292 /* We have two mailboxes. mbox is used to transfer migration data between
 293  * source and destiantion PMs. mbox_ctl is used to detect the completion of a
 294  * migration. The names of these mailboxes must not conflict with others. */
 295 static inline char *get_mig_mbox_src_dst(const char *vm_name, const char *src_pm_name, const char *dst_pm_name)
 296 {
 297   return bprintf("__mbox_mig_src_dst:%s(%s-%s)", vm_name, src_pm_name, dst_pm_name);
 298 }
 299
 300 static inline char *get_mig_mbox_ctl(const char *vm_name, const char *src_pm_name, const char *dst_pm_name)
 301 {
 302   return bprintf("__mbox_mig_ctl:%s(%s-%s)", vm_name, src_pm_name, dst_pm_name);
 303 }
 304
 305 static inline char *get_mig_process_tx_name(const char *vm_name, const char *src_pm_name, const char *dst_pm_name)
 306 {
 307   return bprintf("__pr_mig_tx:%s(%s-%s)", vm_name, src_pm_name, dst_pm_name);
 308 }
 309
 310 static inline char *get_mig_process_rx_name(const char *vm_name, const char *src_pm_name, const char *dst_pm_name)
 311 {
 312   return bprintf("__pr_mig_rx:%s(%s-%s)", vm_name, src_pm_name, dst_pm_name);
 313 }
 314
 315 static inline char *get_mig_task_name(const char *vm_name, const char *src_pm_name, const char *dst_pm_name, int stage)
 316 {
 317   return bprintf("__task_mig_stage%d:%s(%s-%s)", stage, vm_name, src_pm_name, dst_pm_name);
 318 }
 319
 320 static void launch_deferred_exec_process(msg_host_t host, double computation, double prio);
 321
 322 static int migration_rx_fun(int argc, char *argv[])
 323 {
 324   XBT_DEBUG("mig: rx_start");
 325
 326   xbt_assert(argc == 4);
 327   const char *vm_name = argv[1];
 328   const char *src_pm_name  = argv[2];
 329   const char *dst_pm_name  = argv[3];
 330   msg_vm_t vm = MSG_get_host_by_name(vm_name);
 331   msg_vm_t dst_pm = MSG_get_host_by_name(dst_pm_name);
 332
 333
 334   s_ws_params_t params;
 335   simcall_host_get_params(vm, &params);
 336   const double xfer_cpu_overhead = params.xfer_cpu_overhead;
 337
 338
 339   int need_exit = 0;
 340
 341   char *mbox = get_mig_mbox_src_dst(vm_name, src_pm_name, dst_pm_name);
 342   char *mbox_ctl = get_mig_mbox_ctl(vm_name, src_pm_name, dst_pm_name);
 343   char *finalize_task_name = get_mig_task_name(vm_name, src_pm_name, dst_pm_name, 3);
 344
 345   for (;;) {
 346     msg_task_t task = NULL;
 347     MSG_task_recv(&task, mbox);
 348     {
 349       double received = MSG_task_get_data_size(task);
 350       /* TODO: clean up */
 351       // const double alpha = 0.22L * 1.0E8 / (80L * 1024 * 1024);
 352       launch_deferred_exec_process(vm, received * xfer_cpu_overhead, 1);
 353     }
 354
 355     if (strcmp(task->name, finalize_task_name) == 0)
 356       need_exit = 1;
 357
 358     MSG_task_destroy(task);
 359
 360     if (need_exit)
 361       break;
 362   }
 363
 364
 365   simcall_vm_migrate(vm, dst_pm);
 366   simcall_vm_resume(vm);
 367
 368   {
 369     char *task_name = get_mig_task_name(vm_name, src_pm_name, dst_pm_name, 4);
 370
 371     msg_task_t task = MSG_task_create(task_name, 0, 0, NULL);
 372     msg_error_t ret = MSG_task_send(task, mbox_ctl);
 373     xbt_assert(ret == MSG_OK);
 374
 375     xbt_free(task_name);
 376   }
 377
 378
 379   xbt_free(mbox);
 380   xbt_free(mbox_ctl);
 381   xbt_free(finalize_task_name);
 382
 383   XBT_DEBUG("mig: rx_done");
 384
 385   return 0;
 386 }
 387
 388 static void reset_dirty_pages(msg_vm_t vm)
 389 {
 390   msg_host_priv_t priv = msg_host_resource_priv(vm);
 391
 392   char *key = NULL;
 393   xbt_dict_cursor_t cursor = NULL;
 394   dirty_page_t dp = NULL;
 395   xbt_dict_foreach(priv->dp_objs, cursor, key, dp) {
 396     double remaining = MSG_task_get_remaining_computation(dp->task);
 397     dp->prev_clock = MSG_get_clock();
 398     dp->prev_remaining = remaining;
 399
 400     // XBT_INFO("%s@%s remaining %f", key, sg_host_name(vm), remaining);
 401   }
 402 }
 403
 404 static void start_dirty_page_tracking(msg_vm_t vm)
 405 {
 406   msg_host_priv_t priv = msg_host_resource_priv(vm);
 407   priv->dp_enabled = 1;
 408
 409   reset_dirty_pages(vm);
 410 }
 411
 412 static void stop_dirty_page_tracking(msg_vm_t vm)
 413 {
 414   msg_host_priv_t priv = msg_host_resource_priv(vm);
 415   priv->dp_enabled = 0;
 416 }
 417
 418 #if 0
 419 /* It might be natural that we define dp_rate for each task. But, we will also
 420  * have to care about how each task behavior affects the memory update behavior
 421  * at the operating system level. It may not be easy to model it with a simple algorithm. */
 422 double calc_updated_pages(char *key, msg_vm_t vm, dirty_page_t dp, double remaining, double clock)
 423 {
 424     double computed = dp->prev_remaining - remaining;
 425     double duration = clock - dp->prev_clock;
 426     double updated = dp->task->dp_rate * computed;
 427
 428     XBT_INFO("%s@%s: computated %f ops (remaining %f -> %f) in %f secs (%f -> %f)",
 429         key, sg_host_name(vm), computed, dp->prev_remaining, remaining, duration, dp->prev_clock, clock);
 430     XBT_INFO("%s@%s: updated %f bytes, %f Mbytes/s",
 431         key, sg_host_name(vm), updated, updated / duration / 1000 / 1000);
 432
 433     return updated;
 434 }
 435 #endif
 436
 437 static double get_computed(char *key, msg_vm_t vm, dirty_page_t dp, double remaining, double clock)
 438 {
 439   double computed = dp->prev_remaining - remaining;
 440   double duration = clock - dp->prev_clock;
 441
 442   XBT_DEBUG("%s@%s: computated %f ops (remaining %f -> %f) in %f secs (%f -> %f)",
 443       key, sg_host_name(vm), computed, dp->prev_remaining, remaining, duration, dp->prev_clock, clock);
 444
 445   return computed;
 446 }
 447
 448 static double lookup_computed_flop_counts(msg_vm_t vm, int stage_for_fancy_debug, int stage2_round_for_fancy_debug)
 449 {
 450   msg_host_priv_t priv = msg_host_resource_priv(vm);
 451   double total = 0;
 452
 453   char *key = NULL;
 454   xbt_dict_cursor_t cursor = NULL;
 455   dirty_page_t dp = NULL;
 456   xbt_dict_foreach(priv->dp_objs, cursor, key, dp) {
 457     double remaining = MSG_task_get_remaining_computation(dp->task);
 458     double clock = MSG_get_clock();
 459
 460     // total += calc_updated_pages(key, vm, dp, remaining, clock);
 461     total += get_computed(key, vm, dp, remaining, clock);
 462
 463     dp->prev_remaining = remaining;
 464     dp->prev_clock = clock;
 465   }
 466
 467   total += priv->dp_updated_by_deleted_tasks;
 468
 469   XBT_INFO("mig-stage%d.%d: computed %f flop_counts (including %f by deleted tasks)",
 470       stage_for_fancy_debug,
 471       stage2_round_for_fancy_debug,
 472       total, priv->dp_updated_by_deleted_tasks);
 473
 474
 475
 476   priv->dp_updated_by_deleted_tasks = 0;
 477
 478
 479   return total;
 480 }
 481
 482 // TODO Is this code redundant with the information provided by
 483 // msg_process_t MSG_process_create(const char *name, xbt_main_func_t code, void *data, msg_host_t host)
 484 void MSG_host_add_task(msg_host_t host, msg_task_t task)
 485 {
 486   msg_host_priv_t priv = msg_host_resource_priv(host);
 487   double remaining = MSG_task_get_remaining_computation(task);
 488   char *key = bprintf("%s-%lld", task->name, task->counter);
 489
 490   dirty_page_t dp = xbt_new0(s_dirty_page, 1);
 491   dp->task = task;
 492
 493   /* It should be okay that we add a task onto a migrating VM. */
 494   if (priv->dp_enabled) {
 495     dp->prev_clock = MSG_get_clock();
 496     dp->prev_remaining = remaining;
 497   }
 498
 499   xbt_assert(xbt_dict_get_or_null(priv->dp_objs, key) == NULL);
 500   xbt_dict_set(priv->dp_objs, key, dp, NULL);
 501   XBT_DEBUG("add %s on %s (remaining %f, dp_enabled %d)", key, sg_host_name(host), remaining, priv->dp_enabled);
 502
 503   xbt_free(key);
 504 }
 505
 506 void MSG_host_del_task(msg_host_t host, msg_task_t task)
 507 {
 508   msg_host_priv_t priv = msg_host_resource_priv(host);
 509
 510   char *key = bprintf("%s-%lld", task->name, task->counter);
 511
 512   dirty_page_t dp = xbt_dict_get_or_null(priv->dp_objs, key);
 513   xbt_assert(dp->task == task);
 514
 515   /* If we are in the middle of dirty page tracking, we record how much
 516    * computaion has been done until now, and keep the information for the
 517    * lookup_() function that will called soon. */
 518   if (priv->dp_enabled) {
 519     double remaining = MSG_task_get_remaining_computation(task);
 520     double clock = MSG_get_clock();
 521     // double updated = calc_updated_pages(key, host, dp, remaining, clock);
 522     double updated = get_computed(key, host, dp, remaining, clock);
 523
 524     priv->dp_updated_by_deleted_tasks += updated;
 525   }
 526
 527   xbt_dict_remove(priv->dp_objs, key);
 528   xbt_free(dp);
 529
 530   XBT_DEBUG("del %s on %s", key, sg_host_name(host));
 531
 532   xbt_free(key);
 533 }
 534
 535
 536 static int deferred_exec_fun(int argc, char *argv[])
 537 {
 538   xbt_assert(argc == 3);
 539   const char *comp_str = argv[1];
 540   double computaion = atof(comp_str);
 541   const char *prio_str = argv[2];
 542   double prio = atof(prio_str);
 543
 544   msg_task_t task = MSG_task_create("__task_deferred", computaion, 0, NULL);
 545   // XBT_INFO("exec deferred %f", computaion);
 546
 547   /* dpt is the results of the VM activity */
 548   MSG_task_set_priority(task, prio);
 549   MSG_task_execute(task);
 550
 551
 552
 553   MSG_task_destroy(task);
 554
 555   return 0;
 556 }
 557
 558 static void launch_deferred_exec_process(msg_host_t host, double computation, double prio)
 559 {
 560   char *pr_name = bprintf("__pr_deferred_exec_%s", MSG_host_get_name(host));
 561
 562   int nargvs = 4;
 563   char **argv = xbt_new(char *, nargvs);
 564   argv[0] = xbt_strdup(pr_name);
 565   argv[1] = bprintf("%lf", computation);
 566   argv[2] = bprintf("%lf", prio);
 567   argv[3] = NULL;
 568
 569   MSG_process_create_with_arguments(pr_name, deferred_exec_fun, NULL, host, nargvs - 1, argv);
 570
 571   xbt_free(pr_name);
 572 }
 573
 574
 575 static int task_tx_overhead_fun(int argc, char *argv[])
 576 {
 577   xbt_assert(argc == 2);
 578   const char *mbox = argv[1];
 579
 580   int need_exit = 0;
 581
 582   // XBT_INFO("start %s", mbox);
 583
 584   for (;;) {
 585     msg_task_t task = NULL;
 586     MSG_task_recv(&task, mbox);
 587
 588     // XBT_INFO("task->name %s", task->name);
 589
 590     if (strcmp(task->name, "finalize_making_overhead") == 0)
 591       need_exit = 1;
 592
 593     // XBT_INFO("exec");
 594     // MSG_task_set_priority(task, 1000000);
 595     MSG_task_execute(task);
 596     MSG_task_destroy(task);
 597
 598     if (need_exit)
 599       break;
 600   }
 601
 602   // XBT_INFO("bye");
 603
 604   return 0;
 605 }
 606
 607 static void start_overhead_process(msg_task_t comm_task)
 608 {
 609   char *pr_name = bprintf("__pr_task_tx_overhead_%s", MSG_task_get_name(comm_task));
 610   char *mbox    = bprintf("__mb_task_tx_overhead_%s", MSG_task_get_name(comm_task));
 611
 612   int nargvs = 3;
 613   char **argv = xbt_new(char *, nargvs);
 614   argv[0] = xbt_strdup(pr_name);
 615   argv[1] = xbt_strdup(mbox);
 616   argv[2] = NULL;
 617
 618   // XBT_INFO("micro start: mbox %s", mbox);
 619   MSG_process_create_with_arguments(pr_name, task_tx_overhead_fun, NULL, MSG_host_self(), nargvs - 1, argv);
 620
 621   xbt_free(pr_name);
 622   xbt_free(mbox);
 623 }
 624
 625 static void shutdown_overhead_process(msg_task_t comm_task)
 626 {
 627   char *mbox = bprintf("__mb_task_tx_overhead_%s", MSG_task_get_name(comm_task));
 628
 629   msg_task_t task = MSG_task_create("finalize_making_overhead", 0, 0, NULL);
 630
 631   // XBT_INFO("micro shutdown: mbox %s", mbox);
 632   msg_error_t ret = MSG_task_send(task, mbox);
 633   xbt_assert(ret == MSG_OK);
 634
 635   xbt_free(mbox);
 636   // XBT_INFO("shutdown done");
 637 }
 638
 639 static void request_overhead(msg_task_t comm_task, double computation)
 640 {
 641   char *mbox = bprintf("__mb_task_tx_overhead_%s", MSG_task_get_name(comm_task));
 642
 643   msg_task_t task = MSG_task_create("micro", computation, 0, NULL);
 644
 645   // XBT_INFO("req overhead");
 646   msg_error_t ret = MSG_task_send(task, mbox);
 647   xbt_assert(ret == MSG_OK);
 648
 649   xbt_free(mbox);
 650 }
 651
 652 /* alpha is (floating_operations / bytes).
 653  *
 654  * When actual migration traffic was 32 mbytes/s, we observed the CPU
 655  * utilization of the main thread of the Qemu process was 10 %.
 656  *   alpha = 0.1 * C / (32 * 1024 * 1024)
 657  * where the CPU capacity of the PM is C flops/s.
 658  *
 659  * */
 660 static void task_send_bounded_with_cpu_overhead(msg_task_t comm_task, char *mbox, double mig_speed, double alpha)
 661 {
 662   const double chunk_size = 1024 * 1024 * 10;
 663   double remaining = MSG_task_get_data_size(comm_task);
 664
 665   start_overhead_process(comm_task);
 666
 667
 668   while (remaining > 0) {
 669     double data_size = chunk_size;
 670     if (remaining < chunk_size)
 671       data_size = remaining;
 672
 673     remaining -= data_size;
 674
 675     // XBT_INFO("remaining %f bytes", remaining);
 676
 677
 678     double clock_sta = MSG_get_clock();
 679
 680     /* create a micro task */
 681     {
 682       char *mtask_name = bprintf("__micro_%s", MSG_task_get_name(comm_task));
 683       msg_task_t mtask = MSG_task_create(mtask_name, 0, data_size, NULL);
 684
 685       request_overhead(comm_task, data_size * alpha);
 686
 687       msg_error_t ret = MSG_task_send(mtask, mbox);
 688       xbt_assert(ret == MSG_OK);
 689
 690       xbt_free(mtask_name);
 691     }
 692
 693 #if 0
 694     {
 695       /* In the real world, sending data involves small CPU computation. */
 696       char *mtask_name = bprintf("__micro_%s", MSG_task_get_name(comm_task));
 697       msg_task_t mtask = MSG_task_create(mtask_name, data_size * alpha, data_size, NULL);
 698       MSG_task_execute(mtask);
 699       MSG_task_destroy(mtask);
 700       xbt_free(mtask_name);
 701     }
 702 #endif
 703
 704     /* TODO */
 705
 706     double clock_end = MSG_get_clock();
 707
 708
 709     if (mig_speed > 0) {
 710       /*
 711        * (max bandwidth) > data_size / ((elapsed time) + time_to_sleep)
 712        *
 713        * Thus, we get
 714        *   time_to_sleep > data_size / (max bandwidth) - (elapsed time)
 715        *
 716        * If time_to_sleep is smaller than zero, the elapsed time was too big. We
 717        * do not need a micro sleep.
 718        **/
 719       double time_to_sleep = data_size / mig_speed - (clock_end - clock_sta);
 720       if (time_to_sleep > 0)
 721         MSG_process_sleep(time_to_sleep);
 722
 723
 724       //XBT_INFO("duration %f", clock_end - clock_sta);
 725       //XBT_INFO("time_to_sleep %f", time_to_sleep);
 726     }
 727   }
 728
 729   // XBT_INFO("%s", MSG_task_get_name(comm_task));
 730   shutdown_overhead_process(comm_task);
 731
 732 }
 733
 734
 735 #if 0
 736 static void make_cpu_overhead_of_data_transfer(msg_task_t comm_task, double init_comm_size)
 737 {
 738   double prev_remaining = init_comm_size;
 739
 740   for (;;) {
 741     double remaining = MSG_task_get_remaining_communication(comm_task);
 742     if (remaining == 0)
 743       need_exit = 1;
 744
 745     double sent = prev_remaining - remaining;
 746     double comp_size = sent * overhead;
 747
 748
 749     char *comp_task_name = bprintf("__sender_overhead%s", MSG_task_get_name(comm_task));
 750     msg_task_t comp_task = MSG_task_create(comp_task_name, comp_size, 0, NULL);
 751     MSG_task_execute(comp_task);
 752     MSG_task_destroy(comp_task);
 753
 754     if (need_exit)
 755       break;
 756
 757     prev_remaining = remaining;
 758
 759   }
 760
 761   xbt_free(comp_task_name);
 762 }
 763 #endif
 764
 765 #define USE_MICRO_TASK 1
 766
 767 #if 0
 768 // const double alpha = 0.1L * 1.0E8 / (32L * 1024 * 1024);
 769 // const double alpha = 0.25L * 1.0E8 / (85L * 1024 * 1024);
 770 // const double alpha = 0.20L * 1.0E8 / (85L * 1024 * 1024);
 771 // const double alpha = 0.25L * 1.0E8 / (85L * 1024 * 1024);
 772 // const double alpha = 0.32L * 1.0E8 / (24L * 1024 * 1024);   // makes super good values for 32 mbytes/s
 773 //const double alpha = 0.32L * 1.0E8 / (32L * 1024 * 1024);
 774 // const double alpha = 0.56L * 1.0E8 / (80L * 1024 * 1024);
 775 ////const double alpha = 0.20L * 1.0E8 / (80L * 1024 * 1024);
 776 // const double alpha = 0.56L * 1.0E8 / (90L * 1024 * 1024);
 777 // const double alpha = 0.66L * 1.0E8 / (90L * 1024 * 1024);
 778 // const double alpha = 0.20L * 1.0E8 / (80L * 1024 * 1024);
 779
 780 /* CPU 22% when 80Mbyte/s */
 781 const double alpha = 0.22L * 1.0E8 / (80L * 1024 * 1024);
 782 #endif
 783
 784
 785 static void send_migration_data(const char *vm_name, const char *src_pm_name, const char *dst_pm_name,
 786     double size, char *mbox, int stage, int stage2_round, double mig_speed, double xfer_cpu_overhead)
 787 {
 788   char *task_name = get_mig_task_name(vm_name, src_pm_name, dst_pm_name, stage);
 789   msg_task_t task = MSG_task_create(task_name, 0, size, NULL);
 790
 791   /* TODO: clean up */
 792
 793   double clock_sta = MSG_get_clock();
 794
 795 #ifdef USE_MICRO_TASK
 796
 797   task_send_bounded_with_cpu_overhead(task, mbox, mig_speed, xfer_cpu_overhead);
 798
 799 #else
 800   msg_error_t ret;
 801   if (mig_speed > 0)
 802     ret = MSG_task_send_bounded(task, mbox, mig_speed);
 803   else
 804     ret = MSG_task_send(task, mbox);
 805   xbt_assert(ret == MSG_OK);
 806 #endif
 807
 808   double clock_end = MSG_get_clock();
 809   double duration = clock_end - clock_sta;
 810   double actual_speed = size / duration;
 811 #ifdef USE_MICRO_TASK
 812   double cpu_utilization = size * xfer_cpu_overhead / duration / 1.0E8;
 813 #else
 814   double cpu_utilization = 0;
 815 #endif
 816
 817
 818
 819
 820   if (stage == 2)
 821     XBT_INFO("mig-stage%d.%d: sent %f duration %f actual_speed %f (target %f) cpu %f", stage, stage2_round, size, duration, actual_speed, mig_speed, cpu_utilization);
 822   else
 823     XBT_INFO("mig-stage%d: sent %f duration %f actual_speed %f (target %f) cpu %f", stage, size, duration, actual_speed, mig_speed, cpu_utilization);
 824
 825   xbt_free(task_name);
 826
 827
 828
 829 #ifdef USE_MICRO_TASK
 830   /* The name of a micro task starts with __micro, which does not match the
 831    * special name that finalizes the receiver loop. Thus, we send the special task.
 832    **/
 833   {
 834     if (stage == 3) {
 835       char *task_name = get_mig_task_name(vm_name, src_pm_name, dst_pm_name, stage);
 836       msg_task_t task = MSG_task_create(task_name, 0, 0, NULL);
 837       msg_error_t ret = MSG_task_send(task, mbox);
 838       xbt_assert(ret == MSG_OK);
 839       xbt_free(task_name);
 840     }
 841   }
 842 #endif
 843 }
 844
 845 static double get_updated_size(double computed, double dp_rate, double dp_cap)
 846 {
 847   double updated_size = computed * dp_rate;
 848   XBT_INFO("updated_size %f dp_rate %f", updated_size, dp_rate);
 849   if (updated_size > dp_cap) {
 850     // XBT_INFO("mig-stage2.%d: %f bytes updated, but cap it with the working set size %f", stage2_round, updated_size, dp_cap);
 851     updated_size = dp_cap;
 852   }
 853
 854   return updated_size;
 855 }
 856
 857 static double send_stage1(msg_host_t vm, const char *src_pm_name, const char *dst_pm_name,
 858     long ramsize, double mig_speed, double xfer_cpu_overhead, double dp_rate, double dp_cap, double dpt_cpu_overhead)
 859 {
 860   const char *vm_name = MSG_host_get_name(vm);
 861   char *mbox = get_mig_mbox_src_dst(vm_name, src_pm_name, dst_pm_name);
 862
 863   const long chunksize = 1024 * 1024 * 100;
 864   long remaining = ramsize;
 865   double computed_total = 0;
 866
 867   while (remaining > 0) {
 868     long datasize = chunksize;
 869     if (remaining < chunksize)
 870       datasize = remaining;
 871
 872     remaining -= datasize;
 873
 874     send_migration_data(vm_name, src_pm_name, dst_pm_name, datasize, mbox, 1, 0, mig_speed, xfer_cpu_overhead);
 875
 876     double computed = lookup_computed_flop_counts(vm, 1, 0);
 877     computed_total += computed;
 878
 879     {
 880       double updated_size = get_updated_size(computed, dp_rate, dp_cap);
 881
 882       double overhead = dpt_cpu_overhead * updated_size;
 883       launch_deferred_exec_process(vm, overhead, 10000);
 884     }
 885   }
 886
 887   return computed_total;
 888 }
 889
 890
 891
 892
 893 static int migration_tx_fun(int argc, char *argv[])
 894 {
 895   XBT_DEBUG("mig: tx_start");
 896
 897   xbt_assert(argc == 4);
 898   const char *vm_name = argv[1];
 899   const char *src_pm_name  = argv[2];
 900   const char *dst_pm_name  = argv[3];
 901   msg_vm_t vm = MSG_get_host_by_name(vm_name);
 902
 903
 904   s_ws_params_t params;
 905   simcall_host_get_params(vm, &params);
 906   const long ramsize        = params.ramsize;
 907   const long devsize        = params.devsize;
 908   const int skip_stage1     = params.skip_stage1;
 909   const int skip_stage2     = params.skip_stage2;
 910   const double dp_rate      = params.dp_rate;
 911   const double dp_cap       = params.dp_cap;
 912   const double mig_speed    = params.mig_speed;
 913   const double xfer_cpu_overhead = params.xfer_cpu_overhead;
 914   const double dpt_cpu_overhead = params.dpt_cpu_overhead;
 915
 916   double remaining_size = ramsize + devsize;
 917
 918   double max_downtime = params.max_downtime;
 919   if (max_downtime == 0) {
 920     XBT_WARN("use the default max_downtime value 30ms");
 921     max_downtime = 0.03;
 922   }
 923
 924   /* This value assumes the network link is 1Gbps. */
 925   double threshold = max_downtime * 125 * 1024 * 1024;
 926
 927   /* setting up parameters has done */
 928
 929
 930   if (ramsize == 0)
 931     XBT_WARN("migrate a VM, but ramsize is zero");
 932
 933   char *mbox = get_mig_mbox_src_dst(vm_name, src_pm_name, dst_pm_name);
 934
 935   XBT_INFO("mig-stage1: remaining_size %f", remaining_size);
 936
 937   /* Stage1: send all memory pages to the destination. */
 938   start_dirty_page_tracking(vm);
 939
 940   double computed_during_stage1 = 0;
 941   if (!skip_stage1) {
 942     // send_migration_data(vm_name, src_pm_name, dst_pm_name, ramsize, mbox, 1, 0, mig_speed, xfer_cpu_overhead);
 943
 944     /* send ramsize, but split it */
 945     computed_during_stage1 = send_stage1(vm, src_pm_name, dst_pm_name, ramsize, mig_speed, xfer_cpu_overhead, dp_rate, dp_cap, dpt_cpu_overhead);
 946     remaining_size -= ramsize;
 947   }
 948
 949
 950   /* Stage2: send update pages iteratively until the size of remaining states
 951    * becomes smaller than the threshold value. */
 952   if (skip_stage2)
 953     goto stage3;
 954   if (max_downtime == 0) {
 955     XBT_WARN("no max_downtime parameter, skip stage2");
 956     goto stage3;
 957   }
 958
 959
 960   int stage2_round = 0;
 961   for (;;) {
 962
 963     double updated_size = 0;
 964     if (stage2_round == 0)  {
 965       /* just after stage1, nothing has been updated. But, we have to send the data updated during stage1 */
 966       updated_size = get_updated_size(computed_during_stage1, dp_rate, dp_cap);
 967     } else {
 968       double computed = lookup_computed_flop_counts(vm, 2, stage2_round);
 969       updated_size = get_updated_size(computed, dp_rate, dp_cap);
 970     }
 971
 972     XBT_INFO("%d updated_size %f computed_during_stage1 %f dp_rate %f dp_cap %f",
 973         stage2_round, updated_size, computed_during_stage1, dp_rate, dp_cap);
 974
 975
 976     if (stage2_round != 0) {
 977       /* during stage1, we have already created overhead tasks */
 978       double overhead = dpt_cpu_overhead * updated_size;
 979       XBT_INFO("updated %f overhead %f", updated_size, overhead);
 980       launch_deferred_exec_process(vm, overhead, 10000);
 981     }
 982
 983
 984     {
 985       remaining_size += updated_size;
 986
 987       XBT_INFO("mig-stage2.%d: remaining_size %f (%s threshold %f)", stage2_round,
 988           remaining_size, (remaining_size < threshold) ? "<" : ">", threshold);
 989
 990       if (remaining_size < threshold)
 991         break;
 992     }
 993
 994
 995     send_migration_data(vm_name, src_pm_name, dst_pm_name, updated_size, mbox, 2, stage2_round, mig_speed, xfer_cpu_overhead);
 996
 997     remaining_size -= updated_size;
 998     stage2_round += 1;
 999   }
1000
1001
1002 stage3:
1003   /* Stage3: stop the VM and copy the rest of states. */
1004   XBT_INFO("mig-stage3: remaining_size %f", remaining_size);
1005   simcall_vm_suspend(vm);
1006   stop_dirty_page_tracking(vm);
1007
1008   send_migration_data(vm_name, src_pm_name, dst_pm_name, remaining_size, mbox, 3, 0, mig_speed, xfer_cpu_overhead);
1009
1010   xbt_free(mbox);
1011
1012   XBT_DEBUG("mig: tx_done");
1013
1014   return 0;
1015 }
1016
1017
1018
1019 static void do_migration(msg_vm_t vm, msg_host_t src_pm, msg_host_t dst_pm)
1020 {
1021   char *mbox_ctl = get_mig_mbox_ctl(sg_host_name(vm), sg_host_name(src_pm), sg_host_name(dst_pm));
1022
1023   {
1024     char *pr_name = get_mig_process_rx_name(sg_host_name(vm), sg_host_name(src_pm), sg_host_name(dst_pm));
1025     int nargvs = 5;
1026     char **argv = xbt_new(char *, nargvs);
1027     argv[0] = xbt_strdup(pr_name);
1028     argv[1] = xbt_strdup(sg_host_name(vm));
1029     argv[2] = xbt_strdup(sg_host_name(src_pm));
1030     argv[3] = xbt_strdup(sg_host_name(dst_pm));
1031     argv[4] = NULL;
1032
1033     MSG_process_create_with_arguments(pr_name, migration_rx_fun, NULL, dst_pm, nargvs - 1, argv);
1034
1035     xbt_free(pr_name);
1036   }
1037
1038   {
1039     char *pr_name = get_mig_process_tx_name(sg_host_name(vm), sg_host_name(src_pm), sg_host_name(dst_pm));
1040     int nargvs = 5;
1041     char **argv = xbt_new(char *, nargvs);
1042     argv[0] = xbt_strdup(pr_name);
1043     argv[1] = xbt_strdup(sg_host_name(vm));
1044     argv[2] = xbt_strdup(sg_host_name(src_pm));
1045     argv[3] = xbt_strdup(sg_host_name(dst_pm));
1046     argv[4] = NULL;
1047     MSG_process_create_with_arguments(pr_name, migration_tx_fun, NULL, src_pm, nargvs - 1, argv);
1048
1049     xbt_free(pr_name);
1050   }
1051
1052   /* wait until the migration have finished */
1053   {
1054     msg_task_t task = NULL;
1055     msg_error_t ret = MSG_task_recv(&task, mbox_ctl);
1056     xbt_assert(ret == MSG_OK);
1057
1058     char *expected_task_name = get_mig_task_name(sg_host_name(vm), sg_host_name(src_pm), sg_host_name(dst_pm), 4);
1059     xbt_assert(strcmp(task->name, expected_task_name) == 0);
1060     xbt_free(expected_task_name);
1061   }
1062
1063   xbt_free(mbox_ctl);
1064 }
1065
1066
1067 /** @brief Migrate the VM to the given host.
1068  *  @ingroup msg_VMs
1069  *
1070  * FIXME: No migration cost occurs. If you want to simulate this too, you want to use a
1071  * MSG_task_send() before or after, depending on whether you want to do cold or hot
1072  * migration.
1073  */
1074 void MSG_vm_migrate(msg_vm_t vm, msg_host_t new_pm)
1075 {
1076   /* some thoughts:
1077    * - One approach is ...
1078    *   We first create a new VM (i.e., destination VM) on the destination
1079    *   physical host. The destination VM will receive the state of the source
1080    *   VM over network. We will finally destroy the source VM.
1081    *   - This behavior is similar to the way of migration in the real world.
1082    *     Even before a migration is completed, we will see a destination VM,
1083    *     consuming resources.
1084    *   - We have to relocate all processes. The existing process migraion code
1085    *     will work for this?
1086    *   - The name of the VM is a somewhat unique ID in the code. It is tricky
1087    *     for the destination VM?
1088    *
1089    * - Another one is ...
1090    *   We update the information of the given VM to place it to the destination
1091    *   physical host.
1092    *
1093    * The second one would be easier.
1094    *
1095    */
1096
1097   msg_host_t old_pm = simcall_vm_get_pm(vm);
1098
1099   if (simcall_vm_get_state(vm) != SURF_VM_STATE_RUNNING)
1100     THROWF(vm_error, 0, "VM(%s) is not running", sg_host_name(vm));
1101
1102   do_migration(vm, old_pm, new_pm);
1103
1104
1105
1106   XBT_DEBUG("VM(%s) moved from PM(%s) to PM(%s)", vm->key, old_pm->key, new_pm->key);
1107
1108   #ifdef HAVE_TRACING
1109   TRACE_msg_vm_change_host(vm, old_pm, new_pm);
1110   #endif
1111 }
1112
1113
1114 /** @brief Immediately suspend the execution of all processes within the given VM.
1115  *  @ingroup msg_VMs
1116  *
1117  * This function stops the exection of the VM. All the processes on this VM
1118  * will pause. The state of the VM is perserved. We can later resume it again.
1119  *
1120  * No suspension cost occurs.
1121  */
1122 void MSG_vm_suspend(msg_vm_t vm)
1123 {
1124   simcall_vm_suspend(vm);
1125
1126   XBT_DEBUG("vm_suspend done");
1127
1128   #ifdef HAVE_TRACING
1129   TRACE_msg_vm_suspend(vm);
1130   #endif
1131 }
1132
1133
1134 /** @brief Resume the execution of the VM. All processes on the VM run again.
1135  *  @ingroup msg_VMs
1136  *
1137  * No resume cost occurs.
1138  */
1139 void MSG_vm_resume(msg_vm_t vm)
1140 {
1141   simcall_vm_resume(vm);
1142
1143   #ifdef HAVE_TRACING
1144   TRACE_msg_vm_resume(vm);
1145   #endif
1146 }
1147
1148
1149 /** @brief Immediately save the execution of all processes within the given VM.
1150  *  @ingroup msg_VMs
1151  *
1152  * This function stops the exection of the VM. All the processes on this VM
1153  * will pause. The state of the VM is perserved. We can later resume it again.
1154  *
1155  * FIXME: No suspension cost occurs. If you want to simulate this too, you want to
1156  * use a \ref MSG_file_write() before or after, depending on the exact semantic
1157  * of VM save to you.
1158  */
1159 void MSG_vm_save(msg_vm_t vm)
1160 {
1161   simcall_vm_save(vm);
1162   #ifdef HAVE_TRACING
1163   TRACE_msg_vm_save(vm);
1164   #endif
1165 }
1166
1167 /** @brief Restore the execution of the VM. All processes on the VM run again.
1168  *  @ingroup msg_VMs
1169  *
1170  * FIXME: No restore cost occurs. If you want to simulate this too, you want to
1171  * use a \ref MSG_file_read() before or after, depending on the exact semantic
1172  * of VM restore to you.
1173  */
1174 void MSG_vm_restore(msg_vm_t vm)
1175 {
1176   simcall_vm_restore(vm);
1177
1178   #ifdef HAVE_TRACING
1179   TRACE_msg_vm_restore(vm);
1180   #endif
1181 }
1182
1183
1184 /** @brief Get the physical host of a given VM.
1185  *  @ingroup msg_VMs
1186  */
1187 msg_host_t MSG_vm_get_pm(msg_vm_t vm)
1188 {
1189   return simcall_vm_get_pm(vm);
1190 }
1191
1192
1193 /** @brief Set a CPU bound for a given VM.
1194  *  @ingroup msg_VMs
1195  *
1196  * 1.
1197  * Note that in some cases MSG_task_set_bound() may not intuitively work for VMs.
1198  *
1199  * For example,
1200  *  On PM0, there are Task1 and VM0.
1201  *  On VM0, there is Task2.
1202  * Now we bound 75% to Task1@PM0 and bound 25% to Task2@VM0.
1203  * Then,
1204  *  Task1@PM0 gets 50%.
1205  *  Task2@VM0 gets 25%.
1206  * This is NOT 75% for Task1@PM0 and 25% for Task2@VM0, respectively.
1207  *
1208  * This is because a VM has the dummy CPU action in the PM layer. Putting a
1209  * task on the VM does not affect the bound of the dummy CPU action. The bound
1210  * of the dummy CPU action is unlimited.
1211  *
1212  * There are some solutions for this problem. One option is to update the bound
1213  * of the dummy CPU action automatically. It should be the sum of all tasks on
1214  * the VM. But, this solution might be costy, because we have to scan all tasks
1215  * on the VM in share_resource() or we have to trap both the start and end of
1216  * task execution.
1217  *
1218  * The current solution is to use MSG_vm_set_bound(), which allows us to
1219  * directly set the bound of the dummy CPU action.
1220  *
1221  *
1222  * 2.
1223  * Note that bound == 0 means no bound (i.e., unlimited).
1224  */
1225 void MSG_vm_set_bound(msg_vm_t vm, double bound)
1226 {
1227         return simcall_vm_set_bound(vm, bound);
1228 }