lost data on uart streamreader task

[wohltat]

I'm writing a driver for a RFID reader that communicates over UART. Around every 7th message / frame is broken because of missing bytes. I already checked the pin at ESP32 with a logic analyzer and the data that reaches the pin seems flawless.

How could this happen and how can i solve this?
As i understand it this should not happen since the ESP32 has hardware buffer which is far from being filled up.

Here is the minimal code that i'm using to read and print the messages:

import machine
import asyncio

FRAME_HEADER = b'\xAA'
FRAME_TAIL = b'\xDD'

async def read_serial_data(sreader):
    while True:
        try:
            c = await sreader.read(1)
            print(c.hex(), end='')
            
            # new line on end of message (\xDD)
            if c == FRAME_TAIL:
                c2 = await sreader.read(1)
                if c2 == FRAME_HEADER:
                    print('\n' + FRAME_HEADER.hex(), end='')
                       
            await asyncio.sleep(0) 
            
        except KeyboardInterrupt:
            print("Program terminated.")
            break


async def main():
    uart = machine.UART(2, baudrate=115200, tx=17, rx=16, timeout=50) 
    swriter = asyncio.StreamWriter(uart, {})
    sreader = asyncio.StreamReader(uart)
    
    cmd = b'\xAA\x00\x27\x00\x03\x22\xFF\xFF\x4A\xDD' # start data stream
    swriter.write(cmd)
    await swriter.drain()  

    asyncio.create_task(read_serial_data(sreader))

    while True:
        await asyncio.sleep(20)

asyncio.run(main())

The read data looks like follows. With the logic analyzer the frames are the same lengths. You can see that there is sometimes parts of the message missing if bytes in the middle got lost. Sometimes the message is longer if the end or the beginning of the new message got lost.

aa02220011d73400e2004709c0a06821a520010d4f1794dd
aa02220011d23400e2004709c4606821a55c010c8b27dadd
aa022200606821a55c010c8b27dadd
aa02220011ce3000000000000000000000009746a4d387dd
aa02220011d83000000000000000000000009746a4d391dd
aa02220011cd3400e2004709c4606821a55c010c8b27d5dd
aa02220011d73400e2004709c0a06821a520010d4f1794dd
aa02220000e2004709c4606821a55c010c8b27d5dd
aa02220011d63000000000000000000000009746a4d38fdd
aa02220011c53400e2004709c4606821a55c010c8b27cddd
aa02220011d73400e2004709c0a06821a520010d4f1794dd
aa02220011d83000000000000000000000009746a4d391dd
aa02220011cd3400e2004709c4606821a55c010c8b27d5dd0d4f1794dd
aa02220011db3000000000000000000000009746a4d394dd
aa02220011d23400e2004709c4606821a55c010c8b27dadd
aa02220011d63400e2004709c0a06821a520010d4f1793dd
aa02220011d73400e2004709c0a06821a520010d4f1794dd
aa02220011db3000000000000000000000009746a4d394dd
aa02220011d63400e2004709c0a06821a520010d4f1793dd
aa02220011c63400e2004709c4606821a55c010c8b27cedd0d4f1794dd
aa02220011cc3400e2004709c4606821a55c010c8b27d4dd
aa02220011d73400e2004709c0a06821a520010d4f1794dd
aa02220011d73400e2004709c0a06821a520010d4f1794dd
aa02220011d23400e2004709c4606821a55c010c8b27dadd
aa02220000000000000000000000009746a4d394dd
aa02220011d23400e2004709c4606821a55c010c8b27dadd
aa02220011d63400e2004709c0a06821a520010d4f1793dd
aa02220011d63000000000000000000000009746a4d38fdd
aa02220011d23400e2004709c4606821a55c010c8b27dadd
aa02220011d63400e2004709c0a06821a520010d4f1793dd
aa02220011d23400e2004709c4606821a55c010c8b27dadd
aa022200a06821a520010d4f1791dd
aa02220011d83000000000000000000000009746a4d391dd
aa02220011d53000000000000000000000009746a4d38edd
aa02220011d23400e2004709c4606821a55c010c8b27dadd
aa02220011c93400e2004709c0a06821a520010d4f1786dda06821a520010d4f1793dd
aa02220011d23400e2004709c4606821a55c010c8b27dadd
aa02220011d63000000000000000000000009746a4d38fdd

---

MicroPython v1.22.0-preview.73.g7736ab832.dirty on 2023-11-24; Generic ESP32 module with ESP32

Also failed with older micropython version 1.20.

[peterhinch]

I would set the UART timeout to 0 (wait forever). If you need a timeout it should be applied to the task doing the reading rather than to the UART device driver.

[wohltat]

Are you sure timeout=0 is wait forever? When i set it, the behavior is like a small values for timeout and i get a 'None' when reading if there is no data.

Anyways, I tried that Didn't change anything about the data loss.

This data loss phenomenon seems to be somehow connected to tasks, asyncio or something in the esp-idf.
When i use classic blocking uart, everything works fine:

import machine

FRAME_HEADER = b'\xAA'
FRAME_TAIL = b'\xDD'

uart = machine.UART(2, baudrate=115200, tx=17, rx=16, timeout=500) 

cmd = b'\xAA\x00\x27\x00\x03\x22\xFF\xFF\x4A\xDD' # start data stream
uart.write(cmd)
    
while True:
    try:
        c = uart.read(1)
        print(c.hex(), end='')
        
        # new line on end of message (\xDD)
        if c == FRAME_TAIL:
            c2 = uart.read(1)
            
            print(' ', end='')
            if c2 == FRAME_HEADER:              
                print('\n' + FRAME_HEADER.hex(), end='')
                   
            
    except KeyboardInterrupt:
        print("Program terminated.")
        break

[peterhinch]

Apologies, you are correct about timeout=0. I think the problem is in reading one character at a time. At 115,200baud characters arrive at 87μs intervals. In terms of asyncio scheduling this is fast. Try changing the code to read multiple characters at a time:

async def read_serial_data(sreader):
    while True:
        try:
            chars = await sreader.read(100)  # Return as much data as is available
            for c in chars:
                print(c.hex(), end='')
                # new line on end of message (\xDD)
                if c == FRAME_TAIL:  # We just printed the tail
                    print()  # Newline
        except KeyboardInterrupt:
            print("Program terminated.")
            break

[wohltat]

Thanks for your reply. Changing the number of bytes to read works.
The broken transmission seems to happens because the ESP32 cannot keep up with the the buffer overflow.
Your suggestion chars = await sreader.read(100) works.
But strangely if sreader.s.any(): c = sreader.s.read(1); asyncio.sleep(0)also works.

For some reason asyncio / streamreader makes it really slow. I don't really understand why.
Especially strange because streamreader uses the same read uart.read() function.

https://github.com/micropython/micropython/blob/master/extmod/asyncio/stream.py

# MicroPython asyncio module
# stream.py
  class Stream:
    def __init__(self, s, e={}):
        self.s = s
        self.e = e
        self.out_buf = b""
 ...
    # async
    def read(self, n=-1):
        r = b""
        while True:
            yield core._io_queue.queue_read(self.s)
            r2 = self.s.read(n)
            if r2 is not None:
                if n >= 0:
                    return r2
                if not len(r2):
                    return r
                r += r2
...
StreamReader = Stream

What is the line yield core._io_queue.queue_read(self.s) for?

I don't really see the point in using StreamReader.read that is promoted in the examples, anymore.
I was hoping it worked like some kind of abstraction layer so one can grab data from the buffer one byte at a time without worrying about blocking.

I don't know what makes StreamReader.read() so slow. My case is not crazy demanding, just 24Bytes every 20ms.
I was expecting this is easily possibly on an 160MHz ESP32.

There is another thing that i don't understand.
I was trying to change the buffersize with the rxbuf parameter.
rxbuf <= 128 does not work at all, no bytes are returned
The default is rxbuf=256. Higher buffer sizes work fine.

[wohltat]

There is another thing that i don't understand.
I was trying to change the buffersize with the rxbuf parameter.
rxbuf <= 128 does not work at all, no bytes are returned
The default is rxbuf=256. Higher buffer sizes work fine.

Seems to be a IDF requirement:
self->rxbuf = 256; // IDF minimum
https://github.com/micropython/micropython/blob/master/ports/esp32/machine_uart.c#L341

[wohltat]

sreader.read is called like a coroutine (await sreader.read(N) ) but it doesn't really look like one. More like a generator.
yield core._io_queue.queue_read(self.s) still doesn't make sense. If sreader.read(...) is a generator then why is queue_read / _enqueue not returning something.

https://github.com/micropython/micropython/blob/master/extmod/asyncio/core.py#L75

    def _enqueue(self, s, idx):
        if id(s) not in self.map:
            entry = [None, None, s]
            entry[idx] = cur_task
            self.map[id(s)] = entry
            self.poller.register(s, select.POLLIN if idx == 0 else select.POLLOUT)
        else:
            sm = self.map[id(s)]
            assert sm[idx] is None
            assert sm[1 - idx] is not None
            sm[idx] = cur_task
            self.poller.modify(s, select.POLLIN | select.POLLOUT)
        # Link task to this IOQueue so it can be removed if needed
        cur_task.data = self
...

    def queue_read(self, s):
        self._enqueue(s, 0)

[glenn20]

The initial implementations of asyncio in Cpython were built on generators (Im not sure if that is still the case for Cpython), but what you are seeing in yield core._io_queue.queue_read(self.s) is the underlying generator-based implementation details of asyncio on micropython (essentially, all awaitables are generators). This is the point that the streamreader yields control back to the scheduler after adding the uart to the list of objects that are poll()-ed by the scheduler - so that it knows when this task can be resumed (ie. when chars are available to be read from the uart).

See the asyncio/core.py:run_until_complete() function to see the asyncio scheduler workflow.

asyncio was not designed for rapid low-latency "context switching" and your use case (reading one char at a time at 115200 bps) stetches the current implementation too far. That is easily fixed following Peter's prescription. However, future optimisations may make the micropython asyncio implementation more responsive in these types of applications (eg. see PR #6125).

[zcattacz]

It's not just the missing bytes. It seems stream.read() lacks sensitivity.
I have UHF RFID moduels default to 115,200 baud rate. My code reads and parses buffered bytes as below:

While True:
    last_buf += await sr.read(N) # I tried N = 7~2048 , 7 = minimum valid frame size.
    if len(last_buf) > 7:
        try_parse(last_buf)
    asyncio.sleep_ms(0)

But I found the stream.read() often steps the toes:
the parser can parse multiple frames, there are almost always partial-frame bytes left in last_buf. I tried different Ns, either too small or too big making it inefficient or irresponsive.

Meanwhile, with uart.any(), each parse attempt on buffer yields usually 1 full frame with no remaining bytes, though it does fragment when too many tags present or getting cmd response during inventory.

While True:
    s = sw.s.any()
    if s > 0:
        last_buf += await sr.read(s)
        if len(last_buf) > 7:
            try_parse(last_buf)
    asyncio.sleep_ms(0)

...

[glenn20]

What timeout do you use for the uart? The behaviour you are seeing is what I would expect if the timeout is non-zero.

For async use I would set it to 0 and then call sr.read() (ie. no arg - to read out all available chars for efficiency and to minimise risk of buffer overflow in the uart). sr.read() will just return with whatever chars are available when called. I expect his should then behave like your second example (ie. a low, but not zero, likelihood of partial frames).

I would also delete the asyncio.sleep_ms(0) (it just makes your code a little less responsive to incoming chars). The sr.read() should be enough to give other tasks a chance to run.

Oh, and if you would like to minimise memory fragmentation, you could look into using sr.readinto() to read from the uart into a pre-allocated bytearray.

[glenn20]

Actually, I just realised you don't call await asyncio.sleep_ms(0), so the line with asyncio.sleep_ms(0) has no effect anyway.