migration/multifd-qatzip.c - qemu - Git at Google

 /*
  * Multifd QATzip compression implementation
  *
  * Copyright (c) Bytedance
  *
  * Authors:
  *  Bryan Zhang <bryan.zhang@bytedance.com>
  *  Hao Xiang <hao.xiang@bytedance.com>
  *  Yichen Wang <yichen.wang@bytedance.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 "exec/ramblock.h"
 #include "qapi/error.h"
 #include "qemu/error-report.h"
 #include "qapi/qapi-types-migration.h"
 #include "options.h"
 #include "multifd.h"
 #include <qatzip.h>

 typedef struct {
     /*
      * Unique session for use with QATzip API
      */
     QzSession_T sess;

     /*
      * For compression: Buffer for pages to compress
      * For decompression: Buffer for data to decompress
      */
     uint8_t *in_buf;
     uint32_t in_len;

     /*
      * For compression: Output buffer of compressed data
      * For decompression: Output buffer of decompressed data
      */
     uint8_t *out_buf;
     uint32_t out_len;
 } QatzipData;

 /**
  * qatzip_send_setup: Set up QATzip session and private buffers.
  *
  * @param p    Multifd channel params
  * @param errp Pointer to error, which will be set in case of error
  * @return     0 on success, -1 on error (and *errp will be set)
  */
 static int qatzip_send_setup(MultiFDSendParams *p, Error **errp)
 {
     QatzipData *q;
     QzSessionParamsDeflate_T params;
     const char *err_msg;
     int ret;

     q = g_new0(QatzipData, 1);
     p->compress_data = q;
     /* We need one extra place for the packet header */
     p->iov = g_new0(struct iovec, 2);

     /*
      * Initialize QAT device with software fallback by default. This allows
      * QATzip to use CPU path when QAT hardware reaches maximum throughput.
      */
     ret = qzInit(&q->sess, true);
     if (ret != QZ_OK && ret != QZ_DUPLICATE) {
         err_msg = "qzInit failed";
         goto err;
     }

     ret = qzGetDefaultsDeflate(&params);
     if (ret != QZ_OK) {
         err_msg = "qzGetDefaultsDeflate failed";
         goto err;
     }

     /* Make sure to use configured QATzip compression level. */
     params.common_params.comp_lvl = migrate_multifd_qatzip_level();
     ret = qzSetupSessionDeflate(&q->sess, &params);
     if (ret != QZ_OK && ret != QZ_DUPLICATE) {
         err_msg = "qzSetupSessionDeflate failed";
         goto err;
     }

     if (MULTIFD_PACKET_SIZE > UINT32_MAX) {
         err_msg = "packet size too large for QAT";
         goto err;
     }

     q->in_len = MULTIFD_PACKET_SIZE;
     /*
      * PINNED_MEM is an enum from qatzip headers, which means to use
      * kzalloc_node() to allocate memory for QAT DMA purposes. When QAT device
      * is not available or software fallback is used, the malloc flag needs to
      * be set as COMMON_MEM.
      */
     q->in_buf = qzMalloc(q->in_len, 0, PINNED_MEM);
     if (!q->in_buf) {
         q->in_buf = qzMalloc(q->in_len, 0, COMMON_MEM);
         if (!q->in_buf) {
             err_msg = "qzMalloc failed";
             goto err;
         }
     }

     q->out_len = qzMaxCompressedLength(MULTIFD_PACKET_SIZE, &q->sess);
     q->out_buf = qzMalloc(q->out_len, 0, PINNED_MEM);
     if (!q->out_buf) {
         q->out_buf = qzMalloc(q->out_len, 0, COMMON_MEM);
         if (!q->out_buf) {
             err_msg = "qzMalloc failed";
             goto err;
         }
     }

     return 0;

 err:
     error_setg(errp, "multifd %u: [sender] %s", p->id, err_msg);
     return -1;
 }

 /**
  * qatzip_send_cleanup: Tear down QATzip session and release private buffers.
  *
  * @param p    Multifd channel params
  * @param errp Pointer to error, which will be set in case of error
  * @return     None
  */
 static void qatzip_send_cleanup(MultiFDSendParams *p, Error **errp)
 {
     QatzipData *q = p->compress_data;

     if (q) {
         if (q->in_buf) {
             qzFree(q->in_buf);
         }
         if (q->out_buf) {
             qzFree(q->out_buf);
         }
         (void)qzTeardownSession(&q->sess);
         (void)qzClose(&q->sess);
         g_free(q);
     }

     g_free(p->iov);
     p->iov = NULL;
     p->compress_data = NULL;
 }

 /**
  * qatzip_send_prepare: Compress pages and update IO channel info.
  *
  * @param p    Multifd channel params
  * @param errp Pointer to error, which will be set in case of error
  * @return     0 on success, -1 on error (and *errp will be set)
  */
 static int qatzip_send_prepare(MultiFDSendParams *p, Error **errp)
 {
     MultiFDPages_t *pages = p->pages;
     QatzipData *q = p->compress_data;
     int ret;
     unsigned int in_len, out_len;

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

     /*
      * Unlike other multifd compression implementations, we use a non-streaming
      * API and place all the data into one buffer, rather than sending each
      * page to the compression API at a time. Based on initial benchmarks, the
      * non-streaming API outperforms the streaming API. Plus, the logic in QEMU
      * is friendly to using the non-streaming API anyway. If either of these
      * statements becomes no longer true, we can revisit adding a streaming
      * implementation.
      */
     for (int i = 0; i < pages->normal_num; i++) {
         memcpy(q->in_buf + (i * p->page_size),
                pages->block->host + pages->offset[i],
                p->page_size);
     }

     in_len = pages->normal_num * p->page_size;
     if (in_len > q->in_len) {
         error_setg(errp, "multifd %u: unexpectedly large input", p->id);
         return -1;
     }
     out_len = q->out_len;

     ret = qzCompress(&q->sess, q->in_buf, &in_len, q->out_buf, &out_len, 1);
     if (ret != QZ_OK) {
         error_setg(errp, "multifd %u: QATzip returned %d instead of QZ_OK",
                    p->id, ret);
         return -1;
     }
     if (in_len != pages->normal_num * p->page_size) {
         error_setg(errp, "multifd %u: QATzip failed to compress all input",
                    p->id);
         return -1;
     }

     p->iov[p->iovs_num].iov_base = q->out_buf;
     p->iov[p->iovs_num].iov_len = out_len;
     p->iovs_num++;
     p->next_packet_size = out_len;

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

 /**
  * qatzip_recv_setup: Set up QATzip session and allocate private buffers.
  *
  * @param p    Multifd channel params
  * @param errp Pointer to error, which will be set in case of error
  * @return     0 on success, -1 on error (and *errp will be set)
  */
 static int qatzip_recv_setup(MultiFDRecvParams *p, Error **errp)
 {
     QatzipData *q;
     QzSessionParamsDeflate_T params;
     const char *err_msg;
     int ret;

     q = g_new0(QatzipData, 1);
     p->compress_data = q;

     /*
      * Initialize QAT device with software fallback by default. This allows
      * QATzip to use CPU path when QAT hardware reaches maximum throughput.
      */
     ret = qzInit(&q->sess, true);
     if (ret != QZ_OK && ret != QZ_DUPLICATE) {
         err_msg = "qzInit failed";
         goto err;
     }

     ret = qzGetDefaultsDeflate(&params);
     if (ret != QZ_OK) {
         err_msg = "qzGetDefaultsDeflate failed";
         goto err;
     }

     ret = qzSetupSessionDeflate(&q->sess, &params);
     if (ret != QZ_OK && ret != QZ_DUPLICATE) {
         err_msg = "qzSetupSessionDeflate failed";
         goto err;
     }

     /*
      * Reserve extra spaces for the incoming packets. Current implementation
      * doesn't send uncompressed pages in case the compression gets too big.
      */
     q->in_len = MULTIFD_PACKET_SIZE * 2;
     /*
      * PINNED_MEM is an enum from qatzip headers, which means to use
      * kzalloc_node() to allocate memory for QAT DMA purposes. When QAT device
      * is not available or software fallback is used, the malloc flag needs to
      * be set as COMMON_MEM.
      */
     q->in_buf = qzMalloc(q->in_len, 0, PINNED_MEM);
     if (!q->in_buf) {
         q->in_buf = qzMalloc(q->in_len, 0, COMMON_MEM);
         if (!q->in_buf) {
             err_msg = "qzMalloc failed";
             goto err;
         }
     }

     q->out_len = MULTIFD_PACKET_SIZE;
     q->out_buf = qzMalloc(q->out_len, 0, PINNED_MEM);
     if (!q->out_buf) {
         q->out_buf = qzMalloc(q->out_len, 0, COMMON_MEM);
         if (!q->out_buf) {
             err_msg = "qzMalloc failed";
             goto err;
         }
     }

     return 0;

 err:
     error_setg(errp, "multifd %u: [receiver] %s", p->id, err_msg);
     return -1;
 }

 /**
  * qatzip_recv_cleanup: Tear down QATzip session and release private buffers.
  *
  * @param p    Multifd channel params
  * @return     None
  */
 static void qatzip_recv_cleanup(MultiFDRecvParams *p)
 {
     QatzipData *q = p->compress_data;

     if (q) {
         if (q->in_buf) {
             qzFree(q->in_buf);
         }
         if (q->out_buf) {
             qzFree(q->out_buf);
         }
         (void)qzTeardownSession(&q->sess);
         (void)qzClose(&q->sess);
         g_free(q);
     }
     p->compress_data = NULL;
 }


 /**
  * qatzip_recv: Decompress pages and copy them to the appropriate
  * locations.
  *
  * @param p    Multifd channel params
  * @param errp Pointer to error, which will be set in case of error
  * @return     0 on success, -1 on error (and *errp will be set)
  */
 static int qatzip_recv(MultiFDRecvParams *p, Error **errp)
 {
     QatzipData *q = p->compress_data;
     int ret;
     unsigned int in_len, out_len;
     uint32_t in_size = p->next_packet_size;
     uint32_t expected_size = p->normal_num * p->page_size;
     uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;

     if (in_size > q->in_len) {
         error_setg(errp, "multifd %u: received unexpectedly large packet",
                    p->id);
         return -1;
     }

     if (flags != MULTIFD_FLAG_QATZIP) {
         error_setg(errp, "multifd %u: flags received %x flags expected %x",
                    p->id, flags, MULTIFD_FLAG_QATZIP);
         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 *)q->in_buf, in_size, errp);
     if (ret != 0) {
         return ret;
     }

     in_len = in_size;
     out_len = q->out_len;
     ret = qzDecompress(&q->sess, q->in_buf, &in_len, q->out_buf, &out_len);
     if (ret != QZ_OK) {
         error_setg(errp, "multifd %u: qzDecompress failed", p->id);
         return -1;
     }
     if (out_len != expected_size) {
         error_setg(errp, "multifd %u: packet size received %u size expected %u",
                    p->id, out_len, expected_size);
         return -1;
     }

     /* Copy each page to its appropriate location. */
     for (int i = 0; i < p->normal_num; i++) {
         memcpy(p->host + p->normal[i],
                q->out_buf + p->page_size * i,
                p->page_size);
     }
     return 0;
 }

 static MultiFDMethods multifd_qatzip_ops = {
     .send_setup = qatzip_send_setup,
     .send_cleanup = qatzip_send_cleanup,
     .send_prepare = qatzip_send_prepare,
     .recv_setup = qatzip_recv_setup,
     .recv_cleanup = qatzip_recv_cleanup,
     .recv = qatzip_recv
 };

 static void multifd_qatzip_register(void)
 {
     multifd_register_ops(MULTIFD_COMPRESSION_QATZIP, &multifd_qatzip_ops);
 }

 migration_init(multifd_qatzip_register);
	/*
	* Multifd QATzip compression implementation
	*
	* Copyright (c) Bytedance
	*
	* Authors:
	* Bryan Zhang <bryan.zhang@bytedance.com>
	* Hao Xiang <hao.xiang@bytedance.com>
	* Yichen Wang <yichen.wang@bytedance.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 "exec/ramblock.h"
	#include "qapi/error.h"
	#include "qemu/error-report.h"
	#include "qapi/qapi-types-migration.h"
	#include "options.h"
	#include "multifd.h"
	#include <qatzip.h>

	typedef struct {
	/*
	* Unique session for use with QATzip API
	*/
	QzSession_T sess;

	/*
	* For compression: Buffer for pages to compress
	* For decompression: Buffer for data to decompress
	*/
	uint8_t *in_buf;
	uint32_t in_len;

	/*
	* For compression: Output buffer of compressed data
	* For decompression: Output buffer of decompressed data
	*/
	uint8_t *out_buf;
	uint32_t out_len;
	} QatzipData;

	/**
	* qatzip_send_setup: Set up QATzip session and private buffers.
	*
	* @param p Multifd channel params
	* @param errp Pointer to error, which will be set in case of error
	* @return 0 on success, -1 on error (and *errp will be set)
	*/
	static int qatzip_send_setup(MultiFDSendParams p, Error *errp)
	{
	QatzipData *q;
	QzSessionParamsDeflate_T params;
	const char *err_msg;
	int ret;

	q = g_new0(QatzipData, 1);
	p->compress_data = q;
	/* We need one extra place for the packet header */
	p->iov = g_new0(struct iovec, 2);

	/*
	* Initialize QAT device with software fallback by default. This allows
	* QATzip to use CPU path when QAT hardware reaches maximum throughput.
	*/
	ret = qzInit(&q->sess, true);
	if (ret != QZ_OK && ret != QZ_DUPLICATE) {
	err_msg = "qzInit failed";
	goto err;
	}

	ret = qzGetDefaultsDeflate(&params);
	if (ret != QZ_OK) {
	err_msg = "qzGetDefaultsDeflate failed";
	goto err;
	}

	/* Make sure to use configured QATzip compression level. */
	params.common_params.comp_lvl = migrate_multifd_qatzip_level();
	ret = qzSetupSessionDeflate(&q->sess, &params);
	if (ret != QZ_OK && ret != QZ_DUPLICATE) {
	err_msg = "qzSetupSessionDeflate failed";
	goto err;
	}

	if (MULTIFD_PACKET_SIZE > UINT32_MAX) {
	err_msg = "packet size too large for QAT";
	goto err;
	}

	q->in_len = MULTIFD_PACKET_SIZE;
	/*
	* PINNED_MEM is an enum from qatzip headers, which means to use
	* kzalloc_node() to allocate memory for QAT DMA purposes. When QAT device
	* is not available or software fallback is used, the malloc flag needs to
	* be set as COMMON_MEM.
	*/
	q->in_buf = qzMalloc(q->in_len, 0, PINNED_MEM);
	if (!q->in_buf) {
	q->in_buf = qzMalloc(q->in_len, 0, COMMON_MEM);
	if (!q->in_buf) {
	err_msg = "qzMalloc failed";
	goto err;
	}
	}

	q->out_len = qzMaxCompressedLength(MULTIFD_PACKET_SIZE, &q->sess);
	q->out_buf = qzMalloc(q->out_len, 0, PINNED_MEM);
	if (!q->out_buf) {
	q->out_buf = qzMalloc(q->out_len, 0, COMMON_MEM);
	if (!q->out_buf) {
	err_msg = "qzMalloc failed";
	goto err;
	}
	}

	return 0;

	err:
	error_setg(errp, "multifd %u: [sender] %s", p->id, err_msg);
	return -1;
	}

	/**
	* qatzip_send_cleanup: Tear down QATzip session and release private buffers.
	*
	* @param p Multifd channel params
	* @param errp Pointer to error, which will be set in case of error
	* @return None
	*/
	static void qatzip_send_cleanup(MultiFDSendParams p, Error *errp)
	{
	QatzipData *q = p->compress_data;

	if (q) {
	if (q->in_buf) {
	qzFree(q->in_buf);
	}
	if (q->out_buf) {
	qzFree(q->out_buf);
	}
	(void)qzTeardownSession(&q->sess);
	(void)qzClose(&q->sess);
	g_free(q);
	}

	g_free(p->iov);
	p->iov = NULL;
	p->compress_data = NULL;
	}

	/**
	* qatzip_send_prepare: Compress pages and update IO channel info.
	*
	* @param p Multifd channel params
	* @param errp Pointer to error, which will be set in case of error
	* @return 0 on success, -1 on error (and *errp will be set)
	*/
	static int qatzip_send_prepare(MultiFDSendParams p, Error *errp)
	{
	MultiFDPages_t *pages = p->pages;
	QatzipData *q = p->compress_data;
	int ret;
	unsigned int in_len, out_len;

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

	/*
	* Unlike other multifd compression implementations, we use a non-streaming
	* API and place all the data into one buffer, rather than sending each
	* page to the compression API at a time. Based on initial benchmarks, the
	* non-streaming API outperforms the streaming API. Plus, the logic in QEMU
	* is friendly to using the non-streaming API anyway. If either of these
	* statements becomes no longer true, we can revisit adding a streaming
	* implementation.
	*/
	for (int i = 0; i < pages->normal_num; i++) {
	memcpy(q->in_buf + (i * p->page_size),
	pages->block->host + pages->offset[i],
	p->page_size);
	}

	in_len = pages->normal_num * p->page_size;
	if (in_len > q->in_len) {
	error_setg(errp, "multifd %u: unexpectedly large input", p->id);
	return -1;
	}
	out_len = q->out_len;

	ret = qzCompress(&q->sess, q->in_buf, &in_len, q->out_buf, &out_len, 1);
	if (ret != QZ_OK) {
	error_setg(errp, "multifd %u: QATzip returned %d instead of QZ_OK",
	p->id, ret);
	return -1;
	}
	if (in_len != pages->normal_num * p->page_size) {
	error_setg(errp, "multifd %u: QATzip failed to compress all input",
	p->id);
	return -1;
	}

	p->iov[p->iovs_num].iov_base = q->out_buf;
	p->iov[p->iovs_num].iov_len = out_len;
	p->iovs_num++;
	p->next_packet_size = out_len;

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

	/**
	* qatzip_recv_setup: Set up QATzip session and allocate private buffers.
	*
	* @param p Multifd channel params
	* @param errp Pointer to error, which will be set in case of error
	* @return 0 on success, -1 on error (and *errp will be set)
	*/
	static int qatzip_recv_setup(MultiFDRecvParams p, Error *errp)
	{
	QatzipData *q;
	QzSessionParamsDeflate_T params;
	const char *err_msg;
	int ret;

	q = g_new0(QatzipData, 1);
	p->compress_data = q;

	/*
	* Initialize QAT device with software fallback by default. This allows
	* QATzip to use CPU path when QAT hardware reaches maximum throughput.
	*/
	ret = qzInit(&q->sess, true);
	if (ret != QZ_OK && ret != QZ_DUPLICATE) {
	err_msg = "qzInit failed";
	goto err;
	}

	ret = qzGetDefaultsDeflate(&params);
	if (ret != QZ_OK) {
	err_msg = "qzGetDefaultsDeflate failed";
	goto err;
	}

	ret = qzSetupSessionDeflate(&q->sess, &params);
	if (ret != QZ_OK && ret != QZ_DUPLICATE) {
	err_msg = "qzSetupSessionDeflate failed";
	goto err;
	}

	/*
	* Reserve extra spaces for the incoming packets. Current implementation
	* doesn't send uncompressed pages in case the compression gets too big.
	*/
	q->in_len = MULTIFD_PACKET_SIZE * 2;
	/*
	* PINNED_MEM is an enum from qatzip headers, which means to use
	* kzalloc_node() to allocate memory for QAT DMA purposes. When QAT device
	* is not available or software fallback is used, the malloc flag needs to
	* be set as COMMON_MEM.
	*/
	q->in_buf = qzMalloc(q->in_len, 0, PINNED_MEM);
	if (!q->in_buf) {
	q->in_buf = qzMalloc(q->in_len, 0, COMMON_MEM);
	if (!q->in_buf) {
	err_msg = "qzMalloc failed";
	goto err;
	}
	}

	q->out_len = MULTIFD_PACKET_SIZE;
	q->out_buf = qzMalloc(q->out_len, 0, PINNED_MEM);
	if (!q->out_buf) {
	q->out_buf = qzMalloc(q->out_len, 0, COMMON_MEM);
	if (!q->out_buf) {
	err_msg = "qzMalloc failed";
	goto err;
	}
	}

	return 0;

	err:
	error_setg(errp, "multifd %u: [receiver] %s", p->id, err_msg);
	return -1;
	}

	/**
	* qatzip_recv_cleanup: Tear down QATzip session and release private buffers.
	*
	* @param p Multifd channel params
	* @return None
	*/
	static void qatzip_recv_cleanup(MultiFDRecvParams *p)
	{
	QatzipData *q = p->compress_data;

	if (q) {
	if (q->in_buf) {
	qzFree(q->in_buf);
	}
	if (q->out_buf) {
	qzFree(q->out_buf);
	}
	(void)qzTeardownSession(&q->sess);
	(void)qzClose(&q->sess);
	g_free(q);
	}
	p->compress_data = NULL;
	}


	/**
	* qatzip_recv: Decompress pages and copy them to the appropriate
	* locations.
	*
	* @param p Multifd channel params
	* @param errp Pointer to error, which will be set in case of error
	* @return 0 on success, -1 on error (and *errp will be set)
	*/
	static int qatzip_recv(MultiFDRecvParams p, Error *errp)
	{
	QatzipData *q = p->compress_data;
	int ret;
	unsigned int in_len, out_len;
	uint32_t in_size = p->next_packet_size;
	uint32_t expected_size = p->normal_num * p->page_size;
	uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;

	if (in_size > q->in_len) {
	error_setg(errp, "multifd %u: received unexpectedly large packet",
	p->id);
	return -1;
	}

	if (flags != MULTIFD_FLAG_QATZIP) {
	error_setg(errp, "multifd %u: flags received %x flags expected %x",
	p->id, flags, MULTIFD_FLAG_QATZIP);
	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 *)q->in_buf, in_size, errp);
	if (ret != 0) {
	return ret;
	}

	in_len = in_size;
	out_len = q->out_len;
	ret = qzDecompress(&q->sess, q->in_buf, &in_len, q->out_buf, &out_len);
	if (ret != QZ_OK) {
	error_setg(errp, "multifd %u: qzDecompress failed", p->id);
	return -1;
	}
	if (out_len != expected_size) {
	error_setg(errp, "multifd %u: packet size received %u size expected %u",
	p->id, out_len, expected_size);
	return -1;
	}

	/* Copy each page to its appropriate location. */
	for (int i = 0; i < p->normal_num; i++) {
	memcpy(p->host + p->normal[i],
	q->out_buf + p->page_size * i,
	p->page_size);
	}
	return 0;
	}

	static MultiFDMethods multifd_qatzip_ops = {
	.send_setup = qatzip_send_setup,
	.send_cleanup = qatzip_send_cleanup,
	.send_prepare = qatzip_send_prepare,
	.recv_setup = qatzip_recv_setup,
	.recv_cleanup = qatzip_recv_cleanup,
	.recv = qatzip_recv
	};

	static void multifd_qatzip_register(void)
	{
	multifd_register_ops(MULTIFD_COMPRESSION_QATZIP, &multifd_qatzip_ops);
	}

	migration_init(multifd_qatzip_register);