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