Created
September 11, 2018 04:49
-
-
Save wecsam/9129344b87546fb316e02f2cfb2e2fbb to your computer and use it in GitHub Desktop.
Prevents concurrent instances of a Python script
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
''' | |
Import this module to ensure that at most one instance of the main script can | |
be running at once. If another instance is detected, sys.exit() will be called. | |
The psutil package is required: pip install psutil | |
''' | |
import atexit, logging, math, os, psutil, socket, sys | |
logger = logging.getLogger("prevent_parallel") | |
PID_FILE = \ | |
os.path.abspath(sys.argv[0]) + "-PID-" + socket.gethostname() + ".bin" | |
if __name__ == "__main__": | |
logging.basicConfig(level=logging.DEBUG) | |
logger.debug("PID file: %s", PID_FILE) | |
try: | |
_pid_file = open(PID_FILE, "r+b") | |
except FileNotFoundError: | |
_pid_file = open(PID_FILE, "wb") | |
_read = False | |
else: | |
_read = True | |
with _pid_file: | |
# Only check for the PID of another instance if there was a PID. | |
if _read: | |
# Get the PID of the last instance of this script that was not | |
# terminated because another instance before that was already running. | |
pid = int.from_bytes(_pid_file.read(), "little") | |
logger.debug("Checking for PID %d.", pid) | |
# Check whether that instance is still running right now. | |
try: | |
process = psutil.Process(pid) | |
except psutil.NoSuchProcess: | |
pass | |
else: | |
if process.name().lower().startswith("python"): | |
# The main script is already running in another instance. | |
# Terminate this instance this instant. | |
logger.info( | |
"The script is already running. This instance will now " | |
"exit." | |
) | |
sys.exit() | |
logger.debug("Another instance of this script was not found.") | |
else: | |
logger.debug("A file for the PID of this script was not found.") | |
# This instance will be allowed to continue. Save its PID. | |
logger.debug("This instance will be allowed to continue.") | |
pid = os.getpid() | |
_pid_file.seek(0) | |
_pid_file.truncate() | |
_pid_file.write(pid.to_bytes(math.ceil(pid.bit_length() / 8.0), "little")) | |
atexit.register(lambda: os.remove(PID_FILE)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment