Implement lock file mechanism when handling storage

src/ert/storage/local_experiment.py

@@ -12,15 +12,10 @@
 import xtgeo
 from pydantic import BaseModel
 
-from ert.config import (
-    ExtParamConfig,
-    Field,
-    GenKwConfig,
-    SurfaceConfig,
-)
+from ert.config import ExtParamConfig, Field, GenKwConfig, SurfaceConfig
 from ert.config.parsing.context_values import ContextBoolEncoder
 from ert.config.response_config import ResponseConfig
-from ert.storage.mode import BaseMode, Mode, require_write
+from ert.storage.mode import BaseMode, Mode, lock_access, require_write
 
 if TYPE_CHECKING:
     from ert.config.parameter_config import ParameterConfig
@@ -80,7 +75,8 @@
             (path / "index.json").read_text(encoding="utf-8")
         )
 
     @classmethod
+    @lock_access
     def create(
         cls,
         storage: LocalStorage,

src/ert/storage/mode.py

@@ -1,12 +1,10 @@
 from __future__ import annotations
 
+import fcntl
+import os
 from enum import Enum
 from functools import wraps
-from typing import (
-    TYPE_CHECKING,
-    Callable,
-    Literal,
-)
+from typing import TYPE_CHECKING, Callable, Literal
 
 if TYPE_CHECKING:
     from typing_extensions import Concatenate, ParamSpec, TypeVar
@@ -105,3 +103,21 @@ def inner(self_: C, /, *args: P.args, **kwargs: P.kwargs) -> T:
         return func(self_, *args, **kwargs)
 
     return inner
+
+
+def lock_access(func: F[C, P, T]) -> F[C, P, T]:
+    _LOCK_FILE = "storage.lock"
+
+    @wraps(func)
+    def inner(self_: C, /, *args: P.args, **kwargs: P.kwargs) -> T:
+        lock_fd = os.open(_LOCK_FILE, os.O_CREAT | os.O_RDWR)
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_EX)
+            result = func(self_, *args, **kwargs)
+        finally:
+            fcntl.flock(lock_fd, fcntl.LOCK_UN)
+            os.close(lock_fd)
+
+        return result
+
+    return inner