Uses ``LockFileEx`` to hard lock a byte range of the lock file on Windows systems. Lock file cleanup: Windows attempts to delete the lock file after release, but deletion is not guaranteed in multi-threaded scenarios where another thread holds an open handle. The lock
| 173 | raise ctypes.WinError(ctypes.get_last_error()) |
| 174 | |
| 175 | class WindowsFileLock(BaseFileLock): |
| 176 | """ |
| 177 | Uses ``LockFileEx`` to hard lock a byte range of the lock file on Windows systems. |
| 178 | |
| 179 | Lock file cleanup: Windows attempts to delete the lock file after release, but deletion is |
| 180 | not guaranteed in multi-threaded scenarios where another thread holds an open handle. The lock |
| 181 | file may persist on disk, which does not affect lock correctness. |
| 182 | """ |
| 183 | |
| 184 | def _acquire(self) -> None: |
| 185 | raise_on_not_writable_file(self.lock_file) |
| 186 | ensure_directory_exists(self.lock_file) |
| 187 | |
| 188 | # The reparse test is bound to the opened handle, so a symlink or junction swapped in cannot defeat it |
| 189 | # through a check-then-open TOCTOU race. |
| 190 | fd = _open_non_reparse_fd(self.lock_file, self._open_mode()) |
| 191 | if fd is None: |
| 192 | return # open contention (share conflict or a name pending deletion); let the retry loop try again |
| 193 | try: |
| 194 | locked = _lock_fd_nonblocking(fd) |
| 195 | if locked: |
| 196 | self._mark_descriptor_owned(fd) |
| 197 | except BaseException: # pragma: no cover # cleanup only if the lock attempt itself raises |
| 198 | os.close(fd) |
| 199 | raise |
| 200 | if not locked: |
| 201 | os.close(fd) # another holder owns the byte-range lock; let the retry loop try again |
| 202 | |
| 203 | def _release(self) -> None: |
| 204 | fd = cast("int", self._context.lock_file_fd) |
| 205 | # Retain the descriptor until the OS unlock succeeds: if UnlockFileEx raises, the byte-range lock is still |
| 206 | # held, so is_locked must keep reporting held rather than losing the fd. Only after the unlock commits do |
| 207 | # close and unlink run as post-unlock cleanup; their failure cannot make the lock held again. |
| 208 | _unlock_fd(fd) |
| 209 | self._mark_descriptor_released() |
| 210 | self._close_released_fd(fd, default_suppresses=False) |
| 211 | if not self._preserve_lock_file: # preserve_lock_file keeps a stable file identity for the caller (#605) |
| 212 | with suppress(OSError): |
| 213 | Path(self.lock_file).unlink() |
| 214 | |
| 215 | def _open_non_reparse_fd(path: str, mode: int) -> int | None: |
| 216 | """ |
nothing calls this directly
no outgoing calls
no test coverage detected