eugenegujing commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3779941581
##########
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
pause/backpressure
+ if not item.tag.is_control and self._queue_state:
Review Comment:
Reworked to the dequeue-side option in
1a65167ba49f280df7cfeaf195adb6c87a3689b7
--
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]
