2 * Multifd zlib compression implementation
4 * Copyright (c) 2020 Red Hat Inc
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"
16 #include "exec/ramblock.h"
17 #include "exec/target_page.h"
18 #include "qapi/error.h"
19 #include "migration.h"
24 /* stream for compression */
26 /* stream for decompression */
31 /* compressed buffer */
33 /* size of compressed buffer */
37 /* Multifd zstd compression */
40 * zstd_send_setup: setup send side
42 * Setup each channel with zstd compression.
44 * Returns 0 for success or -1 for error
46 * @p: Params for the channel that we are using
47 * @errp: pointer to an error
49 static int zstd_send_setup(MultiFDSendParams
*p
, Error
**errp
)
51 struct zstd_data
*z
= g_new0(struct zstd_data
, 1);
55 z
->zcs
= ZSTD_createCStream();
58 error_setg(errp
, "multifd %d: zstd createCStream failed", p
->id
);
62 res
= ZSTD_initCStream(z
->zcs
, migrate_multifd_zstd_level());
63 if (ZSTD_isError(res
)) {
64 ZSTD_freeCStream(z
->zcs
);
66 error_setg(errp
, "multifd %d: initCStream failed with error %s",
67 p
->id
, ZSTD_getErrorName(res
));
70 /* To be safe, we reserve twice the size of the packet */
71 z
->zbuff_len
= MULTIFD_PACKET_SIZE
* 2;
72 z
->zbuff
= g_try_malloc(z
->zbuff_len
);
74 ZSTD_freeCStream(z
->zcs
);
76 error_setg(errp
, "multifd %d: out of memory for zbuff", p
->id
);
83 * zstd_send_cleanup: cleanup send side
85 * Close the channel and return memory.
87 * @p: Params for the channel that we are using
88 * @errp: pointer to an error
90 static void zstd_send_cleanup(MultiFDSendParams
*p
, Error
**errp
)
92 struct zstd_data
*z
= p
->data
;
94 ZSTD_freeCStream(z
->zcs
);
103 * zstd_send_prepare: prepare date to be able to send
105 * Create a compressed buffer with all the pages that we are going to
108 * Returns 0 for success or -1 for error
110 * @p: Params for the channel that we are using
111 * @errp: pointer to an error
113 static int zstd_send_prepare(MultiFDSendParams
*p
, Error
**errp
)
115 struct zstd_data
*z
= p
->data
;
116 size_t page_size
= qemu_target_page_size();
120 z
->out
.dst
= z
->zbuff
;
121 z
->out
.size
= z
->zbuff_len
;
124 for (i
= 0; i
< p
->pages
->num
; i
++) {
125 ZSTD_EndDirective flush
= ZSTD_e_continue
;
127 if (i
== p
->pages
->num
- 1) {
128 flush
= ZSTD_e_flush
;
130 z
->in
.src
= p
->pages
->block
->host
+ p
->pages
->offset
[i
];
131 z
->in
.size
= page_size
;
135 * Welcome to compressStream2 semantics
137 * We need to loop while:
139 * - there is input available
140 * - there is output space free
143 ret
= ZSTD_compressStream2(z
->zcs
, &z
->out
, &z
->in
, flush
);
144 } while (ret
> 0 && (z
->in
.size
- z
->in
.pos
> 0)
145 && (z
->out
.size
- z
->out
.pos
> 0));
146 if (ret
> 0 && (z
->in
.size
- z
->in
.pos
> 0)) {
147 error_setg(errp
, "multifd %d: compressStream buffer too small",
151 if (ZSTD_isError(ret
)) {
152 error_setg(errp
, "multifd %d: compressStream error %s",
153 p
->id
, ZSTD_getErrorName(ret
));
157 p
->next_packet_size
= z
->out
.pos
;
158 p
->flags
|= MULTIFD_FLAG_ZSTD
;
164 * zstd_send_write: do the actual write of the data
166 * Do the actual write of the comprresed buffer.
168 * Returns 0 for success or -1 for error
170 * @p: Params for the channel that we are using
171 * @used: number of pages used
172 * @errp: pointer to an error
174 static int zstd_send_write(MultiFDSendParams
*p
, uint32_t used
, Error
**errp
)
176 struct zstd_data
*z
= p
->data
;
178 return qio_channel_write_all(p
->c
, (void *)z
->zbuff
, p
->next_packet_size
,
183 * zstd_recv_setup: setup receive side
185 * Create the compressed channel and buffer.
187 * Returns 0 for success or -1 for error
189 * @p: Params for the channel that we are using
190 * @errp: pointer to an error
192 static int zstd_recv_setup(MultiFDRecvParams
*p
, Error
**errp
)
194 struct zstd_data
*z
= g_new0(struct zstd_data
, 1);
198 z
->zds
= ZSTD_createDStream();
201 error_setg(errp
, "multifd %d: zstd createDStream failed", p
->id
);
205 ret
= ZSTD_initDStream(z
->zds
);
206 if (ZSTD_isError(ret
)) {
207 ZSTD_freeDStream(z
->zds
);
209 error_setg(errp
, "multifd %d: initDStream failed with error %s",
210 p
->id
, ZSTD_getErrorName(ret
));
214 /* To be safe, we reserve twice the size of the packet */
215 z
->zbuff_len
= MULTIFD_PACKET_SIZE
* 2;
216 z
->zbuff
= g_try_malloc(z
->zbuff_len
);
218 ZSTD_freeDStream(z
->zds
);
220 error_setg(errp
, "multifd %d: out of memory for zbuff", p
->id
);
227 * zstd_recv_cleanup: setup receive side
229 * For no compression this function does nothing.
231 * @p: Params for the channel that we are using
233 static void zstd_recv_cleanup(MultiFDRecvParams
*p
)
235 struct zstd_data
*z
= p
->data
;
237 ZSTD_freeDStream(z
->zds
);
246 * zstd_recv_pages: read the data from the channel into actual pages
248 * Read the compressed buffer, and uncompress it into the actual
251 * Returns 0 for success or -1 for error
253 * @p: Params for the channel that we are using
254 * @errp: pointer to an error
256 static int zstd_recv_pages(MultiFDRecvParams
*p
, Error
**errp
)
258 uint32_t in_size
= p
->next_packet_size
;
259 uint32_t out_size
= 0;
260 size_t page_size
= qemu_target_page_size();
261 uint32_t expected_size
= p
->pages
->num
* page_size
;
262 uint32_t flags
= p
->flags
& MULTIFD_FLAG_COMPRESSION_MASK
;
263 struct zstd_data
*z
= p
->data
;
267 if (flags
!= MULTIFD_FLAG_ZSTD
) {
268 error_setg(errp
, "multifd %d: flags received %x flags expected %x",
269 p
->id
, flags
, MULTIFD_FLAG_ZSTD
);
272 ret
= qio_channel_read_all(p
->c
, (void *)z
->zbuff
, in_size
, errp
);
278 z
->in
.src
= z
->zbuff
;
279 z
->in
.size
= in_size
;
282 for (i
= 0; i
< p
->pages
->num
; i
++) {
283 z
->out
.dst
= p
->pages
->block
->host
+ p
->pages
->offset
[i
];
284 z
->out
.size
= page_size
;
288 * Welcome to decompressStream semantics
290 * We need to loop while:
292 * - there is input available
293 * - we haven't put out a full page
296 ret
= ZSTD_decompressStream(z
->zds
, &z
->out
, &z
->in
);
297 } while (ret
> 0 && (z
->in
.size
- z
->in
.pos
> 0)
298 && (z
->out
.pos
< page_size
));
299 if (ret
> 0 && (z
->out
.pos
< page_size
)) {
300 error_setg(errp
, "multifd %d: decompressStream buffer too small",
304 if (ZSTD_isError(ret
)) {
305 error_setg(errp
, "multifd %d: decompressStream returned %s",
306 p
->id
, ZSTD_getErrorName(ret
));
309 out_size
+= z
->out
.pos
;
311 if (out_size
!= expected_size
) {
312 error_setg(errp
, "multifd %d: packet size received %d size expected %d",
313 p
->id
, out_size
, expected_size
);
319 static MultiFDMethods multifd_zstd_ops
= {
320 .send_setup
= zstd_send_setup
,
321 .send_cleanup
= zstd_send_cleanup
,
322 .send_prepare
= zstd_send_prepare
,
323 .send_write
= zstd_send_write
,
324 .recv_setup
= zstd_recv_setup
,
325 .recv_cleanup
= zstd_recv_cleanup
,
326 .recv_pages
= zstd_recv_pages
329 static void multifd_zstd_register(void)
331 multifd_register_ops(MULTIFD_COMPRESSION_ZSTD
, &multifd_zstd_ops
);
334 migration_init(multifd_zstd_register
);