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->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->data; 94 95 ZSTD_freeCStream(z->zcs); 96 z->zcs = NULL; 97 g_free(z->zbuff); 98 z->zbuff = NULL; 99 g_free(p->data); 100 p->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->data; 118 int ret; 119 uint32_t i; 120 121 multifd_send_prepare_header(p); 122 123 z->out.dst = z->zbuff; 124 z->out.size = z->zbuff_len; 125 z->out.pos = 0; 126 127 for (i = 0; i < pages->num; i++) { 128 ZSTD_EndDirective flush = ZSTD_e_continue; 129 130 if (i == pages->num - 1) { 131 flush = ZSTD_e_flush; 132 } 133 z->in.src = p->pages->block->host + pages->offset[i]; 134 z->in.size = p->page_size; 135 z->in.pos = 0; 136 137 /* 138 * Welcome to compressStream2 semantics 139 * 140 * We need to loop while: 141 * - return is > 0 142 * - there is input available 143 * - there is output space free 144 */ 145 do { 146 ret = ZSTD_compressStream2(z->zcs, &z->out, &z->in, flush); 147 } while (ret > 0 && (z->in.size - z->in.pos > 0) 148 && (z->out.size - z->out.pos > 0)); 149 if (ret > 0 && (z->in.size - z->in.pos > 0)) { 150 error_setg(errp, "multifd %u: compressStream buffer too small", 151 p->id); 152 return -1; 153 } 154 if (ZSTD_isError(ret)) { 155 error_setg(errp, "multifd %u: compressStream error %s", 156 p->id, ZSTD_getErrorName(ret)); 157 return -1; 158 } 159 } 160 p->iov[p->iovs_num].iov_base = z->zbuff; 161 p->iov[p->iovs_num].iov_len = z->out.pos; 162 p->iovs_num++; 163 p->next_packet_size = z->out.pos; 164 p->flags |= MULTIFD_FLAG_ZSTD; 165 166 multifd_send_fill_packet(p); 167 168 return 0; 169 } 170 171 /** 172 * zstd_recv_setup: setup receive side 173 * 174 * Create the compressed channel and buffer. 175 * 176 * Returns 0 for success or -1 for error 177 * 178 * @p: Params for the channel that we are using 179 * @errp: pointer to an error 180 */ 181 static int zstd_recv_setup(MultiFDRecvParams *p, Error **errp) 182 { 183 struct zstd_data *z = g_new0(struct zstd_data, 1); 184 int ret; 185 186 p->data = z; 187 z->zds = ZSTD_createDStream(); 188 if (!z->zds) { 189 g_free(z); 190 error_setg(errp, "multifd %u: zstd createDStream failed", p->id); 191 return -1; 192 } 193 194 ret = ZSTD_initDStream(z->zds); 195 if (ZSTD_isError(ret)) { 196 ZSTD_freeDStream(z->zds); 197 g_free(z); 198 error_setg(errp, "multifd %u: initDStream failed with error %s", 199 p->id, ZSTD_getErrorName(ret)); 200 return -1; 201 } 202 203 /* To be safe, we reserve twice the size of the packet */ 204 z->zbuff_len = MULTIFD_PACKET_SIZE * 2; 205 z->zbuff = g_try_malloc(z->zbuff_len); 206 if (!z->zbuff) { 207 ZSTD_freeDStream(z->zds); 208 g_free(z); 209 error_setg(errp, "multifd %u: out of memory for zbuff", p->id); 210 return -1; 211 } 212 return 0; 213 } 214 215 /** 216 * zstd_recv_cleanup: setup receive side 217 * 218 * For no compression this function does nothing. 219 * 220 * @p: Params for the channel that we are using 221 */ 222 static void zstd_recv_cleanup(MultiFDRecvParams *p) 223 { 224 struct zstd_data *z = p->data; 225 226 ZSTD_freeDStream(z->zds); 227 z->zds = NULL; 228 g_free(z->zbuff); 229 z->zbuff = NULL; 230 g_free(p->data); 231 p->data = NULL; 232 } 233 234 /** 235 * zstd_recv_pages: read the data from the channel into actual pages 236 * 237 * Read the compressed buffer, and uncompress it into the actual 238 * pages. 239 * 240 * Returns 0 for success or -1 for error 241 * 242 * @p: Params for the channel that we are using 243 * @errp: pointer to an error 244 */ 245 static int zstd_recv_pages(MultiFDRecvParams *p, Error **errp) 246 { 247 uint32_t in_size = p->next_packet_size; 248 uint32_t out_size = 0; 249 uint32_t expected_size = p->normal_num * p->page_size; 250 uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK; 251 struct zstd_data *z = p->data; 252 int ret; 253 int i; 254 255 if (flags != MULTIFD_FLAG_ZSTD) { 256 error_setg(errp, "multifd %u: flags received %x flags expected %x", 257 p->id, flags, MULTIFD_FLAG_ZSTD); 258 return -1; 259 } 260 ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp); 261 262 if (ret != 0) { 263 return ret; 264 } 265 266 z->in.src = z->zbuff; 267 z->in.size = in_size; 268 z->in.pos = 0; 269 270 for (i = 0; i < p->normal_num; i++) { 271 z->out.dst = p->host + p->normal[i]; 272 z->out.size = p->page_size; 273 z->out.pos = 0; 274 275 /* 276 * Welcome to decompressStream semantics 277 * 278 * We need to loop while: 279 * - return is > 0 280 * - there is input available 281 * - we haven't put out a full page 282 */ 283 do { 284 ret = ZSTD_decompressStream(z->zds, &z->out, &z->in); 285 } while (ret > 0 && (z->in.size - z->in.pos > 0) 286 && (z->out.pos < p->page_size)); 287 if (ret > 0 && (z->out.pos < p->page_size)) { 288 error_setg(errp, "multifd %u: decompressStream buffer too small", 289 p->id); 290 return -1; 291 } 292 if (ZSTD_isError(ret)) { 293 error_setg(errp, "multifd %u: decompressStream returned %s", 294 p->id, ZSTD_getErrorName(ret)); 295 return ret; 296 } 297 out_size += z->out.pos; 298 } 299 if (out_size != expected_size) { 300 error_setg(errp, "multifd %u: packet size received %u size expected %u", 301 p->id, out_size, expected_size); 302 return -1; 303 } 304 return 0; 305 } 306 307 static MultiFDMethods multifd_zstd_ops = { 308 .send_setup = zstd_send_setup, 309 .send_cleanup = zstd_send_cleanup, 310 .send_prepare = zstd_send_prepare, 311 .recv_setup = zstd_recv_setup, 312 .recv_cleanup = zstd_recv_cleanup, 313 .recv_pages = zstd_recv_pages 314 }; 315 316 static void multifd_zstd_register(void) 317 { 318 multifd_register_ops(MULTIFD_COMPRESSION_ZSTD, &multifd_zstd_ops); 319 } 320 321 migration_init(multifd_zstd_register); 322