[mirror_qemu.git] / migration / multifd-zlib.c

/*
 * Multifd zlib compression implementation
 *
 * Copyright (c) 2020 Red Hat Inc
 *
 * Authors:
 *  Juan Quintela <quintela@redhat.com>
 *
 * This work is licensed under the terms of the GNU GPL, version 2 or later.
 * See the COPYING file in the top-level directory.
 */

#include "qemu/osdep.h"
#include <zlib.h>
#include "qemu/rcu.h"
#include "exec/ramblock.h"
#include "exec/target_page.h"
#include "qapi/error.h"
#include "migration.h"
#include "trace.h"
#include "options.h"
#include "multifd.h"

struct zlib_data {
    /* stream for compression */
    z_stream zs;
    /* compressed buffer */
    uint8_t *zbuff;
    /* size of compressed buffer */
    uint32_t zbuff_len;
    /* uncompressed buffer of size qemu_target_page_size() */
    uint8_t *buf;
};

/* Multifd zlib compression */

/**
 * zlib_send_setup: setup send side
 *
 * Setup each channel with zlib compression.
 *
 * Returns 0 for success or -1 for error
 *
 * @p: Params for the channel that we are using
 * @errp: pointer to an error
 */
static int zlib_send_setup(MultiFDSendParams *p, Error **errp)
{
    struct zlib_data *z = g_new0(struct zlib_data, 1);
    z_stream *zs = &z->zs;
    const char *err_msg;

    zs->zalloc = Z_NULL;
    zs->zfree = Z_NULL;
    zs->opaque = Z_NULL;
    if (deflateInit(zs, migrate_multifd_zlib_level()) != Z_OK) {
        err_msg = "deflate init failed";
        goto err_free_z;
    }
    /* This is the maximum size of the compressed buffer */
    z->zbuff_len = compressBound(MULTIFD_PACKET_SIZE);
    z->zbuff = g_try_malloc(z->zbuff_len);
    if (!z->zbuff) {
        err_msg = "out of memory for zbuff";
        goto err_deflate_end;
    }
    z->buf = g_try_malloc(qemu_target_page_size());
    if (!z->buf) {
        err_msg = "out of memory for buf";
        goto err_free_zbuff;
    }
    p->compress_data = z;
    return 0;

err_free_zbuff:
    g_free(z->zbuff);
err_deflate_end:
    deflateEnd(zs);
err_free_z:
    g_free(z);
    error_setg(errp, "multifd %u: %s", p->id, err_msg);
    return -1;
}

/**
 * zlib_send_cleanup: cleanup send side
 *
 * Close the channel and return memory.
 *
 * @p: Params for the channel that we are using
 * @errp: pointer to an error
 */
static void zlib_send_cleanup(MultiFDSendParams *p, Error **errp)
{
    struct zlib_data *z = p->compress_data;

    deflateEnd(&z->zs);
    g_free(z->zbuff);
    z->zbuff = NULL;
    g_free(z->buf);
    z->buf = NULL;
    g_free(p->compress_data);
    p->compress_data = NULL;
}

/**
 * zlib_send_prepare: prepare date to be able to send
 *
 * Create a compressed buffer with all the pages that we are going to
 * send.
 *
 * Returns 0 for success or -1 for error
 *
 * @p: Params for the channel that we are using
 * @errp: pointer to an error
 */
static int zlib_send_prepare(MultiFDSendParams *p, Error **errp)
{
    MultiFDPages_t *pages = p->pages;
    struct zlib_data *z = p->compress_data;
    z_stream *zs = &z->zs;
    uint32_t out_size = 0;
    int ret;
    uint32_t i;

    if (!multifd_send_prepare_common(p)) {
        goto out;
    }

    for (i = 0; i < pages->normal_num; i++) {
        uint32_t available = z->zbuff_len - out_size;
        int flush = Z_NO_FLUSH;

        if (i == pages->normal_num - 1) {
            flush = Z_SYNC_FLUSH;
        }

        /*
         * Since the VM might be running, the page may be changing concurrently
         * with compression. zlib does not guarantee that this is safe,
         * therefore copy the page before calling deflate().
         */
        memcpy(z->buf, p->pages->block->host + pages->offset[i], p->page_size);
        zs->avail_in = p->page_size;
        zs->next_in = z->buf;

        zs->avail_out = available;
        zs->next_out = z->zbuff + out_size;

        /*
         * Welcome to deflate semantics
         *
         * We need to loop while:
         * - return is Z_OK
         * - there are stuff to be compressed
         * - there are output space free
         */
        do {
            ret = deflate(zs, flush);
        } while (ret == Z_OK && zs->avail_in && zs->avail_out);
        if (ret == Z_OK && zs->avail_in) {
            error_setg(errp, "multifd %u: deflate failed to compress all input",
                       p->id);
            return -1;
        }
        if (ret != Z_OK) {
            error_setg(errp, "multifd %u: deflate returned %d instead of Z_OK",
                       p->id, ret);
            return -1;
        }
        out_size += available - zs->avail_out;
    }
    p->iov[p->iovs_num].iov_base = z->zbuff;
    p->iov[p->iovs_num].iov_len = out_size;
    p->iovs_num++;
    p->next_packet_size = out_size;

out:
    p->flags |= MULTIFD_FLAG_ZLIB;
    multifd_send_fill_packet(p);
    return 0;
}

/**
 * zlib_recv_setup: setup receive side
 *
 * Create the compressed channel and buffer.
 *
 * Returns 0 for success or -1 for error
 *
 * @p: Params for the channel that we are using
 * @errp: pointer to an error
 */
static int zlib_recv_setup(MultiFDRecvParams *p, Error **errp)
{
    struct zlib_data *z = g_new0(struct zlib_data, 1);
    z_stream *zs = &z->zs;

    p->compress_data = z;
    zs->zalloc = Z_NULL;
    zs->zfree = Z_NULL;
    zs->opaque = Z_NULL;
    zs->avail_in = 0;
    zs->next_in = Z_NULL;
    if (inflateInit(zs) != Z_OK) {
        error_setg(errp, "multifd %u: inflate init failed", p->id);
        return -1;
    }
    /* To be safe, we reserve twice the size of the packet */
    z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
    z->zbuff = g_try_malloc(z->zbuff_len);
    if (!z->zbuff) {
        inflateEnd(zs);
        error_setg(errp, "multifd %u: out of memory for zbuff", p->id);
        return -1;
    }
    return 0;
}

/**
 * zlib_recv_cleanup: setup receive side
 *
 * For no compression this function does nothing.
 *
 * @p: Params for the channel that we are using
 */
static void zlib_recv_cleanup(MultiFDRecvParams *p)
{
    struct zlib_data *z = p->compress_data;

    inflateEnd(&z->zs);
    g_free(z->zbuff);
    z->zbuff = NULL;
    g_free(p->compress_data);
    p->compress_data = NULL;
}

/**
 * zlib_recv: read the data from the channel into actual pages
 *
 * Read the compressed buffer, and uncompress it into the actual
 * pages.
 *
 * Returns 0 for success or -1 for error
 *
 * @p: Params for the channel that we are using
 * @errp: pointer to an error
 */
