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 contextmanagerimport os, fcntlimport unittestfrom testfixtures import TempDirectoryM_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)@contextmanagerdef 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):                    passThe 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?
Fast Thread- and multiprocess-safe file operations in python under Linux
python;thread safety;file;linux;multiprocessing
Are there any issues with the code that I should be aware of? (besides from what I already said)If you haven't, maybe readEverything you never wanted to know about file locking;I guess the takeaway is that you should make sure that you're actuallygetting 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'tclosed.  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 thefile descriptor anyway.And another issue you missed is that the yield has to be wrapped in atry/catch in order for the cleanup to run,c.f. contextmanager.In summary it should look somewhat like this:@contextmanagerdef 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 andprocesses.  Since you're only locking a file descriptor once this willwork 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 singleapplication only, using in- and shared-memory locks (frommultiprocessing) could be a faster alternative instead of goingthrough the filesystem.  But it's way more effort on your end with allthree access modes as you wrote them here and with multiple processesyou'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.