mirrors/yt-dlp
1
0
Fork 0
mirror of https://github.com/yt-dlp/yt-dlp.git synced 2025-10-31 14:45:14 +00:00
Code Issues Projects Releases Wiki Activity

[utils] Improve file locking

Browse Source 
* Implement non-blocking locks for windows
* Don't raise error when closing a closed file
This commit is contained in:
pukkandan
2022-03-03 18:57:38 +05:30
parent acea8d7cfb
commit 747c0bd127
1 changed files with 12 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
yt_dlp/utils.py
View File

@@ -2122,22 +2122,22 @@ if sys.platform == 'win32':
whole_low = 0xffffffff
whole_high = 0x7fffffff
def _lock_file(f, exclusive, block): # todo: block unused on win32
def _lock_file(f, exclusive, block):
overlapped = OVERLAPPED()
overlapped.Offset = 0
overlapped.OffsetHigh = 0
overlapped.hEvent = 0
f._lock_file_overlapped_p = ctypes.pointer(overlapped)
handle = msvcrt.get_osfhandle(f.fileno())
if not LockFileEx(handle, 0x2 if exclusive else 0x0, 0,
whole_low, whole_high, f._lock_file_overlapped_p):
raise OSError('Locking file failed: %r' % ctypes.FormatError())
if not LockFileEx(msvcrt.get_osfhandle(f.fileno()),
(0x2 if exclusive else 0x0) | (0x0 if block else 0x1),
0, whole_low, whole_high, f._lock_file_overlapped_p):
raise BlockingIOError('Locking file failed: %r' % ctypes.FormatError())
def _unlock_file(f):
assert f._lock_file_overlapped_p
handle = msvcrt.get_osfhandle(f.fileno())
if not UnlockFileEx(handle, 0,
whole_low, whole_high, f._lock_file_overlapped_p):
if not UnlockFileEx(handle, 0, whole_low, whole_high, f._lock_file_overlapped_p):
raise OSError('Unlocking file failed: %r' % ctypes.FormatError())
else:
@@ -2175,6 +2175,8 @@ else:
class locked_file(object):
_closed = False
def __init__(self, filename, mode, block=True, encoding=None):
assert mode in ['r', 'rb', 'a', 'ab', 'w', 'wb']
self.f = io.open(filename, mode, encoding=encoding)
@@ -2192,9 +2194,11 @@ class locked_file(object):
def __exit__(self, etype, value, traceback):
try:
_unlock_file(self.f)
if not self._closed:
_unlock_file(self.f)
finally:
self.f.close()
self._closed = True
def __iter__(self):
return iter(self.f)