gwr_models/
fc_pipeline.rs

1// Copyright (c) 2023 Graphcore Ltd. All rights reserved.
2
3//! Flow Controlled Pipeline.
4//!
5//! This is a pipeline that has a buffer at one end that emits credits to the
6//! other end. There is a latency for data values to travel down the pipeline
7//! and a different latency can be configured for the credits to travel back to
8//! the input. The size of the buffer is also configurable. For maximum
9//! throughput, the buffer should be large enough to overcome the round trip
10//! latency of the credit loop.
11//!
12//! # Ports
13//!
14//! This component has two ports:
15//!  - One [input port](gwr_engine::port::InPort): `rx`
16//!  - One [output port](gwr_engine::port::OutPort): `tx`
17
18use std::cell::RefCell;
19use std::rc::Rc;
20
21use async_trait::async_trait;
22use gwr_components::delay::Delay;
23use gwr_components::flow_controls::credit_issuer::CreditIssuer;
24use gwr_components::flow_controls::credit_limiter::CreditLimiter;
25use gwr_components::store::ObjectStore;
26use gwr_components::types::Credit;
27use gwr_components::{connect_port, connect_tx, port_rx};
28use gwr_engine::engine::Engine;
29use gwr_engine::port::PortStateResult;
30use gwr_engine::time::clock::Clock;
31use gwr_engine::traits::SimObject;
32use gwr_engine::types::{SimError, SimResult};
33use gwr_model_builder::{EntityDisplay, EntityGet, Runnable};
34use gwr_track::build_aka;
35use gwr_track::entity::Entity;
36use gwr_track::tracker::aka::Aka;
37
38/// Configuration for a flow-controlled pipeline.
39pub struct FcPipelineConfig {
40    buffer_size: usize,
41    data_delay_ticks: usize,
42    credit_delay_ticks: usize,
43}
44
45impl FcPipelineConfig {
46    #[must_use]
47    pub fn new(buffer_size: usize, data_delay_ticks: usize, credit_delay_ticks: usize) -> Self {
48        Self {
49            buffer_size,
50            data_delay_ticks,
51            credit_delay_ticks,
52        }
53    }
54}
55
56/// The Flow-Controlled Pipeline.
57#[derive(EntityGet, EntityDisplay, Runnable)]
58pub struct FcPipeline<T>
59where
60    T: SimObject,
61{
62    entity: Rc<Entity>,
63    credit_limiter: RefCell<Option<Rc<CreditLimiter<T>>>>,
64    credit_delay: RefCell<Option<Rc<Delay<Credit>>>>,
65    credit_issuer: RefCell<Option<Rc<CreditIssuer<T>>>>,
66    data_delay: RefCell<Option<Rc<Delay<T>>>>,
67}
68
69impl<T> FcPipeline<T>
70where
71    T: SimObject,
72{
73    pub fn new_and_register_with_renames(
74        engine: &Engine,
75        clock: &Clock,
76        parent: &Rc<Entity>,
77        name: &str,
78        aka: Option<&Aka>,
79        config: &FcPipelineConfig,
80    ) -> Result<Rc<Self>, SimError> {
81        let entity = Rc::new(Entity::new(parent, name));
82
83        let credit_limiter_aka = build_aka!(aka, &entity, &[("rx", "rx")]);
84        let credit_limiter = CreditLimiter::new_and_register(
85            engine,
86            clock,
87            &entity,
88            "credit_limiter",
89            Some(&credit_limiter_aka),
90            config.buffer_size,
91        );
92
93        let data_delay =
94            Delay::new_and_register(engine, clock, &entity, "pipe", config.data_delay_ticks);
95        // The whole point of the flow-controlled pipeline is that the delays should
96        // never have to stall at their outputs
97        data_delay.set_error_on_output_stall();
98
99        let buffer =
100            ObjectStore::new_and_register(engine, clock, &entity, "buf", config.buffer_size)?;
101
102        connect_port!(credit_limiter, tx => data_delay, rx)
103            .expect("Internal ports should connect without error");
104        connect_port!(data_delay, tx => buffer, rx)
105            .expect("Internal ports should connect without error");
106
107        let credit_issuer_aka = build_aka!(aka, &entity, &[("tx", "tx")]);
108        let credit_issuer = CreditIssuer::new_and_register_with_renames(
109            engine,
110            clock,
111            &entity,
112            "credit_issuer",
113            Some(&credit_issuer_aka),
114        );
115        let credit_delay = Delay::new_and_register(
116            engine,
117            clock,
118            &entity,
119            "credit_pipe",
120            config.credit_delay_ticks,
121        );
122        // The whole point of the flow-controlled pipeline is that the delays should
123        // never have to stall at their outputs
124        credit_delay.set_error_on_output_stall();
125
126        connect_port!(buffer, tx => credit_issuer, rx)
127            .expect("Internal ports should connect without error");
128        connect_port!(credit_issuer, credit_tx => credit_delay, rx)
129            .expect("Internal ports should connect without error");
130        connect_port!(credit_delay, tx => credit_limiter, credit_rx)
131            .expect("Internal ports should connect without error");
132
133        let rc_self = Rc::new(Self {
134            entity,
135            credit_limiter: RefCell::new(Some(credit_limiter)),
136            credit_delay: RefCell::new(Some(credit_delay)),
137            credit_issuer: RefCell::new(Some(credit_issuer)),
138            data_delay: RefCell::new(Some(data_delay)),
139        });
140        engine.register(rc_self.clone());
141        Ok(rc_self)
142    }
143
144    pub fn new_and_register(
145        engine: &Engine,
146        clock: &Clock,
147        parent: &Rc<Entity>,
148        name: &str,
149        config: &FcPipelineConfig,
150    ) -> Result<Rc<Self>, SimError> {
151        Self::new_and_register_with_renames(engine, clock, parent, name, None, config)
152    }
153
154    pub fn set_data_delay(&self, delay: usize) -> SimResult {
155        self.data_delay.borrow().as_ref().unwrap().set_delay(delay)
156    }
157
158    pub fn set_credit_delay(&self, delay: usize) -> SimResult {
159        self.credit_delay
160            .borrow()
161            .as_ref()
162            .unwrap()
163            .set_delay(delay)
164    }
165
166    pub fn connect_port_tx(&self, port_state: PortStateResult<T>) -> SimResult {
167        connect_tx!(self.credit_issuer, connect_port_tx ; port_state)
168    }
169
170    pub fn port_rx(&self) -> PortStateResult<T> {
171        port_rx!(self.credit_limiter, port_rx)
172    }
173}
gwr_models/fc_pipeline.rs

gwr_models/
fc_pipeline.rs
