Exclusively create lockfile. Throws when failed
(path)
| 1296 | pass |
| 1297 | |
| 1298 | def create_lockfile(path): |
| 1299 | """Exclusively create lockfile. Throws when failed""" |
| 1300 | mypid = os.getpid() |
| 1301 | lockfile = path.join(".lock") |
| 1302 | if hasattr(lockfile, "mksymlinkto"): |
| 1303 | lockfile.mksymlinkto(str(mypid)) |
| 1304 | else: |
| 1305 | fd = error.checked_call( |
| 1306 | os.open, str(lockfile), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644 |
| 1307 | ) |
| 1308 | with os.fdopen(fd, "w") as f: |
| 1309 | f.write(str(mypid)) |
| 1310 | return lockfile |
| 1311 | |
| 1312 | def atexit_remove_lockfile(lockfile): |
| 1313 | """Ensure lockfile is removed at process exit""" |
nothing calls this directly
no test coverage detected