Fast Thread- and multiprocess-safe file operations in python under Linux

Question

I'm trying to implement an alternative to python built-in open(), but safe for use in multi-threaded and multiprocessing environment. I'm using advisory locking, because all file operations in my application are performed using my locked_open() function and no other process is accessing that data.

It is only required to run on modern linux distributions, no compatibility with Windows or OS X is required. Also it have to be fast, because I'm going to use it for caching operations.

The code and the tests:

from contextlib import contextmanager
import os, fcntl
import unittest
from testfixtures import TempDirectory

M_READ, M_WRITE, M_READWRITE = range(3)
MODES = (
    # open for reading
    (os.O_RDONLY, fcntl.LOCK_SH, 'rb'),
    # create for writing
    (os.O_WRONLY | os.O_CREAT | os.O_TRUNC, fcntl.LOCK_EX, 'wb'),
    # open for reading and writing
    (os.O_RDWR | os.O_CREAT, fcntl.LOCK_EX, 'r+b')
)
BLOCKING_FLAGS = (fcntl.LOCK_NB, 0)

@contextmanager
def locked_open(filename, mode=M_READ, blocking=True):
    open_mode, flock_flags, mode_str = MODES[mode]
    flock_flags = flock_flags | BLOCKING_FLAGS[blocking]

    fd = os.open(filename, open_mode)
    fcntl.flock(fd, flock_flags)
    fileobj = os.fdopen(fd, mode_str)

    yield fileobj

    fileobj.flush()
    os.fdatasync(fd)
    fcntl.flock(fd, fcntl.LOCK_UN)
    fileobj.close()


class LockedOpenTestCase(unittest.TestCase):
    filename = 'testfile.txt'
    test_data = b'random text'

    def setUp(self):
        self.tempdir = TempDirectory()
        self.filepath = os.path.join(self.tempdir.path, self.filename)

    def tearDown(self):
        self.tempdir.cleanup()

    def test_write(self):
        with locked_open(self.filepath, M_WRITE) as f:
            f.write(self.test_data)
        self.assertEqual(self.test_data, self.tempdir.read(self.filename))

    def test_read(self):
        self.tempdir.write(self.filename, self.test_data)
        with locked_open(self.filepath, M_READWRITE) as f:
            self.assertEqual(self.test_data, f.read())

    def test_read_and_write(self):
        self.tempdir.write(self.filename, self.test_data)
        with locked_open(self.filepath, M_READWRITE) as f:
            data = f.read().swapcase()
            f.seek(0)
            f.write(data)

        self.assertEqual(
            self.test_data.swapcase(),
            self.tempdir.read(self.filename))

    def test_concurrent_read(self):
        self.tempdir.write(self.filename, self.test_data)
        with locked_open(self.filepath, M_READ) as f1:
            with locked_open(self.filepath, M_READ) as f2:
                self.assertEqual(self.test_data, f1.read())
                self.assertEqual(self.test_data, f2.read())

    def test_non_blocking_read_during_write_raises(self):
        with locked_open(self.filepath, M_WRITE):
            with self.assertRaises(IOError):
                with locked_open(self.filepath, M_READ, blocking=False):
                    pass

    def test_non_blocking_write_during_read_raises(self):
        self.tempdir.write(self.filename, self.test_data)
        with locked_open(self.filepath, M_READ):
            with self.assertRaises(IOError):
                with locked_open(self.filepath, M_WRITE, blocking=False):
                    pass

The code seems to be correct, and all tests are passing, but I get the feeling that I'm reinventing the wheel here.

Now the questions:

• Are there any issues with the code that I should be aware of? (besides from what I already said)
• Are my tests sufficient?
• Any suggestions regarding performance?
• Are there any easier ways to achieve same result?

Answer 1

• Are there any issues with the code that I should be aware of? (besides from what I already said)

If you haven't, maybe read Everything you never wanted to know about file locking; I guess the takeaway is that you should make sure that you're actually getting what you want with the Python fcntl module.

The code looks good to me in general, apart from the bad error handling.

When fcntl.flock throws, the file descriptor from os.open isn't closed. This is generally bad and can bite you in the future.

You don't have to care about unlocking as that's done then closing the file descriptor anyway.

And another issue you missed is that the yield has to be wrapped in a try/catch in order for the cleanup to run, c.f. contextmanager.

In summary it should look somewhat like this:

@contextmanager
def locked_open(filename, mode=M_READ, blocking=True):
    open_mode, flock_flags, mode_str = MODES[mode]
    flock_flags = flock_flags | BLOCKING_FLAGS[blocking]

    fd = os.open(filename, open_mode)
    try:
        fcntl.flock(fd, flock_flags)
        fileobj = os.fdopen(fd, mode_str)

        try:
            yield fileobj
        finally:
            fileobj.flush()
            os.fdatasync(fd)
    finally:
        os.close(fd)

• Are my tests sufficient?

You didn't actually test the interaction between multiple threads and processes. Since you're only locking a file descriptor once this will work as it is though.

• Any suggestions regarding performance?
• Are there any easier ways to achieve same result?

You'd have to measure that, but since you're using this from a single application only, using in- and shared-memory locks (from multiprocessing) could be a faster alternative instead of going through the filesystem. But it's way more effort on your end with all three access modes as you wrote them here and with multiple processes you'd easily run into more failure modes - I doubt it's worth it. That said, if you can get away with less features for locking, then you could still investigate it.