- 论坛徽章:
- 0
|
本帖最后由 kgd7558 于 2010-06-01 03:32 编辑
最近遇到一个问题,追了好久,如果我用法没错的话,可能是python的Bug吧?具体如下:
Python 2.6 final
Release date: 01-Oct-2008
1、封的一个二进制文件读写类,功能比较简单
2、没有涉及到多线程之类,没有互斥影响
3、所有地址相关整型都用long型,(int型也试过,也有问题)
4、在写入的文件大小比较大时,比如5G情况下,第一次可能可以正常写入
5、关键是在覆盖写入时(先seek,再write),seek正常,但write可能无返回
6、Linux下很正常,只是windows下偶尔出现,在NTFS以及FAT32上都会出现。
7、各位有遇到过没?- import os, threading
- class FileIO():
- def __init__(self, path):
- self.path = path
- self.fd = None
- self.__lock = threading.RLock()
- if not self.open():
- print('open file(%s) FAILED' % self.path)
- def __del__(self):
- self.close()
- def lock(self):
- self.__lock.acquire()
-
- def unlock(self):
- self.__lock.release()
- def open(self):
- if self.fd:
- return True
- try:
- if os.path.isfile(self.path):
- self.fd = open(self.path, 'r+b')
- else:
- self.fd = open(self.path, 'w+b')
- return True
- except:
- print('open device %s FAILED' % self.path)
- return False
- def close(self):
- if self.fd:
- self.fd.close()
- self.fd = None
- def get_size(self):
- try:
- stat = os.stat(self.path)
- return stat.st_size
- except:
- print('get file(%s) status FAILED.' % self.path);
- return 0
- def write(self, offset, buf):
- try:
- self.fd.seek(offset, os.SEEK_SET)
- self.fd.write(buf)
- self.fd.flush()
- return True
- except:
- print('write buffer FAILED(offset=%d, length=%d)' % (offset, len(buf)))
- return False
- def read(self, offset, length):
- try:
- self.fd.seek(offset, os.SEEK_SET)
- buf = self.fd.read(length)
- if len(buf) != length:
- print('read buffer FAILED(offset=%d, length=%d)' % (offset, length))
- buf = None
- return buf
- except:
- print('read buffer FAILED(offset=%d, length=%d)' % (offset, length))
- return None
-
- def flush(self):
- try:
- self.fd.flush()
- except:
- print('flush FAILED')
复制代码 |
|