Is there a memory leak in umqtt.simple or am I doing this wrong?

[Redsandro]

I'm using a Pimoroni Inky Frame 4.0 with a Raspberry Pi Pico W running the pimoroni-pico build of MicroPython 1.21.0 to show the latest cat detected by Frigate over MQTT using umqtt.simple 1.3.4 downloaded using Thonny.

So I was subscribed to topic frigate/pi4/mycamera/# and only check retained messages on connect one by one (about 14 of them including 4 pictures) until I came across a cat picture at frigate/pi4/mycamera/cat/snapshot. Every other picture I'd just ignore, not do anything, assuming no memory references to those messages would exist by the time I do a gc.collect() at some later time.

def on_message_cb(topic, msg):
    if topic == b"frigate/pi4/mycamera/cat/snapshot":
        # Memory consuming code here
    # Ignore every other topic, not making references to `msg`

I had however great difficulty showing the smallest picture due to lack of memory with the final jpegdec.JPEG(graphics).decode() command throwing MemoryError: memory allocation failed, allocating 16112 bytes, even though the gc.collect() right before it gave me 40k to play with (although I think those numbers are misleading).

Now here is the part I don't understand:

When I subscribe to frigate/pi4/mycamera/cat/snapshot directly so there's only 1 callback to handle in stead of a whole bunch of them, the script works flawlessly. I can even double the size of the pictures, and I don't run into MemoryError.

Why does this happen? It should not matter. The messages in the ignored callbacks should simply be reclaimed by gc.collect(). Am I wrong?

To me it feels like a memory leak, but I know in reality it is infinitely more probable that I made mistakes in my code. I hope someone can point that out. Here's the whole script I called mqtt_img.py. For those unfamiliar with the Inky Frame; the "app" (script) is loaded by the launcher main.py and update() is executed:

import gc, io, os, time, machine, jpegdec
import inky_helper as ih
from umqtt.simple import MQTTClient
from json import dumps, loads
from binascii import hexlify, crc32
from helpers import gc_collect, file_exists
from secrets import secret

# just gc.collect() decorated with a print statement and gc.mem_free()
gc_collect('begin')

# umqtt topics need to be binary for successful comparison
TOPIC_SUB = b""
TOPIC_SUB_ETC = b""
CACHE_FILE='mqtt_img.json'
mqtt_client = None

# Pimoroni Inky Frame specific code
UPDATE_INTERVAL = 5
graphics = None
WIDTH = None
HEIGHT = None
leds = ih.LEDController()

# Called by Inky Frame launcher
def update():
    client = connectMQTT()    
    loops = n = 24 # Checking two dozen messages should be enough
    
    while n > 0:
        print(f'Checking message ({loops-n+1})')
        try:
            client.check_msg()
        except OSError as e:
            if e.args[0] == 9:
                # Done on purpose in the callback to conserve memory
                print("MQTT Client was disconnected.")
                break
        n -= 1
        
    if n == 0:
        print(f'Parsed {loops} messages and nothing happened.')
    else:
        print('Client disconnected. Time to go home.')

def connectMQTT():
    global mqtt_client
    client_id = hexlify(machine.unique_id())
    client_id = f"inky_frame-{client_id.decode('ascii')[:6]}"
    client = MQTTClient(client_id=client_id,
        server=b"mymqtt.localhost",
        port=1883,
        user=b"myuser",
        password=b"mypassword",
        keepalive=7200,
    )

    client.set_callback(on_message_cb)
    client.connect()
    client.subscribe(TOPIC_SUB)
    
    print(f'Connected to MQTT broker, subscribed to {TOPIC_SUB}')
    gc_collect('subscribed')
    mqtt_client = client
    
    return client

def on_message_cb(topic, msg):
    global mqtt_client
    
    if topic == TOPIC_SUB:
        mqtt_client.disconnect()
        buffer = io.BytesIO(msg)
        
        # Calculate hash to compare with cached image
        img_hash = crc32(msg)
        gc_collect('hashed')
        
        # Get old hash from cache
        if file_exists(CACHE_FILE):
            try:
                data = loads(open(CACHE_FILE, "r").read())
                if type(data) is dict:
                    # If the e-ink screen was already on this page, don't refresh old pictures
                    if data['imgHash'] == img_hash and ih.state['run'] == ih.state['prev']:
                        # No update needed; Go back to sleep
                        print('Cache hit. Nothing to do.')
                        return
                else:
                    # File corrupted; Remove
                    os.remove('mqtt_img.json')
            except OSError:
                print('Unexpected OSError.')
                pass
            print('Cache compared, needs redraw')
        # Write cache
        with open(CACHE_FILE, "w") as f:
            f.write(dumps({"imgHash": img_hash}))
            f.flush()
        
        # Rewind buffer and write to file
        buffer.seek(0)
        with open('mqtt_img.jpg', "wb") as f:
            while True:
                chunk = buffer.read(1024)
                if not chunk:
                    break
                f.write(chunk)
                
        # We have a serious need for memory in order to draw the jpeg.
        # Since we call draw(), these vars are still referenced.
        del buffer, msg, img_hash, mqtt_client, data, chunk
        gc_collect('img_draw')
        draw(True)
        
def draw(got_img = False):
    # Override call from main.py 
    if not got_img:
        return
    
    # Set LED status normally done by main.py
    leds.on("warning")
    gc_collect('jpeg')
    jpeg = jpegdec.JPEG(graphics) ### MemoryError here ###
    graphics.set_pen(1)
    graphics.clear()
    jpeg.open_file('mqtt_img.jpg')
    gc_collect('jpdecode')
    jpeg.decode()
    graphics.update()

To summarize: The script works as long as I subscribe directly to the snapshot topic containing one single snapshot. I run into memory errors if I subscribe to a parent topic that contains multiple snapshots, even if I ignore everything but the one I am looking for.

[glenn20]

My first thought is that the other incoming messages would cause increased memory fragmentation, even though you ignore them. Just how fragmented the memory becomes would depend on the rate and pattern of message sizes you receive. If the memory becomes sufficiently fragmented, you will be unable to allocate larger blocks of memory.

PS. Apologies in advance for some other unsolicited advice:

• though not relevant to this particular problem, I think you should be using client.wait_msg(), not client.check_msg() in update() - if I am reading your logic correctly.
• explicitly close the CACHE_FILE in on_message_cb() when you open it for reading
• save memory (and memory fragmentation) by just writing the msg to the disk file in update(), rather than wrapping it in a BytesIO instance and writing it in chunks:

  with open('mqtt_img.jpg', "wb") as f:
      f.write(msg)

• and in what I expect is the common case, you may save more memory by drawing the jpeg fom the in-memory msg, rather than reading it back from the file (though that depends on just how the app is used).

[robert-hh]

But I haven't figured out how to run mpy-cross on the Pico.

You have to run mpy-cross on your PC and upload the .mpy files.

[Redsandro]

You have to run mpy-cross on your PC

@robert-hh don't I need one built specifically for the target MCU and its MicroPython firmware? The Pimoroni "Inky Frame" uses a custom build of MicroPython but they haven't documented how to get, use or build mpy-cross for the Inky Frame.

[glenn20]

.mpy files are generally compatible across different ports and, in many cases, micropython release versions. You can see the details in the micropython docs at https://docs.micropython.org/en/latest/reference/mpyfiles.html#versioning-and-compatibility-of-mpy-files. That explains how to identify the bytecode version of your micropython, and which versions are compatible with that bytecode. Then, you should be able to download a compatible release of micropython and compile mpy-cross (https://github.com/micropython/micropython/blob/master/mpy-cross/README.md). I believe there are plans to make a release of mpy-cross that can be distributed without having to build it yourself, but that is not the case yet.

[glenn20]

Thank you for the elaborate educational response. With everything implemented, and saving another kilobyte by shrinking the code (including Pimoroni's scaffolding), everything is running smoothly.

Excellent.. glad to hear it.

Although if I keep the device running long enough, it could or should eventually run into the same problem because slowly but surely it keeps receiving (and discarding) MQTT messages that may end up fragmenting free memory over time.

Yes, that is true. However, if you are not writing and reading files to/from disk, that should signifcantly reduce memory demand. If it is a significant issue, the wait_msg() function could be rewritten to use readinto() to save each msg into a single buffer that gets reused for each msg. There are some complexities to that, but it could substantially reduce memory fragmentation. I'd only consider that if it was necessary.

[glenn20]

A simpler solution would probably be to just do a soft_reset() when you get an ENOMEM exception (assuming it does not happen too frequently).

[Josverl]

you can get information on the memory map using
import micropython; micropython.mem_info(True)
to visualize that I have used a Jupyter notebook like in this example, or this plot
Still that does not give you the info of exactly what object/action is taking space ( although it did help me understand more of the allocation pattern if I slowly step though the allocations map)

[Redsandro]

Thanks, this is promising. I have to take some time and learn to understand the output.