Merge c63fd4f into sapling-pr-archive-shadaj

docs/docs/hydroflow_plus/distributed.mdx

@@ -49,7 +49,7 @@ numbers
 Now that our graph spans multiple nodes, our runtime entrypoint will involve multiple subgraphs. This means we can't get away with `build_single`. Instead, we must take the subgraph ID as a runtime parameter (`subgraph_id`) to select the appropriate graph. In addition, our dataflow involves the network, so we take a `HydroCLI` runtime parameter (`cli`) so that nodes can look up their network connections and instantiate the flow graph with access to it.
 
 ```rust
-use hydroflow::util::cli::HydroCLI;
+use hydroflow_plus::util::cli::HydroCLI;
 use hydroflow_plus_cli_integration::{CLIRuntime, HydroflowPlusMeta};
 
 #[stageleft::entry]
@@ -67,10 +67,8 @@ The corresponding binary in `src/bin/first_ten_distributed.rs` then instantiates
 ```rust
 #[tokio::main]
 async fn main() {
-    let ports = hydroflow::util::cli::init().await;
-
-    hydroflow::util::cli::launch_flow(
-        flow::first_ten_distributed_runtime!(&ports)
+    hydroflow_plus::util::cli::launch(
+        |ports| flow::first_ten_distributed_runtime!(ports)
     ).await;
 }
 ```

hydroflow/src/util/cli.rs

@@ -8,6 +8,15 @@ use serde::de::DeserializeOwned;
 
 use crate::scheduled::graph::Hydroflow;
 
+pub async fn launch<T: DeserializeOwned + Default>(flow: impl FnOnce(&HydroCLI<T>) -> Hydroflow<'_>) {
+    let ports = init_no_ack_start::<T>().await;
+    let flow = flow(&ports);
+
+    println!("ack start");
+
+    launch_flow(flow).await;
+}
+
 pub async fn launch_flow(mut flow: Hydroflow<'_>) {
     let stop = tokio::sync::oneshot::channel();
     tokio::task::spawn_blocking(|| {
@@ -43,7 +52,7 @@ impl<T> HydroCLI<T> {
     }
 }
 
-pub async fn init<T: DeserializeOwned + Default>() -> HydroCLI<T> {
+async fn init_no_ack_start<T: DeserializeOwned + Default>() -> HydroCLI<T> {
     let mut input = String::new();
     std::io::stdin().read_line(&mut input).unwrap();
     let trimmed = input.trim();
@@ -82,8 +91,6 @@ pub async fn init<T: DeserializeOwned + Default>() -> HydroCLI<T> {
         all_connected.insert(name, ServerOrBound::Bound(defn));
     }
 
-    println!("ack start");
-
     HydroCLI {
         ports: RefCell::new(all_connected),
         meta: bind_config
@@ -92,3 +99,11 @@ pub async fn init<T: DeserializeOwned + Default>() -> HydroCLI<T> {
             .unwrap_or_default(),
     }
 }
+
+pub async fn init<T: DeserializeOwned + Default>() -> HydroCLI<T> {
+    let ret = init_no_ack_start::<T>().await;
+
+    println!("ack start");
+
+    ret
+}

hydroflow/src/util/mod.rs

@@ -2,6 +2,7 @@
 //! Helper utilities for the Hydroflow surface syntax.
 
 pub mod clear;
+#[cfg(feature = "hydroflow_macro")]
 pub mod demux_enum;
 pub mod monotonic_map;
 pub mod multiset;

hydroflow_plus/Cargo.toml

@@ -20,7 +20,7 @@ quote = "1.0.0"
 syn = { version = "2.0.0", features = [ "parsing", "extra-traits" ] }
 proc-macro2 = "1.0.57"
 proc-macro-crate = "1.1.0"
-hydroflow = { path = "../hydroflow", version = "^0.5.0" }
+hydroflow = { path = "../hydroflow", version = "^0.5.0", default-features = false }
 hydroflow_lang = { path = "../hydroflow_lang", version = "^0.5.0" }
 serde = { version = "1", features = [ "derive" ] }
 bincode = "1.3"

hydroflow_plus_test/Cargo.toml

@@ -5,7 +5,6 @@ version = "0.0.0"
 edition = "2021"
 
 [dependencies]
-hydroflow = { path = "../hydroflow", version = "^0.5.0", features = [ "cli_integration" ] }
 hydroflow_plus = { path = "../hydroflow_plus", version = "^0.5.0" }
 tokio = { version = "1.16", features = [ "full" ] }
 stageleft = { path = "../stageleft", version = "^0.1.0" }

hydroflow_plus_test/examples/networked_basic.rs

@@ -1,6 +1,6 @@
 use hydro_deploy::{Deployment, HydroflowCrate};
-use hydroflow::futures::SinkExt;
-use hydroflow::util::cli::ConnectedSink;
+use hydroflow_plus::futures::SinkExt;
+use hydroflow_plus::util::cli::ConnectedSink;
 use hydroflow_plus_cli_integration::CLIDeployNodeBuilder;
 
 #[tokio::main]

hydroflow_plus_test/src/bin/first_ten_distributed.rs

@@ -1,10 +1,8 @@
 // cannot use hydroflow::main because connect_local_blocking causes a deadlock
 #[tokio::main]
 async fn main() {
-    let ports = hydroflow::util::cli::init().await;
-
-    hydroflow::util::cli::launch_flow(
-        hydroflow_plus_test::first_ten::first_ten_distributed_runtime!(&ports),
+    hydroflow_plus::util::cli::launch(
+        |ports| hydroflow_plus_test::first_ten::first_ten_distributed_runtime!(ports),
     )
     .await;
 }

hydroflow_plus_test/src/bin/many_to_many.rs

@@ -1,8 +1,6 @@
 // cannot use hydroflow::main because connect_local_blocking causes a deadlock
 #[tokio::main]
 async fn main() {
-    let ports = hydroflow::util::cli::init().await;
-
-    hydroflow::util::cli::launch_flow(hydroflow_plus_test::cluster::many_to_many_runtime!(&ports))
+    hydroflow_plus::util::cli::launch(|ports| hydroflow_plus_test::cluster::many_to_many_runtime!(ports))
         .await;
 }

hydroflow_plus_test/src/bin/map_reduce.rs

@@ -4,8 +4,6 @@ extern crate alloc;
 // cannot use hydroflow::main because connect_local_blocking causes a deadlock
 #[tokio::main]
 async fn main() {
-    let ports = hydroflow::util::cli::init().await;
-
-    hydroflow::util::cli::launch_flow(hydroflow_plus_test::cluster::map_reduce_runtime!(&ports))
+    hydroflow_plus::util::cli::launch(|ports| hydroflow_plus_test::cluster::map_reduce_runtime!(ports))
         .await;
 }

hydroflow_plus_test/src/bin/networked_basic.rs

@@ -1,10 +1,8 @@
 // cannot use hydroflow::main because connect_local_blocking causes a deadlock
 #[tokio::main]
 async fn main() {
-    let ports = hydroflow::util::cli::init().await;
-
-    hydroflow::util::cli::launch_flow(hydroflow_plus_test::networked::networked_basic_runtime!(
-        &ports
+    hydroflow_plus::util::cli::launch(|ports| hydroflow_plus_test::networked::networked_basic_runtime!(
+        ports
     ))
     .await;
 }

hydroflow_plus_test/src/bin/simple_cluster.rs

@@ -1,10 +1,8 @@
 // cannot use hydroflow::main because connect_local_blocking causes a deadlock
 #[tokio::main]
 async fn main() {
-    let ports = hydroflow::util::cli::init().await;
-
-    hydroflow::util::cli::launch_flow(hydroflow_plus_test::cluster::simple_cluster_runtime!(
-        &ports
+    hydroflow_plus::util::cli::launch(|ports| hydroflow_plus_test::cluster::simple_cluster_runtime!(
+        ports
     ))
     .await;
 }

hydroflow_plus_test/src/cluster.rs

@@ -68,7 +68,7 @@ pub fn map_reduce<'a, D: Deploy<'a>>(
     (node, cluster)
 }
 
-use hydroflow::util::cli::HydroCLI;
+use hydroflow_plus::util::cli::HydroCLI;
 use hydroflow_plus_cli_integration::{CLIRuntime, HydroflowPlusMeta};
 
 #[stageleft::entry]
@@ -105,7 +105,6 @@ mod tests {
     use std::time::Duration;
 
     use hydro_deploy::{Deployment, HydroflowCrate};
-    use hydroflow::lattices::cc_traits::Iter;
     use hydroflow_plus_cli_integration::{
         CLIDeployClusterBuilder, CLIDeployNodeBuilder, DeployCrateWrapper,
     };

hydroflow_plus_test/src/first_ten.rs

@@ -34,7 +34,7 @@ pub fn first_ten_distributed<'a, D: Deploy<'a>>(
     second_node
 }
 
-use hydroflow::util::cli::HydroCLI;
+use hydroflow_plus::util::cli::HydroCLI;
 use hydroflow_plus_cli_integration::{CLIRuntime, HydroflowPlusMeta};
 
 #[stageleft::entry]

hydroflow_plus_test/src/networked.rs

@@ -1,5 +1,5 @@
-use hydroflow::bytes::BytesMut;
-use hydroflow::util::cli::HydroCLI;
+use hydroflow_plus::bytes::BytesMut;
+use hydroflow_plus::util::cli::HydroCLI;
 use hydroflow_plus::node::{Deploy, HfNode, NodeBuilder};
 use hydroflow_plus::scheduled::graph::Hydroflow;
 use hydroflow_plus::GraphBuilder;
@@ -43,8 +43,8 @@ mod tests {
     use std::time::Duration;
 
     use hydro_deploy::{Deployment, HydroflowCrate};
-    use hydroflow::futures::SinkExt;
-    use hydroflow::util::cli::ConnectedSink;
+    use hydroflow_plus::futures::SinkExt;
+    use hydroflow_plus::util::cli::ConnectedSink;
     use hydroflow_plus_cli_integration::{CLIDeployNodeBuilder, DeployCrateWrapper};
 
     #[tokio::test]

hydroflow_plus_test_macro/Cargo.toml

@@ -13,7 +13,6 @@ default = ["macro"]
 macro = []
 
 [dependencies]
-hydroflow = { path = "../hydroflow", version = "^0.5.0", features = [ "cli_integration" ] }
 hydroflow_plus = { path = "../hydroflow_plus", version = "^0.5.0" }
 tokio = { version = "1.16", features = [ "full" ] }
 stageleft = { path = "../stageleft", version = "^0.1.0" }