#!/usr/bin/python
# Unix/Linux only (as the 'fcntl' module is). Python >=2.6/3.x compatibile.
# Copyright (c) 2010 Jan Kaliszewski (zuo). Licensed under the MIT License.
"""
flocktests.py: fcntl.flock(LOCK_EX|LOCK_NB) behaviour sampling -- with one
file object or separate file objects (pointing to the same filesystem path),
with/without threading or forking.
"""
from __future__ import print_function
import os
import sys
import threading
from fcntl import flock, LOCK_EX, LOCK_UN, LOCK_NB
from os.path import basename
class lockpath(object):
"Open a file, flock it, generate appropriate info, unflock if necessary"
def __init__(self, path, _keep=False):
"By default, unflock/close the file immediately after setting the lock"
self.file = file = open(path, 'a')
locked = _lockonly(file)
if _keep:
self.locked = locked
else:
file.close()
self.locked = False
@classmethod
def keeping(cls, path):
"Constructor for with-blocks: tries to keep the file open and flocked"
return cls(path, _keep=True)
def __enter__(self):
"Enter a with-block binding the file to the as-clause target"
if self.locked:
return self.file
else:
raise RuntimeError('lockpath().file is not locked '
'-- cannot enter the with-block')
def __exit__(self, *args, **kwargs):
"Leave the with-block closing the file (=> unflocking it)"
self.file.close()
def _help(*args):
print('\n '.join(args), file=sys.stderr)
def _msg(*args):
print('pid:{0}'.format(os.getpid()),
threading.current_thread().name,
*args)
def _lockonly(file):
_msg('got file #', file.fileno())
try:
flock(file, LOCK_EX | LOCK_NB)
except IOError:
_msg('failed to lock')
return False
else:
_msg('locked successfully')
return True
def lockfile(file):
"flock a given file, then unflock it immediately"
if _lockonly(file):
flock(file, LOCK_UN)
# Options
def n(path):
"one file object + no concurrency"
with lockpath.keeping(path) as file:
lockfile(file)
def N(path):
"separate file objects + no concurrency"
with lockpath.keeping(path):
lockpath(path)
def t(path):
"one file object + threading"
with lockpath.keeping(path) as file:
t = threading.Thread(target=lockfile, args=(file,))
t.start()
t.join()
def T(path):
"separate file objects + threading"
with lockpath.keeping(path):
t = threading.Thread(target=lockpath, args=(path,))
t.start()
t.join()
def f(path):
"one file object + forking"
with lockpath.keeping(path) as file:
if os.fork():
os.wait()
else:
lockfile(file)
def F(path):
"separate file objects + forking"
with lockpath.keeping(path):
if os.fork():
os.wait()
else:
lockpath(path)
OPTIONS = 'nNtTfF'
def main(program, option='', path='test.flock'):
"Do one of the tests or print a short help"
flocktests = globals()
option = option.lstrip('-')
if option and (option in OPTIONS):
function = flocktests[option]
function(path)
else:
_help(__doc__.lstrip())
_help('Usage: {0} OPTION [PATH]'.format(basename(program)),
'Default PATH: test.flock', 'OPTIONS:',
*('-{0} {1}'.format(option, flocktests[option].__doc__)
for option in OPTIONS))
if __name__ == '__main__':
main(*sys.argv)
Diff to Previous Revision
--- revision 2 2010-09-22 00:10:22
+++ revision 3 2010-09-22 00:11:12
@@ -70,7 +70,7 @@
return True
def lockfile(file):
- "flock a given file, then unflock it immediately without closing"
+ "flock a given file, then unflock it immediately"
if _lockonly(file):
flock(file, LOCK_UN)
