Notes from October 25 Hackathon

[jhellerstein]

What is this Discussion?

Everybody participating in the Hackathon should use this as a shared whiteboard of ideas or notes that we want to remember from today.

[jhellerstein]

demux_enum please!

We should prioritize demux_enum in hydro_lang. This is nearly the first thing anybody does when writing a server: handle input streams and partition them by command. Currently requires clone which has a bad code smell and replicates each item as many times as there are enum variants!

[MingweiSamuel]

#2138

[jhellerstein]

KeyedStream::join por favor!

We seem to be missing an API for join on KeyedStream

[jhellerstein]

State management is ugly

There's still really no elegant way to build a stateful thing (say a HashMap or a FSM) and probe its state:

• fold followed by join seems like the right pattern but doesn't really work
• even if it did work, it's implying that the state is accumulated twice: ones in the fold, once on one side of the join
• unclear how to delete from state interactively

[MingweiSamuel]

Need a way to have a global client (e.g. http client) which can be called from within Hydro

[MingweiSamuel]

Singleton .resolve_futures does not exist for a one-off API call (client is a singleton, making the call via client_singleton.clone().map(q!(|client| client.do_something_async()))

[MingweiSamuel]

Rewrite for error messages that show the correct path is broken due to moving things to examples

[MingweiSamuel]

Hydro wants a semi-naive (no replay) cross_product, but DFIR 'static,'static currently replays, so multiset_delta is used. We should eliminate the replay.

[jhellerstein]

for_each_commutative

• for_each should at minimum return a helpful error suggesting the use of assume_ordering
• would be nice to have a convenience operator like for_each_commutative

[MingweiSamuel]

desk_sink_commutative maybe would be nice similarly

[jhellerstein]

Error messages for Location mismatch

Helpful error messages would make a big difference, especially for Tick mismatches

[MingweiSamuel]

dest_sink shouldn't require Pin

[jhellerstein]

KeyedSingleton is a non-intuitive name for HashMap

We should think about renaming our live collections, especially Singletons/Optionals, in the context of familiar programming metaphors.

[MingweiSamuel]

Need new default export remaps:

stageleft::add_private_reexport(vec!["futures_util", "sink", "unfold"], vec!["futures_util", "sink"]);
stageleft::add_private_reexport(vec!["tokio", "fs", "file"], vec!["tokio", "fs"]);

[shadaj]

and should improve error messages if we can detect that a re-export is needed

[jhellerstein]

We need "outer joins" for KeyedSingleton.get

Should be a version that returns Option results on the get_* functions.

[jhellerstein]

fold initialization with Default: better error messages

I was puzzled by the fact that my KVS<V> that stores V's was compiling but failing at runtime due to

.fold(q!(|| V::default()), q!(|acc, i| *acc = i));

Turns out that Claude knew that I could use Default::default():

.fold(q!(|| Default::default()), q!(|acc, i| *acc = i));

That makes sense to me based on what I know about q!, but it'd be helpful to have better messages than:

error[E0433]: failed to resolve: use of undeclared type `V`
   --> examples/_process_0_549E49F2.rs:852:48
    |
852 | ...                   || V::default()
    |                          ^ use of undeclared type `V`
    ```

[jhellerstein]

The Tick metaphor is a bit misleading

A "tick" is a single discrete unit of time. But a Hydro Tick is in fact a generator of a sequence of said units of time. We probably mean something like Clock. But that word scares people who don't really know distributed systems (and even I don't love the idea of multiple clocks on a single node).

"Window" is another term of art we could borrow; Hydro Ticks are "tumbling" windows. But that word scares people who don't really know streaming systems.

Maybe it should be called a Ticker?

More generally, we should figure out our temporal nomenclature and make it all very consistent. Right now we have a bunch of stuff like Tick and batch and snapshot and atomic and it's hard to know as a developer how they interrelate. And as a researcher it's hard to know if they are some necessary and sufficient set of constructs to cover some particular set of needs; the needs themeselves aren't well defined.

[MingweiSamuel]

take N is missing

[MingweiSamuel]

Workaround: sort -> enumerate -> filter

[shadaj]

Need top-level source_iter and singleton to emit bounded collections.

[shadaj]

Support .inspect inside a tick as a way to debug (when for_each is unavailable)

[MingweiSamuel]

Local variable needs to have move closure for a usize for unclear reasons

const MAX_SUMMONERS_TO_UPDATE: usize = 100;
    let summoners_to_update = some_stream
        // TODO(mingwei): need take N operator.
        .enumerate()
        .filter(q!(move |&(i, _)| i < MAX_SUMMONERS_TO_UPDATE)) // `move` required for some reason
        .map(q!(|(_, s)| s));

[shadaj]

Support directly borrowing bounded singletons into closures to avoid cross_singleton.

[jhellerstein]

Some thoughts on ergonomics

Here's an idealized (fake) local KVS

fn local_kvs<'a, V>(
    operations: Stream<KVSOperation<V>, Process<'a>, Unbounded>,
    server_process: &Process<'a>,
) ->
    Stream<KVSResult<V>, Process<'a>, Unbounded>,
{

    let put_stream = Stream<(String, V)>.new();
    let get_stream = Stream<String>.new();
    let ht = HashMap<String, V>::new();

    // PUTs go to storage and generate Acks
    // GETs generate Matches
    operations.demux_enum {
        KVSOperation:Put(k, v) => {
            ht.insert(k,v));
            KVSResult::new(KVSResultType::Ack, k, v)
        }
        KVSOperation:Get(k) => {
            get_stream
            .query(ht)
            .map(|(k, v)| KVSResult::new(KVSResultType::Match, k, v))
        }
    }
}

and here is my current (working) version that uses some helper functions (full version here):

fn local_kvs<'a>(
        operations: Stream<KVSOperation<V>, Process<'a>, Unbounded>,
        server_process: &Process<'a>,
    ) -> (
        Stream<(String, V), Process<'a>, Unbounded>,
        Stream<String, Process<'a>, Unbounded>,
    )
    where
        V: Default,
    {
        // we will need ticks for batching.
        let ticker = &server_process.tick();

        let ht = Self::ht_build(operations.clone());
        let gets = Self::batch_gets(
            operations
                .clone()
                .batch(ticker, nondet!(/** gets are non-deterministic */)),
        );
        return Self::ht_query(
            gets,
            ht.snapshot(ticker, nondet!(/** gets are non-deterministic! */)),
        );
    }
}

Some thoughts on simplifying things:

1. State: I think it's very important for us to make a familiar syntax sugar for state. Maybe declaring live variants of familiar Rust collections, like Live<HashMap<K, V>> that have streaming inputs for insertion, and streaming "queries" and "outputs".
   • HashMaps and BTrees would be nice, I'd also like FSMs. What else? How Rust-like and ergonomic can we make that?
2. Tickless Default: I think the need for explicit tick alignment is the rare case and we can offer defaults for ticks that don't require devs to think about them 80% of the time.
   • We could allow top-level Unbounded stuff to race as along as the races insist on nondet!() where needed. For example, "queries" on "state" like the example above shouldn't need explicit ticks, just a nondet! on the query.
   • I bet the next 10% of cases would work with a Dedalus/Bloom-like, single clock "event loop". This is quite a natural concept in distributed systems
   • The next 5% is probably nested "event loops", maybe we could get away with a scoping syntax
   • If I'm right that leaves 5% of the time for explicit named ticks

[shadaj]

So, what you are describing for Live is exactly Singleton! Which raises the question... is Live a better name?

[jhellerstein]

Well it is and it isn't...in the sense that a Rust programmer wanting to use a HashMap or an FSM won't know very immediately how to use a Singleton in place of their favorite data structure. My suggestion ... very much inspired by Shadaj's initial work on hydro_lang ... is that we work backward™️ from "how would you do this in plain Rust" and see how close we can get to that in Hydro without massive surgery.

There's a lot of quality-of-life DX stuff here that we need to start tackling, and I'm trying to take the stance that expectations of "plain Rust" are a default we should try to embrace. My hope is that a lot of this will be easy to implement as syntax sugar.

Among the things I hid in helpers in my code above:

• our workaround for the lack of demux (which will ideally look like match as much as possible)
• HashMap insertion via .into_keyed().fold(q!(|| Default::default()), q!(|acc, i| *acc = i)) which is pure boilerplate
• HashMap::get via .get_many_if_present(key_batch.clone().into_keyed()).entries()
• Singleton get with Option results to support failed gets
• all_ticks cllas to get out of the Tick for get, and get back to top level (I would have hidden the batch/snapshot calls in helpers too, but @shadaj told me it's a bad code smell to pass a Tick as a parameter to a function

My thought is we assess what aspects of those helpers are common-case usage, and make some or all of them defaults. We of course leave the full expressive power around as well for the remaining 15% of cases to be done in special-case ways.

[MingweiSamuel]

@jhellerstein

If I clone a live collection and batch the two different copies using the same Tick, are they guaranteed to pick identical batching?

[jhellerstein]

Kiro read the docs but still assumed that Hydro's cluster abstraction would include automatic replication across the cluster. Not an unreasonable guess for a human. We should clarify what you do and don't get from cluster in the docs.

[jhellerstein]

Let's make sure we have good steering docs in the Hydro repo.