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            *self.state.waiting_get.borrow_mut() = Some(cx.waker().clone());
437            Poll::Pending
438        }
439    }
440}
441
442impl<T> FusedFuture for PortStartGet<T>
443where
444    T: SimObject,
445{
446    fn is_terminated(&self) -> bool {
447        self.done
448    }
449}
450
451#[cfg(test)]
452mod tests {
453    use std::sync::Arc;
454    use std::sync::atomic::{AtomicUsize, Ordering};
455    use std::task::{Wake, Waker};
456
457    use futures::future::FusedFuture;
458    use futures::task::noop_waker;
459    use gwr_track::Tracker;
460    use gwr_track::entity::Entity;
461    use gwr_track::tracker::dev_null_tracker;
462
463    use super::*;
464    use crate::traits::TotalBytes;
465
466    struct TestContext {
467        // Just kept to ensure it isn't dropped
468        _tracker: Tracker,
469        engine: Engine,
470        clock: Clock,
471    }
472
473    fn test_context() -> TestContext {
474        let tracker = dev_null_tracker();
475        let mut engine = Engine::new(&tracker);
476        let clock = engine.default_clock();
477
478        TestContext {
479            _tracker: tracker,
480            engine,
481            clock,
482        }
483    }
484
485    fn test_state<T: SimObject>() -> Rc<PortState<T>> {
486        let context = test_context();
487        let entity = Rc::new(Entity::new(context.engine.top(), "rx"));
488
489        Rc::new(PortState::new(
490            &context.engine,
491            &context.clock,
492            entity,
493            None,
494        ))
495    }
496
497    fn monitored_test_state<T: SimObject>() -> Rc<PortState<T>> {
498        let context = test_context();
499        let entity = Rc::new(Entity::new(context.engine.top(), "rx"));
500
501        Rc::new(PortState::new(
502            &context.engine,
503            &context.clock,
504            entity,
505            Some(1),
506        ))
507    }
508
509    struct WakeCounter {
510        wakes_count: Arc<AtomicUsize>,
511    }
512
513    impl Wake for WakeCounter {
514        fn wake(self: Arc<Self>) {
515            self.wakes_count.fetch_add(1, Ordering::SeqCst);
516        }
517
518        fn wake_by_ref(self: &Arc<Self>) {
519            self.wakes_count.fetch_add(1, Ordering::SeqCst);
520        }
521    }
522
523    fn counting_waker() -> (Arc<AtomicUsize>, Waker) {
524        let wakes_count = Arc::new(AtomicUsize::new(0));
525        let waker = Waker::from(Arc::new(WakeCounter {
526            wakes_count: wakes_count.clone(),
527        }));
528
529        (wakes_count, waker)
530    }
531
532    #[test]
533    fn wake_counter_counts_wake_and_wake_by_ref() {
534        let (wakes_count, waker) = counting_waker();
535
536        waker.wake_by_ref();
537        assert_eq!(wakes_count.load(Ordering::SeqCst), 1);
538
539        waker.wake();
540        assert_eq!(wakes_count.load(Ordering::SeqCst), 2);
541    }
542
543    #[test]
544    fn in_port_state_can_only_connect_once() {
545        let context = test_context();
546        let in_port =
547            InPort::<i32>::new(&context.engine, &context.clock, context.engine.top(), "rx");
548
549        assert!(in_port.state().is_ok());
550
551        let err = in_port
552            .state()
553            .err()
554            .expect("second state call should fail");
555        assert!(format!("{err}").contains("already connected"));
556    }
557
558    #[test]
559    fn out_port_connect_can_only_connect_once() {
560        let context = test_context();
561        let mut out_port = OutPort::<i32>::new(context.engine.top(), "tx");
562        let first_in_port =
563            InPort::new(&context.engine, &context.clock, context.engine.top(), "rx1");
564        let second_in_port =
565            InPort::new(&context.engine, &context.clock, context.engine.top(), "rx2");
566
567        out_port.connect(first_in_port.state()).unwrap();
568
569        let err = out_port.connect(second_in_port.state()).unwrap_err();
570        assert!(format!("{err}").contains("already connected"));
571    }
572
573    #[test]
574    fn out_port_entity_returns_port_entity() {
575        let context = test_context();
576        let out_port = OutPort::<i32>::new(context.engine.top(), "tx");
577
578        assert!(Rc::ptr_eq(out_port.entity(), &out_port.entity));
579    }
580
581    #[test]
582    fn start_get_requires_connection_and_finish_get_wakes_putter() {
583        let context = test_context();
584        let mut in_port =
585            InPort::<i32>::new(&context.engine, &context.clock, context.engine.top(), "rx");
586
587        assert!(in_port.start_get().is_err());
588        assert!(in_port.state().is_ok());
589        assert!(in_port.start_get().is_ok());
590
591        let waker = noop_waker();
592        *in_port.state.waiting_put.borrow_mut() = Some(waker);
593        in_port.finish_get();
594
595        assert!(in_port.state.waiting_put.borrow().is_none());
596    }
597
598    #[test]
599    fn finish_get_without_waiting_putter_is_a_noop() {
600        let context = test_context();
601        let mut in_port =
602            InPort::<i32>::new(&context.engine, &context.clock, context.engine.top(), "rx");
603
604        in_port.finish_get();
605
606        assert!(in_port.state.waiting_put.borrow().is_none());
607    }
608
609    #[test]
610    fn port_put_waits_until_value_is_consumed_before_terminating() {
611        let state = test_state::<i32>();
612        let put = PortPut {
613            state: state.clone(),
614            value: Some(123),
615            done: false,
616        };
617        let mut put = Box::pin(put);
618        let waker = noop_waker();
619        let mut cx = Context::from_waker(&waker);
620
621        assert_eq!(put.as_mut().poll(&mut cx), Poll::Pending);
622        assert!(!put.is_terminated());
623        assert_eq!(*state.value.borrow(), Some(123));
624        assert!(state.waiting_put.borrow().is_some());
625
626        assert_eq!(put.as_mut().poll(&mut cx), Poll::Pending);
627        assert!(!put.is_terminated());
628
629        assert_eq!(state.value.borrow_mut().take(), Some(123));
630        *state.put_released.borrow_mut() = true;
631
632        assert_eq!(put.as_mut().poll(&mut cx), Poll::Ready(()));
633        assert!(put.is_terminated());
634    }
635
636    #[test]
637    fn port_put_waits_for_start_get_to_finish_before_terminating() {
638        let state = test_state::<i32>();
639        let put = PortPut {
640            state: state.clone(),
641            value: Some(123),
642            done: false,
643        };
644        let mut put = Box::pin(put);
645        let start_get = PortStartGet {
646            state: state.clone(),
647            done: false,
648        };
649        let mut start_get = Box::pin(start_get);
650        let waker = noop_waker();
651        let mut cx = Context::from_waker(&waker);
652
653        assert_eq!(put.as_mut().poll(&mut cx), Poll::Pending);
654        assert_eq!(start_get.as_mut().poll(&mut cx), Poll::Ready(123));
655        assert!(state.value.borrow().is_none());
656
657        assert_eq!(put.as_mut().poll(&mut cx), Poll::Pending);
658        assert!(!put.is_terminated());
659
660        *state.put_released.borrow_mut() = true;
661
662        assert_eq!(put.as_mut().poll(&mut cx), Poll::Ready(()));
663        assert!(put.is_terminated());
664    }
665
666    #[test]
667    fn port_try_put_waits_for_getter_then_completes() {
668        let state = test_state::<i32>();
669        let try_put = PortTryPut {
670            state: state.clone(),
671            done: false,
672        };
673        let mut try_put = Box::pin(try_put);
674        let waker = noop_waker();
675        let mut cx = Context::from_waker(&waker);
676
677        assert_eq!(try_put.as_mut().poll(&mut cx), Poll::Pending);
678        assert!(!try_put.is_terminated());
679        assert!(state.waiting_put.borrow().is_some());
680
681        *state.waiting_get.borrow_mut() = Some(noop_waker());
682
683        assert_eq!(try_put.as_mut().poll(&mut cx), Poll::Ready(()));
684        assert!(try_put.is_terminated());
685    }
686
687    #[test]
688    fn connected_out_port_creates_try_put_future() {
689        let context = test_context();
690        let mut out_port = OutPort::<i32>::new(context.engine.top(), "tx");
691        let in_port = InPort::new(&context.engine, &context.clock, context.engine.top(), "rx");
692
693        out_port.connect(in_port.state()).unwrap();
694
695        assert!(out_port.try_put().is_ok());
696    }
697
698    #[test]
699    fn port_get_waits_then_returns_value_and_reports_termination() {
700        let state = test_state::<i32>();
701        let get = PortGet {
702            state: state.clone(),
703            done: false,
704        };
705        let mut get = Box::pin(get);
706        let waker = noop_waker();
707        let mut cx = Context::from_waker(&waker);
708
709        assert_eq!(get.as_mut().poll(&mut cx), Poll::Pending);
710        assert!(!get.is_terminated());
711        assert!(state.waiting_get.borrow().is_some());
712
713        *state.value.borrow_mut() = Some(456);
714        *state.waiting_put.borrow_mut() = Some(noop_waker());
715
716        assert_eq!(get.as_mut().poll(&mut cx), Poll::Ready(456));
717        assert!(get.is_terminated());
718        assert!(state.waiting_put.borrow().is_none());
719    }
720
721    #[test]
722    fn port_get_pending_wakes_waiting_putter() {
723        let state = test_state::<i32>();
724        let get = PortGet {
725            state: state.clone(),
726            done: false,
727        };
728        let mut get = Box::pin(get);
729        let waker = noop_waker();
730        let mut cx = Context::from_waker(&waker);
731        *state.waiting_put.borrow_mut() = Some(noop_waker());
732
733        assert_eq!(get.as_mut().poll(&mut cx), Poll::Pending);
734
735        assert!(state.waiting_put.borrow().is_none());
736        assert!(state.waiting_get.borrow().is_some());
737    }
738
739    #[test]
740    fn port_get_samples_monitored_values() {
741        let state = monitored_test_state::<i32>();
742        let monitor = state
743            .monitor
744            .as_ref()
745            .expect("monitored state should create a monitor");
746        let get = PortGet {
747            state: state.clone(),
748            done: false,
749        };
750        let mut get = Box::pin(get);
751        let waker = noop_waker();
752        let mut cx = Context::from_waker(&waker);
753        *state.value.borrow_mut() = Some(456);
754
755        assert_eq!(get.as_mut().poll(&mut cx), Poll::Ready(456));
756        assert_eq!(monitor.bytes_in_window(), 456_i32.total_bytes());
757    }
758
759    #[test]
760    fn port_start_get_waits_then_returns_value_without_finishing_put() {
761        let state = test_state::<i32>();
762        let start_get = PortStartGet {
763            state: state.clone(),
764            done: false,
765        };
766        let mut start_get = Box::pin(start_get);
767        let waker = noop_waker();
768        let mut cx = Context::from_waker(&waker);
769
770        assert_eq!(start_get.as_mut().poll(&mut cx), Poll::Pending);
771        assert!(!start_get.is_terminated());
772        assert!(state.waiting_get.borrow().is_some());
773
774        let (waiting_put_wakes, waiting_put_waker) = counting_waker();
775        *state.waiting_put.borrow_mut() = Some(waiting_put_waker.clone());
776        *state.value.borrow_mut() = Some(789);
777
778        assert_eq!(start_get.as_mut().poll(&mut cx), Poll::Ready(789));
779        assert!(start_get.is_terminated());
780        assert!(state.waiting_get.borrow().is_none());
781        assert_eq!(waiting_put_wakes.load(Ordering::SeqCst), 0);
782    }
783
784    #[test]
785    fn port_start_get_samples_monitored_values() {
786        let state = monitored_test_state::<i32>();
787        let monitor = state
788            .monitor
789            .as_ref()
790            .expect("monitored state should create a monitor");
791        let start_get = PortStartGet {
792            state: state.clone(),
793            done: false,
794        };
795        let mut start_get = Box::pin(start_get);
796        let waker = noop_waker();
797        let mut cx = Context::from_waker(&waker);
798        *state.value.borrow_mut() = Some(789);
799
800        assert_eq!(start_get.as_mut().poll(&mut cx), Poll::Ready(789));
801        assert_eq!(monitor.bytes_in_window(), 789_i32.total_bytes());
802    }
803}
gwr_engine/port/mod.rs

gwr_engine/port/
mod.rs
