Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_read_write.py: 26%

1from __future__ import annotations 

2 

3import atexit 

4import logging 

5import os 

6import pathlib 

7import sqlite3 

8import threading 

9import time 

10from contextlib import contextmanager, suppress 

11from typing import TYPE_CHECKING, Literal 

12from weakref import WeakValueDictionary 

13 

14from ._api import AcquireReturnProxy 

15from ._error import Timeout 

16 

17if TYPE_CHECKING: 

18 from collections.abc import Generator 

19 

20_LOGGER = logging.getLogger("filelock") 

21 

22_all_connections: set[sqlite3.Connection] = set() 

23_all_connections_lock = threading.Lock() 

24 

25 

26def _cleanup_connections() -> None: 

27 with _all_connections_lock: 

28 for con in list(_all_connections): 

29 with suppress(Exception): 

30 con.close() 

31 _all_connections.clear() 

32 

33 

34atexit.register(_cleanup_connections) 

35 

36# sqlite3_busy_timeout() accepts a C int, max 2_147_483_647 on 32-bit. Use a lower value to be safe (~23 days). 

37_MAX_SQLITE_TIMEOUT_MS = 2_000_000_000 - 1 

38 

39 

40def timeout_for_sqlite(timeout: float, *, blocking: bool, already_waited: float) -> int: 

41 if blocking is False: 

42 return 0 

43 

44 if timeout == -1: 

45 return _MAX_SQLITE_TIMEOUT_MS 

46 

47 if timeout < 0: 

48 msg = "timeout must be a non-negative number or -1" 

49 raise ValueError(msg) 

50 

51 remaining = max(timeout - already_waited, 0) if timeout > 0 else timeout 

52 timeout_ms = int(remaining * 1000) 

53 if timeout_ms > _MAX_SQLITE_TIMEOUT_MS or timeout_ms < 0: 

54 _LOGGER.warning("timeout %s is too large for SQLite, using %s ms instead", timeout, _MAX_SQLITE_TIMEOUT_MS) 

55 return _MAX_SQLITE_TIMEOUT_MS 

56 return timeout_ms 

57 

58 

59class _ReadWriteLockMeta(type): 

60 """ 

61 Metaclass that handles singleton resolution when is_singleton=True. 

62 

63 Singleton logic lives here rather than in ReadWriteLock.get_lock so that ``ReadWriteLock(path)`` transparently 

64 returns cached instances without a 2-arg ``super()`` call that type checkers cannot verify. 

65 

66 """ 

67 

68 _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock] 

69 _instances_lock: threading.Lock 

70 

71 def __call__( 

72 cls, 

73 lock_file: str | os.PathLike[str], 

74 timeout: float = -1, 

75 *, 

76 blocking: bool = True, 

77 is_singleton: bool = True, 

78 ) -> ReadWriteLock: 

79 if not is_singleton: 

80 return super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton) 

81 

82 normalized = pathlib.Path(lock_file).resolve() 

83 with cls._instances_lock: 

84 if normalized not in cls._instances: 

85 instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton) 

86 cls._instances[normalized] = instance 

87 else: 

88 instance = cls._instances[normalized] 

89 

90 if instance.timeout != timeout or instance.blocking != blocking: 

91 msg = ( 

92 f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking}," 

93 f" cannot be changed to timeout={timeout}, blocking={blocking}" 

94 ) 

95 raise ValueError(msg) 

96 return instance 

97 

98 

99class ReadWriteLock(metaclass=_ReadWriteLockMeta): 

100 """ 

101 Cross-process read-write lock backed by SQLite. 

102 

103 Allows concurrent shared readers or a single exclusive writer. The lock is reentrant within the same mode (multiple 

104 ``acquire_read`` calls nest, as do multiple ``acquire_write`` calls from the same thread), but upgrading from read 

105 to write or downgrading from write to read raises :class:`RuntimeError`. Write locks are pinned to the thread that 

106 acquired them. 

107 

108 By default, ``is_singleton=True``: calling ``ReadWriteLock(path)`` with the same resolved path returns the same 

109 instance. The lock file must use a ``.db`` extension (SQLite database). 

110 

111 :param lock_file: path to the SQLite database file used as the lock 

112 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely 

113 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable 

114 :param is_singleton: if ``True``, reuse existing instances for the same resolved path 

115 

116 .. versionadded:: 3.21.0 

117 

118 """ 

119 

120 _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock] = WeakValueDictionary() 

121 _instances_lock = threading.Lock() 

122 

123 @classmethod 

124 def get_lock( 

125 cls, lock_file: str | os.PathLike[str], timeout: float = -1, *, blocking: bool = True 

126 ) -> ReadWriteLock: 

127 """ 

128 Return the singleton :class:`ReadWriteLock` for *lock_file*. 

129 

130 :param lock_file: path to the SQLite database file used as the lock 

131 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely 

132 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable 

133 

134 :returns: the singleton lock instance 

135 

136 :raises ValueError: if an instance already exists for this path with different *timeout* or *blocking* values 

137 

138 """ 

139 return cls(lock_file, timeout, blocking=blocking) 

140 

141 def __init__( 

142 self, 

143 lock_file: str | os.PathLike[str], 

144 timeout: float = -1, 

145 *, 

146 blocking: bool = True, 

147 is_singleton: bool = True, # noqa: ARG002 # consumed by _ReadWriteLockMeta.__call__ 

148 ) -> None: 

149 self.lock_file = os.fspath(lock_file) 

150 self.timeout = timeout 

151 self.blocking = blocking 

152 self._transaction_lock = threading.Lock() # serializes the (possibly blocking) SQLite transaction work 

153 self._internal_lock = threading.Lock() # protects _lock_level / _current_mode updates and rollback 

154 self._lock_level = 0 

155 self._current_mode: Literal["read", "write"] | None = None 

156 self._write_thread_id: int | None = None 

157 self._con = sqlite3.connect(self.lock_file, check_same_thread=False) 

158 with _all_connections_lock: 

159 _all_connections.add(self._con) 

160 

161 def _acquire_transaction_lock(self, *, blocking: bool, timeout: float) -> None: 

162 if not blocking: 

163 acquired = self._transaction_lock.acquire(blocking=False) 

164 elif timeout == -1: 

165 acquired = self._transaction_lock.acquire(blocking=True) 

166 else: 

167 acquired = self._transaction_lock.acquire(blocking=True, timeout=timeout) 

168 if not acquired: 

169 raise Timeout(self.lock_file) from None 

170 

171 def _validate_reentrant(self, mode: Literal["read", "write"]) -> AcquireReturnProxy: 

172 opposite = "write" if mode == "read" else "read" 

173 direction = "downgrade" if mode == "read" else "upgrade" 

174 if self._current_mode != mode: 

175 msg = ( 

176 f"Cannot acquire {mode} lock on {self.lock_file} (lock id: {id(self)}): " 

177 f"already holding a {opposite} lock ({direction} not allowed)" 

178 ) 

179 raise RuntimeError(msg) 

180 if mode == "write" and (cur := threading.get_ident()) != self._write_thread_id: 

181 msg = ( 

182 f"Cannot acquire write lock on {self.lock_file} (lock id: {id(self)}) " 

183 f"from thread {cur} while it is held by thread {self._write_thread_id}" 

184 ) 

185 raise RuntimeError(msg) 

186 self._lock_level += 1 

187 return AcquireReturnProxy(lock=self) 

188 

189 def _configure_and_begin( 

190 self, mode: Literal["read", "write"], timeout: float, *, blocking: bool, start_time: float 

191 ) -> None: 

192 waited = time.perf_counter() - start_time 

193 timeout_ms = timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited) 

194 self._con.execute(f"PRAGMA busy_timeout={timeout_ms};").close() 

195 # Use legacy journal mode (not WAL) because WAL does not block readers when a concurrent EXCLUSIVE 

196 # write transaction is active, making read-write locking impossible without modifying table data. 

197 # MEMORY is safe here since no actual writes happen — crashes cannot corrupt the DB. 

