| 665 | } |
| 666 | |
| 667 | fn send_chunk(&mut self, chunk: MemoryRangeTable) -> Result<(), MigratableError> { |
| 668 | let mut chunk = SendMemoryThreadMessage::Memory(chunk); |
| 669 | // [`Self::message_tx`] has a limited size, so we may have to retry sending the chunk |
| 670 | loop { |
| 671 | if self.worker_error.load(Ordering::Relaxed) { |
| 672 | return self.cleanup(); |
| 673 | } |
| 674 | |
| 675 | // Use try_send() so we can keep checking worker_error while the |
| 676 | // workers catch up. A blocking send() could wait forever if a |
| 677 | // worker failed and stopped making progress. |
| 678 | match self.message_tx.try_send(chunk) { |
| 679 | Ok(()) => { |
| 680 | return Ok(()); |
| 681 | } |
| 682 | Err(TrySendError::Full(unsent_chunk)) => { |
| 683 | // The channel is full. We wait for a short time and retry. |
| 684 | thread::sleep(Duration::from_millis(10)); |
| 685 | chunk = unsent_chunk; |
| 686 | } |
| 687 | Err(TrySendError::Disconnected(_)) => { |
| 688 | // The workers didn't disconnect for no reason, thus we do a cleanup. |
| 689 | return Err(self.cleanup().err().unwrap_or(MigratableError::MigrateSend( |
| 690 | anyhow!("All sending threads disconnected, but none returned an error?"), |
| 691 | ))); |
| 692 | } |
| 693 | } |
| 694 | } |
| 695 | } |
| 696 | |
| 697 | /// Wait until all data that is in-flight has actually been sent and acknowledged. |
| 698 | fn wait_for_pending_data(&mut self) -> Result<(), MigratableError> { |