[Qemu-devel] [PULL 22/57] Return path: Source handling of return path
From: "Dr. David Alan Gilbert" Open a return path, and handle messages that are received upon it. Signed-off-by: Dr. David Alan Gilbert Reviewed-by: Juan Quintela Signed-off-by: Juan Quintela --- include/migration/migration.h | 8 +++ migration/migration.c | 159 +- trace-events | 10 +++ 3 files changed, 175 insertions(+), 2 deletions(-) diff --git a/include/migration/migration.h b/include/migration/migration.h index 3ce3fda..571466b 100644 --- a/include/migration/migration.h +++ b/include/migration/migration.h @@ -80,6 +80,14 @@ struct MigrationState int state; MigrationParams params; + +/* State related to return path */ +struct { +QEMUFile *from_dst_file; +QemuThreadrp_thread; +bool error; +} rp_state; + double mbps; int64_t total_time; int64_t downtime; diff --git a/migration/migration.c b/migration/migration.c index 4317bab..b0782c6 100644 --- a/migration/migration.c +++ b/migration/migration.c @@ -696,6 +696,11 @@ static void migrate_fd_cancel(MigrationState *s) QEMUFile *f = migrate_get_current()->file; trace_migrate_fd_cancel(); +if (s->rp_state.from_dst_file) { +/* shutdown the rp socket, so causing the rp thread to shutdown */ +qemu_file_shutdown(s->rp_state.from_dst_file); +} + do { old_state = s->state; if (!migration_is_setup_or_active(old_state)) { @@ -1030,6 +1035,154 @@ int64_t migrate_xbzrle_cache_size(void) return s->xbzrle_cache_size; } +/* migration thread support */ +/* + * Something bad happened to the RP stream, mark an error + * The caller shall print or trace something to indicate why + */ +static void mark_source_rp_bad(MigrationState *s) +{ +s->rp_state.error = true; +} + +static struct rp_cmd_args { +ssize_t len; /* -1 = variable */ +const char *name; +} rp_cmd_args[] = { +[MIG_RP_MSG_INVALID]= { .len = -1, .name = "INVALID" }, +[MIG_RP_MSG_SHUT] = { .len = 4, .name = "SHUT" }, +[MIG_RP_MSG_PONG] = { .len = 4, .name = "PONG" }, +[MIG_RP_MSG_MAX]= { .len = -1, .name = "MAX" }, +}; + +/* + * Handles messages sent on the return path towards the source VM + * + */ +static void *source_return_path_thread(void *opaque) +{ +MigrationState *ms = opaque; +QEMUFile *rp = ms->rp_state.from_dst_file; +uint16_t header_len, header_type; +const int max_len = 512; +uint8_t buf[max_len]; +uint32_t tmp32, sibling_error; +int res; + +trace_source_return_path_thread_entry(); +while (!ms->rp_state.error && !qemu_file_get_error(rp) && + migration_is_setup_or_active(ms->state)) { +trace_source_return_path_thread_loop_top(); +header_type = qemu_get_be16(rp); +header_len = qemu_get_be16(rp); + +if (header_type >= MIG_RP_MSG_MAX || +header_type == MIG_RP_MSG_INVALID) { +error_report("RP: Received invalid message 0x%04x length 0x%04x", +header_type, header_len); +mark_source_rp_bad(ms); +goto out; +} + +if ((rp_cmd_args[header_type].len != -1 && +header_len != rp_cmd_args[header_type].len) || +header_len > max_len) { +error_report("RP: Received '%s' message (0x%04x) with" +"incorrect length %d expecting %zd", +rp_cmd_args[header_type].name, header_type, header_len, +rp_cmd_args[header_type].len); +mark_source_rp_bad(ms); +goto out; +} + +/* We know we've got a valid header by this point */ +res = qemu_get_buffer(rp, buf, header_len); +if (res != header_len) { +error_report("RP: Failed reading data for message 0x%04x" + " read %d expected %d", + header_type, res, header_len); +mark_source_rp_bad(ms); +goto out; +} + +/* OK, we have the message and the data */ +switch (header_type) { +case MIG_RP_MSG_SHUT: +sibling_error = be32_to_cpup((uint32_t *)buf); +trace_source_return_path_thread_shut(sibling_error); +if (sibling_error) { +error_report("RP: Sibling indicated error %d", sibling_error); +mark_source_rp_bad(ms); +} +/* + * We'll let the main thread deal with closing the RP + * we could do a shutdown(2) on it, but we're the only user + * anyway, so there's nothing gained. + */ +goto out; + +case MIG_RP_MSG_PONG: +tmp32 = be32_to_cpup((uint32_t *)buf); +trace_source_return_path_thread_pong(tmp32); +break; + +default: +break; +} +} +if (rp && qemu_file_get_error(rp)) {
[Qemu-devel] [PULL 22/57] Return path: Source handling of return path
From: "Dr. David Alan Gilbert" Open a return path, and handle messages that are received upon it. Signed-off-by: Dr. David Alan Gilbert Reviewed-by: Juan Quintela Signed-off-by: Juan Quintela --- include/migration/migration.h | 8 +++ migration/migration.c | 159 +- trace-events | 10 +++ 3 files changed, 175 insertions(+), 2 deletions(-) diff --git a/include/migration/migration.h b/include/migration/migration.h index 3ce3fda..571466b 100644 --- a/include/migration/migration.h +++ b/include/migration/migration.h @@ -80,6 +80,14 @@ struct MigrationState int state; MigrationParams params; + +/* State related to return path */ +struct { +QEMUFile *from_dst_file; +QemuThreadrp_thread; +bool error; +} rp_state; + double mbps; int64_t total_time; int64_t downtime; diff --git a/migration/migration.c b/migration/migration.c index 4317bab..295deb8 100644 --- a/migration/migration.c +++ b/migration/migration.c @@ -696,6 +696,11 @@ static void migrate_fd_cancel(MigrationState *s) QEMUFile *f = migrate_get_current()->file; trace_migrate_fd_cancel(); +if (s->rp_state.from_dst_file) { +/* shutdown the rp socket, so causing the rp thread to shutdown */ +qemu_file_shutdown(s->rp_state.from_dst_file); +} + do { old_state = s->state; if (!migration_is_setup_or_active(old_state)) { @@ -1030,6 +1035,154 @@ int64_t migrate_xbzrle_cache_size(void) return s->xbzrle_cache_size; } +/* migration thread support */ +/* + * Something bad happened to the RP stream, mark an error + * The caller shall print or trace something to indicate why + */ +static void mark_source_rp_bad(MigrationState *s) +{ +s->rp_state.error = true; +} + +static struct rp_cmd_args { +ssize_t len; /* -1 = variable */ +const char *name; +} rp_cmd_args[] = { +[MIG_RP_MSG_INVALID]= { .len = -1, .name = "INVALID" }, +[MIG_RP_MSG_SHUT] = { .len = 4, .name = "SHUT" }, +[MIG_RP_MSG_PONG] = { .len = 4, .name = "PONG" }, +[MIG_RP_MSG_MAX]= { .len = -1, .name = "MAX" }, +}; + +/* + * Handles messages sent on the return path towards the source VM + * + */ +static void *source_return_path_thread(void *opaque) +{ +MigrationState *ms = opaque; +QEMUFile *rp = ms->rp_state.from_dst_file; +uint16_t header_len, header_type; +const int max_len = 512; +uint8_t buf[max_len]; +uint32_t tmp32, sibling_error; +int res; + +trace_source_return_path_thread_entry(); +while (!ms->rp_state.error && !qemu_file_get_error(rp) && + migration_is_setup_or_active(ms->state)) { +trace_source_return_path_thread_loop_top(); +header_type = qemu_get_be16(rp); +header_len = qemu_get_be16(rp); + +if (header_type >= MIG_RP_MSG_MAX || +header_type == MIG_RP_MSG_INVALID) { +error_report("RP: Received invalid message 0x%04x length 0x%04x", +header_type, header_len); +mark_source_rp_bad(ms); +goto out; +} + +if ((rp_cmd_args[header_type].len != -1 && +header_len != rp_cmd_args[header_type].len) || +header_len > max_len) { +error_report("RP: Received '%s' message (0x%04x) with" +"incorrect length %d expecting %zu", +rp_cmd_args[header_type].name, header_type, header_len, +(size_t)rp_cmd_args[header_type].len); +mark_source_rp_bad(ms); +goto out; +} + +/* We know we've got a valid header by this point */ +res = qemu_get_buffer(rp, buf, header_len); +if (res != header_len) { +error_report("RP: Failed reading data for message 0x%04x" + " read %d expected %d", + header_type, res, header_len); +mark_source_rp_bad(ms); +goto out; +} + +/* OK, we have the message and the data */ +switch (header_type) { +case MIG_RP_MSG_SHUT: +sibling_error = be32_to_cpup((uint32_t *)buf); +trace_source_return_path_thread_shut(sibling_error); +if (sibling_error) { +error_report("RP: Sibling indicated error %d", sibling_error); +mark_source_rp_bad(ms); +} +/* + * We'll let the main thread deal with closing the RP + * we could do a shutdown(2) on it, but we're the only user + * anyway, so there's nothing gained. + */ +goto out; + +case MIG_RP_MSG_PONG: +tmp32 = be32_to_cpup((uint32_t *)buf); +trace_source_return_path_thread_pong(tmp32); +break; + +default: +break; +} +} +if (rp && qemu_file_get_erro