block: add missing coroutine_fn annotations
[qemu.git] / migration / multifd-zlib.c
blob18213a951302e5dba2c1c364bcb8803e6c346b10
1 /*
2 * Multifd zlib compression implementation
4 * Copyright (c) 2020 Red Hat Inc
6 * Authors:
7 * Juan Quintela <quintela@redhat.com>
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.
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 "multifd.h"
23 struct zlib_data {
24 /* stream for compression */
25 z_stream zs;
26 /* compressed buffer */
27 uint8_t *zbuff;
28 /* size of compressed buffer */
29 uint32_t zbuff_len;
30 /* uncompressed buffer of size qemu_target_page_size() */
31 uint8_t *buf;
34 /* Multifd zlib compression */
36 /**
37 * zlib_send_setup: setup send side
39 * Setup each channel with zlib compression.
41 * Returns 0 for success or -1 for error
43 * @p: Params for the channel that we are using
44 * @errp: pointer to an error
46 static int zlib_send_setup(MultiFDSendParams *p, Error **errp)
48 struct zlib_data *z = g_new0(struct zlib_data, 1);
49 z_stream *zs = &z->zs;
50 const char *err_msg;
52 zs->zalloc = Z_NULL;
53 zs->zfree = Z_NULL;
54 zs->opaque = Z_NULL;
55 if (deflateInit(zs, migrate_multifd_zlib_level()) != Z_OK) {
56 err_msg = "deflate init failed";
57 goto err_free_z;
59 /* This is the maxium size of the compressed buffer */
60 z->zbuff_len = compressBound(MULTIFD_PACKET_SIZE);
61 z->zbuff = g_try_malloc(z->zbuff_len);
62 if (!z->zbuff) {
63 err_msg = "out of memory for zbuff";
64 goto err_deflate_end;
66 z->buf = g_try_malloc(qemu_target_page_size());
67 if (!z->buf) {
68 err_msg = "out of memory for buf";
69 goto err_free_zbuff;
71 p->data = z;
72 return 0;
74 err_free_zbuff:
75 g_free(z->zbuff);
76 err_deflate_end:
77 deflateEnd(&z->zs);
78 err_free_z:
79 g_free(z);
80 error_setg(errp, "multifd %u: %s", p->id, err_msg);
81 return -1;
84 /**
85 * zlib_send_cleanup: cleanup send side
87 * Close the channel and return memory.
89 * @p: Params for the channel that we are using
90 * @errp: pointer to an error
92 static void zlib_send_cleanup(MultiFDSendParams *p, Error **errp)
94 struct zlib_data *z = p->data;
96 deflateEnd(&z->zs);
97 g_free(z->zbuff);
98 z->zbuff = NULL;
99 g_free(z->buf);
100 z->buf = NULL;
101 g_free(p->data);
102 p->data = NULL;
106 * zlib_send_prepare: prepare date to be able to send
108 * Create a compressed buffer with all the pages that we are going to
109 * send.
111 * Returns 0 for success or -1 for error
113 * @p: Params for the channel that we are using
114 * @errp: pointer to an error
116 static int zlib_send_prepare(MultiFDSendParams *p, Error **errp)
118 struct zlib_data *z = p->data;
119 size_t page_size = qemu_target_page_size();
120 z_stream *zs = &z->zs;
121 uint32_t out_size = 0;
122 int ret;
123 uint32_t i;
125 for (i = 0; i < p->normal_num; i++) {
126 uint32_t available = z->zbuff_len - out_size;
127 int flush = Z_NO_FLUSH;
129 if (i == p->normal_num - 1) {
130 flush = Z_SYNC_FLUSH;
134 * Since the VM might be running, the page may be changing concurrently
135 * with compression. zlib does not guarantee that this is safe,
136 * therefore copy the page before calling deflate().
138 memcpy(z->buf, p->pages->block->host + p->normal[i], page_size);
139 zs->avail_in = page_size;
140 zs->next_in = z->buf;
142 zs->avail_out = available;
143 zs->next_out = z->zbuff + out_size;
146 * Welcome to deflate semantics
148 * We need to loop while:
149 * - return is Z_OK
150 * - there are stuff to be compressed
151 * - there are output space free
153 do {
154 ret = deflate(zs, flush);
155 } while (ret == Z_OK && zs->avail_in && zs->avail_out);
156 if (ret == Z_OK && zs->avail_in) {
157 error_setg(errp, "multifd %u: deflate failed to compress all input",
158 p->id);
159 return -1;
161 if (ret != Z_OK) {
162 error_setg(errp, "multifd %u: deflate returned %d instead of Z_OK",
163 p->id, ret);
164 return -1;
166 out_size += available - zs->avail_out;
168 p->iov[p->iovs_num].iov_base = z->zbuff;
169 p->iov[p->iovs_num].iov_len = out_size;
170 p->iovs_num++;
171 p->next_packet_size = out_size;
172 p->flags |= MULTIFD_FLAG_ZLIB;
174 return 0;
178 * zlib_recv_setup: setup receive side
180 * Create the compressed channel and buffer.
182 * Returns 0 for success or -1 for error
184 * @p: Params for the channel that we are using
185 * @errp: pointer to an error
187 static int zlib_recv_setup(MultiFDRecvParams *p, Error **errp)
189 struct zlib_data *z = g_new0(struct zlib_data, 1);
190 z_stream *zs = &z->zs;
192 p->data = z;
193 zs->zalloc = Z_NULL;
194 zs->zfree = Z_NULL;
195 zs->opaque = Z_NULL;
196 zs->avail_in = 0;
197 zs->next_in = Z_NULL;
198 if (inflateInit(zs) != Z_OK) {
199 error_setg(errp, "multifd %u: inflate init failed", p->id);
200 return -1;
202 /* To be safe, we reserve twice the size of the packet */
203 z->zbuff_len = MULTIFD_PACKET_SIZE * 2;
204 z->zbuff = g_try_malloc(z->zbuff_len);
205 if (!z->zbuff) {
206 inflateEnd(zs);
207 error_setg(errp, "multifd %u: out of memory for zbuff", p->id);
208 return -1;
210 return 0;
214 * zlib_recv_cleanup: setup receive side
216 * For no compression this function does nothing.
218 * @p: Params for the channel that we are using
220 static void zlib_recv_cleanup(MultiFDRecvParams *p)
222 struct zlib_data *z = p->data;
224 inflateEnd(&z->zs);
225 g_free(z->zbuff);
226 z->zbuff = NULL;
227 g_free(p->data);
228 p->data = NULL;
232 * zlib_recv_pages: read the data from the channel into actual pages
234 * Read the compressed buffer, and uncompress it into the actual
235 * pages.
237 * Returns 0 for success or -1 for error
239 * @p: Params for the channel that we are using
240 * @errp: pointer to an error
242 static int zlib_recv_pages(MultiFDRecvParams *p, Error **errp)
244 struct zlib_data *z = p->data;
245 size_t page_size = qemu_target_page_size();
246 z_stream *zs = &z->zs;
247 uint32_t in_size = p->next_packet_size;
248 /* we measure the change of total_out */
249 uint32_t out_size = zs->total_out;
250 uint32_t expected_size = p->normal_num * page_size;
251 uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
252 int ret;
253 int i;
255 if (flags != MULTIFD_FLAG_ZLIB) {
256 error_setg(errp, "multifd %u: flags received %x flags expected %x",
257 p->id, flags, MULTIFD_FLAG_ZLIB);
258 return -1;
260 ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp);
262 if (ret != 0) {
263 return ret;
266 zs->avail_in = in_size;
267 zs->next_in = z->zbuff;
269 for (i = 0; i < p->normal_num; i++) {
270 int flush = Z_NO_FLUSH;
271 unsigned long start = zs->total_out;
273 if (i == p->normal_num - 1) {
274 flush = Z_SYNC_FLUSH;
277 zs->avail_out = page_size;
278 zs->next_out = p->host + p->normal[i];
281 * Welcome to inflate semantics
283 * We need to loop while:
284 * - return is Z_OK
285 * - there are input available
286 * - we haven't completed a full page
288 do {
289 ret = inflate(zs, flush);
290 } while (ret == Z_OK && zs->avail_in
291 && (zs->total_out - start) < page_size);
292 if (ret == Z_OK && (zs->total_out - start) < page_size) {
293 error_setg(errp, "multifd %u: inflate generated too few output",
294 p->id);
295 return -1;
297 if (ret != Z_OK) {
298 error_setg(errp, "multifd %u: inflate returned %d instead of Z_OK",
299 p->id, ret);
300 return -1;
303 out_size = zs->total_out - out_size;
304 if (out_size != expected_size) {
305 error_setg(errp, "multifd %u: packet size received %u size expected %u",
306 p->id, out_size, expected_size);
307 return -1;
309 return 0;
312 static MultiFDMethods multifd_zlib_ops = {
313 .send_setup = zlib_send_setup,
314 .send_cleanup = zlib_send_cleanup,
315 .send_prepare = zlib_send_prepare,
316 .recv_setup = zlib_recv_setup,
317 .recv_cleanup = zlib_recv_cleanup,
318 .recv_pages = zlib_recv_pages
321 static void multifd_zlib_register(void)
323 multifd_register_ops(MULTIFD_COMPRESSION_ZLIB, &multifd_zlib_ops);
326 migration_init(multifd_zlib_register);