Using connect on an empty Stream() breaks backpressure

[roveo]

Here's an example:

from time import sleep
from streamz import Stream


class Counter:
    def __init__(self):
        self.n = 0

    def __call__(self, x):
        self.n += 1
        return x

    def __str__(self):
        return f"Counter({self.n})"


cin = Counter()
cout = Counter()

src = Stream.from_textfile("1000_lines.txt")
out = Stream()
out.map(cin).rate_limit(1).sink(cout)
src.connect(out)

src.start()
sleep(1)
print(cin, cout)
assert cin.n < 1000  # AssertionError

I expected this to work the same way as src.map(...).rate_limit(...).sink(...), but yield _emit() in the source doesn't block. If you go into rate_limit.update, you'll see that it gets called multiple times, so data actually accumulates in the form of running blocked coroutines.

You can do out = Stream().map(cin) and then in will work fine, but if you want to additionally emit something from out before the first map (so that it is counted by cin), this isn't a solution.

[martindurant]

Sorry, are you saying that the existence of the line

src.connect(out)

makes the difference in the behaviour of the variable cin, on the other branch of the stream?

I do note that Counter ought to be a subclass of Stream. I imagine that matters.

[roveo]

I do note that Counter ought to be a subclass of Stream. I imagine that matters.

Why? It's just a callable that's passed to map.

Sorry, are you saying that the existence of the line src.connect(out) makes the difference in the behaviour of the variable cin, on the other branch of the stream?

No, that wasn't what I was trying to demonstrate. The counters are there just to show that backpressure doesn't work as expected, but the example would still work the same without them.

If you write it like this:

src = Stream.from_textfile("1000_lines.txt")
out = Stream()
out.map(cin).rate_limit(1).sink(cout)
src.connect(out)

and then start the pipeline, the asynchronous source (from_file) will immediately emit 1000 messages, while I expect them to be emitted one by one every second due to the existence of backpressure.

The counters in my example are to make it clear that 1000 messages passed through .map(cin) (cin stands for "count in"), but only one message went to .sink(cout) (count out). So these messages in this case are somewhere between the 1st and the 2nd .map nodes, namely they are accumulated as running coroutines rate_limit.update.

[roveo]

Here are a couple of tests that demonstrate my point/question (you can copy-paste them into test_core.py)

Test that passes:

def test_backpressure_without_connect():
    @Stream.register_api()
    class from_list(Stream):

        def __init__(self, source, **kwargs):
            self.source = source
            super().__init__(ensure_io_loop=True, **kwargs)

        def start(self):
            self.stopped = False
            self.loop.add_callback(self.run)

        @gen.coroutine
        def run(self):
            while not self.stopped and len(self.source) > 0:
                yield self._emit(self.source.pop(0))

    source_list = [0, 1, 2, 3, 4]

    source = Stream.from_list(source_list)
    L = source.rate_limit(1).sink_to_list()

    source.start()

    sleep(0.1)

    assert L == [0]
    assert len(source_list) > 0

Test that doesn't, but I expected it to:

def test_backpressure_connect_empty_stream():
    @Stream.register_api()
    class from_list(Stream):

        def __init__(self, source, **kwargs):
            self.source = source
            super().__init__(ensure_io_loop=True, **kwargs)

        def start(self):
            self.stopped = False
            self.loop.add_callback(self.run)

        @gen.coroutine
        def run(self):
            while not self.stopped and len(self.source) > 0:
                yield self._emit(self.source.pop(0))

    source_list = [0, 1, 2, 3, 4]

    source = Stream.from_list(source_list)
    sout = Stream()
    L = sout.rate_limit(1).sink_to_list()

    source.connect(sout)

    source.start()

    sleep(0.1)

    assert L == [0]
    assert len(source_list) > 0  # this fails

As you can see, in the first scenario the source list is not consumed entirely because of backpressure, but in the 2nd it is, and it is not obvious why that would be the case. (Maybe because I don't have a good mental model of what Stream() actually is.)

This is not necessarily a bug, just unclear/unexpected behavior, maybe this should just be addressed with updating documentation with "don't do this".