Memory footprint StreamReader.readexactly

[frensjan]

StreamReader.readexactly is a very handy method to implement reading frames ('chunks' of data preceded with the size of the 'chunk') from a stream. E.g.:

frame_len = yield from reader.readexactly(4) # when using 32 bits sizes
frame = yield from reader.readexactly(frame_len)

---

However, it's memory footprint isn't ideal. A list of 'blocks' is kept until all data is read which is then joined with b''.join(blocks). This causes each 'frame' to reside in memory twice until the blocks list and the bytes objects it points to are freed (when leaving the readexactly method). So it's peak memory usage is O(2n).

For small 'frames' the only issue probably is some copying which could be avoided, but for larger 'frames' this hurts. E.g. when sending large files for processing to a machine which are larger than memory / #receivers / 2.

---

A possible solution might be to use a bytearray to gather the data incrementally. E.g. something along the lines of:

@asyncio.coroutine
def readexactly(self, n):
    if self._exception is not None:
        raise self._exception

    if not n:
        return b''

    if len(self._buffer) >= n and not self._eof:
        data = self._buffer[:n]
        del self._buffer[:n]
        self._maybe_resume_transport()
        return data

    data = bytearray(n)
    pos = 0

    while n:
        if not self._buffer:
            yield from self._wait_for_data('read')

        if self._eof or not self._buffer:
            raise asyncio.IncompleteReadError(data[:pos], pos + n)

        available = len(self._buffer)
        if available <= n:
            data[pos:pos + available] = self._buffer
            self._buffer.clear()
            n -= available
            pos += available
        else:
            data[pos:pos + n] = self._buffer[:n]
            del self._buffer[:n]
            n = 0
        self._maybe_resume_transport()

    return data

This implementation would have a peak memory footprint of O(n). Or O(n + max(map(len, blocks))), not sure what the other constant factors might be, probably lots more lower in the stack.

Note that the implementation above also does away with some unnecessary copying by duplicating some code from StreamReader.read.
Note as well that the implementation has a short path for readexactly when the required data is available in the _buffer of the StreamReader.

An obvious issue with the implementation above is that it returns (mutable) bytebuffer instead of bytes objects. For my use case I considered this acceptable; I'm not sure if it would be acceptable in asyncio / the standard library. I'm not a aware of a copy-free mechanism to turn a bytebuffer into an immutable bytes-like object (like asReadOnlyBuffer).

Apart from being more memory friendly in terms of 'volume', because there is less copying, it's also a tad bit faster (around 20% on my machine, see test script here: https://gist.github.com/frensjan/27061b47e1423dd73eeec9f5cad5f4ad

An alternative direction could be to add a socket.recv_into-like method which accepts a mutable buffer.

---

I hope the issue is clear. Curious to get feedback on relevance for the 'broader developer community' and to the solution. If desirable I'll whip up a PR.

Best regards,
Frens Jan

[1st1]

cc @socketpair

[socketpair]

Yes, this need to be fixed. expansion of bytearray is very fast.

[1st1]

I actually noticed a speed regression in streams in 3.5.2. @socketpair would you be able to work on this?

[socketpair]

Yes, I can try. @1st1 can you provide the case where you notice performnce regression? (except this issue, since it already have performance test)

[1st1]

I think I saw it when I was running this benchmark: https://github.com/MagicStack/uvloop/blob/master/examples/bench/rlserver.py -- stresses StreamReader.readline -- it became slower in 3.5.2.

[frensjan]

@1st1 Just ran my own benchmark for readline again (https://gist.github.com/frensjan/27061b47e1423dd73eeec9f5cad5f4ad) but with uvloop: I saw a throughput improvement of 36% with my suggested implementation.

@socketpair note that the only bytearray which gets expanded is StreamReader._buffer. No incremental allocations are done within the suggested implementation of readline. I'm not familiar with the way bytearray objects expand, but I expect allocating bytarray(n) is better. Also, there is the added 'benefit' that a MemoryError will be raised before consuming to much from the reader.

[socketpair]

This test just hang on my PC...
@frensjan can you compare preformance of your variant vs #395 ?

[frensjan]

@socketpair @1st1 I've compared master(at bbe53f3), #395 and the implementation suggested above with (with https://gist.github.com/frensjan/27061b47e1423dd73eeec9f5cad5f4ad):

master:

server done,     sent 4096.0 megabytes
client done, received 4096.0 megabytes
test took 0:00:09.196442
at 0.43 gigabytes per second

#395:

server done,     sent 4096.0 megabytes
client done, received 4096.0 megabytes
test took 0:00:08.648002
at 0.46 gigabytes per second

implementation above:

server done,     sent 4096.0 megabytes
client done, received 4096.0 megabytes
test took 0:00:07.291480
at 0.55 gigabytes per second

Environment:

Python: 3.5.1
Kernel: 4.6.4-301.fc24.x86_64

Curious to hear other experiences / observations.

[methane]

@frensjan Your implementation seems returns bytearray, not bytes. Am I wrong?

#395 returns bytes, via bytes(memoryview(self._buffer)[:n]). (#395 (comment))

But it has significant cost. I researched what cause slowdown, and sent patch to Python.
http://bugs.python.org/issue27704

[frensjan]

@methane yes, it indeed uses bytearray. Would be nice to have a means to turn a bytearray into a bytes object (or readonly view on it) as in asReadOnlyBuffer.

Or a faster implementation of bytes(bytearray) as you suggest. However, I'm a not fully confident on the interplay between StreamReader._limit and waiting until len(self._buffer) > n. If that's no issue ... (I've seen deadlock related commits around this in asyncio, so we should probably be careful).

[frensjan]

@methane perhaps more importantly, the main reason why I filed this issue is not because of performance in terms of throughput, but memory footprint. The #395 implementation uses O(2n) memory (where n is the number of bytes to be read) because of the copy with bytes.

[frensjan]

Finally, I'm also curious about the memory behaviour of del self._buffer[:n] which both the implementation above as #395 use. I've witnessed MemoryErrors on this, so I suspect it to copy any bytes beyond n.

This may be 'catastrophic' when reading an 8 bytes size field for some gigabytes of message following it. Now both implementations would suffer from this issue. However, #395 would suffer from this sooner as self._buffer is growing until it contains n bytes. While in the implementation above self._buffer would contain only the bytes received since the last time the socket buffer was empty.

The above may be a reason to implement something like recv_into where the user would control the buffer. With a usage as e.g.:

length = decode_length(reader.readexactly(4))
frame = bytearray(length)
reader.receiveinto(frame, length)

[methane]

I'm sorry. I've forgot what you explained at first post.

read_into and readexactly_into seems better to me.

[socketpair]

Yes, I think read_into and read_into_exactly is good API.