Provide some standard mechanism for splitting a stream into lines, and other basic protocol tasks

[njsmith]

Most networking libraries provide some standard way to implement basic protocol building blocks like "split a stream into lines", "read exactly N bytes", or "split a stream into length-prefixed frames", e.g.:

• asyncio StreamReader.readline, StreamReader.readexactly, StreamReader.readuntil

• The classes in twisted.protocols.basic

• The stdlib socket module's makefile method, that lets you get access to the full Python file API, including readline and friends

• Tornado IOStream's read_until

We don't have anything like this currently, as I was reminded by this StackOverflow question from @basak.

Note: if you're just looking for a quick way to read lines from a trio Stream, then click on that SO link, it has an example.

Use cases

• Simple protocols used in tutorials, to make it easy for beginners to get something working
• A base for implementing substantial/standard protocols that happen to use one of these framing methods. This mostly applies to the line-based framing, e.g. twisted's LineReceiver and LineOnlyReceiver have subclasses implementing HTTP, IMAP, POP3, SMTP, Ident, Finger, FTP, Memcache, IRC, ... you get the idea.
• Inventing little private mini-protocols where you don't want to have to build basic framing from scratch. I think a lot of cases that used to use this kind of thing nowadays use HTTP or WebSocket or ZeroMQ, but it still comes up occasionally. This mostly involves the length-prefixed framing variants (e.g. twisted AMP subclasses Int16Receiver), though sometimes it involves lines, e.g. newline-terminated JSON, or the log parser in linehaul.
• If you have to script an interactive subprocess that was never designed to be scripted, then readline and read_until are pretty useful. This particular case can also benefit from more sophisticated tools, like TTY emulation and pexpect-style pattern matching.

Considerations

Our approach shouldn't involve adding new methods to Stream, because the point of the Stream interface is to allow for lots of different implementions, and we don't want to force everyone who implements Stream to have to reimplement their own version of the standard frame-splitting algorithms. So this should be some helper function that acts on a Stream, or wrapper class that has-a Stream, something like that.

For "real" protocols like HTTP, you definitely can implement them on top of explicit (async) blocking I/O operations like readline and read_exactly, but these days I'm pretty convinced that you will be happier using Sans I/O. Some of the arguments for sans-io design are kind of pure and theoretical, like "better modularity" and "higher reusability", but having done this twice now (with h11 and wsproto), I really don't feel like it's an eat-your-vegetables thing – the benefits are super practical: like, you can actually understand your protocol code, and test it, and people with totally different use cases show up to fix bugs for you. It's just a more pleasant way to do things.

OTOH, while trio is generally kind of opinionated and we should give confused users helpful nudges in the best direction we can, we don't want to be elitist. If someone's used to hacking together simple protocols using readline, and is comfortable doing that, we don't want to put up barriers to their using trio. And if the sans-i/O approach is harder to get started with, then for some people that will legitimately outweigh the long-term benefits.

There might be one way to have our cake and eat it to: if we can make the sans-I/O version so simple and easy to get started with that even beginners and folks used to readline don't find it a barrier. If we can pull this off, it'd be pretty sweet, because then we can teach the better approach from the beginning, and when they move on to implementing more complex protocols, or integrated existing libraries like h11/h2/wsproto, they're already prepared to do it right.

Alternatively, if we can't... there is really not a lot of harm in having a lines_from_stream generator, or whatever. But anything more than that is going to require exposing some kind of buffering to the user, which is the core of the sans-I/O pattern, so let's think about sans-I/O for a bit.

Can we make sans-I/O accessible and easy?

The core parts of implementing a high-quality streaming line reader, a streaming length-prefixed string reader, or an HTTP parser, are actually all kind of the same:

• You need a buffer
• It needs an efficient append-to-the-end operation
• It needs an efficient extract-from-the-beginning operation
• You need to be able to scan the buffer for a delimiter, with some cleverness to track how far you've scanned to avoid O(n^2) rescans after new data is added
• And some kind of maximum buffer size to avoid memory DoS

h11 internally has a robust implementation of everything here except for specifying delimiters as a regex, and I need to add that anyway to fix python-hyper/h11#7. So I have a plan already to pull that out into a standalone library.

And the API to a sans-I/O line reader, length-prefixed string reader, HTTP parser, or websocket parser for that matter, are also all kind of the same: you wrap them around a Stream, and then call a receive method which tries to pull some "event" out of the internal buffer, while refiling the buffer as necessary.

In fact, if you had sans-I/O versions of any of these, that all followed the same interface conventions, you could even have a single generic wrapper that binds them to a Trio stream, and implements the ReceiveChannel interface! Where the objects being received are lines, or h11.Event objects, or whatever.

So if you really just wanted a way to receive and send lines on a Stream, that might be:

line_channel: trio.abc.Channel[bytes] = sansio_toolbelt.to_trio(sansio_toolbelt.LineProtocol(delimiter=b"\r\n", max_line_length=16384), my_stream)

await line_channel.send(b"hello")
response = await line_channel.receive()

That's maybe a little bit more complicated than I'd want to use in a tutorial, but it's pretty close? Maybe we can slim it down a little more?

This approach is also flexible enough to handle more complex cases, like protocols that switch between lines-oriented and bulk data (HTTP), or that enable TLS half-way through (SMTP's STARTTLS command), which in Twisted's LineReceiver requires some special hooks. You can detach the sans-I/O wrapper from the underlying stream and then wrap it again in a different protocol, so long as you have some way to hand-off the buffer between them.

But while it is flexible enough for that, and that approach is very elegant for Serious Robust Protocol implementations, it might be a lot to ask when someone really just wants to call readline twice and then read N bytes, or something like that. So maybe we'd also want something that wraps a ReceiveStream and provides read_line, read_exactly, read_until, based on the same buffering code described above but without the fancy sans-I/O event layer in between?

[oremanj]

I literally just wrote one of these yesterday, with the interface

@attr.s(auto_attribs=True)
class BufferedReceiveStream(trio.abc.AsyncResource):
    transport_stream: trio.abc.ReceiveStream
    chunk_size: int = 4096
    async def aclose(self) -> None: ...
    async def receive(self, num_bytes: int) -> bytes: ...
    async def receive_all_or_none(self, num_bytes: int) -> Optional[bytes]: ...
    async def receive_exactly(self, num_bytes: int) -> bytes: ...

class TextReceiveStream(trio.abc.AsyncResource):
    transport_stream: trio.abc.ReceiveStream
    encoding: str
    errors: Optional[str]
    newlines: Union[str, Tuple[str, ...], None]
    chunk_size: int
    def __init__(
        self,
        transport_stream: trio.abc.ReceiveStream
        encoding: Optional[str] = None,
        *,
        errors: Optional[str] = None,
        newline: Optional[str] = "",
        chunk_size: int = 8192
    ): ...
    async def aclose(self) -> None: ...
    async def __aiter__(self) -> AsyncIterator[str]: ...
    async def receive_line(self, max_chars: int = -1) -> str: ...

I haven't tested it yet and I'm not totally sure about the interface, but I'll post code once I'm more convinced it works if people think either of these interfaces would be useful. The three receive methods in BufferedReceiveStream only differ in how they handle EOF; I wrote it to help with "you have a length and then that many bytes of data" type binary protocols.

[njsmith]

This may go without saying, but just in case I'll say it anyway :-): I'm being a bit cautious in the issue because if we add something directly to trio itself, then we want to make sure it's the right thing we can support for a long time. But for helpers and utilities and stuff that's outside trio, that doesn't apply, so everyone should at the least feel free to implement what they find useful, share it, whatever.

In fact it can only be useful to see examples of what people come up with...

[njsmith]

Speaking of which, @oremanj, what's your use case? (If you can share.)

[pawarren]

Any movement on this? I'm looking for an equivalent for StreamReader.readuntil so I can continuously listen for special termination chars in websocket messages.

[njsmith]

@pawarren I think this is up to date at the moment. I'm a bit confused about how readuntil and websockets fit together, though – usually websocket messages are small and you read them in as single bytes or str objects? Are you using trio-websocket, or something else?

[pawarren]

I'm currently using asyncio.open_connection() and am referring to https://docs.python.org/3/library/asyncio-stream.html#asyncio.StreamReader.readuntil

I connect to 58 web sockets, keep the connections permanently open, and for each web socket continuously loop over the following steps asynchronously:

1. send one of three different types of messages asking for something
2. listen for one of four different types of responses (one per request type plus a heartbeat)
3. do something with response

The responses themselves are typically small XML messages or images.

It seems like there's a default EOF for most websockets, which I imagine is why libraries like websockets can do things like "for message in socket". But these message generally have a separate end-of-message separator, b'\r\n\r\n'., and normal practice for dealing with that seems to be maintaining a buffer while repeatedly reading chunks of size ~4096 bytes while looking for the separator. reader.readuntil(b'\r\n\r\n') was a nice way of avoiding that.

I'm new to async + websockets, so I might be missing an obvious solution or best practice.

I hadn't seen trio-websockets before; thank you for the reference.

[njsmith]

Ah, OK, I think you mean regular TCP sockets :-). "Web socket" is a specific somewhat complicated protocol that you can use on top of regular sockets. For some reason the people designing WebSocket decided to give it a really confusing name.

It sounds like you have your own simple ad-hoc protocol on top of sockets. You might say you're transmitting a series of "frames", and each frame is terminated by \r\n\r\n. Which is, indeed, exactly the sort of thing that this issue is about :-).

While you're waiting for us to get our ducks in a row and provide a more comprehensive solution, here's some code you can use:

import trio

_RECEIVE_SIZE = 4096  # pretty arbitrary

class TerminatedFrameReceiver:
    """Parse frames out of a Trio stream, where each frame is terminated by a
    fixed byte sequence.

    For example, you can parse newline-terminated lines by setting the
    terminator to b"\n".

    This uses some tricks to protect against denial of service attacks:

    - It puts a limit on the maximum frame size, to avoid memory overflow; you
    might want to adjust the limit for your situation.

    - It uses some algorithmic trickiness to avoid "slow loris" attacks. All
      algorithms are amortized O(n) in the length of the input.

    """
    def __init__(self, stream, terminator, max_frame_length=16384):
        self.stream = stream
        self.terminator = terminator
        self.max_frame_length = max_frame_length
        self._buf = bytearray()
        self._next_find_idx = 0

    async def receive(self):
        while True:
            terminator_idx = self._buf.find(
                self.terminator, self._next_find_idx
            )
            if terminator_idx < 0:
                # no terminator found
                if len(self._buf) > self.max_frame_length:
                    raise ValueError("frame too long")
                # next time, start the search where this one left off
                self._next_find_idx = max(0, len(self._buf) - len(self.terminator) + 1)
                # add some more data, then loop around
                more_data = await self.stream.receive_some(_RECEIVE_SIZE)
                if more_data == b"":
                    if self._buf:
                        raise ValueError("incomplete frame")
                    raise trio.EndOfChannel
                self._buf += more_data
            else:
                # terminator found in buf, so extract the frame
                frame = self._buf[:terminator_idx]
                # Update the buffer in place, to take advantage of bytearray's
                # optimized delete-from-beginning feature.
                del self._buf[:terminator_idx+len(self.terminator)]
                # next time, start the search from the beginning
                self._next_find_idx = 0
                return frame

    def __aiter__(self):
        return self

    async def __anext__(self):
        try:
            return await self.receive()
        except trio.EndOfChannel:
            raise StopAsyncIteration


################################################################
# Example
################################################################

from trio.testing import memory_stream_pair
async def main():
    sender_stream, receiver_stream = memory_stream_pair()

    async def sender():
        await sender_stream.send_all(b"hello\r\n\r\n")
        await trio.sleep(1)
        await sender_stream.send_all(b"split-up ")
        await trio.sleep(1)
        await sender_stream.send_all(b"message\r\n\r")
        await trio.sleep(1)
        await sender_stream.send_all(b"\n")
        await trio.sleep(1)
        await sender_stream.send_all(b"goodbye\r\n\r\n")
        await trio.sleep(1)
        await sender_stream.aclose()

    async def receiver():
        chan = TerminatedFrameReceiver(receiver_stream, b"\r\n\r\n")
        async for message in chan:
            print(f"Got message: {message!r}")

    async with trio.open_nursery() as nursery:
        nursery.start_soon(sender)
        nursery.start_soon(receiver)

