Yicong-Huang commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3782142166
##########
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:
`main` is the branch this docstring is about to become part of, so once
merged the sentence compares the code to itself. The behavior it points at is
real: a channel `disable_data()` already closed does hold its ECM until resume.
Naming that instead of the branch keeps the rationale and survives the merge.
```suggestion
buffering, and already happens for channels disable_data() has
closed.
```
Same wording at `test_internal_queue.py:517`.
##########
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:
`put_first`/`enqueue_first` get no tests in this file's own suite. It is
otherwise thorough about exactly this: `remove()` — the method `put_first`
mirrors and whose locking it cites — has six, covering
first/middle/last/only-item and the disabled-sub-queue accounting. `put` has a
`None`-guard test.
The gap matters because `InternalQueue.get()` always disables the sub-queue
first. So this method's `if self.enabled` branch, its `old_size == 0` wake-up,
and its `None` guard are executed by nothing. A pair of tests mirroring
`test_remove_*` would close it.
##########
amber/src/test/python/core/models/test_internal_queue.py:
##########
@@ -364,3 +383,536 @@ def test_it_can_disable_and_enable_a_single_data_channel(
queue.enable(data_channel)
assert queue.get() is blocked
assert queue.is_empty()
+
+ # Regression tests below: a data channel whose sub-queue is created lazily
+ # (on the channel's first put) while disable_data is in effect comes up
+ # ENABLED, because ECMs ride data channels and an ECM landing first on
+ # such a channel must still be delivered. DataElements are instead
+ # withheld on the way out of get(): the channel is closed and the element
+ # is pushed back to the head of its own sub-queue, so a paused or
+ # backpressured worker never consumes data, nothing is lost or reordered,
+ # and is_data_enabled() stays False for the whole disabled period.
+
+ @pytest.mark.timeout(2)
+ @pytest.mark.parametrize(
+ "disable_type",
+ [
+ InternalQueue.DisableType.DISABLE_BY_PAUSE,
+ InternalQueue.DisableType.DISABLE_BY_BACKPRESSURE,
+ ],
+ )
+ def test_ecm_first_on_a_channel_registered_mid_disable_is_delivered(
+ self, queue, data_channel, disable_type
+ ):
+ # The must-fix: ECMs travel on data channels, so a reconfiguration ECM
+ # that is the FIRST-EVER message of a channel registered mid-pause has
+ # to come out; otherwise it is never acked and the coordinator's await
+ # expires. The timeout turns a regression into a failure, not a hang.
+ queue.disable_data(disable_type)
+ ecm = self.ecm_element(data_channel)
+ queue.put(ecm)
+ assert queue.get() is ecm
+
+ @pytest.mark.timeout(10)
+ @pytest.mark.parametrize(
+ "first_element_kind, expected_delivered",
+ [
+ ("data", False),
+ ("ecm", True),
+ ("dcm", True),
+ ],
+ )
+ def test_first_element_kind_decides_delivery_mid_disable(
+ self,
+ queue,
+ data_channel,
+ first_element_kind,
+ expected_delivered,
+ ):
+ # The matrix dimension that matters is the ELEMENT kind, not just the
+ # channel kind: on a data channel registered mid-disable, only a
+ # DataElement is withheld; control-carrying elements flow. The dcm case
+ # is a type gate rather than a real message shape, since a DCMElement
+ # is never tagged with a data channel in production.
+ first = {
+ "data": self.data_element,
+ "ecm": self.ecm_element,
+ "dcm": self.dcm_element,
+ }[first_element_kind](data_channel)
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ queue.put(first) # the channel's first-ever message, mid-disable
+
+ consumer, taken = self.start_consumer(queue)
+ consumer.join(1)
+ if expected_delivered:
+ assert not consumer.is_alive()
+ assert taken[0] is first
+ else:
+ # withheld: nothing is handed out, and the channel is closed
+ assert consumer.is_alive()
+ assert taken == []
+ assert not queue._queue.is_enabled(data_channel)
+ assert queue.size_data() == 1
+ # released only on resume
+ assert
queue.enable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ consumer.join(5)
+ assert not consumer.is_alive()
+ assert taken[0] is first
+
+ @pytest.mark.timeout(10)
+ def test_data_first_on_a_channel_registered_mid_disable_is_withheld(
+ self, queue, control_channel, data_channel
+ ):
+ # A DataElement handed to get() while data is disabled must be put
+ # back, not consumed: the channel closes, the element stays queued and
+ # keeps its place, and everything queued behind it follows in FIFO
+ # order once data is re-enabled.
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ data_elements = [self.data_element(data_channel) for _ in range(3)]
+ dcm = self.dcm_element(control_channel)
+ queue.put(data_elements[0])
+ queue.put(dcm)
+ in_mem_size_before = queue.in_mem_size()
+
+ # only the control traffic is handed out; this get() is already where
+ # the data element is withheld, closing its channel and leaving it
+ # queued in place before the DCM is returned
+ assert queue.get() is dcm
+ # so a consumer coming back for more now gets nothing
+ consumer, taken = self.start_consumer(queue)
+ consumer.join(1)
+ assert consumer.is_alive()
+ assert taken == []
+ assert not queue._queue.is_enabled(data_channel)
+ assert not queue.is_data_enabled()
+ assert queue.size_data() == 1
+ assert queue.in_mem_size() == in_mem_size_before
+
+ # more data arrives on the now-closed channel and queues up behind it
+ queue.put(data_elements[1])
+ queue.put(data_elements[2])
+ assert queue.size_data() == 3
+ assert consumer.is_alive()
+
+ assert queue.enable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ assert queue.is_data_enabled()
+ consumer.join(5)
+ assert not consumer.is_alive()
+ # FIFO is preserved across the withhold: the released element is the
+ # one that was put back, and the later ones follow it
+ results = taken + [queue.get() for _ in range(2)]
+ assert all(got is put for got, put in zip(results, data_elements))
+ assert queue.is_empty()
+ assert queue.in_mem_size() == 0
+
+ @pytest.mark.timeout(2)
+ def test_an_ecm_queued_behind_withheld_data_is_delayed_until_resume(
+ self, queue, control_channel, data_channel
+ ):
+ # KNOWN, PRE-EXISTING LIMITATION, asserted so nobody "fixes" it
+ # silently: withholding a DataElement closes its channel, which also
+ # holds back an ECM queued behind it on that SAME channel. Per-channel
+ # FIFO, "no data while paused" and "deliver ECMs immediately" cannot
+ # all hold once data comes first, short of unbounded buffering that
+ # would defeat backpressure. main behaves the same way for a channel
+ # already disabled by disable_data.
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ data = self.data_element(data_channel)
+ ecm = self.ecm_element(data_channel)
+ dcm = self.dcm_element(control_channel)
+ queue.put(data)
+ queue.put(ecm)
+ queue.put(dcm)
+ assert queue.get() is dcm
+ # the ECM does not overtake the withheld data element in front of it
+ consumer, taken = self.start_consumer(queue)
+ consumer.join(1)
+ assert consumer.is_alive()
+ assert taken == []
+ assert not queue._queue.is_enabled(data_channel)
+ assert queue.size_data() == 2
+ # both are released, in order, on resume
+ assert queue.enable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ consumer.join(5)
+ assert not consumer.is_alive()
+ assert taken[0] is data
+ assert queue.get() is ecm
+
+ @pytest.mark.timeout(2)
+ def test_is_data_enabled_stays_false_while_a_disable_reason_is_active(
+ self, queue, data_channel, second_data_channel
+ ):
+ # main_loop's pause wait-loop spins while `not is_control_empty() or
+ # not is_data_enabled()`, so a channel registering mid-pause must not
+ # make is_data_enabled() flip back to True and let the loop exit.
+ queue.put(self.data_element(data_channel))
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ assert not queue.is_data_enabled()
+ # a brand-new channel's first-ever message arrives mid-pause
+ queue.put(self.data_element(second_data_channel))
+ assert not queue.is_data_enabled()
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_BACKPRESSURE)
+ # one reason cleared, one still active
+ assert not
queue.enable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ assert not queue.is_data_enabled()
+ assert
queue.enable_data(InternalQueue.DisableType.DISABLE_BY_BACKPRESSURE)
+ assert queue.is_data_enabled()
+
+ @pytest.mark.timeout(5)
+ def test_a_resume_racing_a_withhold_does_not_strand_the_channel(
+ self, queue, data_channel
+ ):
+ # An element taken while data is disabled, with the last disable
+ # reason cleared before the withhold takes effect, must still be
+ # handed out. Closing the channel at that point would leave it closed
+ # with no reason left for enable_data to clear, stranding that channel
+ # for good. The patched get() places the resume exactly in the window
Review Comment:
Nothing patches `get()` — line 583 swaps `queue._lock` for the
`ResumingLock` defined just below, which is what lands the resume inside the
withhold's lock acquisition. Worth correcting because this comment is the only
explanation of how the test reaches a window too narrow to race.
```suggestion
# for good. The patched lock places the resume exactly in the window
```
--
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]
