Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
support timeout of migration
[simgrid.git] / src / msg / msg_vm.c
index 25ec03c..7909bb5 100644 (file)
@@ -607,9 +607,10 @@ void MSG_host_del_task(msg_host_t host, msg_task_t task)
 
 
 
-static void send_migration_data(msg_vm_t vm, msg_host_t src_pm, msg_host_t dst_pm,
-    sg_size_t size, char *mbox, int stage, int stage2_round, double mig_speed)
+static sg_size_t send_migration_data(msg_vm_t vm, msg_host_t src_pm, msg_host_t dst_pm,
+    sg_size_t size, char *mbox, int stage, int stage2_round, double mig_speed, double timeout)
 {
+  sg_size_t sent = 0;
   char *task_name = get_mig_task_name(vm, src_pm, dst_pm, stage);
   msg_task_t task = MSG_task_create(task_name, 0, size, NULL);
 
@@ -619,37 +620,45 @@ static void send_migration_data(msg_vm_t vm, msg_host_t src_pm, msg_host_t dst_p
 
   msg_error_t ret;
   if (mig_speed > 0)
-    ret = MSG_task_send_bounded(task, mbox, mig_speed);
+    ret = MSG_task_send_with_timeout_bounded(task, mbox, timeout, mig_speed);
   else
     ret = MSG_task_send(task, mbox);
 
   xbt_free(task_name);
+
+  if (ret == MSG_OK) {
+    sent = size;
+  } else if (ret == MSG_TIMEOUT) {
+    sg_size_t remaining = MSG_task_get_remaining_communication(task);
+    sent = size - remaining;
+    XBT_INFO("timeout (%lf s) in sending_migration_data, remaining %llu bytes of %llu",
+        timeout, remaining, size);
+  }
+
+  /* FIXME: why try-and-catch is used here? */
   if(ret == MSG_HOST_FAILURE){
-       //XBT_INFO("SRC host failed during migration of %s (stage %d)", sg_host_name(vm), stage);
-       MSG_task_destroy(task);
-       THROWF(host_error, 0, "SRC host failed during migration of %s (stage %d)", sg_host_name(vm), stage);
-   }else if(ret == MSG_TRANSFER_FAILURE){
-       //XBT_INFO("DST host failed during migration of %s (stage %d)", sg_host_name(vm), stage);
-       MSG_task_destroy(task);
-       THROWF(host_error, 0, "DST host failed during migration of %s (stage %d)", sg_host_name(vm), stage);
+    //XBT_INFO("SRC host failed during migration of %s (stage %d)", sg_host_name(vm), stage);
+    MSG_task_destroy(task);
+    THROWF(host_error, 0, "SRC host failed during migration of %s (stage %d)", sg_host_name(vm), stage);
+  }else if(ret == MSG_TRANSFER_FAILURE){
+    //XBT_INFO("DST host failed during migration of %s (stage %d)", sg_host_name(vm), stage);
+    MSG_task_destroy(task);
+    THROWF(host_error, 0, "DST host failed during migration of %s (stage %d)", sg_host_name(vm), stage);
   }
 
   double clock_end = MSG_get_clock();
   double duration = clock_end - clock_sta;
   double actual_speed = size / duration;
-  double cpu_utilization = 0;
-
-  if (stage == 2){
-    XBT_DEBUG("mig-stage%d.%d: sent %llu duration %f actual_speed %f (target %f) cpu %f", stage, stage2_round, size, duration, actual_speed, mig_speed, cpu_utilization);}
-  else{
-    XBT_DEBUG("mig-stage%d: sent %llu duration %f actual_speed %f (target %f) cpu %f", stage, size, duration, actual_speed, mig_speed, cpu_utilization);
-  }
-
 
+  if (stage == 2)
+    XBT_DEBUG("mig-stage%d.%d: sent %llu duration %f actual_speed %f (target %f)", stage, stage2_round, size, duration, actual_speed, mig_speed);
+  else
+    XBT_DEBUG("mig-stage%d: sent %llu duration %f actual_speed %f (target %f)", stage, size, duration, actual_speed, mig_speed);
 
+  return sent;
 }
 
-static double get_updated_size(double computed, double dp_rate, double dp_cap)
+static sg_size_t get_updated_size(double computed, double dp_rate, double dp_cap)
 {
   double updated_size = computed * dp_rate;
   XBT_DEBUG("updated_size %f dp_rate %f", updated_size, dp_rate);
@@ -658,7 +667,7 @@ static double get_updated_size(double computed, double dp_rate, double dp_cap)
     updated_size = dp_cap;
   }
 
-  return updated_size;
+  return (sg_size_t) updated_size;
 }
 
 static double send_stage1(struct migration_session *ms,
@@ -676,16 +685,9 @@ static double send_stage1(struct migration_session *ms,
       datasize = remaining;
 
     remaining -= datasize;
-    send_migration_data(ms->vm, ms->src_pm, ms->dst_pm, datasize, ms->mbox, 1, 0, mig_speed);
+    send_migration_data(ms->vm, ms->src_pm, ms->dst_pm, datasize, ms->mbox, 1, 0, mig_speed, -1);
     double computed = lookup_computed_flop_counts(ms->vm, 1, 0);
     computed_total += computed;
-
-    // {
-    //   double updated_size = get_updated_size(computed, dp_rate, dp_cap);
-
-    //   double overhead = dpt_cpu_overhead * updated_size;
-    //   launch_deferred_exec_process(vm, overhead, 10000);
-    // }
   }
 
   return computed_total;
@@ -695,11 +697,7 @@ static double send_stage1(struct migration_session *ms,
 
 static double get_threshold_value(double bandwidth, double max_downtime)
 {
-  /* This value assumes the network link is 1Gbps. */
-  // double threshold = max_downtime * 125 * 1024 * 1024;
-  double threshold = max_downtime * bandwidth;
-
-  return threshold;
+  return max_downtime * bandwidth;
 }
 
 static int migration_tx_fun(int argc, char *argv[])
@@ -714,53 +712,66 @@ static int migration_tx_fun(int argc, char *argv[])
   const sg_size_t ramsize   = params.ramsize;
   const sg_size_t devsize   = params.devsize;
   const int skip_stage1     = params.skip_stage1;
-  const int skip_stage2     = params.skip_stage2;
+  int skip_stage2           = params.skip_stage2;
   const double dp_rate      = params.dp_rate;
   const double dp_cap       = params.dp_cap;
   const double mig_speed    = params.mig_speed;
+  double max_downtime       = params.max_downtime;
 
-  msg_vm_t vm=ms->vm; 
+  /* hard code it temporally. Fix Me */
+#define MIGRATION_TIMEOUT_DO_NOT_HARDCODE_ME 10000000.0
+  double mig_timeout = MIGRATION_TIMEOUT_DO_NOT_HARDCODE_ME;
 
   double remaining_size = ramsize + devsize;
+  double threshold;
+
+  /* check parameters */
+  if (ramsize == 0)
+    XBT_WARN("migrate a VM, but ramsize is zero");
 
-  double max_downtime = params.max_downtime;
   if (max_downtime == 0) {
     XBT_WARN("use the default max_downtime value 30ms");
     max_downtime = 0.03;
   }
 
-  double threshold = 0.00001; /* TODO: cleanup */
-
-  /* setting up parameters has done */
-
-
-  if (ramsize == 0)
-    XBT_WARN("migrate a VM, but ramsize is zero");
-
-
-  XBT_DEBUG("mig-stage1: remaining_size %f", remaining_size);
-
   /* Stage1: send all memory pages to the destination. */
-  start_dirty_page_tracking(vm);
+  XBT_DEBUG("mig-stage1: remaining_size %f", remaining_size);
+  start_dirty_page_tracking(ms->vm);
 
   double computed_during_stage1 = 0;
   if (!skip_stage1) {
-    // send_migration_data(vm_name, src_pm_name, dst_pm_name, ramsize, mbox, 1, 0, mig_speed, xfer_cpu_overhead);
-
-    /* send ramsize, but split it */
     double clock_prev_send = MSG_get_clock();
 
     TRY {
-      computed_during_stage1 = send_stage1(ms, ramsize, mig_speed, dp_rate, dp_cap);
-    } CATCH_ANONYMOUS{
+      /* At stage 1, we do not need timeout. We have to send all the memory
+       * pages even though the duration of this tranfer exceeds the timeout
+       * value. */
+      sg_size_t sent = send_migration_data(ms->vm, ms->src_pm, ms->dst_pm, ramsize, ms->mbox, 1, 0, mig_speed, -1);
+      remaining_size -= sent;
+      computed_during_stage1 = lookup_computed_flop_counts(ms->vm, 1, 0);
+
+      if (sent < ramsize) {
+        XBT_INFO("mig-stage1: timeout, force moving to stage 3");
+        skip_stage2 = 1;
+      } else if (sent > ramsize)
+        XBT_CRITICAL("bug");
+
+    } CATCH_ANONYMOUS {
       //hostfailure (if you want to know whether this is the SRC or the DST please check directly in send_migration_data code)
       // Stop the dirty page tracking an return (there is no memory space to release)
-      stop_dirty_page_tracking(vm);
+      stop_dirty_page_tracking(ms->vm);
       return 0;
     }
-    remaining_size -= ramsize;
 
     double clock_post_send = MSG_get_clock();
+    mig_timeout -= (clock_post_send - clock_prev_send);
+    if (mig_timeout < 0) {
+      XBT_INFO("The duration of stage 1 exceeds the timeout value (%lf > %lf), skip stage 2",
+          (clock_post_send - clock_prev_send), MIGRATION_TIMEOUT_DO_NOT_HARDCODE_ME);
+      skip_stage2 = 1;
+    }
+
+    /* estimate bandwidth */
     double bandwidth = ramsize / (clock_post_send - clock_prev_send);
     threshold = get_threshold_value(bandwidth, max_downtime);
     XBT_DEBUG("actual bandwidth %f (MB/s), threshold %f", bandwidth / 1024 / 1024, threshold);
@@ -771,83 +782,89 @@ static int migration_tx_fun(int argc, char *argv[])
    * becomes smaller than the threshold value. */
   if (skip_stage2)
     goto stage3;
-  if (max_downtime == 0) {
-    XBT_WARN("no max_downtime parameter, skip stage2");
-    goto stage3;
-  }
 
 
   int stage2_round = 0;
   for (;;) {
 
-    double updated_size = 0;
-    if (stage2_round == 0)  {
-      /* just after stage1, nothing has been updated. But, we have to send the data updated during stage1 */
+    sg_size_t updated_size = 0;
+    if (stage2_round == 0) {
+      /* just after stage1, nothing has been updated. But, we have to send the
+       * data updated during stage1 */
       updated_size = get_updated_size(computed_during_stage1, dp_rate, dp_cap);
     } else {
       double computed = lookup_computed_flop_counts(ms->vm, 2, stage2_round);
       updated_size = get_updated_size(computed, dp_rate, dp_cap);
     }
 
-    XBT_DEBUG("mig-stage 2:%d updated_size %f computed_during_stage1 %f dp_rate %f dp_cap %f",
+    XBT_DEBUG("mig-stage 2:%d updated_size %llu computed_during_stage1 %f dp_rate %f dp_cap %f",
         stage2_round, updated_size, computed_during_stage1, dp_rate, dp_cap);
 
 
-    // if (stage2_round != 0) {
-    //   /* during stage1, we have already created overhead tasks */
-    //   double overhead = dpt_cpu_overhead * updated_size;
-    //   XBT_DEBUG("updated %f overhead %f", updated_size, overhead);
-    //   launch_deferred_exec_process(vm, overhead, 10000);
-    // }
-
-
-    {
-      remaining_size += updated_size;
-
-      XBT_DEBUG("mig-stage2.%d: remaining_size %f (%s threshold %f)", stage2_round,
-          remaining_size, (remaining_size < threshold) ? "<" : ">", threshold);
+    /* Check whether the remaining size is below the threshold value. If so,
+     * move to stage 3. */
+    remaining_size += updated_size;
+    XBT_DEBUG("mig-stage2.%d: remaining_size %f (%s threshold %f)", stage2_round,
+        remaining_size, (remaining_size < threshold) ? "<" : ">", threshold);
+    if (remaining_size < threshold)
+      break;
 
-      if (remaining_size < threshold)
-        break;
-    }
 
+    sg_size_t sent;
     double clock_prev_send = MSG_get_clock();
-    TRY{
-      send_migration_data(ms->vm, ms->src_pm, ms->dst_pm, updated_size, ms->mbox, 2, stage2_round, mig_speed);
-    }CATCH_ANONYMOUS{
+    TRY {
+      sent = send_migration_data(ms->vm, ms->src_pm, ms->dst_pm, updated_size, ms->mbox, 2, stage2_round, mig_speed, mig_timeout);
+    } CATCH_ANONYMOUS {
       //hostfailure (if you want to know whether this is the SRC or the DST please check directly in send_migration_data code)
       // Stop the dirty page tracking an return (there is no memory space to release)
-      stop_dirty_page_tracking(vm);
+      stop_dirty_page_tracking(ms->vm);
       return 0;
     }
     double clock_post_send = MSG_get_clock();
 
-    double bandwidth = updated_size / (clock_post_send - clock_prev_send);
-    threshold = get_threshold_value(bandwidth, max_downtime);
-    XBT_DEBUG("actual bandwidth %f, threshold %f", bandwidth / 1024 / 1024, threshold);
+    if (sent == updated_size) {
+      /* timeout did not happen */
+      double bandwidth = updated_size / (clock_post_send - clock_prev_send);
+      threshold = get_threshold_value(bandwidth, max_downtime);
+      XBT_DEBUG("actual bandwidth %f, threshold %f", bandwidth / 1024 / 1024, threshold);
+      remaining_size -= sent;
+      stage2_round += 1;
+      mig_timeout -= (clock_post_send - clock_prev_send);
+      xbt_assert(mig_timeout > 0);
+
+    } else if (sent < updated_size) {
+      /* When timeout happens, we move to stage 3. The size of memory pages
+       * updated before timeout must be added to the remaining size. */
+      XBT_INFO("mig-stage2.%d: timeout, force moving to stage 3. sent %llu / %llu, eta %lf",
+          stage2_round, sent, updated_size, (clock_post_send - clock_prev_send));
+      remaining_size -= sent;
 
+      double computed = lookup_computed_flop_counts(ms->vm, 2, stage2_round);
+      updated_size = get_updated_size(computed, dp_rate, dp_cap);
+      remaining_size += updated_size;
+      break;
 
-    remaining_size -= updated_size;
-    stage2_round += 1;
+    } else
+      XBT_CRITICAL("bug");
   }
 
 
 stage3:
   /* Stage3: stop the VM and copy the rest of states. */
   XBT_DEBUG("mig-stage3: remaining_size %f", remaining_size);
-  simcall_vm_suspend(vm);
-  stop_dirty_page_tracking(vm);
+  simcall_vm_suspend(ms->vm);
+  stop_dirty_page_tracking(ms->vm);
+
+  TRY {
+    send_migration_data(ms->vm, ms->src_pm, ms->dst_pm, remaining_size, ms->mbox, 3, 0, mig_speed, -1);
+  } CATCH_ANONYMOUS {
+    //hostfailure (if you want to know whether this is the SRC or the DST please check directly in send_migration_data code)
+    // Stop the dirty page tracking an return (there is no memory space to release)
+    simcall_vm_resume(ms->vm);
+    return 0;
+  }
 
- TRY{
-    send_migration_data(ms->vm, ms->src_pm, ms->dst_pm, remaining_size, ms->mbox, 3, 0, mig_speed);
-  }CATCH_ANONYMOUS{
-      //hostfailure (if you want to know whether this is the SRC or the DST please check directly in send_migration_data code)
-      // Stop the dirty page tracking an return (there is no memory space to release)
-      simcall_vm_resume(vm);
-      return 0;
-    }
- // At that point the Migration is considered valid for the SRC node but remind that the DST side should relocate effectively the VM on the DST node.
+  // At that point the Migration is considered valid for the SRC node but remind that the DST side should relocate effectively the VM on the DST node.
 
   XBT_DEBUG("mig: tx_done");