vmbus_channel/
offer.rs

1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4//! Vmbus channel offer support.
5
6use crate::ChannelClosed;
7use crate::RawAsyncChannel;
8use crate::SignalVmbusChannel;
9use crate::bus::ChannelRequest;
10use crate::bus::ChannelServerRequest;
11use crate::bus::OfferInput;
12use crate::bus::OfferParams;
13use crate::bus::OfferResources;
14use crate::bus::OpenRequest;
15use crate::bus::OpenResult;
16use crate::bus::ParentBus;
17use crate::gpadl::GpadlMap;
18use crate::gpadl::GpadlMapView;
19use crate::gpadl_ring;
20use crate::gpadl_ring::GpadlRingMem;
21use crate::gpadl_ring::make_rings;
22use futures::StreamExt;
23use mesh::rpc::Rpc;
24use pal_async::driver::Driver;
25use pal_async::task::Spawn;
26use pal_async::task::Task;
27use pal_event::Event;
28use std::sync::Arc;
29use std::sync::atomic::AtomicBool;
30use std::sync::atomic::Ordering;
31use vmbus_ring::gparange::MultiPagedRangeBuf;
32use vmcore::interrupt::Interrupt;
33use vmcore::notify::Notify;
34use vmcore::notify::PolledNotify;
35
36/// A channel accept error.
37#[derive(Debug, thiserror::Error)]
38pub enum Error {
39    /// channel revoked
40    #[error("the channel has been revoked")]
41    Revoked,
42    /// GPADL ring buffer error
43    #[error(transparent)]
44    GpadlRing(#[from] gpadl_ring::Error),
45    /// Driver error
46    #[error("io driver error")]
47    Driver(#[source] std::io::Error),
48}
49
50/// A channel offer.
51pub struct Offer {
52    task: Task<()>,
53    open_recv: mesh::Receiver<OpenMessage>,
54    gpadl_map: GpadlMapView,
55    event: Notify,
56    offer_resources: OfferResources,
57    _server_request_send: mesh::Sender<ChannelServerRequest>,
58}
59
60impl Offer {
61    /// Offers a new channel.
62    pub async fn new(
63        driver: impl Spawn,
64        bus: &dyn ParentBus,
65        offer_params: OfferParams,
66    ) -> anyhow::Result<Self> {
67        let instance_id = offer_params.instance_id;
68        let event = Event::new();
69        let (request_send, request_recv) = mesh::channel();
70        let (server_request_send, server_request_recv) = mesh::channel();
71        let result = bus
72            .add_child(OfferInput {
73                params: offer_params,
74                request_send,
75                server_request_recv,
76            })
77            .await?;
78
79        let gpadls = GpadlMap::new();
80        let gpadl_map = gpadls.clone().view();
81        let (open_send, open_recv) = mesh::channel();
82        let task = driver.spawn(format!("vmbus-offer-{}", instance_id), {
83            let event = event.clone();
84            async move { Self::task(event, gpadls, request_recv, open_send).await }
85        });
86
87        let offer = Self {
88            offer_resources: result,
89            task,
90            open_recv,
91            gpadl_map,
92            event: Notify::from_event(event),
93            _server_request_send: server_request_send,
94        };
95        Ok(offer)
96    }
97
98    async fn task(
99        event: Event,
100        gpadls: Arc<GpadlMap>,
101        mut request_recv: mesh::Receiver<ChannelRequest>,
102        send: mesh::Sender<OpenMessage>,
103    ) {
104        let mut open_done = None;
105        while let Ok(request) = request_recv.recv().await {
106            match request {
107                ChannelRequest::Open(rpc) => {
108                    let (open_request, response_send) = rpc.split();
109                    let done = Arc::new(AtomicBool::new(false));
110                    send.send(OpenMessage {
111                        open_request,
112                        done: done.clone(),
113                        response: OpenResponse(Some(response_send)),
114                    });
115                    open_done = Some(done);
116                }
117                ChannelRequest::Close(rpc) => {
118                    let _response_send = rpc; // TODO: figure out if we should really just drop this here.
119                    open_done
120                        .take()
121                        .expect("channel must be open")
122                        .store(true, Ordering::Relaxed);
123                    event.signal();
124                }
125                ChannelRequest::Gpadl(rpc) => {
126                    rpc.handle_sync(|gpadl| {
127                        match MultiPagedRangeBuf::new(gpadl.count.into(), gpadl.buf) {
128                            Ok(buf) => {
129                                gpadls.add(gpadl.id, buf);
130                                true
131                            }
132                            Err(err) => {
133                                tracelimit::error_ratelimited!(
134                                    error = &err as &dyn std::error::Error,
135                                    "failed to parse gpadl"
136                                );
137                                false
138                            }
139                        }
140                    })
141                }
142                ChannelRequest::TeardownGpadl(rpc) => {
143                    let (id, response_send) = rpc.split();
144                    if let Some(f) = gpadls.remove(
145                        id,
146                        Box::new(move || {
147                            response_send.complete(());
148                        }),
149                    ) {
150                        f();
151                    }
152                }
153                ChannelRequest::Modify(rpc) => rpc.handle_sync(|_| 0),
154            }
155        }
156    }
157
158    /// Accepts a channel open request from the guest.
159    pub async fn accept(
160        &mut self,
161        driver: &(impl Driver + ?Sized),
162    ) -> Result<OpenChannelResources, Error> {
163        let message = self.open_recv.next().await.ok_or(Error::Revoked)?;
164
165        let (in_ring, out_ring) = make_rings(
166            self.offer_resources.ring_memory(&message.open_request),
167            &self.gpadl_map,
168            &message.open_request.open_data,
169        )?;
170        let event = OfferChannelSignal {
171            event: self.event.clone().pollable(driver).map_err(Error::Driver)?,
172            interrupt: message.open_request.interrupt.clone(),
173            done: message.done,
174        };
175        let channel = RawAsyncChannel {
176            in_ring,
177            out_ring,
178            signal: Box::new(event),
179        };
180        let resources = OpenChannelResources {
181            channel,
182            gpadl_map: self.gpadl_map.clone(),
183        };
184        message.response.respond(Some(OpenResult {
185            guest_to_host_interrupt: self.event.clone().interrupt(),
186        }));
187        Ok(resources)
188    }
189
190    /// Revokes the channel.
191    pub async fn revoke(self) {
192        drop(self.open_recv);
193        self.task.await;
194    }
195}
196
197struct OfferChannelSignal {
198    event: PolledNotify,
199    interrupt: Interrupt,
200    done: Arc<AtomicBool>,
201}
202
203impl SignalVmbusChannel for OfferChannelSignal {
204    fn signal_remote(&self) {
205        self.interrupt.deliver();
206    }
207
208    fn poll_for_signal(
209        &self,
210        cx: &mut std::task::Context<'_>,
211    ) -> std::task::Poll<Result<(), ChannelClosed>> {
212        if self.done.load(Ordering::Relaxed) {
213            return Err(ChannelClosed).into();
214        }
215        self.event.poll_wait(cx).map(Ok)
216    }
217}
218
219struct OpenMessage {
220    open_request: OpenRequest,
221    done: Arc<AtomicBool>,
222    response: OpenResponse,
223}
224
225struct OpenResponse(Option<Rpc<(), Option<OpenResult>>>);
226
227impl OpenResponse {
228    fn respond(mut self, result: Option<OpenResult>) {
229        self.0.take().unwrap().complete(result)
230    }
231}
232
233impl Drop for OpenResponse {
234    fn drop(&mut self) {
235        if let Some(rpc) = self.0.take() {
236            rpc.complete(None);
237        }
238    }
239}
240
241/// Channel resources for an open channel.
242pub struct OpenChannelResources {
243    /// The channel ring buffer and interrupt state.
244    pub channel: RawAsyncChannel<GpadlRingMem>,
245    /// The channel's GPADL map.
246    pub gpadl_map: GpadlMapView,
247}
vmbus_channel/offer.rs

vmbus_channel/
offer.rs
