Skip to content
Snippets Groups Projects
Commit 5480647c authored by Pierre-Yves David's avatar Pierre-Yves David :octopus:
Browse files

stream-clone: skip the memory_target logic when usage is unlimited

The overhead is not huge, but it still better to skip it.
parent 1330278b
No related branches found
No related tags found
2 merge requests!1292Draft: 7.0rc preparation,!1270more streambundle evolution
......@@ -1156,7 +1156,8 @@
else:
fc = _ThreadSafeFileChunker
data_queue = _DataQueue(memory_target=memory_target)
mark_used = data_queue.mark_used
if memory_target is not None:
mark_used = data_queue.mark_used
raw_data = util.chunkbuffer(data_queue)
w = threading.Thread(
......@@ -1271,7 +1272,7 @@
class _DataQueue:
"""A queue passing data from the bundle stream to other thread
It has a "target_memory" optional parameter to avoid buffering too much
It has a "memory_target" optional parameter to avoid buffering too much
information. The implementation is not exact and the memory target might be
exceed for a time in some situation.
"""
......@@ -1314,12 +1315,13 @@
This is meant to be used from another thread than the one filler the queue.
"""
with self._lock:
if offset > self._current_used:
self._current_used = offset
# If the reader is waiting for room, unblock it.
if self._wait.locked() and self._has_free_space():
self._wait.release()
if self._memory_target is not None:
with self._lock:
if offset > self._current_used:
self._current_used = offset
# If the reader is waiting for room, unblock it.
if self._wait.locked() and self._has_free_space():
self._wait.release()
def fill_from(self, data):
"""fill the data queue from a bundle2 part object
......@@ -1333,20 +1335,21 @@
q.put(item)
if self._abort:
break
with self._lock:
while not self._has_free_space():
# make sure the _wait lock is locked
# this is done under lock, so there case be no race with the release logic
self._wait.acquire(blocking=False)
self._lock.release()
# acquiring the lock will block until some other thread release it.
self._wait.acquire()
# lets dive into the locked section again
self._lock.acquire()
# make sure we release the lock we just grabed if
# needed.
if self._wait.locked():
self._wait.release()
if self._memory_target is not None:
with self._lock:
while not self._has_free_space():
# make sure the _wait lock is locked
# this is done under lock, so there case be no race with the release logic
self._wait.acquire(blocking=False)
self._lock.release()
# acquiring the lock will block until some other thread release it.
self._wait.acquire()
# lets dive into the locked section again
self._lock.acquire()
# make sure we release the lock we just grabed if
# needed.
if self._wait.locked():
self._wait.release()
finally:
q.put(None)
......@@ -1367,10 +1370,11 @@
"""
self._abort = True
self._q.put(None)
with self._lock:
# make sure we unstuck the reader thread.
if self._wait.locked():
self._wait.release()
if self._memory_target is not None:
with self._lock:
# make sure we unstuck the reader thread.
if self._wait.locked():
self._wait.release()
class _FileInfoQueue:
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment