Repository: optuna/optuna
Pull request: Prevent Lock Blocking by Adding Timeout to `JournalStorage`
## Motivation
- Currently, if a process is terminated without releasing the lock, the next process fails to resume. This change introduces a mechanism to release the existing lock if acquiring it takes too long.
- ref: [JournalStorage with JournalFileSymlinkLock may fail to resume #5921](https://github.com/optuna/optuna/issues/5921)

## Description of the changes
- Introduced a grace period for acquiring locks in `JournalFileSymlinkLock` and `JournalFileOpenLock`.
- If the lock cannot be acquired within the grace period, it is considered stale and released to prevent blocking new processes.

Problem statement:
`JournalStorage` with `JournalFileSymlinkLock` may fail to resume
### Expected behavior

`JournalStorage` should be able to resume even if the process terminates unexpectedly. However, if the process terminates suddenly, a lock file may remain and it may not be able to resume.

### Environment

- Optuna version:4.2.0.dev
- Python version:3.11.0
- OS:macOS-15.1-x86_64-i386-64bit
- (Optional) Other libraries and their versions:


### Error messages, stack traces, or logs

```shell
If we encounter the above bug, the resumed process will sleep forever.
```


### Steps to reproduce

1. Terminate the process suddenly (not Ctrl-C; it deletes the lock file correctly.)
2. Resume the optimization process


### Additional context (optional)

_No response_

Hints from issue discussion:

Commit message: add test_invalid_grace_period

Patch to review:
diff --git a/optuna/storages/journal/_file.py b/optuna/storages/journal/_file.py
index a1a4e29cb82..4449508e46e 100644
--- a/optuna/storages/journal/_file.py
+++ b/optuna/storages/journal/_file.py
@@ -9,6 +9,7 @@
 import time
 from typing import Any
 import uuid
+import warnings
 
 from optuna._deprecated import deprecated_class
 from optuna.storages.journal._base import BaseJournalBackend
@@ -128,26 +129,55 @@ class JournalFileSymlinkLock(BaseJournalFileLock):
     Args:
         filepath:
             The path of the file whose race condition must be protected.
+        grace_period:
+            Grace period before an existing lock is forcibly released.
     """
 
-    def __init__(self, filepath: str) -> None:
+    def __init__(self, filepath: str, grace_period: int | None = 30) -> None:
         self._lock_target_file = filepath
         self._lock_file = filepath + LOCK_FILE_SUFFIX
+        if grace_period is not None:
+            if grace_period <= 0:
+                raise ValueError("The value of `grace_period` should be a positive integer.")
+            if grace_period < 3:
+                warnings.warn("The value of `grace_period` might be too small. ")
+        self.grace_period = grace_period
 
     def acquire(self) -> bool:
         """Acquire a lock in a blocking way by creating a symbolic link of a file.
 
         Returns:
             :obj:`True` if it succeeded in creating a symbolic link of ``self._lock_target_file``.
-
         """
         sleep_secs = 0.001
+        last_update_monotonic_time = time.monotonic()
+        mtime = None
         while True:
             try:
                 os.symlink(self._lock_target_file, self._lock_file)
                 return True
             except OSError as err:
                 if err.errno == errno.EEXIST:
+                    if self.grace_period is not None:
+                        try:
+                            current_mtime = os.stat(self._lock_file).st_mtime
+                        except OSError:
+                            continue
+                        if current_mtime != mtime:
+                            mtime = current_mtime
+                            last_update_monotonic_time = time.monotonic()
+
+                        if time.monotonic() - last_update_monotonic_time > self.grace_period:
+                            warnings.warn(
+                                "The existing lock file has not been released "
+                                "for an extended period. Forcibly releasing the lock file."
+                            )
+                            try:
+                                self.release()
+                                sleep_secs = 0.001
+                            except RuntimeError:
+                                continue
+
                     time.sleep(sleep_secs)
                     sleep_secs = min(sleep_secs * 2, 1)
                     continue
@@ -181,10 +211,18 @@ class JournalFileOpenLock(BaseJournalFileLock):
     Args:
         filepath:
             The path of the file whose race condition must be protected.
+        grace_period:
+            Grace period before an existing lock is forcibly released.
     """
 
-    def __init__(self, filepath: str) -> None:
+    def __init__(self, filepath: str, grace_period: int | None = 30) -> None:
         self._lock_file = filepath + LOCK_FILE_SUFFIX
+        if grace_period is not None:
+            if grace_period <= 0:
+                raise ValueError("The value of `grace_period` should be a positive integer.")
+            if grace_period < 3:
+                warnings.warn("The value of `grace_period` might be too small. ")
+        self.grace_period = grace_period
 
     def acquire(self) -> bool:
         """Acquire a lock in a blocking way by creating a lock file.
@@ -194,6 +232,8 @@ def acquire(self) -> bool:
 
         """
         sleep_secs = 0.001
+        last_update_monotonic_time = time.monotonic()
+        mtime = None
         while True:
             try:
                 open_flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY
@@ -201,6 +241,26 @@ def acquire(self) -> bool:
                 return True
             except OSError as err:
                 if err.errno == errno.EEXIST:
+                    if self.grace_period is not None:
+                        try:
+                            current_mtime = os.stat(self._lock_file).st_mtime
+                        except OSError:
+                            continue
+                        if current_mtime != mtime:
+                            mtime = current_mtime
+                            last_update_monotonic_time = time.monotonic()
+
+                        if time.monotonic() - last_update_monotonic_time > self.grace_period:
+                            warnings.warn(
+                                "The existing lock file has not been released "
+                                "for an extended period. Forcibly releasing the lock file."
+                            )
+                            try:
+                                self.release()
+                                sleep_secs = 0.001
+                            except RuntimeError:
+                                continue
+
                     time.sleep(sleep_secs)
                     sleep_secs = min(sleep_secs * 2, 1)
                     continue
diff --git a/tests/storages_tests/journal_tests/test_journal.py b/tests/storages_tests/journal_tests/test_journal.py
index c24b27d90f8..710b88ecaad 100644
--- a/tests/storages_tests/journal_tests/test_journal.py
+++ b/tests/storages_tests/journal_tests/test_journal.py
@@ -28,33 +28,43 @@
 from optuna.testing.tempfile_pool import NamedTemporaryFilePool
 
 
-LOG_STORAGE = [
-    "file_with_open_lock",
-    "file_with_link_lock",
-    "redis_default",
-    "redis_with_use_cluster",
+LOG_STORAGE_WITH_PARAMETER = [
+    ("file_with_open_lock", 3),
+    ("file_with_open_lock", 30),
+    ("file_with_open_lock", None),
+    ("file_with_link_lock", 3),
+    ("file_with_link_lock", 30),
+    ("file_with_link_lock", None),
+    ("redis_default", None),
+    ("redis_with_use_cluster", None),
 ]
 
 JOURNAL_STORAGE_SUPPORTING_SNAPSHOT = ["journal_redis"]
 
 
 class JournalLogStorageSupplier:
-    def __init__(self, storage_type: str) -> None:
+    def __init__(self, storage_type: str, grace_period: int | None) -> None:
         self.storage_type = storage_type
         self.tempfile: IO[Any] | None = None
+        self.grace_period = grace_period
 
     def __enter__(self) -> optuna.storages.journal.BaseJournalBackend:
         if self.storage_type.startswith("file"):
             self.tempfile = NamedTemporaryFilePool().tempfile()
             lock: BaseJournalFileLock
             if self.storage_type == "file_with_open_lock":
-                lock = optuna.storages.journal.JournalFileOpenLock(self.tempfile.name)
+                lock = optuna.storages.journal.JournalFileOpenLock(
+                    self.tempfile.name, self.grace_period
+                )
             elif self.storage_type == "file_with_link_lock":
-                lock = optuna.storages.journal.JournalFileSymlinkLock(self.tempfile.name)
+                lock = optuna.storages.journal.JournalFileOpenLock(
+                    self.tempfile.name, self.grace_period
+                )
             else:
                 raise Exception("Must not reach here")
             return optuna.storages.journal.JournalFileBackend(self.tempfile.name, lock)
         elif self.storage_type.startswith("redis"):
+            assert self.grace_period is None
             use_cluster = self.storage_type == "redis_with_use_cluster"
             journal_redis_storage = optuna.storages.journal.JournalRedisBackend(
                 "redis://localhost", use_cluster
@@ -71,8 +81,10 @@ def __exit__(
             self.tempfile.close()
 
 
-@pytest.mark.parametrize("log_storage_type", LOG_STORAGE)
-def test_concurrent_append_logs_for_multi_processes(log_storage_type: str) -> None:
+@pytest.mark.parametrize("log_storage_type,grace_period", LOG_STORAGE_WITH_PARAMETER)
+def test_concurrent_append_logs_for_multi_processes(
+    log_storage_type: str, grace_period: int | None
+) -> None:
     if log_storage_type.startswith("redis"):
         pytest.skip("The `fakeredis` does not support multi process environments.")
 
@@ -80,7 +92,7 @@ def test_concurrent_append_logs_for_multi_processes(log_storage_type: str) -> No
     num_records = 200
     record = {"key": "value"}
 
-    with JournalLogStorageSupplier(log_storage_type) as storage:
+    with JournalLogStorageSupplier(log_storage_type, grace_period) as storage:
         with ProcessPoolExecutor(num_executors) as pool:
             pool.map(storage.append_logs, [[record] for _ in range(num_records)], timeout=20)
 
@@ -88,13 +100,15 @@ def test_concurrent_append_logs_for_multi_processes(log_storage_type: str) -> No
         assert all(record == r for r in storage.read_logs(0))
 
 
-@pytest.mark.parametrize("log_storage_type", LOG_STORAGE)
-def test_concurrent_append_logs_for_multi_threads(log_storage_type: str) -> None:
+@pytest.mark.parametrize("log_storage_type,grace_period", LOG_STORAGE_WITH_PARAMETER)
+def test_concurrent_append_logs_for_multi_threads(
+    log_storage_type: str, grace_period: int | None
+) -> None:
     num_executors = 10
     num_records = 200
     record = {"key": "value"}
 
-    with JournalLogStorageSupplier(log_storage_type) as storage:
+    with JournalLogStorageSupplier(log_storage_type, grace_period) as storage:
         with ThreadPoolExecutor(num_executors) as pool:
             pool.map(storage.append_logs, [[record] for _ in range(num_records)], timeout=20)
 
@@ -244,3 +258,18 @@ def test_raise_error_for_deprecated_class_import_from_journal() -> None:
         journal.JournalRedisStorage  # type: ignore[attr-defined]
     with pytest.raises(AttributeError):
         journal.BaseJournalLogStorage  # type: ignore[attr-defined]
+
+
+@pytest.mark.parametrize(
+    "log_storage_type,grace_period",
+    [
+        ("file_with_open_lock", 0),
+        ("file_with_open_lock", -1),
+        ("file_with_link_lock", 0),
+        ("file_with_link_lock", -1),
+    ],
+)
+def test_invalid_grace_period(log_storage_type: str, grace_period: int) -> None:
+    with pytest.raises(ValueError):
+        with JournalLogStorageSupplier(log_storage_type, grace_period):
+            pass


Code review comments:
diff_hunk: @@ -244,3 +258,18 @@ def test_raise_error_for_deprecated_class_import_from_journal() -> None:
         journal.JournalRedisStorage  # type: ignore[attr-defined]
     with pytest.raises(AttributeError):
         journal.BaseJournalLogStorage  # type: ignore[attr-defined]
+
+
+@pytest.mark.parametrize(
+    "log_storage_type,grace_period",
+    [
+        ("file_with_open_lock", 0),
+        ("file_with_open_lock", -1),
+        ("file_with_link_lock", 0),
+        ("file_with_link_lock", -1),
+    ],
+)
path: tests/storages_tests/journal_tests/test_journal.py
text: @user2:
It can be written more simply.

```suggestion
@user1.mark.parametrize("log_storage_type", ("file_with_open_lock", "file_with_link_lock"))
@user1.mark.parametrize("grace_period", (0, -1))
```