From 69138a4a415c56768d6ce93839b1dd9a7c955476 Mon Sep 17 00:00:00 2001 From: Michael Adkins Date: Wed, 13 Apr 2022 12:32:02 -0500 Subject: [PATCH 1/3] Update `locket._LockFile` implementation to avoid leaving open files The `LockFile` implementation opens a file descriptor to check locks. In the non-blocking case, if the lock cannot be acquired, a `LockError` is thrown and an open file descriptor is left dangling. In normal operation, this file is closed on lock release, but since the lock has already been acquired by another instance of the `LockFile`, release never occurs. --- distributed/locket.py | 31 ++++++++++++++----------------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/distributed/locket.py b/distributed/locket.py index 73529ae03a5..15ba483659c 100644 --- a/distributed/locket.py +++ b/distributed/locket.py @@ -154,26 +154,23 @@ def release(self): class _LockFile: def __init__(self, path): self._path = path - self._file = None self._thread_lock = threading.Lock() def acquire(self, timeout=None, retry_period=None): - if self._file is None: - self._file = open(self._path, "wb") - if timeout is None and _lock_file_blocking_available: - _lock_file_blocking(self._file) - else: - _acquire_non_blocking( - acquire=lambda: _lock_file_non_blocking(self._file), - timeout=timeout, - retry_period=retry_period, - path=self._path, - ) - - def release(self): - _unlock_file(self._file) - self._file.close() - self._file = None + with open(self._path, "wb") as fd: + if timeout is None and _lock_file_blocking_available: + _lock_file_blocking(fd) + else: + _acquire_non_blocking( + acquire=lambda: _lock_file_non_blocking(fd), + timeout=timeout, + retry_period=retry_period, + path=self._path, + ) + + def release(self): + with open(self._path, "wb") as fd: + _unlock_file(fd) class _Locker: From aa72d74b94c123e745abcbbddfbad9c1675e99d6 Mon Sep 17 00:00:00 2001 From: Michael Adkins Date: Wed, 13 Apr 2022 15:11:17 -0500 Subject: [PATCH 2/3] Fix indent of release :facepalm: --- distributed/locket.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/distributed/locket.py b/distributed/locket.py index 15ba483659c..590ee456391 100644 --- a/distributed/locket.py +++ b/distributed/locket.py @@ -168,9 +168,9 @@ def acquire(self, timeout=None, retry_period=None): path=self._path, ) - def release(self): - with open(self._path, "wb") as fd: - _unlock_file(fd) + def release(self): + with open(self._path, "wb") as fd: + _unlock_file(fd) class _Locker: From a37b87ea8bb3dd466abec0fbd69a9a10d9a92c66 Mon Sep 17 00:00:00 2001 From: Michael Adkins Date: Wed, 13 Apr 2022 16:39:09 -0500 Subject: [PATCH 3/3] Add check for missing file on release --- distributed/locket.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/distributed/locket.py b/distributed/locket.py index 590ee456391..3b1c016e1e2 100644 --- a/distributed/locket.py +++ b/distributed/locket.py @@ -3,7 +3,9 @@ from __future__ import annotations +import os import threading +import warnings import weakref from time import sleep @@ -169,8 +171,9 @@ def acquire(self, timeout=None, retry_period=None): ) def release(self): - with open(self._path, "wb") as fd: - _unlock_file(fd) + if os.path.exists(self._path): + with open(self._path, "wb") as fd: + _unlock_file(fd) class _Locker: