blob: 3c787ed87920bbe2dc6deec3acced201d83017ca [file] [log] [blame]
/*
* 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);