eugenegujing commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3801283969
##########
amber/src/main/python/core/models/internal_queue.py:
##########
@@ -72,16 +71,54 @@ def is_empty(self, key=None) -> bool:
return self._queue.is_empty(key)
def get(self) -> T:
- return self._queue.get()
+ """Blocking get of the next available element.
+
+ Data channels register enabled even during a disable window, because
+ ECMs ride data channels and one swallowed by a channel that came up
+ disabled would never be acked. A DataElement arriving here during
+ such a window is withheld instead: its channel is closed and the
+ element goes back to its sub-queue's head, for enable_data() to
+ release. An ECM queued behind it on the same channel is therefore
+ delayed until resume, which is unavoidable without unbounded
+ buffering, and is what main does for channels disable_data() closed.
Review Comment:
Resolved in 87939215dd06e6b6710791c1691a0a74b17f901c
##########
amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py:
##########
@@ -149,6 +157,26 @@ def put(self, obj: T) -> None:
if old_size == 0:
self.owner._signal_not_empty()
+ def put_first(self, obj: T) -> None:
Review Comment:
Resolved in 87939215dd06e6b6710791c1691a0a74b17f901c
--
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]
