opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-10-13 08:51:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

Created integration test for connection pool monitor. Revised connection pool monitor thread creation/destruction.

Browse Source
This commit is contained in:
Kaivan Kamali
2022-09-26 17:12:30 -04:00
parent 38b149b187
commit d42df2cb1a
2 changed files with 47 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/objectstore/irods.py
+15 -17
View File
@@ -30,7 +30,6 @@ from galaxy.util import (
unlink,
)
from galaxy.util.path import safe_relpath
from galaxy.util.sleeper import Sleeper
from ..objectstore import DiskObjectStore
IRODS_IMPORT_MESSAGE = "The Python irods package is required to use this feature, please install it"
@@ -236,17 +235,12 @@ class IRODSObjectStore(DiskObjectStore, CloudConfigMixin):
# Set connection timeout
self.session.connection_timeout = self.timeout
# This Event object is initialized to False
# It is set to True in shutdown(), causing
# the connection pool monitor thread to return/terminate
self.stop_connection_pool_monitor_event = threading.Event()
# Helper for interruptible sleep
self.sleeper = Sleeper()
# Start connection pool monitor
if self.connection_pool_monitor_interval != -1:
self.start_connection_pool_monitor()
# This Event object is initialized to False
# It is set to True in shutdown(), causing
# the connection pool monitor thread to return/terminate
self.stop_connection_pool_monitor_event = threading.Event()
self.connection_pool_monitor_thread = None
log.debug("irods_pt __init__: %s", ipt_timer)
@@ -259,8 +253,11 @@ class IRODSObjectStore(DiskObjectStore, CloudConfigMixin):
except OSError:
pass
# Set to True so the connection pool monitor thread will return/terminate
self.stop_connection_pool_monitor_event.set()
if self.connection_pool_monitor_interval != -1:
# Set to True so the connection pool monitor thread will return/terminate
self.stop_connection_pool_monitor_event.set()
self.connection_pool_monitor_thread.join(5)
log.debug("irods_pt shutdown: %s", ipt_timer)
@classmethod
@@ -275,20 +272,21 @@ class IRODSObjectStore(DiskObjectStore, CloudConfigMixin):
"refresh_time": self.refresh_time,
"connection_pool_monitor_interval": self.connection_pool_monitor_interval,
"stop_connection_pool_monitor_event": self.stop_connection_pool_monitor_event,
"sleeper": self.sleeper,
},
name="ConnectionPoolMonitorThread",
daemon=True,
)
self.connection_pool_monitor_thread.start()
log.info("Connection pool monitor started")
def start(self):
self.start_connection_pool_monitor()
if self.connection_pool_monitor_interval != -1:
self.start_connection_pool_monitor()
def _connection_pool_monitor(self, *args, **kwargs):
refresh_time = kwargs["refresh_time"]
connection_pool_monitor_interval = kwargs["connection_pool_monitor_interval"]
stop_connection_pool_monitor_event = kwargs["stop_connection_pool_monitor_event"]
sleeper = kwargs["sleeper"]
while not stop_connection_pool_monitor_event.is_set():
curr_time = datetime.now()
@@ -303,7 +301,7 @@ class IRODSObjectStore(DiskObjectStore, CloudConfigMixin):
)
)
self.session.pool.release_connection(conn, True)
sleeper.sleep(connection_pool_monitor_interval)
stop_connection_pool_monitor_event.wait(connection_pool_monitor_interval)
def to_dict(self):
as_dict = super().to_dict()
test/integration/objectstore/test_objectstore_datatype_upload.py
+32 -2
View File
@@ -14,13 +14,19 @@ from ..test_datatype_upload import (
UploadTestDatatypeDataTestCase,
)
REFRESH_TIME = 30
CONNECTION_POOL_MONITOR_INTERVAL = 60
SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
IRODS_OBJECT_STORE_HOST = os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_HOST", "localhost")
IRODS_OBJECT_STORE_PORT = int(os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_PORT", 1247))
IRODS_OBJECT_STORE_TIMEOUT = int(os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_TIMEOUT", 30))
IRODS_OBJECT_STORE_REFRESH_TIME = int(os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_REFRESH_TIME", 300))
IRODS_OBJECT_STORE_REFRESH_TIME = int(
os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_REFRESH_TIME", REFRESH_TIME)
)
IRODS_OBJECT_STORE_CONNECTION_POOL_MONITOR_INTERVAL = int(
os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_CONNECTION_POOL_MONITOR_INTERVAL", 600)
os.environ.get(
"GALAXY_INTEGRATION_IRODS_OBJECT_STORE_CONNECTION_POOL_MONITOR_INTERVAL", CONNECTION_POOL_MONITOR_INTERVAL
)
)
IRODS_OBJECT_STORE_USERNAME = os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_USERNAME", "rods")
IRODS_OBJECT_STORE_PASSWORD = os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_PASSWORD", "rods")
@@ -28,6 +34,7 @@ IRODS_OBJECT_STORE_RESOURCE = os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_ST
IRODS_OBJECT_STORE_ZONE = os.environ.get("GALAXY_INTEGRATION_IRODS_OBJECT_STORE_ZONE", "tempZone")
# Run test for only the first 10 test files
TEST_CASES = dict(list(TEST_CASES.items())[0:10])
SINGLE_TEST_CASE = dict(list(TEST_CASES.items())[0:1])
DISTRIBUTED_OBJECT_STORE_CONFIG = string.Template(
"""
<object_store type="distributed">
@@ -194,3 +201,26 @@ def test_upload_datatype_irods(irods_instance, test_data, temp_file):
@pytest.mark.parametrize("test_data", TEST_CASES.values(), ids=list(TEST_CASES.keys()))
def test_upload_datatype_dos_irods_and_disk(distributed_and_irods_instance, test_data, temp_file):
upload_datatype_helper(distributed_and_irods_instance, test_data, temp_file)
@pytest.mark.parametrize("test_data", SINGLE_TEST_CASE.values(), ids=list(SINGLE_TEST_CASE.keys()))
def test_upload_datatype_irods_idle_connections(irods_instance, test_data, temp_file):
# Upload a file to iRods
upload_datatype_helper(irods_instance, test_data, temp_file, True)
# Get Irods object store's connection pool
connection_pool = irods_instance._test_driver.app.object_store.session.pool
# Verify the connection pool has 0 active and 1 idle connections
assert len(connection_pool.active) == 0
assert len(connection_pool.idle) == 1
# Wait for the idle connection to turn stale
time.sleep(REFRESH_TIME)
# Wait for the connection pool monitor thread to run and reclaim the stale idle connection
time.sleep(CONNECTION_POOL_MONITOR_INTERVAL)
# Check that the stale idle connection has been reclaimed
assert len(connection_pool.active) == 0
assert len(connection_pool.idle) == 0