trio.run(main)

[pawarren]

Wow, this is fantastic! I particularly like the use of bytearray's optimized delete-from-beginning feature. What do you mean by "It uses some algorithmic trickiness to avoid "slow loris" attacks."? What's the trickiness? And thank you for pointing me towards the definition of websocket. That makes a lot of the docs I've been reading over the past few days make more sense...

…

On Mon, Mar 11, 2019 at 12:12 AM Nathaniel J. Smith < ***@***.***> wrote: Ah, OK, I think you mean regular TCP sockets :-). "Web socket" is a specific somewhat complicated protocol <https://tools.ietf.org/html/rfc6455> that you can use on top of regular sockets. For some reason the people designing WebSocket decided to give it a really confusing name. It sounds like you have your own simple ad-hoc protocol on top of sockets. You might say you're transmitting a series of "frames", and each frame is terminated by \r\n\r\n. Which is, indeed, exactly the sort of thing that this issue is about :-). While you're waiting for us to get our ducks in a row and provide a more comprehensive solution, here's some code you can use: import trio _RECEIVE_SIZE = 4096 # pretty arbitrary class TerminatedFrameReceiver: """Parse frames out of a Trio stream, where each frame is terminated by a fixed byte sequence. For example, you can parse newline-terminated lines by setting the terminator to b"\n". This uses some tricks to protect against denial of service attacks: - It puts a limit on the maximum frame size, to avoid memory overflow; you might want to adjust the limit for your situation. - It uses some algorithmic trickiness to avoid "slow loris" attacks. All algorithms are amortized O(n) in the length of the input. """ def __init__(self, stream, terminator, max_frame_length=16384): self.stream = stream self.terminator = terminator self.max_frame_length = max_frame_length self._buf = bytearray() self._next_find_idx = 0 async def receive(self): while True: terminator_idx = self._buf.find( self.terminator, self._next_find_idx ) if terminator_idx < 0: # no terminator found if len(self._buf) > self.max_frame_length: raise ValueError("frame too long") # next time, start the search where this one left off self._next_find_idx = max(0, len(self._buf) - len(self.terminator) + 1) # add some more data, then loop around more_data = await self.stream.receive_some(_RECEIVE_SIZE) if more_data == b"": if self._buf: raise ValueError("incomplete frame") raise trio.EndOfChannel self._buf += more_data else: # terminator found in buf, so extract the frame frame = self._buf[:terminator_idx] # Update the buffer in place, to take advantage of bytearray's # optimized delete-from-beginning feature. del self._buf[:terminator_idx+len(self.terminator)] # next time, start the search from the beginning self._next_find_idx = 0 return frame def __aiter__(self): return self async def __anext__(self): try: return await self.receive() except trio.EndOfChannel: raise StopAsyncIteration ################################################################# Example################################################################ from trio.testing import memory_stream_pairasync def main(): sender_stream, receiver_stream = memory_stream_pair() async def sender(): await sender_stream.send_all(b"hello\r\n\r\n") await trio.sleep(1) await sender_stream.send_all(b"split-up ") await trio.sleep(1) await sender_stream.send_all(b"message\r\n\r") await trio.sleep(1) await sender_stream.send_all(b"\n") await trio.sleep(1) await sender_stream.send_all(b"goodbye\r\n\r\n") await trio.sleep(1) await sender_stream.aclose() async def receiver(): chan = TerminatedFrameReceiver(receiver_stream, b"\r\n\r\n") async for message in chan: print(f"Got message: {message!r}") async with trio.open_nursery() as nursery: nursery.start_soon(sender) nursery.start_soon(receiver) trio.run(main) — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#796 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AD71iKCpoWeA2uynp9JT-KhBL2Pe4WO_ks5vVgH5gaJpZM4Y-HF3> .

