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::queue::Queue;
32use rustdv_sim::sync::Event;
33use rustdv_sim::log::Logger;
34use rustdv_sim::{Rng, RustdvPath};
35
36use crate::component::{Component, ComponentNode};
37use crate::port::{bind_or_panic, PortName, PortOwner};
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 { inner: self.inner.clone() }
133 }
134}
135
136impl<RSP> ResponseQueue<RSP> {
137 fn new() -> ResponseQueue<RSP> {
138 ResponseQueue {
139 inner: Rc::new(RespInner {
140 items: RefCell::new(Vec::new()),
141 waiters: RefCell::new(Vec::new()),
142 }),
143 }
144 }
145
146 fn push(&self, id: TxnId, rsp: RSP) {
147 self.inner.items.borrow_mut().push((id, rsp));
148 for w in self.inner.waiters.borrow_mut().drain(..) {
149 w.wake();
150 }
151 }
152
153 pub fn get_response(&self, txn_id: Option<TxnId>) -> GetResponse<RSP> {
156 GetResponse { inner: self.inner.clone(), txn_id }
157 }
158
159 pub fn try_get_response(&self, txn_id: Option<TxnId>) -> Option<RSP> {
167 let mut items = self.inner.items.borrow_mut();
168 let idx = match txn_id {
169 None => (!items.is_empty()).then_some(0),
170 Some(id) => items.iter().position(|(i, _)| *i == id),
171 };
172 idx.map(|i| items.remove(i).1)
173 }
174}
175
176pub struct GetResponse<RSP> {
177 inner: Rc<RespInner<RSP>>,
178 txn_id: Option<TxnId>,
179}
180
181impl<RSP> Future for GetResponse<RSP> {
182 type Output = RSP;
183 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<RSP> {
184 let mut items = self.inner.items.borrow_mut();
185 let idx = match self.txn_id {
186 None => (!items.is_empty()).then_some(0),
187 Some(id) => items.iter().position(|(i, _)| *i == id),
188 };
189 match idx {
190 Some(i) => Poll::Ready(items.remove(i).1),
191 None => {
192 drop(items);
193 self.inner.waiters.borrow_mut().push(cx.waker().clone());
194 Poll::Pending
195 }
196 }
197 }
198}
199
200struct SeqrInner<REQ: 'static, RSP: 'static> {
201 queue: Queue<Rc<ItemSlot<REQ>>>,
205 next_id: Cell<u64>,
206 responses: ResponseQueue<RSP>,
207 current: RefCell<Option<Rc<ItemSlot<REQ>>>>,
210}
211
212pub trait SeqItemIf<REQ: 'static, RSP: 'static>: 'static {
219 fn get_next_item(&self) -> Pin<Box<dyn Future<Output = SeqItem<REQ>> + '_>>;
220 fn try_next_item(&self) -> Option<SeqItem<REQ>>;
225 fn item_done(&self, rsp: Option<RSP>);
226 fn put_response(&self, id: TxnId, rsp: RSP);
230}
231
232impl<REQ: 'static, RSP: 'static> SeqItemIf<REQ, RSP> for SeqrInner<REQ, RSP> {
233 fn get_next_item(&self) -> Pin<Box<dyn Future<Output = SeqItem<REQ>> + '_>> {
234 Box::pin(async move {
235 assert!(
236 self.current.borrow().is_none(),
237 "get_next_item called twice without item_done"
238 );
239 let slot = self.queue.get().await;
240 slot.granted.set();
241 slot.ready.wait().await;
242 let payload = slot
243 .payload
244 .borrow_mut()
245 .take()
246 .expect("item ready but payload missing (rustdv bug)");
247 let item = SeqItem { id: slot.id, payload };
248 *self.current.borrow_mut() = Some(slot);
249 item
250 })
251 }
252
253 fn try_next_item(&self) -> Option<SeqItem<REQ>> {
254 assert!(self.current.borrow().is_none(), "try_next_item called without item_done");
255 let slot = self.queue.try_get()?;
259 slot.granted.set();
260 let taken = slot.payload.borrow_mut().take();
261 let payload = match taken {
262 Some(p) => p,
263 None => {
264 let _ = self.queue.try_put(slot);
266 return None;
267 }
268 };
269 let item = SeqItem { id: slot.id, payload };
270 *self.current.borrow_mut() = Some(slot);
271 Some(item)
272 }
273
274 fn item_done(&self, rsp: Option<RSP>) {
275 let slot = self
276 .current
277 .borrow_mut()
278 .take()
279 .expect("item_done without get_next_item");
280 if let Some(r) = rsp {
281 self.responses.push(slot.id, r);
282 }
283 slot.done.set();
284 }
285
286 fn put_response(&self, id: TxnId, rsp: RSP) {
287 self.responses.push(id, rsp);
288 }
289}
290
291pub type SeqItemPort<REQ, RSP = REQ> = crate::port::Port<dyn SeqItemIf<REQ, RSP>>;
293
294pub struct SeqItemExport<REQ: 'static, RSP: 'static> {
296 iface: Rc<dyn SeqItemIf<REQ, RSP>>,
297}
298
299impl<REQ: 'static, RSP: 'static> SeqItemExport<REQ, RSP> {
300 pub fn connect(&self, owner: &dyn PortOwner, name: PortName<dyn SeqItemIf<REQ, RSP>>) {
301 bind_or_panic(owner, name, self.iface.clone());
302 }
303
304 pub fn connect_port(&self, port: &SeqItemPort<REQ, RSP>) {
307 port.bind_iface(self.iface.clone());
308 }
309}
310
311pub struct Sequencer<REQ: 'static, RSP: 'static = REQ> {
321 inner: Rc<SeqrInner<REQ, RSP>>,
322}
323
324impl<REQ, RSP> Clone for Sequencer<REQ, RSP> {
325 fn clone(&self) -> Self {
326 Sequencer { inner: self.inner.clone() }
327 }
328}
329
330impl<REQ: 'static, RSP: 'static> fmt::Debug for Sequencer<REQ, RSP> {
331 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
334 f.write_str("Sequencer")
335 }
336}
337
338impl<REQ: 'static, RSP: 'static> Default for Sequencer<REQ, RSP> {
339 fn default() -> Self {
340 Self::new()
341 }
342}
343
344impl<REQ: 'static, RSP: 'static> Sequencer<REQ, RSP> {
345 pub fn new() -> Sequencer<REQ, RSP> {
346 Sequencer {
347 inner: Rc::new(SeqrInner {
348 queue: Queue::unbounded(),
349 next_id: Cell::new(1),
350 responses: ResponseQueue::new(),
351 current: RefCell::new(None),
352 }),
353 }
354 }
355
356 pub fn handle(&self) -> Sequencer<REQ, RSP> {
358 self.clone()
359 }
360
361 pub fn seq_item_export(&self) -> SeqItemExport<REQ, RSP> {
363 SeqItemExport { iface: self.inner.clone() }
364 }
365}
366
367impl<REQ: 'static, RSP: 'static> Component for Sequencer<REQ, RSP> {}
370
371impl<REQ: 'static, RSP: 'static> ComponentNode for Sequencer<REQ, RSP> {
372 fn node_name(&self) -> &'static str {
373 "Sequencer"
374 }
375 fn children_mut(&mut self) -> Vec<(String, &mut (dyn ComponentNode + 'static))> {
376 Vec::new()
377 }
378}
379
380pub struct SeqCtx<REQ: 'static, RSP: 'static = REQ> {
388 inner: Option<Rc<SeqrInner<REQ, RSP>>>,
389 current: Option<Rc<ItemSlot<REQ>>>,
390 name: &'static str,
391 logger: Logger,
392 seed: u64,
393}
394
395impl<REQ: 'static, RSP: 'static> SeqCtx<REQ, RSP> {
396 fn new(inner: Option<Rc<SeqrInner<REQ, RSP>>>, name: &'static str, seed: u64) -> Self {
397 SeqCtx {
398 inner,
399 current: None,
400 name,
401 logger: Logger::at(RustdvPath::root(name)),
402 seed,
403 }
404 }
405
406 pub fn name(&self) -> &'static str {
409 self.name
410 }
411
412 pub fn rng(&self) -> Rng {
415 Rng::new(self.seed)
416 }
417
418 pub fn info(&self, msg: &str) {
419 self.logger.info(msg);
420 }
421 pub fn warning(&self, msg: &str) {
422 self.logger.warning(msg);
423 }
424 pub fn error(&self, msg: &str) {
425 self.logger.error(msg);
426 }
427
428 fn seqr(&self) -> Result<&Rc<SeqrInner<REQ, RSP>>, SeqError> {
429 self.inner.as_ref().ok_or_else(|| {
430 SeqError(format!(
431 "{}: start_item in a virtual sequence — it was started without a sequencer",
432 self.name
433 ))
434 })
435 }
436
437 pub async fn start_item(&mut self, _item: &mut REQ) -> Result<(), SeqError> {
440 if self.current.is_some() {
441 return Err(SeqError("start_item called twice without finish_item".into()));
442 }
443 let inner = self.seqr()?.clone();
444 let id = TxnId(inner.next_id.get());
445 inner.next_id.set(id.0 + 1);
446 let slot = Rc::new(ItemSlot {
447 id,
448 granted: Event::new(),
449 ready: Event::new(),
450 done: Event::new(),
451 payload: RefCell::new(None),
452 });
453 self.current = Some(slot.clone());
454 inner.queue.put(slot.clone()).await;
455 slot.granted.wait().await;
456 Ok(())
457 }
458
459 pub async fn finish_item(&mut self, item: REQ) -> Result<TxnId, SeqError> {
462 let slot = self
463 .current
464 .take()
465 .ok_or_else(|| SeqError("finish_item without start_item".into()))?;
466 *slot.payload.borrow_mut() = Some(item);
467 slot.ready.set();
468 slot.done.wait().await;
469 Ok(slot.id)
470 }
471
472 pub fn try_get_response(&mut self, txn_id: Option<TxnId>) -> Option<RSP> {
479 let responses = self.seqr().ok()?.responses.clone();
480 responses.try_get_response(txn_id)
481 }
482
483 pub async fn get_response(&mut self, txn_id: Option<TxnId>) -> RSP {
486 let responses = self.seqr().expect("get_response in a virtual sequence").responses.clone();
487 responses.get_response(txn_id).await
488 }
489}
490
491pub trait Sequence: Sized + 'static {
502 type Req: 'static;
503 type Rsp: 'static;
504
505 fn body(
506 &mut self,
507 ctx: &mut SeqCtx<Self::Req, Self::Rsp>,
508 ) -> impl Future<Output = Result<(), SeqError>>;
509
510 fn seq_name(&self) -> &'static str {
515 let full = std::any::type_name::<Self>();
516 full.rsplit("::").next().unwrap_or(full)
517 }
518
519 fn start(
521 &mut self,
522 seqr: &Sequencer<Self::Req, Self::Rsp>,
523 ) -> impl Future<Output = Result<(), SeqError>> {
524 let inner = seqr.inner.clone();
525 let name = self.seq_name();
526 async move {
527 let mut ctx = SeqCtx::new(Some(inner), name, current_seed());
528 self.body(&mut ctx).await
529 }
530 }
531
532 fn start_virtual(&mut self) -> impl Future<Output = Result<(), SeqError>> {
536 let name = self.seq_name();
537 async move {
538 let mut ctx = SeqCtx::new(None, name, current_seed());
539 self.body(&mut ctx).await
540 }
541 }
542}
543
544thread_local! {
545 static SEED: Cell<u64> = const { Cell::new(1) };
546}
547
548pub fn set_sequence_seed(seed: u64) {
550 SEED.with(|s| s.set(seed));
551}
552
553fn current_seed() -> u64 {
554 SEED.with(|s| s.get())
555}
556
557pub trait DynSequence<REQ: 'static, RSP: 'static> {
564 fn dyn_body<'a>(
565 &'a mut self,
566 ctx: &'a mut SeqCtx<REQ, RSP>,
567 ) -> Pin<Box<dyn Future<Output = Result<(), SeqError>> + 'a>>;
568 fn dyn_name(&self) -> &'static str;
569}
570
571impl<S: Sequence> DynSequence<S::Req, S::Rsp> for S {
572 fn dyn_body<'a>(
573 &'a mut self,
574 ctx: &'a mut SeqCtx<S::Req, S::Rsp>,
575 ) -> Pin<Box<dyn Future<Output = Result<(), SeqError>> + 'a>> {
576 Box::pin(self.body(ctx))
577 }
578 fn dyn_name(&self) -> &'static str {
579 self.seq_name()
580 }
581}
582
583pub struct RustdvSeq<REQ: 'static, RSP: 'static = REQ> {
587 inner: Option<Box<dyn DynSequence<REQ, RSP>>>,
588}
589
590impl<REQ: 'static, RSP: 'static> Default for RustdvSeq<REQ, RSP> {
591 fn default() -> Self {
592 RustdvSeq { inner: None }
593 }
594}
595
596impl<REQ: 'static, RSP: 'static> RustdvSeq<REQ, RSP> {
597 pub fn new(seq: Box<dyn DynSequence<REQ, RSP>>) -> Self {
598 RustdvSeq { inner: Some(seq) }
599 }
600
601 pub fn name(&self) -> &'static str {
603 self.inner.as_ref().map(|s| s.dyn_name()).unwrap_or("<empty>")
604 }
605
606 fn get(&mut self) -> Result<&mut Box<dyn DynSequence<REQ, RSP>>, SeqError> {
607 self.inner.as_mut().ok_or_else(|| SeqError("an empty sequence slot".into()))
608 }
609
610 pub async fn start(&mut self, seqr: &Sequencer<REQ, RSP>) -> Result<(), SeqError> {
611 let inner = seqr.inner.clone();
612 let seq = self.get()?;
613 let mut ctx = SeqCtx::new(Some(inner), seq.dyn_name(), current_seed());
614 seq.dyn_body(&mut ctx).await
615 }
616
617 pub async fn start_virtual(&mut self) -> Result<(), SeqError> {
618 let seq = self.get()?;
619 let mut ctx = SeqCtx::new(None, seq.dyn_name(), current_seed());
620 seq.dyn_body(&mut ctx).await
621 }
622}
623
624type SeqOverrides = HashMap<TypeId, (TypeId, fn() -> Box<dyn Any>)>;
629
630thread_local! {
631 static SEQ_OVERRIDES: RefCell<SeqOverrides> = RefCell::new(HashMap::new());
632}
633
634pub fn clear_seq_overrides() {
636 SEQ_OVERRIDES.with(|o| o.borrow_mut().clear());
637}
638
639pub fn set_seq_override<From, To>()
646where
647 From: Sequence,
648 To: Sequence<Req = From::Req, Rsp = From::Rsp> + Default,
649{
650 fn maker<To: Sequence + Default>() -> Box<dyn Any> {
651 let boxed: Box<dyn DynSequence<To::Req, To::Rsp>> = Box::new(To::default());
652 Box::new(boxed)
653 }
654 SEQ_OVERRIDES.with(|o| {
655 o.borrow_mut()
656 .insert(TypeId::of::<From>(), (TypeId::of::<To>(), maker::<To>));
657 });
658}
659
660pub fn create_seq<S>() -> RustdvSeq<S::Req, S::Rsp>
662where
663 S: Sequence + Default,
664{
665 let over = SEQ_OVERRIDES.with(|o| o.borrow().get(&TypeId::of::<S>()).map(|(_, m)| *m));
666 match over {
667 Some(make) => {
668 let any = make();
669 let boxed = any
670 .downcast::<Box<dyn DynSequence<S::Req, S::Rsp>>>()
671 .expect("sequence override built the wrong request/response types");
672 RustdvSeq::new(*boxed)
673 }
674 None => RustdvSeq::new(Box::new(S::default())),
675 }
676}
677
678#[cfg(test)]
687mod tests {
688 use super::*;
689 use rustdv_sim::executor;
690 use rustdv_sim::testing::{assert_pending, block_on};
691
692 #[derive(Clone, Debug, PartialEq, Eq, Default)]
693 struct Cmd {
694 a: u8,
695 tag: &'static str,
696 }
697
698 #[derive(Clone, Debug, PartialEq, Eq, Default)]
699 struct Rsp {
700 v: u8,
701 }
702
703 fn spawn_one_shot_driver(seqr: &Sequencer<Cmd, Rsp>, answer: u8) {
705 let port = seqr.seq_item_export();
706 let inner = seqr.inner.clone();
707 let _ = port; executor::spawn(async move {
709 let item = SeqItemIf::get_next_item(&*inner).await;
710 let v = item.payload().a.wrapping_add(answer);
711 SeqItemIf::item_done(&*inner, Some(Rsp { v }));
712 });
713 }
714
715 #[test]
716 fn start_item_blocks_until_the_driver_asks() {
717 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
718 let inner = seqr.inner.clone();
719 assert_pending(async move {
721 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
722 let mut cmd = Cmd::default();
723 ctx.start_item(&mut cmd).await.unwrap();
724 });
725 }
726
727 #[test]
731 fn the_driver_sees_what_was_written_after_the_grant() {
732 block_on(async {
733 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
734 let inner = seqr.inner.clone();
735 let seen = Rc::new(RefCell::new(None));
736 let seen2 = seen.clone();
737 let d = inner.clone();
738 executor::spawn(async move {
739 let item = SeqItemIf::get_next_item(&*d).await;
740 *seen2.borrow_mut() = Some(item.payload().clone());
741 SeqItemIf::item_done(&*d, None);
742 });
743
744 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
745 let mut cmd = Cmd { a: 0, tag: "before" };
746 ctx.start_item(&mut cmd).await.unwrap();
747 cmd.a = 42;
749 cmd.tag = "after the grant";
750 ctx.finish_item(cmd).await.unwrap();
751
752 let got = seen.borrow().clone().expect("the driver got an item");
753 assert_eq!(got.a, 42, "the late value reached the driver");
754 assert_eq!(got.tag, "after the grant");
755 });
756 }
757
758 #[test]
759 fn finish_item_returns_the_ticket_and_waits_for_item_done() {
760 block_on(async {
761 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
762 spawn_one_shot_driver(&seqr, 1);
763 let mut ctx = SeqCtx::new(Some(seqr.inner.clone()), "T", 1);
764 let mut cmd = Cmd { a: 10, tag: "x" };
765 ctx.start_item(&mut cmd).await.unwrap();
766 let ticket = ctx.finish_item(cmd).await.unwrap();
767 assert_eq!(ticket, TxnId(1), "tickets start at 1");
768 });
769 }
770
771 #[test]
772 fn tickets_are_unique_and_ascending() {
773 block_on(async {
774 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
775 let inner = seqr.inner.clone();
776 let d = inner.clone();
777 executor::spawn(async move {
778 for _ in 0..3 {
779 let _item = SeqItemIf::get_next_item(&*d).await;
780 SeqItemIf::item_done(&*d, None);
781 }
782 });
783 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
784 let mut tickets = Vec::new();
785 for a in 0..3u8 {
786 let mut cmd = Cmd { a, tag: "" };
787 ctx.start_item(&mut cmd).await.unwrap();
788 tickets.push(ctx.finish_item(cmd).await.unwrap());
789 }
790 assert_eq!(tickets, vec![TxnId(1), TxnId(2), TxnId(3)]);
791 });
792 }
793
794 #[test]
797 #[should_panic(expected = "get_next_item called twice without item_done")]
798 fn two_get_next_items_without_item_done_is_a_bug() {
799 block_on(async {
800 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
801 let inner = seqr.inner.clone();
802 let s = inner.clone();
804 executor::spawn(async move {
805 let mut ctx = SeqCtx::new(Some(s), "T", 1);
806 let mut cmd = Cmd::default();
807 ctx.start_item(&mut cmd).await.unwrap();
808 ctx.finish_item(cmd).await.unwrap();
809 });
810 let _a = SeqItemIf::get_next_item(&*inner).await;
814 let _b = SeqItemIf::get_next_item(&*inner).await;
815 });
816 }
817
818 #[test]
819 fn start_item_twice_without_finish_is_an_error() {
820 block_on(async {
821 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
822 let inner = seqr.inner.clone();
823 let d = inner.clone();
824 executor::spawn(async move {
825 let _ = SeqItemIf::get_next_item(&*d).await;
826 });
827 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
828 let mut a = Cmd::default();
829 ctx.start_item(&mut a).await.unwrap();
830 let mut b = Cmd::default();
831 let err = ctx.start_item(&mut b).await;
832 assert!(err.is_err(), "a second start_item without finish_item");
833 });
834 }
835
836 #[test]
838 fn try_next_item_is_none_on_an_empty_sequencer() {
839 block_on(async {
840 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
841 assert!(SeqItemIf::try_next_item(&*seqr.inner).is_none());
842 });
843 }
844
845 #[test]
846 fn try_next_item_takes_a_waiting_item() {
847 block_on(async {
848 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
849 let inner = seqr.inner.clone();
850 let s = inner.clone();
851 executor::spawn(async move {
852 let mut ctx = SeqCtx::new(Some(s), "T", 1);
853 let mut cmd = Cmd { a: 5, tag: "" };
854 ctx.start_item(&mut cmd).await.unwrap();
855 ctx.finish_item(cmd).await.unwrap();
856 });
857 let mut got = None;
860 for _ in 0..8 {
861 executor::current().run_until_idle();
862 if let Some(item) = SeqItemIf::try_next_item(&*inner) {
863 got = Some(item.payload().a);
864 SeqItemIf::item_done(&*inner, None);
865 break;
866 }
867 }
868 assert_eq!(got, Some(5));
869 });
870 }
871
872 #[test]
873 fn item_done_with_a_response_reaches_get_response() {
874 block_on(async {
875 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
876 spawn_one_shot_driver(&seqr, 100);
877 let mut ctx = SeqCtx::new(Some(seqr.inner.clone()), "T", 1);
878 let mut cmd = Cmd { a: 1, tag: "" };
879 ctx.start_item(&mut cmd).await.unwrap();
880 let ticket = ctx.finish_item(cmd).await.unwrap();
881 let rsp = ctx.get_response(Some(ticket)).await;
882 assert_eq!(rsp.v, 101);
883 });
884 }
885
886 #[test]
889 fn put_response_answers_after_item_done() {
890 block_on(async {
891 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
892 let inner = seqr.inner.clone();
893 let d = inner.clone();
894 executor::spawn(async move {
895 let item = SeqItemIf::get_next_item(&*d).await;
896 let id = item.txn_id();
897 SeqItemIf::item_done(&*d, None); SeqItemIf::put_response(&*d, id, Rsp { v: 77 });
899 });
900 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
901 let mut cmd = Cmd::default();
902 ctx.start_item(&mut cmd).await.unwrap();
903 let ticket = ctx.finish_item(cmd).await.unwrap();
904 assert_eq!(ctx.get_response(Some(ticket)).await.v, 77);
905 });
906 }
907
908 #[test]
911 fn get_response_picks_its_ticket_out_of_order() {
912 block_on(async {
913 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
914 let inner = seqr.inner.clone();
915 inner.responses.push(TxnId(2), Rsp { v: 22 });
917 inner.responses.push(TxnId(1), Rsp { v: 11 });
918 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
919 assert_eq!(ctx.get_response(Some(TxnId(1))).await.v, 11);
920 assert_eq!(ctx.get_response(Some(TxnId(2))).await.v, 22);
921 });
922 }
923
924 #[test]
925 fn get_response_none_takes_the_oldest() {
926 block_on(async {
927 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
928 let inner = seqr.inner.clone();
929 inner.responses.push(TxnId(5), Rsp { v: 50 });
930 inner.responses.push(TxnId(6), Rsp { v: 60 });
931 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
932 assert_eq!(ctx.get_response(None).await.v, 50, "oldest first");
933 assert_eq!(ctx.get_response(None).await.v, 60);
934 });
935 }
936
937 #[test]
938 fn try_get_response_does_not_wait() {
939 block_on(async {
940 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
941 let inner = seqr.inner.clone();
942 let mut ctx = SeqCtx::new(Some(inner.clone()), "T", 1);
943 assert!(ctx.try_get_response(Some(TxnId(1))).is_none(), "nothing yet");
944 inner.responses.push(TxnId(1), Rsp { v: 9 });
945 assert_eq!(ctx.try_get_response(Some(TxnId(1))).unwrap().v, 9);
946 assert!(ctx.try_get_response(Some(TxnId(1))).is_none(), "and it was taken");
947 });
948 }
949
950 #[test]
951 fn a_response_that_never_comes_waits_forever() {
952 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
953 let inner = seqr.inner.clone();
954 assert_pending(async move {
955 let mut ctx = SeqCtx::new(Some(inner), "T", 1);
956 ctx.get_response(Some(TxnId(1043))).await
957 });
958 }
959
960 #[test]
963 fn start_item_in_a_virtual_sequence_is_an_error() {
964 block_on(async {
965 let mut ctx: SeqCtx<Cmd, Rsp> = SeqCtx::new(None, "MyVirtualSeq", 1);
966 let mut cmd = Cmd::default();
967 match ctx.start_item(&mut cmd).await {
968 Err(SeqError(msg)) => {
969 assert!(msg.contains("virtual"), "the error explains: {msg}");
970 assert!(msg.contains("MyVirtualSeq"), "and names the sequence: {msg}");
971 }
972 Ok(()) => panic!("start_item should fail without a sequencer"),
973 }
974 });
975 }
976
977 #[test]
979 fn two_sequences_interleave_one_item_each() {
980 block_on(async {
981 let seqr: Sequencer<Cmd, Rsp> = Sequencer::new();
982 let inner = seqr.inner.clone();
983 let order = Rc::new(RefCell::new(Vec::new()));
984
985 for tag in ["A", "B"] {
986 let s = inner.clone();
987 executor::spawn(async move {
988 let mut ctx = SeqCtx::new(Some(s), tag, 1);
989 for a in 0..2u8 {
990 let mut cmd = Cmd { a, tag };
991 ctx.start_item(&mut cmd).await.unwrap();
992 ctx.finish_item(cmd).await.unwrap();
993 }
994 });
995 }
996
997 let d = inner.clone();
998 let seen = order.clone();
999 executor::spawn(async move {
1000 for _ in 0..4 {
1001 let item = SeqItemIf::get_next_item(&*d).await;
1002 seen.borrow_mut().push(item.payload().tag);
1003 SeqItemIf::item_done(&*d, None);
1004 }
1005 });
1006
1007 for _ in 0..64 {
1008 executor::current().run_until_idle();
1009 }
1010 let got = order.borrow().clone();
1011 assert_eq!(got.len(), 4, "all four items were driven");
1012 assert_eq!(got, vec!["A", "B", "A", "B"], "one item each, in turn");
1013 });
1014 }
1015
1016 #[test]
1018 fn a_sequence_name_defaults_to_its_type() {
1019 #[derive(Default)]
1020 struct MyFancySeq;
1021 impl Sequence for MyFancySeq {
1022 type Req = Cmd;
1023 type Rsp = Rsp;
1024 async fn body(&mut self, _ctx: &mut SeqCtx<Cmd, Rsp>) -> Result<(), SeqError> {
1025 Ok(())
1026 }
1027 }
1028 assert_eq!(MyFancySeq.seq_name(), "MyFancySeq");
1029 }
1030
1031 #[test]
1032 fn a_virtual_sequence_runs_its_body() {
1033 #[derive(Default)]
1034 struct VSeq {
1035 ran: bool,
1036 }
1037 impl Sequence for VSeq {
1038 type Req = Cmd;
1039 type Rsp = Rsp;
1040 async fn body(&mut self, _ctx: &mut SeqCtx<Cmd, Rsp>) -> Result<(), SeqError> {
1041 self.ran = true;
1042 Ok(())
1043 }
1044 }
1045 block_on(async {
1046 let mut s = VSeq::default();
1047 s.start_virtual().await.unwrap();
1048 assert!(s.ran);
1049 });
1050 }
1051}