feat(hydroflow_plus): add API for cycle with initial value (#1430)

hydroflow_plus/src/builder/mod.rs

@@ -12,11 +12,11 @@ use quote::quote;
 use runtime_support::FreeVariable;
 use stageleft::*;
 
-use crate::cycle::CycleCollection;
+use crate::cycle::{CycleCollection, CycleCollectionWithInitial};
 use crate::ir::{HfPlusLeaf, HfPlusNode, HfPlusSource};
 use crate::location::{Cluster, Location, LocationId, Process};
 use crate::stream::{Bounded, NoTick, Tick, Unbounded};
-use crate::{HfCycle, Optional, RuntimeContext, Stream};
+use crate::{HfCycle, Optional, RuntimeContext, Singleton, Stream};
 
 pub mod built;
 pub mod deploy;
@@ -247,6 +247,47 @@ impl<'a> FlowBuilder<'a> {
         )
     }
 
+    pub fn singleton<T: Clone, L: Location>(
+        &self,
+        on: &L,
+        e: impl Quoted<'a, T>,
+    ) -> Singleton<'a, T, Bounded, NoTick, L> {
+        let e_arr = q!([e]);
+        let e = e_arr.splice();
+
+        // we do a double persist here because if the singleton shows up on every tick,
+        // we first persist the source so that we store that value and then persist again
+        // so that it grows every tick
+        Singleton::new(
+            on.id(),
+            self.ir_leaves().clone(),
+            HfPlusNode::Persist(Box::new(HfPlusNode::Persist(Box::new(
+                HfPlusNode::Source {
+                    source: HfPlusSource::Iter(e.into()),
+                    location_kind: on.id(),
+                },
+            )))),
+        )
+    }
+
+    pub fn singleton_first_tick<T: Clone, L: Location>(
+        &self,
+        on: &L,
+        e: impl Quoted<'a, T>,
+    ) -> Optional<'a, T, Bounded, Tick, L> {
+        let e_arr = q!([e]);
+        let e = e_arr.splice();
+
+        Optional::new(
+            on.id(),
+            self.ir_leaves().clone(),
+            HfPlusNode::Source {
+                source: HfPlusSource::Iter(e.into()),
+                location_kind: on.id(),
+            },
+        )
+    }
+
     pub fn source_interval<L: Location>(
         &self,
         on: &L,
@@ -306,4 +347,34 @@ impl<'a> FlowBuilder<'a> {
             S::create_source(ident, self.ir_leaves.clone(), on.id()),
         )
     }
+
+    pub fn cycle_with_initial<S: CycleCollectionWithInitial<'a>>(
+        &self,
+        on: &S::Location,
+        initial: S,
+    ) -> (HfCycle<'a, S>, S) {
+        let next_id = {
+            let on_id = match on.id() {
+                LocationId::Process(id) => id,
+                LocationId::Cluster(id) => id,
+            };
+
+            let mut cycle_ids = self.cycle_ids.borrow_mut();
+            let next_id_entry = cycle_ids.entry(on_id).or_default();
+
+            let id = *next_id_entry;
+            *next_id_entry += 1;
+            id
+        };
+
+        let ident = syn::Ident::new(&format!("cycle_{}", next_id), Span::call_site());
+
+        (
+            HfCycle {
+                ident: ident.clone(),
+                _phantom: PhantomData,
+            },
+            S::create_source(ident, self.ir_leaves.clone(), initial, on.id()),
+        )
+    }
 }

hydroflow_plus/src/cycle.rs

@@ -3,24 +3,37 @@ use std::marker::PhantomData;
 use crate::builder::FlowLeaves;
 use crate::location::{Location, LocationId};
 
-pub trait CycleCollection<'a> {
+pub trait CycleComplete<'a> {
+    fn complete(self, ident: syn::Ident);
+}
+
+pub trait CycleCollection<'a>: CycleComplete<'a> {
     type Location: Location;
 
     fn create_source(ident: syn::Ident, ir_leaves: FlowLeaves<'a>, l: LocationId) -> Self;
+}
 
-    fn complete(self, ident: syn::Ident);
+pub trait CycleCollectionWithInitial<'a>: CycleComplete<'a> {
+    type Location: Location;
+
+    fn create_source(
+        ident: syn::Ident,
+        ir_leaves: FlowLeaves<'a>,
+        initial: Self,
+        l: LocationId,
+    ) -> Self;
 }
 
 /// Represents a fixpoint cycle in the graph that will be fulfilled
 /// by a stream that is not yet known.
 ///
 /// See [`crate::FlowBuilder`] for an explainer on the type parameters.
-pub struct HfCycle<'a, S: CycleCollection<'a>> {
+pub struct HfCycle<'a, S: CycleComplete<'a>> {
     pub(crate) ident: syn::Ident,
     pub(crate) _phantom: PhantomData<(&'a mut &'a (), S)>,
 }
 
-impl<'a, S: CycleCollection<'a>> HfCycle<'a, S> {
+impl<'a, S: CycleComplete<'a>> HfCycle<'a, S> {
     pub fn complete(self, stream: S) {
         let ident = self.ident;
         S::complete(stream, ident)