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