From d42df2cb1a5dad2f90d554a118f86bcd28df2e81 Mon Sep 17 00:00:00 2001 From: Kaivan Kamali Date: Mon, 26 Sep 2022 17:12:30 -0400 Subject: [PATCH] Created integration test for connection pool monitor. Revised connection pool monitor thread creation/destruction. --- lib/galaxy/objectstore/irods.py | 32 ++++++++--------- .../test_objectstore_datatype_upload.py | 34 +++++++++++++++++-- 2 files changed, 47 insertions(+), 19 deletions(-) diff --git a/lib/galaxy/objectstore/irods.py b/lib/galaxy/objectstore/irods.py index 1575f792e03..fbbc2d0ecca 100644 --- a/lib/galaxy/objectstore/irods.py +++ b/lib/galaxy/objectstore/irods.py @@ -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() diff --git a/test/integration/objectstore/test_objectstore_datatype_upload.py b/test/integration/objectstore/test_objectstore_datatype_upload.py index 908077243b3..d8648a46992 100644 --- a/test/integration/objectstore/test_objectstore_datatype_upload.py +++ b/test/integration/objectstore/test_objectstore_datatype_upload.py @@ -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( """ @@ -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