Yicong-Huang commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3739000232
##########
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:
Only half of `_queue_ids`'s accessors take the new guard.
`is_control_empty`, `is_data_empty`, `size_data`, `in_mem_size` and
`is_data_enabled` still iterate it unlocked.
They do race this `add`: `main_loop.py:171` calls `is_data_enabled()` on the
main-loop thread while a Flight thread runs `put`. CPython then raises
`RuntimeError: Set changed size during iteration` (reproduced), killing the
main-loop thread while the heartbeat survives — the same silent hang this PR
fixes.
Pre-existing, so not a blocker; closing it needs no lock, just
`tuple(self._queue_ids)` in those five methods.
##########
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:
The guard reads `_queue_state`, which `disable_data` also fills for
`DISABLE_BY_BACKPRESSURE` (backpressure_handler.py:43), so this branch fires
outside pauses too.
```suggestion
# enqueued), or it would leak data during
pause/backpressure
```
--
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]
