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