migration/multifd-zlib.c

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