Register a cleanup function for removing a lock, by default on atexit.
(lock_path: Path, register=atexit.register)
| 245 | |
| 246 | |
| 247 | def register_cleanup_lock_removal(lock_path: Path, register=atexit.register): |
| 248 | """Register a cleanup function for removing a lock, by default on atexit.""" |
| 249 | pid = os.getpid() |
| 250 | |
| 251 | def cleanup_on_exit(lock_path: Path = lock_path, original_pid: int = pid) -> None: |
| 252 | current_pid = os.getpid() |
| 253 | if current_pid != original_pid: |
| 254 | # fork |
| 255 | return |
| 256 | try: |
| 257 | lock_path.unlink() |
| 258 | except OSError: |
| 259 | pass |
| 260 | |
| 261 | return register(cleanup_on_exit) |
| 262 | |
| 263 | |
| 264 | def maybe_delete_a_numbered_dir(path: Path) -> None: |
no outgoing calls