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