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 static int multifd_zlib_send_setup(MultiFDSendParams *p, Error **errp)
  38 {
  39     struct zlib_data *z = g_new0(struct zlib_data, 1);
  40     z_stream *zs = &z->zs;
  41     const char *err_msg;
  42
  43     zs->zalloc = Z_NULL;
  44     zs->zfree = Z_NULL;
  45     zs->opaque = Z_NULL;
  46     if (deflateInit(zs, migrate_multifd_zlib_level()) != Z_OK) {
  47         err_msg = "deflate init failed";
  48         goto err_free_z;
  49     }
  50     /* This is the maximum size of the compressed buffer */
  51     z->zbuff_len = compressBound(MULTIFD_PACKET_SIZE);
  52     z->zbuff = g_try_malloc(z->zbuff_len);
  53     if (!z->zbuff) {
  54         err_msg = "out of memory for zbuff";
  55         goto err_deflate_end;
  56     }
  57     z->buf = g_try_malloc(qemu_target_page_size());
  58     if (!z->buf) {
  59         err_msg = "out of memory for buf";
  60         goto err_free_zbuff;
  61     }
  62     p->compress_data = z;
  63
  64     /* Needs 2 IOVs, one for packet header and one for compressed data */
  65     p->iov = g_new0(struct iovec, 2);
  66
  67     return 0;
  68
  69 err_free_zbuff:
  70     g_free(z->zbuff);
  71 err_deflate_end:
  72     deflateEnd(zs);
  73 err_free_z:
  74     g_free(z);
  75     error_setg(errp, "multifd %u: %s", p->id, err_msg);
  76     return -1;
  77 }
  78
  79 static void multifd_zlib_send_cleanup(MultiFDSendParams *p, Error **errp)
  80 {
  81     struct zlib_data *z = p->compress_data;
  82
  83     deflateEnd(&z->zs);
  84     g_free(z->zbuff);
  85     z->zbuff = NULL;
  86     g_free(z->buf);
  87     z->buf = NULL;
  88     g_free(p->compress_data);
  89     p->compress_data = NULL;
  90
  91     g_free(p->iov);
  92     p->iov = NULL;
  93 }
  94
  95 static int multifd_zlib_send_prepare(MultiFDSendParams *p, Error **errp)
  96 {
  97     MultiFDPages_t *pages = &p->data->u.ram;
  98     struct zlib_data *z = p->compress_data;
  99     z_stream *zs = &z->zs;
 100     uint32_t out_size = 0;
 101     uint32_t page_size = multifd_ram_page_size();
 102     int ret;
 103     uint32_t i;
 104
 105     if (!multifd_send_prepare_common(p)) {
 106         goto out;
 107     }
 108
 109     for (i = 0; i < pages->normal_num; i++) {
 110         uint32_t available = z->zbuff_len - out_size;
 111         int flush = Z_NO_FLUSH;
 112
 113         if (i == pages->normal_num - 1) {
 114             flush = Z_SYNC_FLUSH;
 115         }
 116
 117         /*
 118          * Since the VM might be running, the page may be changing concurrently
 119          * with compression. zlib does not guarantee that this is safe,
 120          * therefore copy the page before calling deflate().
 121          */
 122         memcpy(z->buf, pages->block->host + pages->offset[i], page_size);
 123         zs->avail_in = page_size;
 124         zs->next_in = z->buf;
 125
 126         zs->avail_out = available;
 127         zs->next_out = z->zbuff + out_size;
 128
 129         /*
 130          * Welcome to deflate semantics
 131          *
 132          * We need to loop while:
 133          * - return is Z_OK
 134          * - there are stuff to be compressed
 135          * - there are output space free
 136          */
 137         do {
 138             ret = deflate(zs, flush);
 139         } while (ret == Z_OK && zs->avail_in && zs->avail_out);
 140         if (ret == Z_OK && zs->avail_in) {
 141             error_setg(errp, "multifd %u: deflate failed to compress all input",
 142                        p->id);
 143             return -1;
 144         }
 145         if (ret != Z_OK) {
 146             error_setg(errp, "multifd %u: deflate returned %d instead of Z_OK",
 147                        p->id, ret);
 148             return -1;
 149         }
 150         out_size += available - zs->avail_out;
 151     }
 152     p->iov[p->iovs_num].iov_base = z->zbuff;
 153     p->iov[p->iovs_num].iov_len = out_size;
 154     p->iovs_num++;
 155     p->next_packet_size = out_size;
 156
 157 out:
 158     p->flags |= MULTIFD_FLAG_ZLIB;
 159     multifd_send_fill_packet(p);
 160     return 0;
 161 }
 162
 163 static int multifd_zlib_recv_setup(MultiFDRecvParams *p, Error **errp)
 164 {
 165     struct zlib_data *z = g_new0(struct zlib_data, 1);
 166     z_stream *zs = &z->zs;
 167
 168     p->compress_data = z;
 169     zs->zalloc = Z_NULL;
 170     zs->zfree = Z_NULL;
 171     zs->opaque = Z_NULL;
 172     zs->avail_in = 0;
 173     zs->next_in = Z_NULL;
 174     if (inflateInit(zs) != Z_OK) {
 175         error_setg(errp, "multifd %u: inflate init failed", p->id);
 176         return -1;
 177     }
 178     /* To be safe, we reserve twice the size of the packet */
 179     z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
 180     z->zbuff = g_try_malloc(z->zbuff_len);
 181     if (!z->zbuff) {
 182         inflateEnd(zs);
 183         error_setg(errp, "multifd %u: out of memory for zbuff", p->id);
 184         return -1;
 185     }
 186     return 0;
 187 }
 188
 189 static void multifd_zlib_recv_cleanup(MultiFDRecvParams *p)
 190 {
 191     struct zlib_data *z = p->compress_data;
 192
 193     inflateEnd(&z->zs);
 194     g_free(z->zbuff);
 195     z->zbuff = NULL;
 196     g_free(p->compress_data);
 197     p->compress_data = NULL;
 198 }
 199
 200 static int multifd_zlib_recv(MultiFDRecvParams *p, Error **errp)
 201 {
 202     struct zlib_data *z = p->compress_data;
 203     z_stream *zs = &z->zs;
 204     uint32_t in_size = p->next_packet_size;
 205     /* we measure the change of total_out */
 206     uint32_t out_size = zs->total_out;
 207     uint32_t page_size = multifd_ram_page_size();
 208     uint32_t expected_size = p->normal_num * page_size;
 209     uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
 210     int ret;
 211     int i;
 212
 213     if (flags != MULTIFD_FLAG_ZLIB) {
 214         error_setg(errp, "multifd %u: flags received %x flags expected %x",
 215                    p->id, flags, MULTIFD_FLAG_ZLIB);
 216         return -1;
 217     }
 218
 219     multifd_recv_zero_page_process(p);
 220
 221     if (!p->normal_num) {
 222         assert(in_size == 0);
 223         return 0;
 224     }
 225
 226     ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp);
 227
 228     if (ret != 0) {
 229         return ret;
 230     }
 231
 232     zs->avail_in = in_size;
 233     zs->next_in = z->zbuff;
 234
 235     for (i = 0; i < p->normal_num; i++) {
 236         int flush = Z_NO_FLUSH;
 237         unsigned long start = zs->total_out;
 238
 239         ramblock_recv_bitmap_set_offset(p->block, p->normal[i]);
 240         if (i == p->normal_num - 1) {
 241             flush = Z_SYNC_FLUSH;
 242         }
 243
 244         zs->avail_out = page_size;
 245         zs->next_out = p->host + p->normal[i];
 246
 247         /*
 248          * Welcome to inflate semantics
 249          *
 250          * We need to loop while:
 251          * - return is Z_OK
 252          * - there are input available
 253          * - we haven't completed a full page
 254          */
 255         do {
 256             ret = inflate(zs, flush);
 257         } while (ret == Z_OK && zs->avail_in
 258                              && (zs->total_out - start) < page_size);
 259         if (ret == Z_OK && (zs->total_out - start) < page_size) {
 260             error_setg(errp, "multifd %u: inflate generated too few output",
 261                        p->id);
 262             return -1;
 263         }
 264         if (ret != Z_OK) {
 265             error_setg(errp, "multifd %u: inflate returned %d instead of Z_OK",
 266                        p->id, ret);
 267             return -1;
 268         }
 269     }
 270     out_size = zs->total_out - out_size;
 271     if (out_size != expected_size) {
 272         error_setg(errp, "multifd %u: packet size received %u size expected %u",
 273                    p->id, out_size, expected_size);
 274         return -1;
 275     }
 276
 277     return 0;
 278 }
 279
 280 static const MultiFDMethods multifd_zlib_ops = {
 281     .send_setup = multifd_zlib_send_setup,
 282     .send_cleanup = multifd_zlib_send_cleanup,
 283     .send_prepare = multifd_zlib_send_prepare,
 284     .recv_setup = multifd_zlib_recv_setup,
 285     .recv_cleanup = multifd_zlib_recv_cleanup,
 286     .recv = multifd_zlib_recv
 287 };
 288
 289 static void multifd_zlib_register(void)
 290 {
 291     multifd_register_ops(MULTIFD_COMPRESSION_ZLIB, &multifd_zlib_ops);
 292 }
 293
 294 migration_init(multifd_zlib_register);