198 # See https://sqlite.org/lang_transaction.html#deferred_immediate_and_exclusive_transactions 

199 # 

200 # Set here (not in __init__) because this pragma itself may block on a locked database, 

201 # so it must run after busy_timeout is configured above. 

202 self._con.execute("PRAGMA journal_mode=MEMORY;").close() 

203 # Recompute remaining timeout after the potentially blocking journal_mode pragma. 

204 waited = time.perf_counter() - start_time 

205 if (recomputed := timeout_for_sqlite(timeout, blocking=blocking, already_waited=waited)) != timeout_ms: 

206 self._con.execute(f"PRAGMA busy_timeout={recomputed};").close() 

207 stmt = "BEGIN EXCLUSIVE TRANSACTION;" if mode == "write" else "BEGIN TRANSACTION;" 

208 self._con.execute(stmt).close() 

209 if mode == "read": 

210 # A SELECT is needed to force SQLite to actually acquire the SHARED lock on the database. 

211 # https://www.sqlite.org/lockingv3.html#transaction_control 

212 self._con.execute("SELECT name FROM sqlite_schema LIMIT 1;").close() 

213 

214 def _acquire(self, mode: Literal["read", "write"], timeout: float, *, blocking: bool) -> AcquireReturnProxy: 

215 with self._internal_lock: 

216 if self._lock_level > 0: 

217 return self._validate_reentrant(mode) 

218 

219 start_time = time.perf_counter() 

220 self._acquire_transaction_lock(blocking=blocking, timeout=timeout) 

221 try: 

222 return self._do_acquire_inner(mode, timeout, blocking=blocking, start_time=start_time) 

223 except sqlite3.Error as exc: 

224 # A read acquire runs BEGIN (a deferred transaction that takes no database lock) and only then the 

225 # SELECT that actually takes the SHARED lock. If a writer grabs the EXCLUSIVE lock between the two, 

226 # the SELECT fails but BEGIN's transaction is left open on the shared connection. Roll it back here, 

227 # while we still hold _transaction_lock, otherwise the next acquire's BEGIN dies with "cannot start a 

228 # transaction within a transaction" and the instance is wedged for good. Catch only sqlite3.Error: a 

229 # reentrant-validation RuntimeError from _do_acquire_inner's double-check must not roll back, or it 

230 # would drop a transaction another thread legitimately holds on the shared connection. 

231 with suppress(sqlite3.Error): 

232 self._con.rollback() 

233 if isinstance(exc, sqlite3.OperationalError) and "database is locked" in str(exc): 

234 raise Timeout(self.lock_file) from None 

235 raise 

236 finally: 

237 self._transaction_lock.release() 

238 

239 def _do_acquire_inner( 

240 self, 

241 mode: Literal["read", "write"], 

242 timeout: float, 

243 *, 

244 blocking: bool, 

245 start_time: float, 

246 ) -> AcquireReturnProxy: 

247 # Double-check: another thread may have acquired the lock while we waited on _transaction_lock. 

248 with self._internal_lock: 

249 if self._lock_level > 0: 

250 return self._validate_reentrant(mode) 

251 self._configure_and_begin(mode, timeout, blocking=blocking, start_time=start_time) 

252 with self._internal_lock: 

253 self._current_mode = mode 

254 self._lock_level = 1 

255 if mode == "write": 

256 self._write_thread_id = threading.get_ident() 

257 return AcquireReturnProxy(lock=self) 

258 

259 def acquire_read(self, timeout: float = -1, *, blocking: bool = True) -> AcquireReturnProxy: 

260 """ 

261 Acquire a shared read lock. 

262 

263 If this instance already holds a read lock, the lock level is incremented (reentrant). Attempting to acquire a 

264 read lock while holding a write lock raises :class:`RuntimeError` (downgrade not allowed). 

265 

266 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely 

267 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable 

268 

269 :returns: a proxy that can be used as a context manager to release the lock 

270 

271 :raises RuntimeError: if a write lock is already held on this instance 

272 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

273 

274 """ 

275 return self._acquire("read", timeout, blocking=blocking) 

276 

