eugenegujing commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3801284924
##########
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:
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]
