Skip to content

Instantly share code, notes, and snippets.

@vtermanis
Created May 12, 2018 14:19
Show Gist options
  • Save vtermanis/199619111521db585d1bc9f451b59af3 to your computer and use it in GitHub Desktop.
Save vtermanis/199619111521db585d1bc9f451b59af3 to your computer and use it in GitHub Desktop.
MySQL Connector/Python reference leak test script
#!/usr/bin/env python3
"""
Schema used:
SET sql_mode = 'NO_ENGINE_SUBSTITUTION,ANSI,MAXDB,TRADITIONAL,NO_AUTO_VALUE_ON_ZERO,ONLY_FULL_GROUP_BY';
CREATE DATABASE "leak_test";
USE "leak_test"
CREATE TABLE "parent" (
"uuid" BINARY(16) PRIMARY KEY
) ENGINE=INNODB;
CREATE TABLE "child" (
"uuid" BINARY(16) PRIMARY KEY,
"parent_uuid" BINARY(16),
"date" DATETIME NOT NULL,
"num" BIGINT UNSIGNED NOT NULL,
CONSTRAINT "fk_c_parent" FOREIGN KEY ("uuid") REFERENCES "parent"("uuid") ON UPDATE CASCADE ON DELETE CASCADE
) ENGINE=INNODB;
"""
from sys import argv, getrefcount
from functools import partial
from contextlib import closing
from datetime import datetime
import tracemalloc
from mysql.connector import connect
from mysql.connector.errors import IntegrityError
def leak_test(do_connect, loops=1):
query = 'INSERT INTO "child" ("uuid", "parent_uuid", "date", "num") VALUES (%s, %s, %s, %s)'
args = (b'0'*16, b'1'*16, datetime.utcnow(), 2**63 - 1)
for _ in range(loops):
try:
with closing(do_connect()) as conn:
with closing(conn.cursor(buffered=True)) as cursor:
cursor.execute(query, params=args)
cursor.fetchall()
except IntegrityError:
# Expecting FK error for parent_uuid as part of test
pass
def main():
try:
loops = int(argv[1])
except:
loops = 10
print('Will run %d loops' % loops)
do_connect = partial(
connect,
host='127.0.0.1',
port=3306,
user='leak_test',
password='p',
database='leak_test',
use_unicode=True,
use_pure=False,
compress=False,
sql_mode='NO_ENGINE_SUBSTITUTION,ANSI,MAXDB,TRADITIONAL,NO_AUTO_VALUE_ON_ZERO,ONLY_FULL_GROUP_BY',
charset='utf8'
)
# Exclude one-off references by running loop once beforehand
leak_test(do_connect)
refs = {item: getrefcount(item) for item in (True, False, None)}
tracemalloc.start()
snap1 = tracemalloc.take_snapshot()
leak_test(do_connect, loops=loops)
snap2 = tracemalloc.take_snapshot()
tracemalloc.stop()
print('** Reference changes (built-in types):')
print('\n'.join('%r: %d' % (item, getrefcount(item) - count) for item, count in refs.items()))
del refs
print('\n** Tracemalloc top 50')
for stat in snap2.compare_to(snap1, 'lineno')[:50]:
print(stat)
if __name__ == '__main__':
main()
@vtermanis
Copy link
Author

vtermanis commented May 12, 2018

Output including mysql/mysql-connector-python#41 and mysql/mysql-connector-python#42

Will run 100 loops
** Reference changes (built-in types):
False: 3
True: 3
None: 3

** Tracemalloc top 50
/usr/lib/python3.5/tracemalloc.py:349: size=528 B (+528 B), count=3 (+3), average=176 B
/usr/lib/python3.5/tracemalloc.py:487: size=504 B (+504 B), count=2 (+2), average=252 B
/usr/lib/python3.5/tracemalloc.py:277: size=32 B (+32 B), count=1 (+1), average=32 B
dbtest/raw.py:76: size=448 B (+0 B), count=1 (+0), average=448 B

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment