"""Portable, NFS-safe file locking with timeouts.
This code implements an NFS-safe file-based locking algorithm influenced by
the GNU/Linux open(2) manpage, under the description of the O_EXCL option.
From RH6.1:
[...] O_EXCL is broken on NFS file systems, programs which rely on it
for performing locking tasks will contain a race condition. The
solution for performing atomic file locking using a lockfile is to
create a unique file on the same fs (e.g., incorporating hostname and
pid), use link(2) to make a link to the lockfile. If link() returns
0, the lock is successful. Otherwise, use stat(2) on the unique file
to check if its link count has increased to 2, in which case the lock
is also successful.
The assumption made here is that there will be no `outside interference',
e.g. no agent external to this code will have access to link() to the affected
lock files.
LockFile objects support lock-breaking so that you can't wedge a process
forever. This is especially helpful in a web environment, but may not be
appropriate for all applications.
Locks have a `lifetime', which is the maximum length of time the process
expects to retain the lock. It is important to pick a good number here
because other processes will not break an existing lock until the expected
lifetime has expired. Too long and other processes will hang; too short and
you'll end up trampling on existing process locks -- and possibly corrupting
data. In a distributed (NFS) environment, you also need to make sure that
your clocks are properly synchronized.
Locks can also log their state to a log file. When running under Mailman, the
log file is placed in a Mailman-specific location, otherwise, the log file is
called `LockFile.log' and placed in the temp directory (calculated from
tempfile.mktemp()).
"""
import os
import socket
import time
import errno
import random
import traceback
from stat import ST_NLINK, ST_MTIME
DEFAULT_LOCK_LIFETIME = 15
CLOCK_SLOP = 10
try:
True, False
except NameError:
True = 1
False = 0
_logfile = None
def _get_logfile():
global _logfile
if _logfile is None:
try:
from Mailman.Logging.StampedLogger import StampedLogger
_logfile = StampedLogger('locks')
except ImportError:
import tempfile
dir = os.path.split(tempfile.mktemp())[0]
path = os.path.join(dir, 'LockFile.log')
class SimpleUserFile:
def __init__(self, path):
self.__fp = open(path, 'a', 1)
self.__prefix = '(%d) ' % os.getpid()
def write(self, msg):
now = '%.3f' % time.time()
self.__fp.write(self.__prefix + now + ' ' + msg)
_logfile = SimpleUserFile(path)
return _logfile
class LockError(Exception):
"""Base class for all exceptions in this module."""
class AlreadyLockedError(LockError):
"""An attempt is made to lock an already locked object."""
class NotLockedError(LockError):
"""An attempt is made to unlock an object that isn't locked."""
class TimeOutError(LockError):
"""The timeout interval elapsed before the lock succeeded."""
class LockFile:
"""A portable way to lock resources by way of the file system.
This class supports the following methods:
__init__(lockfile[, lifetime[, withlogging]]):
Create the resource lock using lockfile as the global lock file. Each
process laying claim to this resource lock will create their own
temporary lock files based on the path specified by lockfile.
Optional lifetime is the number of seconds the process expects to hold
the lock. Optional withlogging, when true, turns on lockfile logging
(see the module docstring for details).
set_lifetime(lifetime):
Set a new lock lifetime. This takes affect the next time the file is
locked, but does not refresh a locked file.
get_lifetime():
Return the lock's lifetime.
refresh([newlifetime[, unconditionally]]):
Refreshes the lifetime of a locked file. Use this if you realize that
you need to keep a resource locked longer than you thought. With
optional newlifetime, set the lock's lifetime. Raises NotLockedError
if the lock is not set, unless optional unconditionally flag is set to
true.
lock([timeout]):
Acquire the lock. This blocks until the lock is acquired unless
optional timeout is greater than 0, in which case, a TimeOutError is
raised when timeout number of seconds (or possibly more) expires
without lock acquisition. Raises AlreadyLockedError if the lock is
already set.
unlock([unconditionally]):
Relinquishes the lock. Raises a NotLockedError if the lock is not
set, unless optional unconditionally is true.
locked():
Return true if the lock is set, otherwise false. To avoid race
conditions, this refreshes the lock (on set locks).
"""
COUNTER = 0
def __init__(self, lockfile,
lifetime=DEFAULT_LOCK_LIFETIME,
withlogging=False):
"""Create the resource lock using lockfile as the global lock file.
Each process laying claim to this resource lock will create their own
temporary lock files based on the path specified by lockfile.
Optional lifetime is the number of seconds the process expects to hold
the lock. Optional withlogging, when true, turns on lockfile logging
(see the module docstring for details).
"""
self.__lockfile = lockfile
self.__lifetime = lifetime
self.__counter = LockFile.COUNTER
LockFile.COUNTER += 1
self.__tmpfname = '%s.%s.%d.%d' % (
lockfile, socket.gethostname(), os.getpid(), self.__counter)
self.__withlogging = withlogging
self.__logprefix = os.path.split(self.__lockfile)[1]
self.__owned = True
def __repr__(self):
return '<LockFile %s: %s [%s: %ssec] pid=%s>' % (
id(self), self.__lockfile,
self.locked() and 'locked' or 'unlocked',
self.__lifetime, os.getpid())
def set_lifetime(self, lifetime):
"""Set a new lock lifetime.
This takes affect the next time the file is locked, but does not
refresh a locked file.
"""
self.__lifetime = lifetime
def get_lifetime(self):
"""Return the lock's lifetime."""
return self.__lifetime
def refresh(self, newlifetime=None, unconditionally=False):
"""Refreshes the lifetime of a locked file.
Use this if you realize that you need to keep a resource locked longer
than you thought. With optional newlifetime, set the lock's lifetime.
Raises NotLockedError if the lock is not set, unless optional
unconditionally flag is set to true.
"""
if newlifetime is not None:
self.set_lifetime(newlifetime)
if not self.locked() and not unconditionally:
raise NotLockedError, '%s: %s' % (repr(self), self.__read())
def lock(self, timeout=0):
"""Acquire the lock.
This blocks until the lock is acquired unless optional timeout is
greater than 0, in which case, a TimeOutError is raised when timeout
number of seconds (or possibly more) expires without lock acquisition.
Raises AlreadyLockedError if the lock is already set.
"""
if timeout:
timeout_time = time.time() + timeout
self.__write()
self.__touch()
self.__writelog('laying claim')
loopcount = -1
while True:
loopcount += 1
try:
os.link(self.__tmpfname, self.__lockfile)
self.__writelog('got the lock')
self.__touch()
break
except OSError, e:
if e.errno == errno.ENOENT:
pass
elif e.errno <> errno.EEXIST:
self.__writelog('unexpected link error: %s' % e,
important=True)
os.unlink(self.__tmpfname)
raise
elif self.__linkcount() <> 2:
self.__writelog('unexpected linkcount: %d' %
self.__linkcount(), important=True)
elif self.__read() == self.__tmpfname:
self.__writelog('already locked')
raise AlreadyLockedError
pass
if timeout and timeout_time < time.time():
os.unlink(self.__tmpfname)
self.__writelog('timed out')
raise TimeOutError
if time.time() > self.__releasetime() + CLOCK_SLOP:
self.__break()
self.__writelog('lifetime has expired, breaking',
important=True)
elif not loopcount % 100:
self.__writelog('waiting for claim')
self.__sleep()
def unlock(self, unconditionally=False):
"""Unlock the lock.
If we don't already own the lock (either because of unbalanced unlock
calls, or because the lock was stolen out from under us), raise a
NotLockedError, unless optional `unconditionally' is true.
"""
islocked = self.locked()
if not islocked and not unconditionally:
raise NotLockedError
if islocked:
try:
os.unlink(self.__lockfile)
except OSError, e:
if e.errno <> errno.ENOENT: raise
try:
os.unlink(self.__tmpfname)
except OSError, e:
if e.errno <> errno.ENOENT: raise
self.__writelog('unlocked')
def locked(self):
"""Return true if we own the lock, false if we do not.
Checking the status of the lock resets the lock's lifetime, which
helps avoid race conditions during the lock status test.
"""
try:
self.__touch()
except OSError, e:
if e.errno == errno.EPERM:
return False
else:
raise
if self.__linkcount() <> 2:
return False
return self.__read() == self.__tmpfname
def finalize(self):
self.unlock(unconditionally=True)
def __del__(self):
if self.__owned:
self.finalize()
def _transfer_to(self, pid):
self.__touch()
winner = self.__read()
self.__tmpfname = '%s.%s.%d' % (
self.__lockfile, socket.gethostname(), pid)
os.link(self.__lockfile, self.__tmpfname)
self.__write()
self.__owned = False
os.unlink(winner)
assert self.__linkcount() == 2
assert self.locked()
self.__writelog('transferred the lock')
def _take_possession(self):
self.__tmpfname = tmpfname = '%s.%s.%d' % (
self.__lockfile, socket.gethostname(), os.getpid())
while self.__linkcount() <> 2 or self.__read() <> tmpfname:
time.sleep(0.25)
self.__writelog('took possession of the lock')
def _disown(self):
self.__owned = False
def __writelog(self, msg, important=0):
if self.__withlogging or important:
logf = _get_logfile()
logf.write('%s %s\n' % (self.__logprefix, msg))
traceback.print_stack(file=logf)
def __write(self):
oldmask = os.umask(002)
try:
fp = open(self.__tmpfname, 'w')
fp.write(self.__tmpfname)
fp.close()
finally:
os.umask(oldmask)
def __read(self):
try:
fp = open(self.__lockfile)
filename = fp.read()
fp.close()
return filename
except EnvironmentError, e:
if e.errno <> errno.ENOENT: raise
return None
def __touch(self, filename=None):
t = time.time() + self.__lifetime
try:
os.utime(filename or self.__tmpfname, (t, t))
except OSError, e:
if e.errno <> errno.ENOENT: raise
def __releasetime(self):
try:
return os.stat(self.__lockfile)[ST_MTIME]
except OSError, e:
if e.errno <> errno.ENOENT: raise
return -1
def __linkcount(self):
try:
return os.stat(self.__lockfile)[ST_NLINK]
except OSError, e:
if e.errno <> errno.ENOENT: raise
return -1
def __break(self):
try:
self.__touch(self.__lockfile)
except OSError, e:
if e.errno <> errno.EPERM: raise
winner = self.__read()
try:
os.unlink(self.__lockfile)
except OSError, e:
if e.errno <> errno.ENOENT: raise
try:
if winner:
os.unlink(winner)
except OSError, e:
if e.errno <> errno.ENOENT: raise
def __sleep(self):
interval = random.random() * 2.0 + 0.01
time.sleep(interval)
def _dochild():
prefix = '[%d]' % os.getpid()
lockfile = LockFile('/tmp/LockTest', withlogging=True, lifetime=120)
workinterval = 5 * random.random()
hitwait = 20 * random.random()
print prefix, 'workinterval:', workinterval
islocked = False
t0 = 0
t1 = 0
t2 = 0
try:
try:
t0 = time.time()
print prefix, 'acquiring...'
lockfile.lock()
print prefix, 'acquired...'
islocked = True
except TimeOutError:
print prefix, 'timed out'
else:
t1 = time.time()
print prefix, 'acquisition time:', t1-t0, 'seconds'
time.sleep(workinterval)
finally:
if islocked:
try:
lockfile.unlock()
t2 = time.time()
print prefix, 'lock hold time:', t2-t1, 'seconds'
except NotLockedError:
print prefix, 'lock was broken'
print prefix, 'webhit sleep:', hitwait
time.sleep(hitwait)
def _seed():
try:
fp = open('/dev/random')
d = fp.read(40)
fp.close()
except EnvironmentError, e:
if e.errno <> errno.ENOENT:
raise
import sha
d = sha.new(`os.getpid()`+`time.time()`).hexdigest()
random.seed(d)
def _onetest():
loopcount = random.randint(1, 100)
for i in range(loopcount):
print 'Loop %d of %d' % (i+1, loopcount)
pid = os.fork()
if pid:
pid, status = os.waitpid(pid, 0)
else:
_seed()
try:
_dochild()
except KeyboardInterrupt:
pass
os._exit(0)
def _reap(kids):
if not kids:
return
pid, status = os.waitpid(-1, os.WNOHANG)
if pid <> 0:
del kids[pid]
def _test(numtests):
kids = {}
for i in range(numtests):
pid = os.fork()
if pid:
kids[pid] = pid
else:
_seed()
try:
_onetest()
except KeyboardInterrupt:
pass
os._exit(0)
while kids:
_reap(kids)
if __name__ == '__main__':
import sys
import random
_test(int(sys.argv[1]))