Async randomly dying

[desultory]

Checks

• I agree to follow the MicroPython Code of Conduct to ensure a safe and respectful space for everyone.

• I've searched for existing issues matching this bug, and didn't find any.

Port, board and/or hardware

RP2

MicroPython version

MicroPython v1.22.2 on 2024-02-22; Raspberry Pi Pico with RP2040

Reproduction

from machine import UART, Pin
from uasyncio import StreamReader, StreamWriter, Event, Lock, sleep_ms, run


LoRa_bands = {'EU': 868.1, 'US': 915.0}


class LoRa:
    def __init__(self, uart=0, tx_pin=0, rx_pin=1, reset_pin=4, baudrate=115200,
                 band='US', address=2):
        self.uart = UART(uart, baudrate=baudrate, tx=tx_pin, rx=rx_pin)
        self.reset_pin = Pin(reset_pin, Pin.OUT)
        self.reader = StreamReader(self.uart)
        self.writer = StreamWriter(self.uart, {})

        self._band = band
        self._address = address

        self.initialized = False
        self.seat = Lock()
        self._recv = Event()
        self._received = Event()

    def start(self):
        from uasyncio import create_task
        create_task(self.dev_init())
        create_task(self.readloop())

    async def send_query(self, query):
        return await self.send_command(query, True)

    async def send_command(self, cmd, query=False):
        """ Sends an AT command to the device """
        # Fix capitalization
        if loc := cmd.find('=') + 1:
            cmd = cmd[:loc].upper() + cmd[loc:]
        else:
            cmd = cmd.upper()

        # Add prefix
        if not cmd.startswith('AT+'):
            orig_cmd = cmd
            cmd = 'AT+' + cmd
        else:
            orig_cmd = cmd[3:]

        # Add suffix for query
        if query and cmd[-1] != '?':
            cmd += '?'

        # Add termination
        cmd += '\r\n'

        async with self.seat:
            print("Sending command: ", cmd.rstrip())
            await self.writer.awrite(cmd)
            await sleep_ms(10 if query else 50)
            while not self.uart.any():
                print("Waiting for response...")
                await sleep_ms(250)
            response = await self.recv()

        response = response[:-2]
        print(response)

        if query:
            if response.startswith(f"+{orig_cmd}="):
                response = response[len(f"+{orig_cmd}="):]
            else:
                raise ValueError("Invalid response: %s" % response)
        elif response != '+OK':
            raise ValueError("Error: %s" % response)
        return response

    async def reset(self):
        print("Resetting LoRa module.")
        self.reset_pin.off()
        await sleep_ms(100)
        self.reset_pin.on()
        status = await self.recv()
        if status != '+READY\r\n':
            raise ValueError("Module not ready after reset: %s" % status)

    async def dev_init(self):
        await self.reset()
        if not self.version or not self.uid:
            raise ValueError("Unknown module.")
        self.address = self._address
        self.band = self._band

    async def recv(self):
        self._recv.set()
        await self._received.wait()
        self._received.clear()
        return self._data

    async def readloop(self):
        while True:
            await self._recv.wait()
            raw_data = await self.reader.readline()
            self._data = raw_data.decode()
            self._recv.clear()
            self._received.set()
        print("wtf")

    @property
    def address(self):
        return run(self.send_query('address'))

    @address.setter
    def address(self, address=None):
        if address is None:
            address = self._address
        elif address >= 0 and address < 2 ** 16:
            self._address = address
        else:
            raise ValueError("Invalid LoRa address: %s", address)

        return run(self.send_command(f'ADDRESS={address}'))

    @property
    def band(self):
        return run(self.send_query('band'))

    @band.setter
    def band(self, band=None):
        if band is None:
            band = self._band
        else:
            self._band = band

        band_freq = int(LoRa_bands[band] * 1_000_000)
        return run(self.send_command(f'BAND={band_freq}'))

    @property
    def parameter(self):
        return run(self.send_query('parameter'))

    @parameter.setter
    def set_parameters(self, spreading_factor, bandwidth, coding_rate):
        return run(self.send_command(f'PARAMETER={spreading_factor},{bandwidth},{coding_rate}'))

    @property
    def uid(self):
        return run(self.send_query('uid'))

    @property
    def version(self):
        return run(self.send_query('ver'))

