Bugs in streams.py

[furbrain]

This code should show a constantly increasing number, interspersed with groups of four characters read from
the usb_cdc.data serial terminal:

"""
example that reads from the cdc data serial port in groups of four and prints
to the console. The USB CDC data serial port will need enabling. This can be done
by copying examples/usb_cdc_boot.py to boot.py in the CIRCUITPY directory

Meanwhile a simple counter counts up every second and also prints
to the console.
"""


import asyncio
import usb_cdc

async def client():
    s = asyncio.StreamReader(usb_cdc.data)
    while True:
        text = await s.read(4)
        print(text)
        await asyncio.sleep(0)

async def counter():
    i = 0
    while True:
        print(i)
        i += 1
        await asyncio.sleep(1)

async def main():
    client_task = asyncio.create_task(client())
    count_task = asyncio.create_task(counter())
    await asyncio.gather(client_task, count_task)

asyncio.run(main())

What actually happens is 0 is printed, then nothing until four characters have been received on usb_cdc.data, and then this error message is produced:

Traceback (most recent call last):
  File "code.py", line 39, in <module>
  File "/lib/asyncio/core.py", line 292, in run
  File "/lib/asyncio/core.py", line 256, in run_until_complete
  File "/lib/asyncio/core.py", line 241, in run_until_complete
  File "code.py", line 36, in main
  File "/lib/asyncio/funcs.py", line 113, in gather
  File "/lib/asyncio/funcs.py", line 108, in gather
  File "/lib/asyncio/core.py", line 241, in run_until_complete
  File "code.py", line 22, in client
  File "/lib/asyncio/stream.py", line 63, in read
  File "/lib/asyncio/core.py", line 152, in queue_read
  File "/lib/asyncio/core.py", line 140, in _enqueue
AssertionError: 

This is due to this code in asyncio.stream.Stream:

    async def read(self, n):
        """Read up to *n* bytes and return them.

        This is a coroutine.
        """

        core._io_queue.queue_read(self.s)
        await core.sleep(0)
        return self.s.read(n)

It seems that the call to core._io_queue.queue_read(self.s) sets up the asyncio main loop to poll for
updates to self.s (usb_cdc.data) in this instance, and the schedules the current task to be re-awoken when it happens.
It then calls await core.sleep(0) which almost immediately returns.
self.s.read is then called which blocks everything until 4 characters are read.
Next time around, again core._io_queue.queue_read(self.s) is read, but this time there is already an entry waiting for usb_cdc.data to be updated and the code gets confused and raises an error.

This can be fixed by adding a yield statement instead of core.sleep(0) - I'll add a PR to that effect. However, I can see that the code previously did use that and I'm not sure why the change was made.

There is also an error in that if we add the yield, then the coroutine will pause until one character has been read (thus triggering the poll in core._io_queue), but it will then (again) block until the other 3 characters have been read. However, this time it will not crash. This probably needs to be fixed in python, probably using a scheme similar to Stream.read_exactly