uasyncio StreamReader readinto not reading all data

[pobbie]

I have an issue sending files to a Pico W running a uasyncio server on micropython v1.20.0 and older.
Example:

BLOCK_SIZE = 512
buffer = bytearray(BLOCK_SIZE)
size = await reader.readinto(buffer)

If I use a larger BLOCK_SIZE than 512, some times only part of the data is read.
For example 800 bytes even though 1024 bytes are sent.
Every thing works great, (but slow), as long as I stick to the 512 bytes block size.

Any one with the same experience?

[peterhinch]

Perhaps you could clarify. With a large buffer, I take it that .readinto is returning with partial data. In a subsequent call to .readinto does the missed data appear? My question is whether data is actually being lost.

[pobbie]

From my, (not very careful), tests I have observed that a second call to .readinto some times return the exact amout of missed data, some times more than the missed data, some times nothing. Från: Peter Hinch ***@***.***> Skickat: den 3 maj 2023 13:21 Till: micropython/micropython ***@***.***> Kopia: Per-Olof Bergström ***@***.***>; Author ***@***.***> Ämne: Re: [micropython/micropython] uasyncio StreamReader readinto not reading all data (Discussion #11403) Perhaps you could clarify. With a large buffer, I take it that .readinto is returning with partial data. In a subsequent call to .readinto does the missed data appear? My question is whether data is actually being lost. - Reply to this email directly, view it on GitHub<#11403 (comment)>, or unsubscribe<https://github.com/notifications/unsubscribe-auth/AC67ZPBMSKJEJ3QXJXCJMF3XEI5QNANCNFSM6AAAAAAXUDGBT4>. You are receiving this because you authored the thread.Message ID: ***@***.******@***.***>>

[peterhinch]

I think you are experiencing a classic synchronisation problem. To clarify the different methods:

• .readexactly(n) will cause the calling task to pause until n characters have been read.
• .readline() will cause the calling task to pause until a complete line (terminated with \n) has been read.
• .readinto(buf) will return immediately with as many bytes are available in the socket's internal buffer.

If the buffer is small, by the time you call .readinto, the socket's internal buffer has enough characters so you are getting a buffer full. If the buffer is large, my guess is that you are calling .readinto when too few characters are ready to fill the buffer. If you immediately call it again, you may do so before any more data has been received, and hence receive zero characters.

This problem of synchronisation is simpler if you have fixed length records or line-structured data, where the first two methods may be used. Otherwise you must keep reading data until you get an end of record, as determined by parsing the data.

[pobbie]

Ok, I can only use .readinto else I get all kinds of memory exceptions because the pico is low on ram. I have learned that from you in another discussion. I have already tested your thoughts on the issue with .readinto by waiting some time before I call .readinto without success. Anyway, I am right now writing a simple code sample focusing on the problem that I might post later for others to test. Från: Peter Hinch ***@***.***> Skickat: den 3 maj 2023 15:51 Till: micropython/micropython ***@***.***> Kopia: Per-Olof Bergström ***@***.***>; Author ***@***.***> Ämne: Re: [micropython/micropython] uasyncio StreamReader readinto not reading all data (Discussion #11403) I think you are experiencing a classic synchronisation problem. To clarify the different methods: * .readexactly(n) will cause the calling task to pause until n characters have been read. * .readline() will cause the calling task to pause until a complete line (terminated with \n) has been read. * .readinto(buf) will return immediately with as many bytes are available in the socket's internal buffer. If the buffer is small, by the time you call .readinto, the socket's internal buffer has enough characters so you are getting a buffer full. If the buffer is large, my guess is that you are calling .readinto when too few characters are ready to fill the buffer. If you immediately call it again, you may do so before any more data has been received, and hence receive zero characters. This problem of synchronisation is simpler if you have fixed length records or line-structured data, where the first two methods may be used. Otherwise you must keep reading data until you get an end of record, as determined by parsing the data. - Reply to this email directly, view it on GitHub<#11403 (comment)>, or unsubscribe<https://github.com/notifications/unsubscribe-auth/AC67ZPAOIPO3RF7UZXGUJMLXEJPEJANCNFSM6AAAAAAXUDGBT4>. You are receiving this because you authored the thread.Message ID: ***@***.******@***.***>>

[pobbie]

I have prepaired a test script below with the sender running on Windows and
receiver running on Pico. (Same script for both platforms).
Type in your WIFI info and run it on the Pico, and you will get the server ip address.

The pico script detects partial data read by .readinto and the rest is read by .readexactly,
that is then inserted in the bytearray. My experience is that it works for some hundred of
blocks before it crash caused by a memory exception in the pico.

DATA_PORT: int = 2023
BLOCK_SIZE: int = 4096

try:
    import asyncio
    # Windows Python code:
    class Sender(object):
        READY: bytes = b'\x01\x02\x07\x03'

        def __init__(self, host: str) -> None:
            self.host: str = host
            self.dbuf = bytearray(BLOCK_SIZE)

        async def run(self) -> None:
            reader, writer = await asyncio.open_connection(self.host, DATA_PORT)
            block_count: int = 0

            while True:
                # Ready?
                if await reader.read(4) == self.READY:
                    # Data size
                    ds = BLOCK_SIZE.to_bytes(4, byteorder='big')
                    writer.write(ds)
                    await writer.drain()

                    # Data
                    writer.write(self.dbuf)
                    await writer.drain()

                    block_count += 1
                    print(f'Block count: {block_count}', end='\r')
                else:
                    break

    
    if __name__ == '__main__':
        host: str = input('Please enter host ip: ')
        if host:
            sender = Sender(host)
            try: 
                asyncio.run(sender.run())
                print('Sender exit')
            except Exception as ex:
                print(ex)
        input('Enter to exit ')

except:
    # Pico MicroPython code:
    import uasyncio as asyncio
    from network import WLAN, STA_IF
    from rp2 import country
    import time
    import gc
    
    class Receiver(object):
        READY: bytes = b'\x01\x02\x07\x03'
        ERROR: bytes = b'\x00\x00\x00\x00'
        
        def __init__(self) -> None:
            self.wifi_info = {'ssid':'Your network', 'password':'Your network password'}
            country('SE') # Put your country
            self.wlan = WLAN(STA_IF)
            self.wlan.active(True)
            self.sbuffer = bytearray(4)
            self.dbuffer = bytearray(BLOCK_SIZE)

        def connect_wifi(self) -> tuple:
            while not self.wlan.isconnected():
                self.wlan.connect(self.wifi_info.get('ssid'), self.wifi_info.get('password'))
                # Wait for connect or fail
                s = self.wlan.status()
                while s > 0 and s < 3:
                    time.sleep(0.1)
                    s = self.wlan.status()
                if s == 3:
                    # Connected
                    break
        
            ssid = self.wlan.config('ssid')
            ip = self.wlan.ifconfig()[0]
            return (ssid, ip)
  
        async def run(self):
            print('Connecting WIFI')
            ssid, ip = self.connect_wifi()
            print(f'Connected to {ssid}')
            asyncio.create_task(asyncio.start_server(self.session, "0.0.0.0", DATA_PORT))
            print(f'Server running on: {ip}')
            while True:
                await asyncio.sleep(1)

        async def session(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
            print('Sender connected')
            block_count: int = 0

            while True:
                # Ready for more
                writer.write(self.READY)
                await writer.drain()

                # Data size
                s = await reader.readinto(self.sbuffer)
                if s == 4:
                    size = int.from_bytes(self.sbuffer, 'big')
                    
                    # Data
                    s = await reader.readinto(self.dbuffer)
                    if s == size:
                        block_count += 1
                        print(f'Block count: {block_count}', end='\r')
                        gc.collect()
                    else:
                        # Partial data received
                        print(f'Block count: {block_count}')
                        print(f'{s} of {size} bytes received')
                        
                        rest = size - s
                        b = await reader.readexactly(rest)
                        if b and len(b) == rest:
                            self.dbuffer[s:0] = b
                            print(f'The rest ({rest} bytes) addded to buffer')
                            b = None
                            gc.collect()


    if __name__ == '__main__':
        receiver = Receiver()
        try: 
            asyncio.run(receiver.run())
            print('Receiver exit')
        except Exception as ex:
            print(ex)
        input('Enter to exit ')

[glenn20]

A possibly simpler alternative is (warning, untested code):

                    # Data
                    s = 0
                    while s < size:
                        s += await reader.readinto(self.dbuffer[s:])

This should keep calling readinto to add more data to the end of the buffer till you have exactly size bytes in self.dbuffer.

[pobbie]

Thank you but... been there done that, it gives an exception Från: glenn20 ***@***.***> Skickat: den 4 maj 2023 09:58 Till: micropython/micropython ***@***.***> Kopia: Per-Olof Bergström ***@***.***>; Author ***@***.***> Ämne: Re: [micropython/micropython] uasyncio StreamReader readinto not reading all data (Discussion #11403) A possibly simpler alternative is (warning, untested code): # Data s = 0 while s < size: s += await reader.readinto(self.dbuffer[s:]) This should keep calling readinto to add more data to the end of the buffer till you have exactly size bytes in self.dbuffer. - Reply to this email directly, view it on GitHub<#11403 (comment)>, or unsubscribe<https://github.com/notifications/unsubscribe-auth/AC67ZPDZ3ZLQEQQ5SRJMFCTXENOQ5ANCNFSM6AAAAAAXUDGBT4>. You are receiving this because you authored the thread.Message ID: ***@***.******@***.***>>

[peterhinch]

The handling of partial data does a lot of allocation, which the suggestion of @glenn20 aims to fix. What is the exception which occurs when you try his approach?