This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/release/v1.2/pr-7101-d3915afe1c81e9d591bba40ffc002a072c9965f2 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 7106df496367311d8e4b8325da4cd62a0561f132 Author: Yicong Huang <[email protected]> AuthorDate: Thu Jul 30 03:42:51 2026 -0400 fix(pyamber, v1.2): repair LinkedBlockingMultiQueue removal paths; add unit tests (#7101) ### What changes were proposed in this PR? Backport of #6903 to `release/v1.2`, cherry-picked from 38d94ead0ec267fb205857d16b61a1eed36172df. Follows the Direct Backport Push convention; opened as a PR (rather than a direct push) as part of a backport-coverage audit for fixes merged to `main` since early June that were never labeled for backport. ### Any related issues, documentation, discussions? Backport of #6903. Originally linked #6899. ### How was this PR tested? Release-branch CI runs on this PR. The cherry-pick applied cleanly onto `release/v1.2`; no manual conflict resolution was needed. ### Was this PR authored or co-authored using generative AI tooling? Yes — backport prepared with Claude Code (mechanical cherry-pick; the change itself is #6903 by its original author). Co-authored-by: Meng Wang <[email protected]> --- .../linked_blocking_multi_queue.py | 11 +- .../test_linked_blocking_multi_queue.py | 275 ++++++++++++++++++++- 2 files changed, 275 insertions(+), 11 deletions(-) diff --git a/amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py b/amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py index 735f0f6dc0..fcf175d6bc 100644 --- a/amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py +++ b/amber/src/main/python/core/util/customized_queue/linked_blocking_multi_queue.py @@ -154,9 +154,11 @@ class LinkedBlockingMultiQueue(IKeyedQueue): return False self.fully_lock() try: + # head is a sentinel, so the candidate node is always + # trail.next and trail is its predecessor. trail = self.head while trail.next is not None: - if trail.item == obj: + if trail.next.item == obj: self.unlink(trail, trail.next) return True trail = trail.next @@ -169,10 +171,11 @@ class LinkedBlockingMultiQueue(IKeyedQueue): trail: LinkedBlockingMultiQueue.Node, next_: LinkedBlockingMultiQueue.Node, ) -> None: - trail.item = None + next_.item = None trail.next = next_.next if self.last == next_: self.last = trail + self.count.dec() if self.enabled: self.owner.total_count.get_and_dec() @@ -418,7 +421,7 @@ class LinkedBlockingMultiQueue(IKeyedQueue): added = True break elif pg.priority > priority: - new_pg = LinkedBlockingMultiQueue.PriorityGroup(priority) + new_pg = self.PriorityGroup(priority) new_pg.add_queue(sub_queue) self.priority_groups.append(new_pg) added = True @@ -426,7 +429,7 @@ class LinkedBlockingMultiQueue(IKeyedQueue): i += 1 if not added: - new_pg = LinkedBlockingMultiQueue.PriorityGroup(priority) + new_pg = self.PriorityGroup(priority) new_pg.add_queue(sub_queue) self.priority_groups.append(new_pg) diff --git a/amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py b/amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py index 1df9423b69..8cdaada6bd 100644 --- a/amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py +++ b/amber/src/test/python/core/util/customized_queue/test_linked_blocking_multi_queue.py @@ -25,14 +25,26 @@ from core.util.customized_queue.linked_blocking_multi_queue import ( ) -class TestLinkedBlockingMultiQueue: - @pytest.fixture - def queue(self): - lbmq = LinkedBlockingMultiQueue() - lbmq.add_sub_queue("control", 0) - lbmq.add_sub_queue("data", 1) - return lbmq [email protected] +def queue(): + """A queue with `control` at priority 0 and `data` at priority 1.""" + lbmq = LinkedBlockingMultiQueue() + lbmq.add_sub_queue("control", 0) + lbmq.add_sub_queue("data", 1) + return lbmq + + +def chain_items(sub_queue): + """The items still reachable from a SubQueue's sentinel head, in order.""" + items = [] + node = sub_queue.head.next + while node is not None: + items.append(node.item) + node = node.next + return items + +class TestLinkedBlockingMultiQueue: def test_sub_can_emit(self, queue): assert queue.is_empty() queue.put("data", 1) @@ -235,3 +247,252 @@ class TestLinkedBlockingMultiQueue: assert total == sum(filter(lambda x: x % 3 != 0, range(11))) reraise() + + +class TestPeek: + """peek() at all three levels. It never waits — LinkedBlockingMultiQueue.peek + short-circuits on total_count == 0 — so these are safe single-threaded.""" + + def test_returns_none_on_empty_queue(self, queue): + assert queue.peek() is None + + def test_does_not_consume_the_item(self, queue): + queue.put("data", 1) + assert queue.peek() == 1 + assert queue.peek() == 1 + assert queue.size() == 1 + assert queue.get() == 1 + + def test_prefers_the_higher_priority_sub_queue(self, queue): + queue.put("data", 1) + queue.put("control", "s") + assert queue.peek() == "s" + + def test_returns_none_when_the_only_non_empty_sub_queue_is_disabled(self, queue): + queue.put("data", 1) + queue.disable("data") + assert queue.peek() is None + + def test_priority_group_peek_perturbs_round_robin_state(self): + lbmq = LinkedBlockingMultiQueue() + lbmq.add_sub_queue("first", 1) + lbmq.add_sub_queue("second", 1) + group = lbmq.get_sub_queue("first").priority_group + lbmq.put("second", "s") + + assert group.next_idx == 0 + assert group.peek() == "s" + # Skipping the empty `first` queue advanced next_idx, so peek is not + # side-effect free on the group's round-robin cursor. + assert group.next_idx == 1 + + def test_priority_group_peek_returns_none_when_all_queues_empty(self, queue): + group = queue.get_sub_queue("control").priority_group + assert group.peek() is None + + def test_selection_peek_returns_first_non_none_across_groups(self, queue): + queue.put("data", 1) + assert queue.sub_queue_selection.peek() == 1 + + +class TestSubQueueClear: + def test_clear_resets_count_and_collapses_the_chain(self, queue): + queue.put("data", 1) + queue.put("data", 2) + sub = queue.get_sub_queue("data") + + sub.clear() + + assert sub.size() == 0 + assert sub.is_empty() + assert sub.head is sub.last + assert chain_items(sub) == [] + assert queue.size() == 0 + + def test_clear_leaves_total_count_alone_when_disabled(self, queue): + queue.put("data", 1) + queue.put("control", "s") + queue.disable("data") + assert queue.size() == 1 + + queue.get_sub_queue("data").clear() + + assert queue.size() == 1 + + def test_clear_does_not_reset_in_mem_size(self, queue): + queue.put("data", 1) + before = queue.in_mem_size("data") + assert before > 0 + + queue.get_sub_queue("data").clear() + + # Current behavior: clear() drops the items but not the in-memory + # size accounting. + assert queue.in_mem_size("data") == before + + +class TestSubQueueRemove: + def test_remove_none_returns_false(self, queue): + assert queue.get_sub_queue("data").remove(None) is False + + def test_remove_missing_item_returns_false(self, queue): + queue.put("data", "a") + sub = queue.get_sub_queue("data") + + assert sub.remove("missing") is False + assert chain_items(sub) == ["a"] + assert sub.size() == queue.size() == 1 + + def test_remove_first_item(self, queue): + for item in ["a", "b", "c"]: + queue.put("data", item) + sub = queue.get_sub_queue("data") + + assert sub.remove("a") is True + assert chain_items(sub) == ["b", "c"] + assert sub.size() == queue.size() == 2 + assert queue.get() == "b" + + def test_remove_middle_item(self, queue): + for item in ["a", "b", "c"]: + queue.put("data", item) + sub = queue.get_sub_queue("data") + + assert sub.remove("b") is True + assert chain_items(sub) == ["a", "c"] + assert sub.size() == queue.size() == 2 + + def test_remove_last_item_fixes_up_the_tail(self, queue): + for item in ["a", "b", "c"]: + queue.put("data", item) + sub = queue.get_sub_queue("data") + + assert sub.remove("c") is True + assert chain_items(sub) == ["a", "b"] + assert sub.last.item == "b" + # The tail pointer must still be usable for the next enqueue. + queue.put("data", "d") + assert chain_items(sub) == ["a", "b", "d"] + + def test_remove_only_item_collapses_the_chain(self, queue): + queue.put("data", "only") + sub = queue.get_sub_queue("data") + + assert sub.remove("only") is True + assert sub.head is sub.last + assert sub.is_empty() + assert queue.size() == 0 + + def test_remove_on_disabled_sub_queue_leaves_total_count(self, queue): + queue.put("data", "a") + queue.put("data", "b") + queue.disable("data") + assert queue.size() == 0 + + assert queue.get_sub_queue("data").remove("a") is True + + # total_count already excludes a disabled sub-queue, so only the + # sub-queue's own count moves. + assert queue.size() == 0 + assert queue.size("data") == 1 + queue.enable("data") + assert queue.size() == 1 + + +class TestPriorityGroupRemoveQueue: + def test_remove_queue_shrinks_the_group(self, queue): + sub = queue.get_sub_queue("data") + group = sub.priority_group + + group.remove_queue(sub) + + assert group.queues == [] + + def test_remove_queue_resets_next_idx_when_it_reached_the_end(self): + lbmq = LinkedBlockingMultiQueue() + lbmq.add_sub_queue("first", 1) + lbmq.add_sub_queue("second", 1) + group = lbmq.get_sub_queue("first").priority_group + lbmq.put("first", "a") + assert lbmq.get() == "a" + assert group.next_idx == 1 + + group.remove_queue(lbmq.get_sub_queue("second")) + + assert len(group.queues) == 1 + assert group.next_idx == 0 + + def test_remove_queue_ignores_a_non_matching_key(self, queue): + control_group = queue.get_sub_queue("control").priority_group + + # `data` lives in a different group, so this is a no-op. + control_group.remove_queue(queue.get_sub_queue("data")) + + assert [q.key for q in control_group.queues] == ["control"] + + +class TestRemoveSubQueue: + def test_missing_key_returns_none_and_mutates_nothing(self, queue): + assert queue.remove_sub_queue("nope") is None + assert sorted(queue.sub_queues) == ["control", "data"] + assert len(queue.priority_groups) == 2 + + def test_removes_an_enabled_sub_queue_and_its_items(self, queue): + queue.put("control", "s") + queue.put("data", 1) + + removed = queue.remove_sub_queue("control") + + assert removed is not None + assert removed.key == "control" + assert sorted(queue.sub_queues) == ["data"] + # The removed sub-queue's item no longer counts toward the total. + assert queue.size() == 1 + + def test_drops_the_priority_group_once_it_is_empty(self, queue): + queue.remove_sub_queue("control") + + assert len(queue.priority_groups) == 1 + assert queue.priority_groups[0].priority == 1 + + +class TestSmallGuards: + def test_is_empty_tracks_size(self, queue): + sub = queue.get_sub_queue("data") + assert sub.is_empty() + queue.put("data", 1) + assert not sub.is_empty() + + def test_str_renders_an_arrow_chain(self, queue): + sub = queue.get_sub_queue("control") + assert str(sub) == "" + for item in ["a", "b", "c"]: + queue.put("control", item) + assert str(sub) == "a -> b -> c -> " + + def test_str_only_supports_str_items(self, queue): + queue.put("data", 1) + with pytest.raises(TypeError): + str(queue.get_sub_queue("data")) + + def test_put_none_raises_value_error(self, queue): + with pytest.raises(ValueError, match="Does not support NoneType."): + queue.put("data", None) + + def test_get_next_returns_none_when_everything_is_empty(self, queue): + # Called directly: q.get() on an empty queue would block forever. + assert queue.sub_queue_selection.get_next() is None + + def test_set_priority_groups_to_empty(self, queue): + queue.put("data", 1) + queue.sub_queue_selection.set_priority_groups([]) + + assert queue.sub_queue_selection.priority_groups == [] + assert queue.sub_queue_selection.get_next() is None + assert queue.sub_queue_selection.peek() is None + + def test_len_matches_size(self, queue): + assert len(queue) == 0 + queue.put("data", 1) + queue.put("control", "s") + assert len(queue) == queue.size() == 2
