gwr_engine/port/
mod.rs

1// Copyright (c) 2024 Graphcore Ltd. All rights reserved.
2
3//! Typed, back-pressure-aware ports used to compose components and models.
4//!
5//! Ports are the main runtime boundary between simulation blocks. An
6//! [`OutPort<T>`] connects to an [`InPort<T>`] through shared [`PortState<T>`],
7//! and the type parameter is part of the contract: only ports carrying the same
8//! simulation object type can be connected.
9//!
10//! Construct components and models, connect every required port, and then
11//! start the engine. An unconnected port is reported
12//! when the active task calls [`get`](InPort::get),
13//! [`start_get`](InPort::start_get), [`put`](OutPort::put), or
14//! [`try_put`](OutPort::try_put), so a "not connected" error should be
15//! treated as a topology construction bug rather than as ordinary simulation
16//! behavior.
17//!
18//! Data transfer is flow controlled. `put` waits until the receiver accepts the
19//! value, `get` waits until a value is available, and `try_put` asynchronously
20//! waits until a receiver is waiting in `get` or `start_get` without
21//! transferring a value.
22//! [`start_get`](InPort::start_get) / [`finish_get`](InPort::finish_get)
23//! support protocols that need to observe a value before releasing the sender.
24
25use std::cell::RefCell;
26use std::fmt;
27use std::pin::Pin;
28use std::rc::Rc;
29use std::task::{Context, Poll, Waker};
30
31use futures::Future;
32use futures::future::FusedFuture;
33use gwr_track::connect;
34use gwr_track::entity::{Entity, GetEntity};
35use gwr_track::tracker::aka::Aka;
36
37use crate::engine::Engine;
38use crate::port::monitor::Monitor;
39use crate::sim_error;
40use crate::time::clock::Clock;
41use crate::traits::SimObject;
42use crate::types::{SimError, SimResult};
43
44pub mod monitor;
45
46pub type PortStateResult<T> = Result<Rc<PortState<T>>, SimError>;
47pub type PortGetResult<T> = Result<PortGet<T>, SimError>;
48pub type PortStartGetResult<T> = Result<PortStartGet<T>, SimError>;
49pub type PortPutResult<T> = Result<PortPut<T>, SimError>;
50pub type PortTryPutResult<T> = Result<PortTryPut<T>, SimError>;
51
52pub struct PortState<T>
53where
54    T: SimObject,
55{
56    value: RefCell<Option<T>>,
57    put_released: RefCell<bool>,
58    waiting_get: RefCell<Option<Waker>>,
59    waiting_put: RefCell<Option<Waker>>,
60    pub in_port_entity: Rc<Entity>,
61    monitor: Option<Rc<Monitor>>,
62}
63
64impl<T> PortState<T>
65where
66    T: SimObject,
67{
68    fn new(
69        engine: &Engine,
70        clock: &Clock,
71        in_port_entity: Rc<Entity>,
72        window_size_ticks: Option<u64>,
73    ) -> Self {
74        let monitor = window_size_ticks.map(|window_size_ticks| {
75            Monitor::new_and_register(engine, &in_port_entity, clock, window_size_ticks)
76        });
77        Self {
78            value: RefCell::new(None),
79            put_released: RefCell::new(true),
80            waiting_get: RefCell::new(None),
81            waiting_put: RefCell::new(None),
82            in_port_entity,
83            monitor,
84        }
85    }
86}
87
88pub struct InPort<T>
89where
90    T: SimObject,
91{
92    entity: Rc<Entity>,
93    state: Rc<PortState<T>>,
94    connected: RefCell<bool>,
95}
96
97impl<T> fmt::Display for InPort<T>
98where
99    T: SimObject,
100{
101    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
102        self.entity.fmt(f)
103    }
104}
105
106impl<T> InPort<T>
107where
108    T: SimObject,
109{
110    #[must_use]
111    pub fn new(engine: &Engine, clock: &Clock, parent: &Rc<Entity>, name: &str) -> Self {
112        Self::new_with_renames(engine, clock, parent, name, None)
113    }
114
115    #[must_use]
116    pub fn new_with_renames(
117        engine: &Engine,
118        clock: &Clock,
119        parent: &Rc<Entity>,
120        name: &str,
121        aka: Option<&Aka>,
122    ) -> Self {
123        let entity = Rc::new(Entity::new_with_renames(parent, name, aka));
124        let monitor_window_size = entity.tracker.monitoring_window_size_for(entity.id);
125        Self {
126            entity: entity.clone(),
127            state: Rc::new(PortState::new(engine, clock, entity, monitor_window_size)),
128            connected: RefCell::new(false),
129        }
130    }
131
132    pub fn state(&self) -> PortStateResult<T> {
133        if *self.connected.borrow() {
134            return sim_error!("{self} already connected");
135        }
136
137        *self.connected.borrow_mut() = true;
138        Ok(self.state.clone())
139    }
140
141    #[must_use]
142    pub fn has_value(&self) -> bool {
143        self.state.value.borrow().is_some()
144    }
145
146    #[must_use = "Futures do nothing unless you `.await` or otherwise use them"]
147    pub fn get(&mut self) -> PortGetResult<T> {
148        if !*self.connected.borrow() {
149            return sim_error!("{self} not connected");
150        }
151
152        Ok(PortGet {
153            state: self.state.clone(),
154            done: false,
155        })
156    }
157
158    /// Must be matched with a `finish_get` to allow the OutPort to continue.
159    #[must_use = "Futures do nothing unless you `.await` or otherwise use them"]
160    pub fn start_get(&mut self) -> PortStartGetResult<T> {
161        if !*self.connected.borrow() {
162            return sim_error!("{self} not connected");
163        }
164
165        Ok(PortStartGet {
166            state: self.state.clone(),
167            done: false,
168        })
169    }
170
171    /// Must be matched with a `start_get ` to consume the value.
172    pub fn finish_get(&mut self) {
173        *self.state.put_released.borrow_mut() = true;
174        if let Some(waker) = self.state.waiting_put.borrow_mut().take() {
175            waker.wake();
176        }
177    }
178}
179
180pub struct OutPort<T>
181where
182    T: SimObject,
183{
184    entity: Rc<Entity>,
185    state: Option<Rc<PortState<T>>>,
186}
187
188impl<T> GetEntity for OutPort<T>
189where
190    T: SimObject,
191{
192    fn entity(&self) -> &Rc<Entity> {
193        &self.entity
194    }
195}
196
197impl<T> fmt::Display for OutPort<T>
198where
199    T: SimObject,
200{
201    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
202        self.entity.fmt(f)
203    }
204}
205
206impl<T> OutPort<T>
207where
208    T: SimObject,
209{
210    #[must_use]
211    pub fn new(parent: &Rc<Entity>, name: &str) -> Self {
212        Self::new_with_renames(parent, name, None)
213    }
214
215    #[must_use]
216    pub fn new_with_renames(parent: &Rc<Entity>, name: &str, aka: Option<&Aka>) -> Self {
217        let entity = Rc::new(Entity::new_with_renames(parent, name, aka));
218        Self {
219            entity,
220            state: None,
221        }
222    }
223
224    pub fn connect(&mut self, port_state: PortStateResult<T>) -> SimResult {
225        let port_state = port_state?;
226
227        connect!(self.entity ; port_state.in_port_entity);
228        match self.state {
229            Some(_) => {
230                return sim_error!("{self} already connected");
231            }
232            None => {
233                self.state = Some(port_state);
234            }
235        }
236        Ok(())
237    }
238
239    #[must_use = "Futures do nothing unless you `.await` or otherwise use them"]
240    pub fn put(&mut self, value: T) -> PortPutResult<T> {
241        let state = match self.state.as_ref() {
242            Some(s) => s.clone(),
243            None => return sim_error!("{self} not connected"),
244        };
245        Ok(PortPut {
246            state,
247            value: Some(value),
248            done: false,
249        })
250    }
251
252    /// Waits until the connected receiver is waiting in [`InPort::get`] or
253    /// [`InPort::start_get`].
254    ///
255    /// This does not transfer a value or return an immediate availability
256    /// result. The returned future remains pending until the receiver starts a
257    /// receive operation, and may therefore wait indefinitely if the receiver
258    /// never does.
259    #[must_use = "Futures do nothing unless you `.await` or otherwise use them"]
260    pub fn try_put(&mut self) -> PortTryPutResult<T> {
261        let state = match self.state.as_ref() {
262            Some(s) => s.clone(),
263            None => return sim_error!("{self} not connected"),
264        };
265        Ok(PortTryPut { state, done: false })
266    }
267}
268
269pub struct PortPut<T>
270where
271    T: SimObject,
272{
273    state: Rc<PortState<T>>,
274    value: Option<T>,
275    done: bool,
276}
277
278impl<T> Future for PortPut<T>
279where
280    T: SimObject,
281{
282    type Output = ();
283
284    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
285        match self.value.take() {
286            Some(value) => {
287                // The state is designed to be shared between one put/get pair
288                // so it should not be possible for the value in the state to be
289                // set at this point.
290                assert!(self.state.value.borrow().is_none());
291
292                *self.state.value.borrow_mut() = Some(value);
293                *self.state.put_released.borrow_mut() = false;
294                if let Some(waker) = self.state.waiting_get.borrow_mut().take() {
295                    waker.wake();
296                }
297                *self.state.waiting_put.borrow_mut() = Some(cx.waker().clone());
298                Poll::Pending
299            }
300            None => {
301                if *self.state.put_released.borrow() {
302                    // Getter has consumed the value and released the putter.
303                    self.done = true;
304                    Poll::Ready(())
305                } else {
306                    // Stay pending as the task was woken before the getter has
307                    // removed the value and released the putter.
308                    *self.state.waiting_put.borrow_mut() = Some(cx.waker().clone());
309                    Poll::Pending
310                }
311            }
312        }
313    }
314}
315
316impl<T> FusedFuture for PortPut<T>
317where
318    T: SimObject,
319{
320    fn is_terminated(&self) -> bool {
321        self.done
322    }
323}
324
325pub struct PortTryPut<T>
326where
327    T: SimObject,
328{
329    state: Rc<PortState<T>>,
330    done: bool,
331}
332
333impl<T> Future for PortTryPut<T>
334where
335    T: SimObject,
336{
337    type Output = ();
338
339    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
340        if self.state.waiting_get.borrow().is_some() {
341            self.done = true;
342            Poll::Ready(())
343        } else {
344            *self.state.waiting_put.borrow_mut() = Some(cx.waker().clone());
345            Poll::Pending
346        }
347    }
348}
349
350impl<T> FusedFuture for PortTryPut<T>
351where
352    T: SimObject,
353{
354    fn is_terminated(&self) -> bool {
355        self.done
356    }
357}
358
359pub struct PortGet<T>
360where
361    T: SimObject,
362{
363    state: Rc<PortState<T>>,
364    done: bool,
365}
366
367impl<T> Future for PortGet<T>
368where
369    T: SimObject,
370{
371    type Output = T;
372
373    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
374        let value = self.state.value.borrow_mut().take();
375        if let Some(value) = value {
376            self.done = true;
377            self.state.waiting_get.borrow_mut().take();
378            *self.state.put_released.borrow_mut() = true;
379
380            // Track the object through the port monitor if there is one
381            if let Some(monitor) = self.state.monitor.as_ref() {
382                monitor.sample(&value);
383            }
384
385            if let Some(waker) = self.state.waiting_put.borrow_mut().take() {
386                waker.wake();
387            }
388            Poll::Ready(value)
389        } else {
390            if let Some(waker) = self.state.waiting_put.borrow_mut().take() {
391                waker.wake();
392            }
393
394            *self.state.waiting_get.borrow_mut() = Some(cx.waker().clone());
395            Poll::Pending
396        }
397    }
398}
399
400impl<T> FusedFuture for PortGet<T>
401where
402    T: SimObject,
403{
404    fn is_terminated(&self) -> bool {
405        self.done
406    }
407}
408
409pub struct PortStartGet<T>
410where
411    T: SimObject,
412{
413    state: Rc<PortState<T>>,
414    done: bool,
415}
416
417impl<T> Future for PortStartGet<T>
418where
419    T: SimObject,
420{
421    type Output = T;
422
423    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
424        let value = self.state.value.borrow_mut().take();
425        if let Some(value) = value {
426            self.done = true;
427            self.state.waiting_get.borrow_mut().take();
428
429            // Track the object through the port monitor if there is one
430            if let Some(monitor) = self.state.monitor.as_ref() {
431                monitor.sample(&value);
432            }
433
434            Poll::Ready(value)
435        } else {
436            // Publish receiver readiness to a producer parked in try_put.
437            // This is not the packet ACK: only finish_get releases the put.
438            if let Some(waker) = self.state.waiting_put.borrow_mut().take() {
439                waker.wake();
440            }
441            *self.state.waiting_get.borrow_mut() = Some(cx.waker().clone());
442            Poll::Pending
443        }
444    }
445}
446
447impl<T> FusedFuture for PortStartGet<T>
448where
449    T: SimObject,
450{
451    fn is_terminated(&self) -> bool {
452        self.done
453    }
454}
455
456#[cfg(test)]
457mod tests {
458    use std::sync::Arc;
459    use std::sync::atomic::{AtomicUsize, Ordering};
460    use std::task::{Wake, Waker};
461
462    use futures::future::FusedFuture;
463    use futures::task::noop_waker;
464    use gwr_track::Tracker;
465    use gwr_track::entity::Entity;
466    use gwr_track::tracker::dev_null_tracker;
467
468    use super::*;
469    use crate::traits::TotalBytes;
470
471    struct TestContext {
472        // Just kept to ensure it isn't dropped
473        _tracker: Tracker,
474        engine: Engine,
475        clock: Clock,
476    }
477
478    fn test_context() -> TestContext {
479        let tracker = dev_null_tracker();
480        let mut engine = Engine::new(&tracker);
481        let clock = engine.default_clock();
482
483        TestContext {
484            _tracker: tracker,
485            engine,
486            clock,
487        }
488    }
489
490    fn test_state<T: SimObject>() -> Rc<PortState<T>> {
491        let context = test_context();
492        let entity = Rc::new(Entity::new(context.engine.top(), "rx"));
493
494        Rc::new(PortState::new(
495            &context.engine,
496            &context.clock,
497            entity,
498            None,
499        ))
500    }
501
502    fn monitored_test_state<T: SimObject>() -> Rc<PortState<T>> {
503        let context = test_context();
504        let entity = Rc::new(Entity::new(context.engine.top(), "rx"));
505
506        Rc::new(PortState::new(
507            &context.engine,
508            &context.clock,
509            entity,
510            Some(1),
511        ))
512    }
513
514    struct WakeCounter {
515        wakes_count: Arc<AtomicUsize>,
516    }
517
518    impl Wake for WakeCounter {
519        fn wake(self: Arc<Self>) {
520            self.wakes_count.fetch_add(1, Ordering::SeqCst);
521        }
522
523        fn wake_by_ref(self: &Arc<Self>) {
524            self.wakes_count.fetch_add(1, Ordering::SeqCst);
525        }
526    }
527
528    fn counting_waker() -> (Arc<AtomicUsize>, Waker) {
529        let wakes_count = Arc::new(AtomicUsize::new(0));
530        let waker = Waker::from(Arc::new(WakeCounter {
531            wakes_count: wakes_count.clone(),
532        }));
533
534        (wakes_count, waker)
535    }
536
537    #[test]
538    fn wake_counter_counts_wake_and_wake_by_ref() {
539        let (wakes_count, waker) = counting_waker();
540
541        waker.wake_by_ref();
542        assert_eq!(wakes_count.load(Ordering::SeqCst), 1);
543
544        waker.wake();
545        assert_eq!(wakes_count.load(Ordering::SeqCst), 2);
546    }
547
548    #[test]
549    fn in_port_state_can_only_connect_once() {
550        let context = test_context();
551        let in_port =
552            InPort::<i32>::new(&context.engine, &context.clock, context.engine.top(), "rx");
553
554        assert!(in_port.state().is_ok());
555
556        let err = in_port
557            .state()
558            .err()
559            .expect("second state call should fail");
560        assert!(format!("{err}").contains("already connected"));
561    }
562
563    #[test]
564    fn out_port_connect_can_only_connect_once() {
565        let context = test_context();
566        let mut out_port = OutPort::<i32>::new(context.engine.top(), "tx");
567        let first_in_port =
568            InPort::new(&context.engine, &context.clock, context.engine.top(), "rx1");
569        let second_in_port =
570            InPort::new(&context.engine, &context.clock, context.engine.top(), "rx2");
571
572        out_port.connect(first_in_port.state()).unwrap();
573
574        let err = out_port.connect(second_in_port.state()).unwrap_err();
575        assert!(format!("{err}").contains("already connected"));
576    }
577
578    #[test]
579    fn out_port_entity_returns_port_entity() {
580        let context = test_context();
581        let out_port = OutPort::<i32>::new(context.engine.top(), "tx");
582
583        assert!(Rc::ptr_eq(out_port.entity(), &out_port.entity));
584    }
585
586    #[test]
587    fn start_get_requires_connection_and_finish_get_wakes_putter() {
588        let context = test_context();
589        let mut in_port =
590            InPort::<i32>::new(&context.engine, &context.clock, context.engine.top(), "rx");
591
592        assert!(in_port.start_get().is_err());
593        assert!(in_port.state().is_ok());
594        assert!(in_port.start_get().is_ok());
595
596        let waker = noop_waker();
597        *in_port.state.waiting_put.borrow_mut() = Some(waker);
598        in_port.finish_get();
599
600        assert!(in_port.state.waiting_put.borrow().is_none());
601    }
602
603    #[test]
604    fn finish_get_without_waiting_putter_is_a_noop() {
605        let context = test_context();
606        let mut in_port =
607            InPort::<i32>::new(&context.engine, &context.clock, context.engine.top(), "rx");
608
609        in_port.finish_get();
610
611        assert!(in_port.state.waiting_put.borrow().is_none());
612    }
613
614    #[test]
615    fn port_put_waits_until_value_is_consumed_before_terminating() {
616        let state = test_state::<i32>();
617        let put = PortPut {
618            state: state.clone(),
619            value: Some(123),
620            done: false,
621        };
622        let mut put = Box::pin(put);
623        let waker = noop_waker();
624        let mut cx = Context::from_waker(&waker);
625
626        assert_eq!(put.as_mut().poll(&mut cx), Poll::Pending);
627        assert!(!put.is_terminated());
628        assert_eq!(*state.value.borrow(), Some(123));
629        assert!(state.waiting_put.borrow().is_some());
630
631        assert_eq!(put.as_mut().poll(&mut cx), Poll::Pending);
632        assert!(!put.is_terminated());
633
634        assert_eq!(state.value.borrow_mut().take(), Some(123));
635        *state.put_released.borrow_mut() = true;
636
637        assert_eq!(put.as_mut().poll(&mut cx), Poll::Ready(()));
638        assert!(put.is_terminated());
639    }
640
641    #[test]
642    fn port_put_waits_for_start_get_to_finish_before_terminating() {
643        let state = test_state::<i32>();
644        let put = PortPut {
645            state: state.clone(),
646            value: Some(123),
647            done: false,
648        };
649        let mut put = Box::pin(put);
650        let start_get = PortStartGet {
651            state: state.clone(),
652            done: false,
653        };
654        let mut start_get = Box::pin(start_get);
655        let waker = noop_waker();
656        let mut cx = Context::from_waker(&waker);
657
658        assert_eq!(put.as_mut().poll(&mut cx), Poll::Pending);
659        assert_eq!(start_get.as_mut().poll(&mut cx), Poll::Ready(123));
660        assert!(state.value.borrow().is_none());
661
662        assert_eq!(put.as_mut().poll(&mut cx), Poll::Pending);
663        assert!(!put.is_terminated());
664
665        *state.put_released.borrow_mut() = true;
666
667        assert_eq!(put.as_mut().poll(&mut cx), Poll::Ready(()));
668        assert!(put.is_terminated());
669    }
670
671    #[test]
672    fn port_try_put_waits_for_getter_then_completes() {
673        let state = test_state::<i32>();
674        let try_put = PortTryPut {
675            state: state.clone(),
676            done: false,
677        };
678        let mut try_put = Box::pin(try_put);
679        let waker = noop_waker();
680        let mut cx = Context::from_waker(&waker);
681
682        assert_eq!(try_put.as_mut().poll(&mut cx), Poll::Pending);
683        assert!(!try_put.is_terminated());
684        assert!(state.waiting_put.borrow().is_some());
685
686        *state.waiting_get.borrow_mut() = Some(noop_waker());
687
688        assert_eq!(try_put.as_mut().poll(&mut cx), Poll::Ready(()));
689        assert!(try_put.is_terminated());
690    }
691
692    #[test]
693    fn connected_out_port_creates_try_put_future() {
694        let context = test_context();
695        let mut out_port = OutPort::<i32>::new(context.engine.top(), "tx");
696        let in_port = InPort::new(&context.engine, &context.clock, context.engine.top(), "rx");
697
698        out_port.connect(in_port.state()).unwrap();
699
700        assert!(out_port.try_put().is_ok());
701    }
702
703    #[test]
704    fn port_get_waits_then_returns_value_and_reports_termination() {
705        let state = test_state::<i32>();
706        let get = PortGet {
707            state: state.clone(),
708            done: false,
709        };
710        let mut get = Box::pin(get);
711        let waker = noop_waker();
712        let mut cx = Context::from_waker(&waker);
713
714        assert_eq!(get.as_mut().poll(&mut cx), Poll::Pending);
715        assert!(!get.is_terminated());
716        assert!(state.waiting_get.borrow().is_some());
717
718        *state.value.borrow_mut() = Some(456);
719        *state.waiting_put.borrow_mut() = Some(noop_waker());
720
721        assert_eq!(get.as_mut().poll(&mut cx), Poll::Ready(456));
722        assert!(get.is_terminated());
723        assert!(state.waiting_put.borrow().is_none());
724    }
725
726    #[test]
727    fn port_get_pending_wakes_waiting_putter() {
728        let state = test_state::<i32>();
729        let get = PortGet {
730            state: state.clone(),
731            done: false,
732        };
733        let mut get = Box::pin(get);
734        let waker = noop_waker();
735        let mut cx = Context::from_waker(&waker);
736        *state.waiting_put.borrow_mut() = Some(noop_waker());
737
738        assert_eq!(get.as_mut().poll(&mut cx), Poll::Pending);
739
740        assert!(state.waiting_put.borrow().is_none());
741        assert!(state.waiting_get.borrow().is_some());
742    }
743
744    #[test]
745    fn port_get_samples_monitored_values() {
746        let state = monitored_test_state::<i32>();
747        let monitor = state
748            .monitor
749            .as_ref()
750            .expect("monitored state should create a monitor");
751        let get = PortGet {
752            state: state.clone(),
753            done: false,
754        };
755        let mut get = Box::pin(get);
756        let waker = noop_waker();
757        let mut cx = Context::from_waker(&waker);
758        *state.value.borrow_mut() = Some(456);
759
760        assert_eq!(get.as_mut().poll(&mut cx), Poll::Ready(456));
761        assert_eq!(monitor.bytes_in_window(), 456_i32.total_bytes());
762    }
763
764    #[test]
765    fn port_start_get_waits_then_returns_value_without_finishing_put() {
766        let state = test_state::<i32>();
767        let start_get = PortStartGet {
768            state: state.clone(),
769            done: false,
770        };
771        let mut start_get = Box::pin(start_get);
772        let waker = noop_waker();
773        let mut cx = Context::from_waker(&waker);
774
775        assert_eq!(start_get.as_mut().poll(&mut cx), Poll::Pending);
776        assert!(!start_get.is_terminated());
777        assert!(state.waiting_get.borrow().is_some());
778
779        let (waiting_put_wakes, waiting_put_waker) = counting_waker();
780        *state.waiting_put.borrow_mut() = Some(waiting_put_waker.clone());
781        *state.value.borrow_mut() = Some(789);
782
783        assert_eq!(start_get.as_mut().poll(&mut cx), Poll::Ready(789));
784        assert!(start_get.is_terminated());
785        assert!(state.waiting_get.borrow().is_none());
786        assert_eq!(waiting_put_wakes.load(Ordering::SeqCst), 0);
787    }
788
789    #[test]
790    fn late_start_get_wakes_try_put_readiness_without_acknowledging_payload() {
791        let state = test_state::<i32>();
792        let (producer_wakes, producer_waker) = counting_waker();
793        let mut producer_cx = Context::from_waker(&producer_waker);
794        let receiver_waker = noop_waker();
795        let mut receiver_cx = Context::from_waker(&receiver_waker);
796        let mut ready = Box::pin(PortTryPut {
797            state: state.clone(),
798            done: false,
799        });
800        assert_eq!(ready.as_mut().poll(&mut producer_cx), Poll::Pending);
801        let mut get = Box::pin(PortStartGet {
802            state: state.clone(),
803            done: false,
804        });
805        assert_eq!(get.as_mut().poll(&mut receiver_cx), Poll::Pending);
806        assert_eq!(
807            producer_wakes.load(Ordering::SeqCst),
808            1,
809            "late getter must publish readiness"
810        );
811        assert_eq!(ready.as_mut().poll(&mut producer_cx), Poll::Ready(()));
812        let mut put = Box::pin(PortPut {
813            state: state.clone(),
814            value: Some(37),
815            done: false,
816        });
817        assert_eq!(put.as_mut().poll(&mut producer_cx), Poll::Pending);
818        assert_eq!(get.as_mut().poll(&mut receiver_cx), Poll::Ready(37));
819        assert!(!*state.put_released.borrow());
820        assert_eq!(
821            put.as_mut().poll(&mut producer_cx),
822            Poll::Pending,
823            "readiness and taking the payload are not its ACK"
824        );
825        *state.put_released.borrow_mut() = true;
826        assert_eq!(put.as_mut().poll(&mut producer_cx), Poll::Ready(()));
827    }
828
829    #[test]
830    fn port_start_get_samples_monitored_values() {
831        let state = monitored_test_state::<i32>();
832        let monitor = state
833            .monitor
834            .as_ref()
835            .expect("monitored state should create a monitor");
836        let start_get = PortStartGet {
837            state: state.clone(),
838            done: false,
839        };
840        let mut start_get = Box::pin(start_get);
841        let waker = noop_waker();
842        let mut cx = Context::from_waker(&waker);
843        *state.value.borrow_mut() = Some(789);
844
845        assert_eq!(start_get.as_mut().poll(&mut cx), Poll::Ready(789));
846        assert_eq!(monitor.bytes_in_window(), 789_i32.total_bytes());
847    }
848}
gwr_engine/port/mod.rs

gwr_engine/port/
mod.rs
