工具，可以幫助在文件鎖定 - 專家提示想

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文件對象試圖讀/寫相同的實際文件。