-#shell-command-next-line
-rustup update
-#shell-command-next-line
-cargo test
-```
-
-:::
-
-
-Let's look a minimal example of a Hydroflow+ program. We'll start with a simple flow graph that prints out the first 10 natural numbers. First, we'll define the **flow graph**.
-
-
-## The Flow Graph
-
-{getLines(firstTenSrc, 1, 17)}
-
-To build a Hydroflow+ application, we need to define a dataflow that spans multiple processes. The `FlowBuilder` parameter captures the global dataflow, and we can instantiate processes to define boundaries between distributed logic. When defining a process, we also pass in a type parameter to a "tag" that identifies the process. When transforming streams, the Rust type system will guarantee that we are operating on streams on the same process.
-
-{getLines(firstTenSrc, 8, 9)}
-
-Now, we can build out the dataflow to run on this process. Every dataflow starts at a source that is bound to a specific process. First, we instantiate a stream that emits the first 10 natural numbers.
-
-{getLines(firstTenSrc, 11)}
-
-In Hydroflow+, whenever there are snippets of Rust code passed to operators (like `source_iter`, `map`, or `for_each`), we use the `q!` macro to mark them. For example, we may use Rust snippets to define static sources of data or closures that transform them.
-
-To print out these numbers, we can use the `for_each` operator (note that the body of `for_each` is a closure wrapped in `q!`):
-
-{getLines(firstTenSrc, 12, 14)}
-
-In the next section, we will look at how to deploy this program to run on multiple processs.
diff --git a/docs/docs/hydroflow_plus/stageleft.mdx b/docs/docs/hydroflow_plus/stageleft.mdx
index f2a4ef062ce2..e4ac5014c9e8 100644
--- a/docs/docs/hydroflow_plus/stageleft.mdx
+++ b/docs/docs/hydroflow_plus/stageleft.mdx
@@ -1,8 +1,10 @@
---
title: Stageleft
-sidebar_position: 6
+sidebar_position: 3
---
import StageleftDocs from '../../../stageleft/README.md'
+Under the hood, Hydroflow+ uses a library called Stageleft to power the `q!` macro and code generation logic. The following docs, from the Stageleft README, outline the core architecture of Stageleft.
+
{paper.conf}
-
+
{paper["authors"]}
- {paper.description[0]} {paper.description[1]}
+ {paper.description[0]} {paper.description[1]}
{paper.links}
diff --git a/docs/src/theme/prism-include-languages.js b/docs/src/theme/prism-include-languages.js
index 827cb9065199..c8fe35db80e1 100644
--- a/docs/src/theme/prism-include-languages.js
+++ b/docs/src/theme/prism-include-languages.js
@@ -4,27 +4,25 @@ export default function prismIncludeLanguages(PrismObject) {
themeConfig: {prism},
} = siteConfig;
const {additionalLanguages} = prism;
- // Prism components work on the Prism instance on the window, while prism-
- // react-renderer uses its own Prism instance. We temporarily mount the
- // instance onto window, import components to enhance it, then remove it to
- // avoid polluting global namespace.
- // You can mutate PrismObject: registering plugins, deleting languages... As
- // long as you don't re-assign it
+
+ const PrismBefore = globalThis.Prism;
globalThis.Prism = PrismObject;
additionalLanguages.forEach((lang) => {
// eslint-disable-next-line global-require, import/no-dynamic-require
require(`prismjs/components/prism-${lang}`);
});
- const rustLanguage = Prism.languages.rust;
Prism.languages["rust,ignore"] = Prism.languages.rust;
const origTokenize = PrismObject.tokenize;
- PrismObject.tokenize = (text, grammar) => {
- if (grammar == rustLanguage) {
- text = text.split("\n").filter(line => !line.startsWith("# ")).join("\n");
+ PrismObject.hooks.add("after-tokenize", function(env) {
+ if (env.language === "rust") {
+ let code = env.code.split("\n").filter(line => !line.startsWith("# ")).join("\n");
+ env.tokens = origTokenize(code, env.grammar);
}
- return origTokenize(text, grammar);
- };
+ });
delete globalThis.Prism;
+ if (typeof PrismBefore !== 'undefined') {
+ globalThis.Prism = PrismObject;
+ }
}
diff --git a/docs/src/util.ts b/docs/src/util.ts
index 4cd8eec79eb9..7ee441194e4d 100644
--- a/docs/src/util.ts
+++ b/docs/src/util.ts
@@ -1,8 +1,41 @@
-/// Grabs the specified lines `[lineStart, lineEnd]` from the string.
+/// Grabs the specified lines from the string.
+///
+/// Can specify a line number range `lineStart, lineEnd`, a specific line number `lineNumber` (no
+/// second arg), or a section name.
///
/// Lines are one-indexed (start with `1`). Both `lineStart` and `lineEnd` are inclusive.
-export function getLines(str: string, lineStart: number, lineEnd?: number): string {
- let lines = str.split('\n').slice(lineStart - 1, lineEnd || lineStart);
+///
+/// Sections are marked in the code with a start tag `//[mysection]//` and and end tag
+/// `//[/mysection]//`. The rest of these tag lines must be whitespace, and these tag lines are not
+/// included in the output. However they are included for line number _counting_.
+export function getLines(str: string, sectionName: string): string;
+export function getLines(str: string, lineStart: number, lineEnd?: number): string;
+export function getLines(str: string, lineStartOrSectionName: number | string, lineEnd?: number): string {
+ // `//[section]//` or `//[/section]//` (rest of line must be whitespace).
+ const SECTION_REGEX = /^\s*\/\/\[(\/?)(\S+)\]\/\/\s*$/;
+ let lines;
+ if ('string' === typeof lineStartOrSectionName) {
+ let inSection = false;
+ lines = str
+ .split('\n')
+ .filter(line => {
+ const match = SECTION_REGEX.exec(line);
+ if (null == match) {
+ return inSection;
+ }
+ const [_, end, name] = match;
+ if (name == lineStartOrSectionName) {
+ inSection = 0 === end.length;
+ }
+ return false;
+ })
+ }
+ else {
+ lines = str
+ .split('\n')
+ .slice(lineStartOrSectionName - 1, lineEnd || lineStartOrSectionName) // Select lines before removing section lines.
+ .filter(line => !SECTION_REGEX.test(line));
+ }
const leadingWhitespace = Math.min(...lines.filter(line => 0 !== line.length).map(line => line.search(/\S/)).map(Number));
if (0 < leadingWhitespace) {
lines = lines.map(line => line.slice(leadingWhitespace));
@@ -10,6 +43,16 @@ export function getLines(str: string, lineStart: number, lineEnd?: number): stri
return lines.join('\n');
}
+/// Adds `// highlight-next-line` annotations to the specified lines.
+export function highlightLines(code: string, lines: number[]): string {
+ return code.split('\n').map((line, i) => {
+ if (lines.includes(i + 1)) {
+ return `// highlight-next-line\n${line}`;
+ }
+ return line;
+ }).join('\n');
+}
+
/// Extract the output from the stdout snapshots created by `surface_examples.rs`.
///
/// This hides the graph output. Use `extractMermaid` to extract the graph output.
@@ -40,7 +83,7 @@ ${stdOut}`;
/// Extract the mermaid graph logged to stdout from the snapshots created by `surface_examples.rs`.
export function extractMermaid(output: string): string {
const outputLines = output.split('\n');
- // Delete the first four lines, which are the snapshot front matter.
+ // Delete the first four lines, which are the snapshot front matter.
outputLines.splice(0, 4);
// Mermaid graph starts with double-percent signs.
if (!outputLines[0].startsWith('%%')) {
diff --git a/docs/static/.gitignore b/docs/static/.gitignore
new file mode 100644
index 000000000000..1d109a9866c6
--- /dev/null
+++ b/docs/static/.gitignore
@@ -0,0 +1 @@
+rustdoc/
diff --git a/docs/static/img/hydro-stack.png b/docs/static/img/hydro-stack.png
new file mode 100644
index 000000000000..5b7cf22da9a3
Binary files /dev/null and b/docs/static/img/hydro-stack.png differ
diff --git a/docs/static/papers/flo.pdf b/docs/static/papers/flo.pdf
new file mode 100644
index 000000000000..146dab24a520
Binary files /dev/null and b/docs/static/papers/flo.pdf differ
diff --git a/docs/static/papers/suki.pdf b/docs/static/papers/suki.pdf
new file mode 100644
index 000000000000..e45a0aae7981
Binary files /dev/null and b/docs/static/papers/suki.pdf differ
diff --git a/hydro_deploy/core/CHANGELOG.md b/hydro_deploy/core/CHANGELOG.md
index 26f8b6922b0e..f526879cdf38 100644
--- a/hydro_deploy/core/CHANGELOG.md
+++ b/hydro_deploy/core/CHANGELOG.md
@@ -5,8 +5,82 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+## v0.10.0 (2024-11-08)
+
+### Chore
+
+ - view details
+
+ * **[#1444](https://github.com/hydro-project/hydroflow/issues/1444)**
+ - Update pinned rust version, clippy lints, remove some dead code ([`d567760`](https://github.com/hydro-project/hydroflow/commit/d5677604e93c07a5392f4229af94a0b736eca382))
+ * **[#1449](https://github.com/hydro-project/hydroflow/issues/1449)**
+ - Add API for external network inputs ([`8a80931`](https://github.com/hydro-project/hydroflow/commit/8a809315cd37929687fcabc34a12042db25d5767))
+ * **[#1450](https://github.com/hydro-project/hydroflow/issues/1450)**
+ - Add ability to have staged flows inside unit tests ([`afe78c3`](https://github.com/hydro-project/hydroflow/commit/afe78c343658472513b34d28658634b253148aee))
+ * **[#1537](https://github.com/hydro-project/hydroflow/issues/1537)**
+ - Fixes for latest nightly clippy ([`8442d1b`](https://github.com/hydro-project/hydroflow/commit/8442d1b524621a9f8b43372a9c25991efb33c25e))
+view details
+
+ * **[#1444](https://github.com/hydro-project/hydroflow/issues/1444)**
+ - Update pinned rust version, clippy lints, remove some dead code ([`d567760`](https://github.com/hydro-project/hydroflow/commit/d5677604e93c07a5392f4229af94a0b736eca382))
+view details
+
+ * **[#1444](https://github.com/hydro-project/hydroflow/issues/1444)**
+ - Update pinned rust version, clippy lints, remove some dead code ([`d567760`](https://github.com/hydro-project/hydroflow/commit/d5677604e93c07a5392f4229af94a0b736eca382))
+view details
+
+ * **[#1444](https://github.com/hydro-project/hydroflow/issues/1444)**
+ - Update pinned rust version, clippy lints, remove some dead code ([`d567760`](https://github.com/hydro-project/hydroflow/commit/d5677604e93c07a5392f4229af94a0b736eca382))
+ * **[#1446](https://github.com/hydro-project/hydroflow/issues/1446)**
+ - Cleanup temp tcp networking code, fix race condition fix #1458 ([`b961233`](https://github.com/hydro-project/hydroflow/commit/b96123369cec3d6d407af973f1896b2734fd92ef))
+ * **[#1497](https://github.com/hydro-project/hydroflow/issues/1497)**
+ - Update `proc-macro2`, fixes span info, fix #729 ([`e564b13`](https://github.com/hydro-project/hydroflow/commit/e564b133a4db192e0331d427aa80ef52cd97608c))
+ * **[#1503](https://github.com/hydro-project/hydroflow/issues/1503)**
+ - Generalized hash trie indexes for relational tuples ([`f7e740f`](https://github.com/hydro-project/hydroflow/commit/f7e740fb2ba36d0fcf3fd196d60333552911e3a4))
+ * **[#1505](https://github.com/hydro-project/hydroflow/issues/1505)**
+ - Fixes for nightly clippy ([`47cb703`](https://github.com/hydro-project/hydroflow/commit/47cb703e771f7d1c451ceb9d185ada96410949da))
+ * **[#1520](https://github.com/hydro-project/hydroflow/issues/1520)**
+ - `cross_singleton()` forgot value if multiple runs in a tick, fix #1518 ([`16b730c`](https://github.com/hydro-project/hydroflow/commit/16b730c75cfca79ea5f869308b1e1e14b3e9c155))
+ * **[#1522](https://github.com/hydro-project/hydroflow/issues/1522)**
+ - Fix #1401 `lattice_bimorphism()` double-emit, add docs ([`e796200`](https://github.com/hydro-project/hydroflow/commit/e796200743f2cc2da5a0e91c492f016ca98008e8))
+ * **[#1537](https://github.com/hydro-project/hydroflow/issues/1537)**
+ - Fixes for latest nightly clippy ([`8442d1b`](https://github.com/hydro-project/hydroflow/commit/8442d1b524621a9f8b43372a9c25991efb33c25e))
+ * **Uncategorized**
+ - Ignore trybuild tests inconsistent on latest nightly ([`656ee32`](https://github.com/hydro-project/hydroflow/commit/656ee328c8710bce7370c851437a80ca3db46a5a))
+
-Other 'tick state will need to be cleared, but existing implementation does that when the iterator runs, which is good enough. There is only a problem if a singleton can reference the state before the iterator runs, in that case. allow use of generics in demux_enum::<...>() opensures turbofish syntax for the match clauses, which is needed when there are generic parameters in the typeadds a test as well yield before collect_ready_async to ensure background async tasks can send to the stream make sure tasks are spawned!!!!fix bug introduced in #978 Make inner for WithTop & WithBot privateOption is not a lattice, so it is unsafe to expose as public.I also updated documentation to lead with intention beforeimplementation (minor cleanup).To emulate this unintuitive behavior, we currently ensure that apersist::<'static>() exists before operator that references thesingleton (filter, in this case). (Note that this is equivalent tocross_join::<'static>() and not cross_join::<'tick>())However singletons also have had a different mechanism that affectsthis- currently singleton references create a next-stratum constraint,that ensures a singleton referencer must be in a later stratum than thesingleton it is referencing.Note that this actually prevents the example situation above fromhappening– the updates to y will be received all at once at the startof the next stratum.This means that actually, currently singletons are equivalent tosomething like:ruststream -> cj[0];
-y -> next_stratum() -> last() -> cj[1];
-cj = cross_join() -> filter(|(item, y)| ...) -> ...
-last() is a hypothetical operator that only keeps the most recent itemoutput by y. next_stratum() -> last() is equivalent to reduce(|acc, item| *acc = item) (since that comes with a stratum barrier). Sotechnically this is a slightly different behavior than just cross_join,but it is more intuitive.ruststream -> cj[0];
-y -> reduce(|acc, item| { *acc = item; }) -> cj[1];
-cj = cross_join() -> filter(|(item, y)| ...) -> ...
-Also fixes #1293>,
);
-impl<'a, const SIZE: usize> Serialize for MapUnionHashMapWrapper<'a, SIZE> {
+impl Serialize for MapUnionHashMapWrapper<'_, SIZE> {
fn serialize(&self, serializer: S) -> Result
where
S: Serializer,
diff --git a/hydroflow/examples/kvs_bench/protocol/serialization/lattices/my_last_write_wins.rs b/hydroflow/examples/kvs_bench/protocol/serialization/lattices/my_last_write_wins.rs
index 91fc1b71dfb4..74b3041bfbe7 100644
--- a/hydroflow/examples/kvs_bench/protocol/serialization/lattices/my_last_write_wins.rs
+++ b/hydroflow/examples/kvs_bench/protocol/serialization/lattices/my_last_write_wins.rs
@@ -11,7 +11,7 @@ use crate::protocol::MyLastWriteWins;
#[repr(transparent)]
pub struct MyLastWriteWinsWrapper<'a, const SIZE: usize>(pub &'a MyLastWriteWins);
-impl<'a, const SIZE: usize> Serialize for MyLastWriteWinsWrapper<'a, SIZE> {
+impl Serialize for MyLastWriteWinsWrapper<'_, SIZE> {
fn serialize(&self, serializer: S) -> Result
where
S: Serializer,
diff --git a/hydroflow/examples/kvs_bench/protocol/serialization/lattices/point.rs b/hydroflow/examples/kvs_bench/protocol/serialization/lattices/point.rs
index b0410e203947..98b2552a5c28 100644
--- a/hydroflow/examples/kvs_bench/protocol/serialization/lattices/point.rs
+++ b/hydroflow/examples/kvs_bench/protocol/serialization/lattices/point.rs
@@ -10,7 +10,7 @@ use crate::buffer_pool::{AutoReturnBuffer, AutoReturnBufferDeserializer, BufferP
#[repr(transparent)]
pub struct PointWrapper<'a, const SIZE: usize>(pub &'a Point, ()>);
-impl<'a, const SIZE: usize> Serialize for PointWrapper<'a, SIZE> {
+impl Serialize for PointWrapper<'_, SIZE> {
fn serialize(&self, serializer: S) -> Result
where
S: Serializer,
diff --git a/hydroflow/examples/kvs_bench/protocol/serialization/lattices/with_bot.rs b/hydroflow/examples/kvs_bench/protocol/serialization/lattices/with_bot.rs
index f5204743b698..d7af4a572f24 100644
--- a/hydroflow/examples/kvs_bench/protocol/serialization/lattices/with_bot.rs
+++ b/hydroflow/examples/kvs_bench/protocol/serialization/lattices/with_bot.rs
@@ -14,7 +14,7 @@ pub struct WithBotWrapper<'a, const SIZE: usize>(
pub &'a WithBot, ()>>,
);
-impl<'a, const SIZE: usize> Serialize for WithBotWrapper<'a, SIZE> {
+impl Serialize for WithBotWrapper<'_, SIZE> {
fn serialize(&self, serializer: S) -> Result
where
S: Serializer,
diff --git a/hydroflow/examples/kvs_bench/protocol/serialization/mod.rs b/hydroflow/examples/kvs_bench/protocol/serialization/mod.rs
index db7686f995a2..29fdfb142503 100644
--- a/hydroflow/examples/kvs_bench/protocol/serialization/mod.rs
+++ b/hydroflow/examples/kvs_bench/protocol/serialization/mod.rs
@@ -116,7 +116,7 @@ enum KvsRequestField {
Delete,
}
struct KvsRequestFieldVisitor;
-impl<'de> Visitor<'de> for KvsRequestFieldVisitor {
+impl Visitor<'_> for KvsRequestFieldVisitor {
type Value = KvsRequestField;
fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
diff --git a/hydroflow/src/compiled/pull/anti_join.rs b/hydroflow/src/compiled/pull/anti_join.rs
index 9e11f691f1c1..97421b7de62d 100644
--- a/hydroflow/src/compiled/pull/anti_join.rs
+++ b/hydroflow/src/compiled/pull/anti_join.rs
@@ -12,7 +12,7 @@ where
pos_state: &'a mut FxHashSet<(Key, V)>,
}
-impl<'a, Key, V, Ipos> Iterator for AntiJoin<'a, Key, V, Ipos>
+impl Iterator for AntiJoin<'_, Key, V, Ipos>
where
Key: Eq + std::hash::Hash + Clone,
V: Eq + std::hash::Hash + Clone,
diff --git a/hydroflow/src/compiled/pull/cross_join.rs b/hydroflow/src/compiled/pull/cross_join.rs
index 90f743d9bc92..6b9629fa65f9 100644
--- a/hydroflow/src/compiled/pull/cross_join.rs
+++ b/hydroflow/src/compiled/pull/cross_join.rs
@@ -30,7 +30,7 @@ where
state: &'a mut CrossJoinState,
}
-impl<'a, I1, V1: 'static, I2, V2: 'static> Iterator for CrossJoin<'a, I1, V1, I2, V2>
+impl Iterator for CrossJoin<'_, I1, V1, I2, V2>
where
V1: Eq + Clone,
V2: Eq + Clone,
diff --git a/hydroflow/src/compiled/pull/symmetric_hash_join.rs b/hydroflow/src/compiled/pull/symmetric_hash_join.rs
index d240e6a347ad..1665c9940a05 100644
--- a/hydroflow/src/compiled/pull/symmetric_hash_join.rs
+++ b/hydroflow/src/compiled/pull/symmetric_hash_join.rs
@@ -18,8 +18,8 @@ where
rhs_state: &'a mut RhsState,
}
-impl<'a, Key, I1, V1, I2, V2, LhsState, RhsState> Iterator
- for SymmetricHashJoin<'a, Key, I1, V1, I2, V2, LhsState, RhsState>
+impl Iterator
+ for SymmetricHashJoin<'_, Key, I1, V1, I2, V2, LhsState, RhsState>
where
Key: Eq + std::hash::Hash + Clone,
V1: Clone,
diff --git a/hydroflow/src/scheduled/graph.rs b/hydroflow/src/scheduled/graph.rs
index 85d71d98bfe0..2d4e96962ee2 100644
--- a/hydroflow/src/scheduled/graph.rs
+++ b/hydroflow/src/scheduled/graph.rs
@@ -47,7 +47,7 @@ pub struct Hydroflow<'a> {
/// See [`Self::diagnostics()`].
diagnostics: Option>>,
}
-impl<'a> Default for Hydroflow<'a> {
+impl Default for Hydroflow<'_> {
fn default() -> Self {
let stratum_queues = vec![Default::default()]; // Always initialize stratum #0.
let (event_queue_send, event_queue_recv) = mpsc::unbounded_channel();
@@ -69,7 +69,7 @@ impl<'a> Default for Hydroflow<'a> {
}
/// Methods for [`TeeingHandoff`] teeing and dropping.
-impl<'a> Hydroflow<'a> {
+impl Hydroflow<'_> {
/// Tees a [`TeeingHandoff`].
pub fn teeing_handoff_tee(
&mut self,
@@ -794,7 +794,7 @@ impl<'a> Hydroflow<'a> {
}
}
-impl<'a> Hydroflow<'a> {
+impl Hydroflow<'_> {
/// Alias for [`Context::request_task`].
pub fn request_task(&mut self, future: Fut)
where
@@ -814,7 +814,7 @@ impl<'a> Hydroflow<'a> {
}
}
-impl<'a> Drop for Hydroflow<'a> {
+impl Drop for Hydroflow<'_> {
fn drop(&mut self) {
self.abort_tasks();
}
diff --git a/hydroflow/src/scheduled/graph_ext.rs b/hydroflow/src/scheduled/graph_ext.rs
index 6f54413d832e..1d0be9d28dd6 100644
--- a/hydroflow/src/scheduled/graph_ext.rs
+++ b/hydroflow/src/scheduled/graph_ext.rs
@@ -126,7 +126,7 @@ pub trait GraphExt {
W: 'static + Handoff + CanReceive;
}
-impl<'a> GraphExt for Hydroflow<'a> {
+impl GraphExt for Hydroflow<'_> {
subgraph_ext!(impl add_subgraph_sink, (recv_port: R), ());
subgraph_ext!(
impl add_subgraph_2sink,
diff --git a/hydroflow/src/scheduled/net/mod.rs b/hydroflow/src/scheduled/net/mod.rs
index 48ffc2e6115f..10664f8a58d3 100644
--- a/hydroflow/src/scheduled/net/mod.rs
+++ b/hydroflow/src/scheduled/net/mod.rs
@@ -96,7 +96,7 @@ impl Message {
}
}
-impl<'a> Hydroflow<'a> {
+impl Hydroflow<'_> {
fn register_read_tcp_stream(&mut self, reader: OwnedReadHalf) -> RecvPort> {
let reader = FramedRead::new(reader, LengthDelimitedCodec::new());
let (send_port, recv_port) = self.make_edge("tcp ingress handoff");
diff --git a/hydroflow/src/scheduled/net/network_vertex.rs b/hydroflow/src/scheduled/net/network_vertex.rs
index 13077e6290e8..b2db71169646 100644
--- a/hydroflow/src/scheduled/net/network_vertex.rs
+++ b/hydroflow/src/scheduled/net/network_vertex.rs
@@ -18,7 +18,7 @@ pub type Address = String;
// These methods can't be wrapped up in a trait because async methods are not
// allowed in traits (yet).
-impl<'a> Hydroflow<'a> {
+impl Hydroflow<'_> {
// TODO(justin): document these, but they're derivatives of inbound_tcp_vertex_internal.
pub async fn inbound_tcp_vertex_port(&mut self, port: u16) -> RecvPort>
where
diff --git a/hydroflow/src/util/simulation.rs b/hydroflow/src/util/simulation.rs
index f1611be636e7..86b2bc785644 100644
--- a/hydroflow/src/util/simulation.rs
+++ b/hydroflow/src/util/simulation.rs
@@ -177,7 +177,7 @@ fn sink_from_fn(mut f: impl FnMut(T)) -> impl Sink {
})
}
-impl<'context> TransducerBuilderContext<'context> {
+impl TransducerBuilderContext<'_> {
/// Create a new inbox on the host with the given interface name. Returns a stream that can
/// be read by the transducer using the source_stream hydroflow operator.
pub fn new_inbox(
diff --git a/hydroflow/src/util/tcp.rs b/hydroflow/src/util/tcp.rs
index 5f1cab36003b..9c50c3094160 100644
--- a/hydroflow/src/util/tcp.rs
+++ b/hydroflow/src/util/tcp.rs
@@ -1,16 +1,16 @@
#![cfg(not(target_arch = "wasm32"))]
-use std::cell::RefCell;
use std::collections::hash_map::Entry::{Occupied, Vacant};
use std::collections::HashMap;
+use std::fmt::Debug;
use std::net::SocketAddr;
-use std::pin::pin;
-use std::rc::Rc;
use futures::{SinkExt, StreamExt};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpSocket, TcpStream};
+use tokio::select;
use tokio::task::spawn_local;
+use tokio_stream::StreamMap;
use tokio_util::codec::{
BytesCodec, Decoder, Encoder, FramedRead, FramedWrite, LengthDelimitedCodec, LinesCodec,
};
@@ -74,69 +74,90 @@ pub type TcpFramedSink = Sender<(T, SocketAddr)>;
pub type TcpFramedStream =
Receiver::Item, SocketAddr), ::Error>>;
+// TODO(mingwei): this temporary code should be replaced with a properly thought out networking system.
/// Create a listening tcp socket, and then as new connections come in, receive their data and forward it to a queue.
-pub async fn bind_tcp>(
+pub async fn bind_tcp(
endpoint: SocketAddr,
codec: Codec,
-) -> Result<(TcpFramedSink, TcpFramedStream, SocketAddr), std::io::Error> {
+) -> Result<(TcpFramedSink- , TcpFramedStream
, SocketAddr), std::io::Error>
+where
+ Item: 'static,
+ Codec: 'static + Clone + Decoder + Encoder- ,
+
>::Error: Debug,
+{
let listener = TcpListener::bind(endpoint).await?;
let bound_endpoint = listener.local_addr()?;
- let (tx_egress, mut rx_egress) = unsync_channel(None);
- let (tx_ingress, rx_ingress) = unsync_channel(None);
-
- let clients = Rc::new(RefCell::new(HashMap::new()));
-
- spawn_local({
- let clients = clients.clone();
-
- async move {
- while let Some((payload, addr)) = rx_egress.next().await {
- let client = clients.borrow_mut().remove(&addr);
-
- if let Some(mut sender) = client {
- let _ = SinkExt::send(&mut sender, payload).await;
- clients.borrow_mut().insert(addr, sender);
- }
- }
- }
- });
+ let (send_egress, mut recv_egress) = unsync_channel::<(Item, SocketAddr)>(None);
+ let (send_ingres, recv_ingres) = unsync_channel(None);
spawn_local(async move {
+ let send_ingress = send_ingres;
+ // Map of `addr -> peers`, to send messages to.
+ let mut peers_send = HashMap::new();
+ // `StreamMap` of `addr -> peers`, to receive messages from. Automatically removes streams
+ // when they disconnect.
+ let mut peers_recv = StreamMap::>::new();
+
loop {
- let (stream, peer_addr) = if let Ok((stream, _)) = listener.accept().await {
- if let Ok(peer_addr) = stream.peer_addr() {
- (stream, peer_addr)
- } else {
- continue;
+ // Calling methods in a loop, futures must be cancel-safe.
+ select! {
+ // `biased` means the cases will be prioritized in the order they are listed.
+ // First we accept any new connections
+ // This is not strictly neccessary, but lets us do our internal work (send outgoing
+ // messages) before accepting more work (receiving more messages, accepting new
+ // clients).
+ biased;
+ // Send outgoing messages.
+ msg_send = recv_egress.next() => {
+ let Some((payload, peer_addr)) = msg_send else {
+ // `None` if the send side has been dropped (no more send messages will ever come).
+ continue;
+ };
+ let Some(stream) = peers_send.get_mut(&peer_addr) else {
+ tracing::warn!("Dropping message to non-connected peer: {}", peer_addr);
+ continue;
+ };
+ if let Err(err) = SinkExt::send(stream, payload).await {
+ tracing::error!("IO or codec error sending message to peer {}, disconnecting: {:?}", peer_addr, err);
+ peers_send.remove(&peer_addr); // `Drop` disconnects.
+ };
}
- } else {
- continue;
- };
-
- let mut tx_ingress = tx_ingress.clone();
-
- let (send, recv) = tcp_framed(stream, codec.clone());
-
- // TODO: Using peer_addr here as the key is a little bit sketchy.
- // It's possible that a client could send a message, disconnect, then another client connects from the same IP address (and the same src port), and then the response could be sent to that new client.
- // This can be solved by using monotonically increasing IDs for each new client, but would break the similarity with the UDP versions of this function.
- clients.borrow_mut().insert(peer_addr, send);
-
- spawn_local({
- let clients = clients.clone();
- async move {
- let mapped = recv.map(|x| Ok(x.map(|x| (x, peer_addr))));
- let _ = tx_ingress.send_all(&mut pin!(mapped)).await;
-
- clients.borrow_mut().remove(&peer_addr);
+ // Receive incoming messages.
+ msg_recv = peers_recv.next(), if !peers_recv.is_empty() => {
+ // If `peers_recv` is empty then `next()` will immediately return `None` which
+ // would cause the loop to spin.
+ let Some((peer_addr, payload_result)) = msg_recv else {
+ continue; // => `peers_recv.is_empty()`.
+ };
+ if let Err(err) = send_ingress.send(payload_result.map(|payload| (payload, peer_addr))).await {
+ tracing::error!("Error passing along received message: {:?}", err);
+ }
}
- });
+ // Accept new clients.
+ new_peer = listener.accept() => {
+ let Ok((stream, _addr)) = new_peer else {
+ continue;
+ };
+ let Ok(peer_addr) = stream.peer_addr() else {
+ continue;
+ };
+ let (peer_send, peer_recv) = tcp_framed(stream, codec.clone());
+
+ // TODO: Using peer_addr here as the key is a little bit sketchy.
+ // It's possible that a peer could send a message, disconnect, then another peer connects from the
+ // same IP address (and the same src port), and then the response could be sent to that new client.
+ // This can be solved by using monotonically increasing IDs for each new peer, but would break the
+ // similarity with the UDP versions of this function.
+ peers_send.insert(peer_addr, peer_send);
+ peers_recv.insert(peer_addr, peer_recv);
+ }
+ }
}
});
- Ok((tx_egress, rx_ingress, bound_endpoint))
+ Ok((send_egress, recv_ingres, bound_endpoint))
}
/// The inverse of [`bind_tcp`].
@@ -144,37 +165,69 @@ pub async fn bind_tcp>
/// When messages enqueued into the returned sender, tcp sockets will be created and connected as
/// necessary to send out the requests. As the responses come back, they will be forwarded to the
/// returned receiver.
-pub fn connect_tcp>(
- codec: Codec,
-) -> (TcpFramedSink, TcpFramedStream) {
- let (tx_egress, mut rx_egress) = unsync_channel(None);
- let (tx_ingress, rx_ingress) = unsync_channel(None);
+pub fn connect_tcp(codec: Codec) -> (TcpFramedSink- , TcpFramedStream
)
+where
+ Item: 'static,
+ Codec: 'static + Clone + Decoder + Encoder- ,
+
>::Error: Debug,
+{
+ let (send_egress, mut recv_egress) = unsync_channel(None);
+ let (send_ingres, recv_ingres) = unsync_channel(None);
spawn_local(async move {
- let mut streams = HashMap::new();
+ let send_ingres = send_ingres;
+ // Map of `addr -> peers`, to send messages to.
+ let mut peers_send = HashMap::new();
+ // `StreamMap` of `addr -> peers`, to receive messages from. Automatically removes streams
+ // when they disconnect.
+ let mut peers_recv = StreamMap::new();
- while let Some((payload, addr)) = rx_egress.next().await {
- let stream = match streams.entry(addr) {
- Occupied(entry) => entry.into_mut(),
- Vacant(entry) => {
- let socket = TcpSocket::new_v4().unwrap();
- let stream = socket.connect(addr).await.unwrap();
-
- let (send, recv) = tcp_framed(stream, codec.clone());
-
- let mut tx_ingress = tx_ingress.clone();
- spawn_local(async move {
- let mapped = recv.map(|x| Ok(x.map(|x| (x, addr))));
- let _ = tx_ingress.send_all(&mut pin!(mapped)).await;
- });
-
- entry.insert(send)
+ loop {
+ // Calling methods in a loop, futures must be cancel-safe.
+ select! {
+ // `biased` means the cases will be prioritized in the order they are listed.
+ // This is not strictly neccessary, but lets us do our internal work (send outgoing
+ // messages) before accepting more work (receiving more messages).
+ biased;
+ // Send outgoing messages.
+ msg_send = recv_egress.next() => {
+ let Some((payload, peer_addr)) = msg_send else {
+ // `None` if the send side has been dropped (no more send messages will ever come).
+ continue;
+ };
+
+ let stream = match peers_send.entry(peer_addr) {
+ Occupied(entry) => entry.into_mut(),
+ Vacant(entry) => {
+ let socket = TcpSocket::new_v4().unwrap();
+ let stream = socket.connect(peer_addr).await.unwrap();
+
+ let (peer_send, peer_recv) = tcp_framed(stream, codec.clone());
+
+ peers_recv.insert(peer_addr, peer_recv);
+ entry.insert(peer_send)
+ }
+ };
+
+ if let Err(err) = stream.send(payload).await {
+ tracing::error!("IO or codec error sending message to peer {}, disconnecting: {:?}", peer_addr, err);
+ peers_send.remove(&peer_addr); // `Drop` disconnects.
+ }
}
- };
-
- let _ = stream.send(payload).await;
+ // Receive incoming messages.
+ msg_recv = peers_recv.next(), if !peers_recv.is_empty() => {
+ // If `peers_recv` is empty then `next()` will immediately return `None` which
+ // would cause the loop to spin.
+ let Some((peer_addr, payload_result)) = msg_recv else {
+ continue; // => `peers_recv.is_empty()`.
+ };
+ if let Err(err) = send_ingres.send(payload_result.map(|payload| (payload, peer_addr))).await {
+ tracing::error!("Error passing along received message: {:?}", err);
+ }
+ }
+ }
}
});
- (tx_egress, rx_ingress)
+ (send_egress, recv_ingres)
}
diff --git a/hydroflow/tests/compile-fail/surface_loop_cycle.rs b/hydroflow/tests/compile-fail/surface_loop_cycle.rs
new file mode 100644
index 000000000000..0ddf384b4564
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_cycle.rs
@@ -0,0 +1,9 @@
+fn main() {
+ let mut df = hydroflow::hydroflow_syntax! {
+ loop {
+ a = identity() -> identity() -> identity() -> identity();
+ a -> a;
+ }
+ };
+ df.run_available();
+}
diff --git a/hydroflow/tests/compile-fail/surface_loop_cycle.stderr b/hydroflow/tests/compile-fail/surface_loop_cycle.stderr
new file mode 100644
index 000000000000..dc056047cbf9
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_cycle.stderr
@@ -0,0 +1,23 @@
+error: Operator forms an illegal cycle within a `loop { ... }` block (1/4).
+ --> tests/compile-fail/surface_loop_cycle.rs:4:31
+ |
+4 | a = identity() -> identity() -> identity() -> identity();
+ | ^^^^^^^^^^
+
+error: Operator forms an illegal cycle within a `loop { ... }` block (2/4).
+ --> tests/compile-fail/surface_loop_cycle.rs:4:45
+ |
+4 | a = identity() -> identity() -> identity() -> identity();
+ | ^^^^^^^^^^
+
+error: Operator forms an illegal cycle within a `loop { ... }` block (3/4).
+ --> tests/compile-fail/surface_loop_cycle.rs:4:59
+ |
+4 | a = identity() -> identity() -> identity() -> identity();
+ | ^^^^^^^^^^
+
+error: Operator forms an illegal cycle within a `loop { ... }` block (4/4).
+ --> tests/compile-fail/surface_loop_cycle.rs:4:17
+ |
+4 | a = identity() -> identity() -> identity() -> identity();
+ | ^^^^^^^^^^
diff --git a/hydroflow/tests/compile-fail/surface_loop_missing_unwindowing.rs b/hydroflow/tests/compile-fail/surface_loop_missing_unwindowing.rs
new file mode 100644
index 000000000000..808bfae7a0e2
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_missing_unwindowing.rs
@@ -0,0 +1,10 @@
+fn main() {
+ let mut df = hydroflow::hydroflow_syntax! {
+ a = source_iter(0..10);
+ loop {
+ b = a -> batch();
+ }
+ b -> null();
+ };
+ df.run_available();
+}
diff --git a/hydroflow/tests/compile-fail/surface_loop_missing_unwindowing.stderr b/hydroflow/tests/compile-fail/surface_loop_missing_unwindowing.stderr
new file mode 100644
index 000000000000..e0343ead3779
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_missing_unwindowing.stderr
@@ -0,0 +1,5 @@
+error: Operator `null(...)` exiting a loop context must be an un-windowing operator, but is not.
+ --> tests/compile-fail/surface_loop_missing_unwindowing.rs:7:14
+ |
+7 | b -> null();
+ | ^^^^^^
diff --git a/hydroflow/tests/compile-fail/surface_loop_missing_windowing.rs b/hydroflow/tests/compile-fail/surface_loop_missing_windowing.rs
new file mode 100644
index 000000000000..2177d2ad0e8c
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_missing_windowing.rs
@@ -0,0 +1,9 @@
+fn main() {
+ let mut df = hydroflow::hydroflow_syntax! {
+ a = source_iter(0..10);
+ loop {
+ a -> null();
+ }
+ };
+ df.run_available();
+}
diff --git a/hydroflow/tests/compile-fail/surface_loop_missing_windowing.stderr b/hydroflow/tests/compile-fail/surface_loop_missing_windowing.stderr
new file mode 100644
index 000000000000..1735ebd48911
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_missing_windowing.stderr
@@ -0,0 +1,5 @@
+error: Operator `null(...)` entering a loop context must be a windowing operator, but is not.
+ --> tests/compile-fail/surface_loop_missing_windowing.rs:5:18
+ |
+5 | a -> null();
+ | ^^^^^^
diff --git a/hydroflow/tests/compile-fail/surface_loop_multiple_window.rs b/hydroflow/tests/compile-fail/surface_loop_multiple_window.rs
new file mode 100644
index 000000000000..eb0eb775bf29
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_multiple_window.rs
@@ -0,0 +1,11 @@
+fn main() {
+ let mut df = hydroflow::hydroflow_syntax! {
+ a = source_iter(0..10);
+ loop {
+ loop {
+ a -> batch() -> null();
+ }
+ }
+ };
+ df.run_available();
+}
diff --git a/hydroflow/tests/compile-fail/surface_loop_multiple_window.stderr b/hydroflow/tests/compile-fail/surface_loop_multiple_window.stderr
new file mode 100644
index 000000000000..00a863b05482
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_multiple_window.stderr
@@ -0,0 +1,5 @@
+error: Operator input edge may not cross multiple loop contexts.
+ --> tests/compile-fail/surface_loop_multiple_window.rs:6:22
+ |
+6 | a -> batch() -> null();
+ | ^^^^^^^
diff --git a/hydroflow/tests/compile-fail/surface_loop_source.rs b/hydroflow/tests/compile-fail/surface_loop_source.rs
new file mode 100644
index 000000000000..edea1a7ad958
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_source.rs
@@ -0,0 +1,8 @@
+fn main() {
+ let mut df = hydroflow::hydroflow_syntax! {
+ loop {
+ source_iter(0..10) -> null();
+ }
+ };
+ df.run_available();
+}
diff --git a/hydroflow/tests/compile-fail/surface_loop_source.stderr b/hydroflow/tests/compile-fail/surface_loop_source.stderr
new file mode 100644
index 000000000000..34bbb210a144
--- /dev/null
+++ b/hydroflow/tests/compile-fail/surface_loop_source.stderr
@@ -0,0 +1,5 @@
+error: Source operator `source_iter(...)` must be at the root level, not within any `loop { ... }` contexts.
+ --> tests/compile-fail/surface_loop_source.rs:4:13
+ |
+4 | source_iter(0..10) -> null();
+ | ^^^^^^^^^^^^^^^^^^
diff --git a/hydroflow/tests/compile-fail/surface_source_iter_badtype.rs b/hydroflow/tests/compile-fail/surface_source_iter_badtype.rs.ignore
similarity index 68%
rename from hydroflow/tests/compile-fail/surface_source_iter_badtype.rs
rename to hydroflow/tests/compile-fail/surface_source_iter_badtype.rs.ignore
index 026c5f14bf66..f97ad44dfcf2 100644
--- a/hydroflow/tests/compile-fail/surface_source_iter_badtype.rs
+++ b/hydroflow/tests/compile-fail/surface_source_iter_badtype.rs.ignore
@@ -2,7 +2,7 @@ use hydroflow::hydroflow_syntax;
fn main() {
let mut df = hydroflow_syntax! {
- source_iter(5) -> for_each(std::mem::drop);
+ source_iter(()) -> for_each(std::mem::drop);
};
df.run_available();
}
diff --git a/hydroflow/tests/compile-fail/surface_source_iter_badtype.stderr b/hydroflow/tests/compile-fail/surface_source_iter_badtype.stderr
index 9866f73a5d6b..85d2c8e16e8d 100644
--- a/hydroflow/tests/compile-fail/surface_source_iter_badtype.stderr
+++ b/hydroflow/tests/compile-fail/surface_source_iter_badtype.stderr
@@ -1,45 +1,44 @@
-error[E0277]: `{integer}` is not an iterator
+error[E0277]: `()` is not an iterator
--> tests/compile-fail/surface_source_iter_badtype.rs:5:21
|
-5 | source_iter(5) -> for_each(std::mem::drop);
- | ------------^-
+5 | source_iter(()) -> for_each(std::mem::drop);
+ | ------------^^-
| | |
- | | `{integer}` is not an iterator
+ | | `()` is not an iterator
| required by a bound introduced by this call
|
- = help: the trait `Iterator` is not implemented for `{integer}`, which is required by `{integer}: IntoIterator`
- = note: if you want to iterate between `start` until a value `end`, use the exclusive range syntax `start..end` or the inclusive range syntax `start..=end`
- = note: required for `{integer}` to implement `IntoIterator`
+ = help: the trait `Iterator` is not implemented for `()`, which is required by `(): IntoIterator`
+ = note: required for `()` to implement `IntoIterator`
note: required by a bound in `check_iter`
--> tests/compile-fail/surface_source_iter_badtype.rs:5:9
|
-5 | source_iter(5) -> for_each(std::mem::drop);
- | ^^^^^^^^^^^^^^ required by this bound in `check_iter`
+5 | source_iter(()) -> for_each(std::mem::drop);
+ | ^^^^^^^^^^^^^^^ required by this bound in `check_iter`
-error[E0277]: `{integer}` is not an iterator
+error[E0277]: `()` is not an iterator
--> tests/compile-fail/surface_source_iter_badtype.rs:5:9
|
-5 | source_iter(5) -> for_each(std::mem::drop);
- | ^^^^^^^^^^^^^^ `{integer}` is not an iterator
+5 | source_iter(()) -> for_each(std::mem::drop);
+ | ^^^^^^^^^^^^^^^ `()` is not an iterator
|
- = help: the trait `Iterator` is not implemented for `{integer}`, which is required by `{integer}: IntoIterator`
- = note: required for `{integer}` to implement `IntoIterator`
+ = help: the trait `Iterator` is not implemented for `()`, which is required by `(): IntoIterator`
+ = note: required for `()` to implement `IntoIterator`
note: required by a bound in `check_iter`
--> tests/compile-fail/surface_source_iter_badtype.rs:5:9
|
-5 | source_iter(5) -> for_each(std::mem::drop);
- | ^^^^^^^^^^^^^^ required by this bound in `check_iter`
+5 | source_iter(()) -> for_each(std::mem::drop);
+ | ^^^^^^^^^^^^^^^ required by this bound in `check_iter`
-error[E0277]: `{integer}` is not an iterator
+error[E0277]: `()` is not an iterator
--> tests/compile-fail/surface_source_iter_badtype.rs:5:9
|
-5 | source_iter(5) -> for_each(std::mem::drop);
- | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `{integer}` is not an iterator
+5 | source_iter(()) -> for_each(std::mem::drop);
+ | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `()` is not an iterator
|
- = help: the trait `Iterator` is not implemented for `{integer}`, which is required by `{integer}: IntoIterator`
- = note: required for `{integer}` to implement `IntoIterator`
+ = help: the trait `Iterator` is not implemented for `()`, which is required by `(): IntoIterator`
+ = note: required for `()` to implement `IntoIterator`
note: required by a bound in `check_iter`
--> tests/compile-fail/surface_source_iter_badtype.rs:5:9
|
-5 | source_iter(5) -> for_each(std::mem::drop);
- | ^^^^^^^^^^^^^^ required by this bound in `check_iter`
+5 | source_iter(()) -> for_each(std::mem::drop);
+ | ^^^^^^^^^^^^^^^ required by this bound in `check_iter`
diff --git a/hydroflow/tests/snapshots/surface_cross_singleton__union_defer_tick@graphvis_dot.snap b/hydroflow/tests/snapshots/surface_cross_singleton__union_defer_tick@graphvis_dot.snap
new file mode 100644
index 000000000000..7957a90e7979
--- /dev/null
+++ b/hydroflow/tests/snapshots/surface_cross_singleton__union_defer_tick@graphvis_dot.snap
@@ -0,0 +1,158 @@
+---
+source: hydroflow/tests/surface_cross_singleton.rs
+expression: "df.meta_graph().unwrap().to_dot(& Default :: default())"
+---
+digraph {
+ node [fontname="Monaco,Menlo,Consolas,"Droid Sans Mono",Inconsolata,"Courier New",monospace", style=filled];
+ edge [fontname="Monaco,Menlo,Consolas,"Droid Sans Mono",Inconsolata,"Courier New",monospace"];
+ n1v1 [label="(n1v1) source_stream(cross_rx)", shape=invhouse, fillcolor="#88aaff"]
+ n2v1 [label="(n2v1) sort()", shape=invhouse, fillcolor="#88aaff"]
+ n3v1 [label="(n3v1) tee()", shape=house, fillcolor="#ffff88"]
+ n4v1 [label="(n4v1) defer_tick_lazy()", shape=invhouse, fillcolor="#88aaff"]
+ n5v1 [label="(n5v1) source_iter([0])", shape=invhouse, fillcolor="#88aaff"]
+ n6v1 [label="(n6v1) persist::<'static>()", shape=invhouse, fillcolor="#88aaff"]
+ n7v1 [label="(n7v1) union()", shape=invhouse, fillcolor="#88aaff"]
+ n8v1 [label="(n8v1) cross_singleton()", shape=invhouse, fillcolor="#88aaff"]
+ n9v1 [label="(n9v1) tee()", shape=house, fillcolor="#ffff88"]
+ n10v1 [label="(n10v1) for_each(|x| egress_tx.send(x).unwrap())", shape=house, fillcolor="#ffff88"]
+ n11v1 [label="(n11v1) fold(|| 0, |_, _| {})", shape=invhouse, fillcolor="#88aaff"]
+ n12v1 [label="(n12v1) cross_singleton()", shape=invhouse, fillcolor="#88aaff"]
+ n13v1 [label="(n13v1) fold(|| 0, |_, _| {})", shape=invhouse, fillcolor="#88aaff"]
+ n14v1 [label="(n14v1) flat_map(|_| [])", shape=invhouse, fillcolor="#88aaff"]
+ n15v1 [label="(n15v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n16v1 [label="(n16v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n17v1 [label="(n17v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n18v1 [label="(n18v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n19v1 [label="(n19v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n20v1 [label="(n20v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n21v1 [label="(n21v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n22v1 [label="(n22v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n23v1 [label="(n23v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n24v1 [label="(n24v1) identity()", shape=invhouse, fillcolor="#88aaff"]
+ n25v1 [label="(n25v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n2v1 -> n3v1
+ n1v1 -> n15v1
+ n3v1 -> n16v1
+ n4v1 -> n7v1 [label="0"]
+ n14v1 -> n17v1
+ n5v1 -> n6v1
+ n6v1 -> n7v1 [label="1"]
+ n7v1 -> n18v1
+ n8v1 -> n9v1
+ n9v1 -> n10v1
+ n9v1 -> n19v1
+ n3v1 -> n20v1
+ n11v1 -> n21v1
+ n13v1 -> n22v1
+ n12v1 -> n23v1
+ n15v1 -> n2v1 [color=red]
+ n16v1 -> n8v1 [label="input"]
+ n17v1 -> n24v1
+ n18v1 -> n8v1 [label="single", color=red]
+ n19v1 -> n11v1 [color=red]
+ n20v1 -> n12v1 [label="input"]
+ n21v1 -> n12v1 [label="single", color=red]
+ n22v1 -> n14v1
+ n23v1 -> n13v1 [color=red]
+ n24v1 -> n25v1
+ n25v1 -> n4v1 [color=red]
+ subgraph "cluster n1v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_1v1\nstratum 0"
+ n1v1
+ subgraph "cluster_sg_1v1_var_teed_in" {
+ label="var teed_in"
+ n1v1
+ }
+ }
+ subgraph "cluster n2v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_2v1\nstratum 1"
+ n2v1
+ n3v1
+ subgraph "cluster_sg_2v1_var_teed_in" {
+ label="var teed_in"
+ n2v1
+ n3v1
+ }
+ }
+ subgraph "cluster n3v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_3v1\nstratum 0"
+ n4v1
+ n5v1
+ n6v1
+ n7v1
+ subgraph "cluster_sg_3v1_var_persisted_stream" {
+ label="var persisted_stream"
+ n5v1
+ n6v1
+ }
+ subgraph "cluster_sg_3v1_var_unioned_stream" {
+ label="var unioned_stream"
+ n7v1
+ }
+ }
+ subgraph "cluster n4v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_4v1\nstratum 1"
+ n8v1
+ n9v1
+ n10v1
+ subgraph "cluster_sg_4v1_var_join" {
+ label="var join"
+ n8v1
+ n9v1
+ }
+ }
+ subgraph "cluster n5v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_5v1\nstratum 2"
+ n11v1
+ subgraph "cluster_sg_5v1_var_folded_thing" {
+ label="var folded_thing"
+ n11v1
+ }
+ }
+ subgraph "cluster n6v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_6v1\nstratum 3"
+ n12v1
+ subgraph "cluster_sg_6v1_var_joined_folded" {
+ label="var joined_folded"
+ n12v1
+ }
+ }
+ subgraph "cluster n7v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_7v1\nstratum 4"
+ n13v1
+ subgraph "cluster_sg_7v1_var_deferred_stream" {
+ label="var deferred_stream"
+ n13v1
+ }
+ }
+ subgraph "cluster n8v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_8v1\nstratum 4"
+ n14v1
+ subgraph "cluster_sg_8v1_var_deferred_stream" {
+ label="var deferred_stream"
+ n14v1
+ }
+ }
+ subgraph "cluster n9v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_9v1\nstratum 5"
+ n24v1
+ }
+}
diff --git a/hydroflow/tests/snapshots/surface_cross_singleton__union_defer_tick@graphvis_mermaid.snap b/hydroflow/tests/snapshots/surface_cross_singleton__union_defer_tick@graphvis_mermaid.snap
new file mode 100644
index 000000000000..ae77d69d57cc
--- /dev/null
+++ b/hydroflow/tests/snapshots/surface_cross_singleton__union_defer_tick@graphvis_mermaid.snap
@@ -0,0 +1,124 @@
+---
+source: hydroflow/tests/surface_cross_singleton.rs
+expression: "df.meta_graph().unwrap().to_mermaid(& Default :: default())"
+---
+%%{init:{'theme':'base','themeVariables':{'clusterBkg':'#ddd','clusterBorder':'#888'}}}%%
+flowchart TD
+classDef pullClass fill:#8af,stroke:#000,text-align:left,white-space:pre
+classDef pushClass fill:#ff8,stroke:#000,text-align:left,white-space:pre
+classDef otherClass fill:#fdc,stroke:#000,text-align:left,white-space:pre
+linkStyle default stroke:#aaa
+1v1[\"(1v1) source_stream(cross_rx)"/]:::pullClass
+2v1[\"(2v1) sort()"/]:::pullClass
+3v1[/"(3v1) tee()"\]:::pushClass
+4v1[\"(4v1) defer_tick_lazy()"/]:::pullClass
+5v1[\"(5v1) source_iter([0])"/]:::pullClass
+6v1[\"(6v1) persist::<'static>()"/]:::pullClass
+7v1[\"(7v1) union()"/]:::pullClass
+8v1[\"(8v1) cross_singleton()"/]:::pullClass
+9v1[/"(9v1) tee()"\]:::pushClass
+10v1[/"(10v1) for_each(|x| egress_tx.send(x).unwrap())"\]:::pushClass
+11v1[\"(11v1) fold(|| 0, |_, _| {})"/]:::pullClass
+12v1[\"(12v1) cross_singleton()"/]:::pullClass
+13v1[\"(13v1) fold(|| 0, |_, _| {})"/]:::pullClass
+14v1[\"(14v1) flat_map(|_| [])"/]:::pullClass
+15v1["(15v1) handoff"]:::otherClass
+16v1["(16v1) handoff"]:::otherClass
+17v1["(17v1) handoff"]:::otherClass
+18v1["(18v1) handoff"]:::otherClass
+19v1["(19v1) handoff"]:::otherClass
+20v1["(20v1) handoff"]:::otherClass
+21v1["(21v1) handoff"]:::otherClass
+22v1["(22v1) handoff"]:::otherClass
+23v1["(23v1) handoff"]:::otherClass
+24v1[\"(24v1) identity()"/]:::pullClass
+25v1["(25v1) handoff"]:::otherClass
+2v1-->3v1
+1v1-->15v1
+3v1-->16v1
+4v1-->|0|7v1
+14v1-->17v1
+5v1-->6v1
+6v1-->|1|7v1
+7v1-->18v1
+8v1-->9v1
+9v1-->10v1
+9v1-->19v1
+3v1-->20v1
+11v1-->21v1
+13v1-->22v1
+12v1-->23v1
+15v1--x2v1; linkStyle 15 stroke:red
+16v1-->|input|8v1
+17v1-->24v1
+18v1--x|single|8v1; linkStyle 18 stroke:red
+19v1--x11v1; linkStyle 19 stroke:red
+20v1-->|input|12v1
+21v1--x|single|12v1; linkStyle 21 stroke:red
+22v1-->14v1
+23v1--x13v1; linkStyle 23 stroke:red
+24v1-->25v1
+25v1--o4v1; linkStyle 25 stroke:red
+subgraph sg_1v1 ["sg_1v1 stratum 0"]
+ 1v1
+ subgraph sg_1v1_var_teed_in ["var teed_in "]
+ 1v1
+ end
+end
+subgraph sg_2v1 ["sg_2v1 stratum 1"]
+ 2v1
+ 3v1
+ subgraph sg_2v1_var_teed_in ["var teed_in "]
+ 2v1
+ 3v1
+ end
+end
+subgraph sg_3v1 ["sg_3v1 stratum 0"]
+ 4v1
+ 5v1
+ 6v1
+ 7v1
+ subgraph sg_3v1_var_persisted_stream ["var persisted_stream "]
+ 5v1
+ 6v1
+ end
+ subgraph sg_3v1_var_unioned_stream ["var unioned_stream "]
+ 7v1
+ end
+end
+subgraph sg_4v1 ["sg_4v1 stratum 1"]
+ 8v1
+ 9v1
+ 10v1
+ subgraph sg_4v1_var_join ["var join "]
+ 8v1
+ 9v1
+ end
+end
+subgraph sg_5v1 ["sg_5v1 stratum 2"]
+ 11v1
+ subgraph sg_5v1_var_folded_thing ["var folded_thing "]
+ 11v1
+ end
+end
+subgraph sg_6v1 ["sg_6v1 stratum 3"]
+ 12v1
+ subgraph sg_6v1_var_joined_folded ["var joined_folded "]
+ 12v1
+ end
+end
+subgraph sg_7v1 ["sg_7v1 stratum 4"]
+ 13v1
+ subgraph sg_7v1_var_deferred_stream ["var deferred_stream "]
+ 13v1
+ end
+end
+subgraph sg_8v1 ["sg_8v1 stratum 4"]
+ 14v1
+ subgraph sg_8v1_var_deferred_stream ["var deferred_stream "]
+ 14v1
+ end
+end
+subgraph sg_9v1 ["sg_9v1 stratum 5"]
+ 24v1
+end
diff --git a/hydroflow/tests/snapshots/surface_examples__example_1_simplest.snap b/hydroflow/tests/snapshots/surface_examples__example_1_simplest.snap
index fa30505c69cd..065e174f4cb3 100644
--- a/hydroflow/tests/snapshots/surface_examples__example_1_simplest.snap
+++ b/hydroflow/tests/snapshots/surface_examples__example_1_simplest.snap
@@ -12,4 +12,3 @@ Hello 6
Hello 7
Hello 8
Hello 9
-
diff --git a/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product@graphvis_dot.snap b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product@graphvis_dot.snap
index e8a4194cc706..9e81d0e6973d 100644
--- a/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product@graphvis_dot.snap
+++ b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product@graphvis_dot.snap
@@ -12,22 +12,18 @@ digraph {
n5v1 [label="(n5v1) map(SetUnionSingletonSet::new_from)", shape=invhouse, fillcolor="#88aaff"]
n6v1 [label="(n6v1) state::<'static, SetUnionHashSet>()", shape=invhouse, fillcolor="#88aaff"]
n7v1 [label="(n7v1) lattice_bimorphism(CartesianProductBimorphism::>::default(), lhs, rhs)", shape=invhouse, fillcolor="#88aaff"]
- n8v1 [label="(n8v1) lattice_reduce()", shape=invhouse, fillcolor="#88aaff"]
- n9v1 [label="(n9v1) for_each(|x| out_send.send(x).unwrap())", shape=house, fillcolor="#ffff88"]
+ n8v1 [label="(n8v1) for_each(|x| out_send.send(x).unwrap())", shape=house, fillcolor="#ffff88"]
+ n9v1 [label="(n9v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
n10v1 [label="(n10v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
- n11v1 [label="(n11v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
- n12v1 [label="(n12v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
n2v1 -> n3v1
n1v1 -> n2v1
n5v1 -> n6v1
n4v1 -> n5v1
- n3v1 -> n10v1
- n6v1 -> n11v1
- n8v1 -> n9v1
- n7v1 -> n12v1
- n10v1 -> n7v1 [label="0"]
- n11v1 -> n7v1 [label="1"]
- n12v1 -> n8v1 [color=red]
+ n3v1 -> n9v1
+ n6v1 -> n10v1
+ n7v1 -> n8v1
+ n9v1 -> n7v1 [label="0"]
+ n10v1 -> n7v1 [label="1"]
n3v1 -> n7v1 [color=red]
n6v1 -> n7v1 [color=red]
subgraph "cluster n1v1" {
@@ -63,22 +59,11 @@ digraph {
style=filled
label = "sg_3v1\nstratum 1"
n7v1
+ n8v1
subgraph "cluster_sg_3v1_var_my_join" {
label="var my_join"
n7v1
- }
- }
- subgraph "cluster n4v1" {
- fillcolor="#dddddd"
- style=filled
- label = "sg_4v1\nstratum 1"
- n8v1
- n9v1
- subgraph "cluster_sg_4v1_var_my_join" {
- label="var my_join"
n8v1
- n9v1
}
}
}
-
diff --git a/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product@graphvis_mermaid.snap b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product@graphvis_mermaid.snap
index 9a8e0c4a1c46..60c4b26b33f7 100644
--- a/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product@graphvis_mermaid.snap
+++ b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product@graphvis_mermaid.snap
@@ -15,24 +15,20 @@ linkStyle default stroke:#aaa
5v1[\"(5v1) map(SetUnionSingletonSet::new_from)"/]:::pullClass
6v1[\"(6v1) state::<'static, SetUnionHashSet<u32>>()"/]:::pullClass
7v1[\"(7v1) lattice_bimorphism(CartesianProductBimorphism::<HashSet<_>>::default(), lhs, rhs)"/]:::pullClass
-8v1[\"(8v1) lattice_reduce()"/]:::pullClass
-9v1[/"(9v1) for_each(|x| out_send.send(x).unwrap())"\]:::pushClass
+8v1[/"(8v1) for_each(|x| out_send.send(x).unwrap())"\]:::pushClass
+9v1["(9v1) handoff"]:::otherClass
10v1["(10v1) handoff"]:::otherClass
-11v1["(11v1) handoff"]:::otherClass
-12v1["(12v1) handoff"]:::otherClass
2v1-->3v1
1v1-->2v1
5v1-->6v1
4v1-->5v1
-3v1-->10v1
-6v1-->11v1
-8v1-->9v1
-7v1-->12v1
-10v1-->|0|7v1
-11v1-->|1|7v1
-12v1-->8v1; linkStyle 10 stroke:#060
-3v1--x7v1; linkStyle 11 stroke:red
-6v1--x7v1; linkStyle 12 stroke:red
+3v1-->9v1
+6v1-->10v1
+7v1-->8v1
+9v1-->|0|7v1
+10v1-->|1|7v1
+3v1--x7v1; linkStyle 9 stroke:red
+6v1--x7v1; linkStyle 10 stroke:red
subgraph sg_1v1 ["sg_1v1 stratum 0"]
1v1
2v1
@@ -55,16 +51,9 @@ subgraph sg_2v1 ["sg_2v1 stratum 0"]
end
subgraph sg_3v1 ["sg_3v1 stratum 1"]
7v1
+ 8v1
subgraph sg_3v1_var_my_join ["var my_join "]
7v1
- end
-end
-subgraph sg_4v1 ["sg_4v1 stratum 1"]
- 8v1
- 9v1
- subgraph sg_4v1_var_my_join ["var my_join "]
8v1
- 9v1
end
end
-
diff --git a/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_1401@graphvis_dot.snap b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_1401@graphvis_dot.snap
new file mode 100644
index 000000000000..a1f6888a2b86
--- /dev/null
+++ b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_1401@graphvis_dot.snap
@@ -0,0 +1,69 @@
+---
+source: hydroflow/tests/surface_lattice_bimorphism.rs
+expression: "df.meta_graph().unwrap().to_dot(& Default :: default())"
+---
+digraph {
+ node [fontname="Monaco,Menlo,Consolas,"Droid Sans Mono",Inconsolata,"Courier New",monospace", style=filled];
+ edge [fontname="Monaco,Menlo,Consolas,"Droid Sans Mono",Inconsolata,"Courier New",monospace"];
+ n1v1 [label="(n1v1) source_iter(0..1)", shape=invhouse, fillcolor="#88aaff"]
+ n2v1 [label="(n2v1) map(SetUnionSingletonSet::new_from)", shape=invhouse, fillcolor="#88aaff"]
+ n3v1 [label="(n3v1) state::<'static, SetUnionHashSet>()", shape=invhouse, fillcolor="#88aaff"]
+ n4v1 [label="(n4v1) source_iter(1..2)", shape=invhouse, fillcolor="#88aaff"]
+ n5v1 [label="(n5v1) map(SetUnionSingletonSet::new_from)", shape=invhouse, fillcolor="#88aaff"]
+ n6v1 [label="(n6v1) state::<'static, SetUnionHashSet>()", shape=invhouse, fillcolor="#88aaff"]
+ n7v1 [label="(n7v1) lattice_bimorphism(CartesianProductBimorphism::>::default(), lhs, rhs)", shape=invhouse, fillcolor="#88aaff"]
+ n8v1 [label="(n8v1) for_each(|x| out_send.send(x).unwrap())", shape=house, fillcolor="#ffff88"]
+ n9v1 [label="(n9v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n10v1 [label="(n10v1) handoff", shape=parallelogram, fillcolor="#ddddff"]
+ n2v1 -> n3v1
+ n1v1 -> n2v1
+ n5v1 -> n6v1
+ n4v1 -> n5v1
+ n3v1 -> n9v1
+ n6v1 -> n10v1
+ n7v1 -> n8v1
+ n9v1 -> n7v1 [label="0"]
+ n10v1 -> n7v1 [label="1"]
+ n3v1 -> n7v1 [color=red]
+ n6v1 -> n7v1 [color=red]
+ subgraph "cluster n1v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_1v1\nstratum 0"
+ n1v1
+ n2v1
+ n3v1
+ subgraph "cluster_sg_1v1_var_lhs" {
+ label="var lhs"
+ n1v1
+ n2v1
+ n3v1
+ }
+ }
+ subgraph "cluster n2v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_2v1\nstratum 0"
+ n4v1
+ n5v1
+ n6v1
+ subgraph "cluster_sg_2v1_var_rhs" {
+ label="var rhs"
+ n4v1
+ n5v1
+ n6v1
+ }
+ }
+ subgraph "cluster n3v1" {
+ fillcolor="#dddddd"
+ style=filled
+ label = "sg_3v1\nstratum 1"
+ n7v1
+ n8v1
+ subgraph "cluster_sg_3v1_var_my_join" {
+ label="var my_join"
+ n7v1
+ n8v1
+ }
+ }
+}
diff --git a/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_1401@graphvis_mermaid.snap b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_1401@graphvis_mermaid.snap
new file mode 100644
index 000000000000..08e6a806fc42
--- /dev/null
+++ b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_1401@graphvis_mermaid.snap
@@ -0,0 +1,59 @@
+---
+source: hydroflow/tests/surface_lattice_bimorphism.rs
+expression: "df.meta_graph().unwrap().to_mermaid(& Default :: default())"
+---
+%%{init:{'theme':'base','themeVariables':{'clusterBkg':'#ddd','clusterBorder':'#888'}}}%%
+flowchart TD
+classDef pullClass fill:#8af,stroke:#000,text-align:left,white-space:pre
+classDef pushClass fill:#ff8,stroke:#000,text-align:left,white-space:pre
+classDef otherClass fill:#fdc,stroke:#000,text-align:left,white-space:pre
+linkStyle default stroke:#aaa
+1v1[\"(1v1) source_iter(0..1)"/]:::pullClass
+2v1[\"(2v1) map(SetUnionSingletonSet::new_from)"/]:::pullClass
+3v1[\"(3v1) state::<'static, SetUnionHashSet<u32>>()"/]:::pullClass
+4v1[\"(4v1) source_iter(1..2)"/]:::pullClass
+5v1[\"(5v1) map(SetUnionSingletonSet::new_from)"/]:::pullClass
+6v1[\"(6v1) state::<'static, SetUnionHashSet<u32>>()"/]:::pullClass
+7v1[\"(7v1) lattice_bimorphism(CartesianProductBimorphism::<HashSet<_>>::default(), lhs, rhs)"/]:::pullClass
+8v1[/"(8v1) for_each(|x| out_send.send(x).unwrap())"\]:::pushClass
+9v1["(9v1) handoff"]:::otherClass
+10v1["(10v1) handoff"]:::otherClass
+2v1-->3v1
+1v1-->2v1
+5v1-->6v1
+4v1-->5v1
+3v1-->9v1
+6v1-->10v1
+7v1-->8v1
+9v1-->|0|7v1
+10v1-->|1|7v1
+3v1--x7v1; linkStyle 9 stroke:red
+6v1--x7v1; linkStyle 10 stroke:red
+subgraph sg_1v1 ["sg_1v1 stratum 0"]
+ 1v1
+ 2v1
+ 3v1
+ subgraph sg_1v1_var_lhs ["var lhs "]
+ 1v1
+ 2v1
+ 3v1
+ end
+end
+subgraph sg_2v1 ["sg_2v1 stratum 0"]
+ 4v1
+ 5v1
+ 6v1
+ subgraph sg_2v1_var_rhs ["var rhs "]
+ 4v1
+ 5v1
+ 6v1
+ end
+end
+subgraph sg_3v1 ["sg_3v1 stratum 1"]
+ 7v1
+ 8v1
+ subgraph sg_3v1_var_my_join ["var my_join "]
+ 7v1
+ 8v1
+ end
+end
diff --git a/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_tick_state@graphvis_dot.snap b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_tick_state@graphvis_dot.snap
index 8e00b09a5bba..239b30948c41 100644
--- a/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_tick_state@graphvis_dot.snap
+++ b/hydroflow/tests/snapshots/surface_lattice_bimorphism__cartesian_product_tick_state@graphvis_dot.snap
@@ -12,24 +12,20 @@ digraph {
n5v1 [label="(n5v1) map(SetUnionSingletonSet::new_from)", shape=invhouse, fillcolor="#88aaff"]
n6v1 [label="(n6v1) state::<'tick, SetUnionHashSet>()", shape=invhouse, fillcolor="#88aaff"]
n7v1 [label="(n7v1) lattice_bimorphism(CartesianProductBimorphism:: 
- Hydroflow
+