Skip to content

Commit

Permalink
add is_running, lock_pid_file
Browse files Browse the repository at this point in the history
  • Loading branch information
lidong committed Jun 13, 2024
1 parent e387832 commit 655a391
Show file tree
Hide file tree
Showing 5 changed files with 141 additions and 0 deletions.
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
### v0.0.6
1. add `xor_encode_decode`
2. add `is_running`, `lock_pid_file`
3.
6 changes: 6 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@ print(morebuiltins.__file__)

## Doc

[Changelog](https://github.com/ClericPy/morebuiltins/blob/master/doc.md) comes after v0.0.6(2024-06-14)

---

Module Docs - https://github.com/ClericPy/morebuiltins/blob/master/doc.md
Expand Down Expand Up @@ -79,6 +81,10 @@ Module Docs - https://github.com/ClericPy/morebuiltins/blob/master/doc.md

1.20 `xor_encode_decode` -

1.21 `is_running` - Check if the given process ID is running.

1.22 `set_pid_file` - Attempt to lock a PID file to ensure only one instance is running, like a singleton-lock.


## 2. morebuiltins.functools

Expand Down
45 changes: 45 additions & 0 deletions doc.md
Original file line number Diff line number Diff line change
Expand Up @@ -424,6 +424,51 @@
True


---


1.21 `is_running` - Check if the given process ID is running.

Parameters:
pid -- The process ID to check.

Returns:
True if the process is running; False otherwise.

Examples:
>>> is_running(os.getpid() * 10) # Assume process is not running
False
>>> is_running(os.getpid()) # Check if the current process is running
True
>>> is_running("not_a_pid") # Invalid PID input should be handled and return False
False



---


1.22 `set_pid_file` - Attempt to lock a PID file to ensure only one instance is running, like a singleton-lock.

Args:
- path (Union[str, Path]): The path to the PID file.
- raise_error (bool): If True, raise an exception when the PID file exists and
corresponds to a running process. Defaults to True.

Returns:
- bool: True if the PID file is successfully locked, otherwise, based on
`raise_error`, either raises an exception or returns False indicating
the lock could not be acquired.

Examples:

>>> set_pid_file("myapp.pid") # Assuming this is the first run, should succeed
True
>>> set_pid_file("myapp.pid") # Simulating second instance trying to start, should raise error if raise_error=True
False
>>> os.unlink("myapp.pid")


---

======================
Expand Down
85 changes: 85 additions & 0 deletions morebuiltins/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,16 @@
import gzip
import hashlib
import json
import os
import re
import sys
import traceback
from collections import UserDict
from enum import IntEnum
from functools import wraps
from itertools import groupby, islice
from os.path import basename
from pathlib import Path
from time import gmtime, mktime, strftime, strptime, time, timezone
from typing import (
Any,
Expand Down Expand Up @@ -46,6 +49,8 @@
"Trie",
"GuessExt",
"xor_encode_decode",
"is_running",
"set_pid_file",
]


Expand Down Expand Up @@ -837,6 +842,86 @@ def xor_encode_decode(data, key):
return bytes([b ^ k for b, k in zip(data, extended_key)])


def is_running(pid):
"""Check if the given process ID is running.
Parameters:
pid -- The process ID to check.
Returns:
True if the process is running; False otherwise.
Examples:
>>> is_running(os.getpid() * 10) # Assume process is not running
False
>>> is_running(os.getpid()) # Check if the current process is running
True
>>> is_running("not_a_pid") # Invalid PID input should be handled and return False
False
"""
try:
pid = int(pid)
except ValueError:
return False
if sys.platform == "win32":
with os.popen('tasklist /fo csv /fi "pid eq %s"' % int(pid)) as f:
f.readline()
text = f.readline()
return bool(text)
else:
try:
os.kill(int(pid), 0)
return True
except OSError:
return False
except SystemError:
# maybe windows?
return True


def set_pid_file(path: Union[str, Path], raise_error=False):
"""Attempt to lock a PID file to ensure only one instance is running, like a singleton-lock.
Args:
- path (Union[str, Path]): The path to the PID file.
- raise_error (bool): If True, raise an exception when the PID file exists and
corresponds to a running process. Defaults to True.
Returns:
- bool: True if the PID file is successfully locked, otherwise, based on
`raise_error`, either raises an exception or returns False indicating
the lock could not be acquired.
Examples:
>>> set_pid_file("myapp.pid") # Assuming this is the first run, should succeed
True
>>> set_pid_file("myapp.pid") # Simulating second instance trying to start, should raise error if raise_error=True
False
>>> os.unlink("myapp.pid")
"""
if isinstance(path, str):
path = Path(path)
running = False
if path.is_file():
try:
old_pid = int(path.read_text().strip())
running = is_running(old_pid)
except ValueError:
# non-int pid, NaN
pass

if running:
if raise_error:
raise RuntimeError(f"{path.as_posix()} is locked by {old_pid}")
else:
return False
else:
path.write_text(f"{os.getpid()}")
return True


if __name__ == "__main__":
__name__ = "morebuiltins.utils"
import doctest
Expand Down
1 change: 1 addition & 0 deletions test.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ def test_all():
module.__name__,
flush=True,
)
time.sleep(1)
print("all test ok", flush=True)


Expand Down

0 comments on commit 655a391

Please sign in to comment.