Skip to content

Instantly share code, notes, and snippets.

@kolypto
Created March 12, 2021 13:36
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save kolypto/218c187dfdde87fc21d60952aab3b057 to your computer and use it in GitHub Desktop.
A global named lock. An easy way to prevent multiple copies of a process from running concurrently
import sys
import socket
class SocketLock:
""" A global named lock. Better than pidfiles.
This uses a bind() on AF_UNIX sockets as a means to acquire a lock that is non-blocking.
Socket files are not created in the real filesystem, but in an "abstract namespace":
a Linux-only feature of hidden socket files.
This way of locking is ideal because it's released as soon as your application quits.
But it does not work in Windows. Don't use Windows.
And it does not work on Mac. It's your fault in this case.
"""
__slots__ = ('socket_name', '_sock', 'is_acquired')
# Only works in Linux
is_supported = (sys.platform == 'linux')
def __init__(self, path: str):
"""
Args:
path: Arbitrary path in the abstract socket namespace.
It's global! Transcends chroots!
"""
self.socket_name = path
self.is_acquired = False
def acquire(self) -> bool:
""" Attempt to acquire the lock
Note that you can do it only once, and then you'll need to release() it.
The lock will be released automatically if your process quits.
Returns:
True/False whether the lock has been acquired
"""
# Not supported? Do nothing
if not self.is_supported:
return True
# Already acquired?
assert not self._sock, 'Already acquired'
# Bind to an abstract socket
self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
try:
# The null byte (\0) means the the socket is created in the abstract namespace. Linux only
self._sock.bind('\0' + self.socket_name)
# Result?
except socket.error:
return False
else:
self.is_acquired = True
return True
def release(self):
""" Release the lock """
if not self.is_supported:
return
assert self._sock, 'Was never acquired'
self._sock.close()
self._sock = None
self.is_acquired = False
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment