fcntl.flock() (Unix file lock) behaviour sampling script (Python recipe) by Jan Kaliszewski

ActiveState Code (http://code.activestate.com/recipes/577404/)

#!/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)