Skip to content

Commit

Permalink
vmbus: delay interrupt creation until open response (#649)
Browse files Browse the repository at this point in the history 
Don't require the guest-to-host interrupt be provided at channel offer
time. Wait until open time. This is more flexible, and in the future
will allow us to remove some extra synchronziation in `vmbus_relay`.
  • Loading branch information
@jstarks
jstarks authored Jan 10, 2025
1 parent 1bd3678 commit ec7cfcb
Show file tree
Hide file tree
Showing 7 changed files with 272 additions and 185 deletions.
49 changes: 25 additions & 24 deletions vm/devices/net/netvsp/src/test.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ use vmbus_channel::bus::OfferInput;
use vmbus_channel::bus::OfferResources;
use vmbus_channel::bus::OpenData;
use vmbus_channel::bus::OpenRequest;
use vmbus_channel::bus::OpenResult;
use vmbus_channel::bus::ParentBus;
use vmbus_channel::channel::offer_channel;
use vmbus_channel::channel::ChannelHandle;
Expand Down Expand Up @@ -74,7 +75,7 @@ use zerocopy::FromZeroes;
const VMNIC_CHANNEL_TYPE_GUID: Guid = Guid::from_static_str("f8615163-df3e-46c5-913f-f2d2f965ed0e");

enum ChannelResponse {
Open(bool),
Open(Option<OpenResult>),
Close,
Gpadl(bool),
// TeardownGpadl(GpadlId),
Expand Down Expand Up @@ -432,21 +433,18 @@ impl TestNicDevice {
.await
.expect("open successful");

if let ChannelResponse::Open(response) = open_response {
assert_eq!(response, true);
} else {
let ChannelResponse::Open(Some(result)) = open_response else {
panic!("Unexpected return value");
}
};

let mem = self.mock_vmbus.memory.clone();
let guest_to_host_interrupt = self.offer_input.event.clone();
TestNicChannel::new(
self,
&mem,
gpadl_map,
ring_gpadl_id,
host_to_guest_event,
guest_to_host_interrupt,
result.guest_to_host_interrupt,
)
}

Expand Down Expand Up @@ -478,7 +476,7 @@ impl TestNicDevice {
next_avail_guest_page: usize,
next_avail_gpadl_id: u32,
host_to_guest_interrupt: Interrupt,
) -> anyhow::Result<()> {
) -> anyhow::Result<Option<Interrupt>> {
// Restore the previous memory settings
assert_eq!(self.next_avail_gpadl_id, 1);
self.next_avail_gpadl_id = next_avail_gpadl_id;
Expand All @@ -497,6 +495,7 @@ impl TestNicDevice {
})
.collect::<Vec<(GpadlId, MultiPagedRangeBuf<Vec<u64>>)>>();

let mut guest_to_host_interrupt = None;
mesh::CancelContext::new()
.with_timeout(Duration::from_millis(1000))
.until_cancelled(async {
Expand All @@ -519,7 +518,8 @@ impl TestNicDevice {
accepted: true,
}
}).collect::<Vec<vmbus_channel::bus::RestoredGpadl>>();
rpc.handle_sync(|_open| {
rpc.handle_sync(|open| {
guest_to_host_interrupt = open.map(|open| open.guest_to_host_interrupt);
Ok(vmbus_channel::bus::RestoreResult {
open_request: Some(OpenRequest {
open_data: OpenData {
Expand All @@ -545,7 +545,9 @@ impl TestNicDevice {
}
})
.await
.unwrap()
.unwrap()?;

Ok(guest_to_host_interrupt)
}
}

Expand Down Expand Up @@ -976,7 +978,6 @@ impl<'a> TestNicChannel<'a> {
buffer: SavedStateBlob,
) -> anyhow::Result<TestNicChannel<'_>> {
let mem = self.nic.mock_vmbus.memory.clone();
let guest_to_host_interrupt = nic.offer_input.event.clone();
let host_to_guest_interrupt = {
let event = self.host_to_guest_event.clone();
Interrupt::from_fn(move || event.signal())
Expand All @@ -986,27 +987,27 @@ impl<'a> TestNicChannel<'a> {
let channel_id = self.channel_id;
let next_avail_guest_page = self.nic.next_avail_guest_page;
let next_avail_gpadl_id = self.nic.next_avail_gpadl_id;
let restored_channel = TestNicChannel::new(
nic,
&mem,
gpadl_map.clone(),
channel_id,
self.host_to_guest_event,
guest_to_host_interrupt,
);

restored_channel
.nic
let guest_to_host_interrupt = nic
.restore(
buffer,
gpadl_map,
gpadl_map.clone(),
channel_id,
next_avail_guest_page,
next_avail_gpadl_id,
host_to_guest_interrupt,
)
.await?;
Ok(restored_channel)
.await?
.expect("should be open");

Ok(TestNicChannel::new(
nic,
&mem,
gpadl_map,
channel_id,
self.host_to_guest_event,
guest_to_host_interrupt,
))
}
}

Expand Down
19 changes: 13 additions & 6 deletions vm/devices/vmbus/vmbus_channel/src/bus.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,6 @@ use vmcore::interrupt::Interrupt;
pub struct OfferInput {
/// Parameters describing the offer.
pub params: OfferParams,
/// The event to signal when the guest needs attention.
pub event: Interrupt,
/// A mesh channel to send channel-related requests to.
pub request_send: mesh::Sender<ChannelRequest>,
/// A mesh channel to receive channel-related requests to.
Expand Down Expand Up @@ -80,7 +78,7 @@ impl OfferResources {
#[derive(Debug, MeshPayload)]
pub enum ChannelRequest {
/// Open the channel.
Open(Rpc<OpenRequest, bool>),
Open(Rpc<OpenRequest, Option<OpenResult>>),
/// Close the channel.
Close(Rpc<(), ()>),
/// Create a new GPADL.
Expand All @@ -91,6 +89,14 @@ pub enum ChannelRequest {
Modify(Rpc<ModifyRequest, i32>),
}

/// The successful result of an open request.
#[derive(Debug, MeshPayload)]
pub struct OpenResult {
/// The interrupt object vmbus should signal when the guest signals the
/// host.
pub guest_to_host_interrupt: Interrupt,
}

/// GPADL information from the guest.
#[derive(Debug, MeshPayload)]
pub struct GpadlRequest {
Expand All @@ -117,8 +123,8 @@ pub enum ModifyRequest {
pub enum ChannelServerRequest {
/// A request to restore the channel.
///
/// The input parameter is whether the channel was saved open.
Restore(FailableRpc<bool, RestoreResult>),
/// The input parameter provides the open result if the channel was saved open.
Restore(FailableRpc<Option<OpenResult>, RestoreResult>),
/// A request to revoke the channel.
///
/// A channel can also be revoked by dropping it. This request is only necessary if you need to
Expand Down Expand Up @@ -165,7 +171,8 @@ pub trait ParentBus: Send + Sync {
/// time.
fn clone_bus(&self) -> Box<dyn ParentBus>;

/// Returns whether [`OfferInput::event`] needs to be backed by an OS event.
/// Returns whether [`OpenResult::guest_to_host_interrupt`] needs to be
/// backed by an OS event.
///
/// TODO: Remove this and just return the appropriate notify type directly
/// once subchannel creation and enable are separated.
Expand Down
60 changes: 31 additions & 29 deletions vm/devices/vmbus/vmbus_channel/src/channel.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ use crate::bus::OfferInput;
use crate::bus::OfferParams;
use crate::bus::OfferResources;
use crate::bus::OpenRequest;
use crate::bus::OpenResult;
use crate::bus::ParentBus;
use crate::gpadl::GpadlMap;
use crate::gpadl::GpadlMapView;
Expand Down Expand Up @@ -335,20 +336,19 @@ async fn offer_generic(
let (state_req_send, state_req_recv) = mesh::channel();

let use_event = bus.use_event();
let new_event = || {
if use_event {
Notify::from_event(Event::new())
} else {
Notify::from_slim_event(Arc::new(SlimEvent::new()))
}
};

let event = new_event();
let subchannel_events: Vec<_> = (0..max_subchannels).map(|_| new_event()).collect();
let events: Vec<_> = (0..max_subchannels + 1)
.map(|_| {
if use_event {
Notify::from_event(Event::new())
} else {
Notify::from_slim_event(Arc::new(SlimEvent::new()))
}
})
.collect();

let request = OfferInput {
params: offer,
event: event.clone().interrupt(),
request_send,
server_request_recv,
};
Expand All @@ -357,12 +357,12 @@ async fn offer_generic(

let offer_result = bus.add_child(request).await?;

let mut resources = vec![ChannelResources { event }];
for idx in 0..max_subchannels {
resources.push(ChannelResources {
event: subchannel_events[idx as usize].clone(),
});
}
let resources = events
.iter()
.map(|event| ChannelResources {
event: event.clone(),
})
.collect();

let (subchannel_enable_send, subchannel_enable_recv) = mesh::channel();
channel.install(DeviceResources {
Expand All @@ -380,7 +380,7 @@ async fn offer_generic(
let device = Device::new(
request_recv,
server_request_send,
subchannel_events,
events,
gpadl_map,
subchannel_enable_recv,
);
Expand Down Expand Up @@ -446,7 +446,7 @@ struct Device {
open: Vec<bool>,
subchannel_gpadls: Vec<BTreeSet<GpadlId>>,
requests: SelectAll<TaggedStream<usize, mesh::Receiver<ChannelRequest>>>,
subchannel_events: Vec<Notify>,
events: Vec<Notify>,
gpadl_map: Arc<GpadlMap>,
subchannel_enable_recv: mesh::Receiver<u16>,
}
Expand All @@ -455,7 +455,7 @@ impl Device {
fn new(
request_recv: mesh::Receiver<ChannelRequest>,
server_request_send: mesh::Sender<ChannelServerRequest>,
subchannel_events: Vec<Notify>,
events: Vec<Notify>,
gpadl_map: Arc<GpadlMap>,
subchannel_enable_recv: mesh::Receiver<u16>,
) -> Self {
Expand All @@ -469,7 +469,7 @@ impl Device {
open,
subchannel_gpadls,
requests,
subchannel_events,
events,
gpadl_map,
subchannel_enable_recv,
}
Expand Down Expand Up @@ -585,7 +585,7 @@ impl Device {
channel: &mut dyn VmbusDevice,
channel_idx: usize,
open_request: OpenRequest,
) -> bool {
) -> Option<OpenResult> {
assert!(!self.open[channel_idx]);
// N.B. Any asynchronous GPADL requests will block while in
// open(). This should be fine for all known devices.
Expand All @@ -594,11 +594,13 @@ impl Device {
error = error.as_ref() as &dyn std::error::Error,
"failed to open channel"
);
false
None
} else {
true
Some(OpenResult {
guest_to_host_interrupt: self.events[channel_idx].clone().interrupt(),
})
};
self.open[channel_idx] = opened;
self.open[channel_idx] = opened.is_some();
opened
}

Expand Down Expand Up @@ -739,9 +741,6 @@ impl Device {
subchannel_index: subchannel_idx as u16,
..offer.clone()
},
event: self.subchannel_events[subchannel_idx - 1]
.clone()
.interrupt(),
request_send,
server_request_recv,
};
Expand Down Expand Up @@ -778,9 +777,12 @@ impl Device {
.map_err(ChannelRestoreError::EnablingSubchannels)?;

let mut results = Vec::with_capacity(states.len());
for (channel_idx, open) in states.iter().copied().enumerate() {
for (channel_idx, (open, event)) in states.iter().copied().zip(&self.events).enumerate() {
let open_result = open.then(|| OpenResult {
guest_to_host_interrupt: event.clone().interrupt(),
});
let result = self.server_requests[channel_idx]
.call_failable(ChannelServerRequest::Restore, open)
.call_failable(ChannelServerRequest::Restore, open_result)
.await
.map_err(|err| ChannelRestoreError::RestoreError(err.into()))?;

Expand Down
14 changes: 8 additions & 6 deletions vm/devices/vmbus/vmbus_channel/src/offer.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ use crate::bus::OfferInput;
use crate::bus::OfferParams;
use crate::bus::OfferResources;
use crate::bus::OpenRequest;
use crate::bus::OpenResult;
use crate::bus::ParentBus;
use crate::gpadl::GpadlMap;
use crate::gpadl::GpadlMapView;
Expand Down Expand Up @@ -70,7 +71,6 @@ impl Offer {
let result = bus
.add_child(OfferInput {
params: offer_params,
event: Interrupt::from_event(event.clone()),
request_send,
server_request_recv,
})
Expand Down Expand Up @@ -181,7 +181,9 @@ impl Offer {
channel,
gpadl_map: self.gpadl_map.clone(),
};
message.response.respond(true);
message.response.respond(Some(OpenResult {
guest_to_host_interrupt: self.event.clone().interrupt(),
}));
Ok(resources)
}

Expand Down Expand Up @@ -220,18 +222,18 @@ struct OpenMessage {
response: OpenResponse,
}

struct OpenResponse(Option<Rpc<(), bool>>);
struct OpenResponse(Option<Rpc<(), Option<OpenResult>>>);

impl OpenResponse {
fn respond(mut self, open: bool) {
self.0.take().unwrap().complete(open)
fn respond(mut self, result: Option<OpenResult>) {
self.0.take().unwrap().complete(result)
}
}

impl Drop for OpenResponse {
fn drop(&mut self) {
if let Some(rpc) = self.0.take() {
rpc.complete(false);
rpc.complete(None);
}
}
}
Expand Down
Loading

0 comments on commit ec7cfcb

Please sign in to comment.