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