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