spatialdude
Possible memory leak
Good day
If I run the attached script (see below), tested on windows, unix and stm32f769 builds, it runs fine until the free memory drops below 5%, and then crashes with the error:
Traceback (most recent call last):
File "E:\scratch\micropython\mp_usqlite\samples\producer_consumer_sd.py", line 167, in
If I run the script with useSqlite = False, the memory usage drops until about 1%, but the the gc kicks in and continues as expected.
Although, if useSqlite = True, when the memory usage drops to around 5%, the above error is displayed. Might it be an issue in the interop between the gc and usqlite?
Regards
Script:
import uasyncio as asyncio from queue import Queue import gc import os import usqlite
Exception raised by get_nowait().
class QueueEmpty(Exception): pass
Exception raised by put_nowait().
class QueueFull(Exception): pass
class Queue: def init(self, maxsize=0): self.maxsize = maxsize self._queue = [] self._evput = asyncio.Event() # Triggered by put, tested by get self._evget = asyncio.Event() # Triggered by get, tested by put
def _get(self):
self._evget.set() # Schedule all tasks waiting on get
self._evget.clear()
return self._queue.pop(0)
async def get(self): # Usage: item = await queue.get()
while self.empty(): # May be multiple tasks waiting on get()
# Queue is empty, suspend task until a put occurs
# 1st of N tasks gets, the rest loop again
await self._evput.wait()
return self._get()
def get_nowait(self): # Remove and return an item from the queue.
# Return an item if one is immediately available, else raise QueueEmpty.
if self.empty():
raise QueueEmpty()
return self._get()
def _put(self, val):
self._evput.set() # Schedule tasks waiting on put
self._evput.clear()
self._queue.append(val)
async def put(self, val): # Usage: await queue.put(item)
while self.full():
# Queue full
await self._evget.wait()
# Task(s) waiting to get from queue, schedule first Task
self._put(val)
def put_nowait(self, val): # Put an item into the queue without blocking.
if self.full():
raise QueueFull()
self._put(val)
def qsize(self): # Number of items in the queue.
return len(self._queue)
def empty(self): # Return True if the queue is empty, False otherwise.
return len(self._queue) == 0
def full(self): # Return True if there are maxsize items in the queue.
# Note: if the Queue was initialized with maxsize=0 (the default) or
# any negative number, then full() is never True.
return self.maxsize > 0 and self.qsize() >= self.maxsize
event_test = asyncio.Event() useSqlite = False
async def produce(queue): icount = 0
while True:
await queue.put(str(icount)) # Put result on queue
icount += 1
await asyncio.sleep(.2)
async def consume(queue, dbConn): while True: await event_test.wait()
while (queue.qsize() > 20):
item = await queue.get() # Blocks until data is ready
print("Pulled item..." + str(item))
if (useSqlite == True):
dbConn.executemany(
"BEGIN TRANSACTION;"
f"INSERT INTO items VALUES ({item}, 0, '{item}', '{item}', False);"
"COMMIT;")
with dbConn.execute("SELECT COUNT(ID) from items;") as cur:
for row in cur:
print("items count:", row[0])
if (row[0] > 100):
dbConn.executemany(
"BEGIN TRANSACTION;"
"DELETE FROM items;"
"COMMIT;")
with open('sd/hello.txt', 'w+') as f:
f.write(item)
await asyncio.sleep(.1)
event_test.clear()
async def watchQueue(queue): while True: if (queue.qsize() > 10): event_test.set()
await asyncio.sleep(1)
async def showQueueLength(queue): while True: print("Queue length..." + str(queue.qsize())) await asyncio.sleep(2)
async def showMemUsage(): while True: print(free(True)) await asyncio.sleep(5)
def free(full=False): F = gc.mem_free() A = gc.mem_alloc() T = F+A P = '{0:.2f}%'.format(F/T*100) if not full: return P else : return ('Total:{0} Free:{1} ({2})'.format(T,F,P))
def InitDb(clearDb = True): dbconn = usqlite.connect("data.db")
if clearDb:
dbconn.execute("DROP TABLE IF EXISTS items;")
dbconn.execute('''CREATE TABLE IF NOT EXISTS items
(ID INTEGER,
VERSION INTEGER NOT NULL,
NAME TEXT NOT NULL,
DESCRIPTION TEXT NOT NULL,
IS_COMPLETE BOOL NOT NULL);''')
dbconn.execute("CREATE UNIQUE INDEX index_item_name ON items(name, description);")
return dbconn
#os.rmdir('sd') #os.mkdir('sd') os.remove("data.db")
async def main():
queue = Queue()
dbConn = InitDb()
asyncio.create_task(produce(queue))
asyncio.create_task(consume(queue, dbConn))
asyncio.create_task(watchQueue(queue))
asyncio.create_task(showQueueLength(queue))
asyncio.create_task(showMemUsage())
print('Tasks are running...')
while True:
await asyncio.sleep(5)
asyncio.run(main())
Apologies for the messy code part of this issue, pasting the code did not work as expected, I have attached the script.
Regards producer_consumer_sd.zip
@emile-cronje μSQLite has been updated to v0.1.5 which includes the latest version of SQLite and support for MicroPython v1.23. This may address some of the problems you've encountered.
Please also take a look at the simple memory test shown in https://github.com/spatialdude/usqlite/issues/27#issuecomment-2143392882 that shows how to trace calls and track peak memory usage that may also assist in narrowing down problems.
