SYS-OSS/FRET-qemu
4
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Migration Pull request

Browse Source 
Hi
 
 This are the reviewed patches for the freeze period:
 
 - colo: fix/optimize several things (rao, chen)
 - shutdown qio channels correctly when an error happens (li)
 - serveral multifd patches for the zero series (me)
 
 Please apply.
 
 Thanks, Juan.
 -----BEGIN PGP SIGNATURE-----
 
 iQIzBAABCAAdFiEEGJn/jt6/WMzuA0uC9IfvGFhy1yMFAmG5xCkACgkQ9IfvGFhy
 1yODJA/8CcxfASdxCdiTGcoOQz43259Awp61vg44qqdGz1R0aBQNCCYIL0BFlHJo
 Ta6m1pZfi8CPWCVkReSvkMp2B3RUSwyX1DhI2WxhTmEhY0Kpa4nS2S5NGbA3hCEJ
 3GFN/zDshnCJB47BXNZAz6GoNfY6pYlcGhCEDduWNcbTMAy9AdvTWq/XllGoKCfu
 A3JXNg0P5PqNzdR2AVr24M5gwqNOR3ral36rr5vLWZF5CNd71jLpMj5jmDGnpiA5
 QCQyLcOf7muSXOOFbIZPCmDh3Kvd5voDuG3UbL+2wycLny/8gpc4GoheYHaVaO4F
 87hcwbLZL99/3BOLhXIuumchA3ZOiJruH/VWbSbY6vnuz7/j5EZMQycCSQQuV/+T
 zYK3jDqIUDTiTiRj+d68bHlfAloNF42kQVAbhF9Uk9YDfZsoOSIkRNl+BP/N+YCQ
 OyXoZZ/U1rSjqRw0pTn8A9VvJ5SAsKNEDizkEnvEtjW9mWEq/h/Vym17Nw50V0CK
 JxVNOkg4PBjoy0WdXrF10YHf/R5bOanNUm3tFSMTarAuEu1Oc/I5e7GrffxoVqS+
 Sj+0QS//fVJdzynybA4SpXLhZ5n8HixYT3SfihveevItZateePtJ7gkP3RugwZ9V
 /eUMuC3PwFeYwQ5EC+dkGfku+of3UE+TOzBP8VdM9ahRwRSh1X8=
 =3ge7
 -----END PGP SIGNATURE-----

Merge tag 'migration-20211214-pull-request' of https://gitlab.com/juan.quintela/qemu into staging

Migration Pull request

Hi

This are the reviewed patches for the freeze period:

- colo: fix/optimize several things (rao, chen)
- shutdown qio channels correctly when an error happens (li)
- serveral multifd patches for the zero series (me)

Please apply.

Thanks, Juan.

# gpg: Signature made Wed 15 Dec 2021 02:32:09 AM PST
# gpg:                using RSA key 1899FF8EDEBF58CCEE034B82F487EF185872D723
# gpg: Good signature from "Juan Quintela <quintela@redhat.com>" [full]
# gpg:                 aka "Juan Quintela <quintela@trasno.org>" [full]

* tag 'migration-20211214-pull-request' of https://gitlab.com/juan.quintela/qemu:
  multifd: Make zlib compression method not use iovs
  multifd: Make zstd compression method not use iovs
  COLO: Move some trace code behind qemu_mutex_unlock_iothread()
  multifd: Shut down the QIO channels to avoid blocking the send threads when they are terminated.
  multifd: Fill offset and block for reception
  multifd: remove used parameter from send_recv_pages() method
  multifd: remove used parameter from send_prepare() method
  multifd: The variable is only used inside the loop
  multifd: Add missing documention
  multifd: Rename used field to num
  migration: Never call twice qemu_target_page_size()
  multifd: Delete useless operation
  dump: Remove is_zero_page()
  migration: Remove is_zero_range()
  migration/colo: Optimize COLO primary node start code path
  Fixed a QEMU hang when guest poweroff in COLO mode
  migration/colo: More accurate update checkpoint time
  migration/ram.c: Remove the qemu_mutex_lock in colo_flush_ram_cache.

Signed-off-by: Richard Henderson <richard.henderson@linaro.org>
This commit is contained in:
Richard Henderson 2021-12-15 09:15:06 -08:00
commit 0da6106883
10 changed files with 131 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
dump/dump.c
View File

@ -1293,14 +1293,6 @@ static size_t get_len_buf_out(size_t page_size, uint32_t flag_compress)
return 0; return 0;
} }
/*
* check if the page is all 0
*/
static inline bool is_zero_page(const uint8_t *buf, size_t page_size)
{
return buffer_is_zero(buf, page_size);
}
static void write_dump_pages(DumpState *s, Error **errp) static void write_dump_pages(DumpState *s, Error **errp)
{ {
int ret = 0; int ret = 0;
@ -1357,7 +1349,7 @@ static void write_dump_pages(DumpState *s, Error **errp)
*/ */
while (get_next_page(&block_iter, &pfn_iter, &buf, s)) { while (get_next_page(&block_iter, &pfn_iter, &buf, s)) {
/* check zero page */ /* check zero page */
if (is_zero_page(buf, s->dump_info.page_size)) { if (buffer_is_zero(buf, s->dump_info.page_size)) {
ret = write_cache(&page_desc, &pd_zero, sizeof(PageDescriptor), ret = write_cache(&page_desc, &pd_zero, sizeof(PageDescriptor),
false); false);
if (ret < 0) { if (ret < 0) {

1
include/migration/colo.h
View File

@ -37,4 +37,5 @@ COLOMode get_colo_mode(void);
void colo_do_failover(void); void colo_do_failover(void);
void colo_checkpoint_notify(void *opaque); void colo_checkpoint_notify(void *opaque);
void colo_shutdown(void);
#endif #endif

33
migration/colo.c
View File

@ -530,7 +530,6 @@ static void colo_process_checkpoint(MigrationState *s)
{ {
QIOChannelBuffer *bioc; QIOChannelBuffer *bioc;
QEMUFile *fb = NULL; QEMUFile *fb = NULL;
int64_t current_time = qemu_clock_get_ms(QEMU_CLOCK_HOST);
Error *local_err = NULL; Error *local_err = NULL;
int ret; int ret;
@ -578,8 +577,8 @@ static void colo_process_checkpoint(MigrationState *s)
qemu_mutex_unlock_iothread(); qemu_mutex_unlock_iothread();
trace_colo_vm_state_change("stop", "run"); trace_colo_vm_state_change("stop", "run");
timer_mod(s->colo_delay_timer, timer_mod(s->colo_delay_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
current_time + s->parameters.x_checkpoint_delay); s->parameters.x_checkpoint_delay);
while (s->state == MIGRATION_STATUS_COLO) { while (s->state == MIGRATION_STATUS_COLO) {
if (failover_get_state() != FAILOVER_STATUS_NONE) { if (failover_get_state() != FAILOVER_STATUS_NONE) {
@ -667,8 +666,6 @@ void migrate_start_colo_process(MigrationState *s)
colo_checkpoint_notify, s); colo_checkpoint_notify, s);
qemu_sem_init(&s->colo_exit_sem, 0); qemu_sem_init(&s->colo_exit_sem, 0);
migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
MIGRATION_STATUS_COLO);
colo_process_checkpoint(s); colo_process_checkpoint(s);
qemu_mutex_lock_iothread(); qemu_mutex_lock_iothread();
} }
@ -683,8 +680,8 @@ static void colo_incoming_process_checkpoint(MigrationIncomingState *mis,
qemu_mutex_lock_iothread(); qemu_mutex_lock_iothread();
vm_stop_force_state(RUN_STATE_COLO); vm_stop_force_state(RUN_STATE_COLO);
trace_colo_vm_state_change("run", "stop");
qemu_mutex_unlock_iothread(); qemu_mutex_unlock_iothread();
trace_colo_vm_state_change("run", "stop");
/* FIXME: This is unnecessary for periodic checkpoint mode */ /* FIXME: This is unnecessary for periodic checkpoint mode */
colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_REPLY, colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_REPLY,
@ -786,8 +783,8 @@ static void colo_incoming_process_checkpoint(MigrationIncomingState *mis,
vmstate_loading = false; vmstate_loading = false;
vm_start(); vm_start();
trace_colo_vm_state_change("stop", "run");
qemu_mutex_unlock_iothread(); qemu_mutex_unlock_iothread();
trace_colo_vm_state_change("stop", "run");
if (failover_get_state() == FAILOVER_STATUS_RELAUNCH) { if (failover_get_state() == FAILOVER_STATUS_RELAUNCH) {
return; return;
@ -820,6 +817,26 @@ static void colo_wait_handle_message(MigrationIncomingState *mis,
} }
} }
void colo_shutdown(void)
{
MigrationIncomingState *mis = NULL;
MigrationState *s = NULL;
switch (get_colo_mode()) {
case COLO_MODE_PRIMARY:
s = migrate_get_current();
qemu_event_set(&s->colo_checkpoint_event);
qemu_sem_post(&s->colo_exit_sem);
break;
case COLO_MODE_SECONDARY:
mis = migration_incoming_get_current();
qemu_sem_post(&mis->colo_incoming_sem);
break;
default:
break;
}
}
void *colo_process_incoming_thread(void *opaque) void *colo_process_incoming_thread(void *opaque)
{ {
MigrationIncomingState *mis = opaque; MigrationIncomingState *mis = opaque;
@ -870,8 +887,8 @@ void *colo_process_incoming_thread(void *opaque)
abort(); abort();
#endif #endif
vm_start(); vm_start();
trace_colo_vm_state_change("stop", "run");
qemu_mutex_unlock_iothread(); qemu_mutex_unlock_iothread();
trace_colo_vm_state_change("stop", "run");
colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_READY, colo_send_message(mis->to_src_file, COLO_MESSAGE_CHECKPOINT_READY,
&local_err); &local_err);

26
migration/migration.c
View File

@ -225,6 +225,12 @@ void migration_cancel(const Error *error)
void migration_shutdown(void) void migration_shutdown(void)
{ {
/*
* When the QEMU main thread exit, the COLO thread
* may wait a semaphore. So, we should wakeup the
* COLO thread before migration shutdown.
*/
colo_shutdown();
/* /*
* Cancel the current migration - that will (eventually) * Cancel the current migration - that will (eventually)
* stop the migration using this structure * stop the migration using this structure
@ -990,6 +996,8 @@ static void populate_time_info(MigrationInfo *info, MigrationState *s)
static void populate_ram_info(MigrationInfo *info, MigrationState *s) static void populate_ram_info(MigrationInfo *info, MigrationState *s)
{ {
size_t page_size = qemu_target_page_size();
info->has_ram = true; info->has_ram = true;
info->ram = g_malloc0(sizeof(*info->ram)); info->ram = g_malloc0(sizeof(*info->ram));
info->ram->transferred = ram_counters.transferred; info->ram->transferred = ram_counters.transferred;
@ -998,12 +1006,11 @@ static void populate_ram_info(MigrationInfo *info, MigrationState *s)
/* legacy value. It is not used anymore */ /* legacy value. It is not used anymore */
info->ram->skipped = 0; info->ram->skipped = 0;
info->ram->normal = ram_counters.normal; info->ram->normal = ram_counters.normal;
info->ram->normal_bytes = ram_counters.normal * info->ram->normal_bytes = ram_counters.normal * page_size;
qemu_target_page_size();
info->ram->mbps = s->mbps; info->ram->mbps = s->mbps;
info->ram->dirty_sync_count = ram_counters.dirty_sync_count; info->ram->dirty_sync_count = ram_counters.dirty_sync_count;
info->ram->postcopy_requests = ram_counters.postcopy_requests; info->ram->postcopy_requests = ram_counters.postcopy_requests;
info->ram->page_size = qemu_target_page_size(); info->ram->page_size = page_size;
info->ram->multifd_bytes = ram_counters.multifd_bytes; info->ram->multifd_bytes = ram_counters.multifd_bytes;
info->ram->pages_per_second = s->pages_per_second; info->ram->pages_per_second = s->pages_per_second;
@ -3607,12 +3614,7 @@ static void migration_iteration_finish(MigrationState *s)
migration_calculate_complete(s); migration_calculate_complete(s);
runstate_set(RUN_STATE_POSTMIGRATE); runstate_set(RUN_STATE_POSTMIGRATE);
break; break;
case MIGRATION_STATUS_COLO:
case MIGRATION_STATUS_ACTIVE:
/*
* We should really assert here, but since it's during
* migration, let's try to reduce the usage of assertions.
*/
if (!migrate_colo_enabled()) { if (!migrate_colo_enabled()) {
error_report("%s: critical error: calling COLO code without " error_report("%s: critical error: calling COLO code without "
"COLO enabled", __func__); "COLO enabled", __func__);
@ -3622,6 +3624,12 @@ static void migration_iteration_finish(MigrationState *s)
* Fixme: we will run VM in COLO no matter its old running state. * Fixme: we will run VM in COLO no matter its old running state.
* After exited COLO, we will keep running. * After exited COLO, we will keep running.
*/ */
/* Fallthrough */
case MIGRATION_STATUS_ACTIVE:
/*
* We should really assert here, but since it's during
* migration, let's try to reduce the usage of assertions.
*/
s->vm_was_running = true; s->vm_was_running = true;
/* Fallthrough */ /* Fallthrough */
case MIGRATION_STATUS_FAILED: case MIGRATION_STATUS_FAILED:

48
migration/multifd-zlib.c
View File

@ -13,6 +13,7 @@
#include "qemu/osdep.h" #include "qemu/osdep.h"
#include <zlib.h> #include <zlib.h>
#include "qemu/rcu.h" #include "qemu/rcu.h"
#include "exec/ramblock.h"
#include "exec/target_page.h" #include "exec/target_page.h"
#include "qapi/error.h" #include "qapi/error.h"
#include "migration.h" #include "migration.h"
@ -42,7 +43,6 @@ struct zlib_data {
*/ */
static int zlib_send_setup(MultiFDSendParams *p, Error **errp) static int zlib_send_setup(MultiFDSendParams *p, Error **errp)
{ {
uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
struct zlib_data *z = g_malloc0(sizeof(struct zlib_data)); struct zlib_data *z = g_malloc0(sizeof(struct zlib_data));
z_stream *zs = &z->zs; z_stream *zs = &z->zs;
@ -54,9 +54,8 @@ static int zlib_send_setup(MultiFDSendParams *p, Error **errp)
error_setg(errp, "multifd %d: deflate init failed", p->id); error_setg(errp, "multifd %d: deflate init failed", p->id);
return -1; return -1;
} }
/* We will never have more than page_count pages */ /* To be safe, we reserve twice the size of the packet */
z->zbuff_len = page_count * qemu_target_page_size(); z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
z->zbuff_len *= 2;
z->zbuff = g_try_malloc(z->zbuff_len); z->zbuff = g_try_malloc(z->zbuff_len);
if (!z->zbuff) { if (!z->zbuff) {
deflateEnd(&z->zs); deflateEnd(&z->zs);
@ -74,6 +73,7 @@ static int zlib_send_setup(MultiFDSendParams *p, Error **errp)
* Close the channel and return memory. * Close the channel and return memory.
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @errp: pointer to an error
*/ */
static void zlib_send_cleanup(MultiFDSendParams *p, Error **errp) static void zlib_send_cleanup(MultiFDSendParams *p, Error **errp)
{ {
@ -95,27 +95,27 @@ static void zlib_send_cleanup(MultiFDSendParams *p, Error **errp)
* Returns 0 for success or -1 for error * Returns 0 for success or -1 for error
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @used: number of pages used * @errp: pointer to an error
*/ */
static int zlib_send_prepare(MultiFDSendParams *p, uint32_t used, Error **errp) static int zlib_send_prepare(MultiFDSendParams *p, Error **errp)
{ {
struct iovec *iov = p->pages->iov;
struct zlib_data *z = p->data; struct zlib_data *z = p->data;
size_t page_size = qemu_target_page_size();
z_stream *zs = &z->zs; z_stream *zs = &z->zs;
uint32_t out_size = 0; uint32_t out_size = 0;
int ret; int ret;
uint32_t i; uint32_t i;
for (i = 0; i < used; i++) { for (i = 0; i < p->pages->num; i++) {
uint32_t available = z->zbuff_len - out_size; uint32_t available = z->zbuff_len - out_size;
int flush = Z_NO_FLUSH; int flush = Z_NO_FLUSH;
if (i == used - 1) { if (i == p->pages->num - 1) {
flush = Z_SYNC_FLUSH; flush = Z_SYNC_FLUSH;
} }
zs->avail_in = iov[i].iov_len; zs->avail_in = page_size;
zs->next_in = iov[i].iov_base; zs->next_in = p->pages->block->host + p->pages->offset[i];
zs->avail_out = available; zs->avail_out = available;
zs->next_out = z->zbuff + out_size; zs->next_out = z->zbuff + out_size;
@ -180,7 +180,6 @@ static int zlib_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
*/ */
static int zlib_recv_setup(MultiFDRecvParams *p, Error **errp) static int zlib_recv_setup(MultiFDRecvParams *p, Error **errp)
{ {
uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
struct zlib_data *z = g_malloc0(sizeof(struct zlib_data)); struct zlib_data *z = g_malloc0(sizeof(struct zlib_data));
z_stream *zs = &z->zs; z_stream *zs = &z->zs;
@ -194,10 +193,8 @@ static int zlib_recv_setup(MultiFDRecvParams *p, Error **errp)
error_setg(errp, "multifd %d: inflate init failed", p->id); error_setg(errp, "multifd %d: inflate init failed", p->id);
return -1; return -1;
} }
/* We will never have more than page_count pages */ /* To be safe, we reserve twice the size of the packet */
z->zbuff_len = page_count * qemu_target_page_size(); z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
/* We know compression "could" use more space */
z->zbuff_len *= 2;
z->zbuff = g_try_malloc(z->zbuff_len); z->zbuff = g_try_malloc(z->zbuff_len);
if (!z->zbuff) { if (!z->zbuff) {
inflateEnd(zs); inflateEnd(zs);
@ -234,17 +231,17 @@ static void zlib_recv_cleanup(MultiFDRecvParams *p)
* Returns 0 for success or -1 for error * Returns 0 for success or -1 for error
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @used: number of pages used
* @errp: pointer to an error * @errp: pointer to an error
*/ */
static int zlib_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp) static int zlib_recv_pages(MultiFDRecvParams *p, Error **errp)
{ {
struct zlib_data *z = p->data; struct zlib_data *z = p->data;
size_t page_size = qemu_target_page_size();
z_stream *zs = &z->zs; z_stream *zs = &z->zs;
uint32_t in_size = p->next_packet_size; uint32_t in_size = p->next_packet_size;
/* we measure the change of total_out */ /* we measure the change of total_out */
uint32_t out_size = zs->total_out; uint32_t out_size = zs->total_out;
uint32_t expected_size = used * qemu_target_page_size(); uint32_t expected_size = p->pages->num * qemu_target_page_size();
uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK; uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
int ret; int ret;
int i; int i;
@ -263,17 +260,16 @@ static int zlib_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
zs->avail_in = in_size; zs->avail_in = in_size;
zs->next_in = z->zbuff; zs->next_in = z->zbuff;
for (i = 0; i < used; i++) { for (i = 0; i < p->pages->num; i++) {
struct iovec *iov = &p->pages->iov[i];
int flush = Z_NO_FLUSH; int flush = Z_NO_FLUSH;
unsigned long start = zs->total_out; unsigned long start = zs->total_out;
if (i == used - 1) { if (i == p->pages->num - 1) {
flush = Z_SYNC_FLUSH; flush = Z_SYNC_FLUSH;
} }
zs->avail_out = iov->iov_len; zs->avail_out = page_size;
zs->next_out = iov->iov_base; zs->next_out = p->pages->block->host + p->pages->offset[i];
/* /*
* Welcome to inflate semantics * Welcome to inflate semantics
@ -286,8 +282,8 @@ static int zlib_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
do { do {
ret = inflate(zs, flush); ret = inflate(zs, flush);
} while (ret == Z_OK && zs->avail_in } while (ret == Z_OK && zs->avail_in
&& (zs->total_out - start) < iov->iov_len); && (zs->total_out - start) < page_size);
if (ret == Z_OK && (zs->total_out - start) < iov->iov_len) { if (ret == Z_OK && (zs->total_out - start) < page_size) {
error_setg(errp, "multifd %d: inflate generated too few output", error_setg(errp, "multifd %d: inflate generated too few output",
p->id); p->id);
return -1; return -1;

47
migration/multifd-zstd.c
View File

@ -13,6 +13,7 @@
#include "qemu/osdep.h" #include "qemu/osdep.h"
#include <zstd.h> #include <zstd.h>
#include "qemu/rcu.h" #include "qemu/rcu.h"
#include "exec/ramblock.h"
#include "exec/target_page.h" #include "exec/target_page.h"
#include "qapi/error.h" #include "qapi/error.h"
#include "migration.h" #include "migration.h"
@ -47,7 +48,6 @@ struct zstd_data {
*/ */
static int zstd_send_setup(MultiFDSendParams *p, Error **errp) static int zstd_send_setup(MultiFDSendParams *p, Error **errp)
{ {
uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
struct zstd_data *z = g_new0(struct zstd_data, 1); struct zstd_data *z = g_new0(struct zstd_data, 1);
int res; int res;
@ -67,9 +67,8 @@ static int zstd_send_setup(MultiFDSendParams *p, Error **errp)
p->id, ZSTD_getErrorName(res)); p->id, ZSTD_getErrorName(res));
return -1; return -1;
} }
/* We will never have more than page_count pages */ /* To be safe, we reserve twice the size of the packet */
z->zbuff_len = page_count * qemu_target_page_size(); z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
z->zbuff_len *= 2;
z->zbuff = g_try_malloc(z->zbuff_len); z->zbuff = g_try_malloc(z->zbuff_len);
if (!z->zbuff) { if (!z->zbuff) {
ZSTD_freeCStream(z->zcs); ZSTD_freeCStream(z->zcs);
@ -86,6 +85,7 @@ static int zstd_send_setup(MultiFDSendParams *p, Error **errp)
* Close the channel and return memory. * Close the channel and return memory.
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @errp: pointer to an error
*/ */
static void zstd_send_cleanup(MultiFDSendParams *p, Error **errp) static void zstd_send_cleanup(MultiFDSendParams *p, Error **errp)
{ {
@ -108,12 +108,12 @@ static void zstd_send_cleanup(MultiFDSendParams *p, Error **errp)
* Returns 0 for success or -1 for error * Returns 0 for success or -1 for error
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @used: number of pages used * @errp: pointer to an error
*/ */
static int zstd_send_prepare(MultiFDSendParams *p, uint32_t used, Error **errp) static int zstd_send_prepare(MultiFDSendParams *p, Error **errp)
{ {
struct iovec *iov = p->pages->iov;
struct zstd_data *z = p->data; struct zstd_data *z = p->data;
size_t page_size = qemu_target_page_size();
int ret; int ret;
uint32_t i; uint32_t i;
@ -121,14 +121,14 @@ static int zstd_send_prepare(MultiFDSendParams *p, uint32_t used, Error **errp)
z->out.size = z->zbuff_len; z->out.size = z->zbuff_len;
z->out.pos = 0; z->out.pos = 0;
for (i = 0; i < used; i++) { for (i = 0; i < p->pages->num; i++) {
ZSTD_EndDirective flush = ZSTD_e_continue; ZSTD_EndDirective flush = ZSTD_e_continue;
if (i == used - 1) { if (i == p->pages->num - 1) {
flush = ZSTD_e_flush; flush = ZSTD_e_flush;
} }
z->in.src = iov[i].iov_base; z->in.src = p->pages->block->host + p->pages->offset[i];
z->in.size = iov[i].iov_len; z->in.size = page_size;
z->in.pos = 0; z->in.pos = 0;
/* /*
@ -191,7 +191,6 @@ static int zstd_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
*/ */
static int zstd_recv_setup(MultiFDRecvParams *p, Error **errp) static int zstd_recv_setup(MultiFDRecvParams *p, Error **errp)
{ {
uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
struct zstd_data *z = g_new0(struct zstd_data, 1); struct zstd_data *z = g_new0(struct zstd_data, 1);
int ret; int ret;
@ -212,10 +211,8 @@ static int zstd_recv_setup(MultiFDRecvParams *p, Error **errp)
return -1; return -1;
} }
/* We will never have more than page_count pages */ /* To be safe, we reserve twice the size of the packet */
z->zbuff_len = page_count * qemu_target_page_size(); z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
/* We know compression "could" use more space */
z->zbuff_len *= 2;
z->zbuff = g_try_malloc(z->zbuff_len); z->zbuff = g_try_malloc(z->zbuff_len);
if (!z->zbuff) { if (!z->zbuff) {
ZSTD_freeDStream(z->zds); ZSTD_freeDStream(z->zds);
@ -254,14 +251,14 @@ static void zstd_recv_cleanup(MultiFDRecvParams *p)
* Returns 0 for success or -1 for error * Returns 0 for success or -1 for error
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @used: number of pages used
* @errp: pointer to an error * @errp: pointer to an error
*/ */
static int zstd_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp) static int zstd_recv_pages(MultiFDRecvParams *p, Error **errp)
{ {
uint32_t in_size = p->next_packet_size; uint32_t in_size = p->next_packet_size;
uint32_t out_size = 0; uint32_t out_size = 0;
uint32_t expected_size = used * qemu_target_page_size(); size_t page_size = qemu_target_page_size();
uint32_t expected_size = p->pages->num * page_size;
uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK; uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
struct zstd_data *z = p->data; struct zstd_data *z = p->data;
int ret; int ret;
@ -282,11 +279,9 @@ static int zstd_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
z->in.size = in_size; z->in.size = in_size;
z->in.pos = 0; z->in.pos = 0;
for (i = 0; i < used; i++) { for (i = 0; i < p->pages->num; i++) {
struct iovec *iov = &p->pages->iov[i]; z->out.dst = p->pages->block->host + p->pages->offset[i];
z->out.size = page_size;
z->out.dst = iov->iov_base;
z->out.size = iov->iov_len;
z->out.pos = 0; z->out.pos = 0;
/* /*
@ -300,8 +295,8 @@ static int zstd_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
do { do {
ret = ZSTD_decompressStream(z->zds, &z->out, &z->in); ret = ZSTD_decompressStream(z->zds, &z->out, &z->in);
} while (ret > 0 && (z->in.size - z->in.pos > 0) } while (ret > 0 && (z->in.size - z->in.pos > 0)
&& (z->out.pos < iov->iov_len)); && (z->out.pos < page_size));
if (ret > 0 && (z->out.pos < iov->iov_len)) { if (ret > 0 && (z->out.pos < page_size)) {
error_setg(errp, "multifd %d: decompressStream buffer too small", error_setg(errp, "multifd %d: decompressStream buffer too small",
p->id); p->id);
return -1; return -1;

70
migration/multifd.c
View File

@ -66,6 +66,7 @@ static int nocomp_send_setup(MultiFDSendParams *p, Error **errp)
* For no compression this function does nothing. * For no compression this function does nothing.
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @errp: pointer to an error
*/ */
static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp) static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp)
{ {
@ -81,13 +82,11 @@ static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp)
* Returns 0 for success or -1 for error * Returns 0 for success or -1 for error
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @used: number of pages used
* @errp: pointer to an error * @errp: pointer to an error
*/ */
static int nocomp_send_prepare(MultiFDSendParams *p, uint32_t used, static int nocomp_send_prepare(MultiFDSendParams *p, Error **errp)
Error **errp)
{ {
p->next_packet_size = used * qemu_target_page_size(); p->next_packet_size = p->pages->num * qemu_target_page_size();
p->flags |= MULTIFD_FLAG_NOCOMP; p->flags |= MULTIFD_FLAG_NOCOMP;
return 0; return 0;
} }
@ -142,10 +141,9 @@ static void nocomp_recv_cleanup(MultiFDRecvParams *p)
* Returns 0 for success or -1 for error * Returns 0 for success or -1 for error
* *
* @p: Params for the channel that we are using * @p: Params for the channel that we are using
* @used: number of pages used
* @errp: pointer to an error * @errp: pointer to an error
*/ */
static int nocomp_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp) static int nocomp_recv_pages(MultiFDRecvParams *p, Error **errp)
{ {
uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK; uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
@ -154,7 +152,7 @@ static int nocomp_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
p->id, flags, MULTIFD_FLAG_NOCOMP); p->id, flags, MULTIFD_FLAG_NOCOMP);
return -1; return -1;
} }
return qio_channel_readv_all(p->c, p->pages->iov, used, errp); return qio_channel_readv_all(p->c, p->pages->iov, p->pages->num, errp);
} }
static MultiFDMethods multifd_nocomp_ops = { static MultiFDMethods multifd_nocomp_ops = {
@ -252,7 +250,7 @@ static MultiFDPages_t *multifd_pages_init(size_t size)
static void multifd_pages_clear(MultiFDPages_t *pages) static void multifd_pages_clear(MultiFDPages_t *pages)
{ {
pages->used = 0; pages->num = 0;
pages->allocated = 0; pages->allocated = 0;
pages->packet_num = 0; pages->packet_num = 0;
pages->block = NULL; pages->block = NULL;
@ -270,7 +268,7 @@ static void multifd_send_fill_packet(MultiFDSendParams *p)
packet->flags = cpu_to_be32(p->flags); packet->flags = cpu_to_be32(p->flags);
packet->pages_alloc = cpu_to_be32(p->pages->allocated); packet->pages_alloc = cpu_to_be32(p->pages->allocated);
packet->pages_used = cpu_to_be32(p->pages->used); packet->pages_used = cpu_to_be32(p->pages->num);
packet->next_packet_size = cpu_to_be32(p->next_packet_size); packet->next_packet_size = cpu_to_be32(p->next_packet_size);
packet->packet_num = cpu_to_be64(p->packet_num); packet->packet_num = cpu_to_be64(p->packet_num);
@ -278,7 +276,7 @@ static void multifd_send_fill_packet(MultiFDSendParams *p)
strncpy(packet->ramblock, p->pages->block->idstr, 256); strncpy(packet->ramblock, p->pages->block->idstr, 256);
} }
for (i = 0; i < p->pages->used; i++) { for (i = 0; i < p->pages->num; i++) {
/* there are architectures where ram_addr_t is 32 bit */ /* there are architectures where ram_addr_t is 32 bit */
uint64_t temp = p->pages->offset[i]; uint64_t temp = p->pages->offset[i];
@ -289,7 +287,8 @@ static void multifd_send_fill_packet(MultiFDSendParams *p)
static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp) static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
{ {
MultiFDPacket_t *packet = p->packet; MultiFDPacket_t *packet = p->packet;
uint32_t pages_max = MULTIFD_PACKET_SIZE / qemu_target_page_size(); size_t page_size = qemu_target_page_size();
uint32_t pages_max = MULTIFD_PACKET_SIZE / page_size;
RAMBlock *block; RAMBlock *block;
int i; int i;
@ -331,18 +330,18 @@ static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
p->pages = multifd_pages_init(packet->pages_alloc); p->pages = multifd_pages_init(packet->pages_alloc);
} }
p->pages->used = be32_to_cpu(packet->pages_used); p->pages->num = be32_to_cpu(packet->pages_used);
if (p->pages->used > packet->pages_alloc) { if (p->pages->num > packet->pages_alloc) {
error_setg(errp, "multifd: received packet " error_setg(errp, "multifd: received packet "
"with %d pages and expected maximum pages are %d", "with %d pages and expected maximum pages are %d",
p->pages->used, packet->pages_alloc) ; p->pages->num, packet->pages_alloc) ;
return -1; return -1;
} }
p->next_packet_size = be32_to_cpu(packet->next_packet_size); p->next_packet_size = be32_to_cpu(packet->next_packet_size);
p->packet_num = be64_to_cpu(packet->packet_num); p->packet_num = be64_to_cpu(packet->packet_num);
if (p->pages->used == 0) { if (p->pages->num == 0) {
return 0; return 0;
} }
@ -355,17 +354,19 @@ static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
return -1; return -1;
} }
for (i = 0; i < p->pages->used; i++) { p->pages->block = block;
for (i = 0; i < p->pages->num; i++) {
uint64_t offset = be64_to_cpu(packet->offset[i]); uint64_t offset = be64_to_cpu(packet->offset[i]);
if (offset > (block->used_length - qemu_target_page_size())) { if (offset > (block->used_length - page_size)) {
error_setg(errp, "multifd: offset too long %" PRIu64 error_setg(errp, "multifd: offset too long %" PRIu64
" (max " RAM_ADDR_FMT ")", " (max " RAM_ADDR_FMT ")",
offset, block->used_length); offset, block->used_length);
return -1; return -1;
} }
p->pages->offset[i] = offset;
p->pages->iov[i].iov_base = block->host + offset; p->pages->iov[i].iov_base = block->host + offset;
p->pages->iov[i].iov_len = qemu_target_page_size(); p->pages->iov[i].iov_len = page_size;
} }
return 0; return 0;
@ -442,13 +443,13 @@ static int multifd_send_pages(QEMUFile *f)
} }
qemu_mutex_unlock(&p->mutex); qemu_mutex_unlock(&p->mutex);
} }
assert(!p->pages->used); assert(!p->pages->num);
assert(!p->pages->block); assert(!p->pages->block);
p->packet_num = multifd_send_state->packet_num++; p->packet_num = multifd_send_state->packet_num++;
multifd_send_state->pages = p->pages; multifd_send_state->pages = p->pages;
p->pages = pages; p->pages = pages;
transferred = ((uint64_t) pages->used) * qemu_target_page_size() transferred = ((uint64_t) pages->num) * qemu_target_page_size()
+ p->packet_len; + p->packet_len;
qemu_file_update_transfer(f, transferred); qemu_file_update_transfer(f, transferred);
ram_counters.multifd_bytes += transferred; ram_counters.multifd_bytes += transferred;
@ -468,12 +469,12 @@ int multifd_queue_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset)
} }
if (pages->block == block) { if (pages->block == block) {
pages->offset[pages->used] = offset; pages->offset[pages->num] = offset;
pages->iov[pages->used].iov_base = block->host + offset; pages->iov[pages->num].iov_base = block->host + offset;
pages->iov[pages->used].iov_len = qemu_target_page_size(); pages->iov[pages->num].iov_len = qemu_target_page_size();
pages->used++; pages->num++;
if (pages->used < pages->allocated) { if (pages->num < pages->allocated) {
return 1; return 1;
} }
} }
@ -523,6 +524,9 @@ static void multifd_send_terminate_threads(Error *err)
qemu_mutex_lock(&p->mutex); qemu_mutex_lock(&p->mutex);
p->quit = true; p->quit = true;
qemu_sem_post(&p->sem); qemu_sem_post(&p->sem);
if (p->c) {
qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
}
qemu_mutex_unlock(&p->mutex); qemu_mutex_unlock(&p->mutex);
} }
} }
@ -585,7 +589,7 @@ void multifd_send_sync_main(QEMUFile *f)
if (!migrate_use_multifd()) { if (!migrate_use_multifd()) {
return; return;
} }
if (multifd_send_state->pages->used) { if (multifd_send_state->pages->num) {
if (multifd_send_pages(f) < 0) { if (multifd_send_pages(f) < 0) {
error_report("%s: multifd_send_pages fail", __func__); error_report("%s: multifd_send_pages fail", __func__);
return; return;
@ -627,7 +631,6 @@ static void *multifd_send_thread(void *opaque)
MultiFDSendParams *p = opaque; MultiFDSendParams *p = opaque;
Error *local_err = NULL; Error *local_err = NULL;
int ret = 0; int ret = 0;
uint32_t flags = 0;
trace_multifd_send_thread_start(p->id); trace_multifd_send_thread_start(p->id);
rcu_register_thread(); rcu_register_thread();
@ -648,13 +651,12 @@ static void *multifd_send_thread(void *opaque)
qemu_mutex_lock(&p->mutex); qemu_mutex_lock(&p->mutex);
if (p->pending_job) { if (p->pending_job) {
uint32_t used = p->pages->used; uint32_t used = p->pages->num;
uint64_t packet_num = p->packet_num; uint64_t packet_num = p->packet_num;
flags = p->flags; uint32_t flags = p->flags;
if (used) { if (used) {
ret = multifd_send_state->ops->send_prepare(p, used, ret = multifd_send_state->ops->send_prepare(p, &local_err);
&local_err);
if (ret != 0) { if (ret != 0) {
qemu_mutex_unlock(&p->mutex); qemu_mutex_unlock(&p->mutex);
break; break;
@ -664,7 +666,7 @@ static void *multifd_send_thread(void *opaque)
p->flags = 0; p->flags = 0;
p->num_packets++; p->num_packets++;
p->num_pages += used; p->num_pages += used;
p->pages->used = 0; p->pages->num = 0;
p->pages->block = NULL; p->pages->block = NULL;
qemu_mutex_unlock(&p->mutex); qemu_mutex_unlock(&p->mutex);
@ -1090,7 +1092,7 @@ static void *multifd_recv_thread(void *opaque)
break; break;
} }
used = p->pages->used; used = p->pages->num;
flags = p->flags; flags = p->flags;
/* recv methods don't know how to handle the SYNC flag */ /* recv methods don't know how to handle the SYNC flag */
p->flags &= ~MULTIFD_FLAG_SYNC; p->flags &= ~MULTIFD_FLAG_SYNC;
@ -1101,7 +1103,7 @@ static void *multifd_recv_thread(void *opaque)
qemu_mutex_unlock(&p->mutex); qemu_mutex_unlock(&p->mutex);
if (used) { if (used) {
ret = multifd_recv_state->ops->recv_pages(p, used, &local_err); ret = multifd_recv_state->ops->recv_pages(p, &local_err);
if (ret != 0) { if (ret != 0) {
break; break;
} }

6
migration/multifd.h
View File

@ -55,7 +55,7 @@ typedef struct {
typedef struct { typedef struct {
/* number of used pages */ /* number of used pages */
uint32_t used; uint32_t num;
/* number of allocated pages */ /* number of allocated pages */
uint32_t allocated; uint32_t allocated;
/* global number of generated multifd packets */ /* global number of generated multifd packets */
@ -159,7 +159,7 @@ typedef struct {
/* Cleanup for sending side */ /* Cleanup for sending side */
void (*send_cleanup)(MultiFDSendParams *p, Error **errp); void (*send_cleanup)(MultiFDSendParams *p, Error **errp);
/* Prepare the send packet */ /* Prepare the send packet */
int (*send_prepare)(MultiFDSendParams *p, uint32_t used, Error **errp); int (*send_prepare)(MultiFDSendParams *p, Error **errp);
/* Write the send packet */ /* Write the send packet */
int (*send_write)(MultiFDSendParams *p, uint32_t used, Error **errp); int (*send_write)(MultiFDSendParams *p, uint32_t used, Error **errp);
/* Setup for receiving side */ /* Setup for receiving side */
@ -167,7 +167,7 @@ typedef struct {
/* Cleanup for receiving side */ /* Cleanup for receiving side */
void (*recv_cleanup)(MultiFDRecvParams *p); void (*recv_cleanup)(MultiFDRecvParams *p);
/* Read all pages */ /* Read all pages */
int (*recv_pages)(MultiFDRecvParams *p, uint32_t used, Error **errp); int (*recv_pages)(MultiFDRecvParams *p, Error **errp);
} MultiFDMethods; } MultiFDMethods;
void multifd_register_ops(int method, MultiFDMethods *ops); void multifd_register_ops(int method, MultiFDMethods *ops);

11
migration/ram.c
View File

@ -81,11 +81,6 @@
/* 0x80 is reserved in migration.h start with 0x100 next */ /* 0x80 is reserved in migration.h start with 0x100 next */
#define RAM_SAVE_FLAG_COMPRESS_PAGE 0x100 #define RAM_SAVE_FLAG_COMPRESS_PAGE 0x100
static inline bool is_zero_range(uint8_t *p, uint64_t size)
{
return buffer_is_zero(p, size);
}
XBZRLECacheStats xbzrle_counters; XBZRLECacheStats xbzrle_counters;
/* struct contains XBZRLE cache and a static page /* struct contains XBZRLE cache and a static page
@ -1180,7 +1175,7 @@ static int save_zero_page_to_file(RAMState *rs, QEMUFile *file,
uint8_t *p = block->host + offset; uint8_t *p = block->host + offset;
int len = 0; int len = 0;
if (is_zero_range(p, TARGET_PAGE_SIZE)) { if (buffer_is_zero(p, TARGET_PAGE_SIZE)) {
len += save_page_header(rs, file, block, offset | RAM_SAVE_FLAG_ZERO); len += save_page_header(rs, file, block, offset | RAM_SAVE_FLAG_ZERO);
qemu_put_byte(file, 0); qemu_put_byte(file, 0);
len += 1; len += 1;
@ -3367,7 +3362,7 @@ static inline void *colo_cache_from_block_offset(RAMBlock *block,
*/ */
void ram_handle_compressed(void *host, uint8_t ch, uint64_t size) void ram_handle_compressed(void *host, uint8_t ch, uint64_t size)
{ {
if (ch != 0 || !is_zero_range(host, size)) { if (ch != 0 || !buffer_is_zero(host, size)) {
memset(host, ch, size); memset(host, ch, size);
} }
} }
@ -3918,7 +3913,6 @@ void colo_flush_ram_cache(void)
unsigned long offset = 0; unsigned long offset = 0;
memory_global_dirty_log_sync(); memory_global_dirty_log_sync();
qemu_mutex_lock(&ram_state->bitmap_mutex);
WITH_RCU_READ_LOCK_GUARD() { WITH_RCU_READ_LOCK_GUARD() {
RAMBLOCK_FOREACH_NOT_IGNORED(block) { RAMBLOCK_FOREACH_NOT_IGNORED(block) {
ramblock_sync_dirty_bitmap(ram_state, block); ramblock_sync_dirty_bitmap(ram_state, block);
@ -3954,7 +3948,6 @@ void colo_flush_ram_cache(void)
} }
} }
trace_colo_flush_ram_cache_end(); trace_colo_flush_ram_cache_end();
qemu_mutex_unlock(&ram_state->bitmap_mutex);
} }
/** /**

5
migration/savevm.c
View File

@ -1685,6 +1685,7 @@ static int loadvm_postcopy_handle_advise(MigrationIncomingState *mis,
{ {
PostcopyState ps = postcopy_state_set(POSTCOPY_INCOMING_ADVISE); PostcopyState ps = postcopy_state_set(POSTCOPY_INCOMING_ADVISE);
uint64_t remote_pagesize_summary, local_pagesize_summary, remote_tps; uint64_t remote_pagesize_summary, local_pagesize_summary, remote_tps;
size_t page_size = qemu_target_page_size();
Error *local_err = NULL; Error *local_err = NULL;
trace_loadvm_postcopy_handle_advise(); trace_loadvm_postcopy_handle_advise();
@ -1741,13 +1742,13 @@ static int loadvm_postcopy_handle_advise(MigrationIncomingState *mis,
} }
remote_tps = qemu_get_be64(mis->from_src_file); remote_tps = qemu_get_be64(mis->from_src_file);
if (remote_tps != qemu_target_page_size()) { if (remote_tps != page_size) {
/* /*
* Again, some differences could be dealt with, but for now keep it * Again, some differences could be dealt with, but for now keep it
* simple. * simple.
*/ */
error_report("Postcopy needs matching target page sizes (s=%d d=%zd)", error_report("Postcopy needs matching target page sizes (s=%d d=%zd)",
(int)remote_tps, qemu_target_page_size()); (int)remote_tps, page_size);
return -1; return -1;
} }