Skip to content

readwritelock

readwritelock

Module for Read/Write lock for fsspec filesystems

Classes

FileLock 

FileLock(retry_time=10, timeout=172800, is_acquired=False)

Bases: ABC

Abstract base class for file locks.

Initialize FileLock

Source code in niceml/utilities/readwritelock.py 
20
21
22
23
24
25
26
def __init__(
    self, retry_time: float = 10, timeout: float = 172800, is_acquired: bool = False
):
    """Initialize FileLock"""
    self.retry_time = retry_time
    self.timeout = timeout
    self.is_acquired: bool = is_acquired
Functions
__enter__ 
__enter__()

context manager enter method

Source code in niceml/utilities/readwritelock.py 
44
45
46
47
def __enter__(self):
    """context manager enter method"""
    self.acquire()
    return self
__eq__ 
__eq__(other)

Compares two file locks

Source code in niceml/utilities/readwritelock.py 
53
54
55
56
57
58
59
def __eq__(self, other: "FileLock"):
    """Compares two file locks"""
    return (
        self.retry_time == other.retry_time
        and self.timeout == other.timeout
        and self.is_acquired == other.is_acquired
    )
__exit__ 
__exit__(exc_type, exc_value, traceback)

context manager exit method

Source code in niceml/utilities/readwritelock.py 
49
50
51
def __exit__(self, exc_type, exc_value, traceback):
    """context manager exit method"""
    self.release()
acquire abstractmethod 
acquire()

Acquire the lock

Source code in niceml/utilities/readwritelock.py 
28
29
30
@abstractmethod
def acquire(self):
    """Acquire the lock"""
force_delete abstractmethod 
force_delete()

Force delete the lock

Source code in niceml/utilities/readwritelock.py 
40
41
42
@abstractmethod
def force_delete(self):
    """Force delete the lock"""
is_acquirable abstractmethod 
is_acquirable()

Checks if lock file does not exist

Source code in niceml/utilities/readwritelock.py 
32
33
34
@abstractmethod
def is_acquirable(self) -> bool:
    """Checks if lock file does not exist"""
release abstractmethod 
release()

Release the lock

Source code in niceml/utilities/readwritelock.py 
36
37
38
@abstractmethod
def release(self):
    """Release the lock"""

ReadLock 

ReadLock(
    path_config,
    retry_time=10,
    retry_await_time=0,
    timeout=172800,
    write_lock_name="write.lock",
    read_lock_name="read.lock",
    is_acquired=False,
)

Bases: FileLock

Read lock for fsspec filesystems.

Initialize ReadLock

Source code in niceml/utilities/readwritelock.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def __init__(  # noqa: PLR0913
    self,
    path_config: Union[LocationConfig, Dict[str, Any]],
    retry_time: float = 10,
    retry_await_time=0,
    timeout: float = 172800,
    write_lock_name: str = "write.lock",
    read_lock_name: str = "read.lock",
    is_acquired: bool = False,
):
    """Initialize ReadLock"""
    if write_lock_name == read_lock_name:
        raise ValueError("write_lock_name and read_lock_name must be different")
    super().__init__(retry_time, timeout, is_acquired)
    self.path_config = path_config
    self.write_lock_name = write_lock_name
    self.read_lock_name = read_lock_name
    self.retry_await_time = retry_await_time
Functions
__eq__ 
__eq__(other)

Compares two file locks

Source code in niceml/utilities/readwritelock.py 
231
232
233
234
235
236
237
238
239
240
241
242
def __eq__(self, other: "ReadLock"):
    """Compares two file locks"""
    return (
        isinstance(other, ReadLock)
        and self.retry_time == other.retry_time
        and self.timeout == other.timeout
        and self.is_acquired == other.is_acquired
        and self.path_config == other.path_config
        and self.write_lock_name == other.write_lock_name
        and self.read_lock_name == other.read_lock_name
        and self.retry_await_time == other.retry_await_time
    )
acquire 
acquire()

Acquire the lock

Source code in niceml/utilities/readwritelock.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def acquire(self):
    """Acquire the lock"""
    if self.is_acquired:
        return
    with open_location(self.path_config) as (cur_fs, root_path):
        start_time = time.monotonic()
        read_lock_path = join_fs_path(cur_fs, root_path, self.read_lock_name)
        while True:
            write_lock_path = join_fs_path(cur_fs, root_path, self.write_lock_name)
            cur_fs.mkdirs(root_path, exist_ok=True)
            if is_lock_file_acquirable(write_lock_path, cur_fs):
                break
            if time.monotonic() - start_time > self.timeout:
                raise TimeoutError(
                    f"Timeout while waiting for write lock release "
                    f"'{write_lock_path}' to acquire read lock at "
                    f"'{read_lock_path}'"
                )
            logging.info(
                f"Waiting for write lock release {self.retry_await_time} "
                f"seconds, to acquire read lock."
            )
            time.sleep(self.retry_time)
            self.retry_await_time += self.retry_time
        increase_lock_file_usage(read_lock_path, cur_fs)
        self.is_acquired = True
force_delete 
force_delete()

Force delete the lock

Source code in niceml/utilities/readwritelock.py 
216
217
218
219
220
221
def force_delete(self):
    """Force delete the lock"""
    self.is_acquired = False
    with open_location(self.path_config) as (cur_fs, root_path):
        read_lock_path = join_fs_path(cur_fs, root_path, self.read_lock_name)
        release_lock_file(read_lock_path, cur_fs)
is_acquirable 
is_acquirable()

Checks if lock file does not exist

Source code in niceml/utilities/readwritelock.py 
209
210
211
212
213
214
def is_acquirable(self) -> bool:
    """Checks if lock file does not exist"""
    with open_location(self.path_config) as (cur_fs, root_path):
        lock_path = join_fs_path(cur_fs, root_path, self.read_lock_name)
        lock_is_acquirable = is_lock_file_acquirable(lock_path, cur_fs)
    return lock_is_acquirable
release 
release()

Release the lock

Source code in niceml/utilities/readwritelock.py 
223
224
225
226
227
228
229
def release(self):
    """Release the lock"""
    if self.is_acquired:
        with open_location(self.path_config) as (cur_fs, root_path):
            read_lock_path = join_fs_path(cur_fs, root_path, self.read_lock_name)
            decrease_lock_file_usage(read_lock_path, cur_fs)
            self.is_acquired = False

WriteLock 

WriteLock(
    path_config,
    retry_time=10,
    retry_await_time=0,
    timeout=172800,
    write_lock_name="write.lock",
    read_lock_name="read.lock",
    is_acquired=False,
)

Bases: FileLock

Write lock for fsspec filesystems.

Initialize WriteLock

Source code in niceml/utilities/readwritelock.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(  # noqa: PLR0913
    self,
    path_config: Union[LocationConfig, Dict[str, Any]],
    retry_time: float = 10,
    retry_await_time: int = 0,
    timeout: float = 172800,
    write_lock_name: str = "write.lock",
    read_lock_name: str = "read.lock",
    is_acquired: bool = False,
):
    """Initialize WriteLock"""
    if write_lock_name == read_lock_name:
        raise ValueError("write_lock_name and read_lock_name must be different")
    super().__init__(retry_time, timeout, is_acquired)
    self.path_config = path_config
    self.write_lock_name = write_lock_name
    self.read_lock_name = read_lock_name
    self.retry_await_time = retry_await_time
Functions
__eq__ 
__eq__(other)

Compares two file locks

Source code in niceml/utilities/readwritelock.py 
146
147
148
149
150
151
152
153
154
155
156
157
def __eq__(self, other: "WriteLock"):
    """Compares two file locks"""
    return (
        isinstance(other, WriteLock)
        and self.retry_time == other.retry_time
        and self.timeout == other.timeout
        and self.is_acquired == other.is_acquired
        and self.path_config == other.path_config
        and self.write_lock_name == other.write_lock_name
        and self.read_lock_name == other.read_lock_name
        and self.retry_await_time == other.retry_await_time
    )
acquire 
acquire()

Acquire the lock

Source code in niceml/utilities/readwritelock.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def acquire(self):
    """Acquire the lock"""
    if self.is_acquired:
        return
    with open_location(self.path_config) as (cur_fs, root_path):
        start_time = time.monotonic()
        cur_fs.mkdirs(root_path, exist_ok=True)
        while True:  # Acquire Write Lock
            write_lock_path = join_fs_path(cur_fs, root_path, self.write_lock_name)
            if is_lock_file_acquirable(write_lock_path, cur_fs):
                acquire_lock_file(write_lock_path, cur_fs)
                self.is_acquired = True
                break
            if time.monotonic() - start_time > self.timeout:
                raise TimeoutError(
                    f"Timeout while waiting for write lock release at"
                    f" '{write_lock_path}'"
                )
            logging.info(
                f"Waiting for write lock release {self.retry_await_time} seconds"
            )
            time.sleep(self.retry_time)
            self.retry_await_time += self.retry_time
        self.retry_await_time = 0
        while True:  # Check if Read Lock is acquired
            read_lock_path = join_fs_path(cur_fs, root_path, self.read_lock_name)
            if is_lock_file_acquirable(read_lock_path, cur_fs):
                break
            if time.monotonic() - start_time > self.timeout:
                self.release()
                raise TimeoutError(
                    f"Timeout while waiting for read lock release at"
                    f" '{read_lock_path}'"
                )
            logging.info(
                f"Waiting for read lock release {self.retry_await_time} seconds"
            )
            time.sleep(self.retry_time)
            self.retry_await_time += self.retry_time
force_delete 
force_delete()

Force delete the lock

Source code in niceml/utilities/readwritelock.py 
131
132
133
134
135
136
def force_delete(self):
    """Force delete the lock"""
    self.is_acquired = False
    with open_location(self.path_config) as (cur_fs, root_path):
        write_lock_path = join_fs_path(cur_fs, root_path, self.write_lock_name)
        release_lock_file(write_lock_path, cur_fs)
is_acquirable 
is_acquirable()

Checks if lock file does not exist

Source code in niceml/utilities/readwritelock.py 
124
125
126
127
128
129
def is_acquirable(self) -> bool:
    """Checks if lock file does not exist"""
    with open_location(self.path_config) as (cur_fs, root_path):
        lock_path = join_fs_path(cur_fs, root_path, self.write_lock_name)
        lock_is_acquirable = is_lock_file_acquirable(lock_path, cur_fs)
    return lock_is_acquirable
release 
release()

Release the lock

Source code in niceml/utilities/readwritelock.py 
138
139
140
141
142
143
144
def release(self):
    """Release the lock"""
    if self.is_acquired:
        with open_location(self.path_config) as (cur_fs, root_path):
            write_lock_path = join_fs_path(cur_fs, root_path, self.write_lock_name)
            release_lock_file(write_lock_path, cur_fs)
            self.is_acquired = False

Functions

acquire_lock_file 

acquire_lock_file(lock_file_path, file_system=None)

Acquire the lock file.

Source code in niceml/utilities/readwritelock.py 
256
257
258
259
260
261
262
263
def acquire_lock_file(
    lock_file_path: str, file_system: Optional[AbstractFileSystem] = None
) -> None:
    """Acquire the lock file."""
    file_system = file_system or LocalFileSystem()
    if not is_lock_file_acquirable(lock_file_path, file_system):
        raise RuntimeError(f"Lock file {lock_file_path} is not available.")
    file_system.touch(lock_file_path)

decrease_lock_file_usage 

decrease_lock_file_usage(lock_file_path, file_system=None)

Decrease the lock file usage.

Source code in niceml/utilities/readwritelock.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def decrease_lock_file_usage(
    lock_file_path: str, file_system: Optional[AbstractFileSystem] = None
) -> None:
    """Decrease the lock file usage."""
    file_system = file_system or LocalFileSystem()
    if not file_system.exists(lock_file_path):
        current_usage: int = 0
    else:
        with file_system.open(lock_file_path, "r") as file:
            current_usage = int(file.read())
    if current_usage > 1:
        with file_system.open(lock_file_path, "w") as file:
            file.write(str(current_usage - 1))
    else:
        release_lock_file(lock_file_path, file_system)

increase_lock_file_usage 

increase_lock_file_usage(lock_file_path, file_system=None)

Increase the lock file usage.

Source code in niceml/utilities/readwritelock.py 
278
279
280
281
282
283
284
285
286
287
288
289
def increase_lock_file_usage(
    lock_file_path: str, file_system: Optional[AbstractFileSystem] = None
) -> None:
    """Increase the lock file usage."""
    file_system = file_system or LocalFileSystem()
    if not file_system.exists(lock_file_path):
        current_usage: int = 0
    else:
        with file_system.open(lock_file_path, "r") as file:
            current_usage = int(file.read())
    with file_system.open(lock_file_path, "w") as file:
        file.write(str(current_usage + 1))

is_lock_file_acquirable 

is_lock_file_acquirable(lock_file_path, file_system=None)

Check if the lock file is available. The lock file is available if it does not exist at lock_file_path.

Source code in niceml/utilities/readwritelock.py 
245
246
247
248
249
250
251
252
253
def is_lock_file_acquirable(
    lock_file_path: str, file_system: Optional[AbstractFileSystem] = None
) -> bool:
    """Check if the lock file is available.
    The lock file is available if it does not exist at lock_file_path."""
    file_system = file_system or LocalFileSystem()
    if file_system.exists(lock_file_path):
        return False
    return True

release_lock_file 

release_lock_file(lock_file_path, file_system=None)

Release the lock file.

Source code in niceml/utilities/readwritelock.py 
266
267
268
269
270
271
272
273
274
275
def release_lock_file(
    lock_file_path: str, file_system: Optional[AbstractFileSystem] = None
) -> None:
    """Release the lock file."""
    file_system = file_system or LocalFileSystem()
    if not file_system.exists(lock_file_path):
        logger = logging.getLogger(__name__)
        logger.warning(f"Lock file {lock_file_path} does not exist.")
    else:
        file_system.rm(lock_file_path)