static int zlib_recv(MultiFDRecvParams *p, Error **errp)
{
    struct zlib_data *z = p->compress_data;
    z_stream *zs = &z->zs;
    uint32_t in_size = p->next_packet_size;
    /* we measure the change of total_out */
    uint32_t out_size = zs->total_out;
    uint32_t expected_size = p->normal_num * p->page_size;
    uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
    int ret;
    int i;

    if (flags != MULTIFD_FLAG_ZLIB) {
        error_setg(errp, "multifd %u: flags received %x flags expected %x",
                   p->id, flags, MULTIFD_FLAG_ZLIB);
        return -1;
    }

    multifd_recv_zero_page_process(p);

    if (!p->normal_num) {
        assert(in_size == 0);
        return 0;
    }

    ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp);

    if (ret != 0) {
        return ret;
    }

    zs->avail_in = in_size;
    zs->next_in = z->zbuff;

    for (i = 0; i < p->normal_num; i++) {
        int flush = Z_NO_FLUSH;
        unsigned long start = zs->total_out;

        ramblock_recv_bitmap_set_offset(p->block, p->normal[i]);
        if (i == p->normal_num - 1) {
            flush = Z_SYNC_FLUSH;
        }

        zs->avail_out = p->page_size;
        zs->next_out = p->host + p->normal[i];

        /*
         * Welcome to inflate semantics
         *
         * We need to loop while:
         * - return is Z_OK
         * - there are input available
         * - we haven't completed a full page
         */
        do {
            ret = inflate(zs, flush);
        } while (ret == Z_OK && zs->avail_in
                             && (zs->total_out - start) < p->page_size);
        if (ret == Z_OK && (zs->total_out - start) < p->page_size) {
            error_setg(errp, "multifd %u: inflate generated too few output",
                       p->id);
            return -1;
        }
        if (ret != Z_OK) {
            error_setg(errp, "multifd %u: inflate returned %d instead of Z_OK",
                       p->id, ret);
            return -1;
        }
    }
    out_size = zs->total_out - out_size;
    if (out_size != expected_size) {
        error_setg(errp, "multifd %u: packet size received %u size expected %u",
                   p->id, out_size, expected_size);
        return -1;
    }

    return 0;
}

static MultiFDMethods multifd_zlib_ops = {
    .send_setup = zlib_send_setup,
    .send_cleanup = zlib_send_cleanup,
    .send_prepare = zlib_send_prepare,
    .recv_setup = zlib_recv_setup,
    .recv_cleanup = zlib_recv_cleanup,
    .recv = zlib_recv
};

static void multifd_zlib_register(void)
{
    multifd_register_ops(MULTIFD_COMPRESSION_ZLIB, &multifd_zlib_ops);
}

migration_init(multifd_zlib_register);
Commit	Line	Data
7ec2c2b3 JQ	1	/*
	2	* Multifd zlib compression implementation
	3	*
	4	* Copyright (c) 2020 Red Hat Inc
	5	*
	6	* Authors:
	7	* Juan Quintela <quintela@redhat.com>
	8	*
	9	* This work is licensed under the terms of the GNU GPL, version 2 or later.
	10	* See the COPYING file in the top-level directory.
	11	*/
	12
	13	#include "qemu/osdep.h"
	14	#include <zlib.h>
	15	#include "qemu/rcu.h"
a5ed2294	16	#include "exec/ramblock.h"
7ec2c2b3 JQ	17	#include "exec/target_page.h"
	18	#include "qapi/error.h"
	19	#include "migration.h"
	20	#include "trace.h"
