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