migration/multifd-zstd.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 <zstd.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 zstd_data {
  25     /* stream for compression */
  26     ZSTD_CStream *zcs;
  27     /* stream for decompression */
  28     ZSTD_DStream *zds;
  29     /* buffers */
  30     ZSTD_inBuffer in;
  31     ZSTD_outBuffer out;
  32     /* compressed buffer */
  33     uint8_t *zbuff;
  34     /* size of compressed buffer */
  35     uint32_t zbuff_len;
  36 };
  37
  38 /* Multifd zstd compression */
  39
  40 /**
  41  * zstd_send_setup: setup send side
  42  *
  43  * Setup each channel with zstd compression.
  44  *
  45  * Returns 0 for success or -1 for error
  46  *
  47  * @p: Params for the channel that we are using
  48  * @errp: pointer to an error
  49  */
  50 static int zstd_send_setup(MultiFDSendParams *p, Error **errp)
  51 {
  52     struct zstd_data *z = g_new0(struct zstd_data, 1);
  53     int res;
  54
  55     p->compress_data = z;
  56     z->zcs = ZSTD_createCStream();
  57     if (!z->zcs) {
  58         g_free(z);
  59         error_setg(errp, "multifd %u: zstd createCStream failed", p->id);
  60         return -1;
  61     }
  62
  63     res = ZSTD_initCStream(z->zcs, migrate_multifd_zstd_level());
  64     if (ZSTD_isError(res)) {
  65         ZSTD_freeCStream(z->zcs);
  66         g_free(z);
  67         error_setg(errp, "multifd %u: initCStream failed with error %s",
  68                    p->id, ZSTD_getErrorName(res));
  69         return -1;
  70     }
  71     /* This is the maximum size of the compressed buffer */
  72     z->zbuff_len = ZSTD_compressBound(MULTIFD_PACKET_SIZE);
  73     z->zbuff = g_try_malloc(z->zbuff_len);
  74     if (!z->zbuff) {
  75         ZSTD_freeCStream(z->zcs);
  76         g_free(z);
  77         error_setg(errp, "multifd %u: out of memory for zbuff", p->id);
  78         return -1;
  79     }
  80     return 0;
  81 }
  82
  83 /**
  84  * zstd_send_cleanup: cleanup send side
  85  *
  86  * Close the channel and return memory.
  87  *
  88  * @p: Params for the channel that we are using
  89  * @errp: pointer to an error
  90  */
  91 static void zstd_send_cleanup(MultiFDSendParams *p, Error **errp)
  92 {
  93     struct zstd_data *z = p->compress_data;
  94
  95     ZSTD_freeCStream(z->zcs);
  96     z->zcs = NULL;
  97     g_free(z->zbuff);
  98     z->zbuff = NULL;
  99     g_free(p->compress_data);
 100     p->compress_data = NULL;
 101 }
 102
 103 /**
 104  * zstd_send_prepare: prepare date to be able to send
 105  *
 106  * Create a compressed buffer with all the pages that we are going to
 107  * send.
 108  *
 109  * Returns 0 for success or -1 for error
 110  *
 111  * @p: Params for the channel that we are using
 112  * @errp: pointer to an error
 113  */
 114 static int zstd_send_prepare(MultiFDSendParams *p, Error **errp)
 115 {
 116     MultiFDPages_t *pages = p->pages;
 117     struct zstd_data *z = p->compress_data;
 118     int ret;
 119     uint32_t i;
 120
 121     if (!multifd_send_prepare_common(p)) {
 122         goto out;
 123     }
 124
 125     z->out.dst = z->zbuff;
 126     z->out.size = z->zbuff_len;
 127     z->out.pos = 0;
 128
 129     for (i = 0; i < pages->normal_num; i++) {
 130         ZSTD_EndDirective flush = ZSTD_e_continue;
 131
 132         if (i == pages->normal_num - 1) {
 133             flush = ZSTD_e_flush;
 134         }
 135         z->in.src = p->pages->block->host + pages->offset[i];
 136         z->in.size = p->page_size;
 137         z->in.pos = 0;
 138
 139         /*
 140          * Welcome to compressStream2 semantics
 141          *
 142          * We need to loop while:
 143          * - return is > 0
 144          * - there is input available
 145          * - there is output space free
 146          */
 147         do {
 148             ret = ZSTD_compressStream2(z->zcs, &z->out, &z->in, flush);
 149         } while (ret > 0 && (z->in.size - z->in.pos > 0)
 150                          && (z->out.size - z->out.pos > 0));
 151         if (ret > 0 && (z->in.size - z->in.pos > 0)) {
 152             error_setg(errp, "multifd %u: compressStream buffer too small",
 153                        p->id);
 154             return -1;
 155         }
 156         if (ZSTD_isError(ret)) {
 157             error_setg(errp, "multifd %u: compressStream error %s",
 158                        p->id, ZSTD_getErrorName(ret));
 159             return -1;
 160         }
 161     }
 162     p->iov[p->iovs_num].iov_base = z->zbuff;
 163     p->iov[p->iovs_num].iov_len = z->out.pos;
 164     p->iovs_num++;
 165     p->next_packet_size = z->out.pos;
 166
 167 out:
 168     p->flags |= MULTIFD_FLAG_ZSTD;
 169     multifd_send_fill_packet(p);
 170     return 0;
 171 }
 172
 173 /**
 174  * zstd_recv_setup: setup receive side
 175  *
 176  * Create the compressed channel and buffer.
 177  *
 178  * Returns 0 for success or -1 for error
 179  *
 180  * @p: Params for the channel that we are using
 181  * @errp: pointer to an error
 182  */
 183 static int zstd_recv_setup(MultiFDRecvParams *p, Error **errp)
 184 {
 185     struct zstd_data *z = g_new0(struct zstd_data, 1);
 186     int ret;
 187
 188     p->compress_data = z;
 189     z->zds = ZSTD_createDStream();
 190     if (!z->zds) {
 191         g_free(z);
 192         error_setg(errp, "multifd %u: zstd createDStream failed", p->id);
 193         return -1;
 194     }
 195
 196     ret = ZSTD_initDStream(z->zds);
 197     if (ZSTD_isError(ret)) {
 198         ZSTD_freeDStream(z->zds);
 199         g_free(z);
 200         error_setg(errp, "multifd %u: initDStream failed with error %s",
 201                    p->id, ZSTD_getErrorName(ret));
 202         return -1;
 203     }
 204
 205     /* To be safe, we reserve twice the size of the packet */
 206     z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
 207     z->zbuff = g_try_malloc(z->zbuff_len);
 208     if (!z->zbuff) {
 209         ZSTD_freeDStream(z->zds);
 210         g_free(z);
 211         error_setg(errp, "multifd %u: out of memory for zbuff", p->id);
 212         return -1;
 213     }
 214     return 0;
 215 }
 216
 217 /**
 218  * zstd_recv_cleanup: setup receive side
 219  *
 220  * For no compression this function does nothing.
 221  *
 222  * @p: Params for the channel that we are using
 223  */
 224 static void zstd_recv_cleanup(MultiFDRecvParams *p)
 225 {
 226     struct zstd_data *z = p->compress_data;
 227
 228     ZSTD_freeDStream(z->zds);
 229     z->zds = NULL;
 230     g_free(z->zbuff);
 231     z->zbuff = NULL;
 232     g_free(p->compress_data);
 233     p->compress_data = NULL;
 234 }
 235
 236 /**
 237  * zstd_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 zstd_recv(MultiFDRecvParams *p, Error **errp)
 248 {
 249     uint32_t in_size = p->next_packet_size;
 250     uint32_t out_size = 0;
 251     uint32_t expected_size = p->normal_num * p->page_size;
 252     uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
 253     struct zstd_data *z = p->compress_data;
 254     int ret;
 255     int i;
 256
 257     if (flags != MULTIFD_FLAG_ZSTD) {
 258         error_setg(errp, "multifd %u: flags received %x flags expected %x",
 259                    p->id, flags, MULTIFD_FLAG_ZSTD);
 260         return -1;
 261     }
 262
 263     multifd_recv_zero_page_process(p);
 264
 265     if (!p->normal_num) {
 266         assert(in_size == 0);
 267         return 0;
 268     }
 269
 270     ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp);
 271
 272     if (ret != 0) {
 273         return ret;
 274     }
 275
 276     z->in.src = z->zbuff;
 277     z->in.size = in_size;
 278     z->in.pos = 0;
 279
 280     for (i = 0; i < p->normal_num; i++) {
 281         z->out.dst = p->host + p->normal[i];
 282         z->out.size = p->page_size;
 283         z->out.pos = 0;
 284
 285         /*
 286          * Welcome to decompressStream semantics
 287          *
 288          * We need to loop while:
 289          * - return is > 0
 290          * - there is input available
 291          * - we haven't put out a full page
 292          */
 293         do {
 294             ret = ZSTD_decompressStream(z->zds, &z->out, &z->in);
 295         } while (ret > 0 && (z->in.size - z->in.pos > 0)
 296                          && (z->out.pos < p->page_size));
 297         if (ret > 0 && (z->out.pos < p->page_size)) {
 298             error_setg(errp, "multifd %u: decompressStream buffer too small",
 299                        p->id);
 300             return -1;
 301         }
 302         if (ZSTD_isError(ret)) {
 303             error_setg(errp, "multifd %u: decompressStream returned %s",
 304                        p->id, ZSTD_getErrorName(ret));
 305             return ret;
 306         }
 307         out_size += z->out.pos;
 308     }
 309     if (out_size != expected_size) {
 310         error_setg(errp, "multifd %u: packet size received %u size expected %u",
 311                    p->id, out_size, expected_size);
 312         return -1;
 313     }
 314     return 0;
 315 }
 316
 317 static MultiFDMethods multifd_zstd_ops = {
 318     .send_setup = zstd_send_setup,
 319     .send_cleanup = zstd_send_cleanup,
 320     .send_prepare = zstd_send_prepare,
 321     .recv_setup = zstd_recv_setup,
 322     .recv_cleanup = zstd_recv_cleanup,
 323     .recv = zstd_recv
 324 };
 325
 326 static void multifd_zstd_register(void)
 327 {
 328     multifd_register_ops(MULTIFD_COMPRESSION_ZSTD, &multifd_zstd_ops);
 329 }
 330
 331 migration_init(multifd_zstd_register);