Source code for cherrypy.lib.lockfile

Platform-independent file locking. Inspired by and modeled after zc.lockfile.

import os

    import msvcrt
except ImportError:

    import fcntl
except ImportError:

[docs]class LockError(Exception): 'Could not obtain a lock' msg = 'Unable to lock %r' def __init__(self, path):
super(LockError, self).__init__(self.msg % path)
[docs]class UnlockError(LockError): 'Could not release a lock'
msg = 'Unable to unlock %r' # first, a default, naive locking implementation
[docs]class NaiveLockFile(object): """ A default, naive locking implementation. Always fails if the file already exists. """ def __init__(self, path): self.path = path try: fd =, os.O_CREAT | os.O_WRONLY | os.O_EXCL) except OSError: raise LockError(self.path) os.close(fd)
[docs] def release(self):
[docs] def remove(self):
[docs]class SystemLockFile(object): """ An abstract base class for platform-specific locking. """ def __init__(self, path): self.path = path try: # Open lockfile for writing without truncation: self.fp = open(path, 'r+') except IOError: # If the file doesn't exist, IOError is raised; Use a+ instead. # Note that there may be a race here. Multiple processes # could fail on the r+ open and open the file a+, but only # one will get the the lock and write a pid. self.fp = open(path, 'a+') try: self._lock_file() except Exception: self.fp.close() del self.fp raise self.fp.write(' %s\n' % os.getpid()) self.fp.truncate() self.fp.flush()
[docs] def release(self): if not hasattr(self, 'fp'): return self._unlock_file() self.fp.close()
del self.fp
[docs] def remove(self): """ Attempt to remove the file """ try: os.remove(self.path) except Exception:
pass def _unlock_file(self): """Attempt to obtain the lock on self.fp. Raise UnlockError if not
[docs]class WindowsLockFile(SystemLockFile): def _lock_file(self): # Lock just the first byte try: msvcrt.locking(self.fp.fileno(), msvcrt.LK_NBLCK, 1) except IOError: raise LockError( def _unlock_file(self): try: msvcrt.locking(self.fp.fileno(), msvcrt.LK_UNLCK, 1) except IOError:
raise UnlockError(
[docs]class UnixLockFile(SystemLockFile): def _lock_file(self): flags = fcntl.LOCK_EX | fcntl.LOCK_NB try: fcntl.flock(self.fp.fileno(), flags) except IOError:
raise LockError( # no need to implement _unlock_file, it will be unlocked on close() LockFile = ( UnixLockFile if 'fcntl' in globals() else WindowsLockFile if 'msvcrt' in globals() else NaiveLockFile )