X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/8ba65b25e98c8241e28fef5266584d47bdd4c0ed..f185f7337939bd2116d71563cedda3c33a39a54c:/src/msg/msg_vm.c diff --git a/src/msg/msg_vm.c b/src/msg/msg_vm.c index 103535ac8b..9ab63aef4e 100644 --- a/src/msg/msg_vm.c +++ b/src/msg/msg_vm.c @@ -61,9 +61,9 @@ xbt_dict_t MSG_vm_get_properties(msg_vm_t vm) * \param value what to change the property to * \param free_ctn the freeing function to use to kill the value on need */ -void MSG_vm_set_property_value(msg_vm_t vm, const char *name, void *value,void_f_pvoid_t free_ctn) { - - xbt_dict_set(MSG_host_get_properties(vm), name, value,free_ctn); +void MSG_vm_set_property_value(msg_vm_t vm, const char *name, void *value, void_f_pvoid_t free_ctn) +{ + xbt_dict_set(MSG_host_get_properties(vm), name, value, free_ctn); } /** \ingroup msg_vm_management @@ -77,7 +77,8 @@ void MSG_vm_set_property_value(msg_vm_t vm, const char *name, void *value,void_f * for each VM/PM. */ -msg_vm_t MSG_vm_get_by_name(const char *name){ +msg_vm_t MSG_vm_get_by_name(const char *name) +{ return MSG_get_host_by_name(name); } @@ -88,13 +89,15 @@ msg_vm_t MSG_vm_get_by_name(const char *name){ * This functions checks whether \a host is a valid pointer or not and return its name. */ -const char *MSG_vm_get_name(msg_vm_t vm) { +const char *MSG_vm_get_name(msg_vm_t vm) +{ return MSG_host_get_name(vm); } /* **** Check state of a VM **** */ -static inline int __MSG_vm_is_state(msg_vm_t vm, e_surf_vm_state_t state) { +static inline int __MSG_vm_is_state(msg_vm_t vm, e_surf_vm_state_t state) +{ return simcall_vm_get_state(vm) == state; } @@ -114,6 +117,7 @@ int MSG_vm_is_running(msg_vm_t vm) return __MSG_vm_is_state(vm, SURF_VM_STATE_RUNNING); } +#if 0 /** @brief Returns whether the given VM is currently migrating * @ingroup msg_VMs */ @@ -121,6 +125,7 @@ int MSG_vm_is_migrating(msg_vm_t vm) { return __MSG_vm_is_state(vm, SURF_VM_STATE_MIGRATING); } +#endif /** @brief Returns whether the given VM is currently suspended, not running. * @ingroup msg_VMs @@ -178,7 +183,7 @@ msg_vm_t MSG_vm_create(msg_host_t ind_pm, const char *name, simcall_host_set_params(vm, ¶ms); } - /* TODO: We will revisit the disk support later. */ + /* TODO: Limit net capability, take into account disk considerations. */ return vm; } @@ -204,7 +209,7 @@ msg_vm_t MSG_vm_create_core(msg_host_t ind_pm, const char *name) msg_vm_t ind_vm = NULL; void *ind_vm_workstation = NULL; - // Ask simix to create the surf vm resource + /* Ask the SIMIX layer to create the surf vm resource */ ind_vm_workstation = simcall_vm_create(name, ind_pm); ind_vm = (msg_vm_t) __MSG_host_create(ind_vm_workstation); @@ -218,7 +223,7 @@ msg_vm_t MSG_vm_create_core(msg_host_t ind_pm, const char *name) } -/** @brief Start a vm (ie. boot) +/** @brief Start a vm (i.e., boot the guest operating system) * @ingroup msg_VMs * * If the VM cannot be started, an exception is generated. @@ -226,7 +231,6 @@ msg_vm_t MSG_vm_create_core(msg_host_t ind_pm, const char *name) */ void MSG_vm_start(msg_vm_t vm) { - //Please note that vm start can raise an exception if the VM cannot be started. simcall_vm_start(vm); #ifdef HAVE_TRACING @@ -253,6 +257,405 @@ void MSG_vm_shutdown(msg_vm_t vm) } + +/* We have two mailboxes. mbox is used to transfer migration data between + * source and destiantion PMs. mbox_ctl is used to detect the completion of a + * migration. The names of these mailboxes must not conflict with others. */ +static inline char *get_mig_mbox_src_dst(const char *vm_name, const char *src_pm_name, const char *dst_pm_name) +{ + return bprintf("__mbox_mig_src_dst:%s(%s-%s)", vm_name, src_pm_name, dst_pm_name); +} + +static inline char *get_mig_mbox_ctl(const char *vm_name, const char *src_pm_name, const char *dst_pm_name) +{ + return bprintf("__mbox_mig_ctl:%s(%s-%s)", vm_name, src_pm_name, dst_pm_name); +} + +static inline char *get_mig_process_tx_name(const char *vm_name, const char *src_pm_name, const char *dst_pm_name) +{ + return bprintf("__pr_mig_tx:%s(%s-%s)", vm_name, src_pm_name, dst_pm_name); +} + +static inline char *get_mig_process_rx_name(const char *vm_name, const char *src_pm_name, const char *dst_pm_name) +{ + return bprintf("__pr_mig_rx:%s(%s-%s)", vm_name, src_pm_name, dst_pm_name); +} + +static inline char *get_mig_task_name(const char *vm_name, const char *src_pm_name, const char *dst_pm_name, int stage) +{ + return bprintf("__task_mig_stage%d:%s(%s-%s)", stage, vm_name, src_pm_name, dst_pm_name); +} + +static int migration_rx_fun(int argc, char *argv[]) +{ + const char *pr_name = MSG_process_get_name(MSG_process_self()); + const char *host_name = MSG_host_get_name(MSG_host_self()); + + XBT_DEBUG("mig: rx_start"); + + xbt_assert(argc == 4); + const char *vm_name = argv[1]; + const char *src_pm_name = argv[2]; + const char *dst_pm_name = argv[3]; + msg_vm_t vm = MSG_get_host_by_name(vm_name); + msg_vm_t dst_pm = MSG_get_host_by_name(dst_pm_name); + + int need_exit = 0; + + char *mbox = get_mig_mbox_src_dst(vm_name, src_pm_name, dst_pm_name); + char *mbox_ctl = get_mig_mbox_ctl(vm_name, src_pm_name, dst_pm_name); + char *finalize_task_name = get_mig_task_name(vm_name, src_pm_name, dst_pm_name, 3); + + for (;;) { + msg_task_t task = NULL; + MSG_task_recv(&task, mbox); + + if (strcmp(task->name, finalize_task_name) == 0) + need_exit = 1; + + MSG_task_destroy(task); + + if (need_exit) + break; + } + + + simcall_vm_migrate(vm, dst_pm); + simcall_vm_resume(vm); + + { + char *task_name = get_mig_task_name(vm_name, src_pm_name, dst_pm_name, 4); + + msg_task_t task = MSG_task_create(task_name, 0, 0, NULL); + msg_error_t ret = MSG_task_send(task, mbox_ctl); + xbt_assert(ret == MSG_OK); + + xbt_free(task_name); + } + + + xbt_free(mbox); + xbt_free(mbox_ctl); + xbt_free(finalize_task_name); + + XBT_DEBUG("mig: rx_done"); + + return 0; +} + + +typedef struct dirty_page { + double prev_clock; + double prev_remaining; + msg_task_t task; +} s_dirty_page, *dirty_page_t; + + +static void reset_dirty_pages(msg_vm_t vm) +{ + msg_host_priv_t priv = msg_host_resource_priv(vm); + + char *key = NULL; + xbt_dict_cursor_t cursor = NULL; + dirty_page_t dp = NULL; + xbt_dict_foreach(priv->dp_objs, cursor, key, dp) { + double remaining = MSG_task_get_remaining_computation(dp->task); + dp->prev_clock = MSG_get_clock(); + dp->prev_remaining = remaining; + + // XBT_INFO("%s@%s remaining %f", key, sg_host_name(vm), remaining); + } +} + +static void start_dirty_page_tracking(msg_vm_t vm) +{ + msg_host_priv_t priv = msg_host_resource_priv(vm); + priv->dp_enabled = 1; + + reset_dirty_pages(vm); +} + +static void stop_dirty_page_tracking(msg_vm_t vm) +{ + msg_host_priv_t priv = msg_host_resource_priv(vm); + priv->dp_enabled = 0; +} + +#if 0 +/* It might be natural that we define dp_rate for each task. But, we will also + * have to care about how each task behavior affects the memory update behavior + * at the operating system level. It may not be easy to model it with a simple algorithm. */ +double calc_updated_pages(char *key, msg_vm_t vm, dirty_page_t dp, double remaining, double clock) +{ + double computed = dp->prev_remaining - remaining; + double duration = clock - dp->prev_clock; + double updated = dp->task->dp_rate * computed; + + XBT_INFO("%s@%s: computated %f ops (remaining %f -> %f) in %f secs (%f -> %f)", + key, sg_host_name(vm), computed, dp->prev_remaining, remaining, duration, dp->prev_clock, clock); + XBT_INFO("%s@%s: updated %f bytes, %f Mbytes/s", + key, sg_host_name(vm), updated, updated / duration / 1000 / 1000); + + return updated; +} +#endif + +double get_computed(char *key, msg_vm_t vm, dirty_page_t dp, double remaining, double clock) +{ + double computed = dp->prev_remaining - remaining; + double duration = clock - dp->prev_clock; + + XBT_DEBUG("%s@%s: computated %f ops (remaining %f -> %f) in %f secs (%f -> %f)", + key, sg_host_name(vm), computed, dp->prev_remaining, remaining, duration, dp->prev_clock, clock); + + return computed; +} + +static double lookup_computed_flop_counts(msg_vm_t vm, int stage2_round_for_fancy_debug) +{ + msg_host_priv_t priv = msg_host_resource_priv(vm); + double total = 0; + + char *key = NULL; + xbt_dict_cursor_t cursor = NULL; + dirty_page_t dp = NULL; + xbt_dict_foreach(priv->dp_objs, cursor, key, dp) { + double remaining = MSG_task_get_remaining_computation(dp->task); + double clock = MSG_get_clock(); + + // total += calc_updated_pages(key, vm, dp, remaining, clock); + total += get_computed(key, vm, dp, remaining, clock); + + dp->prev_remaining = remaining; + dp->prev_clock = clock; + } + + total += priv->dp_updated_by_deleted_tasks; + + XBT_INFO("mig-stage2.%d: computed %f flop_counts (including %f by deleted tasks)", + stage2_round_for_fancy_debug, + total, priv->dp_updated_by_deleted_tasks); + + + + priv->dp_updated_by_deleted_tasks = 0; + + + return total; +} + +void MSG_host_add_task(msg_host_t host, msg_task_t task) +{ + msg_host_priv_t priv = msg_host_resource_priv(host); + double remaining = MSG_task_get_remaining_computation(task); + char *key = bprintf("%s-%lld", task->name, task->counter); + + dirty_page_t dp = xbt_new0(s_dirty_page, 1); + dp->task = task; + + /* It should be okay that we add a task onto a migrating VM. */ + if (priv->dp_enabled) { + dp->prev_clock = MSG_get_clock(); + dp->prev_remaining = remaining; + } + + xbt_assert(xbt_dict_get_or_null(priv->dp_objs, key) == NULL); + xbt_dict_set(priv->dp_objs, key, dp, NULL); + XBT_DEBUG("add %s on %s (remaining %f, dp_enabled %d)", key, sg_host_name(host), remaining, priv->dp_enabled); + + xbt_free(key); +} + +void MSG_host_del_task(msg_host_t host, msg_task_t task) +{ + msg_host_priv_t priv = msg_host_resource_priv(host); + + char *key = bprintf("%s-%lld", task->name, task->counter); + + dirty_page_t dp = xbt_dict_get_or_null(priv->dp_objs, key); + xbt_assert(dp->task == task); + + /* If we are in the middle of dirty page tracking, we record how much + * computaion has been done until now, and keep the information for the + * lookup_() function that will called soon. */ + if (priv->dp_enabled) { + double remaining = MSG_task_get_remaining_computation(task); + double clock = MSG_get_clock(); + // double updated = calc_updated_pages(key, host, dp, remaining, clock); + double updated = get_computed(key, host, dp, remaining, clock); + + priv->dp_updated_by_deleted_tasks += updated; + } + + xbt_dict_remove(priv->dp_objs, key); + xbt_free(dp); + + XBT_DEBUG("del %s on %s", key, sg_host_name(host)); + + xbt_free(key); +} + + +static void send_migration_data(const char *vm_name, const char *src_pm_name, const char *dst_pm_name, + double size, char *mbox, int stage, int stage2_round) +{ + char *task_name = get_mig_task_name(vm_name, src_pm_name, dst_pm_name, stage); + msg_task_t task = MSG_task_create(task_name, 0, size, NULL); + msg_error_t ret = MSG_task_send(task, mbox); + xbt_assert(ret == MSG_OK); + + if (stage == 2) + XBT_INFO("mig-stage%d.%d: sent %f", stage, stage2_round, size); + else + XBT_INFO("mig-stage%d: sent %f", stage, size); + + xbt_free(task_name); +} + + +static int migration_tx_fun(int argc, char *argv[]) +{ + const char *pr_name = MSG_process_get_name(MSG_process_self()); + const char *host_name = MSG_host_get_name(MSG_host_self()); + + XBT_DEBUG("mig: tx_start"); + + xbt_assert(argc == 4); + const char *vm_name = argv[1]; + const char *src_pm_name = argv[2]; + const char *dst_pm_name = argv[3]; + msg_vm_t vm = MSG_get_host_by_name(vm_name); + + + s_ws_params_t params; + simcall_host_get_params(vm, ¶ms); + const long ramsize = params.ramsize; + const long devsize = params.devsize; + const int skip_stage2 = params.skip_stage2; + const double max_downtime = params.max_downtime; + const double dp_rate = params.dp_rate; + const double dp_cap = params.dp_cap; + double remaining_size = ramsize + devsize; + double threshold = max_downtime * 125 * 1000 * 1000; + + + if (ramsize == 0) + XBT_WARN("migrate a VM, but ramsize is zero"); + + char *mbox = get_mig_mbox_src_dst(vm_name, src_pm_name, dst_pm_name); + + XBT_INFO("mig-stage1: remaining_size %f", remaining_size); + + /* Stage1: send all memory pages to the destination. */ + start_dirty_page_tracking(vm); + + send_migration_data(vm_name, src_pm_name, dst_pm_name, ramsize, mbox, 1, 0); + + remaining_size -= ramsize; + + + + /* Stage2: send update pages iteratively until the size of remaining states + * 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 (;;) { + // long updated_size = lookup_dirty_pages(vm); + double updated_size = lookup_computed_flop_counts(vm, stage2_round) * dp_rate; + if (updated_size > dp_cap) { + XBT_INFO("mig-stage2.%d: %f bytes updated, but cap it with the working set size %f", + stage2_round, updated_size, dp_cap); + updated_size = dp_cap; + } + + remaining_size += updated_size; + + XBT_INFO("mig-stage2.%d: remaining_size %f (%s threshold %f)", stage2_round, + remaining_size, (remaining_size < threshold) ? "<" : ">", threshold); + + if (remaining_size < threshold) + break; + + send_migration_data(vm_name, src_pm_name, dst_pm_name, updated_size, mbox, 2, stage2_round); + + remaining_size -= updated_size; + stage2_round += 1; + } + + +stage3: + /* Stage3: stop the VM and copy the rest of states. */ + XBT_INFO("mig-stage3: remaining_size %f", remaining_size); + simcall_vm_suspend(vm); + stop_dirty_page_tracking(vm); + + send_migration_data(vm_name, src_pm_name, dst_pm_name, remaining_size, mbox, 3, 0); + + xbt_free(mbox); + + XBT_DEBUG("mig: tx_done"); + + return 0; +} + + + +static void do_migration(msg_vm_t vm, msg_host_t src_pm, msg_host_t dst_pm) +{ + char *mbox_ctl = get_mig_mbox_ctl(sg_host_name(vm), sg_host_name(src_pm), sg_host_name(dst_pm)); + + { + char *pr_name = get_mig_process_rx_name(sg_host_name(vm), sg_host_name(src_pm), sg_host_name(dst_pm)); + int nargvs = 5; + char **argv = xbt_new(char *, nargvs); + argv[0] = xbt_strdup(pr_name); + argv[1] = xbt_strdup(sg_host_name(vm)); + argv[2] = xbt_strdup(sg_host_name(src_pm)); + argv[3] = xbt_strdup(sg_host_name(dst_pm)); + argv[4] = NULL; + + msg_process_t pr = MSG_process_create_with_arguments(pr_name, migration_rx_fun, NULL, dst_pm, nargvs - 1, argv); + + xbt_free(pr_name); + } + + { + char *pr_name = get_mig_process_tx_name(sg_host_name(vm), sg_host_name(src_pm), sg_host_name(dst_pm)); + int nargvs = 5; + char **argv = xbt_new(char *, nargvs); + argv[0] = xbt_strdup(pr_name); + argv[1] = xbt_strdup(sg_host_name(vm)); + argv[2] = xbt_strdup(sg_host_name(src_pm)); + argv[3] = xbt_strdup(sg_host_name(dst_pm)); + argv[4] = NULL; + + msg_process_t pr = MSG_process_create_with_arguments(pr_name, migration_tx_fun, NULL, src_pm, nargvs - 1, argv); + + xbt_free(pr_name); + } + + /* wait until the migration have finished */ + { + msg_task_t task = NULL; + msg_error_t ret = MSG_task_recv(&task, mbox_ctl); + xbt_assert(ret == MSG_OK); + + char *expected_task_name = get_mig_task_name(sg_host_name(vm), sg_host_name(src_pm), sg_host_name(dst_pm), 4); + xbt_assert(strcmp(task->name, expected_task_name) == 0); + xbt_free(expected_task_name); + } + + xbt_free(mbox_ctl); +} + + /** @brief Migrate the VM to the given host. * @ingroup msg_VMs * @@ -285,7 +688,12 @@ void MSG_vm_migrate(msg_vm_t vm, msg_host_t new_pm) msg_host_t old_pm = simcall_vm_get_pm(vm); - simcall_vm_migrate(vm, new_pm); + if (simcall_vm_get_state(vm) != SURF_VM_STATE_RUNNING) + THROWF(vm_error, 0, "VM(%s) is not running", sg_host_name(vm)); + + do_migration(vm, old_pm, new_pm); + + XBT_DEBUG("VM(%s) moved from PM(%s) to PM(%s)", vm->key, old_pm->key, new_pm->key); @@ -388,3 +796,12 @@ void MSG_vm_destroy(msg_vm_t vm) TRACE_msg_vm_end(vm); #endif } + + +/** @brief Get the physical host of a givne VM. + * @ingroup msg_VMs + */ +msg_host_t MSG_vm_get_pm(msg_vm_t vm) +{ + return simcall_vm_get_pm(vm); +}