[njsmith]

What do you mean by "It uses some algorithmic trickiness to avoid "slow loris" attacks."? What's the trickiness?

There are two mistakes people often make, that cause this kind of code to have O(n^2) complexity:

• They use a regular bytes object as the buffer, or instead of using del to remove stuff from the front, they copy the back (buf = buf[new_start:]). This is especially common on Python 2, because it doesn't have the optimized bytearray del method built in, so you have to implement it by hand. But even on py3, a lot of people don't realize how important it is to write this operation in exactly the right way.

• after a failed search, they restart the next search from the beginning, instead of tracking how much of the buffer they already searched and only searching the new part

If you make either of these mistakes, then it's easy for a peer that's malicious, or just misbehaving, to burn up a ton of your CPU time doing repeated work. For example, say someone sends you a 2,000 byte frame, but they do it one byte at a time. Now a naive implementation ends up scanning through ~1,000,000 bytes looking for the \r\n\r\n, because it scans the first part of the buffer over and over and over.

It's surprisingly subtle!

[JefffHofffman]

Here's my hack on TerminatedFrameReceiver that extends the sans I/O idea, possible too far. It's similar to the sort/cmp split. Just as single definition of cmp allows lots of sort implementations, a single read & write allows interchangeable frame methods. My trio implementation leaves MUCH to be desired. It's more a POC that shows how a definition like receive_some() is sufficient for framing variations without pestering the runner (nursery) too much. Comments welcomed.

import trio
import attr

@attr.s
class Framer:
    '''
    a mash-up of an accumulator with a parser for frames with a terminator
    does no I/O, no runner dependecies, only lazy evaluation via async calls
    '''
    MAX_FRAME_SIZE = 16384
    terminator = attr.ib()
    more = attr.ib()  # async f() returns next chunk for buf
    done = attr.ib()  # async f(buf) put as frame, return ignored
    _buf = attr.ib(factory=bytearray)
    _next_find_idx = attr.ib(init=False, default=0)
    
    async def deframe(self):
        while True:
            terminator_idx = self._buf.find(
                self.terminator, self._next_find_idx
            )
            if terminator_idx < 0:
                more_data = await self.more()
                # paranoid that a goof in more() may return too much 
                if len(self._buf) + len(more_data) > self.MAX_FRAME_SIZE:
                    raise ValueError("frame too long")
                if more_data == b"" and self._buf:
                    raise ValueError("incomplete frame")
                self._next_find_idx = max(0, len(self._buf) - len(self.terminator) + 1)
                self._buf += more_data
            else:
                frame = self._buf[:terminator_idx]
                del self._buf[:terminator_idx+len(self.terminator)]
                self._next_find_idx = 0
                await self.done(frame)

    # be nice to have this here, but need another done() dest
    # ... and another more() source to use as a pump
    async def enframe(self, msg):
        return msg + self.terminator

################################################################
# Example
################################################################
_RECEIVE_SIZE = 4096  # pretty arbitrary

from trio.testing import memory_stream_pair
async def main():    
    # so sender.aclose() ripples to receiver
    wire = trio.StapledStream(*memory_stream_pair())
    
    async def sender():
        await wire.send_all(b"split-up ")
        await trio.sleep(1)
        await wire.send_all(b"message\r\n\r")
        await trio.sleep(1)
        await wire.send_all(b"\n")
        await trio.sleep(1)
        await wire.aclose()
        
    async def trio_more():
        return await wire.receive_some(_RECEIVE_SIZE)
        
    async def trio_done(msg):
        print((f"Got message: {msg}"))
        
    pump = Framer(b"\r\n\r\n", trio_more, trio_done)

    try:
        async with trio.open_nursery() as nursery:
            nursery.start_soon(sender)
            nursery.start_soon(pump.deframe)
    except trio.ClosedResourceError:
        print ("done")
    except ValueError as e:
        print (f'Unrecoverable error: {e}')

trio.run(main)