virtio_accel_transport/
queue.rs

1//! Ownership-safe virtqueue lifecycle and notification ports.
2
3use core::num::{NonZeroU16, NonZeroU64};
4
5use crate::{ChainLayoutError, ChainRegion};
6
7/// Maximum queue size permitted by the split-ring index representation.
8pub const MAX_SPLIT_QUEUE_SIZE: u16 = 32_768;
9
10/// Valid nonzero, power-of-two split-virtqueue size.
11#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
12#[repr(transparent)]
13pub struct QueueSize(NonZeroU16);
14
15impl QueueSize {
16    /// Validate a split-virtqueue size.
17    pub const fn new(value: u16) -> Option<Self> {
18        if value.is_power_of_two() && value <= MAX_SPLIT_QUEUE_SIZE {
19            match NonZeroU16::new(value) {
20                Some(value) => Some(Self(value)),
21                None => None,
22            }
23        } else {
24            None
25        }
26    }
27
28    /// Return the validated queue size.
29    pub const fn get(self) -> u16 {
30        self.0.get()
31    }
32}
33
34/// Monotonic queue-reset epoch used to reject stale chain operations.
35#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
36#[repr(transparent)]
37pub struct QueueEpoch(NonZeroU64);
38
39impl QueueEpoch {
40    /// Initial epoch for a newly constructed queue.
41    pub const INITIAL: Self = match NonZeroU64::new(1) {
42        Some(value) => Self(value),
43        None => panic!("one is nonzero"),
44    };
45
46    /// Construct a nonzero epoch.
47    pub const fn new(value: u64) -> Option<Self> {
48        match NonZeroU64::new(value) {
49            Some(value) => Some(Self(value)),
50            None => None,
51        }
52    }
53
54    /// Return the raw epoch value.
55    pub const fn get(self) -> u64 {
56        self.0.get()
57    }
58
59    /// Return the next epoch, or `None` if the epoch space is exhausted.
60    pub const fn checked_next(self) -> Option<Self> {
61        match self.get().checked_add(1) {
62            Some(value) => Self::new(value),
63            None => None,
64        }
65    }
66}
67
68/// Opaque queue-chain identity scoped to one reset epoch.
69#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
70pub struct ChainId {
71    epoch: QueueEpoch,
72    token: u64,
73}
74
75impl ChainId {
76    /// Construct an identity from an epoch and transport-owned token.
77    pub const fn new(epoch: QueueEpoch, token: u64) -> Self {
78        Self { epoch, token }
79    }
80
81    /// Epoch in which this chain was published.
82    pub const fn epoch(self) -> QueueEpoch {
83        self.epoch
84    }
85
86    /// Opaque transport token, such as a split-ring descriptor head.
87    pub const fn token(self) -> u64 {
88        self.token
89    }
90}
91
92/// Exact number of device-written bytes published with a used chain.
93#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
94#[repr(transparent)]
95pub struct UsedLength(u32);
96
97impl UsedLength {
98    /// Construct an exact used length.
99    pub const fn new(bytes: u32) -> Self {
100        Self(bytes)
101    }
102
103    /// Return the used byte count.
104    pub const fn get(self) -> u32 {
105        self.0
106    }
107}
108
109/// Validated queue configuration and lifecycle snapshot.
110#[derive(Clone, Copy, Debug, PartialEq, Eq)]
111pub struct QueueState {
112    max_size: QueueSize,
113    size: Option<QueueSize>,
114    ready: bool,
115    epoch: QueueEpoch,
116}
117
118impl QueueState {
119    /// Construct an unconfigured queue state.
120    pub const fn unconfigured(max_size: QueueSize, epoch: QueueEpoch) -> Self {
121        Self {
122            max_size,
123            size: None,
124            ready: false,
125            epoch,
126        }
127    }
128
129    /// Construct and validate a queue-state snapshot.
130    pub const fn new(
131        max_size: QueueSize,
132        size: Option<QueueSize>,
133        ready: bool,
134        epoch: QueueEpoch,
135    ) -> Result<Self, QueueConfigError> {
136        if let Some(size) = size {
137            if size.get() > max_size.get() {
138                return Err(QueueConfigError::SizeExceedsMaximum);
139            }
140        }
141        if ready && size.is_none() {
142            return Err(QueueConfigError::ReadyWithoutSize);
143        }
144        Ok(Self {
145            max_size,
146            size,
147            ready,
148            epoch,
149        })
150    }
151
152    /// Maximum queue size supported by the implementation.
153    pub const fn max_size(self) -> QueueSize {
154        self.max_size
155    }
156
157    /// Configured queue size, if any.
158    pub const fn size(self) -> Option<QueueSize> {
159        self.size
160    }
161
162    /// Whether descriptor publication and consumption are enabled.
163    pub const fn ready(self) -> bool {
164        self.ready
165    }
166
167    /// Current reset epoch.
168    pub const fn epoch(self) -> QueueEpoch {
169        self.epoch
170    }
171}
172
173/// Invalid queue configuration.
174#[derive(Clone, Copy, Debug, PartialEq, Eq)]
175pub enum QueueConfigError {
176    /// The configured queue size exceeds the implementation maximum.
177    SizeExceedsMaximum,
178    /// A queue cannot become ready before a size is configured.
179    ReadyWithoutSize,
180    /// A reset attempted to reuse or move backwards from the current epoch.
181    NonIncreasingEpoch,
182}
183
184/// Portable queue operation failure.
185#[derive(Clone, Copy, Debug, PartialEq, Eq)]
186pub enum QueueError<E> {
187    /// The queue is not configured and ready for the requested operation.
188    NotReady,
189    /// The queue configuration is invalid.
190    InvalidConfiguration(QueueConfigError),
191    /// An operation belongs to a different reset epoch.
192    ResetRace {
193        /// Epoch carried by the operation.
194        operation: QueueEpoch,
195        /// Current queue epoch.
196        current: QueueEpoch,
197    },
198    /// Completion used length exceeds writable chain capacity.
199    UsedLengthExceeded {
200        /// Attempted used length.
201        used: UsedLength,
202        /// Writable capacity of the consumed chain.
203        capacity: u64,
204    },
205    /// Concrete transport failure.
206    Transport(E),
207}
208
209/// Structurally malformed descriptor-chain classification.
210#[derive(Clone, Copy, Debug, PartialEq, Eq)]
211pub enum MalformedChain {
212    /// A descriptor chain contains a loop.
213    DescriptorLoop,
214    /// A descriptor index is outside the configured queue.
215    DescriptorIndex,
216    /// A descriptor contains flags outside the negotiated split-ring profile.
217    DescriptorFlags,
218    /// The flattened descriptor count is invalid.
219    DescriptorCount,
220    /// A descriptor has zero length.
221    ZeroLength,
222    /// Readable and writable descriptor ordering is invalid.
223    Direction,
224    /// Descriptor byte totals overflow.
225    LengthOverflow,
226    /// Mapped byte ports do not match descriptor byte totals.
227    PortLengthMismatch,
228    /// An indirect descriptor was offered without negotiation.
229    IndirectUnsupported,
230    /// An indirect descriptor table is structurally invalid.
231    IndirectMalformed,
232    /// A descriptor range cannot be mapped for the required access.
233    Address,
234}
235
236impl From<ChainLayoutError> for MalformedChain {
237    fn from(error: ChainLayoutError) -> Self {
238        match error {
239            ChainLayoutError::DescriptorCount => Self::DescriptorCount,
240            ChainLayoutError::ZeroLength => Self::ZeroLength,
241            ChainLayoutError::Direction => Self::Direction,
242            ChainLayoutError::LengthOverflow => Self::LengthOverflow,
243            ChainLayoutError::PortLengthMismatch => Self::PortLengthMismatch,
244        }
245    }
246}
247
248/// Failure to expose one consumed chain as validated byte ports.
249#[derive(Clone, Copy, Debug, PartialEq, Eq)]
250pub enum ChainError<E> {
251    /// The chain is malformed and should be completed with used length zero when recoverable.
252    Malformed(MalformedChain),
253    /// Reset invalidated this chain before byte access.
254    ResetRace {
255        /// Epoch carried by the chain.
256        chain: QueueEpoch,
257        /// Current queue epoch.
258        current: QueueEpoch,
259    },
260    /// Concrete transport or memory-access failure.
261    Transport(E),
262}
263
264/// Borrowed regions and byte ports for one consumed device chain.
265///
266/// The source and sink types remain generic so this crate does not depend on command semantics or a
267/// particular guest-memory library.
268#[derive(Debug)]
269pub struct ChainIo<'a, R: ?Sized, W: ?Sized> {
270    regions: &'a [ChainRegion],
271    request: &'a R,
272    response: &'a mut W,
273}
274
275impl<'a, R: ?Sized, W: ?Sized> ChainIo<'a, R, W> {
276    /// Construct a borrowed chain view after topology and mapping validation.
277    pub fn new(regions: &'a [ChainRegion], request: &'a R, response: &'a mut W) -> Self {
278        Self {
279            regions,
280            request,
281            response,
282        }
283    }
284
285    /// Consume the view into its region metadata, readable port, and writable port.
286    pub fn into_parts(self) -> (&'a [ChainRegion], &'a R, &'a mut W) {
287        (self.regions, self.request, self.response)
288    }
289}
290
291/// Result of exposing one consumed chain as mapped byte ports.
292pub type ChainIoResult<'a, R, W, E> = Result<ChainIo<'a, R, W>, ChainError<E>>;
293
294/// Consumed device-side descriptor chain.
295///
296/// Implementations must keep the mapped bytes alive until this value is consumed by
297/// [`DeviceQueue::complete`] or dropped during reset. The value must not be `Copy`; consuming it at
298/// completion makes double completion impossible in safe implementations.
299pub trait DeviceChain {
300    /// Device-readable byte-port type.
301    type Request: ?Sized;
302    /// Device-writable byte-port type.
303    type Response: ?Sized;
304    /// Concrete transport or memory-access error.
305    type Error;
306
307    /// Return this chain's reset-scoped identity.
308    ///
309    /// This operation must not block or allocate.
310    fn id(&self) -> ChainId;
311
312    /// Expose mapped request and response ports.
313    ///
314    /// This operation must not block or allocate. A reset-race result must expose no guest bytes.
315    fn io(&mut self) -> ChainIoResult<'_, Self::Request, Self::Response, Self::Error>;
316}
317
318/// Whether the peer should be notified after a publication operation.
319#[derive(Clone, Copy, Debug, PartialEq, Eq)]
320pub enum NotificationHint {
321    /// Notification is suppressed by current queue state.
322    Suppressed,
323    /// The peer should be notified after the publication barrier.
324    Notify,
325}
326
327/// Result of atomically enabling notifications and rechecking queue state.
328#[derive(Clone, Copy, Debug, PartialEq, Eq)]
329pub enum NotificationRecheck {
330    /// No work appeared before notification enablement became visible.
331    Idle,
332    /// Work is already pending and must be processed without sleeping.
333    WorkPending,
334}
335
336/// Read-only state shared by driver, device, and configuration ports.
337pub trait QueuePort {
338    /// Return a coherent queue-state snapshot.
339    ///
340    /// This operation must not block or allocate. Implementations using shared state must use
341    /// acquire ordering sufficient to observe the corresponding configuration and reset writes.
342    fn state(&self) -> QueueState;
343}
344
345/// Queue configuration control plane.
346pub trait QueueControl: QueuePort {
347    /// Concrete transport configuration error.
348    type Error;
349
350    /// Configure queue size while the queue is not ready.
351    ///
352    /// This operation must not block. It may allocate storage bounded by `size`; implementations
353    /// must report allocation failure through `Error` without changing the prior configuration.
354    fn configure(&mut self, size: QueueSize) -> Result<(), QueueError<Self::Error>>;
355
356    /// Enable or disable queue descriptor processing.
357    ///
358    /// This operation must not block or allocate. Making the queue ready is a release boundary for
359    /// the configured queue state.
360    fn set_ready(&mut self, ready: bool) -> Result<(), QueueError<Self::Error>>;
361}
362
363/// Successful driver publication.
364#[derive(Clone, Copy, Debug, PartialEq, Eq)]
365pub struct PublishedChain {
366    id: ChainId,
367    notification: NotificationHint,
368}
369
370impl PublishedChain {
371    /// Construct a publication result.
372    pub const fn new(id: ChainId, notification: NotificationHint) -> Self {
373        Self { id, notification }
374    }
375
376    /// Published chain identity.
377    pub const fn id(self) -> ChainId {
378        self.id
379    }
380
381    /// Notification decision made after publishing the available index.
382    pub const fn notification(self) -> NotificationHint {
383        self.notification
384    }
385}
386
387/// Pre-publication failure classification.
388#[derive(Clone, Copy, Debug, PartialEq, Eq)]
389pub enum PublishErrorKind<E> {
390    /// The queue is not configured and ready.
391    NotReady,
392    /// No free descriptor head or available-ring slot exists.
393    QueueFull,
394    /// The supplied command lacks enough descriptors or writable capacity.
395    InsufficientDescriptors,
396    /// Reset invalidated publication before ownership could transfer.
397    ResetRace {
398        /// Epoch in which publication began.
399        operation: QueueEpoch,
400        /// Current queue epoch.
401        current: QueueEpoch,
402    },
403    /// Concrete transport failure before ownership transfer.
404    Transport(E),
405}
406
407/// Failed driver publication that returns the unpublished chain for retry or reclamation.
408#[derive(Debug, PartialEq, Eq)]
409pub struct PublishError<C, E> {
410    chain: C,
411    kind: PublishErrorKind<E>,
412}
413
414impl<C, E> PublishError<C, E> {
415    /// Construct a publication failure that retains caller ownership.
416    pub const fn new(chain: C, kind: PublishErrorKind<E>) -> Self {
417        Self { chain, kind }
418    }
419
420    /// Borrow the unpublished chain.
421    pub const fn chain(&self) -> &C {
422        &self.chain
423    }
424
425    /// Borrow the failure classification.
426    pub const fn kind(&self) -> &PublishErrorKind<E> {
427        &self.kind
428    }
429
430    /// Recover the unpublished chain and failure classification.
431    pub fn into_parts(self) -> (C, PublishErrorKind<E>) {
432        (self.chain, self.kind)
433    }
434}
435
436/// Driver-owned chain returned by used-ring consumption.
437#[derive(Debug, PartialEq, Eq)]
438pub struct UsedChain<C> {
439    id: ChainId,
440    used: UsedLength,
441    chain: C,
442}
443
444impl<C> UsedChain<C> {
445    /// Construct a used-chain result.
446    pub const fn new(id: ChainId, used: UsedLength, chain: C) -> Self {
447        Self { id, used, chain }
448    }
449
450    /// Returned chain identity.
451    pub const fn id(&self) -> ChainId {
452        self.id
453    }
454
455    /// Exact number of bytes written by the device.
456    pub const fn used(&self) -> UsedLength {
457        self.used
458    }
459
460    /// Borrow the returned driver chain.
461    pub const fn chain(&self) -> &C {
462        &self.chain
463    }
464
465    /// Recover the identity, used length, and driver chain.
466    pub fn into_parts(self) -> (ChainId, UsedLength, C) {
467        (self.id, self.used, self.chain)
468    }
469}
470
471/// Published driver chain reclaimed during reset without a used entry.
472#[derive(Debug, PartialEq, Eq)]
473pub struct ReclaimedChain<C> {
474    id: ChainId,
475    chain: C,
476}
477
478impl<C> ReclaimedChain<C> {
479    /// Construct a reset-reclamation result.
480    pub const fn new(id: ChainId, chain: C) -> Self {
481        Self { id, chain }
482    }
483
484    /// Reclaimed chain identity.
485    pub const fn id(&self) -> ChainId {
486        self.id
487    }
488
489    /// Recover the identity and driver chain.
490    pub fn into_parts(self) -> (ChainId, C) {
491        (self.id, self.chain)
492    }
493}
494
495/// Driver-side descriptor publication and used-ring consumption.
496///
497/// A successful [`DriverQueue::publish`] transfers exclusive chain ownership to the queue until
498/// [`DriverQueue::pop_used`] or [`DriverQueue::reset`] returns it. Implementations must publish all
499/// descriptor contents and the available index before returning a notification decision. A
500/// successful used pop is an acquire boundary for response bytes written before device completion.
501pub trait DriverQueue: QueuePort {
502    /// Driver-owned command and descriptor resources.
503    type Chain;
504    /// Owned collection of chains reclaimed by reset.
505    type Reclaimed: IntoIterator<Item = ReclaimedChain<Self::Chain>>;
506    /// Concrete transport failure.
507    type Error;
508
509    /// Publish a complete chain or return it unchanged on pre-publication backpressure.
510    ///
511    /// This operation must not block or allocate. Success is a release boundary for descriptor and
512    /// request writes. Failure must not transfer ownership or expose a partial chain to the device.
513    fn publish(
514        &mut self,
515        chain: Self::Chain,
516    ) -> Result<PublishedChain, PublishError<Self::Chain, Self::Error>>;
517
518    /// Consume one used entry and recover the corresponding driver chain.
519    ///
520    /// This operation must not block or allocate. It is an acquire boundary for the used entry and
521    /// response bytes. Independent completions may be returned out of publication order.
522    fn pop_used(&mut self) -> Result<Option<UsedChain<Self::Chain>>, QueueError<Self::Error>>;
523
524    /// Disable used-ring notifications before draining completions.
525    ///
526    /// This operation must not block or allocate.
527    fn disable_used_notifications(&mut self) -> Result<(), QueueError<Self::Error>>;
528
529    /// Enable used-ring notifications and atomically recheck for missed completions.
530    ///
531    /// This operation must not block or allocate. `WorkPending` requires the caller to continue
532    /// draining rather than sleep.
533    fn enable_used_notifications(&mut self)
534    -> Result<NotificationRecheck, QueueError<Self::Error>>;
535
536    /// Invalidate the old epoch and recover every chain still owned by the queue.
537    ///
538    /// This operation must not block or allocate. It may move existing queue-owned storage into the
539    /// returned collection. `next_epoch` must be greater than the current epoch. No old used entry
540    /// or response write may become visible after success.
541    fn reset(&mut self, next_epoch: QueueEpoch)
542    -> Result<Self::Reclaimed, QueueError<Self::Error>>;
543}
544
545/// Device-side available-ring consumption and completion publication.
546///
547/// A successful [`DeviceQueue::pop_available`] acquires request and descriptor writes made before
548/// driver publication. [`DeviceQueue::complete`] consumes the chain, publishes response bytes and
549/// the used index with release ordering, and only then returns a notification decision.
550pub trait DeviceQueue: QueuePort {
551    /// Owned consumed-chain type.
552    type Chain: DeviceChain;
553    /// Concrete transport failure.
554    type Error;
555
556    /// Consume one available descriptor chain.
557    ///
558    /// This operation must not block or allocate. Chains are returned in available-ring order.
559    fn pop_available(&mut self) -> Result<Option<Self::Chain>, QueueError<Self::Error>>;
560
561    /// Publish one chain completion and consume its ownership token.
562    ///
563    /// This operation must not block or allocate. It must reject a stale epoch without writing
564    /// guest bytes, a used entry, or a notification. `used` must not exceed writable capacity.
565    fn complete(
566        &mut self,
567        chain: Self::Chain,
568        used: UsedLength,
569    ) -> Result<NotificationHint, QueueError<Self::Error>>;
570
571    /// Disable available-ring notifications before draining work.
572    ///
573    /// This operation must not block or allocate.
574    fn disable_available_notifications(&mut self) -> Result<(), QueueError<Self::Error>>;
575
576    /// Enable available-ring notifications and atomically recheck for missed work.
577    ///
578    /// This operation must not block or allocate. `WorkPending` requires the caller to continue
579    /// draining rather than sleep.
580    fn enable_available_notifications(
581        &mut self,
582    ) -> Result<NotificationRecheck, QueueError<Self::Error>>;
583
584    /// Invalidate all consumed chains from the old epoch and return the queue to its base state.
585    ///
586    /// This operation must not block or allocate. `next_epoch` must be greater than the current
587    /// epoch. No later completion from the old epoch may write guest memory or publish a used entry.
588    fn reset(&mut self, next_epoch: QueueEpoch) -> Result<(), QueueError<Self::Error>>;
589}
590
591#[cfg(test)]
592mod tests {
593    extern crate std;
594
595    use std::collections::VecDeque;
596    use std::sync::Arc;
597    use std::sync::atomic::{AtomicU64, Ordering};
598    use std::vec::Vec;
599
600    use super::*;
601
602    #[derive(Debug, PartialEq, Eq)]
603    struct TestPayload {
604        request: [u8; 4],
605        response: [u8; 4],
606    }
607
608    #[derive(Debug)]
609    struct TestDeviceChain {
610        id: ChainId,
611        current_epoch: Arc<AtomicU64>,
612        payload: TestPayload,
613    }
614
615    impl DeviceChain for TestDeviceChain {
616        type Request = [u8; 4];
617        type Response = [u8; 4];
618        type Error = ();
619
620        fn id(&self) -> ChainId {
621            self.id
622        }
623
624        fn io(
625            &mut self,
626        ) -> Result<ChainIo<'_, Self::Request, Self::Response>, ChainError<Self::Error>> {
627            let current = QueueEpoch::new(self.current_epoch.load(Ordering::Acquire)).unwrap();
628            if current != self.id.epoch() {
629                return Err(ChainError::ResetRace {
630                    chain: self.id.epoch(),
631                    current,
632                });
633            }
634            static REGIONS: [ChainRegion; 2] = [ChainRegion::readable(4), ChainRegion::writable(4)];
635            Ok(ChainIo::new(
636                &REGIONS,
637                &self.payload.request,
638                &mut self.payload.response,
639            ))
640        }
641    }
642
643    struct TestQueue {
644        state: QueueState,
645        current_epoch: Arc<AtomicU64>,
646        next_token: u64,
647        available: VecDeque<(ChainId, TestPayload)>,
648        used: VecDeque<UsedChain<TestPayload>>,
649    }
650
651    impl TestQueue {
652        fn new() -> Self {
653            let epoch = QueueEpoch::INITIAL;
654            Self {
655                state: QueueState::unconfigured(QueueSize::new(8).unwrap(), epoch),
656                current_epoch: Arc::new(AtomicU64::new(epoch.get())),
657                next_token: 0,
658                available: VecDeque::new(),
659                used: VecDeque::new(),
660            }
661        }
662
663        fn check_ready(&self) -> Result<(), QueueError<()>> {
664            if self.state.ready() {
665                Ok(())
666            } else {
667                Err(QueueError::NotReady)
668            }
669        }
670
671        fn advance_epoch(&mut self, next_epoch: QueueEpoch) -> Result<(), QueueError<()>> {
672            if next_epoch <= self.state.epoch() {
673                return Err(QueueError::InvalidConfiguration(
674                    QueueConfigError::NonIncreasingEpoch,
675                ));
676            }
677            self.state = QueueState::unconfigured(self.state.max_size(), next_epoch);
678            self.current_epoch
679                .store(next_epoch.get(), Ordering::Release);
680            Ok(())
681        }
682    }
683
684    impl QueuePort for TestQueue {
685        fn state(&self) -> QueueState {
686            self.state
687        }
688    }
689
690    impl QueueControl for TestQueue {
691        type Error = ();
692
693        fn configure(&mut self, size: QueueSize) -> Result<(), QueueError<Self::Error>> {
694            if self.state.ready() || size > self.state.max_size() {
695                return Err(QueueError::InvalidConfiguration(
696                    QueueConfigError::SizeExceedsMaximum,
697                ));
698            }
699            self.state =
700                QueueState::new(self.state.max_size(), Some(size), false, self.state.epoch())
701                    .unwrap();
702            Ok(())
703        }
704
705        fn set_ready(&mut self, ready: bool) -> Result<(), QueueError<Self::Error>> {
706            self.state = QueueState::new(
707                self.state.max_size(),
708                self.state.size(),
709                ready,
710                self.state.epoch(),
711            )
712            .map_err(QueueError::InvalidConfiguration)?;
713            Ok(())
714        }
715    }
716
717    impl DriverQueue for TestQueue {
718        type Chain = TestPayload;
719        type Reclaimed = Vec<ReclaimedChain<Self::Chain>>;
720        type Error = ();
721
722        fn publish(
723            &mut self,
724            chain: Self::Chain,
725        ) -> Result<PublishedChain, PublishError<Self::Chain, Self::Error>> {
726            if !self.state.ready() {
727                return Err(PublishError::new(chain, PublishErrorKind::NotReady));
728            }
729            if self.available.len() >= usize::from(self.state.size().unwrap().get()) {
730                return Err(PublishError::new(chain, PublishErrorKind::QueueFull));
731            }
732            let id = ChainId::new(self.state.epoch(), self.next_token);
733            self.next_token += 1;
734            self.available.push_back((id, chain));
735            Ok(PublishedChain::new(id, NotificationHint::Notify))
736        }
737
738        fn pop_used(&mut self) -> Result<Option<UsedChain<Self::Chain>>, QueueError<Self::Error>> {
739            self.check_ready()?;
740            Ok(self.used.pop_front())
741        }
742
743        fn disable_used_notifications(&mut self) -> Result<(), QueueError<Self::Error>> {
744            self.check_ready()
745        }
746
747        fn enable_used_notifications(
748            &mut self,
749        ) -> Result<NotificationRecheck, QueueError<Self::Error>> {
750            self.check_ready()?;
751            Ok(if self.used.is_empty() {
752                NotificationRecheck::Idle
753            } else {
754                NotificationRecheck::WorkPending
755            })
756        }
757
758        fn reset(
759            &mut self,
760            next_epoch: QueueEpoch,
761        ) -> Result<Self::Reclaimed, QueueError<Self::Error>> {
762            self.advance_epoch(next_epoch)?;
763            let reclaimed = self
764                .available
765                .drain(..)
766                .map(|(id, chain)| ReclaimedChain::new(id, chain))
767                .collect();
768            self.used.clear();
769            Ok(reclaimed)
770        }
771    }
772
773    impl DeviceQueue for TestQueue {
774        type Chain = TestDeviceChain;
775        type Error = ();
776
777        fn pop_available(&mut self) -> Result<Option<Self::Chain>, QueueError<Self::Error>> {
778            self.check_ready()?;
779            Ok(self
780                .available
781                .pop_front()
782                .map(|(id, payload)| TestDeviceChain {
783                    id,
784                    current_epoch: Arc::clone(&self.current_epoch),
785                    payload,
786                }))
787        }
788
789        fn complete(
790            &mut self,
791            chain: Self::Chain,
792            used: UsedLength,
793        ) -> Result<NotificationHint, QueueError<Self::Error>> {
794            if chain.id.epoch() != self.state.epoch() {
795                return Err(QueueError::ResetRace {
796                    operation: chain.id.epoch(),
797                    current: self.state.epoch(),
798                });
799            }
800            if u64::from(used.get()) > chain.payload.response.len() as u64 {
801                return Err(QueueError::UsedLengthExceeded {
802                    used,
803                    capacity: chain.payload.response.len() as u64,
804                });
805            }
806            self.used
807                .push_back(UsedChain::new(chain.id, used, chain.payload));
808            Ok(NotificationHint::Notify)
809        }
810
811        fn disable_available_notifications(&mut self) -> Result<(), QueueError<Self::Error>> {
812            self.check_ready()
813        }
814
815        fn enable_available_notifications(
816            &mut self,
817        ) -> Result<NotificationRecheck, QueueError<Self::Error>> {
818            self.check_ready()?;
819            Ok(if self.available.is_empty() {
820                NotificationRecheck::Idle
821            } else {
822                NotificationRecheck::WorkPending
823            })
824        }
825
826        fn reset(&mut self, next_epoch: QueueEpoch) -> Result<(), QueueError<Self::Error>> {
827            self.advance_epoch(next_epoch)?;
828            self.available.clear();
829            self.used.clear();
830            Ok(())
831        }
832    }
833
834    #[test]
835    fn queue_sizes_and_states_are_validated() {
836        assert_eq!(QueueSize::new(0), None);
837        assert_eq!(QueueSize::new(3), None);
838        assert_eq!(QueueSize::new(MAX_SPLIT_QUEUE_SIZE + 1), None);
839
840        let max = QueueSize::new(8).unwrap();
841        let too_large = QueueSize::new(16).unwrap();
842        assert_eq!(
843            QueueState::new(max, Some(too_large), false, QueueEpoch::INITIAL),
844            Err(QueueConfigError::SizeExceedsMaximum)
845        );
846        assert_eq!(
847            QueueState::new(max, None, true, QueueEpoch::INITIAL),
848            Err(QueueConfigError::ReadyWithoutSize)
849        );
850    }
851
852    #[test]
853    fn publication_transfers_ownership_until_used() {
854        let mut queue = TestQueue::new();
855        QueueControl::configure(&mut queue, QueueSize::new(2).unwrap()).unwrap();
856        QueueControl::set_ready(&mut queue, true).unwrap();
857
858        let published = DriverQueue::publish(
859            &mut queue,
860            TestPayload {
861                request: *b"ping",
862                response: [0; 4],
863            },
864        )
865        .unwrap();
866        assert_eq!(published.notification(), NotificationHint::Notify);
867
868        let mut chain = DeviceQueue::pop_available(&mut queue).unwrap().unwrap();
869        assert_eq!(chain.id(), published.id());
870        let (regions, request, response) = chain.io().unwrap().into_parts();
871        assert_eq!(regions.len(), 2);
872        assert_eq!(request, b"ping");
873        response.copy_from_slice(b"pong");
874        DeviceQueue::complete(&mut queue, chain, UsedLength::new(4)).unwrap();
875
876        let used = DriverQueue::pop_used(&mut queue).unwrap().unwrap();
877        assert_eq!(used.id(), published.id());
878        assert_eq!(used.used(), UsedLength::new(4));
879        assert_eq!(used.chain().response, *b"pong");
880    }
881
882    #[test]
883    fn publication_backpressure_returns_the_chain() {
884        let mut queue = TestQueue::new();
885        let payload = TestPayload {
886            request: *b"ping",
887            response: [0; 4],
888        };
889        let error = DriverQueue::publish(&mut queue, payload).unwrap_err();
890        let (payload, kind) = error.into_parts();
891        assert_eq!(kind, PublishErrorKind::NotReady);
892        assert_eq!(payload.request, *b"ping");
893    }
894
895    #[test]
896    fn reset_epoch_rejects_late_completion() {
897        let mut queue = TestQueue::new();
898        QueueControl::configure(&mut queue, QueueSize::new(2).unwrap()).unwrap();
899        QueueControl::set_ready(&mut queue, true).unwrap();
900        DriverQueue::publish(
901            &mut queue,
902            TestPayload {
903                request: *b"ping",
904                response: [0; 4],
905            },
906        )
907        .unwrap();
908        let mut chain = DeviceQueue::pop_available(&mut queue).unwrap().unwrap();
909        let old_epoch = chain.id().epoch();
910        let next_epoch = old_epoch.checked_next().unwrap();
911
912        DeviceQueue::reset(&mut queue, next_epoch).unwrap();
913        assert!(matches!(
914            chain.io(),
915            Err(ChainError::ResetRace { chain, current })
916                if chain == old_epoch && current == next_epoch
917        ));
918        assert_eq!(
919            DeviceQueue::complete(&mut queue, chain, UsedLength::new(0)),
920            Err(QueueError::ResetRace {
921                operation: old_epoch,
922                current: next_epoch,
923            })
924        );
925        assert!(queue.used.is_empty());
926    }
927}
virtio_accel_transport/queue.rs

virtio_accel_transport/
queue.rs
