test_util.py 文件源码

def lock_files(self, lock_type=None, offset=0, length=0):
        """Lock all files opened by open_files()."""
        if lock_type is None:
            lock_type = self.lock_type
        ret = []
        mode_str = 'WRITE' if lock_type == fcntl.F_WRLCK else 'READ'
        lockdata = struct.pack('hhllhh', lock_type, 0, long(offset), long(length), 0, 0)
        for fd in self.rfds + self.wfds:
            try:
                self.dprint('DBG3', "Lock file F_SETLKW (%s)" % mode_str)
                rv = fcntl.fcntl(fd, fcntl.F_SETLKW, lockdata)
                ret.append(rv)
            except Exception, e:
                self.warning("Unable to get lock on file: %r" % e)
        return ret