diff --git a/lib/propolis/src/vsock/poller.rs b/lib/propolis/src/vsock/poller.rs index 0646d9766..df908d228 100644 --- a/lib/propolis/src/vsock/poller.rs +++ b/lib/propolis/src/vsock/poller.rs @@ -11,7 +11,6 @@ use std::io::Read; use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd, RawFd}; use std::sync::mpsc; use std::sync::Arc; -use std::sync::Barrier; use std::sync::Condvar; use std::sync::Mutex; use std::thread::JoinHandle; @@ -94,7 +93,7 @@ impl PollerState { /// Commands that can be sent to a paused `poller_loop`. enum PausedCmd { /// Continue execution - Resume, + Resume { oneshot: mpsc::SyncSender<()> }, /// Cleanup all connection state and queued packets Reset { oneshot: mpsc::SyncSender<()> }, /// Shutdown the event-loop @@ -132,21 +131,23 @@ impl VsockPollerNotify { } } - pub fn pause(&self) -> std::io::Result<()> { + pub fn pause_evloop(&self) -> std::io::Result<()> { self.port_send(VsockEvent::Pause) } - pub fn resume(&self) { - self.pause_tx.send(PausedCmd::Resume).unwrap(); + pub fn resume_evloop(&self) { + let (tx, rx) = mpsc::sync_channel(0); + self.pause_tx.send(PausedCmd::Resume { oneshot: tx }).unwrap(); + rx.recv().unwrap(); } - pub fn reset(&self) { + pub fn reset_evloop(&self) { let (tx, rx) = mpsc::sync_channel(0); self.pause_tx.send(PausedCmd::Reset { oneshot: tx }).unwrap(); rx.recv().unwrap(); } - pub fn halt(&self) { + pub fn halt_evloop(&self) { self.pause_tx.send(PausedCmd::Halt).unwrap(); } @@ -193,7 +194,6 @@ struct ClosingConn { pub struct VsockPoller { log: Logger, - start_barrier: Arc, /// The guest context id guest_cid: GuestCid, /// Port mappings we are proxying packets to and from @@ -223,7 +223,6 @@ impl VsockPoller { /// the guest VM and host sockets. pub fn new( log: Logger, - start_barrier: Arc, cid: GuestCid, queues: VsockVq, port_mappings: IdHashMap, @@ -257,7 +256,6 @@ impl VsockPoller { Ok(Self { log, - start_barrier, guest_cid: cid, port_mappings, port_fd: Arc::new(unsafe { OwnedFd::from_raw_fd(port_fd) }), @@ -908,37 +906,32 @@ impl VsockPoller { } } - // This is the gerneral flow of the single-threaded processing event-loop: + // This is the general flow of the single-threaded processing event-loop: // - // ┌─────────virtio-socket─event-loop───────┐ - // │ │ - // │ ┌─────────────────────┐ │ - // │ │ start (set_running) ◄────┐ │ - // │ └──────────┬──────────┘ │ │ - // ┌───────────┐ │ │ │ │ - // │vq rx event│ │ │ │ │ - // │vq tx event│ │ ┌──────────▼──────────┐ │ │ - // │fd pollset ├───┼─────────► handle_events() │ │ │ - // │pause event│ │ └──────────┬──────────┘ │ │ - // └───────────┘ │ │ │ │ - // │ │ │ │ - // │ ┌──────────▼──────────┐ │ │ - // │ │ paused (set_stopped)│ │ │ - // │ └──────────┬──────────┘ │ │ - // │ │ │ │ - // │ │ │ │ - // │ ┌──────▼───────┐ │ │ - // │ │ pause_rx │ │ │ - // │ │ │ │ │ - // │ │ - resume────┼───────┘ │ - // │ │ - reset─────┼─►cleanup │ - // │ │ - halt───┐ │ state │ - // │ │ │ │ │ - // │ └───────────┼──┘ │ - // │ │ │ - // └─────────────────────────┼──────────────┘ - // ▼ - // Exit + // ┌─────────────┐ + // │ vq rx event │ + // thread spawn │ vq tx event │ + // │ │ fd pollset │ + // │ │ pause event │ + // │ └──────┬──────┘ + // │ │ + // ┌──────────────┼──virtio-socket─event-loop──────┼─────────────┐ + // │ ▼ │ │ + // │ ┌──────────────────────┐ ┌───────────▼───────────┐ │ + // │┌►│ paused (set_stopped) ◄───pause─┤ handle_events() │ │ + // ││ └───────────┬──────────┘ └───────────▲───────────┘ │ + // ││ │ │ │ + // ││ ┌──────▼───────┐ │ │ + // ││ │ pause_rx │ ┌───────────┴───────────┐ │ + // ││ │ run ──────┼────────────► running (set_running) │ │ + // ││ │ ┌─ reset │ └───────────────────────┘ │ + // ││ │ │ halt ─┐ │ │ + // ││ └─┼────────┤───┘ │ + // ││ ▼ │ │ + // │└─ cleanup state │ │ + // └──────────────────┼──────────────────────────────────────────┘ + // ▼ + // Exit // // The event-loop is executing in a dedicated thread and therefore must // be able to handle triggers from propolis as it manages the device @@ -953,46 +946,34 @@ impl VsockPoller { // every tracked fd and later re-associating them, this allows us to yield // execution until one of the desired next state events is delivered. fn poller_loop(&mut self) { - // Wait until propolis tells us to start the virtio-socket device. - // If we are never told to start that is okay as there will be no need - // to ever transition the device from running->paused->halt. - // - // TODO: If propolis-server is started but the VM is never transitioned - // to running this will leave a useless thread around that will be - // cleanned up when propolis-server exits. This also means if one sends - // a new instance spec to the pre-existing propolis-server we will - // accumulate these "dead" threads, howerver we never do this in the - // actual product. - self.start_barrier.wait(); - loop { - // We are running! - self.state.set_running(); + match self.pause_rx.recv() { + Ok(PausedCmd::Resume { oneshot }) => { + oneshot.send(()).unwrap(); - // Handle events until we are told to pause. - self.handle_events(); + // We are running! + self.state.set_running(); - // Transition to stopped and await for next steps. - self.state.set_stopped(); + // Handle events until we are told to pause. + self.handle_events(); - loop { - match self.pause_rx.recv() { - Ok(PausedCmd::Resume) => break, - Ok(PausedCmd::Reset { oneshot }) => { - self.reset(); - oneshot.send(()).unwrap(); - } - Ok(PausedCmd::Halt) => { - self.reset(); - return; - } - Err(_) => { - error!( + // Transition to stopped and await for next steps. + self.state.set_stopped(); + } + Ok(PausedCmd::Reset { oneshot }) => { + self.reset(); + oneshot.send(()).unwrap(); + } + Ok(PausedCmd::Halt) => { + self.reset(); + return; + } + Err(_) => { + error!( &self.log, "all VsockPoller pause_tx senders have been dropped" ); - return; - } + return; } } } @@ -1070,7 +1051,6 @@ mod test { use std::net::TcpListener; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; - use std::sync::Barrier; use std::time::Duration; use iddqd::IdHashMap; @@ -1235,21 +1215,13 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); harness.add_rx_writable(256); let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); let mut hdr = VsockPacketHeader::new(); hdr.set_src_cid(guest_cid) @@ -1276,8 +1248,8 @@ mod test { assert_eq!(resp_hdr.dst_port(), guest_port); assert_eq!(resp_hdr.socket_type(), Some(VsockSocketType::Stream)); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1288,21 +1260,14 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - IdHashMap::new(), - ) - .unwrap(); + let poller = + VsockPoller::new(log, guest_cid, vq, IdHashMap::new()).unwrap(); harness.add_rx_writable(256); let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); let mut hdr = VsockPacketHeader::new(); hdr.set_src_cid(guest_cid) @@ -1328,8 +1293,8 @@ mod test { assert_eq!(resp_hdr.src_port(), 8080); assert_eq!(resp_hdr.dst_port(), 5555); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1344,15 +1309,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); for _ in 0..4 { harness.add_rx_writable(4096); @@ -1360,7 +1317,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Send REQUEST let mut req_hdr = VsockPacketHeader::new(); @@ -1410,8 +1367,8 @@ mod test { accepted.read_exact(&mut buf).unwrap(); assert_eq!(&buf, payload); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1426,15 +1383,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Provide plenty of RX descriptors for RESPONSE + credit updates for _ in 0..16 { @@ -1443,7 +1392,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish connection let mut req_hdr = VsockPacketHeader::new(); @@ -1537,8 +1486,8 @@ mod test { } assert!(found_credit_update, "expected a CREDIT_UPDATE on RX queue"); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1553,15 +1502,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); for _ in 0..4 { harness.add_rx_writable(4096); @@ -1569,7 +1510,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Send REQUEST let mut req_hdr = VsockPacketHeader::new(); @@ -1623,8 +1564,8 @@ mod test { Ok(n) => panic!("expected EOF or error, got {n} bytes"), } - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1639,15 +1580,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Pre-populate RX queue with writable descriptors for RESPONSE + data for _ in 0..8 { @@ -1656,7 +1589,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Write REQUEST packet into TX queue let mut req_hdr = VsockPacketHeader::new(); @@ -1724,8 +1657,8 @@ mod test { assert_eq!(resp_hdr.dst_port(), guest_port); assert_eq!(&host_buf, host_payload, "host->guest data mismatch"); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1740,22 +1673,14 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Provide only one RX descriptor, just enough for the RESPONSE. harness.add_rx_writable(4096); let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Send REQUEST let mut req_hdr = VsockPacketHeader::new(); @@ -1804,8 +1729,8 @@ mod test { assert_eq!(rw_hdr.dst_port(), guest_port); assert_eq!(&payload, host_data); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1825,15 +1750,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Provide initial RX descriptors for RESPONSE + credit updates for _ in 0..8 { @@ -1842,7 +1759,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish connection // Use a large buf_alloc so host->guest credit doesn't run out @@ -2023,8 +1940,8 @@ mod test { assert_eq!(host_to_guest.len(), total_bytes); assert!(host_to_guest == host_data, "host->guest data mismatch"); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2042,15 +1959,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Provide RX descriptors for RESPONSE + SHUTDOWN for _ in 0..4 { @@ -2059,7 +1968,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish connection let mut req_hdr = VsockPacketHeader::new(); @@ -2116,8 +2025,8 @@ mod test { assert_eq!(hdr.src_port(), vsock_port); assert_eq!(hdr.dst_port(), guest_port); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2132,15 +2041,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Pre-populate RX queue with writable descriptors for RESPONSE + data for _ in 0..8 { @@ -2149,7 +2050,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Write REQUEST packet into TX queue let mut req_hdr = VsockPacketHeader::new(); @@ -2230,8 +2131,8 @@ mod test { assert_eq!(rst_hdr.src_port(), vsock_port); assert_eq!(rst_hdr.dst_port(), guest_port); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2249,15 +2150,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); for _ in 0..4 { harness.add_rx_writable(4096); @@ -2265,7 +2158,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish a connection let mut req_hdr = VsockPacketHeader::new(); @@ -2290,9 +2183,9 @@ mod test { wait_for_condition(|| harness.rx_used_idx() >= 1, 5000); // Pause then reset. - notify.pause().unwrap(); + notify.pause_evloop().unwrap(); notify.wait_stopped(); - notify.reset(); + notify.reset_evloop(); // NOTE: We don't have a way to actually validate that the reset call // above has not left behind a `GuestAddr` as apart of the `VsockVq`. @@ -2306,8 +2199,8 @@ mod test { Ok(n) => panic!("expected EOF or error, got {n} bytes"), } - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2324,15 +2217,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); for _ in 0..8 { harness.add_rx_writable(4096); @@ -2340,7 +2225,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish a connection let mut req_hdr = VsockPacketHeader::new(); @@ -2365,9 +2250,9 @@ mod test { wait_for_condition(|| harness.rx_used_idx() >= 1, 5000); // Pause and resume — connection state should survive. - notify.pause().unwrap(); + notify.pause_evloop().unwrap(); notify.wait_stopped(); - notify.resume(); + notify.resume_evloop(); // Write data from the host side through the still-open TCP // socket. The poller should deliver it to the guest. @@ -2381,8 +2266,8 @@ mod test { assert_eq!(rw_hdr.op(), Some(VsockPacketOp::ReadWrite)); assert_eq!(&data, host_payload); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2393,22 +2278,14 @@ mod test { let harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - IdHashMap::new(), - ) - .unwrap(); + let poller = + VsockPoller::new(log, guest_cid, vq, IdHashMap::new()).unwrap(); let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); - - notify.pause().unwrap(); - notify.halt(); + notify.resume_evloop(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } } diff --git a/lib/propolis/src/vsock/poller_stub.rs b/lib/propolis/src/vsock/poller_stub.rs index 712a4dc40..72c81ddd6 100644 --- a/lib/propolis/src/vsock/poller_stub.rs +++ b/lib/propolis/src/vsock/poller_stub.rs @@ -2,8 +2,6 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/. -use std::sync::Arc; -use std::sync::Barrier; use std::thread::JoinHandle; use iddqd::IdHashMap; @@ -29,15 +27,15 @@ impl VsockPollerNotify { )); } - pub fn pause(&self) -> std::io::Result<()> { + pub fn pause_evloop(&self) -> std::io::Result<()> { Ok(()) } - pub fn resume(&self) {} + pub fn resume_evloop(&self) {} - pub fn reset(&self) {} + pub fn reset_evloop(&self) {} - pub fn halt(&self) {} + pub fn halt_evloop(&self) {} pub fn wait_stopped(&self) {} } @@ -47,7 +45,6 @@ pub struct VsockPoller; impl VsockPoller { pub fn new( _log: Logger, - _start_barrier: Arc, _cid: GuestCid, _queues: VsockVq, _port_mappings: IdHashMap, diff --git a/lib/propolis/src/vsock/proxy.rs b/lib/propolis/src/vsock/proxy.rs index 01085b157..6ebb71f5b 100644 --- a/lib/propolis/src/vsock/proxy.rs +++ b/lib/propolis/src/vsock/proxy.rs @@ -8,8 +8,6 @@ use std::num::NonZeroUsize; use std::num::Wrapping; use std::os::fd::AsRawFd; use std::os::fd::RawFd; -use std::sync::Arc; -use std::sync::Barrier; use std::thread::JoinHandle; use std::time::Duration; @@ -324,7 +322,6 @@ impl IdHashItem for VsockPortMapping { /// virtio-socket backend that proxies between a guest and a host UDS. pub struct VsockProxy { log: Logger, - start_barrier: Arc, poller: VsockPollerNotify, _evloop_handle: JoinHandle<()>, } @@ -336,23 +333,12 @@ impl VsockProxy { queues: VsockVq, port_mappings: IdHashMap, ) -> Self { - // We use a `Barrier` to gate when the spawned `VsockPoller` thread is - // allowed to transition to the running state and is allowed to start - // using guest memory. Forward progress will be made when the - // virtio-socket device calls `Lifecycle::start`. - let start_barrier = Arc::new(Barrier::new(2)); - let evloop = VsockPoller::new( - log.clone(), - start_barrier.clone(), - cid, - queues, - port_mappings, - ) - .unwrap(); + let evloop = + VsockPoller::new(log.clone(), cid, queues, port_mappings).unwrap(); let poller = evloop.notify_handle(); let jh = evloop.run(); - Self { log, start_barrier, poller, _evloop_handle: jh } + Self { log, poller, _evloop_handle: jh } } /// Notification from the vsock device that one of the queues has had an @@ -365,11 +351,11 @@ impl VsockProxy { // TODO this *should* only be called once given the guarantees propolis // gives us. We might want to start using the `Indicator` type in the // future. - self.start_barrier.wait(); + self.poller.resume_evloop(); } pub fn pause(&self) -> std::io::Result<()> { - self.poller.pause() + self.poller.pause_evloop() } pub fn wait_stopped(&self) { @@ -377,15 +363,15 @@ impl VsockProxy { } pub fn resume(&self) { - self.poller.resume(); + self.poller.resume_evloop(); } pub fn reset(&self) { - self.poller.reset(); + self.poller.reset_evloop(); } pub fn halt(&self) { - self.poller.halt(); + self.poller.halt_evloop(); self.poller.wait_stopped(); } }