- 論壇徽章:
- 0
|
本帖最后由 kgd7558 于 2010-06-01 03:32 編輯
最近遇到一個(gè)問題,追了好久,如果我用法沒錯(cuò)的話,可能是python的Bug吧?具體如下:
Python 2.6 final
Release date: 01-Oct-2008
1、封的一個(gè)二進(jìn)制文件讀寫類,功能比較簡(jiǎn)單
2、沒有涉及到多線程之類,沒有互斥影響
3、所有地址相關(guān)整型都用long型,(int型也試過,也有問題)
4、在寫入的文件大小比較大時(shí),比如5G情況下,第一次可能可以正常寫入
5、關(guān)鍵是在覆蓋寫入時(shí)(先seek,再write),seek正常,但write可能無返回
6、Linux下很正常,只是windows下偶爾出現(xiàn),在NTFS以及FAT32上都會(huì)出現(xiàn)。
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')
復(fù)制代碼 |
|