A problem with a self-referential Switch

[kevinbaldor]

I have attempted to write a sieve of Eratosthenes as a test of self-modifying FRP. The idea is that when fed with a stream of the integers from 2 up, it will emit the series of prime numbers. It works as follows:

Initially, the output stream is defined as equal to the input
input -> output
Whenever any value arrives at the output, a new stage is defined that takes that value and filters any even multiple of it. So as values are passed in it grows like
2: input -> output: yields 2, adds sieve(2) stage
3: input -> sieve(2) -> output: yields 3, adds sieve(3) stage
4: input -> sieve(2) -> sieve(3) -> output: yields nothing (the value is filtered by sieve(2))
5: input -> sieve(2) -> sieve(3) -> output: yields 5, adds sieve(5) stage

and so on.

Unfortunately, this isn't working for me. I've attempted to strip it down to the minimal example that demonstrates the issue at https://github.com/kevinbaldor/sodium-rust-test/blob/main/src/main.rs.

I've inserted listeners with println! statements to demonstrate that subscribing to the output works as expected when taking snapshots of normal cells (demonstrated with a constant), but the snapshot of the CellLoop that contains the output stream doesn't seem to work, the println! statements don't appear and the Cell::switch_s never changes from its initial value (the input).

It feels like this is something that should be possible. Is this a bug in sodium-rust, or am I using it incorrectly?

[the-real-blackh]

Hi Kevin,

I translated it into Java (because I am not good with Rust), and I encountered a bug with switchS failing at runtime because its argument was a loop. The issue here is that the implementation for switchS is not able to deal with loops. So that's a problem, and I'll come back to that.

So I changed the code so that we are looping a stream instead of a cell:

import nz.sodium.*;

public class Sieve2
{
    public static void main(String[] args)
    {
        StreamSink<Integer> input = new StreamSink<Integer>();
        Stream<Integer> output_ = Transaction.run(() -> {
            StreamLoop<Stream<Integer>> new_output = new StreamLoop<Stream<Integer>>();
            Cell<Stream<Integer>> output_cell = new_output.hold(input);
            Stream<Integer> output = Cell.switchS(output_cell);
            Cell constant = new Cell<Integer>(100);

            new_output.loop(output.snapshot(output_cell,
                (prime, old_output) -> {
                    System.out.println("Adding filter stage for "+Integer.toString(prime));
                    return old_output.filter((x) -> (x % prime) != 0);
                }));

            return output;
        });

        Listener l = output_.listen((x) -> { System.out.println("output "+Integer.toString(x)); });

        for (int x = 2; x <= 20; x++) {
            System.out.println("Sending "+Integer.toString(x));
            input.send(x);
            System.out.println("");
        }

        l.unlisten();
    }
}

This works:

Sending 2
Adding filter stage for 2
output 2

Sending 3
Adding filter stage for 3
output 3

Sending 4

Sending 5
Adding filter stage for 5
output 5

Sending 6

Sending 7
Adding filter stage for 7
output 7

Sending 8

Sending 9

Sending 10

Sending 11
Adding filter stage for 11
output 11

Sending 12

Sending 13
Adding filter stage for 13
output 13

Sending 14

Sending 15

Sending 16

Sending 17
Adding filter stage for 17
output 17

Sending 18

Sending 19
Adding filter stage for 19
output 19

Sending 20

So, as a first step, please try this. Maybe the Rust implementation has a similar bug but fails differently.

[kevinbaldor]

Thank you for taking the time to convert. I'm glad to see that there is a way to do this in the reference implementation. Unfortunately, converting your Java code to Rust doesn't work either. It behaves exactly as the CellLoop version.

Here is my Rust version of your StreamLoop solution (with all references to the constant removed).

fn main() {
    let sodium_ctx    = SodiumCtx::new();

    let input : StreamSink<i64> = sodium_ctx.new_stream_sink();

    let output = sodium_ctx.transaction(|| {
        let new_output:  StreamLoop<Stream<i64>> = sodium_ctx.new_stream_loop();
        let output_cell: Cell<Stream<i64>> = new_output.stream().hold(input.stream());
        let output: Stream<i64> = Cell::switch_s(&output_cell);

       new_output.loop_(&output.snapshot(&output_cell,
            |prime: &i64, old_output: &Stream<i64>|
            {
                let prime = *prime;
                println!("Adding filter stage for {}",prime);
                old_output.filter(move |x: &i64| (*x%prime)!=0)
            }));

        output });

    let l = output.listen(|x:&i64| println!("output {}",x));

    for x in 2..20 {
        println!("Sending {}",x);
        input.send(x);
        println!();
    }
    l.unlisten();
}

[clinuxrulz]

I believe there may be a node-cycle. (The engine might think output needs to be updated before output in the transacction.) I left you a few comments on github. Will look into it further later.

Also remember the current switch_s test for sodium. The resulting stream behaves like the new stream supplied by the cell after the current transaction. Because the value of a cell changes after the transaction but before any new transactions. So I expect the values flowing through are from the stream before the switch has finished within the current transaction.

