feat(hydroflow): add initial test using Hydro CLI from Hydroflow+

This also required a change to Hydroflow core to make it possible to run the dataflow itself on a single thread (using a LocalSet), even if the surrounding runtime is not single-threaded (required to work around deadlocks because we can't use async APIs inside Hydroflow+). This requires us to spawn any Hydroflow tasks (only for `dest_sink` at the moment) right next to when we run the dataflow rather than when the Hydroflow graph is initialized. From a conceptual perspective, this seems _more right_, since now creating a Hydroflow program will not result in any actual tasks running.

In the third PR of this series, I aim to add a new Hydroflow+ operator that will automate the setup of a `dest_sink`/`source_stream` pair that span nodes.

.github/workflows/ci.yml

@@ -216,11 +216,6 @@ jobs:
           pip install maturin
           maturin develop
 
-      - name: Configure core dumps
-        run: |
-          mkdir cli-core-dumps
-          echo "$PWD/cli-core-dumps/corefile-%e-%p-%t" | sudo tee /proc/sys/kernel/core_pattern
-
       - name: Run Python tests
         run: |
           ulimit -c unlimited
@@ -229,24 +224,16 @@ jobs:
           cd python_tests
           pip install -r requirements.txt
           RUST_BACKTRACE=1 pytest *.py
+          cd ../..
 
-      - name: Print backtraces
-        if: ${{ failure() }}
+      - name: Run Hydroflow+ Python tests
         run: |
-          cd cli-core-dumps
-          sudo apt-get install gdb
-          for file in $(ls); do
-            echo "Backtrace for $file"
-            rust-gdb -ex "thread apply all bt" -batch $PWD/../.venv/bin/python $file
-          done
-          cd ..
-
-      - name: Upload core dumps
-        if: ${{ failure() }}
-        uses: actions/upload-artifact@master
-        with:
-          name: cli-core-dumps
-          path: ./cli-core-dumps/*
+          ulimit -c unlimited
+          cd hydro_cli
+          source .venv/bin/activate
+          cd ../hydroflow_plus_test/python_tests
+          pip install -r requirements.txt
+          RUST_BACKTRACE=1 pytest *.py
 
   lints:
     name: Lints

docs/docs/deploy/your-first-deploy.md

@@ -29,7 +29,7 @@ Let's open up `src/main.rs` in the generated project and write a new `main` func
 ```rust
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
 }
 ```
 
@@ -72,7 +72,7 @@ use hydroflow::hydroflow_syntax;
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
 
     let input_recv = ports
         .port("input")

hydro_cli_examples/examples/dedalus_2pc_coordinator/main.rs

@@ -4,7 +4,7 @@ use hydroflow_datalog::datalog;
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
     let vote_to_participant_port = ports
         .port("vote_to_participant")
         .connect::<ConnectedDemux<ConnectedDirect>>()

hydro_cli_examples/examples/dedalus_2pc_participant/main.rs

@@ -4,7 +4,7 @@ use hydroflow_datalog::datalog;
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
     let vote_to_participant_source = ports
         .port("vote_to_participant")
         .connect::<ConnectedDirect>()

hydro_cli_examples/examples/dedalus_receiver/main.rs

@@ -4,7 +4,7 @@ use hydroflow_datalog::datalog;
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
     let broadcast_recv = ports
         .port("broadcast")
         .connect::<ConnectedDirect>()

hydro_cli_examples/examples/dedalus_sender/main.rs

@@ -5,7 +5,7 @@ use hydroflow_datalog::datalog;
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
     let broadcast_port = ports
         .port("broadcast")
         .connect::<ConnectedDemux<ConnectedDirect>>()

hydro_cli_examples/examples/dedalus_vote_leader/main.rs

@@ -4,7 +4,7 @@ use hydroflow_datalog::datalog;
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
     let to_replica_port = ports
         .port("to_replica")
         .connect::<ConnectedDemux<ConnectedDirect>>()

hydro_cli_examples/examples/dedalus_vote_participant/main.rs

@@ -4,7 +4,7 @@ use hydroflow_datalog::datalog;
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
     let to_replica_source = ports
         .port("to_replica")
         .connect::<ConnectedDirect>()

hydro_cli_examples/examples/stdout_receiver/main.rs

@@ -3,7 +3,7 @@ use hydroflow::util::cli::{ConnectedDirect, ConnectedSource};
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
     let echo_recv = ports
         .port("echo")
         .connect::<ConnectedDirect>()

hydro_cli_examples/examples/tagged_stdout_receiver/main.rs

@@ -3,7 +3,7 @@ use hydroflow::util::cli::{ConnectedDirect, ConnectedSource, ConnectedTagged};
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
     let echo_recv = ports
         .port("echo")
         .connect::<ConnectedTagged<ConnectedDirect>>()

hydro_cli_examples/examples/ws_chat_server/main.rs

@@ -29,7 +29,7 @@ struct ChatMessage {
 
 #[hydroflow::main]
 async fn main() {
-    let mut ports = hydroflow::util::cli::init().await;
+    let ports = hydroflow::util::cli::init().await;
 
     let from_peer = ports
         .port("from_peer")

hydroflow/src/scheduled/context.rs

@@ -3,6 +3,7 @@
 use std::any::Any;
 use std::future::Future;
 use std::marker::PhantomData;
+use std::pin::Pin;
 
 use instant::Instant;
 use tokio::sync::mpsc::UnboundedSender;
@@ -37,6 +38,8 @@ pub struct Context {
     /// meaningless.
     pub(crate) subgraph_id: SubgraphId,
 
+    pub(crate) tasks_to_spawn: Vec<Pin<Box<dyn Future<Output = ()> + 'static>>>,
+
     /// Join handles for spawned tasks.
     pub(crate) task_join_handles: Vec<JoinHandle<()>>,
 }
@@ -155,23 +158,29 @@ impl Context {
             .expect("StateHandle wrong type T for casting.")
     }
 
-    /// Spawns an async task on the internal Tokio executor.
-    pub fn spawn_task<Fut>(&mut self, future: Fut)
+    /// Prepares an async task to be launched by [`Self::spawn_tasks`].
+    pub fn request_task<Fut>(&mut self, future: Fut)
     where
         Fut: Future<Output = ()> + 'static,
     {
-        self.task_join_handles
-            .push(tokio::task::spawn_local(future));
+        self.tasks_to_spawn.push(Box::pin(future));
+    }
+
+    /// Launches all tasks requested with [`Self::request_task`] on the internal Tokio executor.
+    pub fn spawn_tasks(&mut self) {
+        for task in self.tasks_to_spawn.drain(..) {
+            self.task_join_handles.push(tokio::task::spawn_local(task));
+        }
     }
 
-    /// Aborts all tasks spawned with [`Self::spawn_task`].
+    /// Aborts all tasks spawned with [`Self::spawn_tasks`].
     pub fn abort_tasks(&mut self) {
         for task in self.task_join_handles.drain(..) {
             task.abort();
         }
     }
 
-    /// Waits for all tasks spawned with [`Self::spawn_task`] to complete.
+    /// Waits for all tasks spawned with [`Self::spawn_tasks`] to complete.
     ///
     /// Will probably just hang.
     pub async fn join_tasks(&mut self) {

hydroflow/src/scheduled/graph.rs

@@ -61,6 +61,7 @@ impl<'a> Default for Hydroflow<'a> {
 
             subgraph_id: SubgraphId(0),
 
+            tasks_to_spawn: Vec::new(),
             task_join_handles: Vec::new(),
         };
         Self {
@@ -351,6 +352,7 @@ impl<'a> Hydroflow<'a> {
     /// TODO(mingwei): Currently blockes forever, no notion of "completion."
     #[tracing::instrument(level = "trace", skip(self), ret)]
     pub async fn run_async(&mut self) -> Option<Never> {
+        self.context.spawn_tasks();
         loop {
             // Run any work which is immediately available.
             self.run_available_async().await;
@@ -681,12 +683,12 @@ impl<'a> Hydroflow<'a> {
 }
 
 impl<'a> Hydroflow<'a> {
-    /// Alias for [`Context::spawn_task`].
-    pub fn spawn_task<Fut>(&mut self, future: Fut)
+    /// Alias for [`Context::request_task`].
+    pub fn request_task<Fut>(&mut self, future: Fut)
     where
         Fut: Future<Output = ()> + 'static,
     {
-        self.context.spawn_task(future);
+        self.context.request_task(future);
     }
 
     /// Alias for [`Context::abort_tasks`].

hydroflow/src/util/cli.rs

@@ -1,5 +1,6 @@
 #![allow(missing_docs)] // TODO(mingwei)
 
+use std::cell::RefCell;
 use std::collections::HashMap;
 
 pub use hydroflow_cli_integration::*;
@@ -15,19 +16,28 @@ pub async fn launch_flow(mut flow: Hydroflow<'_>) {
         stop.0.send(()).unwrap();
     });
 
+    let local_set = tokio::task::LocalSet::new();
+    let flow = local_set.run_until(async move {
+        flow.run_async().await;
+    });
+
     tokio::select! {
         _ = stop.1 => {},
-        _ = flow.run_async() => {}
+        _ = flow => {}
     }
 }
 
 pub struct HydroCLI {
-    ports: HashMap<String, ServerOrBound>,
+    ports: RefCell<HashMap<String, ServerOrBound>>,
 }
 
 impl HydroCLI {
-    pub fn port(&mut self, name: &str) -> ServerOrBound {
-        self.ports.remove(name).unwrap()
+    pub fn port(&self, name: &str) -> ServerOrBound {
+        self.ports
+            .try_borrow_mut()
+            .unwrap()
+            .remove(name)
+            .unwrap_or_else(|| panic!("port {} not found", name))
     }
 }
 
@@ -73,6 +83,6 @@ pub async fn init() -> HydroCLI {
     println!("ack start");
 
     HydroCLI {
-        ports: all_connected,
+        ports: RefCell::new(all_connected),
     }
 }

hydroflow_cli_integration/src/lib.rs

@@ -175,6 +175,12 @@ impl ServerOrBound {
         T::from_defn(self).await
     }
 
+    pub fn connect_local_blocking<T: Connected>(self) -> T {
+        let handle = tokio::runtime::Handle::current();
+        let _guard = handle.enter();
+        futures::executor::block_on(T::from_defn(self))
+    }
+
     pub async fn accept_tcp(&mut self) -> TcpStream {
         if let ServerOrBound::Bound(BoundConnection::TcpPort(handle, _)) = self {
             handle.recv().await.unwrap().unwrap()

hydroflow_lang/src/graph/ops/dest_sink.rs

@@ -139,7 +139,7 @@ pub const DEST_SINK: OperatorConstraints = OperatorConstraints {
                     }
                 }
                 #hydroflow
-                    .spawn_task(sink_feed_flush(#recv_ident, #sink_arg));
+                    .request_task(sink_feed_flush(#recv_ident, #sink_arg));
             }
         };
 

hydroflow_plus/src/stream.rs

@@ -1,6 +1,7 @@
 use std::hash::Hash;
 use std::marker::PhantomData;
 
+use hydroflow::futures::Sink;
 use proc_macro2::Span;
 use stageleft::{IntoQuotedMut, Quoted};
 use syn::parse_quote;
@@ -332,6 +333,22 @@ impl<'a, T> HfStream<'a, T> {
                 #ident = #self_ident -> for_each(#f);
             });
     }
+
+    pub fn dest_sink<S: Unpin + Sink<T> + 'a>(&self, sink: impl Quoted<'a, S>) {
+        let sink = sink.splice();
+        let self_ident = &self.ident;
+
+        self.graph
+            .builders
+            .borrow_mut()
+            .as_mut()
+            .unwrap()
+            .entry(self.node_id)
+            .or_default()
+            .add_statement(parse_quote! {
+                #self_ident -> dest_sink(#sink);
+            });
+    }
 }
 
 impl<'a, K, V1> HfStream<'a, (K, V1)> {

hydroflow_plus_test/Cargo.toml

@@ -5,7 +5,9 @@ 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" }
 regex = "1"
 serde = "1"

hydroflow_plus_test/examples/networked_basic.rs

@@ -0,0 +1,12 @@
+use hydroflow_plus_test::*;
+
+// cannot use hydroflow::main because connect_local_blocking causes a deadlock
+#[tokio::main]
+async fn main() {
+    let node_id: usize = std::env::args().nth(1).unwrap().parse().unwrap();
+    let ports = hydroflow::util::cli::init().await;
+
+    let joined = hydroflow_plus_test::networked::networked_basic!(&ports, node_id);
+
+    hydroflow::util::cli::launch_flow(joined).await;
+}