Skip to content

Instantly share code, notes, and snippets.

@wecsam
Created September 11, 2018 04:49
Show Gist options
  • Save wecsam/9129344b87546fb316e02f2cfb2e2fbb to your computer and use it in GitHub Desktop.
Save wecsam/9129344b87546fb316e02f2cfb2e2fbb to your computer and use it in GitHub Desktop.
Prevents concurrent instances of a Python script
#!/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