from asyncio import get_event_loop
from lora import LoRa

lora = LoRa()

loop = get_event_loop()

lora.start()

loop.run_forever()

print("Ended.")

Connected to MicroPython at /dev/ttyACM2
Use Ctrl-] or Ctrl-x to exit this shell

MPY: soft reboot
Resetting LoRa module.
Sending command:  AT+VER?
+VER=RYLR998_REYAX_V1.2.2
Sending command:  AT+UID?
+UID=000500110457F1B400003E76
Sending command:  AT+ADDRESS=2
+OK
Sending command:  AT+BAND=915000000
+OK
Ended.
MicroPython v1.22.2 on 2024-02-22; Raspberry Pi Pico with RP2040
Type "help()" for more information.

The infinite loop seems to end withot reason, and then the run_forever is not followed.

Expected behaviour

This script should loop infinitely, instead it seems to stop after dev_init completes.

Observed behaviour

The loop stops with no exceptions, even though it's an infinite loop.

Additional Information

No, I've provided everything above.

[dpgeorge]

Asyncio loops will terminate early if there are no more tasks left to run. This can happen if you have a "deadlock" situation where all tasks are waiting on events that can never occur.

I think that's what is going wrong with your code, the readloop() task is waiting on the self._recv event, but there's no task that's able to set that event.

[desultory]

I've updated it and I'm still getting similar results, and interestingly, if I run this:

lora = LoRa(address=3, reset_pin=2)
webserver = WebServer()
webserver.serve()


async def main():
    await lora.dev_init()

    while True:
        print("asdf")
        await sleep(1)

    while True:
        await lora.send_message('hello', 2)
        await sleep(5)

create_task(main())

get_event_loop().run_forever()

It only prints "asdf" once, and then seems to halt.

from asyncio import run
from lora import LoRa

lora = LoRa()
run(lora.dev_init())


while True:
    run(lora.get_message())

Seems to run correctly:

MPY: soft reboot
Resetting LoRa module.
Sending command:  AT+VER?
Response size: 27
+VER=RYLR998_REYAX_V1.2.2
Sending command:  AT+UID?
Response size: 31
+UID=000500110457F1B400003E76
Sending command:  AT+ADDRESS=2
Response size: 5
+OK
Sending command:  AT+BAND=915000000
Response size: 5
+OK
Sending command:  AT+ADDRESS?
Response size: 12
+ADDRESS=2
Sending command:  AT+NETWORKID?
Response size: 15
+NETWORKID=18
LoRa module initialized: 2@18
+RCV=2,5,hello,-6,11

+RCV=2,5,hello,-7,11

+RCV=3,5,hello,-7,12

+RCV=3,5,hello,-6,11

+RCV=3,5,hello,-6,11

+RCV=3,5,hello,-1,12

+RCV=3,5,hello,-1,11

+RCV=3,5,hello,-1,11

Each printout is a result of me rebooting/reflashing the device which sends the message, it should be repeatedly sending messages.

The current class code is:

from machine import UART, Pin
from asyncio import StreamReader, StreamWriter, Event, Lock, sleep_ms, run


LoRa_bands = {'EU': 868.1, 'US': 915.0}


class LoRa:
    def __init__(self, uart=0, tx_pin=0, rx_pin=1, reset_pin=4, baudrate=115200,
                 band='US', address=2):
        self.uart = UART(uart, baudrate=baudrate, tx=tx_pin, rx=rx_pin)
        self.reset_pin = Pin(reset_pin, Pin.OUT)
        self.reader = StreamReader(self.uart)
        self.writer = StreamWriter(self.uart, {})

        self._band = band
        self._address = address

        self.initialized = Event()
        self.seat = Lock()

    async def send_query(self, query):
        return await self.send_command(query, True)

    async def send_message(self, message, target=0):
        print(f"[{target}@{self.network}] Sending message: {message}")
        return await self.send_command(f'send={target},{len(message)},{message}')

    async def get_message(self):
        msg = await self.recv()
        print(msg)

    async def send_command(self, cmd, query=False):
        """ Sends an AT command to the device """
        # Fix capitalization
        if loc := cmd.find('=') + 1:
            cmd = cmd[:loc].upper() + cmd[loc:]
        else:
            cmd = cmd.upper()

        # Add prefix
        if not cmd.startswith('AT+'):
            orig_cmd = cmd
            cmd = 'AT+' + cmd
        else:
            orig_cmd = cmd[3:]

        # Add suffix for query
        if query and cmd[-1] != '?':
            cmd += '?'

        # Add termination
        cmd += '\r\n'

        print("Waiting for seat")

        async with self.seat:
            print("Sending command: ", cmd.rstrip())
            await self.writer.awrite(cmd)
            await self.writer.drain()
            await sleep_ms(10 if query else 50)
            while not self.uart.any():
                print("Waiting for response...")
                await sleep_ms(250)
            response = await self.recv()

        response = response[:-2]
        print(response)

        if query:
            if response.startswith(f"+{orig_cmd}="):
                response = response[len(f"+{orig_cmd}="):]
            else:
                raise ValueError("Invalid response: %s" % response)
        elif response != '+OK':
            raise ValueError("Error: %s" % response)
        return response

    async def reset(self):
        print("Resetting LoRa module.")
        self.reset_pin.off()
        await sleep_ms(100)
        self.reset_pin.on()
        status = await self.recv()
        if status != '+READY\r\n':
            raise ValueError("Module not ready after reset: %s" % status)

    async def dev_init(self):
        await self.reset()
        if not self.version or not self.uid:
            raise ValueError("Unknown module.")
        self.address = self._address
        self.band = self._band
        print(f"LoRa module initialized: {self.address}@{self.network}")
        self.initialized.set()

    async def recv(self):
        data = await self.reader.readline()
        return data.decode()

    @property
    def address(self):
        return run(self.send_query('address'))

    @address.setter
    def address(self, address=None):
        if address is None:
            address = self._address
        elif address >= 0 and address < 2 ** 16:
            self._address = address
        else:
            raise ValueError("Invalid LoRa address: %s", address)

        return run(self.send_command(f'ADDRESS={address}'))

    @property
    def network(self):
        return run(self.send_query('networkid'))

    @property
    def band(self):
        return run(self.send_query('band'))

    @band.setter
    def band(self, band=None):
        if band is None:
            band = self._band
        else:
            self._band = band

        band_freq = int(LoRa_bands[band] * 1_000_000)
        return run(self.send_command(f'BAND={band_freq}'))

    @property
    def parameter(self):
        return run(self.send_query('parameter'))

    @parameter.setter
    def set_parameters(self, spreading_factor, bandwidth, coding_rate):
        return run(self.send_command(f'PARAMETER={spreading_factor},{bandwidth},{coding_rate}'))

    @property
    def uid(self):
        return run(self.send_query('uid'))

    @property
    def version(self):
        return run(self.send_query('ver'))

I've tested without the "webserver" module, and it does the same thing. The "webserver" is:

class WebServer:
    def __init__(self, address='0.0.0.0', port=80):
        self.address = address
        self.port = port

    def serve(self):
        from asyncio import start_server, create_task
        create_task(start_server(self.handle_request, self.address, self.port))
        print('Listening on: %s:%s' % (self.address, self.port))

    async def blink(self):
        from asyncio import sleep_ms
        from machine import Pin
        Pin("LED", Pin.OUT).toggle()
        await sleep_ms(100)
        Pin("LED", Pin.OUT).toggle()

    async def read_adc(self, path):
        from machine import ADC, Pin
        from re import search
        if s := search(r'adc/(\d+)', path):
            adc_pin = int(s.group(1))
        else:
            adc_pin = 26

        if s := search(r'adc/\d+/(\d+)', path):
            reads = int(s.group(1))
        else:
            reads = 1

        try:
            adc = ADC(Pin(adc_pin))
        except ValueError:
            return b'Invalid ADC pin: %s' % adc_pin

        return '\n'.join([str(adc.read_u16() >> 4) for _ in range(min(reads, 32))]).encode()

    async def send_memory(self, writer):
        from micropython import mem_info
        host, port = writer.get_extra_info('peername')
        info = mem_info()
        print('[%s:%s] <200>: Memory info: %s' % (host, port, info))
        response = b'HTTP/1.1 200 OK\r\nContent-Length: %s\r\n\r\n%s' % (len(info), info.encode())
        writer.write(response)

    async def send_200(self, writer, data):
        host, port = writer.get_extra_info('peername')
        print('[%s:%s] <200>: %s' % (host, port, data))
        response = b'HTTP/1.1 200 OK\r\nContent-Length: %s\r\n\r\n%s' % (len(data), data)
        writer.write(response)

    async def process_request(self, content, writer):
        from micropython import mem_info
        from re import search
        await self.blink()
        path = search(r'^GET /(.+?) HTTP', content).group(1)
        host, port = writer.get_extra_info('peername')
        print('[%s:%s] Got request with path: %s' % (host, port, path))

        mem_info()
        if path.startswith('adc'):
            await self.send_200(writer, await self.read_adc(path))
        else:
            await self.send_200(writer, b'Hello, World!')

    async def handle_request(self, reader, writer):
        data = bytearray(2048)
        if content_length := await reader.readinto(data):
            await self.process_request(str(memoryview(data)[:content_length], 'utf-8'), writer)
        await reader.wait_closed()

[peterhinch]

I am not sure about this:

        async with self.seat:
            print("Sending command: ", cmd.rstrip())
            await self.writer.awrite(cmd)
            await sleep_ms(10 if query else 50)
            while not self.uart.any():
                print("Waiting for response...")
                await sleep_ms(250)
            response = await self.recv()

This will hold the lock forever if no response is received. In any radio communication there is always possibility of a missed response. It is possible to subject the StreamReader to a timeout as described in this section of the tutorial.

I am also doubtful about querying the UART directly when it is being handled by a StreamReader: in my opinion it's best to regard the StreamReader as the device. Querying the UART may be OK if you can be sure that no other task is also reading it... Again the more general solution is to subject the StreamReader to a timeout.

[desultory]

I am not sure about this:

        async with self.seat:
            print("Sending command: ", cmd.rstrip())
            await self.writer.awrite(cmd)
            await sleep_ms(10 if query else 50)
            while not self.uart.any():
                print("Waiting for response...")
                await sleep_ms(250)
            response = await self.recv()

This will hold the lock forever if no response is received. In any radio communication there is always possibility of a missed response. It is possible to subject the StreamReader to a timeout as described in this section of the tutorial.

I am also doubtful about querying the UART directly when it is being handled by a StreamReader: in my opinion it's best to regard the StreamReader as the device. Querying the UART may be OK if you can be sure that no other task is also reading it... Again the more general solution is to subject the StreamReader to a timeout.

I'm not sure how else i can make it keep sleeping until data is ready, and I was going to use a "python" timeout, but I see that's not in mp, I'll check out that tutorial.

I tested and I don't think it's an issue with the lock being held, the program never executes send_message a second time, or any other async stuff repeatedly as long as this lora object is created.

On the topic of no response, the response here is from the module, not necessarily a network response. So after you send the device some command, it should respond with something, if not, I may make it reset the module or something.

[desultory]

Thanks for sharing that, I considered doing a solution like that before, but it seemed a bit messy. I mean using a timeout like that is sorta like a watchdog in a sense right? Where a process to check the task sleeps for however long and cancels it if it's still running. I thought using uart.any here lets me not start reading/blocking until there is data there, and with the "seat" it should only let one thing drain the reader at a time.

[desultory]

I am not sure about this:

        async with self.seat:
            print("Sending command: ", cmd.rstrip())
            await self.writer.awrite(cmd)
            await sleep_ms(10 if query else 50)
            while not self.uart.any():
                print("Waiting for response...")
                await sleep_ms(250)
            response = await self.recv()

This will hold the lock forever if no response is received. In any radio communication there is always possibility of a missed response. It is possible to subject the StreamReader to a timeout as described in this section of the tutorial.

I am also doubtful about querying the UART directly when it is being handled by a StreamReader: in my opinion it's best to regard the StreamReader as the device. Querying the UART may be OK if you can be sure that no other task is also reading it... Again the more general solution is to subject the StreamReader to a timeout.

I'm not able to import "Delay_ms", I can't even import primitives.

[desultory]

I've refactored my code quite a bit and have uploaded it here: https://github.com/desultory/rylr998

The "server" code works fine but the code for a "client" that can just press 2 buttons to send different messages breaks, and it breaks unexplainably. It just stalls and the repl breaks, and it can send a few messages before going totally unresponsive.

The "server" can send messages just fine, using the same methods, I can't explain why it just stalls, but it's related to the async stuff almost certainly.

It stalls out here: https://github.com/desultory/rylr998/blob/main/lora.py#L178-L183

and output is like:

sleeping BEEPer
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Got data: +RCV=3,5,hello,-7,6
Waiting for data...
[3 -> 5 <rssi: -7 snr:6>] hello
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Waiting to boop
sleeping booper
Waiting to BEEP
sleeping BEEPer
Waiting to boop
boop
[0@18] Sending message: boop
Sending command: AT+SEND=0,4,boop
Waiting to BEEP
sleeping BEEPer
Waiting for response, elapsed: 3ms

[desultory]

I've been testing this a lot, I can sometimes restart the board (unplugging and plugging back in) and then I can spam both buttons and it will seemingly never crash, other times it crashes after the first or a few presses. I can't notice any trend, it doesn't seem to matter if I wait a while, spam, alternate buttons, whatever.

[peterhinch]

I'm not able to import "Delay_ms", I can't even import primitives.

Installation instructions are here - it can be done with mpremote mip.

[desultory]

I'm not able to import "Delay_ms", I can't even import primitives.

Installation instructions are here - it can be done with mpremote mip.

Thanks, but I've been using wait_for which seems to do what i need. I'd prefer to use standard library stuff as much as possible.

I've refactored the code a ton and I'm facing other async issues now

[desultory]

Are there any recommended debug steps when REPL stalls?

I'm facing a similar issue with another project using UART devices, and it will stall out sometimes after writing to the UART, with or without async. The device has a battery and I've noticed that the REPL will stall, and it will make the program grind to a halt, but I can detach the usb, wait some time, and then reattach and it works momentarily.

[peterhinch]

The nuclear option is micropython-monitor and a logic analyser. This can trigger the LA on an absence of activity - typically used to detect processor hogging.

[desultory]

I've been looking at the UART to the modem with a logic analyzer, and everything looks totally normal when this error occurs.

I'm not sure how I'd even use that code, because in this case, the runloop is likely where it's stalling. My code only has 2 loops going, and one just makes a light blink. I was getting this REPL issue without async even being used, so im pretty sure they are not related.

Maybe this has something to do with UART usage breaking the USB serial interface?

[peterhinch]

The RP2 UART has had a good deal of testing, including tests where the transmitter was shared between two cores (with suitable locking). These tests ran for hours, I think any interaction between the UART and USB would probably have been spotted.