mqtt_as: PUBACKs of messages sent with qos=1 not handled and stacking up, resulting in 1) resends, then 2) no further messages being sent and 3) eventually resulting in mem exhaustion about micropython-mqtt HOT 10 CLOSED

Dear Mirko and Peter,

thanks a stack for this discussion. I believe it is very much on the spot.

So, while I also second Peter that it is probably not a good idea to publish MQTT messages directly from within an interrupt handler, I would very much appreciate to see a respective full working example program which resonates with Mirko's use case.

Thoughts

On vanilla CPython, I would use queue.Queue in a threading scenario and asyncio.Queue in an asyncio scenario in order to connect both worlds. In other words, the interrupt handler would put messages into the queue which would be drained by an MQTT publishing/communication loop living within the main thread.

Something along the lines of:

queue = asyncio.Queue()

def interrupt_handler():
    message = "Hello world"
    queue.put_nowait(message)

async def communication_loop():
    while True:
        message = await queue.get()
        await mqtt.publish("foo/bar", message, qos=1)

Searching for uasyncio.Queue

However, I don't know whether MicroPython's uasyncio implementation actually provides a Queue class [1]. While @peterhinch's micropython-async implementation has something at [2], I haven't been able to find back a pendant with @dpgeorge's uasyncio v3 implementation quickly , but didn't investigate further.

Patches and respective discussions at micropython/micropython#6125 and micropython/micropython#6515 as well as [4] might be related.

Maybe you can just use @peterhinch's primitives.queue.Queue class [5] based on @pfalcon's original uasyncio.queues.Queue implementation [6] and add it to your userspace code?

Keep up the spirit and with kind regards,
Andreas.

[1] Unfortunately, I haven't been able to play with the recent excellent releases of vanilla MicroPython yet, because the last time we worked with it, we've been stuck with Pycom MicroPython 1.11 as this is very much behind mainline development, which I consider a very unfortunate situation. But, well, C'est la vie.
[2] https://github.com/peterhinch/micropython-async/blob/master/v3/docs/TUTORIAL.md#35-queue
[3] https://docs.micropython.org/en/latest/library/uasyncio.html
[4] Something like Threading/Queue in MicroPython?: https://forum.micropython.org/viewtopic.php?t=6084
[5] https://github.com/peterhinch/micropython-async/blob/master/v3/primitives/queue.py
[6] https://github.com/micropython/micropython-lib/blob/master/uasyncio.queues/uasyncio/queues.py

from micropython-mqtt.

This is a special case in micropython uasyncio not being interrupt aware. The moment you call "res=await q.get()" the uasyncio task will get paused until an item is available in the queue. To do that, the task is being removed from the main loop and parked at the queue object. But now the main uasyncio loop is empty and uasyncio will exit (there would be no way to fill that queue. At least not without interrupts but uasyncio is not aware of those - yet).
It will work if you have a simple other task running like:

async def do_nothing():
    while True:
        await asyncio.sleep(1)

This behaviour might puzzle you during testing but in a real program you'd actually never have a single task waiting for a queue. That would make uasyncio completely useless and you could program without uasyncio. But for testing this is indeed a case where it surprises you.

from micropython-mqtt.

The issue of interfacing uasyncio and interrupts is under discussion. One proposal is that it will be possible to trigger an Event from a soft IRQ, but this is currently unsupported. @jimmo has proposed another solution using the stream mechanism. Until an official solution emerges, the only safe way is to have the hard ISR set a flag which is polled by a coroutine. My Message primitive does this. Not very efficient, but it's currently the only safe way.

Re Queue this version is compatible with uasyncio V3: I suggest you use this until an official version is released.

Regarding the code sample, I think this may be a bug in the Event class connected with the fact that only one task is running. With a second task, everything works as expected. I will try to produce a minimal test case and raise an issue. This runs on a Pyboard:

import uasyncio as asyncio
from primitives import queue

q = queue.Queue()

async def main_loop():
    while True:
        print("-- MAIN LOOP")
        res = await q.get()
        print(res)
        await asyncio.sleep(1)
        print("End of round in in main loop.")

async def main():
    asyncio.create_task(main_loop())
    for _ in range(5):
        await asyncio.sleep(1)  # Waits here as expected
    await q.put(1)
    while True:
        await asyncio.sleep(1)

asyncio.run(main())

from micropython-mqtt.

Your use of uasyncio is (in my opinion) highly unorthodox, in particular scheduling a coroutine from an interrupt. I have never tried this and I suspect it's a bad idea. In the discussions on uasyncio I don't think this was ever considered as a possible design pattern. Bear in mind that CPython does not support interrupts and uasyncio seeks to be a micro version of asyncio.

