use std::fmt;
use std::fmt::Display;
use std::sync::Arc;
use std::sync::MutexGuard;
use anyhow::Context;
use base::debug;
use base::error;
use base::info;
use base::Error as SysError;
use base::Event;
use base::EventType;
use remain::sorted;
use sync::Mutex;
use thiserror::Error;
use vm_memory::GuestAddress;
use vm_memory::GuestMemory;
use super::ring_buffer::RingBuffer;
use super::ring_buffer_stop_cb::RingBufferStopCallback;
use super::xhci_abi::TransferDescriptor;
use crate::utils;
use crate::utils::EventHandler;
use crate::utils::EventLoop;
#[sorted]
#[derive(Error, Debug)]
pub enum Error {
#[error("failed to add event to event loop: {0}")]
AddEvent(utils::Error),
#[error("failed to create event: {0}")]
CreateEvent(SysError),
}
type Result<T> = std::result::Result<T, Error>;
#[derive(PartialEq, Copy, Clone, Eq)]
enum RingBufferState {
Running,
Stopping,
Stopped,
}
pub trait TransferDescriptorHandler {
fn handle_transfer_descriptor(
&self,
descriptor: TransferDescriptor,
complete_event: Event,
) -> anyhow::Result<()>;
fn stop(&self) -> bool {
true
}
}
pub struct RingBufferController<T: 'static + TransferDescriptorHandler> {
name: String,
state: Mutex<RingBufferState>,
stop_callback: Mutex<Vec<RingBufferStopCallback>>,
ring_buffer: Mutex<RingBuffer>,
handler: Mutex<T>,
event_loop: Arc<EventLoop>,
event: Event,
}
impl<T: 'static + TransferDescriptorHandler> Display for RingBufferController<T> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "RingBufferController `{}`", self.name)
}
}
impl<T: Send> RingBufferController<T>
where
T: 'static + TransferDescriptorHandler,
{
pub fn new_with_handler(
name: String,
mem: GuestMemory,
event_loop: Arc<EventLoop>,
handler: T,
) -> Result<Arc<RingBufferController<T>>> {
let evt = Event::new().map_err(Error::CreateEvent)?;
let controller = Arc::new(RingBufferController {
name: name.clone(),
state: Mutex::new(RingBufferState::Stopped),
stop_callback: Mutex::new(Vec::new()),
ring_buffer: Mutex::new(RingBuffer::new(name, mem)),
handler: Mutex::new(handler),
event_loop: event_loop.clone(),
event: evt,
});
let event_handler: Arc<dyn EventHandler> = controller.clone();
event_loop
.add_event(
&controller.event,
EventType::Read,
Arc::downgrade(&event_handler),
)
.map_err(Error::AddEvent)?;
Ok(controller)
}
fn lock_ring_buffer(&self) -> MutexGuard<RingBuffer> {
self.ring_buffer.lock()
}
pub fn get_dequeue_pointer(&self) -> GuestAddress {
self.lock_ring_buffer().get_dequeue_pointer()
}
pub fn set_dequeue_pointer(&self, ptr: GuestAddress) {
xhci_trace!("{}: set_dequeue_pointer({:x})", self.name, ptr.0);
self.lock_ring_buffer().set_dequeue_pointer(ptr);
}
pub fn get_consumer_cycle_state(&self) -> bool {
self.lock_ring_buffer().get_consumer_cycle_state()
}
pub fn set_consumer_cycle_state(&self, state: bool) {
xhci_trace!("{}: set consumer cycle state: {}", self.name, state);
self.lock_ring_buffer().set_consumer_cycle_state(state);
}
pub fn start(&self) {
xhci_trace!("start {}", self.name);
let mut state = self.state.lock();
if *state != RingBufferState::Running {
*state = RingBufferState::Running;
if let Err(e) = self.event.signal() {
error!("cannot start event ring: {}", e);
}
}
}
pub fn stop(&self, callback: RingBufferStopCallback) {
xhci_trace!("stop {}", self.name);
let mut state = self.state.lock();
if *state == RingBufferState::Stopped {
info!("xhci: {} is already stopped", self.name);
return;
}
if self.handler.lock().stop() {
*state = RingBufferState::Stopping;
self.stop_callback.lock().push(callback);
} else {
*state = RingBufferState::Stopped;
}
}
}
impl<T> Drop for RingBufferController<T>
where
T: 'static + TransferDescriptorHandler,
{
fn drop(&mut self) {
if let Err(e) = self.event_loop.remove_event_for_descriptor(&self.event) {
error!(
"cannot remove ring buffer controller from event loop: {}",
e
);
}
}
}
impl<T> EventHandler for RingBufferController<T>
where
T: 'static + TransferDescriptorHandler + Send,
{
fn on_event(&self) -> anyhow::Result<()> {
self.event.wait().context("cannot read from event")?;
let mut state = self.state.lock();
match *state {
RingBufferState::Stopped => return Ok(()),
RingBufferState::Stopping => {
debug!("xhci: {}: stopping ring buffer controller", self.name);
*state = RingBufferState::Stopped;
self.stop_callback.lock().clear();
return Ok(());
}
RingBufferState::Running => {}
}
let transfer_descriptor = self
.lock_ring_buffer()
.dequeue_transfer_descriptor()
.context("cannot dequeue transfer descriptor")?;
let transfer_descriptor = match transfer_descriptor {
Some(t) => t,
None => {
*state = RingBufferState::Stopped;
self.stop_callback.lock().clear();
return Ok(());
}
};
let event = self.event.try_clone().context("cannot clone event")?;
self.handler
.lock()
.handle_transfer_descriptor(transfer_descriptor, event)
}
}
#[cfg(test)]
mod tests {
use std::mem::size_of;
use std::sync::mpsc::channel;
use std::sync::mpsc::Sender;
use base::pagesize;
use super::super::xhci_abi::LinkTrb;
use super::super::xhci_abi::NormalTrb;
use super::super::xhci_abi::Trb;
use super::super::xhci_abi::TrbType;
use super::*;
struct TestHandler {
sender: Sender<i32>,
}
impl TransferDescriptorHandler for TestHandler {
fn handle_transfer_descriptor(
&self,
descriptor: TransferDescriptor,
complete_event: Event,
) -> anyhow::Result<()> {
for atrb in descriptor {
assert_eq!(atrb.trb.get_trb_type().unwrap(), TrbType::Normal);
self.sender.send(atrb.trb.get_parameter() as i32).unwrap();
}
complete_event.signal().unwrap();
Ok(())
}
}
fn setup_mem() -> GuestMemory {
let trb_size = size_of::<Trb>() as u64;
let gm = GuestMemory::new(&[(GuestAddress(0), pagesize() as u64)]).unwrap();
let mut trb = NormalTrb::new();
trb.set_trb_type(TrbType::Normal);
trb.set_data_buffer(1);
trb.set_chain(true);
gm.write_obj_at_addr(trb, GuestAddress(0x100)).unwrap();
trb.set_data_buffer(2);
gm.write_obj_at_addr(trb, GuestAddress(0x100 + trb_size))
.unwrap();
let mut ltrb = LinkTrb::new();
ltrb.set_trb_type(TrbType::Link);
ltrb.set_ring_segment_pointer(0x200);
gm.write_obj_at_addr(ltrb, GuestAddress(0x100 + 2 * trb_size))
.unwrap();
trb.set_data_buffer(3);
gm.write_obj_at_addr(trb, GuestAddress(0x200)).unwrap();
trb.set_data_buffer(4);
trb.set_chain(false);
gm.write_obj_at_addr(trb, GuestAddress(0x200 + 1 * trb_size))
.unwrap();
ltrb.set_ring_segment_pointer(0x300);
gm.write_obj_at_addr(ltrb, GuestAddress(0x200 + 2 * trb_size))
.unwrap();
trb.set_data_buffer(5);
trb.set_chain(true);
gm.write_obj_at_addr(trb, GuestAddress(0x300)).unwrap();
trb.set_data_buffer(6);
trb.set_chain(false);
gm.write_obj_at_addr(trb, GuestAddress(0x300 + 1 * trb_size))
.unwrap();
ltrb.set_ring_segment_pointer(0x100);
gm.write_obj_at_addr(ltrb, GuestAddress(0x300 + 2 * trb_size))
.unwrap();
gm
}
#[test]
fn test_ring_buffer_controller() {
let (tx, rx) = channel();
let mem = setup_mem();
let (l, j) = EventLoop::start("test".to_string(), None).unwrap();
let l = Arc::new(l);
let controller = RingBufferController::new_with_handler(
"".to_string(),
mem,
l.clone(),
TestHandler { sender: tx },
)
.unwrap();
controller.set_dequeue_pointer(GuestAddress(0x100));
controller.set_consumer_cycle_state(false);
controller.start();
assert_eq!(rx.recv().unwrap(), 1);
assert_eq!(rx.recv().unwrap(), 2);
assert_eq!(rx.recv().unwrap(), 3);
assert_eq!(rx.recv().unwrap(), 4);
assert_eq!(rx.recv().unwrap(), 5);
assert_eq!(rx.recv().unwrap(), 6);
l.stop();
j.join().unwrap();
}
}