It might be possible to use Operational::defer to patch up the node cycle, if there is one.

[clinuxrulz]

Changing:

output_cell_loop.loop_(&new_output.hold(input.stream()));

To:

output_cell_loop.loop_(&Operational::defer(&new_output).hold(input.stream()));

Then listen to new_output at the end instead of output to avoid the after-transaction switching.

In your original code will probably do the trick. (Untested)

Seems like the node cycle between output and output just need to be broken. And defer will make output only update output for the next transaction.

[clinuxrulz]

A not-so-elegant but working solution for Rust. I do believe their is a node-cycle between output and itself that needs to be broken, so I ended up using post() to break the cycle.

fn main() {
    let sodium_ctx = SodiumCtx::new();
    let sodium_ctx = &sodium_ctx;
    let ss_input : StreamSink<i64>;
    let s_output_2;
    {
        let _t = sodium_ctx.new_transaction();

        ss_input = sodium_ctx.new_stream_sink();
        let ss_s_output : StreamSink<Stream<i64>> = sodium_ctx.new_stream_sink();
        let c_s_output = ss_s_output.stream().hold(ss_input.stream());
        let s_output = Cell::switch_s(&c_s_output);
        let s_output_next = s_output
            .snapshot(
                &c_s_output,
                |prime: &i64, old_s_output: &Stream<i64>| {
                    let prime = *prime;
                    old_s_output.filter(move |x: &i64| (*x%prime)!=0)
                }
            );
        let keep_alive;
        {
            let sodium_ctx = sodium_ctx.clone();
            let ss_s_output = ss_s_output.clone();
            keep_alive = s_output_next.listen_weak(move |s_output_next: &Stream<i64>| {
                let s_output_next = s_output_next.clone();
                let ss_s_output = ss_s_output.clone();
                sodium_ctx.post(move || {
                    ss_s_output.send(s_output_next.clone());
                });
            });
        }
        s_output_2 = s_output.map(move |x: &i64| {
            let _keep_alive = &keep_alive; // <-- keeps the weak listener alive for the lifespan of s_output_2
            return *x;
        });
    }

    let l = s_output_2.listen(|x:&i64| println!("output {}",x));

    for x in 2..20 {
        println!("Sending {}",x);
        ss_input.send(x);
        println!();
    }
    l.unlisten();
}

It should of also been possible to use Operational::defer as a work around rather than the dirty hack above. I believe a bug must be lurking around somewhere in sodium-rust still.

---

Edit: I've resolve the bug in sodium-rust version 2.1.1. Operational::defer must be used to break the node-cycle.

By node-cycle I mean.

Sodium does not like this: (conceptually)

x = x + 1

But sodium is OK with this:

x_new = x_old + 1 <-- x_old is before current transaction, so it is OK.

That is why we need to use Operational::defer.

See test:

SodiumFRP/sodium-rust/blob/dad65115ba3b627b7e9250188789ecfe12905839/src/tests.rs#L1504

[kevinbaldor]

But doesn't snapshot already return x_old?

That is, accumulators can be implemented as (in sodium pseudocode)

x = input.snapshot(x, (a,b) -> a+b).hold(0)

It seems that this should be analogous, no?

If I understand correctly, defer actually resolves in a later transaction. That makes no difference in this case, but in a larger reactive network, it might matter that the transaction finishes without having updated the output. I suspect that the introduction of additional defers could ultimately produce the correct behavior, but it would be nice to not have to to worry about that if the semantics of snapshot should already handle it.

[clinuxrulz]

Yes. But there was also a non-snapshot cycle.

Normally a cellloop / streamloop is constructing a forward reference to the actual output. But in your case your feeding the new_output stream back into output which is defined in terms of output.

Conceptually you have:

output = output.filter(...).filter(...)...

For the expression a.snapshot(b,...) the snapshot protects you for a cycle through b, because it is just calling sample() on b internally during transactions. But does not protect you for a cycle through a.

So here is the devil in the original code:

let new_output = output.snapshot(&output_cell_loop.cell(),

Followed by

output_cell_loop.loop_(&new_output.hold(input.stream()))

That's where the cycle occurred. "CYCLE HERE".snapshot("NOT HERE",...).

Using defer can be kind of magic, you can change new into old.

Conceptually:

new_output = output.filter(...)
output_stream_loop.loop(Operational::defer(new_output));

Defer is like an old function (the stream before current transaction) "old(new_output) is output" --> "defer(new_output) is output".

[clinuxrulz]

Honestly though defer is not a magical old() operator normally.

Just works for your use case because:

output has type Stream<i64>
new_output has type Stream<Stream<i64>>

and we want output to be a forward reference to the output of the current transaction, not what it would be after the current transaction. So new_output needs to be deferred before it is feed through the loop and switched into output.

[clinuxrulz]

There might be a little more to the story. Because switch_s behaves like the new network after the current transaction, making it a little like a defer in itself. So maybe the Rust version can work without the Operational::defer for the original pattern. I'll need to give this more thought.

---

However the initial stream produced by switch_s has no choice but to be the 1st one, and that first one will have a node-cycle. This is tricky stuff.