In my designs a uasyncio application has only one instance of uasyncio.run(): this starts the whole application and, in typical firmware applications, never actually returns. I would definitely remove the timer: uasyncio has its own means of scheduling coroutines - that is its purpose.

The other thing to bear in mind with public brokers is latency, which depends on the quality of your internet connection. While mqtt_as allows concurrent publications, if communications are slow it is safer to await qos==1 publications: concurrent publications risk spawning multiple coroutines all waiting on PUBACK packets. However I would fix the other issues first, as I think the timer code is the most likely to be the source of the problem.

from micropython-mqtt.

Thanks for your very helpful answer and clarification.
Re public broker and delay: I'm also experiencing this issue with a mosquitto setup in the same LAN, public broker only used for being able to have a minimal test which can be just run.
Also the timer is solely used for this test to be able to just run, work and reproduce the issue I experience in a slightly different scenario.
My real use case: the ISR (intrpt()) is not triggered by a timer, but by an GPIO (Pin(23, Pin.IN).irq(handler=intrpt, trigger=Pin.IRQ_FALLING)).
Internally, at least on ESP32, both ISRs are executed/scheduled the same way, via mp_sched_schedule() (that's why I substituted the GPIO with the Timer while still experiencing the issue).
When you say now, the timer is the issue, then migh high-level question would be: how do I run async code (in particular: how to I publish MQTT messages) once a GPIO got triggered (on falling edge)?

from micropython-mqtt.

Maybe you can just use @peterhinch's primitives.Queue class [5] based on @pfalcon's original uasyncio.queues.Queue implementation [6] and add it to your userspace code?

That sounds quite promising, however clearly I'm struggling with how to use asyncio and its concepts, as my first test code again does not what I'd have expected:

import uasyncio as asyncio
from primitives import queue

q = queue.Queue()

async def main_loop():
    global q
    while True:
        print("-- MAIN LOOP")
        res = await q.get()
        print(res)
        await asyncio.sleep(1)
        print("End of round in in main loop.")

asyncio.run(main_loop())
print("Dropped out of main loop - why?")

drops out of the main loop after its first round:

>>> import test
-- MAIN LOOP
Dropped out of main loop - why?
>>>

Happy for any pointers on how to make myself familiar in a way that I'm not puzzled by such behaviour anymore.

from micropython-mqtt.

Hi Mirko,

while I am not having any MCU at hand, I tried your example slightly modified for vanilla CPython and it also bailed out like

RuntimeError: Task <Task pending name='Task-1' coro=<main_loop() running at testme.py:9> cb=[_run_until_complete_cb() at /Users/amo/.pyenv/versions/3.8.6/lib/python3.8/asyncio/base_events.py:184]>
got Future <Future pending> attached to a different loop

Stack Overflow to the rescue, I have been able to find [1]:

Your queues must be created inside the loop. You created them outside the loop created for asyncio.run(), so they use events.get_event_loop(). asyncio.run() creates a new loop, and futures created for the queue in one loop can't then be used in the other.

So, with this modification, the example would work on CPython - at least it will neither croak nor drop out of the main loop. Maybe it also does work on MicroPython?

import asyncio

q = None

async def main_loop():
    global q
    q = asyncio.Queue()
    while True:
        print("-- MAIN LOOP")
        res = await q.get()
        print(res)
        await asyncio.sleep(1)
        print("End of round in in main loop.")

asyncio.run(main_loop())
print("Dropped out of main loop - why?")

With kind regards,
Andreas.

[1] https://stackoverflow.com/questions/53724665/using-queues-results-in-asyncio-exception-got-future-future-pending-attached/53724990#53724990

from micropython-mqtt.

EDIT: This comment was written before @kevinkk525's answer (#54 (comment)) and as a reply to @amotl (#54 (comment))

Your queues must be created inside the loop. You created them outside the loop created for asyncio.run(), so they use events.get_event_loop(). asyncio.run() creates a new loop, and futures created for the queue in one loop can't then be used in the other.

With the following code it still drops out unfortunately:

import uasyncio as asyncio
from primitives import queue

q = None

async def main_loop():
    global q
    q = queue.Queue()
    while True:
        print("-- MAIN LOOP")
        res = await q.get()
        print(res)
        await asyncio.sleep(1)
        print("End of round in in main loop.")

asyncio.run(main_loop())
print("Dropped out of main loop - why?")

from micropython-mqtt.

@peterhinch the issue of uasyncio exiting is already open since uasyncio v3 was released: micropython/micropython#5843

from micropython-mqtt.

Ah, thank you. I had a feeling I'd come across this before but couldn't find the issue.

from micropython-mqtt.