Skip to content

Instantly share code, notes, and snippets.

@leeclemens
Last active February 28, 2021 02:03
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save leeclemens/0795d957588410f69fc0a6453ecec94f to your computer and use it in GitHub Desktop.
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

Do not call .ping() explicitly

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