| 1 | /* |
| 2 | * Multifd common code |
| 3 | * |
| 4 | * Copyright (c) 2019-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 "qemu/cutils.h" |
| 15 | #include "qemu/iov.h" |
| 16 | #include "qemu/rcu.h" |
| 17 | #include "exec/target_page.h" |
| 18 | #include "system/system.h" |
| 19 | #include "system/ramblock.h" |
| 20 | #include "qemu/error-report.h" |
| 21 | #include "qapi/error.h" |
| 22 | #include "file.h" |
| 23 | #include "migration/misc.h" |
| 24 | #include "migration.h" |
| 25 | #include "migration-stats.h" |
| 26 | #include "savevm.h" |
| 27 | #include "socket.h" |
| 28 | #include "tls.h" |
| 29 | #include "qemu-file.h" |
| 30 | #include "trace.h" |
| 31 | #include "multifd.h" |
| 32 | #include "multifd-colo.h" |
| 33 | #include "options.h" |
| 34 | #include "qemu/yank.h" |
| 35 | #include "io/channel-file.h" |
| 36 | #include "io/channel-socket.h" |
| 37 | #include "yank_functions.h" |
| 38 | |
| 39 | typedef struct { |
| 40 | uint32_t magic; |
| 41 | uint32_t version; |
| 42 | unsigned char uuid[16]; /* QemuUUID */ |
| 43 | uint8_t id; |
| 44 | uint8_t unused1[7]; /* Reserved for future use */ |
| 45 | uint64_t unused2[4]; /* Reserved for future use */ |
| 46 | } __attribute__((packed)) MultiFDInit_t; |
| 47 | |
| 48 | struct { |
| 49 | MultiFDSendParams *params; |
| 50 | |
| 51 | /* multifd_send() body is not thread safe, needs serialization */ |
| 52 | QemuMutex multifd_send_mutex; |
| 53 | |
| 54 | /* |
| 55 | * Global number of generated multifd packets. |
| 56 | * |
| 57 | * Note that we used 'uintptr_t' because it'll naturally support atomic |
| 58 | * operations on both 32bit / 64 bits hosts. It means on 32bit systems |
| 59 | * multifd will overflow the packet_num easier, but that should be |
| 60 | * fine. |
| 61 | */ |
| 62 | uintptr_t packet_num; |
| 63 | /* |
| 64 | * Synchronization point past which no more channels will be |
| 65 | * created. |
| 66 | */ |
| 67 | QemuSemaphore channels_created; |
| 68 | /* send channels ready */ |
| 69 | QemuSemaphore channels_ready; |
| 70 | /* |
| 71 | * Have we already run terminate threads. There is a race when it |
| 72 | * happens that we got one error while we are exiting. |
| 73 | * We will use atomic operations. Only valid values are 0 and 1. |
| 74 | */ |
| 75 | int exiting; |
| 76 | /* multifd ops */ |
| 77 | const MultiFDMethods *ops; |
| 78 | } *multifd_send_state; |
| 79 | |
| 80 | struct { |
| 81 | MultiFDRecvParams *params; |
| 82 | MultiFDRecvData *data; |
| 83 | /* number of created threads */ |
| 84 | int count; |
| 85 | /* |
| 86 | * This is always posted by the recv threads, the migration thread |
| 87 | * uses it to wait for recv threads to finish assigned tasks. |
| 88 | */ |
| 89 | QemuSemaphore sem_sync; |
| 90 | /* global number of generated multifd packets */ |
| 91 | uint64_t packet_num; |
| 92 | int exiting; |
| 93 | /* multifd ops */ |
| 94 | const MultiFDMethods *ops; |
| 95 | } *multifd_recv_state; |
| 96 | |
| 97 | MultiFDSendData *multifd_send_data_alloc(void) |
| 98 | { |
| 99 | MultiFDSendData *new = g_new0(MultiFDSendData, 1); |
| 100 | |
| 101 | multifd_ram_payload_alloc(&new->u.ram); |
| 102 | /* Device state allocates its payload on-demand */ |
| 103 | |
| 104 | return new; |
| 105 | } |
| 106 | |
| 107 | void multifd_send_data_clear(MultiFDSendData *data) |
| 108 | { |
| 109 | if (multifd_payload_empty(data)) { |
| 110 | return; |
| 111 | } |
| 112 | |
| 113 | switch (data->type) { |
| 114 | case MULTIFD_PAYLOAD_DEVICE_STATE: |
| 115 | multifd_send_data_clear_device_state(&data->u.device_state); |
| 116 | break; |
| 117 | default: |
| 118 | /* Nothing to do */ |
| 119 | break; |
| 120 | } |
| 121 | |
| 122 | data->type = MULTIFD_PAYLOAD_NONE; |
| 123 | } |
| 124 | |
| 125 | void multifd_send_data_free(MultiFDSendData *data) |
| 126 | { |
| 127 | if (!data) { |
| 128 | return; |
| 129 | } |
| 130 | |
| 131 | /* This also free's device state payload */ |
| 132 | multifd_send_data_clear(data); |
| 133 | |
| 134 | multifd_ram_payload_free(&data->u.ram); |
| 135 | |
| 136 | g_free(data); |
| 137 | } |
| 138 | |
| 139 | static bool multifd_use_packets(void) |
| 140 | { |
| 141 | return !migrate_mapped_ram(); |
| 142 | } |
| 143 | |
| 144 | void multifd_send_channel_created(void) |
| 145 | { |
| 146 | qemu_sem_post(&multifd_send_state->channels_created); |
| 147 | } |
| 148 | |
| 149 | static const MultiFDMethods *multifd_ops[MULTIFD_COMPRESSION__MAX] = {}; |
| 150 | |
| 151 | void multifd_register_ops(int method, const MultiFDMethods *ops) |
| 152 | { |
| 153 | assert(0 <= method && method < MULTIFD_COMPRESSION__MAX); |
| 154 | assert(!multifd_ops[method]); |
| 155 | multifd_ops[method] = ops; |
| 156 | } |
| 157 | |
| 158 | static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp) |
| 159 | { |
| 160 | MultiFDInit_t msg = {}; |
| 161 | size_t size = sizeof(msg); |
| 162 | int ret; |
| 163 | |
| 164 | msg.magic = cpu_to_be32(MULTIFD_MAGIC); |
| 165 | msg.version = cpu_to_be32(MULTIFD_VERSION); |
| 166 | msg.id = p->id; |
| 167 | memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid)); |
| 168 | |
| 169 | ret = qio_channel_write_all(p->c, (char *)&msg, size, errp); |
| 170 | if (ret != 0) { |
| 171 | return -1; |
| 172 | } |
| 173 | qatomic_add(&mig_stats.multifd_bytes, size); |
| 174 | return 0; |
| 175 | } |
| 176 | |
| 177 | static int multifd_recv_initial_packet(QIOChannel *c, Error **errp) |
| 178 | { |
| 179 | MultiFDInit_t msg; |
| 180 | int ret; |
| 181 | |
| 182 | ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp); |
| 183 | if (ret != 0) { |
| 184 | return -1; |
| 185 | } |
| 186 | |
| 187 | msg.magic = be32_to_cpu(msg.magic); |
| 188 | msg.version = be32_to_cpu(msg.version); |
| 189 | |
| 190 | if (msg.magic != MULTIFD_MAGIC) { |
| 191 | error_setg(errp, "multifd: received packet magic %x " |
| 192 | "expected %x", msg.magic, MULTIFD_MAGIC); |
| 193 | return -1; |
| 194 | } |
| 195 | |
| 196 | if (msg.version != MULTIFD_VERSION) { |
| 197 | error_setg(errp, "multifd: received packet version %u " |
| 198 | "expected %u", msg.version, MULTIFD_VERSION); |
| 199 | return -1; |
| 200 | } |
| 201 | |
| 202 | if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) { |
| 203 | char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid); |
| 204 | char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid); |
| 205 | |
| 206 | error_setg(errp, "multifd: received uuid '%s' and expected " |
| 207 | "uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id); |
| 208 | g_free(uuid); |
| 209 | g_free(msg_uuid); |
| 210 | return -1; |
| 211 | } |
| 212 | |
| 213 | if (msg.id >= migrate_multifd_channels()) { |
| 214 | error_setg(errp, "multifd: received channel id %u exceeds " |
| 215 | "channel count %u", msg.id, migrate_multifd_channels()); |
| 216 | return -1; |
| 217 | } |
| 218 | |
| 219 | return msg.id; |
| 220 | } |
| 221 | |
| 222 | /* Fills a RAM multifd packet */ |
| 223 | void multifd_send_fill_packet(MultiFDSendParams *p) |
| 224 | { |
| 225 | MultiFDPacket_t *packet = p->packet; |
| 226 | uint64_t packet_num; |
| 227 | bool sync_packet = p->flags & MULTIFD_FLAG_SYNC; |
| 228 | |
| 229 | memset(packet, 0, p->packet_len); |
| 230 | |
| 231 | packet->hdr.magic = cpu_to_be32(MULTIFD_MAGIC); |
| 232 | packet->hdr.version = cpu_to_be32(MULTIFD_VERSION); |
| 233 | |
| 234 | packet->hdr.flags = cpu_to_be32(p->flags); |
| 235 | packet->next_packet_size = cpu_to_be32(p->next_packet_size); |
| 236 | |
| 237 | packet_num = qatomic_fetch_inc(&multifd_send_state->packet_num); |
| 238 | packet->packet_num = cpu_to_be64(packet_num); |
| 239 | |
| 240 | p->packets_sent++; |
| 241 | |
| 242 | if (!sync_packet) { |
| 243 | multifd_ram_fill_packet(p); |
| 244 | } |
| 245 | |
| 246 | trace_multifd_send_fill(p->id, packet_num, |
| 247 | p->flags, p->next_packet_size); |
| 248 | } |
| 249 | |
| 250 | static int multifd_recv_unfill_packet_header(MultiFDRecvParams *p, |
| 251 | const MultiFDPacketHdr_t *hdr, |
| 252 | Error **errp) |
| 253 | { |
| 254 | uint32_t magic = be32_to_cpu(hdr->magic); |
| 255 | uint32_t version = be32_to_cpu(hdr->version); |
| 256 | |
| 257 | if (magic != MULTIFD_MAGIC) { |
| 258 | error_setg(errp, "multifd: received packet magic %x, expected %x", |
| 259 | magic, MULTIFD_MAGIC); |
| 260 | return -1; |
| 261 | } |
| 262 | |
| 263 | if (version != MULTIFD_VERSION) { |
| 264 | error_setg(errp, "multifd: received packet version %u, expected %u", |
| 265 | version, MULTIFD_VERSION); |
| 266 | return -1; |
| 267 | } |
| 268 | |
| 269 | p->flags = be32_to_cpu(hdr->flags); |
| 270 | |
| 271 | return 0; |
| 272 | } |
| 273 | |
| 274 | static int multifd_recv_unfill_packet_device_state(MultiFDRecvParams *p, |
| 275 | Error **errp) |
| 276 | { |
| 277 | MultiFDPacketDeviceState_t *packet = p->packet_dev_state; |
| 278 | |
| 279 | packet->instance_id = be32_to_cpu(packet->instance_id); |
| 280 | p->next_packet_size = be32_to_cpu(packet->next_packet_size); |
| 281 | |
| 282 | return 0; |
| 283 | } |
| 284 | |
| 285 | static int multifd_recv_unfill_packet_ram(MultiFDRecvParams *p, Error **errp) |
| 286 | { |
| 287 | const MultiFDPacket_t *packet = p->packet; |
| 288 | int ret = 0; |
| 289 | |
| 290 | p->next_packet_size = be32_to_cpu(packet->next_packet_size); |
| 291 | p->packet_num = be64_to_cpu(packet->packet_num); |
| 292 | |
| 293 | /* Always unfill, old QEMUs (<9.0) send data along with SYNC */ |
| 294 | ret = multifd_ram_unfill_packet(p, errp); |
| 295 | |
| 296 | trace_multifd_recv_unfill(p->id, p->packet_num, p->flags, |
| 297 | p->next_packet_size); |
| 298 | |
| 299 | return ret; |
| 300 | } |
| 301 | |
| 302 | static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp) |
| 303 | { |
| 304 | p->packets_recved++; |
| 305 | |
| 306 | if (p->flags & MULTIFD_FLAG_DEVICE_STATE) { |
| 307 | return multifd_recv_unfill_packet_device_state(p, errp); |
| 308 | } |
| 309 | |
| 310 | return multifd_recv_unfill_packet_ram(p, errp); |
| 311 | } |
| 312 | |
| 313 | static bool multifd_send_should_exit(void) |
| 314 | { |
| 315 | return qatomic_read(&multifd_send_state->exiting); |
| 316 | } |
| 317 | |
| 318 | static bool multifd_recv_should_exit(void) |
| 319 | { |
| 320 | return qatomic_read(&multifd_recv_state->exiting); |
| 321 | } |
| 322 | |
| 323 | /* |
| 324 | * The migration thread can wait on either of the two semaphores. This |
| 325 | * function can be used to kick the main thread out of waiting on either of |
| 326 | * them. Should mostly only be called when something wrong happened with |
| 327 | * the current multifd send thread. |
| 328 | */ |
| 329 | static void multifd_send_kick_main(MultiFDSendParams *p) |
| 330 | { |
| 331 | qemu_sem_post(&p->sem_sync); |
| 332 | qemu_sem_post(&multifd_send_state->channels_ready); |
| 333 | } |
| 334 | |
| 335 | /* |
| 336 | * multifd_send() works by exchanging the MultiFDSendData object |
| 337 | * provided by the caller with an unused MultiFDSendData object from |
| 338 | * the next channel that is found to be idle. |
| 339 | * |
| 340 | * The channel owns the data until it finishes transmitting and the |
| 341 | * caller owns the empty object until it fills it with data and calls |
| 342 | * this function again. No locking necessary. |
| 343 | * |
| 344 | * Switching is safe because both the migration thread and the channel |
| 345 | * thread have barriers in place to serialize access. |
| 346 | * |
| 347 | * Returns true if succeed, false otherwise. |
| 348 | */ |
| 349 | bool multifd_send(MultiFDSendData **send_data) |
| 350 | { |
| 351 | int i; |
| 352 | static int next_channel; |
| 353 | MultiFDSendParams *p = NULL; /* make happy gcc */ |
| 354 | MultiFDSendData *tmp; |
| 355 | |
| 356 | if (multifd_send_should_exit()) { |
| 357 | return false; |
| 358 | } |
| 359 | |
| 360 | QEMU_LOCK_GUARD(&multifd_send_state->multifd_send_mutex); |
| 361 | |
| 362 | /* We wait here, until at least one channel is ready */ |
| 363 | qemu_sem_wait(&multifd_send_state->channels_ready); |
| 364 | |
| 365 | int thread_count = migrate_multifd_channels(); |
| 366 | |
| 367 | /* |
| 368 | * next_channel can remain from a previous migration that was |
| 369 | * using more channels, so ensure it doesn't overflow if the |
| 370 | * limit is lower now. |
| 371 | */ |
| 372 | next_channel %= thread_count; |
| 373 | for (i = next_channel;; i = (i + 1) % thread_count) { |
| 374 | if (multifd_send_should_exit()) { |
| 375 | return false; |
| 376 | } |
| 377 | p = &multifd_send_state->params[i]; |
| 378 | /* |
| 379 | * Lockless read to p->pending_job is safe, because only multifd |
| 380 | * sender thread can clear it. |
| 381 | */ |
| 382 | if (qatomic_read(&p->pending_job) == false) { |
| 383 | next_channel = (i + 1) % thread_count; |
| 384 | break; |
| 385 | } |
| 386 | } |
| 387 | |
| 388 | /* |
| 389 | * Make sure we read p->pending_job before all the rest. Pairs with |
| 390 | * qatomic_store_release() in multifd_send_thread(). |
| 391 | */ |
| 392 | smp_mb_acquire(); |
| 393 | |
| 394 | assert(multifd_payload_empty(p->data)); |
| 395 | |
| 396 | /* |
| 397 | * Swap the pointers. The channel gets the client data for |
| 398 | * transferring and the client gets back an unused data slot. |
| 399 | */ |
| 400 | tmp = *send_data; |
| 401 | *send_data = p->data; |
| 402 | p->data = tmp; |
| 403 | |
| 404 | /* |
| 405 | * Making sure p->data is setup before marking pending_job=true. Pairs |
| 406 | * with the qatomic_load_acquire() in multifd_send_thread(). |
| 407 | */ |
| 408 | qatomic_store_release(&p->pending_job, true); |
| 409 | qemu_sem_post(&p->sem); |
| 410 | |
| 411 | return true; |
| 412 | } |
| 413 | |
| 414 | /* Multifd send side hit an error; remember it and prepare to quit */ |
| 415 | static void multifd_send_error_propagate(Error *err) |
| 416 | { |
| 417 | MigrationState *s = migrate_get_current(); |
| 418 | |
| 419 | /* |
| 420 | * There may be independent errors in each thread. Propagate the |
| 421 | * first and free the subsequent ones. |
| 422 | */ |
| 423 | if (qatomic_xchg(&multifd_send_state->exiting, 1)) { |
| 424 | error_free(err); |
| 425 | return; |
| 426 | } |
| 427 | |
| 428 | migrate_error_propagate(s, err); |
| 429 | |
| 430 | if (s->state == MIGRATION_STATUS_SETUP || |
| 431 | s->state == MIGRATION_STATUS_PRE_SWITCHOVER || |
| 432 | s->state == MIGRATION_STATUS_DEVICE || |
| 433 | s->state == MIGRATION_STATUS_ACTIVE) { |
| 434 | migrate_set_state(&s->state, s->state, |
| 435 | MIGRATION_STATUS_FAILING); |
| 436 | } |
| 437 | } |
| 438 | |
| 439 | /* |
| 440 | * Gracefully shutdown IOChannels. Only needed for successful migrations on |
| 441 | * top of TLS channels. Otherwise it is same to qio_channel_shutdown(). |
| 442 | * |
| 443 | * A successful migration also guarantees multifd sender threads are |
| 444 | * properly flushed and halted. It is only safe to send BYE in the |
| 445 | * migration thread here when we know there's no other thread writting to |
| 446 | * the channel, because GnuTLS doesn't support concurrent writers. |
| 447 | */ |
| 448 | static void migration_ioc_shutdown_gracefully(QIOChannel *ioc) |
| 449 | { |
| 450 | Error *local_err = NULL; |
| 451 | |
| 452 | if (!migration_has_failed(migrate_get_current()) && |
| 453 | object_dynamic_cast((Object *)ioc, TYPE_QIO_CHANNEL_TLS)) { |
| 454 | |
| 455 | /* |
| 456 | * The destination expects the TLS session to always be properly |
| 457 | * terminated. This helps to detect a premature termination in the |
| 458 | * middle of the stream. Note that older QEMUs always break the |
| 459 | * connection on the source and the destination always sees |
| 460 | * GNUTLS_E_PREMATURE_TERMINATION. |
| 461 | */ |
| 462 | migration_tls_channel_end(ioc, &local_err); |
| 463 | if (local_err) { |
| 464 | warn_reportf_err(local_err, |
| 465 | "Failed to gracefully terminate TLS connection: "); |
| 466 | } |
| 467 | } |
| 468 | |
| 469 | qio_channel_shutdown(ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL); |
| 470 | } |
| 471 | |
| 472 | static void multifd_send_terminate_threads(void) |
| 473 | { |
| 474 | int i; |
| 475 | |
| 476 | trace_multifd_send_terminate_threads(); |
| 477 | |
| 478 | /* |
| 479 | * Tell everyone we're quitting. No xchg() needed here; we simply |
| 480 | * always set it. |
| 481 | */ |
| 482 | qatomic_set(&multifd_send_state->exiting, 1); |
| 483 | |
| 484 | /* |
| 485 | * Firstly, kick all threads out; no matter whether they are just idle, |
| 486 | * or blocked in an IO system call. |
| 487 | */ |
| 488 | for (i = 0; i < migrate_multifd_channels(); i++) { |
| 489 | MultiFDSendParams *p = &multifd_send_state->params[i]; |
| 490 | |
| 491 | qemu_sem_post(&p->sem); |
| 492 | if (p->c) { |
| 493 | migration_ioc_shutdown_gracefully(p->c); |
| 494 | } |
| 495 | } |
| 496 | |
| 497 | /* |
| 498 | * Finally recycle all the threads. |
| 499 | */ |
| 500 | for (i = 0; i < migrate_multifd_channels(); i++) { |
| 501 | MultiFDSendParams *p = &multifd_send_state->params[i]; |
| 502 | |
| 503 | if (p->tls_thread_created) { |
| 504 | qemu_thread_join(&p->tls_thread); |
| 505 | } |
| 506 | |
| 507 | if (p->thread_created) { |
| 508 | qemu_thread_join(&p->thread); |
| 509 | } |
| 510 | } |
| 511 | } |
| 512 | |
| 513 | static bool multifd_send_cleanup_channel(MultiFDSendParams *p, Error **errp) |
| 514 | { |
| 515 | if (p->c) { |
| 516 | migration_ioc_unregister_yank(p->c); |
| 517 | /* |
| 518 | * The object_unref() cannot guarantee the fd will always be |
| 519 | * released because finalize() of the iochannel is only |
| 520 | * triggered on the last reference and it's not guaranteed |
| 521 | * that we always hold the last refcount when reaching here. |
| 522 | * |
| 523 | * Closing the fd explicitly has the benefit that if there is any |
| 524 | * registered I/O handler callbacks on such fd, that will get a |
| 525 | * POLLNVAL event and will further trigger the cleanup to finally |
| 526 | * release the IOC. |
| 527 | * |
| 528 | * FIXME: It should logically be guaranteed that all multifd |
| 529 | * channels have no I/O handler callback registered when reaching |
| 530 | * here, because migration thread will wait for all multifd channel |
| 531 | * establishments to complete during setup. Since |
| 532 | * migration_cleanup() will be scheduled in main thread too, all |
| 533 | * previous callbacks should guarantee to be completed when |
| 534 | * reaching here. See multifd_send_state.channels_created and its |
| 535 | * usage. In the future, we could replace this with an assert |
| 536 | * making sure we're the last reference, or simply drop it if above |
| 537 | * is more clear to be justified. |
| 538 | */ |
| 539 | qio_channel_close(p->c, &error_abort); |
| 540 | object_unref(OBJECT(p->c)); |
| 541 | p->c = NULL; |
| 542 | } |
| 543 | qemu_sem_destroy(&p->sem); |
| 544 | qemu_sem_destroy(&p->sem_sync); |
| 545 | g_free(p->name); |
| 546 | p->name = NULL; |
| 547 | g_clear_pointer(&p->data, multifd_send_data_free); |
| 548 | p->packet_len = 0; |
| 549 | g_clear_pointer(&p->packet_device_state, g_free); |
| 550 | g_free(p->packet); |
| 551 | p->packet = NULL; |
| 552 | multifd_send_state->ops->send_cleanup(p, errp); |
| 553 | assert(!p->iov); |
| 554 | |
| 555 | return *errp == NULL; |
| 556 | } |
| 557 | |
| 558 | static void multifd_send_cleanup_state(void) |
| 559 | { |
| 560 | file_cleanup_outgoing_migration(); |
| 561 | socket_cleanup_outgoing_migration(); |
| 562 | multifd_device_state_send_cleanup(); |
| 563 | qemu_sem_destroy(&multifd_send_state->channels_created); |
| 564 | qemu_sem_destroy(&multifd_send_state->channels_ready); |
| 565 | qemu_mutex_destroy(&multifd_send_state->multifd_send_mutex); |
| 566 | g_free(multifd_send_state->params); |
| 567 | multifd_send_state->params = NULL; |
| 568 | g_free(multifd_send_state); |
| 569 | multifd_send_state = NULL; |
| 570 | } |
| 571 | |
| 572 | void multifd_send_shutdown(void) |
| 573 | { |
| 574 | int i; |
| 575 | |
| 576 | if (!migrate_multifd()) { |
| 577 | return; |
| 578 | } |
| 579 | |
| 580 | multifd_send_terminate_threads(); |
| 581 | |
| 582 | for (i = 0; i < migrate_multifd_channels(); i++) { |
| 583 | MultiFDSendParams *p = &multifd_send_state->params[i]; |
| 584 | Error *local_err = NULL; |
| 585 | |
| 586 | if (!multifd_send_cleanup_channel(p, &local_err)) { |
| 587 | migrate_error_propagate(migrate_get_current(), local_err); |
| 588 | } |
| 589 | } |
| 590 | |
| 591 | multifd_send_cleanup_state(); |
| 592 | } |
| 593 | |
| 594 | static int multifd_zero_copy_flush(QIOChannel *c) |
| 595 | { |
| 596 | int ret; |
| 597 | Error *err = NULL; |
| 598 | |
| 599 | ret = qio_channel_flush(c, &err); |
| 600 | if (ret < 0) { |
| 601 | error_report_err(err); |
| 602 | return -1; |
| 603 | } |
| 604 | if (ret == 1) { |
| 605 | qatomic_add(&mig_stats.dirty_sync_missed_zero_copy, 1); |
| 606 | } |
| 607 | |
| 608 | return ret; |
| 609 | } |
| 610 | |
| 611 | int multifd_send_sync_main(MultiFDSyncReq req) |
| 612 | { |
| 613 | int i; |
| 614 | int thread_count; |
| 615 | bool flush_zero_copy; |
| 616 | |
| 617 | assert(req != MULTIFD_SYNC_NONE); |
| 618 | |
| 619 | thread_count = migrate_multifd_channels(); |
| 620 | flush_zero_copy = migrate_zero_copy_send(); |
| 621 | |
| 622 | for (i = 0; i < thread_count; i++) { |
| 623 | MultiFDSendParams *p = &multifd_send_state->params[i]; |
| 624 | |
| 625 | if (multifd_send_should_exit()) { |
| 626 | return -1; |
| 627 | } |
| 628 | |
| 629 | trace_multifd_send_sync_main_signal(p->id); |
| 630 | |
| 631 | /* |
| 632 | * We should be the only user so far, so not possible to be set by |
| 633 | * others concurrently. |
| 634 | */ |
| 635 | assert(qatomic_read(&p->pending_sync) == MULTIFD_SYNC_NONE); |
| 636 | qatomic_set(&p->pending_sync, req); |
| 637 | qemu_sem_post(&p->sem); |
| 638 | } |
| 639 | for (i = 0; i < thread_count; i++) { |
| 640 | MultiFDSendParams *p = &multifd_send_state->params[i]; |
| 641 | |
| 642 | if (multifd_send_should_exit()) { |
| 643 | return -1; |
| 644 | } |
| 645 | |
| 646 | qemu_sem_wait(&multifd_send_state->channels_ready); |
| 647 | trace_multifd_send_sync_main_wait(p->id); |
| 648 | qemu_sem_wait(&p->sem_sync); |
| 649 | |
| 650 | if (flush_zero_copy && p->c && (multifd_zero_copy_flush(p->c) < 0)) { |
| 651 | return -1; |
| 652 | } |
| 653 | } |
| 654 | trace_multifd_send_sync_main(multifd_send_state->packet_num); |
| 655 | |
| 656 | return 0; |
| 657 | } |
| 658 | |
| 659 | static void *multifd_send_thread(void *opaque) |
| 660 | { |
| 661 | MultiFDSendParams *p = opaque; |
| 662 | Error *local_err = NULL; |
| 663 | int ret = 0; |
| 664 | bool use_packets = multifd_use_packets(); |
| 665 | |
| 666 | trace_multifd_send_thread_start(p->id); |
| 667 | rcu_register_thread(); |
| 668 | |
| 669 | if (use_packets) { |
| 670 | if (multifd_send_initial_packet(p, &local_err) < 0) { |
| 671 | ret = -1; |
| 672 | goto out; |
| 673 | } |
| 674 | } |
| 675 | |
| 676 | while (true) { |
| 677 | qemu_sem_post(&multifd_send_state->channels_ready); |
| 678 | qemu_sem_wait(&p->sem); |
| 679 | |
| 680 | if (multifd_send_should_exit()) { |
| 681 | break; |
| 682 | } |
| 683 | |
| 684 | /* |
| 685 | * Read pending_job flag before p->data. Pairs with the |
| 686 | * qatomic_store_release() in multifd_send(). |
| 687 | */ |
| 688 | if (qatomic_load_acquire(&p->pending_job)) { |
| 689 | bool is_device_state = multifd_payload_device_state(p->data); |
| 690 | size_t total_size; |
| 691 | int write_flags_masked = 0; |
| 692 | |
| 693 | p->flags = 0; |
| 694 | p->iovs_num = 0; |
| 695 | assert(!multifd_payload_empty(p->data)); |
| 696 | |
| 697 | if (is_device_state) { |
| 698 | multifd_device_state_send_prepare(p); |
| 699 | |
| 700 | /* Device state packets cannot be sent via zerocopy */ |
| 701 | write_flags_masked |= QIO_CHANNEL_WRITE_FLAG_ZERO_COPY; |
| 702 | } else { |
| 703 | ret = multifd_send_state->ops->send_prepare(p, &local_err); |
| 704 | if (ret != 0) { |
| 705 | break; |
| 706 | } |
| 707 | } |
| 708 | |
| 709 | /* |
| 710 | * The packet header in the zerocopy RAM case is accounted for |
| 711 | * in multifd_nocomp_send_prepare() - where it is actually |
| 712 | * being sent. |
| 713 | */ |
| 714 | total_size = iov_size(p->iov, p->iovs_num); |
| 715 | |
| 716 | if (migrate_mapped_ram()) { |
| 717 | assert(!is_device_state); |
| 718 | |
| 719 | ret = file_write_ramblock_iov(p->c, p->iov, p->iovs_num, |
| 720 | &p->data->u.ram, &local_err); |
| 721 | } else { |
| 722 | ret = qio_channel_writev_full_all(p->c, p->iov, p->iovs_num, |
| 723 | NULL, 0, |
| 724 | p->write_flags & ~write_flags_masked, |
| 725 | &local_err); |
| 726 | } |
| 727 | |
| 728 | if (ret != 0) { |
| 729 | break; |
| 730 | } |
| 731 | |
| 732 | qatomic_add(&mig_stats.multifd_bytes, total_size); |
| 733 | |
| 734 | p->next_packet_size = 0; |
| 735 | multifd_send_data_clear(p->data); |
| 736 | |
| 737 | /* |
| 738 | * Making sure p->data is published before saying "we're |
| 739 | * free". Pairs with the smp_mb_acquire() in |
| 740 | * multifd_send(). |
| 741 | */ |
| 742 | qatomic_store_release(&p->pending_job, false); |
| 743 | } else { |
| 744 | MultiFDSyncReq req = qatomic_read(&p->pending_sync); |
| 745 | |
| 746 | /* |
| 747 | * If not a normal job, must be a sync request. Note that |
| 748 | * pending_sync is a standalone flag (unlike pending_job), so |
| 749 | * it doesn't require explicit memory barriers. |
| 750 | */ |
| 751 | assert(req != MULTIFD_SYNC_NONE); |
| 752 | |
| 753 | /* Only push the SYNC message if it involves a remote sync */ |
| 754 | if (req == MULTIFD_SYNC_ALL) { |
| 755 | p->flags = MULTIFD_FLAG_SYNC; |
| 756 | multifd_send_fill_packet(p); |
| 757 | ret = qio_channel_write_all(p->c, (void *)p->packet, |
| 758 | p->packet_len, &local_err); |
| 759 | if (ret != 0) { |
| 760 | break; |
| 761 | } |
| 762 | /* p->next_packet_size will always be zero for a SYNC packet */ |
| 763 | qatomic_add(&mig_stats.multifd_bytes, p->packet_len); |
| 764 | } |
| 765 | |
| 766 | qatomic_set(&p->pending_sync, MULTIFD_SYNC_NONE); |
| 767 | qemu_sem_post(&p->sem_sync); |
| 768 | } |
| 769 | } |
| 770 | |
| 771 | out: |
| 772 | if (ret) { |
| 773 | assert(local_err); |
| 774 | trace_multifd_send_error(p->id); |
| 775 | multifd_send_error_propagate(local_err); |
| 776 | } |
| 777 | |
| 778 | /* |
| 779 | * Always kick the main thread: The main thread might wait on this thread |
| 780 | * while another thread encounters an error and signals this thread to exit. |
| 781 | */ |
| 782 | multifd_send_kick_main(p); |
| 783 | |
| 784 | rcu_unregister_thread(); |
| 785 | trace_multifd_send_thread_end(p->id, p->packets_sent); |
| 786 | |
| 787 | return NULL; |
| 788 | } |
| 789 | |
| 790 | static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque); |
| 791 | |
| 792 | typedef struct { |
| 793 | MultiFDSendParams *p; |
| 794 | QIOChannelTLS *tioc; |
| 795 | } MultiFDTLSThreadArgs; |
| 796 | |
| 797 | static void *multifd_tls_handshake_thread(void *opaque) |
| 798 | { |
| 799 | MultiFDTLSThreadArgs *args = opaque; |
| 800 | |
| 801 | qio_channel_tls_handshake(args->tioc, |
| 802 | multifd_new_send_channel_async, |
| 803 | args->p, |
| 804 | NULL, |
| 805 | NULL); |
| 806 | g_free(args); |
| 807 | |
| 808 | return NULL; |
| 809 | } |
| 810 | |
| 811 | static bool multifd_tls_channel_connect(MultiFDSendParams *p, |
| 812 | QIOChannel *ioc, |
| 813 | Error **errp) |
| 814 | { |
| 815 | MultiFDTLSThreadArgs *args; |
| 816 | QIOChannelTLS *tioc; |
| 817 | |
| 818 | tioc = migration_tls_client_create(ioc, errp); |
| 819 | if (!tioc) { |
| 820 | return false; |
| 821 | } |
| 822 | |
| 823 | /* |
| 824 | * Ownership of the socket channel now transfers to the newly |
| 825 | * created TLS channel, which has already taken a reference. |
| 826 | */ |
| 827 | object_unref(OBJECT(ioc)); |
| 828 | trace_multifd_tls_outgoing_handshake_start(ioc, tioc); |
| 829 | qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing"); |
| 830 | |
| 831 | args = g_new0(MultiFDTLSThreadArgs, 1); |
| 832 | args->tioc = tioc; |
| 833 | args->p = p; |
| 834 | |
| 835 | p->tls_thread_created = true; |
| 836 | qemu_thread_create(&p->tls_thread, MIGRATION_THREAD_SRC_TLS, |
| 837 | multifd_tls_handshake_thread, args, |
| 838 | QEMU_THREAD_JOINABLE); |
| 839 | return true; |
| 840 | } |
| 841 | |
| 842 | void multifd_channel_connect(MultiFDSendParams *p, QIOChannel *ioc) |
| 843 | { |
| 844 | qio_channel_set_delay(ioc, false); |
| 845 | |
| 846 | migration_ioc_register_yank(ioc); |
| 847 | /* Setup p->c only if the channel is completely setup */ |
| 848 | p->c = ioc; |
| 849 | |
| 850 | p->thread_created = true; |
| 851 | qemu_thread_create(&p->thread, p->name, multifd_send_thread, p, |
| 852 | QEMU_THREAD_JOINABLE); |
| 853 | } |
| 854 | |
| 855 | /* |
| 856 | * When TLS is enabled this function is called once to establish the |
| 857 | * TLS connection and a second time after the TLS handshake to create |
| 858 | * the multifd channel. Without TLS it goes straight into the channel |
| 859 | * creation. |
| 860 | */ |
| 861 | static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque) |
| 862 | { |
| 863 | MultiFDSendParams *p = opaque; |
| 864 | QIOChannel *ioc = QIO_CHANNEL(qio_task_get_source(task)); |
| 865 | Error *local_err = NULL; |
| 866 | bool ret; |
| 867 | |
| 868 | trace_multifd_new_send_channel_async(p->id); |
| 869 | |
| 870 | if (qio_task_propagate_error(task, &local_err)) { |
| 871 | ret = false; |
| 872 | goto out; |
| 873 | } |
| 874 | |
| 875 | trace_multifd_set_outgoing_channel(ioc, object_get_typename(OBJECT(ioc))); |
| 876 | |
| 877 | if (migrate_channel_requires_tls_upgrade(ioc)) { |
| 878 | ret = multifd_tls_channel_connect(p, ioc, &local_err); |
| 879 | if (ret) { |
| 880 | return; |
| 881 | } |
| 882 | } else { |
| 883 | multifd_channel_connect(p, ioc); |
| 884 | ret = true; |
| 885 | } |
| 886 | |
| 887 | out: |
| 888 | /* |
| 889 | * Here we're not interested whether creation succeeded, only that |
| 890 | * it happened at all. |
| 891 | */ |
| 892 | multifd_send_channel_created(); |
| 893 | |
| 894 | if (ret) { |
| 895 | return; |
| 896 | } |
| 897 | |
| 898 | trace_multifd_new_send_channel_async_error(p->id, local_err); |
| 899 | multifd_send_error_propagate(local_err); |
| 900 | /* |
| 901 | * For error cases (TLS or non-TLS), IO channel is always freed here |
| 902 | * rather than when cleanup multifd: since p->c is not set, multifd |
| 903 | * cleanup code doesn't even know its existence. |
| 904 | */ |
| 905 | object_unref(OBJECT(ioc)); |
| 906 | } |
| 907 | |
| 908 | static bool multifd_new_send_channel_create(gpointer opaque, Error **errp) |
| 909 | { |
| 910 | if (!multifd_use_packets()) { |
| 911 | return file_send_channel_create(opaque, errp); |
| 912 | } |
| 913 | |
| 914 | socket_send_channel_create(multifd_new_send_channel_async, opaque); |
| 915 | return true; |
| 916 | } |
| 917 | |
| 918 | bool multifd_send_setup(void) |
| 919 | { |
| 920 | MigrationState *s = migrate_get_current(); |
| 921 | int thread_count, ret = 0; |
| 922 | uint32_t page_count = multifd_ram_page_count(); |
| 923 | bool use_packets = multifd_use_packets(); |
| 924 | uint8_t i; |
| 925 | |
| 926 | if (!migrate_multifd()) { |
| 927 | return true; |
| 928 | } |
| 929 | |
| 930 | thread_count = migrate_multifd_channels(); |
| 931 | multifd_send_state = g_malloc0(sizeof(*multifd_send_state)); |
| 932 | multifd_send_state->params = g_new0(MultiFDSendParams, thread_count); |
| 933 | qemu_mutex_init(&multifd_send_state->multifd_send_mutex); |
| 934 | qemu_sem_init(&multifd_send_state->channels_created, 0); |
| 935 | qemu_sem_init(&multifd_send_state->channels_ready, 0); |
| 936 | qatomic_set(&multifd_send_state->exiting, 0); |
| 937 | multifd_send_state->ops = multifd_ops[migrate_multifd_compression()]; |
| 938 | |
| 939 | for (i = 0; i < thread_count; i++) { |
| 940 | MultiFDSendParams *p = &multifd_send_state->params[i]; |
| 941 | Error *local_err = NULL; |
| 942 | |
| 943 | qemu_sem_init(&p->sem, 0); |
| 944 | qemu_sem_init(&p->sem_sync, 0); |
| 945 | p->id = i; |
| 946 | p->data = multifd_send_data_alloc(); |
| 947 | |
| 948 | if (use_packets) { |
| 949 | p->packet_len = sizeof(MultiFDPacket_t) |
| 950 | + sizeof(uint64_t) * page_count; |
| 951 | p->packet = g_malloc0(p->packet_len); |
| 952 | p->packet_device_state = g_malloc0(sizeof(*p->packet_device_state)); |
| 953 | p->packet_device_state->hdr.magic = cpu_to_be32(MULTIFD_MAGIC); |
| 954 | p->packet_device_state->hdr.version = cpu_to_be32(MULTIFD_VERSION); |
| 955 | } |
| 956 | p->name = g_strdup_printf(MIGRATION_THREAD_SRC_MULTIFD, i); |
| 957 | p->write_flags = 0; |
| 958 | |
| 959 | if (!multifd_new_send_channel_create(p, &local_err)) { |
| 960 | migrate_error_propagate(s, local_err); |
| 961 | ret = -1; |
| 962 | } |
| 963 | } |
| 964 | |
| 965 | /* |
| 966 | * Wait until channel creation has started for all channels. The |
| 967 | * creation can still fail, but no more channels will be created |
| 968 | * past this point. |
| 969 | */ |
| 970 | for (i = 0; i < thread_count; i++) { |
| 971 | qemu_sem_wait(&multifd_send_state->channels_created); |
| 972 | } |
| 973 | |
| 974 | if (ret) { |
| 975 | goto err; |
| 976 | } |
| 977 | |
| 978 | for (i = 0; i < thread_count; i++) { |
| 979 | MultiFDSendParams *p = &multifd_send_state->params[i]; |
| 980 | Error *local_err = NULL; |
| 981 | |
| 982 | ret = multifd_send_state->ops->send_setup(p, &local_err); |
| 983 | if (ret) { |
| 984 | migrate_error_propagate(s, local_err); |
| 985 | goto err; |
| 986 | } |
| 987 | assert(p->iov); |
| 988 | } |
| 989 | |
| 990 | multifd_device_state_send_setup(); |
| 991 | |
| 992 | return true; |
| 993 | |
| 994 | err: |
| 995 | migrate_set_state(&s->state, MIGRATION_STATUS_SETUP, |
| 996 | MIGRATION_STATUS_FAILING); |
| 997 | return false; |
| 998 | } |
| 999 | |
| 1000 | bool multifd_recv(void) |
| 1001 | { |
| 1002 | int i; |
| 1003 | static int next_recv_channel; |
| 1004 | MultiFDRecvParams *p = NULL; |
| 1005 | int thread_count = migrate_multifd_channels(); |
| 1006 | MultiFDRecvData *data = multifd_recv_state->data; |
| 1007 | |
| 1008 | /* |
| 1009 | * next_channel can remain from a previous migration that was |
| 1010 | * using more channels, so ensure it doesn't overflow if the |
| 1011 | * limit is lower now. |
| 1012 | */ |
| 1013 | next_recv_channel %= thread_count; |
| 1014 | for (i = next_recv_channel;; i = (i + 1) % thread_count) { |
| 1015 | if (multifd_recv_should_exit()) { |
| 1016 | return false; |
| 1017 | } |
| 1018 | |
| 1019 | p = &multifd_recv_state->params[i]; |
| 1020 | |
| 1021 | if (qatomic_read(&p->pending_job) == false) { |
| 1022 | next_recv_channel = (i + 1) % thread_count; |
| 1023 | break; |
| 1024 | } |
| 1025 | } |
| 1026 | |
| 1027 | /* |
| 1028 | * Order pending_job read before manipulating p->data below. Pairs |
| 1029 | * with qatomic_store_release() at multifd_recv_thread(). |
| 1030 | */ |
| 1031 | smp_mb_acquire(); |
| 1032 | |
| 1033 | assert(!p->data->size); |
| 1034 | multifd_recv_state->data = p->data; |
| 1035 | p->data = data; |
| 1036 | |
| 1037 | /* |
| 1038 | * Order p->data update before setting pending_job. Pairs with |
| 1039 | * qatomic_load_acquire() at multifd_recv_thread(). |
| 1040 | */ |
| 1041 | qatomic_store_release(&p->pending_job, true); |
| 1042 | qemu_sem_post(&p->sem); |
| 1043 | |
| 1044 | return true; |
| 1045 | } |
| 1046 | |
| 1047 | MultiFDRecvData *multifd_get_recv_data(void) |
| 1048 | { |
| 1049 | return multifd_recv_state->data; |
| 1050 | } |
| 1051 | |
| 1052 | static void multifd_recv_terminate_threads(Error *err) |
| 1053 | { |
| 1054 | int i; |
| 1055 | |
| 1056 | trace_multifd_recv_terminate_threads(err != NULL); |
| 1057 | |
| 1058 | if (qatomic_xchg(&multifd_recv_state->exiting, 1)) { |
| 1059 | error_free(err); |
| 1060 | return; |
| 1061 | } |
| 1062 | |
| 1063 | if (err) { |
| 1064 | MigrationState *s = migrate_get_current(); |
| 1065 | |
| 1066 | migrate_error_propagate(s, err); |
| 1067 | |
| 1068 | if (s->state == MIGRATION_STATUS_SETUP || |
| 1069 | s->state == MIGRATION_STATUS_ACTIVE) { |
| 1070 | migrate_set_state(&s->state, s->state, |
| 1071 | MIGRATION_STATUS_FAILED); |
| 1072 | } |
| 1073 | } |
| 1074 | |
| 1075 | for (i = 0; i < migrate_multifd_channels(); i++) { |
| 1076 | MultiFDRecvParams *p = &multifd_recv_state->params[i]; |
| 1077 | |
| 1078 | /* |
| 1079 | * The migration thread and channels interact differently |
| 1080 | * depending on the presence of packets. |
| 1081 | */ |
| 1082 | if (multifd_use_packets()) { |
| 1083 | /* |
| 1084 | * The channel receives as long as there are packets. When |
| 1085 | * packets end (i.e. MULTIFD_FLAG_SYNC is reached), the |
| 1086 | * channel waits for the migration thread to sync. If the |
| 1087 | * sync never happens, do it here. |
| 1088 | */ |
| 1089 | qemu_sem_post(&p->sem_sync); |
| 1090 | } else { |
| 1091 | /* |
| 1092 | * The channel waits for the migration thread to give it |
| 1093 | * work. When the migration thread runs out of work, it |
| 1094 | * releases the channel and waits for any pending work to |
| 1095 | * finish. If we reach here (e.g. due to error) before the |
| 1096 | * work runs out, release the channel. |
| 1097 | */ |
| 1098 | qemu_sem_post(&p->sem); |
| 1099 | } |
| 1100 | |
| 1101 | /* |
| 1102 | * We could arrive here for two reasons: |
| 1103 | * - normal quit, i.e. everything went fine, just finished |
| 1104 | * - error quit: We close the channels so the channel threads |
| 1105 | * finish the qio_channel_read_all_eof() |
| 1106 | */ |
| 1107 | if (p->c) { |
| 1108 | qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL); |
| 1109 | } |
| 1110 | } |
| 1111 | } |
| 1112 | |
| 1113 | void multifd_recv_shutdown(void) |
| 1114 | { |
| 1115 | if (migrate_multifd()) { |
| 1116 | multifd_recv_terminate_threads(NULL); |
| 1117 | } |
| 1118 | } |
| 1119 | |
| 1120 | static void multifd_recv_cleanup_channel(MultiFDRecvParams *p) |
| 1121 | { |
| 1122 | migration_ioc_unregister_yank(p->c); |
| 1123 | object_unref(OBJECT(p->c)); |
| 1124 | p->c = NULL; |
| 1125 | qemu_mutex_destroy(&p->mutex); |
| 1126 | qemu_sem_destroy(&p->sem_sync); |
| 1127 | qemu_sem_destroy(&p->sem); |
| 1128 | g_free(p->data); |
| 1129 | p->data = NULL; |
| 1130 | g_free(p->name); |
| 1131 | p->name = NULL; |
| 1132 | p->packet_len = 0; |
| 1133 | g_free(p->packet); |
| 1134 | p->packet = NULL; |
| 1135 | g_clear_pointer(&p->packet_dev_state, g_free); |
| 1136 | g_free(p->normal); |
| 1137 | p->normal = NULL; |
| 1138 | g_free(p->zero); |
| 1139 | p->zero = NULL; |
| 1140 | multifd_recv_state->ops->recv_cleanup(p); |
| 1141 | } |
| 1142 | |
| 1143 | static void multifd_recv_cleanup_state(void) |
| 1144 | { |
| 1145 | qemu_sem_destroy(&multifd_recv_state->sem_sync); |
| 1146 | g_free(multifd_recv_state->params); |
| 1147 | multifd_recv_state->params = NULL; |
| 1148 | g_free(multifd_recv_state->data); |
| 1149 | multifd_recv_state->data = NULL; |
| 1150 | g_free(multifd_recv_state); |
| 1151 | multifd_recv_state = NULL; |
| 1152 | } |
| 1153 | |
| 1154 | void multifd_recv_cleanup(void) |
| 1155 | { |
| 1156 | int i; |
| 1157 | |
| 1158 | if (!migrate_multifd()) { |
| 1159 | return; |
| 1160 | } |
| 1161 | multifd_recv_terminate_threads(NULL); |
| 1162 | for (i = 0; i < migrate_multifd_channels(); i++) { |
| 1163 | MultiFDRecvParams *p = &multifd_recv_state->params[i]; |
| 1164 | |
| 1165 | if (p->thread_created) { |
| 1166 | qemu_thread_join(&p->thread); |
| 1167 | } |
| 1168 | } |
| 1169 | for (i = 0; i < migrate_multifd_channels(); i++) { |
| 1170 | multifd_recv_cleanup_channel(&multifd_recv_state->params[i]); |
| 1171 | } |
| 1172 | multifd_recv_cleanup_state(); |
| 1173 | } |
| 1174 | |
| 1175 | void multifd_recv_sync_main(void) |
| 1176 | { |
| 1177 | int thread_count = migrate_multifd_channels(); |
| 1178 | bool file_based = !multifd_use_packets(); |
| 1179 | int i; |
| 1180 | |
| 1181 | if (!migrate_multifd()) { |
| 1182 | return; |
| 1183 | } |
| 1184 | |
| 1185 | /* |
| 1186 | * File-based channels don't use packets and therefore need to |
| 1187 | * wait for more work. Release them to start the sync. |
| 1188 | */ |
| 1189 | if (file_based) { |
| 1190 | for (i = 0; i < thread_count; i++) { |
| 1191 | MultiFDRecvParams *p = &multifd_recv_state->params[i]; |
| 1192 | |
| 1193 | trace_multifd_recv_sync_main_signal(p->id); |
| 1194 | qemu_sem_post(&p->sem); |
| 1195 | } |
| 1196 | } |
| 1197 | |
| 1198 | /* |
| 1199 | * Initiate the synchronization by waiting for all channels. |
| 1200 | * |
| 1201 | * For socket-based migration this means each channel has received |
| 1202 | * the SYNC packet on the stream. |
| 1203 | * |
| 1204 | * For file-based migration this means each channel is done with |
| 1205 | * the work (pending_job=false). |
| 1206 | */ |
| 1207 | for (i = 0; i < thread_count; i++) { |
| 1208 | trace_multifd_recv_sync_main_wait(i); |
| 1209 | qemu_sem_wait(&multifd_recv_state->sem_sync); |
| 1210 | } |
| 1211 | |
| 1212 | if (file_based) { |
| 1213 | /* |
| 1214 | * For file-based loading is done in one iteration. We're |
| 1215 | * done. |
| 1216 | */ |
| 1217 | return; |
| 1218 | } |
| 1219 | |
| 1220 | /* |
| 1221 | * Sync done. Release the channels for the next iteration. |
| 1222 | */ |
| 1223 | for (i = 0; i < thread_count; i++) { |
| 1224 | MultiFDRecvParams *p = &multifd_recv_state->params[i]; |
| 1225 | |
| 1226 | WITH_QEMU_LOCK_GUARD(&p->mutex) { |
| 1227 | if (multifd_recv_state->packet_num < p->packet_num) { |
| 1228 | multifd_recv_state->packet_num = p->packet_num; |
| 1229 | } |
| 1230 | } |
| 1231 | trace_multifd_recv_sync_main_signal(p->id); |
| 1232 | qemu_sem_post(&p->sem_sync); |
| 1233 | } |
| 1234 | trace_multifd_recv_sync_main(multifd_recv_state->packet_num); |
| 1235 | } |
| 1236 | |
| 1237 | static int multifd_device_state_recv(MultiFDRecvParams *p, Error **errp) |
| 1238 | { |
| 1239 | g_autofree char *dev_state_buf = NULL; |
| 1240 | int ret; |
| 1241 | |
| 1242 | dev_state_buf = g_malloc(p->next_packet_size); |
| 1243 | |
| 1244 | ret = qio_channel_read_all(p->c, dev_state_buf, p->next_packet_size, errp); |
| 1245 | if (ret != 0) { |
| 1246 | return ret; |
| 1247 | } |
| 1248 | |
| 1249 | if (p->packet_dev_state->idstr[sizeof(p->packet_dev_state->idstr) - 1] |
| 1250 | != 0) { |
| 1251 | error_setg(errp, "unterminated multifd device state idstr"); |
| 1252 | return -1; |
| 1253 | } |
| 1254 | |
| 1255 | if (!qemu_loadvm_load_state_buffer(p->packet_dev_state->idstr, |
| 1256 | p->packet_dev_state->instance_id, |
| 1257 | dev_state_buf, p->next_packet_size, |
| 1258 | errp)) { |
| 1259 | ret = -1; |
| 1260 | } |
| 1261 | |
| 1262 | return ret; |
| 1263 | } |
| 1264 | |
| 1265 | static int multifd_ram_state_recv(MultiFDRecvParams *p, Error **errp) |
| 1266 | { |
| 1267 | int ret; |
| 1268 | |
| 1269 | ret = multifd_recv_state->ops->recv(p, errp); |
| 1270 | if (ret != 0) { |
| 1271 | return ret; |
| 1272 | } |
| 1273 | |
| 1274 | if (migrate_colo()) { |
| 1275 | multifd_colo_process_recv(p); |
| 1276 | } |
| 1277 | |
| 1278 | return ret; |
| 1279 | } |
| 1280 | |
| 1281 | static void *multifd_recv_thread(void *opaque) |
| 1282 | { |
| 1283 | MigrationState *s = migrate_get_current(); |
| 1284 | MultiFDRecvParams *p = opaque; |
| 1285 | Error *local_err = NULL; |
| 1286 | bool use_packets = multifd_use_packets(); |
| 1287 | int ret; |
| 1288 | |
| 1289 | trace_multifd_recv_thread_start(p->id); |
| 1290 | rcu_register_thread(); |
| 1291 | |
| 1292 | if (!s->multifd_clean_tls_termination) { |
| 1293 | p->read_flags = QIO_CHANNEL_READ_FLAG_RELAXED_EOF; |
| 1294 | } |
| 1295 | |
| 1296 | while (true) { |
| 1297 | MultiFDPacketHdr_t hdr; |
| 1298 | uint32_t flags = 0; |
| 1299 | bool is_device_state = false; |
| 1300 | bool has_data = false; |
| 1301 | uint8_t *pkt_buf; |
| 1302 | size_t pkt_len; |
| 1303 | |
| 1304 | p->normal_num = 0; |
| 1305 | |
| 1306 | if (use_packets) { |
| 1307 | struct iovec iov = { |
| 1308 | .iov_base = (void *)&hdr, |
| 1309 | .iov_len = sizeof(hdr) |
| 1310 | }; |
| 1311 | |
| 1312 | if (multifd_recv_should_exit()) { |
| 1313 | break; |
| 1314 | } |
| 1315 | |
| 1316 | ret = qio_channel_readv_full_all_eof(p->c, &iov, 1, NULL, NULL, |
| 1317 | p->read_flags, &local_err); |
| 1318 | if (!ret) { |
| 1319 | /* EOF */ |
| 1320 | assert(!local_err); |
| 1321 | break; |
| 1322 | } |
| 1323 | |
| 1324 | if (ret == -1) { |
| 1325 | break; |
| 1326 | } |
| 1327 | |
| 1328 | ret = multifd_recv_unfill_packet_header(p, &hdr, &local_err); |
| 1329 | if (ret) { |
| 1330 | break; |
| 1331 | } |
| 1332 | |
| 1333 | is_device_state = p->flags & MULTIFD_FLAG_DEVICE_STATE; |
| 1334 | if (is_device_state) { |
| 1335 | pkt_buf = (uint8_t *)p->packet_dev_state + sizeof(hdr); |
| 1336 | pkt_len = sizeof(*p->packet_dev_state) - sizeof(hdr); |
| 1337 | } else { |
| 1338 | pkt_buf = (uint8_t *)p->packet + sizeof(hdr); |
| 1339 | pkt_len = p->packet_len - sizeof(hdr); |
| 1340 | } |
| 1341 | |
| 1342 | ret = qio_channel_read_all_eof(p->c, (char *)pkt_buf, pkt_len, |
| 1343 | &local_err); |
| 1344 | if (!ret) { |
| 1345 | /* EOF */ |
| 1346 | error_setg(&local_err, "multifd: unexpected EOF after packet header"); |
| 1347 | break; |
| 1348 | } |
| 1349 | |
| 1350 | if (ret == -1) { |
| 1351 | break; |
| 1352 | } |
| 1353 | |
| 1354 | qemu_mutex_lock(&p->mutex); |
| 1355 | ret = multifd_recv_unfill_packet(p, &local_err); |
| 1356 | if (ret) { |
| 1357 | qemu_mutex_unlock(&p->mutex); |
| 1358 | break; |
| 1359 | } |
| 1360 | |
| 1361 | flags = p->flags; |
| 1362 | /* recv methods don't know how to handle the SYNC flag */ |
| 1363 | p->flags &= ~MULTIFD_FLAG_SYNC; |
| 1364 | |
| 1365 | if (is_device_state) { |
| 1366 | has_data = p->next_packet_size > 0; |
| 1367 | } else { |
| 1368 | /* |
| 1369 | * Even if it's a SYNC packet, this needs to be set |
| 1370 | * because older QEMUs (<9.0) still send data along with |
| 1371 | * the SYNC packet. |
| 1372 | */ |
| 1373 | has_data = p->normal_num || p->zero_num; |
| 1374 | } |
| 1375 | |
| 1376 | qemu_mutex_unlock(&p->mutex); |
| 1377 | } else { |
| 1378 | /* |
| 1379 | * No packets, so we need to wait for the vmstate code to |
| 1380 | * give us work. |
| 1381 | */ |
| 1382 | qemu_sem_wait(&p->sem); |
| 1383 | |
| 1384 | if (multifd_recv_should_exit()) { |
| 1385 | break; |
| 1386 | } |
| 1387 | |
| 1388 | /* pairs with qatomic_store_release() at multifd_recv() */ |
| 1389 | if (!qatomic_load_acquire(&p->pending_job)) { |
| 1390 | /* |
| 1391 | * Migration thread did not send work, this is |
| 1392 | * equivalent to pending_sync on the sending |
| 1393 | * side. Post sem_sync to notify we reached this |
| 1394 | * point. |
| 1395 | */ |
| 1396 | qemu_sem_post(&multifd_recv_state->sem_sync); |
| 1397 | continue; |
| 1398 | } |
| 1399 | |
| 1400 | has_data = !!p->data->size; |
| 1401 | } |
| 1402 | |
| 1403 | if (has_data) { |
| 1404 | /* |
| 1405 | * multifd thread should not be active and receive data |
| 1406 | * when migration is in the Postcopy phase. Two threads |
| 1407 | * writing the same memory area could easily corrupt |
| 1408 | * the guest state. |
| 1409 | */ |
| 1410 | assert(!migration_in_postcopy()); |
| 1411 | if (is_device_state) { |
| 1412 | assert(use_packets); |
| 1413 | ret = multifd_device_state_recv(p, &local_err); |
| 1414 | } else { |
| 1415 | ret = multifd_ram_state_recv(p, &local_err); |
| 1416 | } |
| 1417 | if (ret != 0) { |
| 1418 | break; |
| 1419 | } |
| 1420 | } else if (is_device_state) { |
| 1421 | error_setg(&local_err, |
| 1422 | "multifd: received empty device state packet"); |
| 1423 | break; |
| 1424 | } |
| 1425 | |
| 1426 | if (use_packets) { |
| 1427 | if (flags & MULTIFD_FLAG_SYNC) { |
| 1428 | if (is_device_state) { |
| 1429 | error_setg(&local_err, |
| 1430 | "multifd: received SYNC device state packet"); |
| 1431 | break; |
| 1432 | } |
| 1433 | |
| 1434 | qemu_sem_post(&multifd_recv_state->sem_sync); |
| 1435 | qemu_sem_wait(&p->sem_sync); |
| 1436 | } |
| 1437 | } else { |
| 1438 | p->data->size = 0; |
| 1439 | /* |
| 1440 | * Order data->size update before clearing |
| 1441 | * pending_job. Pairs with smp_mb_acquire() at |
| 1442 | * multifd_recv(). |
| 1443 | */ |
| 1444 | qatomic_store_release(&p->pending_job, false); |
| 1445 | } |
| 1446 | } |
| 1447 | |
| 1448 | if (local_err) { |
| 1449 | multifd_recv_terminate_threads(local_err); |
| 1450 | } |
| 1451 | |
| 1452 | rcu_unregister_thread(); |
| 1453 | trace_multifd_recv_thread_end(p->id, p->packets_recved); |
| 1454 | |
| 1455 | return NULL; |
| 1456 | } |
| 1457 | |
| 1458 | int multifd_recv_setup(Error **errp) |
| 1459 | { |
| 1460 | int thread_count; |
| 1461 | uint32_t page_count = multifd_ram_page_count(); |
| 1462 | bool use_packets = multifd_use_packets(); |
| 1463 | uint8_t i; |
| 1464 | |
| 1465 | /* |
| 1466 | * Return successfully if multiFD recv state is already initialised |
| 1467 | * or multiFD is not enabled. |
| 1468 | */ |
| 1469 | if (multifd_recv_state || !migrate_multifd()) { |
| 1470 | return 0; |
| 1471 | } |
| 1472 | |
| 1473 | thread_count = migrate_multifd_channels(); |
| 1474 | multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state)); |
| 1475 | multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count); |
| 1476 | |
| 1477 | multifd_recv_state->data = g_new0(MultiFDRecvData, 1); |
| 1478 | multifd_recv_state->data->size = 0; |
| 1479 | |
| 1480 | qatomic_set(&multifd_recv_state->count, 0); |
| 1481 | qatomic_set(&multifd_recv_state->exiting, 0); |
| 1482 | qemu_sem_init(&multifd_recv_state->sem_sync, 0); |
| 1483 | multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()]; |
| 1484 | |
| 1485 | for (i = 0; i < thread_count; i++) { |
| 1486 | MultiFDRecvParams *p = &multifd_recv_state->params[i]; |
| 1487 | |
| 1488 | qemu_mutex_init(&p->mutex); |
| 1489 | qemu_sem_init(&p->sem_sync, 0); |
| 1490 | qemu_sem_init(&p->sem, 0); |
| 1491 | p->pending_job = false; |
| 1492 | p->id = i; |
| 1493 | |
| 1494 | p->data = g_new0(MultiFDRecvData, 1); |
| 1495 | p->data->size = 0; |
| 1496 | |
| 1497 | if (use_packets) { |
| 1498 | p->packet_len = sizeof(MultiFDPacket_t) |
| 1499 | + sizeof(uint64_t) * page_count; |
| 1500 | p->packet = g_malloc0(p->packet_len); |
| 1501 | p->packet_dev_state = g_malloc0(sizeof(*p->packet_dev_state)); |
| 1502 | } |
| 1503 | p->name = g_strdup_printf(MIGRATION_THREAD_DST_MULTIFD, i); |
| 1504 | p->normal = g_new0(ram_addr_t, page_count); |
| 1505 | p->zero = g_new0(ram_addr_t, page_count); |
| 1506 | } |
| 1507 | |
| 1508 | for (i = 0; i < thread_count; i++) { |
| 1509 | MultiFDRecvParams *p = &multifd_recv_state->params[i]; |
| 1510 | int ret; |
| 1511 | |
| 1512 | ret = multifd_recv_state->ops->recv_setup(p, errp); |
| 1513 | if (ret) { |
| 1514 | return ret; |
| 1515 | } |
| 1516 | } |
| 1517 | return 0; |
| 1518 | } |
| 1519 | |
| 1520 | bool multifd_recv_all_channels_created(void) |
| 1521 | { |
| 1522 | int thread_count = migrate_multifd_channels(); |
| 1523 | |
| 1524 | if (!migrate_multifd()) { |
| 1525 | return true; |
| 1526 | } |
| 1527 | |
| 1528 | if (!multifd_recv_state) { |
| 1529 | /* Called before any connections created */ |
| 1530 | return false; |
| 1531 | } |
| 1532 | |
| 1533 | return thread_count == qatomic_read(&multifd_recv_state->count); |
| 1534 | } |
| 1535 | |
| 1536 | /* |
| 1537 | * Try to receive all multifd channels to get ready for the migration. |
| 1538 | * Sets @errp when failing to receive the current channel. |
| 1539 | */ |
| 1540 | bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp) |
| 1541 | { |
| 1542 | MultiFDRecvParams *p; |
| 1543 | Error *local_err = NULL; |
| 1544 | bool use_packets = multifd_use_packets(); |
| 1545 | int id; |
| 1546 | |
| 1547 | if (use_packets) { |
| 1548 | id = multifd_recv_initial_packet(ioc, &local_err); |
| 1549 | if (id < 0) { |
| 1550 | multifd_recv_terminate_threads(error_copy(local_err)); |
| 1551 | error_propagate_prepend(errp, local_err, |
| 1552 | "failed to receive packet" |
| 1553 | " via multifd channel %d: ", |
| 1554 | qatomic_read(&multifd_recv_state->count)); |
| 1555 | return false; |
| 1556 | } |
| 1557 | trace_multifd_recv_new_channel(id); |
| 1558 | } else { |
| 1559 | id = qatomic_read(&multifd_recv_state->count); |
| 1560 | } |
| 1561 | |
| 1562 | p = &multifd_recv_state->params[id]; |
| 1563 | if (p->c != NULL) { |
| 1564 | error_setg(&local_err, "multifd: received id '%d' already setup'", |
| 1565 | id); |
| 1566 | multifd_recv_terminate_threads(error_copy(local_err)); |
| 1567 | error_propagate(errp, local_err); |
| 1568 | return false; |
| 1569 | } |
| 1570 | p->c = ioc; |
| 1571 | object_ref(OBJECT(ioc)); |
| 1572 | |
| 1573 | p->thread_created = true; |
| 1574 | qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p, |
| 1575 | QEMU_THREAD_JOINABLE); |
| 1576 | qatomic_inc(&multifd_recv_state->count); |
| 1577 | |
| 1578 | return true; |
| 1579 | } |