Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Test to reproduce segmentation fault, caused by conn.ping()
"""Test to reproduce segmentation fault, caused by conn.ping()
"""
import threading
import time
import mariadb
HOST = ''
PORT = ''
USER = ''
PASSWORD = ''
DATABASE = ''
POOL_SIZE = 3
PING_INTERVAL = 2
"""
SIGSEGV during loop # / POOL_SIZE
3 / 1
5 / 2
6 / 3
7 / 4
8 / 5
9 / 6
10 / 7
11 / 8
12 / 9
13 / 10
"""
class PingThread(threading.Thread):
def __init__(self, conn_pool):
super().__init__()
self.get_count = 0
self.conn_pool = conn_pool
def run(self) -> None:
print('PingThread running')
while True:
self.get_count += 1
print('starting loop {}'.format(self.get_count))
try:
conn = self.conn_pool.get_connection()
# conn.ping()
conn.close()
except Exception as ex:
print(ex)
print(' sleeping after {} get of {} conns'.format(self.get_count, POOL_SIZE))
time.sleep(PING_INTERVAL)
def main():
conn_pool = mariadb.ConnectionPool(
pool_name='test_pool',
pool_size=POOL_SIZE,
pool_reset_connection=False,
host=HOST,
port=PORT,
user=USER,
password=PASSWORD,
database=DATABASE,
autocommit=True,
)
ping_thread = PingThread(conn_pool)
ping_thread.start()
ping_thread.join()
if __name__ == '__main__':
main()
@leeclemens
Copy link
Author

leeclemens commented Feb 28, 2021

Do not call .ping() explicitly

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