277 def acquire_write(self, timeout: float = -1, *, blocking: bool = True) -> AcquireReturnProxy: 

278 """ 

279 Acquire an exclusive write lock. 

280 

281 If this instance already holds a write lock from the same thread, the lock level is incremented (reentrant). 

282 Attempting to acquire a write lock while holding a read lock raises :class:`RuntimeError` (upgrade not allowed). 

283 Write locks are pinned to the acquiring thread: a different thread trying to re-enter also raises 

284 :class:`RuntimeError`. 

285 

286 :param timeout: maximum wait time in seconds; ``-1`` means block indefinitely 

287 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately when the lock is unavailable 

288 

289 :returns: a proxy that can be used as a context manager to release the lock 

290 

291 :raises RuntimeError: if a read lock is already held, or a write lock is held by a different thread 

292 :raises Timeout: if the lock cannot be acquired within *timeout* seconds 

293 

294 """ 

295 return self._acquire("write", timeout, blocking=blocking) 

296 

297 def release(self, *, force: bool = False) -> None: 

298 """ 

299 Release one level of the current lock. 

300 

301 When the lock level reaches zero the underlying SQLite transaction is rolled back, releasing the database lock. 

302 

303 :param force: if ``True``, release the lock completely regardless of the current lock level 

304 

305 :raises RuntimeError: if no lock is currently held and *force* is ``False`` 

306 

307 """ 

308 should_rollback = False 

309 with self._internal_lock: 

310 if self._lock_level == 0: 

311 if force: 

312 return 

313 msg = f"Cannot release a lock on {self.lock_file} (lock id: {id(self)}) that is not held" 

314 raise RuntimeError(msg) 

315 if force: 

316 self._lock_level = 0 

317 else: 

318 self._lock_level -= 1 

319 if self._lock_level == 0: 

320 self._current_mode = None 

321 self._write_thread_id = None 

322 should_rollback = True 

323 if should_rollback: 

324 # The rollback ends the transaction on the shared connection, so it has to be serialized against 

325 # acquire()'s BEGIN the same way acquire() already serializes itself with _transaction_lock. Without 

326 # this, another thread that sees lock_level back at 0 can start its BEGIN while this rollback's 

327 # transaction is still open (raising "cannot start a transaction within a transaction") or, in the 

328 # other ordering, have its freshly started transaction rolled back here, dropping the database lock 

329 # while it still believes it holds it. 

330 with self._transaction_lock: 

331 self._con.rollback() 

332 

333 @contextmanager 

334 def read_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]: 

335 """ 

336 Context manager that acquires and releases a shared read lock. 

337 

338 Falls back to instance defaults for *timeout* and *blocking* when ``None``. 

339 

340 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default 

341 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default 

342 

343 """ 

344 if timeout is None: 

345 timeout = self.timeout 

346 if blocking is None: 

347 blocking = self.blocking 

348 self.acquire_read(timeout, blocking=blocking) 

349 try: 

350 yield 

351 finally: 

352 self.release() 

353 

354 @contextmanager 

355 def write_lock(self, timeout: float | None = None, *, blocking: bool | None = None) -> Generator[None]: 

356 """ 

357 Context manager that acquires and releases an exclusive write lock. 

358 

359 Falls back to instance defaults for *timeout* and *blocking* when ``None``. 

360 

361 :param timeout: maximum wait time in seconds, or ``None`` to use the instance default 

362 :param blocking: if ``False``, raise :class:`~filelock.Timeout` immediately; ``None`` uses the instance default 

363 

364 """ 

365 if timeout is None: 

366 timeout = self.timeout 

367 if blocking is None: 

368 blocking = self.blocking 

369 self.acquire_write(timeout, blocking=blocking) 

370 try: 

371 yield 

372 finally: 

373 self.release() 

374 

375 def close(self) -> None: 

376 """ 

377 Release the lock (if held) and close the underlying SQLite connection. 

378 

379 After calling this method, the lock instance is no longer usable. 

380 

381 """ 

382 self.release(force=True) 

383 self._con.close() 

384 with _all_connections_lock: 

385 _all_connections.discard(self._con)