工具，可以帮助在文件锁定 - 专家提示想

Question

我写文件a）项规定的方法来使用的fcntl方便地锁定它（的子类，所以只支持Unix，然而这是OK，我ATM）和b ）当读或写断言该文件被适当地锁定时。

现在我不是这方面的专家（我刚刚读了one paper [de]了解它），并希望得到一些反馈：它是否安全，是否存在竞争条件，是否有其他方面可以做得更好......在这里是代码：

from fcntl import flock, LOCK_EX, LOCK_SH, LOCK_UN, LOCK_NB 

class LockedFile(file): 
    """ 
    A wrapper around `file` providing locking. Requires a shared lock to read 
    and a exclusive lock to write. 

    Main differences: 
    * Additional methods: lock_ex, lock_sh, unlock 
    * Refuse to read when not locked, refuse to write when not locked 
     exclusivly. 
    * mode cannot be `w` since then the file would be truncated before 
     it could be locked. 

    You have to lock the file yourself, it won't be done for you implicitly. 
    Only you know what lock you need. 

    Example usage:: 
     def get_config(): 
      f = LockedFile(CONFIG_FILENAME, 'r') 
      f.lock_sh() 
      config = parse_ini(f.read()) 
      f.close() 

     def set_config(key, value): 
      f = LockedFile(CONFIG_FILENAME, 'r+') 
      f.lock_ex() 
      config = parse_ini(f.read()) 
      config[key] = value 
      f.truncate() 
      f.write(make_ini(config)) 
      f.close() 
    """ 

    def __init__(self, name, mode='r', *args, **kwargs): 
     if 'w' in mode: 
      raise ValueError('Cannot open file in `w` mode') 

     super(LockedFile, self).__init__(name, mode, *args, **kwargs) 

     self.locked = None 

    def lock_sh(self, **kwargs): 
     """ 
     Acquire a shared lock on the file. If the file is already locked 
     exclusively, do nothing. 

     :returns: Lock status from before the call (one of 'sh', 'ex', None). 
     :param nonblocking: Don't wait for the lock to be available. 
     """ 
     if self.locked == 'ex': 
      return # would implicitly remove the exclusive lock 
     return self._lock(LOCK_SH, **kwargs) 

    def lock_ex(self, **kwargs): 
     """ 
     Acquire an exclusive lock on the file. 

     :returns: Lock status from before the call (one of 'sh', 'ex', None). 
     :param nonblocking: Don't wait for the lock to be available. 
     """ 
     return self._lock(LOCK_EX, **kwargs) 

    def unlock(self): 
     """ 
     Release all locks on the file. 
     Flushes if there was an exclusive lock. 

     :returns: Lock status from before the call (one of 'sh', 'ex', None). 
     """ 
     if self.locked == 'ex': 
      self.flush() 
     return self._lock(LOCK_UN) 

    def _lock(self, mode, nonblocking=False): 
     flock(self, mode | bool(nonblocking) * LOCK_NB) 
     before = self.locked 
     self.locked = {LOCK_SH: 'sh', LOCK_EX: 'ex', LOCK_UN: None}[mode] 
     return before 

    def _assert_read_lock(self): 
     assert self.locked, "File is not locked" 

    def _assert_write_lock(self): 
     assert self.locked == 'ex', "File is not locked exclusively" 


    def read(self, *args): 
     self._assert_read_lock() 
     return super(LockedFile, self).read(*args) 

    def readline(self, *args): 
     self._assert_read_lock() 
     return super(LockedFile, self).readline(*args) 

    def readlines(self, *args): 
     self._assert_read_lock() 
     return super(LockedFile, self).readlines(*args) 

    def xreadlines(self, *args): 
     self._assert_read_lock() 
     return super(LockedFile, self).xreadlines(*args) 

    def __iter__(self): 
     self._assert_read_lock() 
     return super(LockedFile, self).__iter__() 

    def next(self): 
     self._assert_read_lock() 
     return super(LockedFile, self).next() 


    def write(self, *args): 
     self._assert_write_lock() 
     return super(LockedFile, self).write(*args) 

    def writelines(self, *args): 
     self._assert_write_lock() 
     return super(LockedFile, self).writelines(*args) 

    def flush(self): 
     self._assert_write_lock() 
     return super(LockedFile, self).flush() 

    def truncate(self, *args): 
     self._assert_write_lock() 
     return super(LockedFile, self).truncate(*args) 

    def close(self): 
     self.unlock() 
     return super(LockedFile, self).close() 

（在文档字符串的例子，也是我目前使用的情况下，此）为已经阅读直到到这里

感谢，甚至可能回答:)

Answer 1

我也不是专家，但有一件事是OU应该改变，和一对夫妇其他考虑：

首先，使用assert这种方式是一个坏主意：如果Python是运行与-O或者-OO断言被关闭，你的两个assert_*_lock()的方法将总是返回True。

二 - 你需要一些测试。 :)我冒昧地添加了一个自定义错误类并编写了一些测试。前四次传球，最后一次失败;这引出了一个问题，如果文件正常打开（和其他非LockedFile对象一样）并且数据被写入它会发生什么？

哦，最后 - 名LockableFile让我更有意义，因为该文件可以是未锁定状态。

下面是我所做的更改：

class LockedFileError(OSError): # might want IOError instead 
    pass 

if __name__ == '__main__': 
    import unittest 
    import tempfile 
    import shutil 
    import os 

    class TestLockedFile(unittest.TestCase): 
     def setUp(self): 
      self.dir = tempfile.mkdtemp() 
      self.testfile = testfile = os.path.join(self.dir, 'opened.txt') 
      temp = open(testfile, 'w') 
      temp.write('[global]\nsetting1=99\nsetting2=42\n') 
      temp.close() 

     def tearDown(self): 
      shutil.rmtree(self.dir, ignore_errors=True) 

     def test_01(self): 
      "writes fail if not locked exclusively" 
      testfile = self.testfile 
      temp = LockedFile(testfile, 'r+') 
      self.assertRaises(LockedFileError, temp.write, 'arbitrary data') 
      temp.lock_sh() 
      self.assertRaises(LockedFileError, temp.write, 'arbitrary data') 

     def test_02(self): 
      "reads fail if not locked" 
      testfile = self.testfile 
      temp = LockedFile(testfile, 'r') 
      self.assertRaises(LockedFileError, temp.read) 

     def test_03(self): 
      "writes succeed if locked exclusively" 
      testfile = self.testfile 
      temp = LockedFile(testfile, 'r+') 
      temp.lock_ex() 
      temp.write('arbitrary data\n') 

     def test_04(self): 
      "reads succeed if locked" 
      testfile = self.testfile 
      temp = LockedFile(testfile, 'r') 
      temp.lock_sh() 
      temp.readline() 
      temp.lock_ex() 
      temp.readline() 

     def test_05(self): 
      "other writes fail if locked exclusively" 
      testfile = self.testfile 
      temp = LockedFile(testfile, 'r') 
      temp.lock_ex() 
      testing = open(testfile, 'r+') 
      # not sure if this should be OSError, IOError, or something else... 
      self.assertRaises(OSError, testing.write, 'this should fail\n') 

    unittest.main() 

有更多的测试应该被写入覆盖LockedFile的各种组合方式读取，写入，以及其他非LockedFile文件对象试图读/写相同的实际文件。