chipset_device_worker/
worker.rs

1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4//! A worker for running ChipsetDevice implementations in a separate process.
5//!
6//! This worker provides process isolation for any device implementing the
7//! ChipsetDevice trait. It handles serialization and deserialization of
8//! device operations across process boundaries.
9
10#![forbid(unsafe_code)]
11
12mod configure;
13
14use crate::RemoteDynamicResolvers;
15use crate::guestmem::GuestMemoryRemoteBuilder;
16use crate::protocol::*;
17use anyhow::Context;
18use chipset_device::ChipsetDevice;
19use chipset_device::io::IoResult;
20use chipset_device::io::deferred::DeferredToken;
21use chipset_device::pci::ByteEnabledDwordRead;
22use chipset_device::pci::ByteEnabledDwordWrite;
23use chipset_device_resources::ErasedChipsetDevice;
24use chipset_device_resources::ResolveChipsetDeviceHandleParams;
25use mesh::MeshPayload;
26use mesh::error::RemoteError;
27use mesh_worker::Worker;
28use mesh_worker::WorkerId;
29use mesh_worker::WorkerRpc;
30use pal_async::DefaultPool;
31use std::task::Poll;
32use vm_resource::Resource;
33use vm_resource::ResourceResolver;
34use vm_resource::kind::ChipsetDeviceHandleKind;
35use vmcore::device_state::ChangeDeviceState;
36use vmcore::save_restore::ProtobufSaveRestore;
37
38/// Worker ID for ChipsetDevice workers.
39pub(crate) const fn remote_chipset_device_worker_id<T: RemoteDynamicResolvers>()
40-> WorkerId<RemoteChipsetDeviceWorkerParameters<T>> {
41    WorkerId::new(T::WORKER_ID_STR)
42}
43
44/// Parameters for launching a remote chipset device worker.
45#[derive(MeshPayload)]
46pub struct RemoteChipsetDeviceWorkerParameters<T> {
47    pub(crate) device: Resource<ChipsetDeviceHandleKind>,
48    pub(crate) dyn_resolvers: T,
49    pub(crate) inputs: RemoteChipsetDeviceHandleParams,
50
51    pub(crate) req_recv: mesh::Receiver<DeviceRequest>,
52    pub(crate) resp_send: mesh::Sender<DeviceResponse>,
53    pub(crate) cap_send: mesh::OneshotSender<DeviceInit>,
54}
55
56#[derive(MeshPayload)]
57pub(crate) struct RemoteChipsetDeviceHandleParams {
58    pub device_name: String,
59    pub is_restoring: bool,
60    pub vmtime: vmcore::vmtime::VmTimeSourceBuilder,
61    pub guest_memory: GuestMemoryRemoteBuilder,
62    pub encrypted_guest_memory: GuestMemoryRemoteBuilder,
63}
64
65/// The chipset device worker.
66///
67/// This worker wraps any device implementing ChipsetDevice and handles
68/// device operations sent via mesh channels.
69pub struct RemoteChipsetDeviceWorker<T> {
70    device: ErasedChipsetDevice,
71    pool: Option<DefaultPool>,
72    req_recv: mesh::Receiver<DeviceRequest>,
73    resp_send: mesh::Sender<DeviceResponse>,
74    deferred_reads: Vec<DeferredRead>,
75    deferred_writes: Vec<DeferredWrite>,
76
77    _phantom_resolvers: std::marker::PhantomData<T>,
78}
79
80struct DeferredRead {
81    id: usize,
82    token: DeferredToken,
83    size: usize,
84}
85
86struct DeferredWrite {
87    id: usize,
88    token: DeferredToken,
89}
90
91impl<T: RemoteDynamicResolvers> Worker for RemoteChipsetDeviceWorker<T> {
92    type Parameters = RemoteChipsetDeviceWorkerParameters<T>;
93    type State = ();
94    const ID: WorkerId<Self::Parameters> = remote_chipset_device_worker_id();
95
96    fn new(params: Self::Parameters) -> anyhow::Result<Self> {
97        let mut pool = DefaultPool::new();
98
99        let RemoteChipsetDeviceWorkerParameters {
100            device,
101            dyn_resolvers,
102            inputs,
103
104            req_recv,
105            resp_send,
106            cap_send,
107        } = params;
108
109        let mut resolver = ResourceResolver::new();
110
111        let driver = pool.driver();
112        let mut device = pool
113            .run_until(async move {
114                dyn_resolvers
115                    .register_remote_dynamic_resolvers(&mut resolver)
116                    .await?;
117                resolver
118                    .resolve(
119                        device,
120                        ResolveChipsetDeviceHandleParams {
121                            device_name: &inputs.device_name,
122                            guest_memory: &inputs.guest_memory.build("remote_gm"),
123                            encrypted_guest_memory: &inputs
124                                .encrypted_guest_memory
125                                .build("remote_enc_gm"),
126                            vmtime: &inputs
127                                .vmtime
128                                .build(&driver)
129                                .await
130                                .context("failed to build vmtime source")?,
131                            is_restoring: inputs.is_restoring,
132                            task_driver_source: &vmcore::vm_task::VmTaskDriverSource::new(
133                                vmcore::vm_task::thread::ThreadDriverBackend::new(driver),
134                            ),
135                            // TODO: Actually wire these up
136                            configure: &mut configure::RemoteConfigureChipsetDevice {},
137                            register_mmio: &mut configure::RemoteRegisterMmio {},
138                            register_pio: &mut configure::RemoteRegisterPio {},
139                        },
140                    )
141                    .await
142                    .context("failed to resolve device")
143            })?
144            .0;
145
146        if device.supports_acknowledge_pic_interrupt().is_some()
147            || device.supports_handle_eoi().is_some()
148            || device.supports_line_interrupt_target().is_some()
149            || device.supports_tdisp().is_some()
150        {
151            anyhow::bail!("remote device requires unimplemented functionality");
152        }
153
154        cap_send.send(DeviceInit {
155            mmio: device.supports_mmio().map(|m| MmioInit {
156                static_regions: m
157                    .get_static_regions()
158                    .iter()
159                    .map(|(name, range)| ((*name).into(), *range.start(), *range.end()))
160                    .collect(),
161            }),
162            pio: device.supports_pio().map(|p| PioInit {
163                static_regions: p
164                    .get_static_regions()
165                    .iter()
166                    .map(|(name, range)| ((*name).into(), *range.start(), *range.end()))
167                    .collect(),
168            }),
169            pci: device.supports_pci().map(|pci| PciInit {
170                suggested_bdf: pci.suggested_bdf(),
171            }),
172        });
173
174        Ok(Self {
175            device,
176            pool: Some(pool),
177            req_recv,
178            resp_send,
179            deferred_reads: Vec::new(),
180            deferred_writes: Vec::new(),
181            _phantom_resolvers: std::marker::PhantomData,
182        })
183    }
184
185    fn restart(_state: Self::State) -> anyhow::Result<Self> {
186        todo!()
187    }
188
189    fn run(mut self, mut rpc_recv: mesh::Receiver<WorkerRpc<Self::State>>) -> anyhow::Result<()> {
190        self.pool.take().unwrap().run_until(async move {
191            loop {
192                enum WorkerEvent {
193                    Rpc(WorkerRpc<()>),
194                    DeviceRequest(DeviceRequest),
195                }
196
197                let event = std::future::poll_fn(|cx| {
198                    if let Some(poll_device) = self.device.supports_poll_device() {
199                        poll_device.poll_device(cx);
200                    }
201
202                    self.deferred_reads
203                        .extract_if(.., |read| {
204                            let mut data = vec![0; read.size];
205                            match read.token.poll_read(cx, &mut data) {
206                                Poll::Ready(r) => {
207                                    self.resp_send.send(DeviceResponse::Read {
208                                        id: read.id,
209                                        result: r.map(|_| data),
210                                    });
211                                    true
212                                }
213                                Poll::Pending => false,
214                            }
215                        })
216                        .for_each(|_| ());
217
218                    self.deferred_writes
219                        .extract_if(.., |write| match write.token.poll_write(cx) {
220                            Poll::Ready(r) => {
221                                self.resp_send.send(DeviceResponse::Write {
222                                    id: write.id,
223                                    result: r,
224                                });
225                                true
226                            }
227                            Poll::Pending => false,
228                        })
229                        .for_each(|_| ());
230
231                    // If either of these channels fail, we fail the worker too.
232                    if let Poll::Ready(r) = rpc_recv.poll_recv(cx) {
233                        return Poll::Ready(r.map(WorkerEvent::Rpc));
234                    }
235                    if let Poll::Ready(r) = self.req_recv.poll_recv(cx) {
236                        return Poll::Ready(r.map(WorkerEvent::DeviceRequest));
237                    }
238                    Poll::Pending
239                })
240                .await?;
241
242                match event {
243                    WorkerEvent::Rpc(rpc) => match rpc {
244                        WorkerRpc::Inspect(deferred) => {
245                            deferred.inspect(&mut self.device);
246                        }
247                        WorkerRpc::Stop => {
248                            return Ok(());
249                        }
250                        WorkerRpc::Restart(rpc) => {
251                            rpc.complete(Err(RemoteError::new(anyhow::anyhow!("not supported"))));
252                        }
253                    },
254                    WorkerEvent::DeviceRequest(req) => match req {
255                        DeviceRequest::Start => self.device.start(),
256                        DeviceRequest::Stop(rpc) => {
257                            rpc.handle(async |()| self.device.stop().await).await
258                        }
259                        DeviceRequest::Reset(rpc) => {
260                            self.deferred_reads.clear();
261                            self.deferred_writes.clear();
262                            rpc.handle(async |()| self.device.reset().await).await
263                        }
264                        DeviceRequest::MmioRead(ReadRequest { id, address, size }) => {
265                            let mut data = vec![0; size];
266                            let result = self
267                                .device
268                                .supports_mmio()
269                                .unwrap()
270                                .mmio_read(address, &mut data);
271                            self.handle_read_result(id, result, data);
272                        }
273                        DeviceRequest::MmioWrite(WriteRequest { id, address, data }) => {
274                            let result = self
275                                .device
276                                .supports_mmio()
277                                .unwrap()
278                                .mmio_write(address, &data);
279                            self.handle_write_result(id, result);
280                        }
281                        DeviceRequest::PioRead(ReadRequest { id, address, size }) => {
282                            let mut data = vec![0; size];
283                            let result = self
284                                .device
285                                .supports_pio()
286                                .unwrap()
287                                .io_read(address, &mut data);
288                            self.handle_read_result(id, result, data);
289                        }
290                        DeviceRequest::PioWrite(WriteRequest { id, address, data }) => {
291                            let result =
292                                self.device.supports_pio().unwrap().io_write(address, &data);
293                            self.handle_write_result(id, result);
294                        }
295                        DeviceRequest::PciConfigRead(
296                            ReadRequest { id, address, size },
297                            byte_enable,
298                        ) => {
299                            assert_eq!(size, 4);
300                            let mut data_u32 = 0;
301                            let mut value = ByteEnabledDwordRead::new(&mut data_u32, byte_enable);
302                            let result = self
303                                .device
304                                .supports_pci()
305                                .unwrap()
306                                .pci_cfg_read(address, value.reborrow());
307                            self.handle_read_result(id, result, data_u32.to_ne_bytes().to_vec());
308                        }
309                        DeviceRequest::PciConfigWrite(
310                            WriteRequest { id, address, data },
311                            byte_enable,
312                        ) => {
313                            let value = ByteEnabledDwordWrite::new(data, byte_enable);
314                            let result = self
315                                .device
316                                .supports_pci()
317                                .unwrap()
318                                .pci_cfg_write(address, value);
319                            self.handle_write_result(id, result);
320                        }
321                        DeviceRequest::Save(rpc) => {
322                            rpc.handle_failable_sync(|()| self.device.save())
323                        }
324                        DeviceRequest::Restore(rpc) => {
325                            rpc.handle_failable_sync(|state| self.device.restore(state))
326                        }
327                    },
328                }
329            }
330        })
331    }
332}
333
334impl<T> RemoteChipsetDeviceWorker<T> {
335    fn handle_read_result(&mut self, id: usize, result: IoResult, data: Vec<u8>) {
336        match result {
337            IoResult::Ok => self.resp_send.send(DeviceResponse::Read {
338                id,
339                result: Ok(data),
340            }),
341            IoResult::Err(io_error) => self.resp_send.send(DeviceResponse::Read {
342                id,
343                result: Err(io_error),
344            }),
345            IoResult::Defer(token) => self.deferred_reads.push(DeferredRead {
346                id,
347                token,
348                size: data.len(),
349            }),
350        }
351    }
352
353    fn handle_write_result(&mut self, id: usize, result: IoResult) {
354        match result {
355            IoResult::Ok => self
356                .resp_send
357                .send(DeviceResponse::Write { id, result: Ok(()) }),
358            IoResult::Err(io_error) => self.resp_send.send(DeviceResponse::Write {
359                id,
360                result: Err(io_error),
361            }),
362            IoResult::Defer(token) => self.deferred_writes.push(DeferredWrite { id, token }),
363        }
364    }
365}
chipset_device_worker/worker.rs

chipset_device_worker/
worker.rs
