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

1from __future__ import annotations 

2 

3import os 

4import socket 

5import sys 

6import time 

7from contextlib import suppress 

8from errno import EACCES, EEXIST, EPERM, ESRCH 

9from pathlib import Path 

10 

11from ._api import BaseFileLock 

12from ._util import ensure_directory_exists, raise_on_not_writable_file 

13 

14_WIN_SYNCHRONIZE = 0x100000 

15_WIN_ERROR_INVALID_PARAMETER = 87 

16 

17 

18class SoftFileLock(BaseFileLock): 

19 """ 

20 Portable file lock based on file existence. 

21 

22 Unlike :class:`UnixFileLock <filelock.UnixFileLock>` and :class:`WindowsFileLock <filelock.WindowsFileLock>`, this 

23 lock does not use OS-level locking primitives. Instead, it creates the lock file with ``O_CREAT | O_EXCL`` and 

24 treats its existence as the lock indicator. This makes it work on any filesystem but leaves stale lock files behind 

25 if the process crashes without releasing the lock. 

26 

27 To mitigate stale locks, the lock file contains the PID and hostname of the holding process. On contention, if the 

28 holder is on the same host and its PID no longer exists, the stale lock is broken automatically. 

29 

30 """ 

31 

32 def _acquire(self) -> None: 

33 raise_on_not_writable_file(self.lock_file) 

34 ensure_directory_exists(self.lock_file) 

35 flags = ( 

36 os.O_WRONLY # open for writing only 

37 | os.O_CREAT 

38 | os.O_EXCL # together with above raise EEXIST if the file specified by filename exists 

39 | os.O_TRUNC # truncate the file to zero byte 

40 ) 

41 if (o_nofollow := getattr(os, "O_NOFOLLOW", None)) is not None: 

42 flags |= o_nofollow 

43 try: 

44 file_handler = os.open(self.lock_file, flags, self._open_mode()) 

45 except OSError as exception: 

46 if not ( 

47 exception.errno == EEXIST or (exception.errno == EACCES and sys.platform == "win32") 

48 ): # pragma: win32 no cover 

49 raise 

50 if exception.errno == EEXIST and sys.platform != "win32": # pragma: win32 no cover 

51 self._try_break_stale_lock() 

52 else: 

53 self._write_lock_info(file_handler) 

54 self._context.lock_file_fd = file_handler 

55 

56 def _try_break_stale_lock(self) -> None: 

57 with suppress(OSError): 

58 content = Path(self.lock_file).read_text(encoding="utf-8") 

59 lines = content.strip().splitlines() 

60 if len(lines) != 2: # noqa: PLR2004 

61 return 

62 pid_str, hostname = lines 

63 if hostname != socket.gethostname(): 

64 return 

65 pid = int(pid_str) 

66 if self._is_process_alive(pid): 

67 return 

68 break_path = f"{self.lock_file}.break.{os.getpid()}" 

69 Path(self.lock_file).rename(break_path) 

70 Path(break_path).unlink() 

71 

72 @staticmethod 

73 def _is_process_alive(pid: int) -> bool: 

74 if sys.platform == "win32": # pragma: win32 cover 

75 import ctypes # noqa: PLC0415 

76 

77 kernel32 = ctypes.windll.kernel32 

78 handle = kernel32.OpenProcess(_WIN_SYNCHRONIZE, 0, pid) 

79 if handle: 

80 kernel32.CloseHandle(handle) 

81 return True 

82 return kernel32.GetLastError() != _WIN_ERROR_INVALID_PARAMETER 

83 try: 

84 os.kill(pid, 0) 

85 except OSError as exc: 

86 if exc.errno == ESRCH: 

87 return False 

88 if exc.errno == EPERM: 

89 return True 

90 raise 

91 return True 

92 

93 @staticmethod 

94 def _write_lock_info(fd: int) -> None: 

95 with suppress(OSError): 

96 os.write(fd, f"{os.getpid()}\n{socket.gethostname()}\n".encode()) 

97 

98 def _release(self) -> None: 

99 assert self._context.lock_file_fd is not None # noqa: S101 

100 os.close(self._context.lock_file_fd) 

101 self._context.lock_file_fd = None 

102 if sys.platform == "win32": 

103 self._windows_unlink_with_retry() 

104 else: 

105 with suppress(OSError): 

106 Path(self.lock_file).unlink() 

107 

108 def _windows_unlink_with_retry(self) -> None: 

109 max_retries = 10 

110 retry_delay = 0.001 

111 for attempt in range(max_retries): 

112 # Windows doesn't immediately release file handles after close, causing EACCES/EPERM on unlink 

113 try: 

114 Path(self.lock_file).unlink() 

115 except OSError as exc: # noqa: PERF203 

116 if exc.errno not in {EACCES, EPERM}: 

117 return 

118 if attempt < max_retries - 1: 

119 time.sleep(retry_delay) 

120 retry_delay *= 2 

121 else: 

122 return 

123 

124 

125__all__ = [ 

126 "SoftFileLock", 

127]