[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->data = z;
    return 0;

err_free_zbuff:
    g_free(z->zbuff);
err_deflate_end:
    deflateEnd(&z->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->data;

    deflateEnd(&z->zs);
    g_free(z->zbuff);
    z->zbuff = NULL;
    g_free(z->buf);
    z->buf = NULL;
    g_free(p->data);
    p->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)
{
    struct zlib_data *z = p->data;
    z_stream *zs = &z->zs;
    uint32_t out_size = 0;
    int ret;
    uint32_t i;

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

        if (i == p->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 + p->normal[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;
    p->flags |= MULTIFD_FLAG_ZLIB;

    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->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->data;

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

/**
 * zlib_recv_pages: 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_pages(MultiFDRecvParams *p, Error **errp)
{
    struct zlib_data *z = p->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;
    }
    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;

        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_pages = zlib_recv_pages
};

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 JQ	71	}
	72	p->data = z;
	73	return 0;
007e179e IL	74
	75	err_free_zbuff:
	76	g_free(z->zbuff);
	77	err_deflate_end:
	78	deflateEnd(&z->zs);
	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	{
	95	struct zlib_data *z = p->data;
	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;
7ec2c2b3 JQ	102	g_free(p->data);
	103	p->data = NULL;
	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	{
7ec2c2b3 JQ	119	struct zlib_data *z = p->data;
	120	z_stream *zs = &z->zs;
	121	uint32_t out_size = 0;
	122	int ret;
	123	uint32_t i;
	124
815956f0	125	for (i = 0; i < p->normal_num; i++) {
7ec2c2b3 JQ	126	uint32_t available = z->zbuff_len - out_size;
	127	int flush = Z_NO_FLUSH;
	128
815956f0	129	if (i == p->normal_num - 1) {
7ec2c2b3 JQ	130	flush = Z_SYNC_FLUSH;
	131	}
	132
007e179e IL	133	/*
	134	* Since the VM might be running, the page may be changing concurrently
	135	* with compression. zlib does not guarantee that this is safe,
	136	* therefore copy the page before calling deflate().
	137	*/
ddec20f8 JQ	138	memcpy(z->buf, p->pages->block->host + p->normal[i], p->page_size);
ddec20f8 JQ	139	zs->avail_in = p->page_size;
007e179e	140	zs->next_in = z->buf;
7ec2c2b3 JQ	141
	142	zs->avail_out = available;
	143	zs->next_out = z->zbuff + out_size;
	144
	145	/*
	146	* Welcome to deflate semantics
	147	*
	148	* We need to loop while:
	149	* - return is Z_OK
	150	* - there are stuff to be compressed
	151	* - there are output space free
	152	*/
	153	do {
	154	ret = deflate(zs, flush);
	155	} while (ret == Z_OK && zs->avail_in && zs->avail_out);
	156	if (ret == Z_OK && zs->avail_in) {
04e11404	157	error_setg(errp, "multifd %u: deflate failed to compress all input",
7ec2c2b3 JQ	158	p->id);
	159	return -1;
	160	}
	161	if (ret != Z_OK) {
04e11404	162	error_setg(errp, "multifd %u: deflate returned %d instead of Z_OK",
7ec2c2b3 JQ	163	p->id, ret);
	164	return -1;
	165	}
	166	out_size += available - zs->avail_out;
	167	}
48a4a44c JQ	168	p->iov[p->iovs_num].iov_base = z->zbuff;
	169	p->iov[p->iovs_num].iov_len = out_size;
	170	p->iovs_num++;
7ec2c2b3 JQ	171	p->next_packet_size = out_size;
	172	p->flags \|= MULTIFD_FLAG_ZLIB;
	173
	174	return 0;
	175	}
	176
7ec2c2b3 JQ	177	/**
	178	* zlib_recv_setup: setup receive side
	179	*
	180	* Create the compressed channel and buffer.
	181	*
	182	* Returns 0 for success or -1 for error
	183	*
	184	* @p: Params for the channel that we are using
	185	* @errp: pointer to an error
	186	*/
	187	static int zlib_recv_setup(MultiFDRecvParams p, Error *errp)
	188	{
b21e2380	189	struct zlib_data *z = g_new0(struct zlib_data, 1);
7ec2c2b3 JQ	190	z_stream *zs = &z->zs;
	191
	192	p->data = z;
	193	zs->zalloc = Z_NULL;
	194	zs->zfree = Z_NULL;
	195	zs->opaque = Z_NULL;
	196	zs->avail_in = 0;
	197	zs->next_in = Z_NULL;
	198	if (inflateInit(zs) != Z_OK) {
04e11404	199	error_setg(errp, "multifd %u: inflate init failed", p->id);
7ec2c2b3 JQ	200	return -1;
7ec2c2b3 JQ	201	}
47a17824 JQ	202	/* To be safe, we reserve twice the size of the packet */
47a17824 JQ	203	z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
7ec2c2b3 JQ	204	z->zbuff = g_try_malloc(z->zbuff_len);
	205	if (!z->zbuff) {
	206	inflateEnd(zs);
04e11404	207	error_setg(errp, "multifd %u: out of memory for zbuff", p->id);
7ec2c2b3 JQ	208	return -1;
	209	}
	210	return 0;
	211	}
	212
	213	/**
	214	* zlib_recv_cleanup: setup receive side
	215	*
	216	* For no compression this function does nothing.
	217	*
	218	* @p: Params for the channel that we are using
	219	*/
	220	static void zlib_recv_cleanup(MultiFDRecvParams *p)
	221	{
	222	struct zlib_data *z = p->data;
	223
	224	inflateEnd(&z->zs);
	225	g_free(z->zbuff);
	226	z->zbuff = NULL;
	227	g_free(p->data);
	228	p->data = NULL;
	229	}
	230
	231	/**
	232	* zlib_recv_pages: read the data from the channel into actual pages
	233	*
	234	* Read the compressed buffer, and uncompress it into the actual
	235	* pages.
	236	*
	237	* Returns 0 for success or -1 for error
	238	*
	239	* @p: Params for the channel that we are using
7ec2c2b3 JQ	240	* @errp: pointer to an error
7ec2c2b3 JQ	241	*/
40a4bfe9	242	static int zlib_recv_pages(MultiFDRecvParams p, Error *errp)
7ec2c2b3 JQ	243	{
	244	struct zlib_data *z = p->data;
	245	z_stream *zs = &z->zs;
	246	uint32_t in_size = p->next_packet_size;
	247	/* we measure the change of total_out */
	248	uint32_t out_size = zs->total_out;
ddec20f8	249	uint32_t expected_size = p->normal_num * p->page_size;
7ec2c2b3 JQ	250	uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
	251	int ret;
	252	int i;
	253
	254	if (flags != MULTIFD_FLAG_ZLIB) {
04e11404	255	error_setg(errp, "multifd %u: flags received %x flags expected %x",
7ec2c2b3 JQ	256	p->id, flags, MULTIFD_FLAG_ZLIB);
	257	return -1;
	258	}
	259	ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp);
	260
	261	if (ret != 0) {
	262	return ret;
	263	}
	264
	265	zs->avail_in = in_size;
	266	zs->next_in = z->zbuff;
	267
cf2d4aa8	268	for (i = 0; i < p->normal_num; i++) {
7ec2c2b3 JQ	269	int flush = Z_NO_FLUSH;
	270	unsigned long start = zs->total_out;
	271
cf2d4aa8	272	if (i == p->normal_num - 1) {
7ec2c2b3 JQ	273	flush = Z_SYNC_FLUSH;
	274	}
	275
ddec20f8	276	zs->avail_out = p->page_size;
faf60935	277	zs->next_out = p->host + p->normal[i];
7ec2c2b3 JQ	278
	279	/*
	280	* Welcome to inflate semantics
	281	*
	282	* We need to loop while:
	283	* - return is Z_OK
	284	* - there are input available
	285	* - we haven't completed a full page
	286	*/
	287	do {
	288	ret = inflate(zs, flush);
	289	} while (ret == Z_OK && zs->avail_in
ddec20f8 JQ	290	&& (zs->total_out - start) < p->page_size);
ddec20f8 JQ	291	if (ret == Z_OK && (zs->total_out - start) < p->page_size) {
04e11404	292	error_setg(errp, "multifd %u: inflate generated too few output",
7ec2c2b3 JQ	293	p->id);
	294	return -1;
	295	}
	296	if (ret != Z_OK) {
04e11404	297	error_setg(errp, "multifd %u: inflate returned %d instead of Z_OK",
7ec2c2b3 JQ	298	p->id, ret);
	299	return -1;
	300	}
	301	}
	302	out_size = zs->total_out - out_size;
	303	if (out_size != expected_size) {
04e11404	304	error_setg(errp, "multifd %u: packet size received %u size expected %u",
7ec2c2b3 JQ	305	p->id, out_size, expected_size);
	306	return -1;
	307	}
	308	return 0;
	309	}
	310
	311	static MultiFDMethods multifd_zlib_ops = {
	312	.send_setup = zlib_send_setup,
	313	.send_cleanup = zlib_send_cleanup,
	314	.send_prepare = zlib_send_prepare,
7ec2c2b3 JQ	315	.recv_setup = zlib_recv_setup,
	316	.recv_cleanup = zlib_recv_cleanup,
	317	.recv_pages = zlib_recv_pages
	318	};
	319
	320	static void multifd_zlib_register(void)
	321	{
	322	multifd_register_ops(MULTIFD_COMPRESSION_ZLIB, &multifd_zlib_ops);
	323	}
	324
	325	migration_init(multifd_zlib_register);