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