bpo-41279: Add StreamReaderBufferedProtocol

[tontinton]

https://bugs.python.org/issue41279

I got way better performance on await reader.read() using this branch on linux (check out the chart on the server.py script's comments).

The way I tested was writing a server / client:
server.py:

import asyncio
import contextlib
import time


async def client_connected(reader, writer):
    start = time.time()

    with contextlib.closing(writer):
        for i in range(1000):
            # the more this parameter's distance from 65536 is greater the better the performance global_buffer gives
            # On linux:
            # 65536 * 2 -> Gives about 160% better performance
            # 4096 -> Gives about 155% better performance
            # 65536 -> Gives about 125% better performance

            # On windows using 65536 gives the same performance for some reason, which is interesting
            # But any other value gives a bit better performance, for example 65536 * 2 gives about 120% better performance
            await reader.read(65536 * 2)

    print(f'{time.time() - start}')


async def main():
    server = await asyncio.start_server(client_connected, '127.0.0.1', 8888, global_buffer=True)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()


if __name__ == "__main__":
    asyncio.run(main())

client.py:

import asyncio
import contextlib


async def flood(ip, port):
    message = b'A' * 1024 * 64  # tweak this parameter as much as you want
    reader, writer = await asyncio.open_connection(ip, port)
    with contextlib.closing(writer):
        while True:
            writer.write(message)
            await writer.drain()


if __name__ == "__main__":
    asyncio.run(flood('127.0.0.1', 8888))

[1st1]

Do we have proper functional tests that hit this branch? If not we need to add them.

[1st1]

You can probably copy some tests from the uvloop project (as I own the project there will be no licensing issues even if it's a literal copy)

[tontinton]

I am thinking of doing parameterize on all the appropriate test in test_proactor_events.py.

That way we can use the already existing tests on both Protocol and BufferedProtocol

[tontinton]

Ok I did this using the parameterize function, it did made me write a bit of ugly code I want you to look at.

Check out what I did in the test adding commit.

By the way shouldn't I add the same tests for UnixReadPipeTransportTests?

[tontinton]

Ok I added the tests to UnixReadPipeTransportTests as well using the same patches

[1st1]

Ok I did this using the parameterize function, it did made me write a bit of ugly code I want you to look at.

Yeah, let's try to simplify it. See my other comment.

By the way shouldn't I add the same tests for UnixReadPipeTransportTests?

I guess since we're updating _UnixReadPipeTransport we should.

[tontinton]

Yeah I simplified it a lot

[mpaolini]

Suggested change

	except (SystemExit, KeyboardInterrupt):
	raise

the two llines can be removed

[tontinton]

Fixed

[mpaolini]

Suggested change

return

just a small change ;)

[tontinton]

Fixed

[bedevere-bot]

Thanks for making the requested changes!

@1st1, @asvetlov: please review the changes made to this pull request.

[tontinton]

Hey, just remembered that I did this, bumping! :)

[gst]

why this double pause_reading() above and then resume_reading() here ?

and same question for below in test_pause_resume_reading.

[tontinton]

Fixed

[gst]

before the _run_once() would not it be better to assert that the msg is not in the _mock_call_args_list of the mock object ?

and same below for the others data_received.assert_called_with following a _run_once().

[tontinton]

I don't understand which mock object you are talking about, and I don't see any other test reading _mock_call_args_list.

Can you elaborate?

[gst]

something like:

for msg in msgs[:2]:
            self.assertFalse(unittest.mock.call(msg) in self.protocol.data_received.mock_calls)
            self.loop._run_once()
            self.protocol.data_received.assert_called_with(msg)

to ensure that each msg is effectively processed during the _run_once() call.

[tontinton]

Oh I see, but protocol's buffer_updated (data_received is only in a regular stream object, not buffered) is not a mock, it's an actual function, so no mock_calls field.

[gst]

ah ok :/

[tontinton]

I'll fix the ssl tests sometime this weekend

[arhadthedev]

@tontinton You can turn your PR into a draft and not worry about timeframes.