Created
February 22, 2022 18:49
-
-
Save elibroftw/68f7998b3edfc580c09161665057ca7f to your computer and use it in GitHub Desktop.
Python Single Instance Alternative?
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
## HELPERS AND CONSTANTS | |
def is_already_running(look_for='Music Caster', threshold=1): | |
raise NotImplementedError('https://gist.github.com/elibroftw/44844582287ce7386d1d9d5060e52972') | |
MUTEX_NAME = 'MusicCasterMutex{{FBE8A652-58D6-482D-B6A9-B3D7931CC9C5}' | |
PID_FILENAME = 'music_caster.pid' | |
LOCK_FILENAME = 'music_caster.lock' | |
def create_pid_file(port=None): | |
with open(PID_FILENAME, 'w', encoding='utf-8') as f: | |
f.write(str(os.getpid())) | |
if port is not None: | |
f.write(f'\n{port}') | |
def parse_pid_file() -> (int, int): | |
with suppress(FileNotFoundError): | |
with open(PID_FILENAME, encoding='utf-8') as f: | |
pid = int(f.readline().strip()) | |
try: | |
port = int(f.readline().strip()) | |
except ValueError: | |
port = 2001 | |
return pid, port | |
return None, 2001 | |
######################################################################################################### | |
# using portalocker | |
# this works if program instance are started simultaneously | |
def ensure_single_instance(is_debug=False): | |
import portalocker | |
from portalocker.exceptions import LockException | |
file = open(LOCK_FILENAME, 'w+', encoding='utf-8') | |
# no old running instances found, try locking file | |
try: | |
# exclusively locked | |
portalocker.lock(file, portalocker.LockFlags.NON_BLOCKING) | |
create_pid_file() | |
if is_debug: | |
print(f'Locked {LOCK_FILENAME} pid = {os.getpid()}') | |
except LockException: | |
# another instance is probably running | |
# wait a bit for pid to be written to file | |
time.sleep(0.1) | |
pid, port = parse_pid_file() | |
look_for = 'Music Caster' if IS_FROZEN else 'python' | |
# double check if it's already running | |
# if more than one instance, there's definitely >3 processes | |
threshold = 3 if pid is None else 0 | |
if is_already_running(threshold=threshold, look_for=look_for, pid=pid): | |
if is_debug: | |
print('not exiting because we are DEBUGGING') | |
else: | |
activate_instance(port=port, timeout=5) | |
sys.exit() | |
else: | |
print('instance not found, lock broken?') | |
return file | |
portalocker.unlock(lock_file) | |
############################################################################### | |
# Windows alternative way | |
# https://code.activestate.com/recipes/474070-creating-a-single-instance-application/ | |
class SingleInstance: | |
def __init__(self, is_debug=False): | |
self.is_debug = is_debug | |
if platform.system() == 'Windows': | |
from win32event import CreateMutex | |
from win32api import GetLastError | |
from winerror import ERROR_ALREADY_EXISTS | |
self.mutex = CreateMutex(None, False, MUTEX_NAME) | |
if GetLastError() == ERROR_ALREADY_EXISTS: | |
self.on_already_running() | |
else: | |
self.on_lock(is_debug=is_debug) | |
else: | |
import portalocker | |
from portalocker.exceptions import LockException | |
self.file = open(LOCK_FILENAME, 'w+', encoding='utf-8') | |
try: | |
portalocker.lock(self.file, portalocker.LockFlags.NON_BLOCKING) | |
self.on_lock(is_debug=is_debug) | |
except LockException: | |
self.on_already_running() | |
def on_lock(self, is_debug=False): | |
create_pid_file() | |
if is_debug: | |
print(f'Locked {LOCK_FILENAME} pid = {os.getpid()}') | |
def on_already_running(self): | |
# single instance is apparently running, so wait for pid file to be written | |
time.sleep(0.1) | |
pid, port = parse_pid_file() | |
look_for = 'Music Caster' if IS_FROZEN else 'python' | |
# double check if it's already running | |
# if more than one instance, there's definitely >3 processes | |
threshold = 3 if pid is None else 0 | |
if is_already_running(threshold=threshold, look_for=look_for, pid=pid): | |
if not is_debug: | |
activate_instance(port=port, timeout=5) | |
sys.exit() | |
else: | |
print('not exiting because we are DEBUGGING') | |
else: | |
print('ERROR: locking mechanic broken') | |
def release(self): | |
if platform.system() == 'Windows': | |
from win32api import CloseHandle | |
CloseHandle(self.mutex) | |
else: | |
import portalocker | |
portalocker.unlock(self.file) | |
single_instance = SingleInstance(is_debug=DEBUG) | |
single_instance.release() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment