Merge "Clean up connection resources upon garbage collection"
This commit is contained in:
commit
dde46af207
2 files changed
+52
-26
No files matched your search
@@ -51,13 +51,28 @@ if TYPE_CHECKING:
|
||||
_T = TypeVar('_T', bound=Mapping[str, Any])
|
||||
|
||||
|
||||
def _close_requests_session(session: 'ks_session.Session | None') -> None:
|
||||
if session is None:
|
||||
return
|
||||
requests_session = getattr(session, '_session', None)
|
||||
if requests_session is not None:
|
||||
try:
|
||||
requests_session.close()
|
||||
except (OSError, requests.exceptions.RequestException):
|
||||
pass
|
||||
|
||||
|
||||
# plain function so nothing keeps ref to connection
|
||||
def _run_close(conn_ref: 'weakref.ref[_OpenStackCloudMixin]') -> None:
|
||||
conn = conn_ref()
|
||||
# Only resolves at interpreter shutdown: once the connection is collected
|
||||
# the weakref is already cleared, and there is nothing left to release.
|
||||
if conn is not None:
|
||||
conn.close()
|
||||
def _run_close(
|
||||
config: cloud_region.CloudRegion,
|
||||
pool_executor_holder: list[concurrent.futures.Executor | None],
|
||||
session_holder: list['ks_session.Session | None'],
|
||||
) -> None:
|
||||
config.set_auth_cache()
|
||||
_close_requests_session(session_holder[0])
|
||||
pool_executor = pool_executor_holder[0]
|
||||
if pool_executor:
|
||||
pool_executor.shutdown()
|
||||
|
||||
|
||||
class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
|
||||
@@ -234,9 +249,11 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
self._session: ks_session.Session | None = None
|
||||
self._session_holder: list[ks_session.Session | None] = [None]
|
||||
self._pool_executor_holder: list[
|
||||
concurrent.futures.Executor | None
|
||||
] = [pool_executor]
|
||||
self._proxies: dict[str, proxy.Proxy] = {}
|
||||
self.__pool_executor = pool_executor
|
||||
self._global_request_id = global_request_id
|
||||
self.use_direct_get = use_direct_get or False
|
||||
self.strict_mode = strict
|
||||
@@ -292,7 +309,21 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
|
||||
# `self` would keep this connection, its session and all other
|
||||
# internal objects (including TLS contexts) alive until the process
|
||||
# exits.
|
||||
self._finalizer = weakref.finalize(self, _run_close, weakref.ref(self))
|
||||
self._finalizer = weakref.finalize(
|
||||
self,
|
||||
_run_close,
|
||||
self.config,
|
||||
self._pool_executor_holder,
|
||||
self._session_holder,
|
||||
)
|
||||
|
||||
@property
|
||||
def _session(self) -> 'ks_session.Session | None':
|
||||
return self._session_holder[0]
|
||||
|
||||
@_session.setter
|
||||
def _session(self, value: 'ks_session.Session | None') -> None:
|
||||
self._session_holder[0] = value
|
||||
|
||||
@property
|
||||
def session(self) -> 'ks_session.Session':
|
||||
@@ -306,20 +337,18 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
|
||||
|
||||
@property
|
||||
def _pool_executor(self) -> concurrent.futures.Executor:
|
||||
if not self.__pool_executor:
|
||||
self.__pool_executor = concurrent.futures.ThreadPoolExecutor(
|
||||
max_workers=5
|
||||
)
|
||||
return self.__pool_executor
|
||||
executor = self._pool_executor_holder[0]
|
||||
if not executor:
|
||||
executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
|
||||
self._pool_executor_holder[0] = executor
|
||||
return executor
|
||||
|
||||
def close(self) -> None:
|
||||
"""Release any resources held open."""
|
||||
self.config.set_auth_cache()
|
||||
if self._session is not None:
|
||||
self._close_session()
|
||||
if self.__pool_executor:
|
||||
self.__pool_executor.shutdown()
|
||||
self._finalizer.detach()
|
||||
_run_close(
|
||||
self.config, self._pool_executor_holder, self._session_holder
|
||||
)
|
||||
|
||||
def _close_session(self) -> None:
|
||||
"""Release the connection pool of the keystoneauth1 session.
|
||||
@@ -335,12 +364,7 @@ class _OpenStackCloudMixin(_services_mixin.ServicesMixin):
|
||||
propagate out of ``close()``, which also runs from the finalizer at
|
||||
interpreter shutdown and from the context manager exit.
|
||||
"""
|
||||
requests_session = getattr(self._session, '_session', None)
|
||||
if requests_session is not None:
|
||||
try:
|
||||
requests_session.close()
|
||||
except (OSError, requests.exceptions.RequestException):
|
||||
pass
|
||||
_close_requests_session(self._session)
|
||||
|
||||
def __enter__(self) -> Self:
|
||||
return self
|
||||
|
||||
@@ -136,7 +136,8 @@ class TestCloud(base.TestCase):
|
||||
with mock.patch.object(ks.session, 'close', side_effect=exc):
|
||||
self.cloud.close() # must not raise
|
||||
|
||||
def test_unclosed_connection_is_garbage_collected(self):
|
||||
@mock.patch('openstack.cloud.openstackcloud._run_close')
|
||||
def test_unclosed_connection_is_garbage_collected(self, mock_run_close):
|
||||
conn = connection.Connection(config=self.cloud.config)
|
||||
conn_ref = weakref.ref(conn)
|
||||
|
||||
@@ -144,6 +145,7 @@ class TestCloud(base.TestCase):
|
||||
gc.collect()
|
||||
|
||||
self.assertIsNone(conn_ref())
|
||||
mock_run_close.assert_called_once()
|
||||
|
||||
def test_close_keeps_session_usable(self):
|
||||
# close() only releases pooled connections; the session must
|
||||
|
||||
Reference in new issue
Block a user