[Python] asyncio httpio boto3 how to find memory leaks

I'm struggling since a while to determine where my memory leaks are coming from. I'm using asyncio, httpio boto3 and so on. The basic idea of my app is :

• getting urls of the newest documents
• transforming the documents in db files on s3 (aws services)

Like everyone else in such a situation, I don't really understand the origin of my memory leaks since I work with batches and I thought that everything should be released when the async batch work is finished. I have read some articles about it, I can see there are memory leaks but I still don't understand where it comes from.

I'll try to write some summary of my code to have a basis for discussions with you.

# main_file.py
# Function to process a single subject
async def process_subject(subject, api_call_semaphore):
await update_doc_data(subject, api_call_semaphore, True)

#@profile
async def process_subjects():
tracemalloc.start(25) # Start tracking memory usage
api_call_semaphore = asyncio.Semaphore(3)
start_index = 0
batch_size = 5
subjects = get_subjects_to_process()

for i in range(start_index, len(subjects), batch_size):
print("alived tasks")
print(len(asyncio.all_tasks()))
batch_subjects = subjects[i:i+batch_size]
print(len(batch_subjects))
tasks = [

process_subjects(subject, api_call_semaphore)
for subject in batch_subjects
]
results = await asyncio.gather(*tasks, return_exceptions=True)

for result in results:
if isinstance(result, Exception):
print(f"Exception occurred: {result}")
# Print the traceback
traceback.print_exception(type(result), result, result.__traceback__)
result.__traceback__ = None
i = len(items)
else:
print(f"Successful result: {result}")

del results
del tasks
gc.collect()

#Print top memory allocations
# snapshot = tracemalloc.take_snapshot()
# top_stats = snapshot.statistics('lineno')

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('traceback')

# pick the biggest memory block
stat = top_stats[0]
print("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))
for line in stat.traceback.format():
print(line)

print("\n Top Memory Usage:")
for stat in top_stats[:10]: # Show top 10 lines consuming memory
print(stat)

gc.collect() # Force garbage collection
unreleased = gc.get_objects() # Get all objects still in memory
unreleased_tasks = [obj for obj in gc.get_objects() if isinstance(obj, asyncio.Task) and not obj.done()]
print(f"Total objects in memory: {len(unreleased)}")

for obj in unreleased[:10]: # Limit to 10 for readability
print(type(obj), repr(obj))



print("\n Most Common Objects in Memory:")
objgraph.show_most_common_types(limit=10)

print("\n References to asyncio Tasks:")
objgraph.show_backrefs(unreleased_tasks, max_depth=5)






if __name__ == "__main__":
os.environ["TOKENIZERS_PARALLELISM"] = "false"
asyncio.run(process_items())



In the others files are the processing


async def fetch_json_with_retry(session, url, headers, retries=3):
timeout = aiohttp.ClientTimeout(total=30) # Increase the global timeout
for attempt in range(retries):
try:
async with session.get(url, headers=headers, timeout=timeout) as response:
if response.status == 200:
return await response.json()# ✅ Await response.json()
else:
print(f"⚠️ HTTP {response.status} for {url}")
return []
except asyncio.TimeoutError :
print(f" timeout {url}. Check network.")
except aiohttp.ClientConnectorError:
print(f" Cannot connect to {url}. Check network.")
except aiohttp.ServerDisconnectedError:
print(f" Attempt {attempt + 1}/{retries}: Server disconnected.")
except aiohttp.ClientOSError as e:
print(f"⚠️ Client error: {e}. Retrying in 2s...")
except Exception as e:
print(f"❓ Unexpected error: {e}. Retrying in 2s...")
await asyncio.sleep(10) # ✅ Wait before retrying

print("❌ Failed after multiple attempts.")
return []

async def get_docs(subject, min_year, api_call_semaphore):
# Output: empty list or a list of docs if not empty
doc_source_url = f""

headers = {
"User-Agent": ""
}

async with aiohttp.ClientSession() as session:

data = await fetch_json_with_retry(session, doc_source_url, headers)
#print(data)
filtered_data = [entry for entry in data if entry.get("year", "2000") >= min_year]
return filtered_data


async def update_doc_data(subject, api_call_semaphore, forced):
docs = await get_docs(subject, "2024", api_call_semaphore)
//processing documents
//saving documents



Processing documents is basically using an embedding api on text chuncks of the docs where docs is a list of dictionnaries with meta data and the text itself.

As said earlier I thought that by doing

for i in range(start_index, len(subjects), batch_size):
print("alived tasks")
print(len(asyncio.all_tasks()))
batch_subjects = subjects[i:i+batch_size]
print(len(batch_subjects))
tasks = [

process_subjects(subject, api_call_semaphore)
for subject in batch_subjects
]
results = await asyncio.gather(*tasks, return_exceptions=True)


it should avoid any issues since the tasks should be finished to go to the next batch. But it's not the case, mprofile is showing a memory usage increasing like stairs.

Tracemalloc gives me this for example :

Most Common Objects in Memory:
dict 111418
function 82692
tuple 61973
weakref 31820
OrderedDict 25662
list 19923
getset_descriptor 15822
builtin_function_or_method 12864
Statistic 11595
Traceback 11595



Most Common Objects in Memory:
dict 112494
function 82896
tuple 78201
Statistic 49685
Traceback 49685
weakref 32050
OrderedDict 25686
list 20334
getset_descriptor 15822
builtin_function_or_method 13032


We can see all objects increasing over time, especially statistic and traceback which I don't really understand.

Please can you tell me how i'm suppose to find the lines responsible of the memory leaks and or giving some tips for a work around. I already tried to free traceback as you can see but it was not really working as you can see.

Continue reading...