1use std::any::{Any, TypeId};
23use std::cell::{Cell, RefCell};
24use std::collections::HashMap;
25use std::fmt;
26use std::future::Future;
27use std::pin::Pin;
28use std::rc::Rc;
29use std::task::{Context, Poll, Waker};
30
31use rustdv_sim::log::Logger;
32use rustdv_sim::queue::Queue;
33use rustdv_sim::sync::Event;
34use rustdv_sim::{Rng, RustdvPath};
35
36use crate::component::{Component, ComponentNode};
37use crate::port::{PortName, PortOwner, bind_or_panic};
38
39#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
47pub struct TxnId(pub u64);
48
49impl fmt::Display for TxnId {
50 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51 write!(f, "#{}", self.0)
52 }
53}
54
55pub struct SeqItem<REQ> {
62 id: TxnId,
63 payload: REQ,
64}
65
66impl<REQ> SeqItem<REQ> {
67 pub fn txn_id(&self) -> TxnId {
68 self.id
69 }
70 pub fn payload(&self) -> &REQ {
71 &self.payload
72 }
73 pub fn payload_mut(&mut self) -> &mut REQ {
74 &mut self.payload
75 }
76 pub fn into_payload(self) -> REQ {
77 self.payload
78 }
79}
80
81#[derive(Debug, Clone)]
82pub struct SeqError(pub String);
83
84impl fmt::Display for SeqError {
85 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
86 write!(f, "sequence error: {}", self.0)
87 }
88}
89impl std::error::Error for SeqError {}
90impl From<&str> for SeqError {
91 fn from(s: &str) -> SeqError {
92 SeqError(s.to_string())
93 }
94}
95impl From<String> for SeqError {
96 fn from(s: String) -> SeqError {
97 SeqError(s)
98 }
99}
100impl From<crate::config::ConfigError> for SeqError {
101 fn from(e: crate::config::ConfigError) -> SeqError {
102 SeqError(e.to_string())
103 }
104}
105
106struct ItemSlot<REQ> {
113 id: TxnId,
114 granted: Event,
115 ready: Event,
116 done: Event,
117 payload: RefCell<Option<REQ>>,
118}
119
120struct RespInner<RSP> {
121 items: RefCell<Vec<(TxnId, RSP)>>,
122 waiters: RefCell<Vec<Waker>>,
123}
124
125pub struct ResponseQueue<RSP> {
127 inner: Rc<RespInner<RSP>>,
128}
129
130impl<RSP> Clone for ResponseQueue<RSP> {
131 fn clone(&self) -> Self {
132 ResponseQueue {
133 inner: self.inner.clone(),
134 }
135 }
136}
137
138impl<RSP> ResponseQueue<RSP> {
139 fn new() -> ResponseQueue<RSP> {
140 ResponseQueue {
141 inner: Rc::new(RespInner {
142 items: RefCell::new(Vec::new()),
143 waiters: RefCell::new(Vec::new()),
144 }),
145 }
146 }
147
148 fn push(&self, id: TxnId, rsp: RSP) {
149 self.inner.items.borrow_mut().push((id, rsp));
150 for w in self.inner.waiters.borrow_mut().drain(..) {
151 w.wake();
152 }
153 }
154
155 pub fn get_response(&self, txn_id: Option<TxnId>) -> GetResponse<RSP> {
158 GetResponse {
159 inner: self.inner.clone(),
160 txn_id,
161 }
162 }
163
164 pub fn try_get_response(&self, txn_id: Option<TxnId>) -> Option<RSP> {
172 let mut items = self.inner.items.borrow_mut();
173 let idx = match txn_id {
174 None => (!items.is_empty()).then_some(0),
175 Some(id) => items.iter().position(|(i, _)| *i == id),
176 };
177 idx.map(|i| items.remove(i).1)
178 }
179}
180
181pub struct GetResponse<RSP> {
182 inner: Rc<RespInner<RSP>>,
183 txn_id: Option<TxnId>,
184}
185
186impl<RSP> Future for GetResponse<RSP> {
187 type Output = RSP;
188 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<RSP> {
189 let mut items = self.inner.items.borrow_mut();
190 let idx = match self.txn_id {
191 None => (!items.is_empty()).then_some(0),
192 Some(id) => items.iter().position(|(i, _)| *i == id),
193 };
194 match idx {
195 Some(i) => Poll::Ready(items.remove(i).1),
196 None => {
197 drop(items);
198 self.inner.waiters.borrow_mut().push(cx.waker().clone());
199 Poll::Pending
200 }
201 }
202 }
203}
204
205struct SeqrInner<REQ: 'static, RSP: 'static> {
206 queue: Queue<Rc<ItemSlot<REQ>>>,
210 next_id: Cell<u64>,
211 responses: ResponseQueue<RSP>,
212 current: RefCell<Option<Rc<ItemSlot<REQ>>>>,
215}
216
217pub trait SeqItemIf<REQ: 'static, RSP: 'static>: 'static {
224 fn get_next_item(&self) -> Pin<Box<dyn Future<Output = SeqItem<REQ>> + '_>>;
225 fn try_next_item(&self) -> Option<SeqItem<REQ>>;
230 fn item_done(&self, rsp: Option<RSP>);
231 fn put_response(&self, id: TxnId, rsp: RSP);
235}
236
237impl<REQ: 'static, RSP: 'static> SeqItemIf<REQ, RSP> for SeqrInner<REQ, RSP> {
238 fn get_next_item(&self) -> Pin<Box<dyn Future<Output = SeqItem<REQ>> + '_>> {
239 Box::pin(async move {
240 assert!(
241 self.current.borrow().is_none(),
242 "get_next_item called twice without item_done"
243 );
244 let slot = self.queue.get().await;
245 slot.granted.set();
246 slot.ready.wait().await;
247 let payload = slot
248 .payload
249 .borrow_mut()
250 .take()
251 .expect("item ready but payload missing (rustdv bug)");
252 let item = SeqItem {
253 id: slot.id,
254 payload,
255 };
256 *self.current.borrow_mut() = Some(slot);
257 item
258 })
259 }
260
261 fn try_next_item(&self) -> Option<SeqItem<REQ>> {
262 assert!(
263 self.current.borrow().is_none(),
264 "try_next_item called without item_done"
265 );
266 let slot = self.queue.try_get()?;
270 slot.granted.set();
271 let taken = slot.payload.borrow_mut().take();
272 let payload = match taken {
273 Some(p) => p,
274 None => {
275 let _ = self.queue.try_put(slot);
277 return None;
278 }
279 };
280 let item = SeqItem {
281 id: slot.id,
282 payload,
283 };
284 *self.current.borrow_mut() = Some(slot);
285 Some(item)
286 }
287
288 fn item_done(&self, rsp: Option<RSP>) {
289 let slot = self
290 .current
291 .borrow_mut()
292 .take()
293 .expect("item_done without get_next_item");
294 if let Some(r) = rsp {
295 self.responses.push(slot.id, r);
296 }
297 slot.done.set();
298 }
299
300 fn put_response(&self, id: TxnId, rsp: RSP) {
301 self.responses.push(id, rsp);
302 }
303}
304
305pub type SeqItemPort<REQ, RSP = REQ> = crate::port::Port<dyn SeqItemIf<REQ, RSP>>;
307
308pub struct SeqItemExport<REQ: 'static, RSP: 'static> {
310 iface: Rc<dyn SeqItemIf<REQ, RSP>>,
311}
312
313impl<REQ: 'static, RSP: 'static> SeqItemExport<REQ, RSP> {
314 pub fn connect(&self, owner: &dyn PortOwner, name: PortName<dyn SeqItemIf<REQ, RSP>>) {
315 bind_or_panic(owner, name, self.iface.clone());
316 }
317
318 pub fn connect_port(&self, port: &SeqItemPort<REQ, RSP>) {
321 port.bind_iface(self.iface.clone());
322 }
323}
324
325pub struct Sequencer<REQ: 'static, RSP: 'static = REQ> {
335 inner: Rc<SeqrInner<REQ, RSP>>,
336}
337
338impl<REQ, RSP> Clone for Sequencer<REQ, RSP> {
339 fn clone(&self) -> Self {
340 Sequencer {
341 inner: self.inner.clone(),
342 }
343 }
344}
345
346impl<REQ: 'static, RSP: 'static> fmt::Debug for Sequencer<REQ, RSP> {
347 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
350 f.write_str("Sequencer")
351 }
352}
353
354impl<REQ: 'static, RSP: 'static> Default for Sequencer<REQ, RSP> {
355 fn default() -> Self {
356 Self::new()
357 }
358}
359
360impl<REQ: 'static, RSP: 'static> Sequencer<REQ, RSP> {
361 pub fn new() -> Sequencer<REQ, RSP> {
362 Sequencer {
363 inner: Rc::new(SeqrInner {
364 queue: Queue::unbounded(),
365 next_id: Cell::new(1),
366 responses: ResponseQueue::new(),
367 current: RefCell::new(None),
368 }),
369 }
370 }
371
372 pub fn handle(&self) -> Sequencer<REQ, RSP> {
374 self.clone()
375 }
376
377 pub fn seq_item_export(&self) -> SeqItemExport<REQ, RSP> {
379 SeqItemExport {
380 iface: self.inner.clone(),
381 }
382 }
383}
384
385impl<REQ: 'static, RSP: 'static> Component for Sequencer<REQ, RSP> {}
388
389impl<REQ: 'static, RSP: 'static> ComponentNode for Sequencer<REQ, RSP> {
390 fn node_name(&self) -> &'static str {
391 "Sequencer"
392 }
393 fn children_mut(&mut self) -> Vec<(String, &mut (dyn ComponentNode + 'static))> {
394 Vec::new()
395 }
396}
397
398pub struct SeqCtx<REQ: 'static, RSP: 'static = REQ> {
406 inner: Option<Rc<SeqrInner<REQ, RSP>>>,
407 current: Option<Rc<ItemSlot<REQ>>>,
408 name: &'static str,
409 logger: Logger,
410 seed: u64,
411}
412
413impl<REQ: 'static, RSP: 'static> SeqCtx<REQ, RSP> {
414 fn new(inner: Option<Rc<SeqrInner<REQ, RSP>>>, name: &'static str, seed: u64) -> Self {
415 SeqCtx {
416 inner,
417 current: None,
418 name,
419 logger: Logger::at(RustdvPath::root(name)),
420 seed,
421 }
422 }
423
424 pub fn name(&self) -> &'static str {
427 self.name
428 }
429
430 pub fn rng(&self) -> Rng {
433 Rng::new(self.seed)
434 }
435
436 pub fn info(&self, msg: &str) {
437 self.logger.info(msg);
438 }
439 pub fn warning(&self, msg: &str) {
440 self.logger.warning(msg);
441 }
442 pub fn error(&self, msg: &str) {
443 self.logger.error(msg);
444 }
445
446 fn seqr(&self) -> Result<&Rc<SeqrInner<REQ, RSP>>, SeqError> {
447 self.inner.as_ref().ok_or_else(|| {
448 SeqError(format!(
449 "{}: start_item in a virtual sequence — it was started without a sequencer",
450 self.name
451 ))
452 })
453 }
454
455 pub async fn start_item(&mut self, _item: &mut REQ) -> Result<(), SeqError> {
458 if self.current.is_some() {
459 return Err(SeqError(
460 "start_item called twice without finish_item".into(),
461 ));
462 }
463 let inner = self.seqr()?.clone();
464 let id = TxnId(inner.next_id.get());
465 inner.next_id.set(id.0 + 1);
466 let slot = Rc::new(ItemSlot {
467 id,
468 granted: Event::new(),
469 ready: Event::new(),
470 done: Event::new(),
471 payload: RefCell::new(None),
472 });
473 self.current = Some(slot.clone());
474 inner.queue.put(slot.clone()).await;
475 slot.granted.wait().await;
476 Ok(())
477 }
478
479 pub async fn finish_item(&mut self, item: REQ) -> Result<TxnId, SeqError> {
482 let slot = self
483 .current
484 .take()
485 .ok_or_else(|| SeqError("finish_item without start_item".into()))?;
486 *slot.payload.borrow_mut() = Some(item);
487 slot.ready.set();
488 slot.done.wait().await;
489 Ok(slot.id)
490 }
491
492 pub fn try_get_response(&mut self, txn_id: Option<TxnId>) -> Option<RSP> {
499 let responses = self.seqr().ok()?.responses.clone();
500 responses.try_get_response(txn_id)
501 }
502
503 pub async fn get_response(&mut self, txn_id: Option<TxnId>) -> RSP {
506 let responses = self
507 .seqr()
508 .expect("get_response in a virtual sequence")
509 .responses
510 .clone();
511 responses.get_response(txn_id).await
512 }
513}
514
515pub trait Sequence: Sized + 'static {
526 type Req: 'static;
527 type Rsp: 'static;
528
529 fn body(
530 &mut self,
531 ctx: &mut SeqCtx<Self::Req, Self::Rsp>,
532 ) -> impl Future<Output = Result<(), SeqError>>;
533
534 fn seq_name(&self) -> &'static str {
539 let full = std::any::type_name::<Self>();
540 full.rsplit("::").next().unwrap_or(full)
541 }
542
543 fn start(
545 &mut self,
546 seqr: &Sequencer<Self::Req, Self::Rsp>,
547 ) -> impl Future<Output = Result<(), SeqError>> {
548 let inner = seqr.inner.clone();
549 let name = self.seq_name();
550 async move {
551 let mut ctx = SeqCtx::new(Some(inner), name, current_seed());
552 self.body(&mut ctx).await
553 }
554 }
555
556 fn start_virtual(&mut self) -> impl Future<Output = Result<(), SeqError>> {
560 let name = self.seq_name();
561 async move {
562 let mut ctx = SeqCtx::new(None, name, current_seed());
563 self.body(&mut ctx).await
564 }
565 }
566}
567
568thread_local! {
569 static SEED: Cell<u64> = const { Cell::new(1) };
570}
571
572pub fn set_sequence_seed(seed: u64) {
574 SEED.with(|s| s.set(seed));
575}
576
577fn current_seed() -> u64 {
578 SEED.with(|s| s.get())
579}
580
581pub trait DynSequence<REQ: 'static, RSP: 'static> {
588 fn dyn_body<'a>(
589 &'a mut self,
590 ctx: &'a mut SeqCtx<REQ, RSP>,
591 ) -> Pin<Box<dyn Future<Output = Result<(), SeqError>> + 'a>>;
592 fn dyn_name(&self) -> &'static str;
593}
594
595impl<S: Sequence> DynSequence<S::Req, S::Rsp> for S {
596 fn dyn_body<'a>(
597 &'a mut self,
598 ctx: &'a mut SeqCtx<S::Req, S::Rsp>,
599 ) -> Pin<Box<dyn Future<Output = Result<(), SeqError>> + 'a>> {
600 Box::pin(self.body(ctx))
601 }
602 fn dyn_name(&self) -> &'static str {
603 self.seq_name()
604 }
605}
606
607pub struct RustdvSeq<REQ: 'static, RSP: 'static = REQ> {
611 inner: Option<Box<dyn DynSequence<REQ, RSP>>>,
612}
613
614impl<REQ: 'static, RSP: 'static> Default for RustdvSeq<REQ, RSP> {
615 fn default() -> Self {
616 RustdvSeq { inner: None }
617 }
618}
619
620impl<REQ: 'static, RSP: 'static> RustdvSeq<REQ, RSP> {
621 pub fn new(seq: Box<dyn DynSequence<REQ, RSP>>) -> Self {
622 RustdvSeq { inner: Some(seq) }
623 }
624
625 pub fn name(&self) -> &'static str {
627 self.inner
628 .as_ref()
629 .map(|s| s.dyn_name())
630 .unwrap_or("<empty>")
631 }
632
633 fn get(&mut self) -> Result<&mut Box<dyn DynSequence<REQ, RSP>>, SeqError> {
634 self.inner
635 .as_mut()
636 .ok_or_else(|| SeqError("an empty sequence slot".into()))
637 }
638
639 pub async fn start(&mut self, seqr: &Sequencer<REQ, RSP>) -> Result<(), SeqError> {
640 let inner = seqr.inner.clone();
641 let seq = self.get()?;
642 let mut ctx = SeqCtx::new(Some(inner), seq.dyn_name(), current_seed());
643 seq.dyn_body(&mut ctx).await
644 }
645
646 pub async fn start_virtual(&mut self) -> Result<(), SeqError> {
647 let seq = self.get()?;
648 let mut ctx = SeqCtx::new(None, seq.dyn_name(), current_seed());
649 seq.dyn_body(&mut ctx).await
650 }
651}
652
653type SeqOverrides = HashMap<TypeId, (TypeId, fn() -> Box<dyn Any>)>;
658
659thread_local! {
660 static SEQ_OVERRIDES: RefCell<SeqOverrides> = RefCell::new(HashMap::new());
661}
662
663pub fn clear_seq_overrides() {
665 SEQ_OVERRIDES.with(|o| o.borrow_mut().clear());
666}
667
668pub fn set_seq_override<From, To>()
675where
676 From: Sequence,
677 To: Sequence<Req = From::Req, Rsp = From::Rsp> + Default,
678{
679 fn maker<To: Sequence + Default>() -> Box<dyn Any> {
680 let boxed: Box<dyn DynSequence<To::Req, To::Rsp>> = Box::new(To::default());
681 Box::new(boxed)
682 }
683 SEQ_OVERRIDES.with(|o| {
684 o.borrow_mut()
685 .insert(TypeId::of::<From>(), (TypeId::of::<To>(), maker::<To>));
686 });
687}
688
689pub fn create_seq<S>() -> RustdvSeq<S::Req, S::Rsp>
691where
692 S: Sequence + Default,
693{
694 let over = SEQ_OVERRIDES.with(|o| o.borrow().get(&TypeId::of::<S>()).map(|(_, m)| *m));
695 match over {
696 Some(make) => {
697 let any = make();
698 let boxed = any
699 .downcast::<Box<dyn DynSequence<S::Req, S::Rsp>>>()
700 .expect("sequence override built the wrong request/response types");
701 RustdvSeq::new(*boxed)
702 }
703 None => RustdvSeq::new(Box::new(S::default())),
704 }
705}
706
707#[cfg(test)]
716mod tests {
717 use super::*;
718 use rustdv_sim::executor;
719 use rustdv_sim::testing::{assert_pending, block_on};
720
721 #[derive(Clone, Debug, PartialEq, Eq, Default)]
722 struct Cmd {
723 a: u8,
724 tag: &'static str,
725 }
726
727 #[derive(Clone, Debug, PartialEq, Eq, Default)]
728 struct Rsp {
729 v: u8,
730 }
731
732 fn spawn_one_shot_driver(seqr: &Sequencer<Cmd, Rsp>, answer: u8) {
734 let port = seqr.seq_item_export();
735 let inner = seqr.inner.clone();
736 let _ = port; executor::spawn(async move {
738 let item = SeqItemIf::get_next_item(&*inner).await;
739 let v = item.payload().a.wrapping_add(answer);
740 SeqItemIf::item_done(&*inner, Some(Rsp { v }));
741 });
742 }
743
744 #[test]
745 fn start_item_blocks_until_the_driver_asks() {
746 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
747 let inner = seqr.inner.clone();
748 assert_pending(async move {
750 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
751 let mut cmd = Cmd::default();
752 ctx.start_item(&mut cmd).await.unwrap();
753 });
754 }
755
756 #[test]
760 fn the_driver_sees_what_was_written_after_the_grant() {
761 block_on(async {
762 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
763 let inner = seqr.inner.clone();
764 let seen = Rc::new(RefCell::new(None));
765 let seen2 = seen.clone();
766 let d = inner.clone();
767 executor::spawn(async move {
768 let item = SeqItemIf::get_next_item(&*d).await;
769 *seen2.borrow_mut() = Some(item.payload().clone());
770 SeqItemIf::item_done(&*d, None);
771 });
772
773 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
774 let mut cmd = Cmd {
775 a: 0,
776 tag: "before",
777 };
778 ctx.start_item(&mut cmd).await.unwrap();
779 cmd.a = 42;
781 cmd.tag = "after the grant";
782 ctx.finish_item(cmd).await.unwrap();
783
784 let got = seen.borrow().clone().expect("the driver got an item");
785 assert_eq!(got.a, 42, "the late value reached the driver");
786 assert_eq!(got.tag, "after the grant");
787 });
788 }
789
790 #[test]
791 fn finish_item_returns_the_ticket_and_waits_for_item_done() {
792 block_on(async {
793 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
794 spawn_one_shot_driver(&seqr, 1);
795 let mut ctx = SeqCtx::new(Some(seqr.inner.clone()), "T", 1);
796 let mut cmd = Cmd { a: 10, tag: "x" };
797 ctx.start_item(&mut cmd).await.unwrap();
798 let ticket = ctx.finish_item(cmd).await.unwrap();
799 assert_eq!(ticket, TxnId(1), "tickets start at 1");
800 });
801 }
802
803 #[test]
804 fn tickets_are_unique_and_ascending() {
805 block_on(async {
806 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
807 let inner = seqr.inner.clone();
808 let d = inner.clone();
809 executor::spawn(async move {
810 for _ in 0..3 {
811 let _item = SeqItemIf::get_next_item(&*d).await;
812 SeqItemIf::item_done(&*d, None);
813 }
814 });
815 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
816 let mut tickets = Vec::new();
817 for a in 0..3u8 {
818 let mut cmd = Cmd { a, tag: "" };
819 ctx.start_item(&mut cmd).await.unwrap();
820 tickets.push(ctx.finish_item(cmd).await.unwrap());
821 }
822 assert_eq!(tickets, vec![TxnId(1), TxnId(2), TxnId(3)]);
823 });
824 }
825
826 #[test]
829 #[should_panic(expected = "get_next_item called twice without item_done")]
830 fn two_get_next_items_without_item_done_is_a_bug() {
831 block_on(async {
832 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
833 let inner = seqr.inner.clone();
834 let s = inner.clone();
836 executor::spawn(async move {
837 let mut ctx = SeqCtx::new(Some(s), "T", 1);
838 let mut cmd = Cmd::default();
839 ctx.start_item(&mut cmd).await.unwrap();
840 ctx.finish_item(cmd).await.unwrap();
841 });
842 let _a = SeqItemIf::get_next_item(&*inner).await;
846 let _b = SeqItemIf::get_next_item(&*inner).await;
847 });
848 }
849
850 #[test]
851 fn start_item_twice_without_finish_is_an_error() {
852 block_on(async {
853 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
854 let inner = seqr.inner.clone();
855 let d = inner.clone();
856 executor::spawn(async move {
857 let _ = SeqItemIf::get_next_item(&*d).await;
858 });
859 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
860 let mut a = Cmd::default();
861 ctx.start_item(&mut a).await.unwrap();
862 let mut b = Cmd::default();
863 let err = ctx.start_item(&mut b).await;
864 assert!(err.is_err(), "a second start_item without finish_item");
865 });
866 }
867
868 #[test]
870 fn try_next_item_is_none_on_an_empty_sequencer() {
871 block_on(async {
872 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
873 assert!(SeqItemIf::try_next_item(&*seqr.inner).is_none());
874 });
875 }
876
877 #[test]
878 fn try_next_item_takes_a_waiting_item() {
879 block_on(async {
880 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
881 let inner = seqr.inner.clone();
882 let s = inner.clone();
883 executor::spawn(async move {
884 let mut ctx = SeqCtx::new(Some(s), "T", 1);
885 let mut cmd = Cmd { a: 5, tag: "" };
886 ctx.start_item(&mut cmd).await.unwrap();
887 ctx.finish_item(cmd).await.unwrap();
888 });
889 let mut got = None;
892 for _ in 0..8 {
893 executor::current().run_until_idle();
894 if let Some(item) = SeqItemIf::try_next_item(&*inner) {
895 got = Some(item.payload().a);
896 SeqItemIf::item_done(&*inner, None);
897 break;
898 }
899 }
900 assert_eq!(got, Some(5));
901 });
902 }
903
904 #[test]
905 fn item_done_with_a_response_reaches_get_response() {
906 block_on(async {
907 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
908 spawn_one_shot_driver(&seqr, 100);
909 let mut ctx = SeqCtx::new(Some(seqr.inner.clone()), "T", 1);
910 let mut cmd = Cmd { a: 1, tag: "" };
911 ctx.start_item(&mut cmd).await.unwrap();
912 let ticket = ctx.finish_item(cmd).await.unwrap();
913 let rsp = ctx.get_response(Some(ticket)).await;
914 assert_eq!(rsp.v, 101);
915 });
916 }
917
918 #[test]
921 fn put_response_answers_after_item_done() {
922 block_on(async {
923 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
924 let inner = seqr.inner.clone();
925 let d = inner.clone();
926 executor::spawn(async move {
927 let item = SeqItemIf::get_next_item(&*d).await;
928 let id = item.txn_id();
929 SeqItemIf::item_done(&*d, None); SeqItemIf::put_response(&*d, id, Rsp { v: 77 });
931 });
932 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
933 let mut cmd = Cmd::default();
934 ctx.start_item(&mut cmd).await.unwrap();
935 let ticket = ctx.finish_item(cmd).await.unwrap();
936 assert_eq!(ctx.get_response(Some(ticket)).await.v, 77);
937 });
938 }
939
940 #[test]
943 fn get_response_picks_its_ticket_out_of_order() {
944 block_on(async {
945 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
946 let inner = seqr.inner.clone();
947 inner.responses.push(TxnId(2), Rsp { v: 22 });
949 inner.responses.push(TxnId(1), Rsp { v: 11 });
950 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
951 assert_eq!(ctx.get_response(Some(TxnId(1))).await.v, 11);
952 assert_eq!(ctx.get_response(Some(TxnId(2))).await.v, 22);
953 });
954 }
955
956 #[test]
957 fn get_response_none_takes_the_oldest() {
958 block_on(async {
959 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
960 let inner = seqr.inner.clone();
961 inner.responses.push(TxnId(5), Rsp { v: 50 });
962 inner.responses.push(TxnId(6), Rsp { v: 60 });
963 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
964 assert_eq!(ctx.get_response(None).await.v, 50, "oldest first");
965 assert_eq!(ctx.get_response(None).await.v, 60);
966 });
967 }
968
969 #[test]
970 fn try_get_response_does_not_wait() {
971 block_on(async {
972 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
973 let inner = seqr.inner.clone();
974 let mut ctx = SeqCtx::new(Some(inner.clone()), "T", 1);
975 assert!(
976 ctx.try_get_response(Some(TxnId(1))).is_none(),
977 "nothing yet"
978 );
979 inner.responses.push(TxnId(1), Rsp { v: 9 });
980 assert_eq!(ctx.try_get_response(Some(TxnId(1))).unwrap().v, 9);
981 assert!(
982 ctx.try_get_response(Some(TxnId(1))).is_none(),
983 "and it was taken"
984 );
985 });
986 }
987
988 #[test]
989 fn a_response_that_never_comes_waits_forever() {
990 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
991 let inner = seqr.inner.clone();
992 assert_pending(async move {
993 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
994 ctx.get_response(Some(TxnId(1043))).await
995 });
996 }
997
998 #[test]
1001 fn start_item_in_a_virtual_sequence_is_an_error() {
1002 block_on(async {
1003 let mut ctx: SeqCtx<Cmd, Rsp> = SeqCtx::new(None, "MyVirtualSeq", 1);
1004 let mut cmd = Cmd::default();
1005 match ctx.start_item(&mut cmd).await {
1006 Err(SeqError(msg)) => {
1007 assert!(msg.contains("virtual"), "the error explains: {msg}");
1008 assert!(
1009 msg.contains("MyVirtualSeq"),
1010 "and names the sequence: {msg}"
1011 );
1012 }
1013 Ok(()) => panic!("start_item should fail without a sequencer"),
1014 }
1015 });
1016 }
1017
1018 #[test]
1020 fn two_sequences_interleave_one_item_each() {
1021 block_on(async {
1022 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
1023 let inner = seqr.inner.clone();
1024 let order = Rc::new(RefCell::new(Vec::new()));
1025
1026 for tag in ["A", "B"] {
1027 let s = inner.clone();
1028 executor::spawn(async move {
1029 let mut ctx = SeqCtx::new(Some(s), tag, 1);
1030 for a in 0..2u8 {
1031 let mut cmd = Cmd { a, tag };
1032 ctx.start_item(&mut cmd).await.unwrap();
1033 ctx.finish_item(cmd).await.unwrap();
1034 }
1035 });
1036 }
1037
1038 let d = inner.clone();
1039 let seen = order.clone();
1040 executor::spawn(async move {
1041 for _ in 0..4 {
1042 let item = SeqItemIf::get_next_item(&*d).await;
1043 seen.borrow_mut().push(item.payload().tag);
1044 SeqItemIf::item_done(&*d, None);
1045 }
1046 });
1047
1048 for _ in 0..64 {
1049 executor::current().run_until_idle();
1050 }
1051 let got = order.borrow().clone();
1052 assert_eq!(got.len(), 4, "all four items were driven");
1053 assert_eq!(got, vec!["A", "B", "A", "B"], "one item each, in turn");
1054 });
1055 }
1056
1057 #[test]
1059 fn a_sequence_name_defaults_to_its_type() {
1060 #[derive(Default)]
1061 struct MyFancySeq;
1062 impl Sequence for MyFancySeq {
1063 type Req = Cmd;
1064 type Rsp = Rsp;
1065 async fn body(&mut self, _ctx: &mut SeqCtx<Cmd, Rsp>) -> Result<(), SeqError> {
1066 Ok(())
1067 }
1068 }
1069 assert_eq!(MyFancySeq.seq_name(), "MyFancySeq");
1070 }
1071
1072 #[test]
1073 fn a_virtual_sequence_runs_its_body() {
1074 #[derive(Default)]
1075 struct VSeq {
1076 ran: bool,
1077 }
1078 impl Sequence for VSeq {
1079 type Req = Cmd;
1080 type Rsp = Rsp;
1081 async fn body(&mut self, _ctx: &mut SeqCtx<Cmd, Rsp>) -> Result<(), SeqError> {
1082 self.ran = true;
1083 Ok(())
1084 }
1085 }
1086 block_on(async {
1087 let mut s = VSeq::default();
1088 s.start_virtual().await.unwrap();
1089 assert!(s.ran);
1090 });
1091 }
1092}