feat: add decouple and simple test and two_pc

.gitignore

@@ -1,3 +1,4 @@
+.vscode/
 /target
 /rust.git
 /*.dot

hydroflow_plus/src/stream.rs

@@ -14,13 +14,13 @@
 use stageleft::{q, IntoQuotedMut, Quoted};
 use syn::parse_quote;
 
-use crate::builder::FlowState;
+use crate::builder::{self, ClusterIds, FlowLeaves, FlowState};
 use crate::cycle::{CycleCollection, CycleComplete};
 use crate::ir::{DebugInstantiate, HfPlusLeaf, HfPlusNode, HfPlusSource};
 use crate::location::{
     CanSend, ExternalBincodeStream, ExternalBytesPort, ExternalProcess, Location, LocationId,
 };
-use crate::{Cluster, Optional, Singleton};
+use crate::{Cluster, Optional, Process, Singleton};
 
 /// Marks the stream as being unbounded, which means that it is not
 /// guaranteed to be complete in finite time.
@@ -690,6 +690,42 @@
 }
 
 impl<'a, T, W, N: Location<'a>> Stream<T, W, NoTick, N> {
+    pub fn decouple_process<P2>(
+        self,
+        other: &Process<P2>,
+    ) -> Stream<'a, T, Unbounded, NoTick, Process<P2>>
+    where
+        N: CanSend<Process<P2>, In<T> = T, Out<T> = T>,
+        T: Clone + Serialize + DeserializeOwned,
+    {
+        match self.location_kind {
+            LocationId::Process(_) => (),
+            _ => panic!("decouple_process must be called on a process"),
+        };
+        self.send_bincode::<Process<P2>, T>(other)
+    }
+
+    pub fn decouple_cluster<C2>(
+        self,
+        other: &Cluster<C2>,
+    ) -> Stream<N::Out<T>, Unbounded, NoTick, Cluster<'a, C2>>
+    where
+        N: CanSend<'a, Cluster<'a, C2>, In<T> = (u32, T)>,
+        T: Clone + Serialize + DeserializeOwned,
+    {
+        // Get the self node ID within the cluster
+        let self_node_id = match self.location_kind {
+            LocationId::Cluster(cluster_id) => builder::ClusterSelfId {
+                id: cluster_id,
+                _phantom: PhantomData,
+            },
+            _ => panic!("decouple_cluster must be called on a cluster"),
+        };
+
+        self.map(q!(move |b| (self_node_id, b.clone(),)))
+            .send_bincode(other)
+    }
+
     pub fn send_bincode<N2: Location<'a>, CoreType>(
         self,
         other: &N2,
@@ -847,8 +883,10 @@
         N: CanSend<'a, Cluster<'a, C2>, In<T> = (u32, T)>,
         T: Clone + Serialize + DeserializeOwned,
     {
-        let ids = other.members();
-
+        let ids = ClusterIds::<'a> {
+            id: other.id,
+            _phantom: PhantomData,
+        };
         self.flat_map(q!(|b| ids.iter().map(move |id| (
             ::std::clone::Clone::clone(id),
             ::std::clone::Clone::clone(&b)

hydroflow_plus_test/examples/two_pc.rs

@@ -0,0 +1,34 @@
+use hydro_deploy::Deployment;
+use hydroflow_plus::deploy::TrybuildHost;
+
+#[tokio::main]
+async fn main() {
+    let mut deployment = Deployment::new();
+    let localhost = deployment.Localhost();
+
+    let builder = hydroflow_plus::FlowBuilder::new();
+    let num_participants: u32 = 3;
+
+    let (coordinator, participants, client) =
+        hydroflow_plus_test::cluster::two_pc::two_pc(&builder, num_participants);
+
+    let rustflags = "-C opt-level=3 -C codegen-units=1 -C strip=none -C debuginfo=2 -C lto=off";
+
+    let _nodes = builder
+        .with_default_optimize()
+        .with_process(&coordinator, TrybuildHost::new(deployment.Localhost()))
+        .with_cluster(
+            &participants,
+            (0..num_participants)
+                .map(|_| TrybuildHost::new(deployment.Localhost()))
+                .collect::<Vec<_>>(),
+        )
+        .with_process(&client, TrybuildHost::new(deployment.Localhost()))
+        .deploy(&mut deployment);
+
+    deployment.deploy().await.unwrap();
+
+    deployment.start().await.unwrap();
+
+    tokio::signal::ctrl_c().await.unwrap();
+}

hydroflow_plus_test/src/cluster/mod.rs

@@ -4,3 +4,4 @@ pub mod map_reduce;
 pub mod paxos;
 pub mod paxos_bench;
 pub mod simple_cluster;
+pub mod two_pc;