"""Named mutex handling (for Win32).""" import ctypes from ctypes import wintypes # Create ctypes wrapper for Win32 functions we need, with correct argument/return types _CreateMutex = ctypes.windll.kernel32.CreateMutexA _CreateMutex.argtypes = [wintypes.LPCVOID, wintypes.BOOL, wintypes.LPCSTR] _CreateMutex.restype = wintypes.HANDLE _WaitForSingleObject = ctypes.windll.kernel32.WaitForSingleObject _WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD] _WaitForSingleObject.restype = wintypes.DWORD _ReleaseMutex = ctypes.windll.kernel32.ReleaseMutex _ReleaseMutex.argtypes = [wintypes.HANDLE] _ReleaseMutex.restype = wintypes.BOOL _CloseHandle = ctypes.windll.kernel32.CloseHandle _CloseHandle.argtypes = [wintypes.HANDLE] _CloseHandle.restype = wintypes.BOOL class NamedMutex(object): """A named, system-wide mutex that can be acquired and released.""" def __init__(self, name, acquired=False): """Create named mutex with given name, also acquiring mutex if acquired is True. Mutex names are case sensitive, and a filename (with backslashes in it) is not a valid mutex name. Raises WindowsError on error. """ self.name = name self.acquired = acquired ret = _CreateMutex(None, False, name) if not ret: raise ctypes.WinError() self.handle = ret if acquired: self.acquire() def acquire(self, timeout=None): """Acquire ownership of the mutex, returning True if acquired. If a timeout is specified, it will wait a maximum of timeout seconds to acquire the mutex, returning True if acquired, False on timeout. Raises WindowsError on error. """ if timeout is None: # Wait forever (INFINITE) timeout = 0xFFFFFFFF else: timeout = int(round(timeout * 1000)) ret = _WaitForSingleObject(self.handle, timeout) if ret in (0, 0x80): # Note that this doesn't distinguish between normally acquired (0) and # acquired due to another owning process terminating without releasing (0x80) self.acquired = True return True elif ret == 0x102: # Timeout self.acquired = False return False else: # Waiting failed raise ctypes.WinError() def release(self): """Relase an acquired mutex. Raises WindowsError on error.""" ret = _ReleaseMutex(self.handle) if not ret: raise ctypes.WinError() self.acquired = False def close(self): """Close the mutex and release the handle.""" if self.handle is None: # Already closed return ret = _CloseHandle(self.handle) if not ret: raise ctypes.WinError() self.handle = None __del__ = close def __repr__(self): """Return the Python representation of this mutex.""" return '{0}({1!r}, acquired={2})'.format( self.__class__.__name__, self.name, self.acquired) __str__ = __repr__ # Make it a context manager so it can be used with the "with" statement def __enter__(self): self.acquire() return self def __exit__(self, exc_type, exc_val, exc_tb): self.release()