1dfc4b9e	21	#include "options.h"
7ec2c2b3 JQ	22	#include "multifd.h"
	23
	24	struct zlib_data {
	25	/* stream for compression */
	26	z_stream zs;
	27	/* compressed buffer */
	28	uint8_t *zbuff;
	29	/* size of compressed buffer */
	30	uint32_t zbuff_len;
007e179e IL	31	/* uncompressed buffer of size qemu_target_page_size() */
007e179e IL	32	uint8_t *buf;
7ec2c2b3 JQ	33	};
	34
	35	/* Multifd zlib compression */
	36
	37	/**
	38	* zlib_send_setup: setup send side
	39	*
	40	* Setup each channel with zlib compression.
	41	*
	42	* Returns 0 for success or -1 for error
	43	*
	44	* @p: Params for the channel that we are using
	45	* @errp: pointer to an error
	46	*/
	47	static int zlib_send_setup(MultiFDSendParams p, Error *errp)
	48	{
b21e2380	49	struct zlib_data *z = g_new0(struct zlib_data, 1);
7ec2c2b3	50	z_stream *zs = &z->zs;
007e179e	51	const char *err_msg;
7ec2c2b3 JQ	52
	53	zs->zalloc = Z_NULL;
	54	zs->zfree = Z_NULL;
	55	zs->opaque = Z_NULL;
	56	if (deflateInit(zs, migrate_multifd_zlib_level()) != Z_OK) {
007e179e IL	57	err_msg = "deflate init failed";
007e179e IL	58	goto err_free_z;
7ec2c2b3	59	}
d8b71d96	60	/* This is the maximum size of the compressed buffer */
fc670522	61	z->zbuff_len = compressBound(MULTIFD_PACKET_SIZE);
7ec2c2b3 JQ	62	z->zbuff = g_try_malloc(z->zbuff_len);
7ec2c2b3 JQ	63	if (!z->zbuff) {
007e179e IL	64	err_msg = "out of memory for zbuff";
	65	goto err_deflate_end;
	66	}
	67	z->buf = g_try_malloc(qemu_target_page_size());
	68	if (!z->buf) {
	69	err_msg = "out of memory for buf";
	70	goto err_free_zbuff;
7ec2c2b3	71	}
402dd7ac	72	p->compress_data = z;
7ec2c2b3	73	return 0;
007e179e IL	74
	75	err_free_zbuff:
	76	g_free(z->zbuff);
	77	err_deflate_end:
ee1004bb	78	deflateEnd(zs);
007e179e IL	79	err_free_z:
	80	g_free(z);
	81	error_setg(errp, "multifd %u: %s", p->id, err_msg);
	82	return -1;
7ec2c2b3 JQ	83	}
	84
	85	/**
	86	* zlib_send_cleanup: cleanup send side
	87	*
	88	* Close the channel and return memory.
	89	*
	90	* @p: Params for the channel that we are using
18ede636	91	* @errp: pointer to an error
7ec2c2b3 JQ	92	*/
	93	static void zlib_send_cleanup(MultiFDSendParams p, Error *errp)
	94	{
402dd7ac	95	struct zlib_data *z = p->compress_data;
7ec2c2b3 JQ	96
	97	deflateEnd(&z->zs);
	98	g_free(z->zbuff);
	99	z->zbuff = NULL;
007e179e IL	100	g_free(z->buf);
007e179e IL	101	z->buf = NULL;
402dd7ac FR	102	g_free(p->compress_data);
402dd7ac FR	103	p->compress_data = NULL;
7ec2c2b3 JQ	104	}
	105
	106	/**
	107	* zlib_send_prepare: prepare date to be able to send
	108	*
	109	* Create a compressed buffer with all the pages that we are going to
	110	* send.
	111	*
	112	* Returns 0 for success or -1 for error
	113	*
	114	* @p: Params for the channel that we are using
18ede636	115	* @errp: pointer to an error
7ec2c2b3	116	*/
02fb8104	117	static int zlib_send_prepare(MultiFDSendParams p, Error *errp)
7ec2c2b3	118	{
efd8c543	119	MultiFDPages_t *pages = p->pages;
402dd7ac	120	struct zlib_data *z = p->compress_data;
7ec2c2b3 JQ	121	z_stream *zs = &z->zs;
	122	uint32_t out_size = 0;
	123	int ret;
	124	uint32_t i;
	125
303e6f54 HX	126	if (!multifd_send_prepare_common(p)) {
	127	goto out;
	128	}
25a1f878	129
303e6f54	130	for (i = 0; i < pages->normal_num; i++) {
7ec2c2b3 JQ	131	uint32_t available = z->zbuff_len - out_size;
	132	int flush = Z_NO_FLUSH;
	133
303e6f54	134	if (i == pages->normal_num - 1) {
7ec2c2b3 JQ	135	flush = Z_SYNC_FLUSH;
	136	}
	137
007e179e IL	138	/*
	139	* Since the VM might be running, the page may be changing concurrently
	140	* with compression. zlib does not guarantee that this is safe,
	141	* therefore copy the page before calling deflate().
	142	*/
efd8c543	143	memcpy(z->buf, p->pages->block->host + pages->offset[i], p->page_size);
ddec20f8	144	zs->avail_in = p->page_size;
007e179e	145	zs->next_in = z->buf;
7ec2c2b3 JQ	146
	147	zs->avail_out = available;
	148	zs->next_out = z->zbuff + out_size;
	149
	150	/*
	151	* Welcome to deflate semantics
	152	*
	153	* We need to loop while:
	154	* - return is Z_OK
	155	* - there are stuff to be compressed
	156	* - there are output space free
	157	*/
	158	do {
	159	ret = deflate(zs, flush);
	160	} while (ret == Z_OK && zs->avail_in && zs->avail_out);
	161	if (ret == Z_OK && zs->avail_in) {
04e11404	162	error_setg(errp, "multifd %u: deflate failed to compress all input",
7ec2c2b3 JQ	163	p->id);
	164	return -1;
	165	}
	166	if (ret != Z_OK) {
04e11404	167	error_setg(errp, "multifd %u: deflate returned %d instead of Z_OK",
7ec2c2b3 JQ	168	p->id, ret);
	169	return -1;
	170	}
	171	out_size += available - zs->avail_out;
	172	}
48a4a44c JQ	173	p->iov[p->iovs_num].iov_base = z->zbuff;
	174	p->iov[p->iovs_num].iov_len = out_size;
	175	p->iovs_num++;
7ec2c2b3	176	p->next_packet_size = out_size;
7ec2c2b3	177
303e6f54 HX	178	out:
303e6f54 HX	179	p->flags \|= MULTIFD_FLAG_ZLIB;
25a1f878	180	multifd_send_fill_packet(p);
7ec2c2b3 JQ	181	return 0;
	182	}
	183
7ec2c2b3 JQ	184	/**
	185	* zlib_recv_setup: setup receive side
	186	*
	187	* Create the compressed channel and buffer.
	188	*
	189	* Returns 0 for success or -1 for error
	190	*
	191	* @p: Params for the channel that we are using
	192	* @errp: pointer to an error
	193	*/
	194	static int zlib_recv_setup(MultiFDRecvParams p, Error *errp)
	195	{
b21e2380	196	struct zlib_data *z = g_new0(struct zlib_data, 1);
7ec2c2b3 JQ	197	z_stream *zs = &z->zs;
7ec2c2b3 JQ	198
402dd7ac	199	p->compress_data = z;
7ec2c2b3 JQ	200	zs->zalloc = Z_NULL;
	201	zs->zfree = Z_NULL;
	202	zs->opaque = Z_NULL;
	203	zs->avail_in = 0;
	204	zs->next_in = Z_NULL;
	205	if (inflateInit(zs) != Z_OK) {
04e11404	206	error_setg(errp, "multifd %u: inflate init failed", p->id);
7ec2c2b3 JQ	207	return -1;
7ec2c2b3 JQ	208	}
47a17824 JQ	209	/* To be safe, we reserve twice the size of the packet */
47a17824 JQ	210	z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
7ec2c2b3 JQ	211	z->zbuff = g_try_malloc(z->zbuff_len);
	212	if (!z->zbuff) {
	213	inflateEnd(zs);
04e11404	214	error_setg(errp, "multifd %u: out of memory for zbuff", p->id);
7ec2c2b3 JQ	215	return -1;
	216	}
	217	return 0;
	218	}
	219
	220	/**
	221	* zlib_recv_cleanup: setup receive side
	222	*
	223	* For no compression this function does nothing.
	224	*
	225	* @p: Params for the channel that we are using
	226	*/
	227	static void zlib_recv_cleanup(MultiFDRecvParams *p)
	228	{
402dd7ac	229	struct zlib_data *z = p->compress_data;
7ec2c2b3 JQ	230
	231	inflateEnd(&z->zs);
	232	g_free(z->zbuff);
	233	z->zbuff = NULL;
402dd7ac FR	234	g_free(p->compress_data);
402dd7ac FR	235	p->compress_data = NULL;
7ec2c2b3 JQ	236	}
	237
	238	/**
9db19125	239	* zlib_recv: read the data from the channel into actual pages
7ec2c2b3 JQ	240	*
	241	* Read the compressed buffer, and uncompress it into the actual
	242	* pages.
	243	*
	244	* Returns 0 for success or -1 for error
	245	*
	246	* @p: Params for the channel that we are using
7ec2c2b3 JQ	247	* @errp: pointer to an error
7ec2c2b3 JQ	248	*/
9db19125	249	static int zlib_recv(MultiFDRecvParams p, Error *errp)
7ec2c2b3	250	{
402dd7ac	251	struct zlib_data *z = p->compress_data;
7ec2c2b3 JQ	252	z_stream *zs = &z->zs;
	253	uint32_t in_size = p->next_packet_size;
	254	/* we measure the change of total_out */
	255	uint32_t out_size = zs->total_out;
ddec20f8	256	uint32_t expected_size = p->normal_num * p->page_size;
7ec2c2b3 JQ	257	uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
	258	int ret;
	259	int i;
	260
	261	if (flags != MULTIFD_FLAG_ZLIB) {
04e11404	262	error_setg(errp, "multifd %u: flags received %x flags expected %x",
7ec2c2b3 JQ	263	p->id, flags, MULTIFD_FLAG_ZLIB);
	264	return -1;
	265	}
303e6f54 HX	266
	267	multifd_recv_zero_page_process(p);
	268
	269	if (!p->normal_num) {
	270	assert(in_size == 0);
	271	return 0;
	272	}
	273
7ec2c2b3 JQ	274	ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp);
	275
	276	if (ret != 0) {
	277	return ret;
	278	}
	279
	280	zs->avail_in = in_size;
	281	zs->next_in = z->zbuff;
	282
cf2d4aa8	283	for (i = 0; i < p->normal_num; i++) {
7ec2c2b3 JQ	284	int flush = Z_NO_FLUSH;
	285	unsigned long start = zs->total_out;
	286
5ef7e26b	287	ramblock_recv_bitmap_set_offset(p->block, p->normal[i]);
cf2d4aa8	288	if (i == p->normal_num - 1) {
7ec2c2b3 JQ	289	flush = Z_SYNC_FLUSH;
	290	}
	291
ddec20f8	292	zs->avail_out = p->page_size;
faf60935	293	zs->next_out = p->host + p->normal[i];
7ec2c2b3 JQ	294
	295	/*
	296	* Welcome to inflate semantics
	297	*
	298	* We need to loop while:
	299	* - return is Z_OK
	300	* - there are input available
	301	* - we haven't completed a full page
	302	*/
	303	do {
	304	ret = inflate(zs, flush);
	305	} while (ret == Z_OK && zs->avail_in
ddec20f8 JQ	306	&& (zs->total_out - start) < p->page_size);
ddec20f8 JQ	307	if (ret == Z_OK && (zs->total_out - start) < p->page_size) {
04e11404	308	error_setg(errp, "multifd %u: inflate generated too few output",
7ec2c2b3 JQ	309	p->id);
	310	return -1;
	311	}
	312	if (ret != Z_OK) {
04e11404	313	error_setg(errp, "multifd %u: inflate returned %d instead of Z_OK",
7ec2c2b3 JQ	314	p->id, ret);
	315	return -1;
	316	}
	317	}
	318	out_size = zs->total_out - out_size;
	319	if (out_size != expected_size) {
04e11404	320	error_setg(errp, "multifd %u: packet size received %u size expected %u",
7ec2c2b3 JQ	321	p->id, out_size, expected_size);
	322	return -1;
	323	}
303e6f54	324
7ec2c2b3 JQ	325	return 0;
	326	}
	327
	328	static MultiFDMethods multifd_zlib_ops = {
	329	.send_setup = zlib_send_setup,
	330	.send_cleanup = zlib_send_cleanup,
	331	.send_prepare = zlib_send_prepare,
7ec2c2b3 JQ	332	.recv_setup = zlib_recv_setup,
7ec2c2b3 JQ	333	.recv_cleanup = zlib_recv_cleanup,
9db19125	334	.recv = zlib_recv
7ec2c2b3 JQ	335	};
	336
	337	static void multifd_zlib_register(void)
	338	{
	339	multifd_register_ops(MULTIFD_COMPRESSION_ZLIB, &multifd_zlib_ops);
	340	}
	341
	342	migration_init(multifd_zlib_register);