1 // Copyright 2021 The ChromiumOS Authors
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4
5 pub mod sys;
6
7 use anyhow::anyhow;
8 use anyhow::Context;
9 use base::error;
10 use base::AsRawDescriptors;
11 use cros_async::EventAsync;
12 use cros_async::Executor;
13 use cros_async::IntoAsync;
14 use cros_async::TaskHandle;
15 use futures::channel::oneshot;
16 use futures::pin_mut;
17 use futures::select_biased;
18 use futures::FutureExt;
19 use net_util::TapT;
20 use once_cell::sync::OnceCell;
21 use serde::Deserialize;
22 use serde::Serialize;
23 pub use sys::start_device as run_net_device;
24 pub use sys::Options;
25 use vm_memory::GuestMemory;
26 use vmm_vhost::message::VhostUserProtocolFeatures;
27 use zerocopy::AsBytes;
28
29 use crate::virtio;
30 use crate::virtio::net::build_config;
31 use crate::virtio::net::process_ctrl;
32 use crate::virtio::net::process_tx;
33 use crate::virtio::net::virtio_features_to_tap_offload;
34 use crate::virtio::vhost::user::device::handler::DeviceRequestHandler;
35 use crate::virtio::vhost::user::device::handler::Error as DeviceError;
36 use crate::virtio::vhost::user::device::handler::VhostUserDevice;
37 use crate::virtio::vhost::user::VhostUserDeviceBuilder;
38 use crate::virtio::Queue;
39
40 thread_local! {
41 pub(crate) static NET_EXECUTOR: OnceCell<Executor> = const { OnceCell::new() };
42 }
43
44 // TODO(b/188947559): Come up with better way to include these constants. Compiler errors happen
45 // if they are kept in the trait.
46 const MAX_QUEUE_NUM: usize = 3; /* rx, tx, ctrl */
47
run_tx_queue<T: TapT>( mut queue: Queue, mut tap: T, kick_evt: EventAsync, mut stop_rx: oneshot::Receiver<()>, ) -> Queue48 async fn run_tx_queue<T: TapT>(
49 mut queue: Queue,
50 mut tap: T,
51 kick_evt: EventAsync,
52 mut stop_rx: oneshot::Receiver<()>,
53 ) -> Queue {
54 let kick_evt_future = kick_evt.next_val().fuse();
55 pin_mut!(kick_evt_future);
56 loop {
57 select_biased! {
58 kick = kick_evt_future => {
59 kick_evt_future.set(kick_evt.next_val().fuse());
60 if let Err(e) = kick {
61 error!("Failed to read kick event for tx queue: {}", e);
62 break;
63 }
64 }
65 _ = stop_rx => {
66 break;
67 }
68 }
69
70 process_tx(&mut queue, &mut tap);
71 }
72 queue
73 }
74
run_ctrl_queue<T: TapT>( mut queue: Queue, mut tap: T, kick_evt: EventAsync, acked_features: u64, vq_pairs: u16, mut stop_rx: oneshot::Receiver<()>, ) -> Queue75 async fn run_ctrl_queue<T: TapT>(
76 mut queue: Queue,
77 mut tap: T,
78 kick_evt: EventAsync,
79 acked_features: u64,
80 vq_pairs: u16,
81 mut stop_rx: oneshot::Receiver<()>,
82 ) -> Queue {
83 let kick_evt_future = kick_evt.next_val().fuse();
84 pin_mut!(kick_evt_future);
85 loop {
86 select_biased! {
87 kick = kick_evt_future => {
88 kick_evt_future.set(kick_evt.next_val().fuse());
89 if let Err(e) = kick {
90 error!("Failed to read kick event for tx queue: {}", e);
91 break;
92 }
93 }
94 _ = stop_rx => {
95 break;
96 }
97 }
98
99 if let Err(e) = process_ctrl(&mut queue, &mut tap, acked_features, vq_pairs) {
100 error!("Failed to process ctrl queue: {}", e);
101 break;
102 }
103 }
104 queue
105 }
106
107 pub struct NetBackend<T: TapT + IntoAsync> {
108 tap: T,
109 avail_features: u64,
110 acked_features: u64,
111 mtu: u16,
112 #[cfg(all(windows, feature = "slirp"))]
113 slirp_kill_event: base::Event,
114 workers: [Option<(TaskHandle<Queue>, oneshot::Sender<()>)>; MAX_QUEUE_NUM],
115 }
116
117 #[derive(Serialize, Deserialize)]
118 pub struct NetBackendSnapshot {
119 acked_feature: u64,
120 }
121
122 impl<T: 'static> NetBackend<T>
123 where
124 T: TapT + IntoAsync,
125 {
max_vq_pairs() -> usize126 fn max_vq_pairs() -> usize {
127 MAX_QUEUE_NUM / 2
128 }
129 }
130
131 impl<T: 'static> AsRawDescriptors for NetBackend<T>
132 where
133 T: TapT + IntoAsync + AsRawDescriptors,
134 {
as_raw_descriptors(&self) -> Vec<base::RawDescriptor>135 fn as_raw_descriptors(&self) -> Vec<base::RawDescriptor> {
136 self.tap.as_raw_descriptors()
137 }
138 }
139
140 impl<T: 'static> VhostUserDevice for NetBackend<T>
141 where
142 T: TapT + IntoAsync,
143 {
max_queue_num(&self) -> usize144 fn max_queue_num(&self) -> usize {
145 MAX_QUEUE_NUM
146 }
147
features(&self) -> u64148 fn features(&self) -> u64 {
149 self.avail_features
150 }
151
ack_features(&mut self, value: u64) -> anyhow::Result<()>152 fn ack_features(&mut self, value: u64) -> anyhow::Result<()> {
153 self.acked_features |= value;
154
155 self.tap
156 .set_offload(virtio_features_to_tap_offload(self.acked_features))
157 .context("failed to set tap offload to match features")?;
158
159 Ok(())
160 }
161
protocol_features(&self) -> VhostUserProtocolFeatures162 fn protocol_features(&self) -> VhostUserProtocolFeatures {
163 VhostUserProtocolFeatures::CONFIG | VhostUserProtocolFeatures::DEVICE_STATE
164 }
165
read_config(&self, offset: u64, data: &mut [u8])166 fn read_config(&self, offset: u64, data: &mut [u8]) {
167 let config_space = build_config(Self::max_vq_pairs() as u16, self.mtu, None);
168 virtio::copy_config(data, 0, config_space.as_bytes(), offset);
169 }
170
reset(&mut self)171 fn reset(&mut self) {}
172
start_queue( &mut self, idx: usize, queue: virtio::Queue, mem: GuestMemory, ) -> anyhow::Result<()>173 fn start_queue(
174 &mut self,
175 idx: usize,
176 queue: virtio::Queue,
177 mem: GuestMemory,
178 ) -> anyhow::Result<()> {
179 sys::start_queue(self, idx, queue, mem)
180 }
181
stop_queue(&mut self, idx: usize) -> anyhow::Result<virtio::Queue>182 fn stop_queue(&mut self, idx: usize) -> anyhow::Result<virtio::Queue> {
183 if let Some((task, stop_tx)) = self.workers.get_mut(idx).and_then(Option::take) {
184 if stop_tx.send(()).is_err() {
185 return Err(anyhow!("Failed to request stop for net queue future"));
186 }
187
188 // Wait for queue_task to be aborted.
189 let queue = NET_EXECUTOR
190 .with(|ex| {
191 let ex = ex.get().expect("Executor not initialized");
192 ex.run_until(task)
193 })
194 .context("Failed to resolve queue worker future")?;
195
196 Ok(queue)
197 } else {
198 Err(anyhow::Error::new(DeviceError::WorkerNotFound))
199 }
200 }
201
enter_suspended_state(&mut self) -> anyhow::Result<()>202 fn enter_suspended_state(&mut self) -> anyhow::Result<()> {
203 // No non-queue workers.
204 Ok(())
205 }
206
snapshot(&mut self) -> anyhow::Result<serde_json::Value>207 fn snapshot(&mut self) -> anyhow::Result<serde_json::Value> {
208 serde_json::to_value(NetBackendSnapshot {
209 acked_feature: self.acked_features,
210 })
211 .context("Failed to serialize NetBackendSnapshot")
212 }
213
restore(&mut self, data: serde_json::Value) -> anyhow::Result<()>214 fn restore(&mut self, data: serde_json::Value) -> anyhow::Result<()> {
215 let net_backend_snapshot: NetBackendSnapshot =
216 serde_json::from_value(data).context("Failed to deserialize NetBackendSnapshot")?;
217 self.acked_features = net_backend_snapshot.acked_feature;
218 Ok(())
219 }
220 }
221
222 impl<T> VhostUserDeviceBuilder for NetBackend<T>
223 where
224 T: TapT + IntoAsync + 'static,
225 {
build(self: Box<Self>, ex: &Executor) -> anyhow::Result<Box<dyn vmm_vhost::Backend>>226 fn build(self: Box<Self>, ex: &Executor) -> anyhow::Result<Box<dyn vmm_vhost::Backend>> {
227 NET_EXECUTOR.with(|thread_ex| {
228 let _ = thread_ex.set(ex.clone());
229 });
230 let handler = DeviceRequestHandler::new(*self);
231
232 Ok(Box::new(handler))
233 }
234 }
235