eugenegujing commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3754054635
##########
amber/src/main/python/core/models/internal_queue.py:
##########
@@ -77,11 +76,19 @@ def get(self) -> T:
def put(self, item: T) -> None:
if isinstance(item, InternalQueueElement):
if item.tag not in self._queue_ids:
- self._queue.add_sub_queue(item.tag, 1 if item.tag.is_control
else 2)
- self._queue_ids.add(item.tag)
- if isinstance(item, (DataElement, InternalMarker, ECMElement)):
- self._queue.put(item.tag, item)
- elif isinstance(item, DCMElement):
+ # registration must not interleave with
disable_data/enable_data
+ with self._lock:
+ if item.tag not in self._queue_ids:
+ self._queue.add_sub_queue(
+ item.tag, 1 if item.tag.is_control else 2
+ )
+ # while data is disabled, a new data sub-queue must
+ # start disabled too (before its first element is
+ # enqueued), or it would leak data during the pause
+ if not item.tag.is_control and self._queue_state:
+ self._queue.disable(item.tag)
+ self._queue_ids.add(item.tag)
Review Comment:
Fixed in 9dbbec219ea59a650aed63da74507aa4615c0ded. Function size_control has
the identical unlocked iteration, so I included it as well
##########
amber/src/main/python/core/models/internal_queue.py:
##########
@@ -77,11 +76,19 @@ def get(self) -> T:
def put(self, item: T) -> None:
if isinstance(item, InternalQueueElement):
if item.tag not in self._queue_ids:
- self._queue.add_sub_queue(item.tag, 1 if item.tag.is_control
else 2)
- self._queue_ids.add(item.tag)
- if isinstance(item, (DataElement, InternalMarker, ECMElement)):
- self._queue.put(item.tag, item)
- elif isinstance(item, DCMElement):
+ # registration must not interleave with
disable_data/enable_data
+ with self._lock:
+ if item.tag not in self._queue_ids:
+ self._queue.add_sub_queue(
+ item.tag, 1 if item.tag.is_control else 2
+ )
+ # while data is disabled, a new data sub-queue must
+ # start disabled too (before its first element is
+ # enqueued), or it would leak data during the pause
Review Comment:
Applied in 9dbbec219ea59a650aed63da74507aa4615c0ded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
