add contexts for locked files

python/lsst/daf/persistence/fmtPosixRepositoryCfg.py

@@ -23,6 +23,7 @@
 #
 
 import copy
+import errno
 import fcntl
 import yaml
 import os
@@ -56,37 +57,35 @@ def setRoot(cfg, loc):
                 cfg.root = None
             return cfg
 
-        loc = butlerLocation.storage.root
-
-        log = Log.getLogger("daf.persistence.butler")
-        if loc is None:
-            loc = cfg.root
         # This class supports schema 'file' and also treats no schema as 'file'.
-        # Split the URI and take only the path; remove the schema fom loc if it's there.
-        parseRes = urllib.parse.urlparse(loc)
-        loc = parseRes.path
-        if not os.path.exists(loc):
-            os.makedirs(loc)
-        loc = os.path.join(loc, butlerLocation.getLocations()[0])
+        # Split the URI and take only the path; remove the schema from loc if it's there.
+        loc = butlerLocation.storage.root
+        parseRes = urllib.parse.urlparse(loc if loc is not None else cfg.root)
+        loc = os.path.join(parseRes.path, butlerLocation.getLocations()[0])
         try:
-            with safeFileIo.FileForWriteOnceCompareSame(loc) as f:
+            with safeFileIo.LockedFileForRead(loc) as f:
+                existingCfg = RepositoryCfgPosixFormatter._read(f, parseRes.path)
+                if existingCfg == cfg:
+                    cfg.dirty = False
+                    return
+        except IOError as e:
+            if e.errno != errno.ENOENT:  # ENOENT is 'No such file or directory'
+                raise e
+        with safeFileIo.LockedFileForWrite(loc) as f:
+            existingCfg = RepositoryCfgPosixFormatter._read(f, parseRes.path)
+            if existingCfg is None:
                 cfgToWrite = setRoot(cfg, loc)
-                yaml.dump(cfgToWrite, f)
-                cfg.dirty = False
-        except safeFileIo.FileForWriteOnceCompareSameFailure:
-            with open(loc, 'r') as fileForRead:
-                log.debug("Acquiring blocking exclusive lock on {}", loc)
-                fcntl.flock(fileForRead, fcntl.LOCK_EX)
-                existingCfg = RepositoryCfgPosixFormatter._read(fileForRead, parseRes.path)
+            else:
+                if existingCfg == cfg:
+                    cfg.dirty = False
+                    return
                 try:
                     existingCfg.extend(cfg)
+                    cfgToWrite = setRoot(existingCfg, loc)
                 except ParentsMismatch as e:
                     raise RuntimeError("Can not extend existing repository cfg because: {}".format(e))
-                with open(loc, 'w') as fileForWrite:
-                    cfgToWrite = setRoot(cfg, loc)
-                    yaml.dump(cfg, fileForWrite)
-                    cfg.dirty = False
-                log.debug("Releasing blocking exclusive lock on {}", loc)
+            yaml.dump(cfgToWrite, f)
+            cfg.dirty = False
 
     @classmethod
     def _read(cls, fileObject, uri):
@@ -102,8 +101,9 @@ def _read(cls, fileObject, uri):
         A RepositoryCfg instance or None
         """
         repositoryCfg = yaml.load(fileObject)
-        if repositoryCfg.root is None:
-            repositoryCfg.root = uri
+        if repositoryCfg is not None:
+            if repositoryCfg.root is None:
+                repositoryCfg.root = uri
         return repositoryCfg
 
     @classmethod

python/lsst/daf/persistence/safeFileIo.py

@@ -25,9 +25,15 @@
 """
 from contextlib import contextmanager
 import errno
+import fcntl
 import filecmp
 import os
 import tempfile
+from lsst.log import Log
+
+
+class DoNotWrite(RuntimeError):
+    pass
 
 
 def safeMakeDir(directory):
@@ -106,12 +112,16 @@ def SafeFile(name):
     """
     outDir, outName = os.path.split(name)
     safeMakeDir(outDir)
+    doWrite = True
     with tempfile.NamedTemporaryFile(mode="w", dir=outDir, prefix=outName, delete=False) as temp:
         try:
             yield temp
+        except DoNotWrite:
+            doWrite = False
         finally:
-            os.rename(temp.name, name)
-            setFileMode(name)
+            if doWrite:
+                os.rename(temp.name, name)
+                setFileMode(name)
 
 
 @contextmanager
@@ -132,3 +142,95 @@ def SafeFilename(name):
     finally:
         os.rename(tempName, name)
         setFileMode(name)
+
+
+@contextmanager
+def SafeLockedFileForRead(name):
+    """Context manager for reading a file that may be locked with an exclusive lock via
+    SafeLockedFileForWrite.
+
+    This first tries to acquire the lock without blocking. If that fails then it will take a blocking lock
+    on the file. However, because SafeLockedFileForWrite uses a temporary file and then replaces the old file
+    with the temporary file, the fd that the blocking lock is against may no longer be valid when it unblocks.
+    To work around that, this will close the (possibly) old file and try again from the beginning by trying to
+    acquire the lock without blocking.
+
+    Parameters
+    ----------
+    name : string
+        The file name to be opened, may include path.
+
+    Yields
+    ------
+    file object
+        The file to be read from.
+    """
+    log = Log.getLogger("daf.persistence.butler")
+    try:
+        while True:
+            try:
+                with open(name, 'r') as f:
+                    log.debug("Acquiring non-blocking shared lock on {}", name)
+                    fcntl.flock(f, fcntl.LOCK_SH | fcntl.LOCK_NB)
+                    yield f
+                    break
+            except IOError as e:
+                if e.errno == errno.EAGAIN:
+                    with open(name, 'r') as f:
+                        log.debug("Acquiring blocking shared lock on {}", name)
+                        fcntl.flock(f, fcntl.LOCK_SH)
+    finally:
+        log.trace("Releasing blocking shared lock on {}", name)
+
+
+class _RWFile:
+    """File-like object that is used to contain readable and writable files to be yielded by
+    SafeLockedFileForWrite
+    """
+    def __init__(self, readable, writable):
+        self.readable = readable
+        self.writeable = writable
+        self.dirty = False
+
+    def read(self, size=None):
+        if size is not None:
+            return self.readable.read(size)
+        return self.readable.read()
+
+    def write(self, str):
+        self.writeable.write(str)
+        self.dirty = True
+
+
+@contextmanager
+def SafeLockedFileForWrite(name):
+    """Context manager to write a file in a manner that prevents other files from reading the old file while
+    the new one is being written. It opens the named file (creating it if needed) and locks it. It then
+    returns a file-like object that reads from the named file and writes to a temporary file. This allows a
+    file to atomically be opened, inspected, and changed if needed. After the user is done, if a write
+    operation has been performed we move the temporary file into the desired place and close the fd.
+
+    Parameters
+    ----------
+    name : string
+        The file name to be opened, may include path.
+
+    Yields
+    ------
+    file-like object
+        The file to be read from and written to.
+    """
+    log = Log.getLogger("daf.persistence.butler")
+    safeMakeDir(os.path.split(name)[0])
+    try:
+        with open(name, 'a+') as writeFile:
+            log.debug("Acquiring blocking exclusive lock on {}", name)
+            fcntl.flock(writeFile, fcntl.LOCK_EX)
+            with open(name, 'r') as readFile:
+                with SafeFile(name) as safeFile:
+                    rwFile = _RWFile(readFile, safeFile)
+                    yield rwFile
+                    if rwFile.dirty is False:
+                        raise DoNotWrite
+    finally:
+        log.debug("Released blocking exclusive lock on {}", name)

tests/testSafeFileIo.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python
+
+#
+# LSST Data Management System
+# Copyright 2017 LSST Corporation.
+#
+# This product includes software developed by the
+# LSST Project (http://www.lsst.org/).
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the LSST License Statement and
+# the GNU General Public License along with this program.  If not,
+# see <http://www.lsstcorp.org/LegalNotices/>.
+#
+
+import os
+import unittest
+import lsst.daf.persistence as dp
+from lsst.utils import getPackageDir
+import lsst.utils.tests
+import multiprocessing
+import shutil
+import time
+
+packageDir = os.path.join(getPackageDir('daf_persistence'))
+
+
+def setup_module(module):
+    lsst.utils.tests.init()
+
+
+def readFile(filename, readQueue):
+    readQueue.put("waiting")
+    readQueue.get()
+    with dp.safeFileIo.SafeLockedFileForRead(filename) as f:
+        readQueue.put(f.read())
+
+
+class TestFileLocking(unittest.TestCase):
+    """A test case for safeFileIo file read and write locking"""
+
+    testDir = os.path.join(packageDir, 'tests', 'TestFileLocking')
+
+    def setUp(self):
+        if os.path.exists(self.testDir):
+            shutil.rmtree(self.testDir)
+
+    def tearDown(self):
+        if os.path.exists(self.testDir):
+            shutil.rmtree(self.testDir)
+
+    def testWriteLock(self):
+        """Test SafeLockedFileForWrite by
+        1. open a file for write
+        2. spawn a second process that tries to read the file but should be blocked by the file lock
+        3. then write the file it and closing it (in the first process)
+        4. the second process should then be unblocked
+        5. read the file in the second process and return the result to the first process
+        6. compare what was written and read
+        """
+        readQueue = multiprocessing.Queue()
+        fileName = os.path.join(self.testDir, "testfile.txt")
+        proc = multiprocessing.Process(target=readFile, args=(fileName, readQueue))
+        testStr = "foobarbaz"
+        proc.start()
+        self.assertEqual(readQueue.get(), "waiting")
+        with dp.safeFileIo.SafeLockedFileForWrite(fileName) as f:
+            readQueue.put("go")
+            time.sleep(1)
+            f.write(testStr)
+        self.assertEqual(readQueue.get(), testStr)
+        proc.join()
+
+    def testNoChange(self):
+        """Test that if a file is opened and not changed that the file does not get changed"""
+        fileName = os.path.join(self.testDir, "testfile.txt")
+        # create the file with some contents
+        with dp.safeFileIo.SafeLockedFileForWrite(fileName) as f:
+            f.write("some test string")
+        # open the file but do not change it
+        with dp.safeFileIo.SafeLockedFileForWrite(fileName) as f:
+            pass
+        # open the file for read and test that it still contains the original test contents
+        with dp.safeFileIo.SafeLockedFileForRead(fileName) as f:
+            self.assertEqual(f.read(), "some test string")
+
+
+class MemoryTester(lsst.utils.tests.MemoryTestCase):
+    pass
+
+
+if __name__ == '__main__':
+    lsst.utils.tests.init()
+    unittest.main()