| 579 | } |
| 580 | |
| 581 | fn worker_send_memory( |
| 582 | socket: &mut SocketStream, |
| 583 | guest_memory: &GuestMemoryAtomic<GuestMemoryMmap>, |
| 584 | message_rx: &Mutex<Receiver<SendMemoryThreadMessage>>, |
| 585 | worker_error: &AtomicBool, |
| 586 | notify_tx: &Sender<SendMemoryThreadNotify>, |
| 587 | ) -> Result<(), MigratableError> { |
| 588 | info!("Spawned thread to send VM memory."); |
| 589 | loop { |
| 590 | // Every memory sending thread receives messages from the main thread through this |
| 591 | // channel. The lock is necessary to synchronize the multiple consumers. If the |
| 592 | // workers are very quick, lock contention could become a performance issue. |
| 593 | let message = message_rx |
| 594 | .lock() |
| 595 | .map_err(|_| MigratableError::MigrateSend(anyhow!("message_rx mutex is poisoned"))) |
| 596 | .inspect_err(|_| { |
| 597 | worker_error.store(true, Ordering::Relaxed); |
| 598 | // We ignore errors during error handling. |
| 599 | notify_tx.send(SendMemoryThreadNotify::Error).ok(); |
| 600 | })? |
| 601 | .recv() |
| 602 | .context("Error receiving message from main thread") |
| 603 | .map_err(MigratableError::MigrateSend) |
| 604 | .inspect_err(|_| { |
| 605 | worker_error.store(true, Ordering::Relaxed); |
| 606 | notify_tx.send(SendMemoryThreadNotify::Error).ok(); |
| 607 | })?; |
| 608 | match message { |
| 609 | SendMemoryThreadMessage::Memory(table) => { |
| 610 | send_memory_ranges(guest_memory, &table, socket) |
| 611 | .inspect_err(|_| { |
| 612 | worker_error.store(true, Ordering::Relaxed); |
| 613 | notify_tx.send(SendMemoryThreadNotify::Error).ok(); |
| 614 | }) |
| 615 | .context("Error sending memory to receiver side") |
| 616 | .map_err(MigratableError::MigrateSend)?; |
| 617 | } |
| 618 | SendMemoryThreadMessage::Gate(gate) => { |
| 619 | notify_tx |
| 620 | .send(SendMemoryThreadNotify::Gate) |
| 621 | .context("Error sending gate notification to main thread") |
| 622 | .map_err(MigratableError::MigrateSend) |
| 623 | .inspect_err(|_| { |
| 624 | // Sending via `notify_tx` just failed, so we don't try to send another |
| 625 | // message via it. |
| 626 | worker_error.store(true, Ordering::Relaxed); |
| 627 | })?; |
| 628 | gate.wait(); |
| 629 | } |
| 630 | SendMemoryThreadMessage::Disconnect => { |
| 631 | return Ok(()); |
| 632 | } |
| 633 | } |
| 634 | } |
| 635 | } |
| 636 | |
| 637 | /// Send memory via all connections that we have. `socket` is the original socket |
| 638 | /// that was used to connect to the destination. Returns Ok(true) if memory was |