X-Git-Url: https://git.proxmox.com/?a=blobdiff_plain;f=migration.c;h=3eb0fad147a40f25dde8cfaf1b9767775cbbb8e5;hb=d7108d90100d5bac5965abef5ed73f2602adae14;hp=11725ae3fc00ca95be6f19d484ec847eb939e8ba;hpb=864a556e9a800116a305f10fbb714268ca7e9bc3;p=qemu.git diff --git a/migration.c b/migration.c index 11725ae3f..3eb0fad14 100644 --- a/migration.c +++ b/migration.c @@ -23,6 +23,7 @@ #include "migration/block.h" #include "qemu/thread.h" #include "qmp-commands.h" +#include "trace.h" //#define DEBUG_MIGRATION @@ -98,7 +99,7 @@ static void process_incoming_migration_co(void *opaque) qemu_fclose(f); if (ret < 0) { fprintf(stderr, "load of migration failed\n"); - exit(0); + exit(EXIT_FAILURE); } qemu_announce_self(); DPRINTF("successfully loaded vm state\n"); @@ -120,7 +121,7 @@ void process_incoming_migration(QEMUFile *f) int fd = qemu_get_fd(f); assert(fd != -1); - socket_set_nonblock(fd); + qemu_set_nonblock(fd); qemu_coroutine_enter(co, f); } @@ -196,11 +197,11 @@ MigrationInfo *qmp_query_migrate(Error **errp) info->ram->remaining = ram_bytes_remaining(); info->ram->total = ram_bytes_total(); info->ram->duplicate = dup_mig_pages_transferred(); + info->ram->skipped = skipped_mig_pages_transferred(); info->ram->normal = norm_mig_pages_transferred(); info->ram->normal_bytes = norm_mig_bytes_transferred(); info->ram->dirty_pages_rate = s->dirty_pages_rate; - if (blk_mig_active()) { info->has_disk = true; info->disk = g_malloc0(sizeof(*info->disk)); @@ -226,6 +227,7 @@ MigrationInfo *qmp_query_migrate(Error **errp) info->ram->remaining = 0; info->ram->total = ram_bytes_total(); info->ram->duplicate = dup_mig_pages_transferred(); + info->ram->skipped = skipped_mig_pages_transferred(); info->ram->normal = norm_mig_pages_transferred(); info->ram->normal_bytes = norm_mig_bytes_transferred(); break; @@ -260,81 +262,54 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params, /* shared migration helpers */ -static int migrate_fd_cleanup(MigrationState *s) +static void migrate_fd_cleanup(void *opaque) { - int ret = 0; + MigrationState *s = opaque; + + qemu_bh_delete(s->cleanup_bh); + s->cleanup_bh = NULL; if (s->file) { DPRINTF("closing file\n"); - ret = qemu_fclose(s->file); + qemu_mutex_unlock_iothread(); + qemu_thread_join(&s->thread); + qemu_mutex_lock_iothread(); + + qemu_fclose(s->file); s->file = NULL; } - assert(s->fd == -1); - return ret; -} + assert(s->state != MIG_STATE_ACTIVE); + + if (s->state != MIG_STATE_COMPLETED) { + qemu_savevm_state_cancel(); + } -void migrate_fd_error(MigrationState *s) -{ - DPRINTF("setting error state\n"); - s->state = MIG_STATE_ERROR; notifier_list_notify(&migration_state_notifiers, s); - migrate_fd_cleanup(s); } -static void migrate_fd_completed(MigrationState *s) +static void migrate_finish_set_state(MigrationState *s, int new_state) { - DPRINTF("setting completed state\n"); - if (migrate_fd_cleanup(s) < 0) { - s->state = MIG_STATE_ERROR; - } else { - s->state = MIG_STATE_COMPLETED; - runstate_set(RUN_STATE_POSTMIGRATE); + if (__sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, + new_state) == new_state) { + trace_migrate_set_state(new_state); } - notifier_list_notify(&migration_state_notifiers, s); } -static ssize_t migrate_fd_put_buffer(MigrationState *s, const void *data, - size_t size) +void migrate_fd_error(MigrationState *s) { - ssize_t ret; - - if (s->state != MIG_STATE_ACTIVE) { - return -EIO; - } - - do { - ret = s->write(s, data, size); - } while (ret == -1 && ((s->get_error(s)) == EINTR)); - - if (ret == -1) - ret = -(s->get_error(s)); - - return ret; + DPRINTF("setting error state\n"); + assert(s->file == NULL); + s->state = MIG_STATE_ERROR; + trace_migrate_set_state(MIG_STATE_ERROR); + notifier_list_notify(&migration_state_notifiers, s); } static void migrate_fd_cancel(MigrationState *s) { - if (s->state != MIG_STATE_ACTIVE) - return; - DPRINTF("cancelling migration\n"); - s->state = MIG_STATE_CANCELLED; - notifier_list_notify(&migration_state_notifiers, s); - qemu_savevm_state_cancel(); - - migrate_fd_cleanup(s); -} - -int migrate_fd_close(MigrationState *s) -{ - int rc = 0; - if (s->fd != -1) { - rc = s->close(s); - s->fd = -1; - } - return rc; + migrate_finish_set_state(s, MIG_STATE_CANCELLED); } void add_migration_state_change_notifier(Notifier *notify) @@ -382,8 +357,9 @@ static MigrationState *migrate_init(const MigrationParams *params) s->bandwidth_limit = bandwidth_limit; s->state = MIG_STATE_SETUP; - s->total_time = qemu_get_clock_ms(rt_clock); + trace_migrate_set_state(MIG_STATE_SETUP); + s->total_time = qemu_get_clock_ms(rt_clock); return s; } @@ -480,10 +456,15 @@ void qmp_migrate_set_speed(int64_t value, Error **errp) if (value < 0) { value = 0; } + if (value > SIZE_MAX) { + value = SIZE_MAX; + } s = migrate_get_current(); s->bandwidth_limit = value; - qemu_file_set_rate_limit(s->file, s->bandwidth_limit); + if (s->file) { + qemu_file_set_rate_limit(s->file, s->bandwidth_limit / XFER_LIMIT_RATIO); + } } void qmp_migrate_set_downtime(double value, Error **errp) @@ -513,224 +494,53 @@ int64_t migrate_xbzrle_cache_size(void) /* migration thread support */ - -static ssize_t buffered_flush(MigrationState *s) -{ - size_t offset = 0; - ssize_t ret = 0; - - DPRINTF("flushing %zu byte(s) of data\n", s->buffer_size); - - while (s->bytes_xfer < s->xfer_limit && offset < s->buffer_size) { - size_t to_send = MIN(s->buffer_size - offset, s->xfer_limit - s->bytes_xfer); - ret = migrate_fd_put_buffer(s, s->buffer + offset, to_send); - if (ret <= 0) { - DPRINTF("error flushing data, %zd\n", ret); - break; - } else { - DPRINTF("flushed %zd byte(s)\n", ret); - offset += ret; - s->bytes_xfer += ret; - } - } - - DPRINTF("flushed %zu of %zu byte(s)\n", offset, s->buffer_size); - memmove(s->buffer, s->buffer + offset, s->buffer_size - offset); - s->buffer_size -= offset; - - if (ret < 0) { - return ret; - } - return offset; -} - -static int buffered_put_buffer(void *opaque, const uint8_t *buf, - int64_t pos, int size) -{ - MigrationState *s = opaque; - ssize_t error; - - DPRINTF("putting %d bytes at %" PRId64 "\n", size, pos); - - error = qemu_file_get_error(s->file); - if (error) { - DPRINTF("flush when error, bailing: %s\n", strerror(-error)); - return error; - } - - if (size <= 0) { - return size; - } - - if (size > (s->buffer_capacity - s->buffer_size)) { - DPRINTF("increasing buffer capacity from %zu by %zu\n", - s->buffer_capacity, size + 1024); - - s->buffer_capacity += size + 1024; - - s->buffer = g_realloc(s->buffer, s->buffer_capacity); - } - - memcpy(s->buffer + s->buffer_size, buf, size); - s->buffer_size += size; - - return size; -} - -static int buffered_close(void *opaque) -{ - MigrationState *s = opaque; - ssize_t ret = 0; - int ret2; - - DPRINTF("closing\n"); - - s->xfer_limit = INT_MAX; - while (!qemu_file_get_error(s->file) && s->buffer_size) { - ret = buffered_flush(s); - if (ret < 0) { - break; - } - } - - ret2 = migrate_fd_close(s); - if (ret >= 0) { - ret = ret2; - } - s->complete = true; - return ret; -} - -static int buffered_get_fd(void *opaque) -{ - MigrationState *s = opaque; - - return s->fd; -} - -/* - * The meaning of the return values is: - * 0: We can continue sending - * 1: Time to stop - * negative: There has been an error - */ -static int buffered_rate_limit(void *opaque) -{ - MigrationState *s = opaque; - int ret; - - ret = qemu_file_get_error(s->file); - if (ret) { - return ret; - } - - if (s->bytes_xfer >= s->xfer_limit) { - return 1; - } - - return 0; -} - -static int64_t buffered_set_rate_limit(void *opaque, int64_t new_rate) -{ - MigrationState *s = opaque; - if (qemu_file_get_error(s->file)) { - goto out; - } - if (new_rate > SIZE_MAX) { - new_rate = SIZE_MAX; - } - - s->xfer_limit = new_rate / XFER_LIMIT_RATIO; - -out: - return s->xfer_limit; -} - -static int64_t buffered_get_rate_limit(void *opaque) -{ - MigrationState *s = opaque; - - return s->xfer_limit; -} - -static void *buffered_file_thread(void *opaque) +static void *migration_thread(void *opaque) { MigrationState *s = opaque; int64_t initial_time = qemu_get_clock_ms(rt_clock); int64_t sleep_time = 0; + int64_t initial_bytes = 0; int64_t max_size = 0; - bool last_round = false; - int ret; + int64_t start_time = initial_time; + bool old_vm_running = false; - qemu_mutex_lock_iothread(); DPRINTF("beginning savevm\n"); - ret = qemu_savevm_state_begin(s->file, &s->params); - if (ret < 0) { - DPRINTF("failed, %d\n", ret); - qemu_mutex_unlock_iothread(); - goto out; - } - qemu_mutex_unlock_iothread(); + qemu_savevm_state_begin(s->file, &s->params); - while (true) { + while (s->state == MIG_STATE_ACTIVE) { int64_t current_time; uint64_t pending_size; - qemu_mutex_lock_iothread(); - if (s->state != MIG_STATE_ACTIVE) { - DPRINTF("put_ready returning because of non-active state\n"); - qemu_mutex_unlock_iothread(); - break; - } - if (s->complete) { - qemu_mutex_unlock_iothread(); - break; - } - if (s->bytes_xfer < s->xfer_limit) { + if (!qemu_file_rate_limit(s->file)) { DPRINTF("iterate\n"); pending_size = qemu_savevm_state_pending(s->file, max_size); DPRINTF("pending size %lu max %lu\n", pending_size, max_size); if (pending_size && pending_size >= max_size) { - ret = qemu_savevm_state_iterate(s->file); - if (ret < 0) { - qemu_mutex_unlock_iothread(); - break; - } + qemu_savevm_state_iterate(s->file); } else { - int old_vm_running = runstate_is_running(); - int64_t start_time, end_time; - DPRINTF("done iterating\n"); + qemu_mutex_lock_iothread(); start_time = qemu_get_clock_ms(rt_clock); qemu_system_wakeup_request(QEMU_WAKEUP_REASON_OTHER); - if (old_vm_running) { - vm_stop(RUN_STATE_FINISH_MIGRATE); - } else { - vm_stop_force_state(RUN_STATE_FINISH_MIGRATE); - } - ret = qemu_savevm_state_complete(s->file); - if (ret < 0) { - qemu_mutex_unlock_iothread(); + old_vm_running = runstate_is_running(); + vm_stop_force_state(RUN_STATE_FINISH_MIGRATE); + qemu_file_set_rate_limit(s->file, INT_MAX); + qemu_savevm_state_complete(s->file); + qemu_mutex_unlock_iothread(); + if (!qemu_file_get_error(s->file)) { + migrate_finish_set_state(s, MIG_STATE_COMPLETED); break; - } else { - migrate_fd_completed(s); } - end_time = qemu_get_clock_ms(rt_clock); - s->total_time = end_time - s->total_time; - s->downtime = end_time - start_time; - if (s->state != MIG_STATE_COMPLETED) { - if (old_vm_running) { - vm_start(); - } - } - last_round = true; } } - qemu_mutex_unlock_iothread(); + + if (qemu_file_get_error(s->file)) { + migrate_finish_set_state(s, MIG_STATE_ERROR); + break; + } current_time = qemu_get_clock_ms(rt_clock); if (current_time >= initial_time + BUFFER_DELAY) { - uint64_t transferred_bytes = s->bytes_xfer; + uint64_t transferred_bytes = qemu_ftell(s->file) - initial_bytes; uint64_t time_spent = current_time - initial_time - sleep_time; double bandwidth = transferred_bytes / time_spent; max_size = bandwidth * migrate_max_downtime() / 1000000; @@ -744,54 +554,48 @@ static void *buffered_file_thread(void *opaque) s->expected_downtime = s->dirty_bytes_rate / bandwidth; } - s->bytes_xfer = 0; + qemu_file_reset_rate_limit(s->file); sleep_time = 0; initial_time = current_time; + initial_bytes = qemu_ftell(s->file); } - if (!last_round && (s->bytes_xfer >= s->xfer_limit)) { + if (qemu_file_rate_limit(s->file)) { /* usleep expects microseconds */ g_usleep((initial_time + BUFFER_DELAY - current_time)*1000); sleep_time += qemu_get_clock_ms(rt_clock) - current_time; } - ret = buffered_flush(s); - if (ret < 0) { - break; - } } -out: - if (ret < 0) { - migrate_fd_error(s); + qemu_mutex_lock_iothread(); + if (s->state == MIG_STATE_COMPLETED) { + int64_t end_time = qemu_get_clock_ms(rt_clock); + s->total_time = end_time - s->total_time; + s->downtime = end_time - start_time; + runstate_set(RUN_STATE_POSTMIGRATE); + } else { + if (old_vm_running) { + vm_start(); + } } - g_free(s->buffer); + qemu_bh_schedule(s->cleanup_bh); + qemu_mutex_unlock_iothread(); + return NULL; } -static const QEMUFileOps buffered_file_ops = { - .get_fd = buffered_get_fd, - .put_buffer = buffered_put_buffer, - .close = buffered_close, - .rate_limit = buffered_rate_limit, - .get_rate_limit = buffered_get_rate_limit, - .set_rate_limit = buffered_set_rate_limit, -}; - void migrate_fd_connect(MigrationState *s) { s->state = MIG_STATE_ACTIVE; - s->bytes_xfer = 0; - s->buffer = NULL; - s->buffer_size = 0; - s->buffer_capacity = 0; + trace_migrate_set_state(MIG_STATE_ACTIVE); + /* This is a best 1st approximation. ns to ms */ s->expected_downtime = max_downtime/1000000; + s->cleanup_bh = qemu_bh_new(migrate_fd_cleanup, s); - s->xfer_limit = s->bandwidth_limit / XFER_LIMIT_RATIO; - s->complete = false; - - s->file = qemu_fopen_ops(s, &buffered_file_ops); + qemu_file_set_rate_limit(s->file, + s->bandwidth_limit / XFER_LIMIT_RATIO); - qemu_thread_create(&s->thread, buffered_file_thread, s, - QEMU_THREAD_DETACHED); + qemu_thread_create(&s->thread, migration_thread, s, + QEMU_THREAD_JOINABLE); notifier_list_notify(&migration_state_notifiers, s); }