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