src/plugins/vm/VmLiveMigration.cpp

   1 /* Copyright (c) 2013-2017. 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 #include <simgrid/plugins/live_migration.h>
   7 #include <simgrid/s4u.hpp>
   8 #include <simgrid/s4u/VirtualMachine.hpp>
   9 #include <simgrid/vm.h>
  10 #include <src/instr/instr_private.hpp>
  11 #include <src/plugins/vm/VirtualMachineImpl.hpp>
  12 #include <src/plugins/vm/VmHostExt.hpp>
  13 #include <src/plugins/vm/VmLiveMigration.hpp>
  14 #include <xbt/ex.hpp>
  15
  16 XBT_LOG_NEW_DEFAULT_CATEGORY(vm_live_migration, "S4U virtual machines live migration");
  17
  18 namespace simgrid {
  19 namespace vm {
  20 simgrid::xbt::Extension<s4u::Host, VmMigrationExt> VmMigrationExt::EXTENSION_ID;
  21
  22 void VmMigrationExt::ensureVmMigrationExtInstalled()
  23 {
  24   if (not EXTENSION_ID.valid())
  25     EXTENSION_ID = simgrid::s4u::Host::extension_create<VmMigrationExt>();
  26 }
  27
  28 void MigrationRx::operator()()
  29 {
  30   XBT_DEBUG("mig: rx_start");
  31   bool received_finalize = false;
  32
  33   std::string finalize_task_name =
  34       std::string("__mig_stage3:") + vm_->getCname() + "(" + src_pm_->getCname() + "-" + dst_pm_->getCname() + ")";
  35
  36   while (not received_finalize) {
  37     std::string* payload = static_cast<std::string*>(mbox->get());
  38
  39     if (finalize_task_name == *payload)
  40       received_finalize = true;
  41
  42     delete payload;
  43   }
  44
  45   // Here Stage 1, 2  and 3 have been performed.
  46   // Hence complete the migration
  47
  48   /* Update the vm location */
  49   /* precopy migration makes the VM temporally paused */
  50   xbt_assert(vm_->getState() == SURF_VM_STATE_SUSPENDED);
  51
  52   /* Update the vm location and resume it */
  53   vm_->setPm(dst_pm_);
  54   vm_->resume();
  55
  56   // Now the VM is running on the new host (the migration is completed) (even if the SRC crash)
  57   vm_->getImpl()->isMigrating = false;
  58   XBT_DEBUG("VM(%s) moved from PM(%s) to PM(%s)", vm_->getCname(), src_pm_->getCname(), dst_pm_->getCname());
  59
  60   if (TRACE_vm_is_enabled()) {
  61     static long long int counter = 0;
  62     std::string key              = std::to_string(counter);
  63     counter++;
  64
  65     // start link
  66     container_t msg = simgrid::instr::Container::byName(vm_->getName());
  67     simgrid::instr::Container::getRoot()->getLink("MSG_VM_LINK")->startEvent(msg, "M", key);
  68
  69     // destroy existing container of this vm
  70     simgrid::instr::Container::byName(vm_->getName())->removeFromParent();
  71
  72     // create new container on the new_host location
  73     new simgrid::instr::Container(vm_->getCname(), "MSG_VM", simgrid::instr::Container::byName(dst_pm_->getName()));
  74
  75     // end link
  76     msg = simgrid::instr::Container::byName(vm_->getName());
  77     simgrid::instr::Container::getRoot()->getLink("MSG_VM_LINK")->endEvent(msg, "M", key);
  78   }
  79   // Inform the SRC that the migration has been correctly performed
  80   std::string* payload = new std::string("__mig_stage4:");
  81   *payload             = *payload + vm_->getCname() + "(" + src_pm_->getCname() + "-" + dst_pm_->getCname() + ")";
  82
  83   mbox_ctl->put(payload, 0);
  84
  85   XBT_DEBUG("mig: rx_done");
  86 }
  87
  88 static sg_size_t get_updated_size(double computed, double dp_rate, sg_size_t dp_cap)
  89 {
  90   sg_size_t updated_size = static_cast<sg_size_t>(computed * dp_rate);
  91   XBT_DEBUG("updated_size %llu dp_rate %f", updated_size, dp_rate);
  92   if (updated_size > dp_cap) {
  93     updated_size = dp_cap;
  94   }
  95
  96   return updated_size;
  97 }
  98
  99 sg_size_t MigrationTx::sendMigrationData(sg_size_t size, int stage, int stage2_round, double mig_speed, double timeout)
 100 {
 101   sg_size_t sent   = size;
 102   std::string* msg = new std::string("__mig_stage");
 103   *msg = *msg + std::to_string(stage) + ":" + vm_->getCname() + "(" + src_pm_->getCname() + "-" + dst_pm_->getCname() +
 104          ")";
 105
 106   double clock_sta = s4u::Engine::getClock();
 107
 108   s4u::Activity* comm = nullptr;
 109   try {
 110     if (mig_speed > 0)
 111       comm = mbox->put_init(msg, size)->setRate(mig_speed)->wait(timeout);
 112     else
 113       comm = mbox->put_async(msg, size)->wait();
 114   } catch (xbt_ex& e) {
 115     if (comm) {
 116       sg_size_t remaining = static_cast<sg_size_t>(comm->getRemains());
 117       XBT_VERB("timeout (%lf s) in sending_migration_data, remaining %llu bytes of %llu", timeout, remaining, size);
 118       sent -= remaining;
 119     }
 120     delete msg;
 121   }
 122
 123   double clock_end    = s4u::Engine::getClock();
 124   double duration     = clock_end - clock_sta;
 125   double actual_speed = size / duration;
 126
 127   if (stage == 2)
 128     XBT_DEBUG("mig-stage%d.%d: sent %llu duration %f actual_speed %f (target %f)", stage, stage2_round, size, duration,
 129               actual_speed, mig_speed);
 130   else
 131     XBT_DEBUG("mig-stage%d: sent %llu duration %f actual_speed %f (target %f)", stage, size, duration, actual_speed,
 132               mig_speed);
 133
 134   return sent;
 135 }
 136
 137 void MigrationTx::operator()()
 138 {
 139   XBT_DEBUG("mig: tx_start");
 140
 141   double host_speed = vm_->getPm()->getSpeed();
 142   const sg_size_t ramsize = vm_->getRamsize();
 143   const double dp_rate =
 144       host_speed ? (sg_vm_get_migration_speed(vm_) * sg_vm_get_dirty_page_intensity(vm_)) / host_speed : 1;
 145   const sg_size_t dp_cap = sg_vm_get_working_set_memory(vm_);
 146   const double mig_speed = sg_vm_get_migration_speed(vm_);
 147   double max_downtime    = sg_vm_get_max_downtime(vm_);
 148
 149   double mig_timeout = 10000000.0;
 150   bool skip_stage2   = false;
 151
 152   size_t remaining_size = ramsize;
 153   size_t threshold      = 0.0;
 154
 155   /* check parameters */
 156   if (ramsize == 0)
 157     XBT_WARN("migrate a VM, but ramsize is zero");
 158
 159   if (max_downtime <= 0) {
 160     XBT_WARN("use the default max_downtime value 30ms");
 161     max_downtime = 0.03;
 162   }
 163
 164   /* Stage1: send all memory pages to the destination. */
 165   XBT_DEBUG("mig-stage1: remaining_size %zu", remaining_size);
 166   sg_vm_start_dirty_page_tracking(vm_);
 167
 168   double computed_during_stage1 = 0;
 169   double clock_prev_send        = s4u::Engine::getClock();
 170
 171   try {
 172     /* At stage 1, we do not need timeout. We have to send all the memory pages even though the duration of this
 173      * transfer exceeds the timeout value. */
 174     XBT_VERB("Stage 1: Gonna send %llu bytes", ramsize);
 175     sg_size_t sent = sendMigrationData(ramsize, 1, 0, mig_speed, -1);
 176     remaining_size -= sent;
 177     computed_during_stage1 = sg_vm_lookup_computed_flops(vm_);
 178
 179     if (sent < ramsize) {
 180       XBT_VERB("mig-stage1: timeout, force moving to stage 3");
 181       skip_stage2 = true;
 182     } else if (sent > ramsize)
 183       XBT_CRITICAL("bug");
 184
 185   } catch (xbt_ex& e) {
 186     // hostfailure (if you want to know whether this is the SRC or the DST check directly in send_migration_data code)
 187     // Stop the dirty page tracking an return (there is no memory space to release)
 188     sg_vm_stop_dirty_page_tracking(vm_);
 189     return;
 190   }
 191
 192   double clock_post_send = s4u::Engine::getClock();
 193   mig_timeout -= (clock_post_send - clock_prev_send);
 194   if (mig_timeout < 0) {
 195     XBT_VERB("The duration of stage 1 exceeds the timeout value, skip stage 2");
 196     skip_stage2 = true;
 197   }
 198
 199   /* estimate bandwidth */
 200   double bandwidth = ramsize / (clock_post_send - clock_prev_send);
 201   threshold        = bandwidth * max_downtime;
 202   XBT_DEBUG("actual bandwidth %f (MB/s), threshold %zu", bandwidth / 1024 / 1024, threshold);
 203
 204   /* Stage2: send update pages iteratively until the size of remaining states becomes smaller than threshold value. */
 205   if (not skip_stage2) {
 206
 207     int stage2_round = 0;
 208     for (;;) {
 209       sg_size_t updated_size = 0;
 210       if (stage2_round == 0) {
 211         /* just after stage1, nothing has been updated. But, we have to send the data updated during stage1 */
 212         updated_size = get_updated_size(computed_during_stage1, dp_rate, dp_cap);
 213       } else {
 214         double computed = sg_vm_lookup_computed_flops(vm_);
 215         updated_size    = get_updated_size(computed, dp_rate, dp_cap);
 216       }
 217
 218       XBT_DEBUG("mig-stage 2:%d updated_size %llu computed_during_stage1 %f dp_rate %f dp_cap %llu", stage2_round,
 219                 updated_size, computed_during_stage1, dp_rate, dp_cap);
 220
 221       /* Check whether the remaining size is below the threshold value. If so, move to stage 3. */
 222       remaining_size += updated_size;
 223       XBT_DEBUG("mig-stage2.%d: remaining_size %zu (%s threshold %zu)", stage2_round, remaining_size,
 224                 (remaining_size < threshold) ? "<" : ">", threshold);
 225       if (remaining_size < threshold)
 226         break;
 227
 228       sg_size_t sent         = 0;
 229       double clock_prev_send = s4u::Engine::getClock();
 230       try {
 231         XBT_DEBUG("Stage 2, gonna send %llu", updated_size);
 232         sent = sendMigrationData(updated_size, 2, stage2_round, mig_speed, mig_timeout);
 233       } catch (xbt_ex& e) {
 234         // hostfailure (if you want to know whether this is the SRC or the DST check directly in send_migration_data
 235         // code)
 236         // Stop the dirty page tracking an return (there is no memory space to release)
 237         sg_vm_stop_dirty_page_tracking(vm_);
 238         return;
 239       }
 240       double clock_post_send = s4u::Engine::getClock();
 241
 242       if (sent == updated_size) {
 243         /* timeout did not happen */
 244         double bandwidth = updated_size / (clock_post_send - clock_prev_send);
 245         threshold        = bandwidth * max_downtime;
 246         XBT_DEBUG("actual bandwidth %f, threshold %zu", bandwidth / 1024 / 1024, threshold);
 247         remaining_size -= sent;
 248         stage2_round += 1;
 249         mig_timeout -= (clock_post_send - clock_prev_send);
 250         xbt_assert(mig_timeout > 0);
 251
 252       } else if (sent < updated_size) {
 253         /* When timeout happens, we move to stage 3. The size of memory pages
 254          * updated before timeout must be added to the remaining size. */
 255         XBT_VERB("mig-stage2.%d: timeout, force moving to stage 3. sent %llu / %llu, eta %lf", stage2_round, sent,
 256                  updated_size, (clock_post_send - clock_prev_send));
 257         remaining_size -= sent;
 258
 259         double computed = sg_vm_lookup_computed_flops(vm_);
 260         updated_size    = get_updated_size(computed, dp_rate, dp_cap);
 261         remaining_size += updated_size;
 262         break;
 263       } else
 264         XBT_CRITICAL("bug");
 265     }
 266   }
 267
 268   /* Stage3: stop the VM and copy the rest of states. */
 269   XBT_DEBUG("mig-stage3: remaining_size %zu", remaining_size);
 270   vm_->suspend();
 271   sg_vm_stop_dirty_page_tracking(vm_);
 272
 273   try {
 274     XBT_DEBUG("Stage 3: Gonna send %zu bytes", remaining_size);
 275     sendMigrationData(remaining_size, 3, 0, mig_speed, -1);
 276   } catch (xbt_ex& e) {
 277     // hostfailure (if you want to know whether this is the SRC or the DST check directly in send_migration_data code)
 278     // Stop the dirty page tracking an return (there is no memory space to release)
 279     vm_->resume();
 280     return;
 281   }
 282
 283   // At that point the Migration is considered valid for the SRC node but remind that the DST side should relocate
 284   // effectively the VM on the DST node.
 285   XBT_DEBUG("mig: tx_done");
 286 }
 287 }
 288 }
 289
 290 SG_BEGIN_DECL()
 291
 292 static void onVirtualMachineShutdown(simgrid::s4u::VirtualMachine* vm)
 293 {
 294   if (vm->getImpl()->isMigrating) {
 295     vm->extension<simgrid::vm::VmMigrationExt>()->rx_->kill();
 296     vm->extension<simgrid::vm::VmMigrationExt>()->tx_->kill();
 297     vm->extension<simgrid::vm::VmMigrationExt>()->issuer_->kill();
 298     vm->getImpl()->isMigrating = false;
 299   }
 300 }
 301
 302 void sg_vm_live_migration_plugin_init()
 303 {
 304   sg_vm_dirty_page_tracking_init();
 305   simgrid::vm::VmMigrationExt::ensureVmMigrationExtInstalled();
 306   simgrid::s4u::VirtualMachine::onVmShutdown.connect(&onVirtualMachineShutdown);
 307 }
 308
 309 /* Deprecated. Please use MSG_vm_create_migratable() instead */
 310 msg_vm_t MSG_vm_create(msg_host_t ind_pm, const char* name, int coreAmount, int ramsize, int mig_netspeed,
 311                        int dp_intensity)
 312 {
 313   return sg_vm_create_migratable(ind_pm, name, coreAmount, ramsize, mig_netspeed, dp_intensity);
 314 }
 315
 316 simgrid::s4u::VirtualMachine* sg_vm_create_migratable(simgrid::s4u::Host* pm, const char* name, int coreAmount,
 317                                                       int ramsize, int mig_netspeed, int dp_intensity)
 318 {
 319   simgrid::vm::VmHostExt::ensureVmExtInstalled();
 320
 321   /* For the moment, intensity_rate is the percentage against the migration bandwidth */
 322
 323   sg_vm_t vm = new simgrid::s4u::VirtualMachine(name, pm, coreAmount, static_cast<sg_size_t>(ramsize) * 1024 * 1024);
 324   sg_vm_set_dirty_page_intensity(vm, dp_intensity / 100.0);
 325   sg_vm_set_working_set_memory(vm, vm->getRamsize() * 0.9); // assume working set memory is 90% of ramsize
 326   sg_vm_set_migration_speed(vm, mig_netspeed * 1024 * 1024.0);
 327
 328   XBT_DEBUG("migspeed : %f intensity mem : %d", mig_netspeed * 1024 * 1024.0, dp_intensity);
 329
 330   return vm;
 331 }
 332
 333 int sg_vm_is_migrating(simgrid::s4u::VirtualMachine* vm)
 334 {
 335   return vm->getImpl()->isMigrating;
 336 }
 337
 338 void sg_vm_migrate(simgrid::s4u::VirtualMachine* vm, simgrid::s4u::Host* dst_pm)
 339 {
 340   simgrid::s4u::Host* src_pm = vm->getPm();
 341
 342   if (src_pm->isOff())
 343     THROWF(vm_error, 0, "Cannot migrate VM '%s' from host '%s', which is offline.", vm->getCname(), src_pm->getCname());
 344   if (dst_pm->isOff())
 345     THROWF(vm_error, 0, "Cannot migrate VM '%s' to host '%s', which is offline.", vm->getCname(), dst_pm->getCname());
 346   if (vm->getState() != SURF_VM_STATE_RUNNING)
 347     THROWF(vm_error, 0, "Cannot migrate VM '%s' that is not running yet.", vm->getCname());
 348   if (vm->getImpl()->isMigrating)
 349     THROWF(vm_error, 0, "Cannot migrate VM '%s' that is already migrating.", vm->getCname());
 350
 351   vm->getImpl()->isMigrating = true;
 352
 353   std::string rx_name =
 354       std::string("__pr_mig_rx:") + vm->getCname() + "(" + src_pm->getCname() + "-" + dst_pm->getCname() + ")";
 355   std::string tx_name =
 356       std::string("__pr_mig_tx:") + vm->getCname() + "(" + src_pm->getCname() + "-" + dst_pm->getCname() + ")";
 357
 358   simgrid::s4u::ActorPtr rx =
 359       simgrid::s4u::Actor::createActor(rx_name.c_str(), dst_pm, simgrid::vm::MigrationRx(vm, dst_pm));
 360   simgrid::s4u::ActorPtr tx =
 361       simgrid::s4u::Actor::createActor(tx_name.c_str(), src_pm, simgrid::vm::MigrationTx(vm, dst_pm));
 362
 363   vm->extension_set<simgrid::vm::VmMigrationExt>(new simgrid::vm::VmMigrationExt(simgrid::s4u::Actor::self(), rx, tx));
 364
 365   /* wait until the migration have finished or on error has occurred */
 366   XBT_DEBUG("wait for reception of the final ACK (i.e. migration has been correctly performed");
 367   simgrid::s4u::MailboxPtr mbox_ctl = simgrid::s4u::Mailbox::byName(
 368       std::string("__mbox_mig_ctl:") + vm->getCname() + "(" + src_pm->getCname() + "-" + dst_pm->getCname() + ")");
 369   delete static_cast<std::string*>(mbox_ctl->get());
 370   tx->join();
 371   rx->join();
 372
 373   vm->getImpl()->isMigrating = false;
 374 }
 375
 376 SG_END_DECL()