eugenegujing commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3634756751
##########
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:
Review Comment:
I think the lock might not quite be on the hot path as it's only reachable
on a channel's first put. I instrumented it to verify: for 100k puts across 10
channels acquire the lock exactly 10 times. Also, I think the layer below
already locks on every element: SubQueue.put takes put_lock and get takes
take_lock on each call. I did try a lock-free version (publish the channel id
first, then re-check _queue_state), but there's a window where the element
lands in the still-enabled sub-queue before disable_data gets to it, which
reopens the same leak this PR fixes. The lock-free issues could probably be
worked around by creating the sub-queue disabled and only enabling it after
rechecking the state, switching _queue_ids to copy-on-write, but that version
might be overcomplicated. This lock might feel like the simpler safe choice.
Happy to discuss!
--
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]
