xref: /openbmc/qemu/migration/multifd-qpl.c (revision f6fe9fea)
1 /*
2  * Multifd qpl compression accelerator implementation
3  *
4  * Copyright (c) 2023 Intel Corporation
5  *
6  * Authors:
7  *  Yuan Liu<yuan1.liu@intel.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 "qemu/module.h"
15 #include "qapi/error.h"
16 #include "qapi/qapi-types-migration.h"
17 #include "exec/ramblock.h"
18 #include "multifd.h"
19 #include "qpl/qpl.h"
20 
21 /* Maximum number of retries to resubmit a job if IAA work queues are full */
22 #define MAX_SUBMIT_RETRY_NUM (3)
23 
24 typedef struct {
25     /* the QPL hardware path job */
26     qpl_job *job;
27     /* indicates if fallback to software path is required */
28     bool fallback_sw_path;
29     /* output data from the software path */
30     uint8_t *sw_output;
31     /* output data length from the software path */
32     uint32_t sw_output_len;
33 } QplHwJob;
34 
35 typedef struct {
36     /* array of hardware jobs, the number of jobs equals the number pages */
37     QplHwJob *hw_jobs;
38     /* the QPL software job for the slow path and software fallback */
39     qpl_job *sw_job;
40     /* the number of pages that the QPL needs to process at one time */
41     uint32_t page_num;
42     /* array of compressed page buffers */
43     uint8_t *zbuf;
44     /* array of compressed page lengths */
45     uint32_t *zlen;
46     /* the status of the hardware device */
47     bool hw_avail;
48 } QplData;
49 
50 /**
51  * check_hw_avail: check if IAA hardware is available
52  *
53  * If the IAA hardware does not exist or is unavailable,
54  * the QPL hardware job initialization will fail.
55  *
56  * Returns true if IAA hardware is available, otherwise false.
57  *
58  * @job_size: indicates the hardware job size if hardware is available
59  */
check_hw_avail(uint32_t * job_size)60 static bool check_hw_avail(uint32_t *job_size)
61 {
62     qpl_path_t path = qpl_path_hardware;
63     uint32_t size = 0;
64     qpl_job *job;
65 
66     if (qpl_get_job_size(path, &size) != QPL_STS_OK) {
67         return false;
68     }
69     assert(size > 0);
70     job = g_malloc0(size);
71     if (qpl_init_job(path, job) != QPL_STS_OK) {
72         g_free(job);
73         return false;
74     }
75     g_free(job);
76     *job_size = size;
77     return true;
78 }
79 
80 /**
81  * multifd_qpl_free_sw_job: clean up software job
82  *
83  * Free the software job resources.
84  *
85  * @qpl: pointer to the QplData structure
86  */
multifd_qpl_free_sw_job(QplData * qpl)87 static void multifd_qpl_free_sw_job(QplData *qpl)
88 {
89     assert(qpl);
90     if (qpl->sw_job) {
91         qpl_fini_job(qpl->sw_job);
92         g_free(qpl->sw_job);
93         qpl->sw_job = NULL;
94     }
95 }
96 
97 /**
98  * multifd_qpl_free_jobs: clean up hardware jobs
99  *
100  * Free all hardware job resources.
101  *
102  * @qpl: pointer to the QplData structure
103  */
multifd_qpl_free_hw_job(QplData * qpl)104 static void multifd_qpl_free_hw_job(QplData *qpl)
105 {
106     assert(qpl);
107     if (qpl->hw_jobs) {
108         for (int i = 0; i < qpl->page_num; i++) {
109             qpl_fini_job(qpl->hw_jobs[i].job);
110             g_free(qpl->hw_jobs[i].job);
111             qpl->hw_jobs[i].job = NULL;
112         }
113         g_free(qpl->hw_jobs);
114         qpl->hw_jobs = NULL;
115     }
116 }
117 
118 /**
119  * multifd_qpl_init_sw_job: initialize a software job
120  *
121  * Use the QPL software path to initialize a job
122  *
123  * @qpl: pointer to the QplData structure
124  * @errp: pointer to an error
125  */
multifd_qpl_init_sw_job(QplData * qpl,Error ** errp)126 static int multifd_qpl_init_sw_job(QplData *qpl, Error **errp)
127 {
128     qpl_path_t path = qpl_path_software;
129     uint32_t size = 0;
130     qpl_job *job = NULL;
131     qpl_status status;
132 
133     status = qpl_get_job_size(path, &size);
134     if (status != QPL_STS_OK) {
135         error_setg(errp, "qpl_get_job_size failed with error %d", status);
136         return -1;
137     }
138     job = g_malloc0(size);
139     status = qpl_init_job(path, job);
140     if (status != QPL_STS_OK) {
141         error_setg(errp, "qpl_init_job failed with error %d", status);
142         g_free(job);
143         return -1;
144     }
145     qpl->sw_job = job;
146     return 0;
147 }
148 
149 /**
150  * multifd_qpl_init_jobs: initialize hardware jobs
151  *
152  * Use the QPL hardware path to initialize jobs
153  *
154  * @qpl: pointer to the QplData structure
155  * @size: the size of QPL hardware path job
156  * @errp: pointer to an error
157  */
multifd_qpl_init_hw_job(QplData * qpl,uint32_t size,Error ** errp)158 static void multifd_qpl_init_hw_job(QplData *qpl, uint32_t size, Error **errp)
159 {
160     qpl_path_t path = qpl_path_hardware;
161     qpl_job *job = NULL;
162     qpl_status status;
163 
164     qpl->hw_jobs = g_new0(QplHwJob, qpl->page_num);
165     for (int i = 0; i < qpl->page_num; i++) {
166         job = g_malloc0(size);
167         status = qpl_init_job(path, job);
168         /* the job initialization should succeed after check_hw_avail */
169         assert(status == QPL_STS_OK);
170         qpl->hw_jobs[i].job = job;
171     }
172 }
173 
174 /**
175  * multifd_qpl_init: initialize QplData structure
176  *
177  * Allocate and initialize a QplData structure
178  *
179  * Returns a QplData pointer on success or NULL on error
180  *
181  * @num: the number of pages
182  * @size: the page size
183  * @errp: pointer to an error
184  */
multifd_qpl_init(uint32_t num,uint32_t size,Error ** errp)185 static QplData *multifd_qpl_init(uint32_t num, uint32_t size, Error **errp)
186 {
187     uint32_t job_size = 0;
188     QplData *qpl;
189 
190     qpl = g_new0(QplData, 1);
191     qpl->page_num = num;
192     if (multifd_qpl_init_sw_job(qpl, errp) != 0) {
193         g_free(qpl);
194         return NULL;
195     }
196     qpl->hw_avail = check_hw_avail(&job_size);
197     if (qpl->hw_avail) {
198         multifd_qpl_init_hw_job(qpl, job_size, errp);
199     }
200     qpl->zbuf = g_malloc0(size * num);
201     qpl->zlen = g_new0(uint32_t, num);
202     return qpl;
203 }
204 
205 /**
206  * multifd_qpl_deinit: clean up QplData structure
207  *
208  * Free jobs, buffers and the QplData structure
209  *
210  * @qpl: pointer to the QplData structure
211  */
multifd_qpl_deinit(QplData * qpl)212 static void multifd_qpl_deinit(QplData *qpl)
213 {
214     if (qpl) {
215         multifd_qpl_free_sw_job(qpl);
216         multifd_qpl_free_hw_job(qpl);
217         g_free(qpl->zbuf);
218         g_free(qpl->zlen);
219         g_free(qpl);
220     }
221 }
222 
223 /**
224  * multifd_qpl_send_setup: set up send side
225  *
226  * Set up the channel with QPL compression.
227  *
228  * Returns 0 on success or -1 on error
229  *
230  * @p: Params for the channel being used
231  * @errp: pointer to an error
232  */
multifd_qpl_send_setup(MultiFDSendParams * p,Error ** errp)233 static int multifd_qpl_send_setup(MultiFDSendParams *p, Error **errp)
234 {
235     QplData *qpl;
236 
237     qpl = multifd_qpl_init(p->page_count, p->page_size, errp);
238     if (!qpl) {
239         return -1;
240     }
241     p->compress_data = qpl;
242 
243     /*
244      * the page will be compressed independently and sent using an IOV. The
245      * additional two IOVs are used to store packet header and compressed data
246      * length
247      */
248     p->iov = g_new0(struct iovec, p->page_count + 2);
249     return 0;
250 }
251 
252 /**
253  * multifd_qpl_send_cleanup: clean up send side
254  *
255  * Close the channel and free memory.
256  *
257  * @p: Params for the channel being used
258  * @errp: pointer to an error
259  */
multifd_qpl_send_cleanup(MultiFDSendParams * p,Error ** errp)260 static void multifd_qpl_send_cleanup(MultiFDSendParams *p, Error **errp)
261 {
262     multifd_qpl_deinit(p->compress_data);
263     p->compress_data = NULL;
264     g_free(p->iov);
265     p->iov = NULL;
266 }
267 
268 /**
269  * multifd_qpl_prepare_job: prepare the job
270  *
271  * Set the QPL job parameters and properties.
272  *
273  * @job: pointer to the qpl_job structure
274  * @is_compression: indicates compression and decompression
275  * @input: pointer to the input data buffer
276  * @input_len: the length of the input data
277  * @output: pointer to the output data buffer
278  * @output_len: the length of the output data
279  */
multifd_qpl_prepare_job(qpl_job * job,bool is_compression,uint8_t * input,uint32_t input_len,uint8_t * output,uint32_t output_len)280 static void multifd_qpl_prepare_job(qpl_job *job, bool is_compression,
281                                     uint8_t *input, uint32_t input_len,
282                                     uint8_t *output, uint32_t output_len)
283 {
284     job->op = is_compression ? qpl_op_compress : qpl_op_decompress;
285     job->next_in_ptr = input;
286     job->next_out_ptr = output;
287     job->available_in = input_len;
288     job->available_out = output_len;
289     job->flags = QPL_FLAG_FIRST | QPL_FLAG_LAST | QPL_FLAG_OMIT_VERIFY;
290     /* only supports compression level 1 */
291     job->level = 1;
292 }
293 
294 /**
295  * multifd_qpl_prepare_comp_job: prepare the compression job
296  *
297  * Set the compression job parameters and properties.
298  *
299  * @job: pointer to the qpl_job structure
300  * @input: pointer to the input data buffer
301  * @output: pointer to the output data buffer
302  * @size: the page size
303  */
multifd_qpl_prepare_comp_job(qpl_job * job,uint8_t * input,uint8_t * output,uint32_t size)304 static void multifd_qpl_prepare_comp_job(qpl_job *job, uint8_t *input,
305                                          uint8_t *output, uint32_t size)
306 {
307     /*
308      * Set output length to less than the page size to force the job to
309      * fail in case it compresses to a larger size. We'll send that page
310      * without compression and skip the decompression operation on the
311      * destination.
312      */
313     multifd_qpl_prepare_job(job, true, input, size, output, size - 1);
314 }
315 
316 /**
317  * multifd_qpl_prepare_decomp_job: prepare the decompression job
318  *
319  * Set the decompression job parameters and properties.
320  *
321  * @job: pointer to the qpl_job structure
322  * @input: pointer to the input data buffer
323  * @len: the length of the input data
324  * @output: pointer to the output data buffer
325  * @size: the page size
326  */
multifd_qpl_prepare_decomp_job(qpl_job * job,uint8_t * input,uint32_t len,uint8_t * output,uint32_t size)327 static void multifd_qpl_prepare_decomp_job(qpl_job *job, uint8_t *input,
328                                            uint32_t len, uint8_t *output,
329                                            uint32_t size)
330 {
331     multifd_qpl_prepare_job(job, false, input, len, output, size);
332 }
333 
334 /**
335  * multifd_qpl_fill_iov: fill in the IOV
336  *
337  * Fill in the QPL packet IOV
338  *
339  * @p: Params for the channel being used
340  * @data: pointer to the IOV data
341  * @len: The length of the IOV data
342  */
multifd_qpl_fill_iov(MultiFDSendParams * p,uint8_t * data,uint32_t len)343 static void multifd_qpl_fill_iov(MultiFDSendParams *p, uint8_t *data,
344                                  uint32_t len)
345 {
346     p->iov[p->iovs_num].iov_base = data;
347     p->iov[p->iovs_num].iov_len = len;
348     p->iovs_num++;
349     p->next_packet_size += len;
350 }
351 
352 /**
353  * multifd_qpl_fill_packet: fill the compressed page into the QPL packet
354  *
355  * Fill the compressed page length and IOV into the QPL packet
356  *
357  * @idx: The index of the compressed length array
358  * @p: Params for the channel being used
359  * @data: pointer to the compressed page buffer
360  * @len: The length of the compressed page
361  */
multifd_qpl_fill_packet(uint32_t idx,MultiFDSendParams * p,uint8_t * data,uint32_t len)362 static void multifd_qpl_fill_packet(uint32_t idx, MultiFDSendParams *p,
363                                     uint8_t *data, uint32_t len)
364 {
365     QplData *qpl = p->compress_data;
366 
367     qpl->zlen[idx] = cpu_to_be32(len);
368     multifd_qpl_fill_iov(p, data, len);
369 }
370 
371 /**
372  * multifd_qpl_submit_job: submit a job to the hardware
373  *
374  * Submit a QPL hardware job to the IAA device
375  *
376  * Returns true if the job is submitted successfully, otherwise false.
377  *
378  * @job: pointer to the qpl_job structure
379  */
multifd_qpl_submit_job(qpl_job * job)380 static bool multifd_qpl_submit_job(qpl_job *job)
381 {
382     qpl_status status;
383     uint32_t num = 0;
384 
385 retry:
386     status = qpl_submit_job(job);
387     if (status == QPL_STS_QUEUES_ARE_BUSY_ERR) {
388         if (num < MAX_SUBMIT_RETRY_NUM) {
389             num++;
390             goto retry;
391         }
392     }
393     return (status == QPL_STS_OK);
394 }
395 
396 /**
397  * multifd_qpl_compress_pages_slow_path: compress pages using slow path
398  *
399  * Compress the pages using software. If compression fails, the uncompressed
400  * page will be sent.
401  *
402  * @p: Params for the channel being used
403  */
multifd_qpl_compress_pages_slow_path(MultiFDSendParams * p)404 static void multifd_qpl_compress_pages_slow_path(MultiFDSendParams *p)
405 {
406     QplData *qpl = p->compress_data;
407     uint32_t size = p->page_size;
408     qpl_job *job = qpl->sw_job;
409     uint8_t *zbuf = qpl->zbuf;
410     uint8_t *buf;
411 
412     for (int i = 0; i < p->pages->normal_num; i++) {
413         buf = p->pages->block->host + p->pages->offset[i];
414         multifd_qpl_prepare_comp_job(job, buf, zbuf, size);
415         if (qpl_execute_job(job) == QPL_STS_OK) {
416             multifd_qpl_fill_packet(i, p, zbuf, job->total_out);
417         } else {
418             /* send the uncompressed page */
419             multifd_qpl_fill_packet(i, p, buf, size);
420         }
421         zbuf += size;
422     }
423 }
424 
425 /**
426  * multifd_qpl_compress_pages: compress pages
427  *
428  * Submit the pages to the IAA hardware for compression. If hardware
429  * compression fails, it falls back to software compression. If software
430  * compression also fails, the uncompressed page is sent.
431  *
432  * @p: Params for the channel being used
433  */
multifd_qpl_compress_pages(MultiFDSendParams * p)434 static void multifd_qpl_compress_pages(MultiFDSendParams *p)
435 {
436     QplData *qpl = p->compress_data;
437     MultiFDPages_t *pages = p->pages;
438     uint32_t size = p->page_size;
439     QplHwJob *hw_job;
440     uint8_t *buf;
441     uint8_t *zbuf;
442 
443     for (int i = 0; i < pages->normal_num; i++) {
444         buf = pages->block->host + pages->offset[i];
445         zbuf = qpl->zbuf + (size * i);
446         hw_job = &qpl->hw_jobs[i];
447         multifd_qpl_prepare_comp_job(hw_job->job, buf, zbuf, size);
448         if (multifd_qpl_submit_job(hw_job->job)) {
449             hw_job->fallback_sw_path = false;
450         } else {
451             /*
452              * The IAA work queue is full, any immediate subsequent job
453              * submission is likely to fail, sending the page via the QPL
454              * software path at this point gives us a better chance of
455              * finding the queue open for the next pages.
456              */
457             hw_job->fallback_sw_path = true;
458             multifd_qpl_prepare_comp_job(qpl->sw_job, buf, zbuf, size);
459             if (qpl_execute_job(qpl->sw_job) == QPL_STS_OK) {
460                 hw_job->sw_output = zbuf;
461                 hw_job->sw_output_len = qpl->sw_job->total_out;
462             } else {
463                 hw_job->sw_output = buf;
464                 hw_job->sw_output_len = size;
465             }
466         }
467     }
468 
469     for (int i = 0; i < pages->normal_num; i++) {
470         buf = pages->block->host + pages->offset[i];
471         zbuf = qpl->zbuf + (size * i);
472         hw_job = &qpl->hw_jobs[i];
473         if (hw_job->fallback_sw_path) {
474             multifd_qpl_fill_packet(i, p, hw_job->sw_output,
475                                     hw_job->sw_output_len);
476             continue;
477         }
478         if (qpl_wait_job(hw_job->job) == QPL_STS_OK) {
479             multifd_qpl_fill_packet(i, p, zbuf, hw_job->job->total_out);
480         } else {
481             /* send the uncompressed page */
482             multifd_qpl_fill_packet(i, p, buf, size);
483         }
484     }
485 }
486 
487 /**
488  * multifd_qpl_send_prepare: prepare data to be able to send
489  *
490  * Create a compressed buffer with all the pages that we are going to
491  * send.
492  *
493  * Returns 0 on success or -1 on error
494  *
495  * @p: Params for the channel being used
496  * @errp: pointer to an error
497  */
multifd_qpl_send_prepare(MultiFDSendParams * p,Error ** errp)498 static int multifd_qpl_send_prepare(MultiFDSendParams *p, Error **errp)
499 {
500     QplData *qpl = p->compress_data;
501     uint32_t len = 0;
502 
503     if (!multifd_send_prepare_common(p)) {
504         goto out;
505     }
506 
507     /* The first IOV is used to store the compressed page lengths */
508     len = p->pages->normal_num * sizeof(uint32_t);
509     multifd_qpl_fill_iov(p, (uint8_t *) qpl->zlen, len);
510     if (qpl->hw_avail) {
511         multifd_qpl_compress_pages(p);
512     } else {
513         multifd_qpl_compress_pages_slow_path(p);
514     }
515 
516 out:
517     p->flags |= MULTIFD_FLAG_QPL;
518     multifd_send_fill_packet(p);
519     return 0;
520 }
521 
522 /**
523  * multifd_qpl_recv_setup: set up receive side
524  *
525  * Create the compressed channel and buffer.
526  *
527  * Returns 0 on success or -1 on error
528  *
529  * @p: Params for the channel being used
530  * @errp: pointer to an error
531  */
multifd_qpl_recv_setup(MultiFDRecvParams * p,Error ** errp)532 static int multifd_qpl_recv_setup(MultiFDRecvParams *p, Error **errp)
533 {
534     QplData *qpl;
535 
536     qpl = multifd_qpl_init(p->page_count, p->page_size, errp);
537     if (!qpl) {
538         return -1;
539     }
540     p->compress_data = qpl;
541     return 0;
542 }
543 
544 /**
545  * multifd_qpl_recv_cleanup: set up receive side
546  *
547  * Close the channel and free memory.
548  *
549  * @p: Params for the channel being used
550  */
multifd_qpl_recv_cleanup(MultiFDRecvParams * p)551 static void multifd_qpl_recv_cleanup(MultiFDRecvParams *p)
552 {
553     multifd_qpl_deinit(p->compress_data);
554     p->compress_data = NULL;
555 }
556 
557 /**
558  * multifd_qpl_process_and_check_job: process and check a QPL job
559  *
560  * Process the job and check whether the job output length is the
561  * same as the specified length
562  *
563  * Returns true if the job execution succeeded and the output length
564  * is equal to the specified length, otherwise false.
565  *
566  * @job: pointer to the qpl_job structure
567  * @is_hardware: indicates whether the job is a hardware job
568  * @len: Specified output length
569  * @errp: pointer to an error
570  */
multifd_qpl_process_and_check_job(qpl_job * job,bool is_hardware,uint32_t len,Error ** errp)571 static bool multifd_qpl_process_and_check_job(qpl_job *job, bool is_hardware,
572                                               uint32_t len, Error **errp)
573 {
574     qpl_status status;
575 
576     status = (is_hardware ? qpl_wait_job(job) : qpl_execute_job(job));
577     if (status != QPL_STS_OK) {
578         error_setg(errp, "qpl job failed with error %d", status);
579         return false;
580     }
581     if (job->total_out != len) {
582         error_setg(errp, "qpl decompressed len %u, expected len %u",
583                    job->total_out, len);
584         return false;
585     }
586     return true;
587 }
588 
589 /**
590  * multifd_qpl_decompress_pages_slow_path: decompress pages using slow path
591  *
592  * Decompress the pages using software
593  *
594  * Returns 0 on success or -1 on error
595  *
596  * @p: Params for the channel being used
597  * @errp: pointer to an error
598  */
multifd_qpl_decompress_pages_slow_path(MultiFDRecvParams * p,Error ** errp)599 static int multifd_qpl_decompress_pages_slow_path(MultiFDRecvParams *p,
600                                                   Error **errp)
601 {
602     QplData *qpl = p->compress_data;
603     uint32_t size = p->page_size;
604     qpl_job *job = qpl->sw_job;
605     uint8_t *zbuf = qpl->zbuf;
606     uint8_t *addr;
607     uint32_t len;
608 
609     for (int i = 0; i < p->normal_num; i++) {
610         len = qpl->zlen[i];
611         addr = p->host + p->normal[i];
612         /* the page is uncompressed, load it */
613         if (len == size) {
614             memcpy(addr, zbuf, size);
615             zbuf += size;
616             continue;
617         }
618         multifd_qpl_prepare_decomp_job(job, zbuf, len, addr, size);
619         if (!multifd_qpl_process_and_check_job(job, false, size, errp)) {
620             return -1;
621         }
622         zbuf += len;
623     }
624     return 0;
625 }
626 
627 /**
628  * multifd_qpl_decompress_pages: decompress pages
629  *
630  * Decompress the pages using the IAA hardware. If hardware
631  * decompression fails, it falls back to software decompression.
632  *
633  * Returns 0 on success or -1 on error
634  *
635  * @p: Params for the channel being used
636  * @errp: pointer to an error
637  */
multifd_qpl_decompress_pages(MultiFDRecvParams * p,Error ** errp)638 static int multifd_qpl_decompress_pages(MultiFDRecvParams *p, Error **errp)
639 {
640     QplData *qpl = p->compress_data;
641     uint32_t size = p->page_size;
642     uint8_t *zbuf = qpl->zbuf;
643     uint8_t *addr;
644     uint32_t len;
645     qpl_job *job;
646 
647     for (int i = 0; i < p->normal_num; i++) {
648         addr = p->host + p->normal[i];
649         len = qpl->zlen[i];
650         /* the page is uncompressed if received length equals the page size */
651         if (len == size) {
652             memcpy(addr, zbuf, size);
653             zbuf += size;
654             continue;
655         }
656 
657         job = qpl->hw_jobs[i].job;
658         multifd_qpl_prepare_decomp_job(job, zbuf, len, addr, size);
659         if (multifd_qpl_submit_job(job)) {
660             qpl->hw_jobs[i].fallback_sw_path = false;
661         } else {
662             /*
663              * The IAA work queue is full, any immediate subsequent job
664              * submission is likely to fail, sending the page via the QPL
665              * software path at this point gives us a better chance of
666              * finding the queue open for the next pages.
667              */
668             qpl->hw_jobs[i].fallback_sw_path = true;
669             job = qpl->sw_job;
670             multifd_qpl_prepare_decomp_job(job, zbuf, len, addr, size);
671             if (!multifd_qpl_process_and_check_job(job, false, size, errp)) {
672                 return -1;
673             }
674         }
675         zbuf += len;
676     }
677 
678     for (int i = 0; i < p->normal_num; i++) {
679         /* ignore pages that have already been processed */
680         if (qpl->zlen[i] == size || qpl->hw_jobs[i].fallback_sw_path) {
681             continue;
682         }
683 
684         job = qpl->hw_jobs[i].job;
685         if (!multifd_qpl_process_and_check_job(job, true, size, errp)) {
686             return -1;
687         }
688     }
689     return 0;
690 }
691 /**
692  * multifd_qpl_recv: read the data from the channel into actual pages
693  *
694  * Read the compressed buffer, and uncompress it into the actual
695  * pages.
696  *
697  * Returns 0 on success or -1 on error
698  *
699  * @p: Params for the channel being used
700  * @errp: pointer to an error
701  */
multifd_qpl_recv(MultiFDRecvParams * p,Error ** errp)702 static int multifd_qpl_recv(MultiFDRecvParams *p, Error **errp)
703 {
704     QplData *qpl = p->compress_data;
705     uint32_t in_size = p->next_packet_size;
706     uint32_t flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
707     uint32_t len = 0;
708     uint32_t zbuf_len = 0;
709     int ret;
710 
711     if (flags != MULTIFD_FLAG_QPL) {
712         error_setg(errp, "multifd %u: flags received %x flags expected %x",
713                    p->id, flags, MULTIFD_FLAG_QPL);
714         return -1;
715     }
716     multifd_recv_zero_page_process(p);
717     if (!p->normal_num) {
718         assert(in_size == 0);
719         return 0;
720     }
721 
722     /* read compressed page lengths */
723     len = p->normal_num * sizeof(uint32_t);
724     assert(len < in_size);
725     ret = qio_channel_read_all(p->c, (void *) qpl->zlen, len, errp);
726     if (ret != 0) {
727         return ret;
728     }
729     for (int i = 0; i < p->normal_num; i++) {
730         qpl->zlen[i] = be32_to_cpu(qpl->zlen[i]);
731         assert(qpl->zlen[i] <= p->page_size);
732         zbuf_len += qpl->zlen[i];
733     }
734 
735     /* read compressed pages */
736     assert(in_size == len + zbuf_len);
737     ret = qio_channel_read_all(p->c, (void *) qpl->zbuf, zbuf_len, errp);
738     if (ret != 0) {
739         return ret;
740     }
741 
742     if (qpl->hw_avail) {
743         return multifd_qpl_decompress_pages(p, errp);
744     }
745     return multifd_qpl_decompress_pages_slow_path(p, errp);
746 }
747 
748 static MultiFDMethods multifd_qpl_ops = {
749     .send_setup = multifd_qpl_send_setup,
750     .send_cleanup = multifd_qpl_send_cleanup,
751     .send_prepare = multifd_qpl_send_prepare,
752     .recv_setup = multifd_qpl_recv_setup,
753     .recv_cleanup = multifd_qpl_recv_cleanup,
754     .recv = multifd_qpl_recv,
755 };
756 
multifd_qpl_register(void)757 static void multifd_qpl_register(void)
758 {
759     multifd_register_ops(MULTIFD_COMPRESSION_QPL, &multifd_qpl_ops);
760 }
761 
762 migration_init(multifd_qpl_register);
763